Subtle timing bug with `.remember()` dropping values

[matthewjamesadam]

Thank you for writing xstream, we use it heavily in a sizable project and have enjoyed using it.

There appears to be a subtle timing bug that seems to be caused by the way streams are asynchronously disconnected. The following unit test shows the bug. In the test, I have a producer (FiveProducer). that emits 5 but never completes, this is simply meant to mimic a stream that emits a value and may possibly emit more in the future. This producer has .compose(dropRepeats()).remember() added.

If I subscribe to this stream once (sub1 / spy1), and then unsubscribe from it, wait zero milliseconds, then subscribe a second time (spy2), I would expect both spy1 and spy2 to receive the value 5. However, spy2 does not.

class FiveProducer implements InternalProducer<number> {
    public type = 'FiveProducer';

    _start(out: InternalListener<number>): void {
        out._n(5);
    }

    _stop(): void {}
}

test('Verify dropRepeats behaviour', async () => {
    const stream = new Stream(new FiveProducer()).compose(dropRepeats()).remember();

    const spy1 = jest.fn();
    const sub1 = stream.subscribe({ next: spy1 });

    sub1.unsubscribe();

    // This is important -- must wait exactly zero.
    await jest.advanceTimersByTimeAsync(0);

    const spy2 = jest.fn();
    stream.subscribe({ next: spy2 });

    await jest.advanceTimersByTimeAsync(10);

    // This succeeds -- spy1 is called once with the right value
    expect(spy1).toHaveBeenCalledTimes(1);
    expect(spy1).toHaveBeenNthCalledWith(1, 5);

    // This fails -- spy2 is not called
    expect(spy2).toHaveBeenCalledTimes(1);
    expect(spy2).toHaveBeenNthCalledWith(1, 5);
});

What seems to be happening is:

• sub1 is unsubscribed, which schedules an asynchronous stop of the .remember() stream.
• Asynchronously, the .remember stream is stopped, which causes an asynchronous operation to be scheduled, to stop the .compose stream
• However, before this runs, my second spy attempts to subscribe to the stream. This starts the .remember() stream, which doesn't emit a remembered value immediately (it has been stopped, and .has==false. Instead it starts the .compose stream.
• The .compose stream runs this code, which cancels the stop timer, but does not cause any other action to happen:

        if (this._stopID !== NO) {
            clearTimeout(this._stopID);
            this._stopID = NO;
        }

The end effect is that the stream never receives any data at all. This feels like a race condition bug to me, but I do not know exactly where/how to fix this. I'd appreciate any advice, as we run into this bug somewhat often, and unpredictably, since our app heavily uses many streams asynchronously.

Thank you!

[matthewjamesadam]

This might be related to #310, in that I suspect if MemoryStream remembered its value I would not have run into this problem -- but since this affects stopping any type of stream, I suspect it might affect other streams and operators.

[matthewjamesadam]

I have been looking into fixing this issue -- the root cause does seem to be that Stream._add is synchronous, but Stream._remove is asynchronous.

Every Stream asynchronously unsubscribes from its Producer, but subscribing happens synchronously -- since the chain of Streams and Producers can be of arbitrary length, that means that after unsubscribing, behaviour when resubscribing can depend on exactly how long the Stream's chain of operators is, and how many milliseconds elapses between unsubscription and resubscription. The subscription operation will synchronously walk up the chain of Streams and Producers until it hits the point where the last Stream is waiting to asynchronously unsubscribe from its Producer. The end result is unpredictable.

Of course, this only matters in situations where unsubscribing and resubscribing happens in rapid succession -- but in our app we have seen this happen fairly frequently. Specifically, we use streams in React views, and when navigating between routes, an old view may unmount its stream while a new view subscribes to that same stream -- depending on the timing, this may cause the stream to completely stop emitting values.

Unfortunately, I do not believe there is any way to make the current behaviour deterministic or predictable. The best option I've thought of is to make only the user-accessible API (Stream.removeListener, StreamSub.unsubscribe) asynchronous, so that if user code rapidly unsubscribes and resubscribes, the stream is not shut down. I have put up a PR that implements this here: #341. This seems to work in our app.