Regarding memoryleak in Subscription service.

[s-n-o-r-r-e]

Howdy!

Starting by thanking for an awesome package. :)

I would like to ask about a memory leak found while using an earlier version of this package. The library structure has gone through major changes so it has probably being addressed already but could still be good to be aware.

This leak was found in version 6.0.0, in file: src/Transports.Subscriptions.Abstractions/SubscriptionServer.cs

public SubscriptionServer(
    IMessageTransport transport,
    ISubscriptionManager subscriptions,
    IEnumerable<IOperationMessageListener> messageListeners,
    ILogger<SubscriptionServer> logger)
{
    _messageListeners = messageListeners;
    _logger = logger;
    Subscriptions = subscriptions;
    Transport = transport;
}

public IMessageTransport Transport { get; }

public ISubscriptionManager Subscriptions { get; }

public IReaderPipeline TransportReader { get; set; }

public IWriterPipeline TransportWriter { get; set; }

public async Task OnConnect()
{
    _logger.LogDebug("Connected...");
    LinkToTransportWriter();
    LinkToTransportReader();

    LogServerInformation();

    // when transport reader is completed it should propagate here
    await _handler.Completion.ConfigureAwait(false);

    // complete write buffer
    await TransportWriter.Complete().ConfigureAwait(false);
    await TransportWriter.Completion.ConfigureAwait(false);
}

public Task OnDisconnect() => Terminate();

public async Task Terminate()
{
    foreach (var subscription in Subscriptions)
        await Subscriptions.UnsubscribeAsync(subscription.Id).ConfigureAwait(false);

    // this should propagate to handler completion
    await TransportReader.Complete().ConfigureAwait(false);
}

private void LogServerInformation()
{
    // list listeners
    var builder = new StringBuilder();
    builder.AppendLine("Message listeners:");
    foreach (var listener in _messageListeners)
        builder.AppendLine(listener.GetType().FullName);

    _logger.LogDebug(builder.ToString());
}

private void LinkToTransportReader()
{
    _logger.LogDebug("Creating reader pipeline");
    TransportReader = Transport.Reader;
    _handler = new ActionBlock<OperationMessage>(HandleMessageAsync, new ExecutionDataflowBlockOptions
    {
        EnsureOrdered = true,
        BoundedCapacity = 1
    });

    TransportReader.LinkTo(_handler);
    _logger.LogDebug("Reader pipeline created");
}

private async Task HandleMessageAsync(OperationMessage message)
{
    _logger.LogDebug("Handling message: {id} of type: {type}", message.Id, message.Type);
    using (var context = await BuildMessageHandlingContext(message).ConfigureAwait(false))
    {
        await OnBeforeHandleAsync(context).ConfigureAwait(false);

        if (context.Terminated)
            return;

        await OnHandleAsync(context).ConfigureAwait(false);
        await OnAfterHandleAsync(context).ConfigureAwait(false);
    }
}

private async Task OnBeforeHandleAsync(MessageHandlingContext context)
{
    foreach (var listener in _messageListeners)
    {
        await listener.BeforeHandleAsync(context).ConfigureAwait(false);
    }
}

private Task<MessageHandlingContext> BuildMessageHandlingContext(OperationMessage message)
    => Task.FromResult(new MessageHandlingContext(this, message));

private async Task OnHandleAsync(MessageHandlingContext context)
{
    foreach (var listener in _messageListeners)
    {
        await listener.HandleAsync(context).ConfigureAwait(false);
    }
}

private async Task OnAfterHandleAsync(MessageHandlingContext context)
{
    foreach (var listener in _messageListeners)
    {
        await listener.AfterHandleAsync(context).ConfigureAwait(false);
    }
}

private void LinkToTransportWriter()
{
    _logger.LogDebug("Creating writer pipeline");
    TransportWriter = Transport.Writer;
    _logger.LogDebug("Writer pipeline created");
}

public virtual void Dispose() => (Subscriptions as IDisposable)?.Dispose();

}

As can be seen in the SubscriptionServer class the ActionBlock<OperationMessage>, _handler does not appear to call the Complete() method, which makes the block remain alive. Any new invocation of OnConnect() also creates a new _hander block by the LinkToTransportReader(). Indeed there are implementations for disposal of TransportReader and TransportWriter alas the _handler block does not seem to have any implementation for this. Multiple calls to OnConnect() has showed to accumulate memory usage most probably because of this.

[Shane32]

The subscription implementation was rewritten from scratch for version 7. You should not encounter any memory leaks with the current version. It offers many other improvements also - see https://github.com/graphql-dotnet/server/blob/master/docs/migration/migration7.md

When upgrading to v7, be sure to read the GraphQL.NET migration guides here:

• https://github.com/graphql-dotnet/graphql-dotnet?tab=readme-ov-file#upgrade-guides
• https://github.com/graphql-dotnet/server?tab=readme-ov-file#migration-from-older-version

While we are not pushing updates for v6, if you want to issue a PR against the v6 branch, feel free to do so, and I can review how easy/difficult it would be to release an update.