fix: do not send Complete after Error for subscriptions (#667)

Author: endigma

.changeset/khaki-buses-trade.md

@@ -0,0 +1,26 @@
+---
+'graphql-ws': patch
+---
+
+Fix the server sending a `Complete` message after an `Error` message for subscriptions.
+
+Previously, when a subscription's async iterable threw an error, the server would send:
+
+```
+{"id":"1","type":"error","payload":[{"message":"..."}]}
+{"id":"1","type":"complete"}
+```
+
+Per the protocol spec:
+
+> **Error:** This message terminates the operation and no further messages will be sent.
+
+> **Complete (Server → Client):** If the server dispatched the `Error` message relative to the original `Subscribe` message, no `Complete` message will be emitted.
+
+The server now correctly sends only the `Error` message:
+
+```
+{"id":"1","type":"error","payload":[{"message":"..."}]}
+```
+
+Clients that correctly follow the spec should be unaffected, as they are expected to ignore messages for operations they consider already completed.

src/server.ts

@@ -880,6 +880,9 @@ export function makeServer<
                       ],
                       message,
                     );
+                    // error terminates the operation per spec — prevent
+                    // emit.complete from notifying the client
+                    delete ctx.subscriptions[id];
                   }
                 }
               } else {

tests/server.test.ts

@@ -2015,6 +2015,43 @@ describe.concurrent('Subscribe', () => {
       );
     });
   });
+
+  it('should not send complete after error when subscription iterable throws', async ({
+    expect,
+  }) => {
+    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 { throwing }',
+        },
+      }),
+    );
+    await server.waitForOperation();
+
+    await client.waitForMessage((msg) => {
+      expect(msg.data).toMatchInlineSnapshot(
+        `"{"id":"1","type":"error","payload":[{"message":"Subscribe Kaboom!"}]}"`,
+      );
+    });
+
+    // per the protocol spec, no complete message should follow an error
+    await client.waitForMessage((msg) => {
+      throw new Error(`Unexpected message after error: ${msg.data}`);
+    }, 30);
+  });
 });
 
 describe.concurrent('Disconnect/close', () => {