fix(server): subscription operations are distinct on the message ID

PROTOCOL.md

@@ -73,7 +73,9 @@ The client is now **ready** to request subscription operations.
 
 Direction: **Client -> Server**
 
-Requests a operation specified in the message `payload`. This message leverages the unique ID field to connect future server messages to the operation started by this message.
+Requests a operation specified in the message `payload`. This message provides a unique ID field to connect future server messages to the operation started by this message.
+
+If there is already an active subscriber for a `subscription` operation matching the provided ID, the server will close the socket immediately with the event `4409: Subscriber for <unique-operation-id> already exists`. Since `query` and `mutation` resolve to a single emitted value, their subscription does not require reservations for additional future events. Having this in mind, the server may not assert this rule for these operations.
 
 ```typescript
 import { DocumentNode } from 'graphql';

src/server.ts

@@ -461,6 +461,13 @@ export function createServer(
             if (operationAST.operation === 'subscription') {
               const subscriptionOrResult = await subscribe(execArgs);
               if (isAsyncIterable(subscriptionOrResult)) {
+                // iterable subscriptions are distinct on ID
+                if (ctx.subscriptions[message.id]) {
+                  return ctx.socket.close(
+                    4409,
+                    `Subscriber for ${message.id} already exists`,
+                  );
+                }
                 ctx.subscriptions[message.id] = subscriptionOrResult;
 
                 try {

src/tests/server.ts

@@ -819,6 +819,64 @@ describe('Subscribe', () => {
 
     expect(onMessageFn).toBeCalledTimes(3); // ack, next, complete
   });
+
+  it('should close the socket on duplicate `subscription` operation subscriptions request', async () => {
+    expect.assertions(3);
+
+    await makeServer();
+
+    const client = new WebSocket(url, GRAPHQL_TRANSPORT_WS_PROTOCOL);
+    client.onopen = () => {
+      client.send(
+        stringifyMessage<MessageType.ConnectionInit>({
+          type: MessageType.ConnectionInit,
+        }),
+      );
+    };
+
+    client.onmessage = ({ data }) => {
+      const message = parseMessage(data);
+      if (message.type === MessageType.ConnectionAck) {
+        client.send(
+          stringifyMessage<MessageType.Subscribe>({
+            id: 'not-unique',
+            type: MessageType.Subscribe,
+            payload: {
+              query: `subscription {
+              boughtBananas {
+                name
+              }
+            }`,
+            },
+          }),
+        );
+
+        // try subscribing with a live subscription id
+        setTimeout(() => {
+          client.send(
+            stringifyMessage<MessageType.Subscribe>({
+              id: 'not-unique',
+              type: MessageType.Subscribe,
+              payload: {
+                query: `subscription {
+                  greetings
+                }`,
+              },
+            }),
+          );
+        }, 10);
+      }
+    };
+
+    await new Promise((resolve) => {
+      client.onclose = (event) => {
+        expect(event.code).toBe(4409);
+        expect(event.reason).toBe('Subscriber for not-unique already exists');
+        expect(event.wasClean).toBeTruthy();
+        resolve(); // done
+      };
+    });
+  });
 });
 
 describe('Keep-Alive', () => {