PubSub iterator promise never resolves if return() called too soon

[arendjr]

The following snippet will result in an iterator promise returned by next() that never resolves:

const iterator = new PubSub().asyncIterator();
const promise = iterator.next();
iterator.return();
await promise; // will wait forever

The trick is that return() gets called synchronously while the first call to next() is still setting up its subscriptions.

This appears to be a regression since 1.0 since I have a test suite that was green and started failing on this issue when I upgraded to 1.1.

I have tried various solutions, including monkey-patching emptyQueue() like this:

PubSubAsyncIterator.prototype.emptyQueue = async function() {
    if (this.running) {
        this.running = false;
        this.pullQueue.forEach(resolve => resolve({ value: undefined, done: true }));
        this.pullQueue.length = 0;
        this.pushQueue.length = 0;
        const subscriptionIds = await this.allSubscribed;
        if (subscriptionIds) {
            this.unsubscribeAll(subscriptionIds);
        }
        // This one should make sure new next() calls while we were awaiting and resolved:
        this.pullQueue.forEach(resolve => resolve({ value: undefined, done: true }));
    }
};

But interestingly, doing so made other tests in my suite fail for reasons I'm afraid I don't understand right now... Ignore this last part, that failure seemed to be caused by me mis-using Jest's .resolves method (forgetting to await it).

[arendjr]

Okay, I think I have gotten to the bottom of it. I can get all my tests green again as long as I call iterator.next() before doing anything else in my tests. The fact that things that got published previously (despite no one waiting for it yet), which don't get published anymore unless iterator.next() has been called (due to the lazy initialization of the subscriptions) wreaked quite some havoc in my tests. But I can solve it now without needing any further patching of the graphql-subscriptions library.

Still, given this experience, I believe it would have been more appropriate to push this change with a major version bump rather than a minor one, because despite the API technically not changing, the semantics were certainly changed by the introduction of this lazy initialization.

I'm also not entirely sure whether it's safe to close this issue now, because I find it hard to reason whether a situation like the snippet I originally described could be triggered in production usage. To be on the safe side, I'd say the PubSub implementation should still guard against it.

[grantwwu]

I'll take a look at this. I feel like the behavior in the original snippet you posted should be that promise should resolve immediately with no values.

[arendjr]

Thanks!

[TimSusa]

JFI: axelspringer/graphql-google-pubsub#16

[grantwwu]

@TimSusa I don't know what "JFI" means here or how that issue is relevant.

@arendjr I've been really busy at work, but at first glance, I think the issue is that next() doesn't do any checking to see if the iterator has been terminated. I'll try to get a PR out later today.

[TimSusa]

@grantwwu Sry for being quiet. I just wanted to point out and tag, that our issue could maybe have something to do with the issue meant here. If this is not the case in your opinion, just drop a comment. Thanks for help and Greets from Berlin!

[JeffML]

Update:
It seems setImmediate() isn't as immediate as the name suggests. By executing publish() through setImmediate, the publish event is delayed long enough for the subscriber to receive it via sub.next(). Otherwise, the publish sends the event before sub.next() is called.

---

This might be related:

I'm writing a Jest test and I have to call setImmediate to invoke pubsub.publish() before I call the async iterator from subscribe, otherwise next() never returns:

import { graphql, validateSchema, subscribe, parse, ExecutionResult } from 'graphql'
import { schema, client, pubsub } from './schema'   // pubsub => graphql-subscription instance

  test('subscriptions', async () => {
    const document = parse(`
      subscription {
        create 
      }
    `)

    const sub = <AsyncIterator<ExecutionResult>>await subscribe(schema, document);
    expect(sub.next).toBeDefined()

    /** why is setImmediate needed here? **/
    setImmediate(() => pubsub.publish('CREATE_ONE', { create: "FLUM!" }))

    /** if setImmediate isn't used, the following promise is never resolved and test will timeout
    const { value: { errors, data } } = await sub.next()

    expect(errors).toBeUndefined()
    expect(data).toBeDefined()
    expect(data.create).toBe('FLUM!')
  })

The need for setImmediate() is confusing, non-obvious and seems to rely on a side-effect. Awaiting form the publish() promise doesn't work, either.

More details here.