websocket‎: Deliver the received signature notification instead of closing the connection

[kartsan03]

Problem

signature_subscribe(enable_received_notification=True) (new in #703, not released yet) closes the whole websocket connection as soon as the node reports the signature as received.

With that option the node sends an extra notification ahead of the processed one, and its value is a plain string instead of an object (ReceivedSignatureResult::ReceivedSignature in Agave's rpc-client-types/src/response.rs):

{"jsonrpc":"2.0","method":"signatureNotification","params":{"result":{"context":{"slot":505544335},"value":"receivedSignature"},"subscription":13566433}}

solders only models the processed shape (RpcSignatureResponse { err }), so parse_websocket_message raises SerdeJSONError. In SolanaWsClient that ends the reader task, which abandons the connection: every subscription on it is dropped, recv() raises, and the client has to be rebuilt.

I found this while going through the new signature_subscribe options from #703: solders' SignatureNotification only covers the processed result, so I checked what the node actually sends for the received one. Reproduced on devnet at 96047a5 by subscribing to a transfer's signature with enable_received_notification=True and then sending the transfer:

recv raised SerdeJSONError: a list or single element
connection state: ConnectionState.CLOSED

Fix

• _parse_frame recognizes the "receivedSignature" notification before the frame reaches solders and returns a new SignatureReceivedNotification(subscription, slot). Every other frame still goes through parse_websocket_message.
• recv() and iteration yield it like any other notification. Only the processed SignatureNotification drops the local handle, since the server keeps the subscription until then, so unsubscribe() still works in between.
• The signature_subscribe docstring and the cookbook page mention the extra notification.

The underlying gap is in solders, which could model this as an enum like Agave does. Handling it here keeps the option usable with the solders version this release depends on.

I left CHANGELOG.md alone since #705 is rewriting the Unreleased section. Happy to add a line to either PR.

Verification

• New unit test test_received_signature_notification_is_delivered feeds the received notification and then the processed one through the fake socket, and checks that the connection stays open, the handle survives the first notification and is dropped by the second. Removing the parsing change makes it fail with SerdeJSONError; letting the received notification drop the handle makes it fail too.
• make lint and make unit-tests pass (202 passed).
• Devnet with this branch, same script as above:

SignatureReceivedNotification(subscription=13566433, slot=505544335)
SignatureNotification { result: SignatureNotificationResult { context: RpcResponseContext { slot: 505544335, api_version: None }, value: RpcSignatureResponse { err: None } }, subscription: 13566433 }
connection state: ConnectionState.OPEN

Devnet script

import asyncio

from solders.keypair import Keypair
from solders.message import Message
from solders.system_program import TransferParams, transfer
from solders.transaction import Transaction

from solana.rpc.async_api import AsyncClient
from solana.rpc.websocket_api import SolanaWsClient

PAYER = Keypair()  # a funded devnet keypair


async def main() -> None:
    async with AsyncClient("https://api.devnet.solana.com") as http:
        blockhash = (await http.get_latest_blockhash()).value.blockhash
        ix = transfer(TransferParams(from_pubkey=PAYER.pubkey(), to_pubkey=Keypair().pubkey(), lamports=1_000_000))
        tx = Transaction([PAYER], Message([ix], PAYER.pubkey()), blockhash)

        async with SolanaWsClient("wss://api.devnet.solana.com") as ws:
            await ws.signature_subscribe(signature=tx.signatures[0], enable_received_notification=True)
            await http.send_raw_transaction(bytes(tx))
            try:
                for _ in range(2):
                    print(await asyncio.wait_for(ws.recv(), 60))
            except Exception as exc:
                print(f"recv raised {type(exc).__name__}: {exc}")
            print("connection state:", ws.connection_state)


asyncio.run(main())

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 95.30%. Comparing base (96047a5) to head (ed47e7d).

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #706      +/-   ##
==========================================
+ Coverage   95.26%   95.30%   +0.03%     
==========================================
  Files          25       25              
  Lines        1902     1917      +15     
==========================================
+ Hits         1812     1827      +15     
  Misses         90       90              

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[kingsznhone]

looks good

But I think fix in solders is better. I'll try open a pr to solders

[kartsan03]

Thanks! Agreed that solders is the right place for the parsing. I can open that PR since I already have the repro, unless you've started on it.

Since solana-py pins solders<0.30 and the fix needs a solders release, could we keep this one as the stopgap until then? Once a release with the fix is out, I'll send a follow-up here that bumps solders and drops the workaround.

[kingsznhone]

Maybe we can add this temp fix in solana-py and waiting for solders fix?
@michaelhly

[michaelhly]

Thank you. Let me make a minor release on master first.