Duplicate subscribe/complete operations, a possible bug?

[constantind]

Hi, @enisdenjo

I'm writing this as a discussion as i'm not sure if this is a websocket bug ([email protected]) or [email protected] subscription bug, but in the following scenario where the client just sends double subscribe the server should return 4409 however it does spin happily two subscriptions on the same client since frames on the websocket can be sent simultaneously (code snippet below), then observe in Chrome or Firefox two websocket operations subscribe spin at the exact same millisecond.
There is also the scenario where a little bit of time takes in graphql executing the subscription (say resolver parsing something or reading data from file, that takes more than 10ms) between the subscribe operation completes and returns the AsyncIterator, complete operation is sent in the middle and the subscription is null (utils.isAsyncIterable=false) inside of onComplete on the server, if complete is sent before the subscribe finishes, that can be reproduced in below example instead of double subscribe send subscribe followed by immediate complete.
How would the client solve it, by using retryWait ?
How would the server prevent this, by using a mutex with Atomics ?

` this.reloadGraphiql= () => {

            let customError = {};

            const asyncUrl = this.urlInputValue();

            const wsClient = graphqlWs.createClient({
              url: `${asyncUrl}/subscriptions`,
              lazy: true,
              keepAlive: 10000,
              on: {
                connected: () => {
                  customError.message = "";
                },
                closed: (event) => {
                  customError.message = `Socket closed with event ${ event.code } ${ event.reason }`;
                  if (dispose) {
                    dispose();
                  }
                }
              }
            });

            let dispose;

            function subscribe(payload) {
              let deferred = null;
              const pending = [];
              let throwMe = null,
                  done = false;

              dispose = wsClient.subscribe(payload, {
                next: (data) => {
                  pending.push(data);
                  deferred?.resolve(false);
                },
                error: (err) => {
                  if (err instanceof Error) {
                    throwMe = err;
                  } else if (err instanceof CloseEvent) {
                    throwMe = new Error(
                      `Socket closed with event ${err.code} ${
                        err.reason || ""
                      }`.trim()
                    );
                  } else {
                    // GraphQLError[]
                    throwMe = new Error(err.map(({ message }) => message).join(", "));
                  }
                  deferred?.reject({
                    message: throwMe.message
                  });
                },
                complete: () => {
                  done = true;
                  if (customError && customError.message) {
                    deferred?.reject(customError);
                  } else {
                    deferred?.resolve(true);
                  }
                },
              });
              
			  //sent double subscribe
			  dispose = wsClient.subscribe(payload, {
                next: (data) => {
                  pending.push(data);
                  deferred?.resolve(false);
                },
                error: (err) => {
                  if (err instanceof Error) {
                    throwMe = err;
                  } else if (err instanceof CloseEvent) {
                    throwMe = new Error(
                      `Socket closed with event ${err.code} ${
                        err.reason || ""
                      }`.trim()
                    );
                  } else {
                    // GraphQLError[]
                    throwMe = new Error(err.map(({ message }) => message).join(", "));
                  }
                  deferred?.reject({
                    message: throwMe.message
                  });
                },
                complete: () => {
                  done = true;
                  if (customError && customError.message) {
                    deferred?.reject(customError);
                  } else {
                    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(
                    (resolve, reject) => (deferred = { resolve, reject })
                  ))
                    ? { done: true, value: undefined }
                    : { value: pending.shift() };
                },
                async return() {
                  dispose();
                  return { done: true, value: undefined };
                },
              };
            }

`

[enisdenjo]

Hi there, thanks for writing!

Are you using graphql-ws also on the backend? Do the subscriptions have the same ID? Can you share the WS frames/messages from the DevTools?

[constantind]

In this case you see only one next because i have added mutex to block the other one with 4409, but otherwise both of them run on the server.

[enisdenjo]

Only if the subscription ID is the same between subscriptions would there be a problem. Here's what the spec says:

If there is already an active subscriber for an operation matching the provided ID, regardless of the operation type, the server must close the socket immediately with the event 4409: Subscriber for <unique-operation-id> already exists.

Subscribe message section in GraphQL over WebSocket Protocol

There is also the scenario where a little bit of time takes in graphql executing the subscription (say resolver parsing something or reading data from file, that takes more than 10ms) between the subscribe operation completes and returns the AsyncIterator, complete operation is sent in the middle and the subscription is null (utils.isAsyncIterable=false) inside of onComplete on the server

I don't quite understand, what do you mean by "subscription is null (utils.isAsyncIterable=false)"? Why is this a problem?

If the client immediately sends a complete event, then the server should not send any messages about that subscription upon receiving the complete message - is this not the case?

I would be very grateful if you could elaborate a bit more, or ideally create a repro with failing tests where I can fidget around.

[constantind]

Yes, correct in the first case however because the id is optional the client may not be a browser or obey the spec or sent an id in which case, the server still needs to handle the scenario in our case, so my question in the first case is if the handling with a mutex makes sense to fail one with 4409 and allow the other to proceed. The ask of this is because i have seen clients sending duplicates or in 10ms next subscribe with same id.
the NodeJS 14.20.1 or latest 14.x appears to run this code at the same time https://github.com/enisdenjo/graphql-ws/blob/master/src/server.ts#L659 and there is no id so both continue, similarly with this pubsub with apollo express: https://github.com/davidyaha/graphql-redis-subscriptions/blob/master/src/redis-pubsub.ts#L97 the increment ++ does not work without a mutex as both turn into 1 are not found in the map and return the subscriptions index 1, so both spin, creating similar issue to apollographql/subscriptions-transport-ws#433

In the second case let me explain on the server very similar to this apollographql/subscriptions-transport-ws#645:
onSubscribe takes 300ms |--------------| including resolver after which the resolver returns asyncIterator.return() as from subscribe: https://github.com/apollographql/graphql-subscriptions/blob/master/src/pubsub-engine.ts#L5 and the context on complete have subscriptions
onComplete is sent here |-------| before the subscribe finishes in this case the Object.keys(context.subscriptions).length is 0
In this edge case on complete cannot be used to stop the pending subscribe or free resources, it can only trigger "unsubscribe (https://github.com/apollographql/graphql-subscriptions/blob/master/src/pubsub-engine.ts#L6)" and stop it" otherwise the socket on close only will trigger the async iterator event to shutdown when the connection drops and that is guaranteed as long as its emitter triggers. It feels more like subscribe needs ack before it can accept complete, unless i'm missing something, however if the asyncIterator does not call return() and cancels out it will never trigger the "unsubscribe", so i end up with calling "unsubscribe" onComplete as well as from the the pubsub iterator which ever wins or just have it rolling complete operation which does nothing and subscribe operation does actually "complete(and cleanup previous) followed by subscribe (allocate next)".
I will see if i can build something reproducible and let you know.

[enisdenjo]

Sorry for the late reply, am quite busy.

the id is optional the client may not be a browser or obey the spec or sent an id in which case

The id is not optional. Please read Communication section in the protocol again.

if the handling with a mutex makes sense to fail one with 4409 and allow the other to proceed

Yes, only one should succeed.

The ask of this is because i have seen clients sending duplicates or in 10ms next subscribe with same id.

graphql-ws clients? This would be a serious bug, if so - can you point me to where I can see this behaviour?

the NodeJS 14.20.1 or latest 14.x appears to run this code at the same time

NodeJS is not concurrent, it's not possible that that code part ran in the same time - because of this, the check will always be correct.

About the rest, I am a bit lost to be honest 😬. If you're able to send me a repro, I'd be more than happy to look at it? Thank you!

[constantind]

Hope all is well.
Here is the bad client: bad-client.zip
To reproduce the server is based on apollo express 3.x by the example you provided and you can choose in memory or redis pubsub does not matter for example with package json:
"dependencies": {
"@graphql-tools/schema": "8.0.3",
"apollo-server-core": "3.11.1",
"apollo-server-plugin-base": "3.11.1",
"apollo-server-types": "3.11.1",
"cors": "2.8.5",
"express": "4.17.2",
"graphql": "15.5.0",
"graphql-subscriptions": "1.2.1",
"graphql-ws": "5.11.2",
"hpp": "0.2.3",
"winston": "3.4.0",
"ws": "7.5.9"
}
with schema resolver something like this:

`
export const resolvers = {
Subscription: {
newEvent: {
subscribe: async (_parent: any, args: NewEventArguments, context: ConnectionContext) => newEventResolver(args, context)
},
}
};

async function newEventResolver(args: NewEventArguments, context: ConnectionContext) {
//note:currently only one subscription per consumer allowed
let streamObj;
try{
//since chrome is able to send WS at same time below just runs twice from a bad client
//local pod lock, multiple pods would need to use external lock
GlobalMutex.lock();
streamObj = pubSub.find( stream => stream.name === streamName);
} finally {
GlobalMutex.unlock();
}
if (!streamObj) {
const subscriptionObj = {
name: streamName,
index: 1,
iterator: pubSub.consumer.asyncIterator(streamName)
}
pubSub.streams.push(subscriptionObj);
try {
subscriptionObj.index = await pubSub.consumer.subscribe(streamName, onMessageReceived, {
streamPool: context.name,
offset: args.input.offset,
delta: args.input.delta || false
});
} catch(err) {
//unlock the local lock
try {
GlobalMutex.lock();
pubSub.streams = pubSub.streams.filter(stream => stream.name !== streamName);
} finally {
GlobalMutex.unlock();
}
throw err;
}
return subscriptionObj.iterator;
} else {
throw new GraphQLError(ALREADY_SUBSCRIBED);
}
}
`

[enisdenjo]

I checked the bad client you attached with #422 (reply in thread). The reason why you have 2 subscriptions issued is because you actually do 2 subscribes... Just remove the second subscribe and there wont be any double subscription requests.

function subscribe(payload) {
  let deferred = null;
  const pending = [];
  let throwMe = null,
    done = false;

  dispose = wsClient.subscribe(payload, {
    next: (data) => {
      pending.push(data);
      deferred?.resolve(false);
    },
    error: (err) => {
      if (err instanceof Error) {
        throwMe = err;
      } else if (err instanceof CloseEvent) {
        throwMe = new Error(
          `Socket closed with event ${err.code} ${
            err.reason || ""
          }`.trim()
        );
      } else {
        // GraphQLError[]
        throwMe = new Error(err.map(({ message }) => message).join(", "));
      }
      deferred?.reject({
        message: throwMe.message
      });
    },
    complete: () => {
      done = true;
      if (customError && customError.message) {
        deferred?.reject(customError);
      } else {
        deferred?.resolve(true);
      }
    },
  });
- // The next block sends double subscribe
-   dispose = wsClient.subscribe(payload, {
-     next: (data) => {
-       pending.push(data);
-       deferred?.resolve(false);
-     },
-     error: (err) => {
-       if (err instanceof Error) {
-         throwMe = err;
-       } else if (err instanceof CloseEvent) {
-         throwMe = new Error(
-           `Socket closed with event ${err.code} ${
-             err.reason || ""
-           }`.trim()
-         );
-       } else {
-         // GraphQLError[]
-         throwMe = new Error(err.map(({ message }) => message).join(", "));
-       }
-       deferred?.reject({
-         message: throwMe.message
-       });
-     },
-     complete: () => {
-       done = true;
-       if (customError && customError.message) {
-         deferred?.reject(customError);
-       } else {
-         deferred?.resolve(true);
-       }
-     },
-   });
- // ------------------end of double subscribe-------------------- //
  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(
        (resolve, reject) => (deferred = { resolve, reject })
      ))
        ? { done: true, value: undefined }
        : { value: pending.shift() };
    },
    async return() {
      dispose();
      return { done: true, value: undefined };
    },
  };
}

[enisdenjo]

Please note that Node is not concurrent, there is absolutely no need for overcomplicating and overoptimizing the server with mutexes or atomic locks alike.

This part will execute only once at a time guaranteeing that there'll always be just one subscriber for one ID.

graphql-ws/src/server.ts

Lines 659 to 667 in 2e6eb13

	if (id in ctx.subscriptions)
	return socket.close(
	CloseCode.SubscriberAlreadyExists,
	`Subscriber for ${id} already exists`,
	);
	// if this turns out to be a streaming operation, the subscription value
	// will change to an `AsyncIterable`, otherwise it will stay as is
	ctx.subscriptions[id] = null;

[constantind]

Right but different ids (also cleaned after close) per subscribe we use a "key" or "auth" to identify a client when sent in the subscribe message so both "bad clients" sending same key execute in this case for the same bad client code with two different ids, the code above will not close the socket since the id is not in the ctx, wasting resources on the server. imagine you wire redis pubsub (there will be multiple calls to that store) like you have commented out here: apollographql/subscriptions-transport-ws#433
Also when using multiple servers (containers) to serve the ctx of one is not available in the memory of the other therefore allowing the same client (for us same "key" with different ids) to connect the times the servers are. Or can try with two buttons start/stop in two browsers and clicking them very quickly, its just a a legitimate client trying to abuse the server in this case and so trying to prevent wasting resources on the server. if not mutually exclusive both will execute as node will just run this:
concurrently running below in node is about less than 1ms according to node console.log:
---first subscribe:
ctx.subscription = [];
"1" in ctx.subscriptions is false
--second subscribe:
ctx.subscription = [];
"2" in ctx.subscriotions is false
100 ms later both 1 and 2 are in the subscriptions list since both subscribe methods return their own iterator id
Since this id cannot be used to identify a client we use the "key" or "auth" in message which is the same in the case of the "bad client"

[enisdenjo]

Sorry, I am having a difficult time grasping what you're trying to explain.

Can you please create a failing test? Code can speak more than words.

[constantind]

Hi, @enisdenjo

Sorry didnt have time to write one, but here is another example query followed by subscription:

The lock could only help to serialize it but unfairly and creates more problems including deadlock. The way i solve this problem is the following, since clearly the query type subscribe sends complete and the subscription ones request the client to send complete, effectively makes each connection calls init -> subscribe -> complete, the client ids cannot be trusted, the protocol says regardless of the type but not regardless of operation AST. Which should it execute query or subscription when they work differently? Looks like apollo 3 executes both but returns to client only the first in this case response from the query, while it spins subscription but there is nothing to sent next, the server then sends complete from query and closes the connection leaving a ghost subscription running, as i looked in the code you either handle subscriptions or queries, as we dont use queries below fixes the issue for us.

onInit {
  context.extras.operationId = randomUUID()
  operationLock.clear()
}
onSubscribe {
   //...
   if (context.extra.operationId === operationLock.get(message.type)) {
    return [new GraphQLError(ALREADY_SUBSCRIBED_DUPLICATE_INPROGRESS)];
  }

  operationLock.set(message.type, context.extra.operationId?context.extra.operationId:"");
  //...
}

[enisdenjo]

Please write a failing test if you're still having issues. This thread is getting too convoluted and I am having a hard time understanding you.