OperationContext is not disposed in subsriptions (memory leak)

[akolpachev]

Looks like HotChocolate has serious memory leaks in subscriptions when InMemoryPubSub is used (and probably with other PubSub too). Subscription and all related stuffs like operation context, execution context, etc are leaked because of ExecutionContext subscribes to event TaskEnqueued in batch dispatcher, but never unsubscribes from it. Taking into account that static batch dispatcher NoopBatchDispatcher is used for subscriptions it causes leak. Look at the code below

            // subscribe will use the subscribe resolver to create a source stream that yields
            // the event messages from the underlying pub/sub-system.
            private async ValueTask<ISourceStream> SubscribeAsync()
            {
                OperationContext operationContext = _operationContextPool.Get();
                try
                {
                    // first we will create the root value which essentially
                    // is the subscription object. In some cases this object is null.
                    object? rootValue = RootValueResolver.Resolve(
                        _requestContext,
                        _requestContext.Services,
                        _subscriptionType,
                        ref _cachedRootValue);
                    // next we need to initialize our operation context so that we have access to
                    // variables services and other things.
                    // The subscribe resolver will use a noop dispatcher and all DataLoader are
                    // dispatched immediately.
                    operationContext.Initialize(
                        _requestContext,
                        _requestContext.Services,
                        NoopBatchDispatcher.Default,
                        _requestContext.Operation!,
                        _requestContext.Variables!,
                        rootValue,
                        _resolveQueryRootValue);
                    // next we need a result map so that we can store the subscribe temporarily
                    // while executing the subscribe pipeline.
                    ResultMap resultMap = operationContext.Result.RentResultMap(1);
                    ISelection rootSelection = _rootSelections.Selections[0];
                    // we create a temporary middleware context so that we can use the standard
                    // resolver pipeline.
                    var middlewareContext = new MiddlewareContext();
                    middlewareContext.Initialize(
                        operationContext,
                        rootSelection,
                        resultMap,
                        1,
                        rootValue,
                        Path.New(rootSelection.ResponseName),
                        ImmutableDictionary<string, object?>.Empty);
                    // it is important that we correctly coerce the arguments before
                    // invoking subscribe.
                    if (!rootSelection.Arguments.TryCoerceArguments(
                        middlewareContext.Variables,
                        middlewareContext.ReportError,
                        out IReadOnlyDictionary<NameString, ArgumentValue>? coercedArgs))
                    {
                        // the middleware context reports errors to the operation context,
                        // this means if we failed, we need to grab the coercion errors from there
                        // and just throw a GraphQLException.
                        throw new GraphQLException(operationContext.Result.Errors);
                    }
                    // if everything is fine with the arguments we still need to assign them.
                    middlewareContext.Arguments = coercedArgs;
                    // last but not least we can invoke the subscribe resolver which will subscribe
                    // to the underlying pub/sub-system yielding the source stream.
                    ISourceStream sourceStream =
                        await rootSelection.Field.SubscribeResolver!.Invoke(middlewareContext)
                            .ConfigureAwait(false);
                    if (operationContext.Result.Errors.Count > 0)
                    {
                        // again if we have any errors we will just throw them and not opening
                        // any subscription context.
                        throw new GraphQLException(operationContext.Result.Errors);
                    }
                    return sourceStream;
                }
                finally
                {
                    operationContext.Result.DropResult();
                    _operationContextPool.Return(operationContext);
                }
            }

Unsubscribe from event is happened only if operation context is returned to the pool, but it doesn't happen because of object doesnt meet pool return policy OperationContextPoolPolicy requirements (condition obj.Execution.TaskStats.IsCompleted is not true).