fix(server): Shouldn't send a complete message if client sent it

Closes #403

src/__tests__/fixtures/simple.ts

@@ -75,6 +75,31 @@ export const schemaConfig: GraphQLSchemaConfig = {
           };
         },
       },
+      lateReturn: {
+        type: new GraphQLNonNull(GraphQLString),
+        subscribe() {
+          let completed = () => {
+            // noop
+          };
+          return {
+            [Symbol.asyncIterator]() {
+              return this;
+            },
+            async next() {
+              await new Promise<void>((resolve) => (completed = resolve));
+              return { done: true };
+            },
+            return() {
+              completed();
+
+              // resolve return in next tick so that the generator loop breaks first
+              return new Promise((resolve) =>
+                setTimeout(() => resolve({ done: true }), 0),
+              );
+            },
+          };
+        },
+      },
     },
   }),
 };

src/__tests__/server.ts

@@ -1704,6 +1704,42 @@ describe('Subscribe', () => {
       {},
     );
   });
+
+  it('should not send a complete message back if the client sent it', async () => {
+    const server = await startTServer();
+
+    const client = await createTClient(server.url);
+
+    client.ws.send(
+      stringifyMessage({
+        type: MessageType.ConnectionInit,
+      }),
+    );
+    await client.waitForMessage(); // MessageType.ConnectionAck
+
+    client.ws.send(
+      stringifyMessage({
+        id: '1',
+        type: MessageType.Subscribe,
+        payload: {
+          query: 'subscription { lateReturn }',
+        },
+      }),
+    );
+    await server.waitForOperation();
+
+    client.ws.send(
+      stringifyMessage({
+        id: '1',
+        type: MessageType.Complete,
+      }),
+    );
+    await server.waitForComplete();
+
+    await client.waitForMessage(() => {
+      fail("Shouldn't have received a message");
+    }, 20);
+  });
 });
 
 describe('Disconnect/close', () => {

src/server.ts

@@ -831,9 +831,9 @@ export function makeServer<
           }
           case MessageType.Complete: {
             const subscription = ctx.subscriptions[message.id];
+            delete ctx.subscriptions[message.id]; // deleting the subscription means no further activity should take place
             if (isAsyncGenerator(subscription))
               await subscription.return(undefined);
-            delete ctx.subscriptions[message.id]; // deleting the subscription means no further activity should take place
             return;
           }
           default: