Eventing on server-side close of stream WITHOUT final msg.event complete

[brentmjohnson]

Really enjoying the work folks have put into this - so thank you!

Opening for discussion if we should be considering a server-side initiated close of the event stream without a final "complete" message an error reported back through the graphql-sse client. Currently this scenario leaves a dangling client.subscribe sink since the operation completes without a final next(), error(), or complete() callback to the sink.

If this makes sense, it seems like it could be handled in the finally block here: https://github.com/enisdenjo/graphql-sse/blob/master/src/client.ts#:~:text=Object.values(waiting).forEach((%7B%20proceed%20%7D)%20%3D%3E%20proceed())%3B

[enisdenjo]

Hmm, I am not fully clear on what you're trying to describe. Can you elaborate a bit more or provide an example?

All subscription sinks must end with a complete or error callback. If you're confident the operation really finishes but the client isn't notified accordingly - this is a bug and a repro would be greatly appreciated!

[brentmjohnson]

Sure thing - thanks for the quick response:

Environment:
server-env: graphql-sse served via node http2

client-env: node environment running graphql-sse createClient
fetchFn: got-fetch with http2 enabled
client implementation: following https://github.com/enisdenjo/graphql-sse#async-iterator pattern

Sample server schema.js:

import { GraphQLSchema, GraphQLObjectType, GraphQLString } from 'graphql';

/**
 * Construct a GraphQL schema and define the necessary resolvers.
 *
 * type Query {
 *   hello: String
 * }
 * type Subscription {
 *   greetings: String
 * }
 */
 export default new GraphQLSchema({
  query: new GraphQLObjectType({
    name: 'Query',
    fields: {
      hello: {
        type: GraphQLString,
        resolve: () => 'world',
      },
    },
  }),
  subscription: new GraphQLObjectType({
    name: 'Subscription',
    fields: {
      greetings: {
        type: GraphQLString,
        subscribe: async function* () {
          const greetings = ['Hi', 'Bonjour', 'Hola', 'Ciao', 'Zdravo']
          for await (const hi of greetings) {
            console.log({ greetings: hi })
            yield { greetings: hi };
            greetings.push(hi);
            await new Promise((resolve) => { setTimeout(() => { resolve() }, 3000) });
          }
        },
      },
    },
  }),
});

Sample client implementation streamingPosts.server.ts (getStreamingPosts() returning an async iterable in this case):

import { createClient, SubscribePayload } from 'graphql-sse';
import http2wrapper from 'http2-wrapper';
import got from 'got';
const gotClient = got.extend({
    http2: true,
    request: http2wrapper.auto,
    https: {
      rejectUnauthorized: false,
    },
});

import { createFetch } from 'got-fetch';
const fetch = createFetch(gotClient);

const retry = async (retries) => {
  let retryDelay = 1000; // start with 1s delay
  for (let i = 0; i < retries; i++) {
    retryDelay *= 2;
  }
  await new Promise((resolve) =>
    setTimeout(
      resolve,
      retryDelay +
        // add random timeout from 300ms to 3s
        Math.floor(Math.random() * (3000 - 300) + 300),
    ),
  );
}

export async function* getStreamingPosts() {
  let subscription = undefined;
  try {
    const client = createClient({
      // singleConnection: true, preferred for HTTP/1 enabled servers. read more below
      url: "https://localhost:4000/graphql/stream",
      fetchFn: (url, init) => {
        init.signal.addEventListener('abort', () => {
          console.log('abort');
        }, { signal: init.signal });
        return fetch(url, init);
        },
      retryAttempts: -1,
      retry,
      onMessage: console.log,
    });

    function subscribe<T>(payload: SubscribePayload): AsyncGenerator<T> {
      // console.log(this.Request.body.on('close', () => {
      //   console.log('Connection closed');
      // }));
      let deferred: {
        resolve: (done: boolean) => void;
        reject: (err: unknown) => void;
      } | null = null;
      const pending: T[] = [];
      let throwMe: unknown = null,
        done = false;
      const dispose = client.subscribe<T>(payload, {
        next: (data) => {
          console.log(data);
          pending.push({ event: "next", ...data });
          deferred?.resolve(false);
        },
        error: (err) => {
          throwMe = err;
          deferred?.reject(throwMe);
        },
        complete: () => {
          done = true;
          pending.push({ event: "complete" });
          deferred?.resolve(true);
        },
      });
      return {
        [Symbol.asyncIterator]() {
          return this;
        },
        async next() {
          if (done) {
            return { done: true, value: undefined };
          }
          if (throwMe) throw throwMe;
          if (pending.length) return { value: pending.shift()! };
          return (await new Promise<boolean>(
            (resolve, reject) => (deferred = { resolve, reject }),
          ))
            ? { done: true, value: undefined }
            : { value: pending.shift()! };
        },
        async throw(err) {
          throw err;
        },
        async return() {
          dispose();
          return { done: true, value: undefined };
        }
      };
    }

    subscription = subscribe({
      query: 'subscription { greetings }',
    });

    for await (const result of subscription) {
      yield result
    }

    console.log("done");
  } catch (error) {
    console.error(error);
    subscription?.return && subscription.return()
  }
}

Running:

Repro steps:

1. set breakpoint in transpiled graphql-sse client.js in node-modules:
   
2. terminate graphql-sse server instance. breakpoint should be reached within client.js:
   
3. observe the node res.body stream closes and for loop is exited without handling the unexpected close event with a 'complete' msg.event or exception
4. this means the sample client implementation never receives a callback to identify the sink as "closed":
   (breakpoints not reached)
   

Hopefully this is enough to follow the behavior. Let me know if you need additional info!

[enisdenjo]

Since you're shutting down the server, the complete callback won't be called since the subscription didn't complete. Instead, the error callback should be called, can you check whether that is the case by logging the sink.error callback (maybe the async iterator recipe is broken)?

I've added a test confirming the error callback should be called when the server shuts down. Maybe the problem is with got-fetch 🤔? Can you try the same with node-fetch instead?

[brentmjohnson]

I think the issue in that new test, was that the server was destroying the socket before the client started iterating over the res.body. This should be able to reproduce what I am seeing: https://github.com/brentmjohnson/graphql-sse/commit/143883758ff32a2a8996c46db42da39a77bffa73

[enisdenjo]

Wow @brentmjohnson, that's an awesome catch! Thank you for the extensive replication steps and helping out. Let's track the progress over at #22.