Fixed RPC subscriptions leak when subscription stream is finished

[svyatonik]

closes paritytech/parity-bridges-common#3000

Recently we've changed our bridge configuration for Rococo <> Westend and our new relayer has started to submit transactions every ~ 30 seconds. Eventually, it switches itself into limbo state, where it can't submit more transactions - all author_submitAndWatchExtrinsic calls are failing with the following error: ERROR bridge Failed to send transaction to BridgeHubRococo node: Call(ErrorObject { code: ServerError(-32006), message: "Too many subscriptions on the connection", data: Some(RawValue("Exceeded max limit of 1024")) }).

Some links for those who want to explore:

• server side (node) has a strict limit on a number of active subscriptions. It fails to open a new subscription if this limit is hit: https://github.com/paritytech/jsonrpsee/blob/a4533966b997e83632509ad97eea010fc7c3efc0/server/src/middleware/rpc/layer/rpc_service.rs#L122-L132. The limit is set to 1024 by default;
• internally this limit is a semaphore with limit permits: https://github.com/paritytech/jsonrpsee/blob/a4533966b997e83632509ad97eea010fc7c3efc0/core/src/server/subscription.rs#L461-L485;
• semaphore permit is acquired in the first link;
• the permit is "returned" when the SubscriptionSink is dropped: https://github.com/paritytech/jsonrpsee/blob/a4533966b997e83632509ad97eea010fc7c3efc0/core/src/server/subscription.rs#L310-L325;
• the SubscriptionSink is dropped when this polkadot-sdk function returns. In other words - when the connection is closed, the stream is finished or internal subscription buffer limit is hit;
• the subscription has the internal buffer, so sending an item contains of two steps: reading an item from the underlying stream and sending it over the connection;
• when the underlying stream is finished, the inner_pipe_from_stream wants to ensure that all items are sent to the subscriber. So it: waits until the current send operation completes and then send all remaining items from the internal buffer. Once it is done, the function returns, the SubscriptionSink is dropped, semaphore permit is dropped and we are ready to accept new subscriptions;
• unfortunately, the code just calls the pending_fut.await.is_err() to ensure that the current send operation completes. But if there are no current send operation (which is normal), then the pending_fut is set to terminated future and the await never completes. Hence, no return from the function, no drop of SubscriptionSink, no drop of semaphore permit, no new subscriptions allowed (once number of susbcriptions hits the limit.

I've illustrated the issue with small test - you may ensure that if e.g. the stream is initially empty, the subscription_is_dropped_when_stream_is_empty will hang because pipe_from_stream never exits.

[acatangiu]

Nice catch! 💯

[niklasad1]

@svyatonik nice digging and you got it right.

I guess we could maybe remove this subscription limit since the server is backpressured these days