2 subscribes on different subscription result in duplicated event

[JonathanDuvalV]

Hi,

First of all, thanks for this awesome library, it's been quite helpful so far! I just wanted to point out one issue I came across recently while using this package using the version 2.0.0 and 2.1.0. When I create to Subscribe as such:

public static void main ()
{
    startEntity1Subscription();
    startEntity2Subscription();
}

public void startEntity1Subscription()
        {

            string queryMessage = @"
                entity1 (serverId:""1234"", mutationType: CREATED) {
                    node {
                        id,
                        commandText
                    }
                }";
            StringBuilder requestBuilder = new StringBuilder("");

            requestBuilder.Append("subscription");
            requestBuilder.Append("{" + queryMessage + "}");

            var testRequest = new GraphQL.GraphQLRequest
            {
                Query = requestBuilder.ToString()
            };

            var stream = client.CreateSubscriptionStream<CommandSubscritionResult>(testRequest);

            stream.Subscribe(response => {
                // this is where the subscribe get two request through
                CommandResponseHandling(response, server);
            });
        }

public void startEntity2Subscription()
{
    string queryMessage = @"
                entity2 (mutationType: CREATED) {
                    node {
                        id,
                        name,
                        slug
                    }
                }";
            StringBuilder requestBuilder = new StringBuilder("");

            requestBuilder.Append("subscription");
            requestBuilder.Append("{" + queryMessage + "}");

            var testRequest = new GraphQL.GraphQLRequest
            {
                Query = requestBuilder.ToString()
            };

            var stream = client.CreateSubscriptionStream<ServerCreationResult>(testRequest);
            stream.Subscribe(response => {
                // this is where the subscribe get two request through
                ServerCreationResHandling(response, agent);
            });
        }

So as the code stated above, the subscriptions receives duplicated events when I call. However, if I comment out on of the subscription method call (in main), events will be rightfully received once. Basically if we send ONE mutation for entity1, we receive 2 responses for the same mutation, but only if there'S two subscribe done.

We checked our endpoint and the mutation is only getting through once. We tried it on playground and it was also working right. I think this could be due to something behind the scene on the websocket handler.

Thanks, I don't know if this is clear for you

[rose-a]

Hi, I've never experienced something like this using that library.

What type of server are u using? How did you try to test this using Playground? Does Playground use a single websocket for multiple subscriptions (I doubt it...)?

There already is a test in place for this library which proves the simultaneous creation of 2 subscriptions on a single websocket connection, but I think its not yet testing for excess payloads and this kind of interference. I will add that in the near future.

Maybe the server implementation you're using does not expect multiple independent subscriptions on a single websocket connection?

[MysticFragilist]

Hi, I was the guy that wrote this message, it was from a colleague account. As of our technology stack I think it's best if I let @Willbill360 describe what our graphql stack is as he is the official maintainer of our API. I think we uses graphql-yoga from prisma. I don't have any more details on implementation. @Willbill360 could you clarify?

[Willbill360]

The API is fairly simple. We've based it upon the boilerplate generated by the graphql-cli. It's a very basic graphql-yoga setup with prisma behind it. Not that I think it matters, but we chose to go with TypeScript instead of basic JavaScript if that helps in any way.

The subscription are created using prisma-binding. Here is an exemple:

command: {
    subscribe: (
      _parent,
      args: {
        serverId: string
        mutationType: 'CREATED' | 'UPDATED' | 'DELETED' | MutationType[]
      },
      ctx: Context,
      info
    ) => {
      return ctx.db.subscription.command(
        {
          where: {
            node: {
              server: { id: args.serverId }
            },
            mutation_in: args.mutationType
          }
        },
        info
      )
    }
  }

As mentioned before, subscribing to this exact subscription (and a second, very similar one) through Playground works as expected. For reference, this is the same subscription that startEntity1Subscription() is subscribing to.

I have no idea how Playground handles listening to websockets, but it seems like there is only one connection made, yet it understand clearly that one is listening to some event and the other one is listening to another event.

Here is an example working in Playground: As expected, the "command" subscription receives an entry since the command was updated and the "gameServer" subscription is still hanging since it didn't receive anything.

Does that helps? I could provide you with an account on our development server if you'd like to try it for yourself.

[rose-a]

Open the devtools of your browser and switch to the network analyzer before starting the subscriptions...

If you see a new websocket connection opening up on each subscription then those are separated and I'd suspect that this would be the cause for the error not showing on Playground...

[Willbill360]

Yes, in fact, it seems to create two connections...

Should we try to replicate that in the C# application?

[rose-a]

To replicate that you would have to use two instances of GraphQLHttpClient...

[Willbill360]

Alright, we will try to implement that, thank you very much!

[rose-a]

@Willbill360 IMO this is a bad workaround...

I added the "crossinterference" test locally, and it shows no issues. I really think this is an issue with your server (graphql-yoga) which handles multiple subscriptions on a single websocket badly...

Since other libraries (like apollo-graphql) also support multiple subscriptions over a single websocket this should be supported by any GraphQL server. (see also https://github.com/apollographql/subscriptions-transport-ws#full-websocket-transport)

[Willbill360]

The link provided at that specific position is for the client, the server is just above. We were going to change to Apollo Server in the futur, so I'll just bump the upgrade to now. Thank you, I'll advice when the changes are done we tested the new solution.

[JonathanDuvalV]

We fixed the bug!

The problem is not with the server, we are still using graphql-yoga. Actually we found out that graphql-yoga is based on apollo-server, the setup is basically the same.

Here is how we fixed it:

client = new GraphQLHttpClient(o =>
            {
o.EndPoint = new Uri("https://develop.api.odeyr.org");
o.UseWebSocketForQueriesAndMutations = false;
o.OnWebsocketConnected = OnWebSocketConnect;
o.JsonSerializer = new NewtonsoftJsonSerializer();
});
client.InitializeWebsocketConnection().Wait();

Adding client.InitializeWebsocketConnection().Wait(): seems to have fixed our issue. The code posted above in the first message of the thread hasn't changed since.

[rose-a]

Thanks for your investigations! Care to post a complete example? The creation of the GraphQLHttpClient was missing from your fist post.

Calling client.InitializeWebsocketConnection() shouldn't be necessary since that's automatically invoked (with thread-safety in mind) when creating a subscription.

Thus I'd really like to try that for myself....

[JonathanDuvalV]

Here you go! I have removed some of the unnecessary code.

    class ApiConnect
    {
        GraphQLHttpClient client;
        ILogger<OdeyrWorker> logger;

        public ApiConnect(ILogger<OdeyrWorker> logger)
        {
            this.logger = logger;

            client = new GraphQLHttpClient(o =>
            {
                o.EndPoint = new Uri("https://develop.api.odeyr.org");
                o.UseWebSocketForQueriesAndMutations = false;
                o.OnWebsocketConnected = OnWebSocketConnect;
                o.JsonSerializer = new NewtonsoftJsonSerializer();
            });

            client.InitializeWebsocketConnection().Wait();

        }

        public Task OnWebSocketConnect(GraphQLHttpClient client)
        {
            logger.LogInformation("Main websocket is open");

            return Task.CompletedTask;
        }

        public async Task<string> sendLoginMutationAsync(string email, string password)
        {
            string queryMessage = "login(email: \"" + email + "\", password: \"" + password + "\") { id }";
            StringBuilder requestBuilder = new StringBuilder("");

            requestBuilder.Append("mutation");
            requestBuilder.Append("{" + queryMessage + "}");

            var testRequest = new GraphQL.GraphQLRequest
            {
                Query = requestBuilder.ToString()
            };

            var response = await client.SendQueryAsync<LoginMutationResult>(testRequest);

            if (response.Errors == null)
            {
                var data = response.Data.Login.Id;
                logger.LogInformation("Id data : " + data);
                return data;
            }
            else
            {
                var data = response.Errors;

                foreach (GraphQLError error in response.Errors)
                {
                    logger.LogError(error.Message);
                }
                return null;
            }
        }

        public void startServerCreationSubscription(SetupAgent agent)
        {
            string queryMessage = @"
                gameServer (mutationType: CREATED) {
                    node {
                        id,
                        name,
                        slug
                    }
                }";
            StringBuilder requestBuilder = new StringBuilder("");

            requestBuilder.Append("subscription");
            requestBuilder.Append("{" + queryMessage + "}");

            var testRequest = new GraphQL.GraphQLRequest
            {
                Query = requestBuilder.ToString()
            };

            var stream = client.CreateSubscriptionStream<ServerCreationResult>(testRequest);
            stream.Subscribe(response => ServerCreationResHandling(response, agent));
        }

        public void startCommandSubscription(Server server)
        {

            string queryMessage = @"
                command (serverId:""" + server.CUID + @""", mutationType: CREATED) {
                    node {
                        id,
                        commandText
                    }
                }";
            StringBuilder requestBuilder = new StringBuilder("");

            requestBuilder.Append("subscription");
            requestBuilder.Append("{" + queryMessage + "}");

            var testRequest = new GraphQL.GraphQLRequest
            {
                Query = requestBuilder.ToString()
            };

            var stream = client.CreateSubscriptionStream<CommandSubscritionResult>(testRequest);

            stream.Subscribe(response =>
            {
                CommandResponseHandling(response, server);
            });
        }

        private void CommandResponseHandling(GraphQLResponse<CommandSubscritionResult> response, Server server)
        {
            var cmd = response.Data.Command.Node.CommandText;
            var cmdId = response.Data.Command.Node.Id;
            logger.LogInformation(cmd);
            switch (cmd)
            {
                case "start":
                    server.Start(cmdId);
                    break;
                case "stop":
                    server.Stop(cmdId);
                    break;
                default:
                    server.SendSpecificCommand(cmd, cmdId);
                    break;
            }
        }

        private void ServerCreationResHandling(GraphQLResponse<ServerCreationResult> response, SetupAgent agent)
        {
            var data = response.Data.GameServer.Node;
            Task.Run(() => agent.CreateNewServerAsync(data.Id, data.Slug, data.Name, ServerType.Vanilla, new Version(1, 15, 2)));

        }
    }  

And here is where we instantiate our API and start the subscriptions

private async Task Initialization()
        {
            logger.LogInformation("Started Odeyr Service");
            api = new ApiConnect(logger);
            storage = new DataStorage(logger);

            //Login to api
            string userId = await api.sendLoginMutationAsync(storage.userLogin.Email, 
            storage.userLogin.Password);
            storage.userLogin.Id = userId;

            storage.Servers.ToList().ForEach(x =>
            {
                api.startCommandSubscription(x.Value);
                x.Value.CommandSaved += api.OnCommandSaved;
            });

            api.startServerCreationSubscription(setupAgent);
        }

[rose-a]

@JonathanDuvalV does this issue still persist with the latest release? There have been some changes regarding threading with subscriptions...

[MysticFragilist]

Hi! I'll be looking into that this weekend, it's been a while since we finished this project so I'll need to update to the new graphql-client version, and reinstall it on my personal computer. Do you have a specific version that you want me to try it with?

[rose-a]

use the latest, currently 3.1.9