Subscription messages can silently fail to post back to client

[braudabaugh]

When generating subscription messages in rapid succession, some messages may never make it back to the client. I've debugged the issue to the Subscription.OnNext function calling _writer.Post here:

server/src/Transports.Subscriptions.Abstractions/Subscription.cs

Line 62 in a41d7c3

_writer.Post(new OperationMessage

The call to _writer.Post returns false when the messages are lost. A false return value is meant to indicate that the Post could not be completed without blocking, but the Subscription.OnNext function does not handle this case.

My solution is to use _write.SendAsync instead, as it will send the messages even if it must block. Here are my implementation of Subscription.OnNext:

        public void OnNext(ExecutionResult value)
        {
            _logger.LogDebug("Subscription: {subscriptionId} got data", Id);
            var task = Task.Run(() => _writer.SendAsync(new OperationMessage
            {
                Type = MessageType.GQL_DATA,
                Id = Id,
                Payload = value
            }));
            task.Wait();
        }

I am not sure if this is the best solution, but it is working for my use case.

[sungam3r]

Why to wrap SendAsync in Task.Run and then wait for it? Why not just use

public void OnNext(ExecutionResult value)
        {
            _logger.LogDebug("Subscription: {subscriptionId} got data", Id);
            _writer.SendAsync(new OperationMessage
            {
                Type = MessageType.GQL_DATA,
                Id = Id,
                Payload = value
            }).GetAwaiter().GetResult();
        }

Also note that SendAsync returns Task, but underlying ITargetBlock.SendAsync returns Task<bool> - https://docs.microsoft.com/en-us/dotnet/api/system.threading.tasks.dataflow.dataflowblock.sendasync?f1url=https%3A%2F%2Fmsdn.microsoft.com%2Fquery%2Fdev15.query%3FappId%3DDev15IDEF1%26l%3DEN-US%26k%3Dk(System.Threading.Tasks.Dataflow.DataflowBlock.SendAsync%60%601);k(DevLang-csharp)%26rd%3Dtrue&view=netcore-2.2

[braudabaugh]

Using the Task.Run technique avoids causing a possible deadlock by allocating a new thread to do the actual blocking. See this reference:

https://stackoverflow.com/questions/17248680/await-works-but-calling-task-result-hangs-deadlocks/32429753#32429753

Since the OnNext method did not currently cause the calling thread to block, I was hesitant to change that behavior. Since I'm not super familiar with the graphql-dotnet code base, I'll defer to others to determine if blocking the calling thread is okay or not.

[sungam3r]

Task.Run does not necessary allocate new thread. In any case you introduce blocking on calling thread which is not good.

[braudabaugh]

You're right. Unless Task.Wait was removed which would cause another set of problems.
Perhaps there is a solution lower down in the code.

[jinhong-]

Any updates regarding this issue? We are currently facing the same problem as well. Messages sent in quick successions causes missing messages sent to the client

[sungam3r]

I did not intend to study this problem in the near future. I'm not well prepared for the topic of subscriptions yet.

[braudabaugh]

I settled on the following change in my fork based on conversations with @sungam3r:

        public void OnNext(ExecutionResult value)
        {
            _logger.LogDebug("Subscription: {subscriptionId} got data", Id);
            var result = _writer.Post(new OperationMessage
            {
                Type = MessageType.GQL_DATA,
                Id = Id,
                Payload = value
            });
            if (!result)
            {
                _writer.SendAsync(new OperationMessage
                {
                    Type = MessageType.GQL_DATA,
                    Id = Id,
                    Payload = value
                }).GetAwaiter().GetResult();
            }
        }

It seems to be working, however some of the other open issues related to subscriptions, #255 for example, are preventing us from doing much with this functionality.

[jinhong-]

What do you think about this solution? I was thinking, instead of handling the message sending as a subscriber, why not use the reactive extensions to extend the observable pipeline instead. This way i was able to use the async APIs without using any synchronous task calls:

    public class Subscription
    {
        private readonly Action<Subscription> _completed;
        private readonly ILogger<Subscription> _logger;
        private readonly IWriterPipeline _writer;
        private IDisposable _unsubscribe;

        public Subscription(string id,
            OperationMessagePayload payload,
            SubscriptionExecutionResult result,
            IWriterPipeline writer,
            Action<Subscription> completed,
            ILogger<Subscription> logger)
        {
            _writer = writer;
            _completed = completed;
            _logger = logger;
            Id = id;
            OriginalPayload = payload;

            Subscribe(result);
        }

        public string Id { get; }

        public OperationMessagePayload OriginalPayload { get; }


        private async Task SendComplete()
        {
            _logger.LogDebug("Subscription: {subscriptionId} completing", Id);
            await _writer.SendAsync(new OperationMessage
            {
                Type = MessageType.GQL_COMPLETE,
                Id = Id
            });

            _completed?.Invoke(this);
            _unsubscribe?.Dispose();
        }

        private async Task SendData(ExecutionResult value)
        {
            _logger.LogDebug("Subscription: {subscriptionId} got data", Id);
            await _writer.SendAsync(new OperationMessage
            {
                Type = MessageType.GQL_DATA,
                Id = Id,
                Payload = value
            });
        }

        public Task UnsubscribeAsync()
        {
            _logger.LogDebug("Subscription: {subscriptionId} unsubscribing", Id);
            _unsubscribe.Dispose();
            return _writer.SendAsync(new OperationMessage
            {
                Type = MessageType.GQL_COMPLETE,
                Id = Id
            });
        }

        private void Subscribe(SubscriptionExecutionResult result)
        {
            var stream = result.Streams.Values.Single();
            _unsubscribe = stream.Synchronize()
                .Select(value => Observable.FromAsync(() => SendData(value)))
                .Merge(1)
                .Concat(Observable.FromAsync(SendComplete))
                .Subscribe();

            _logger.LogDebug("Subscription: {subscriptionId} subscribed", Id);
        }
    }

[jinhong-]

I raised PR #279 with the new code. Tested it in my project and it seems to work

[jinhong-]

@sungam3r any chance to get my PR merged into the main code for the next release?

[sungam3r]

Sorry but I'm not good enough yet at the topic of subscriptions, so I can’t say if this change will lead to problems. Maybe @joemcbride can help. Subscriptions is the next big topic that I have to figure out, now I'm doing other things.

[danmalcolm]

We are seeing the same issue in a project. We've tested with a build based on PR #279 which resolves the problem for us.

This is a good PR. I'd really like to see this merged in.

[sungam3r]

I'd really like to see this merged in too but I just can't afford to merge code that I don't understand. For now, the subscription mechanism is out of my attention.

[jinhong-]

Is there anything I can do to help to increase the confidence if having this PR merged?