Prevent accumulation of subscription notifications when the last AsyncIterator falls out of scope

[steveluscher]

Overview

There's something about the chain of AsyncGenerators in the subscriptions transport that results in orphaned AsyncIterators stuffing messages into queuedMessages all day long until you run out of memory.

1. If you abort the subscription (unsubscribe) these go away. Good.
2. If, however, you're holding on to an AsyncIterator and you stop iterating over it, an inner iterator won't learn of that fact; it will keep pumping notifications into queuedMessages forever.

Steps to reproduce

Instrument around here where messages are added to queuedMessages and watch as the number of messages destined for the AsyncIterator in (A) grows forever.

import { createSolanaRpcSubscriptions_UNSTABLE } from '@solana/web3.js';

const rpcSubscriptions = createSolanaRpcSubscriptions_UNSTABLE(
    'wss://api.devnet.solana.com',
);

const abortSignal = AbortSignal.timeout(5000);

// Imagine this code lives in one part of your app (A).
(async () => {
    const slotNotifications = await rpcSubscriptions
        .slotsUpdatesNotifications()
        .subscribe({ abortSignal });

    console.log('>>>>> (A) start caring about slot notifications');
    for await (const notification of slotNotifications) {
        if (notification.type === 'firstShredReceived') {
            // Stop iterating after receiving a particular notification.
            break;
        }
    }
    console.log('>>>>> (A) stop caring about slot notifications');
    // The `AsyncIterator` created by `of slotNotifications` has fallen
    // out of scope and should be eligible for garbage collection now.
})();

// Imagine this code lives in a different part of your app, (B).
(async () => {
    const slotNotifications = await rpcSubscriptions
        .slotsUpdatesNotifications()
        .subscribe({ abortSignal });

    console.log('>>>>> (B) start caring about slot notifications');
    for await (const _notification of slotNotifications) {
        // Use every notification. Never stop.
    }
    console.log('>>>>> (B) stop caring about slot notifications');
})();

Description of bug

Right now the AsyncIterable returned by subscribe() offers certain guarantees about message delivery that we might have to reevaluate. In particular, it guarantees that if there's a delay between having processed a value and the time you ask the AsyncIterator for the next value, any subscription notifications received during that time will be queued and delivered to you all at once. If you never ask for the next value and you never abort the subscription (unsubscribe) then this queue grows forever.

We might need to eliminate this guarantee. Alternatively, maybe this is one of those rare use cases for FinalizationRegistry, if only the thing that's doing the queueing can know when the thing consuming the messages has fallen out of scope.