Fixed executor swap logic when using the proxy . (#8538)

Co-authored-by: Glen <[email protected]>

Author: michaelstaib

src/HotChocolate/AspNetCore/src/AspNetCore.Pipeline/HttpRequestExecutorProxy.cs

@@ -22,16 +22,10 @@ public async ValueTask<ExecutorSession> GetOrCreateSessionAsync(CancellationToke
         return executor.Features.GetRequired<ExecutorSession>();
     }
 
-    protected override void OnRequestExecutorUpdated(IRequestExecutor? executor)
+    protected override void OnConfigureRequestExecutor(IRequestExecutor newExecutor, IRequestExecutor? oldExecutor)
     {
-        if (executor is null)
-        {
-            _session = null;
-            return;
-        }
-
-        var session = new ExecutorSession(executor);
-        executor.Features.Set(session);
+        var session = new ExecutorSession(newExecutor);
+        newExecutor.Features.Set(session);
         _session = session;
     }
 

src/HotChocolate/AspNetCore/test/AspNetCore.Tests/EvictSchemaTests.cs

@@ -28,7 +28,9 @@ public async Task Evict_Default_Schema()
                 }));
 
         // act
+        await Task.Delay(1000, cts.Token);
         await server.GetAsync(new ClientQueryRequest { Query = "{ evict }" });
+        await Task.Delay(2000, cts.Token);
         newExecutorCreatedResetEvent.Wait(cts.Token);
 
         // assert
@@ -58,9 +60,11 @@ public async Task Evict_Named_Schema()
         }));
 
         // act
+        await Task.Delay(1000, cts.Token);
         await server.GetAsync(
             new ClientQueryRequest { Query = "{ evict }" },
             "/evict");
+        await Task.Delay(2000, cts.Token);
         newExecutorCreatedResetEvent.Wait(cts.Token);
 
         // assert

src/HotChocolate/Core/src/Execution.Abstractions/Execution/RequestExecutorProxy.cs

@@ -1,4 +1,6 @@
+using System.Threading.Channels;
 using HotChocolate.Features;
+using HotChocolate.Utilities;
 
 namespace HotChocolate.Execution;
 
@@ -14,6 +16,15 @@ public class RequestExecutorProxy : IRequestExecutor, IDisposable
     private readonly IRequestExecutorProvider _executorProvider;
     private readonly string _schemaName;
     private readonly IDisposable? _eventSubscription;
+    private readonly CancellationTokenSource _cts = new();
+    private readonly Channel<RequestExecutorEvent> _events =
+        Channel.CreateBounded<RequestExecutorEvent>(
+            new BoundedChannelOptions(1)
+            {
+                SingleReader = true,
+                SingleWriter = true,
+                FullMode = BoundedChannelFullMode.DropOldest
+            });
     private bool _disposed;
 
     /// <summary>
@@ -42,6 +53,7 @@ public RequestExecutorProxy(
 
         var observer = new RequestExecutorEventObserver(OnRequestExecutorEvent);
         _eventSubscription = executorEvents.Subscribe(observer);
+        OnUpdateRequestExecutorAsync(_cts.Token).FireAndForget();
     }
 
     /// <summary>
@@ -178,7 +190,7 @@ public async ValueTask<IRequestExecutor> GetExecutorAsync(
                     .ConfigureAwait(false);
 
                 CurrentExecutor = executor;
-                OnRequestExecutorUpdated(executor);
+                OnConfigureRequestExecutor(executor, null);
             }
             else
             {
@@ -193,7 +205,11 @@ public async ValueTask<IRequestExecutor> GetExecutorAsync(
         return executor;
     }
 
-    protected virtual void OnRequestExecutorUpdated(IRequestExecutor? executor)
+    protected virtual void OnConfigureRequestExecutor(IRequestExecutor newExecutor, IRequestExecutor? oldExecutor)
+    {
+    }
+
+    protected virtual void OnAfterRequestExecutorSwapped(IRequestExecutor newExecutor, IRequestExecutor oldExecutor)
     {
     }
 
@@ -204,33 +220,42 @@ private void OnRequestExecutorEvent(RequestExecutorEvent eventArgs)
             return;
         }
 
-        if (eventArgs.Type is RequestExecutorEventType.Evicted)
+        if (eventArgs.Type is RequestExecutorEventType.Created)
         {
-            _semaphore.Wait();
-
-            try
-            {
-                CurrentExecutor = null;
-                OnRequestExecutorUpdated(null);
-            }
-            finally
-            {
-                _semaphore.Release();
-            }
+            _events.Writer.TryWrite(eventArgs);
         }
-        else if (eventArgs.Type is RequestExecutorEventType.Created)
+    }
+
+    private async Task OnUpdateRequestExecutorAsync(CancellationToken ct)
+    {
+        await foreach (var eventArgs in _events.Reader.ReadAllAsync(ct).ConfigureAwait(false))
         {
-            _semaphore.Wait();
+            IRequestExecutor newExecutor;
+            IRequestExecutor oldExecutor;
+
+            await _semaphore.WaitAsync(ct);
 
             try
             {
-                CurrentExecutor = eventArgs.Executor;
-                OnRequestExecutorUpdated(eventArgs.Executor);
+                // events are only raised when there is an initial executor
+                // so its guaranteed that we have a CurrentExecutor at this point.
+                oldExecutor = CurrentExecutor!;
+                newExecutor = eventArgs.Executor;
+
+                OnConfigureRequestExecutor(newExecutor, oldExecutor);
+
+                // we only assign the executor after we have run configuration logic
+                // on the new instance.
+                CurrentExecutor = newExecutor;
             }
             finally
             {
                 _semaphore.Release();
             }
+
+            // after the swap of the executors we allow classes extending this proxy
+            // to run notification logic that does not run within the lock context.
+            OnAfterRequestExecutorSwapped(newExecutor, oldExecutor);
         }
     }
 
@@ -239,8 +264,17 @@ public void Dispose()
         if (!_disposed)
         {
             CurrentExecutor = null;
+            _cts.Cancel();
+            _cts.Dispose();
+            _events.Writer.TryComplete();
             _eventSubscription?.Dispose();
             _semaphore.Dispose();
+
+            while (_events.Reader.TryRead(out _))
+            {
+                // we drain the channel
+            }
+
             _disposed = true;
         }
     }

src/HotChocolate/Core/src/Execution.Abstractions/HotChocolate.Execution.Abstractions.csproj

@@ -23,6 +23,7 @@
     <ProjectReference Include="..\..\..\Language\src\Language.SyntaxTree\HotChocolate.Language.SyntaxTree.csproj" />
     <ProjectReference Include="..\..\..\Language\src\Language.Web\HotChocolate.Language.Web.csproj" />
     <ProjectReference Include="..\..\..\Primitives\src\Primitives\HotChocolate.Primitives.csproj" />
+    <ProjectReference Include="..\..\..\Utilities\src\Utilities.Tasks\HotChocolate.Utilities.Tasks.csproj" />
     <ProjectReference Include="..\Features\HotChocolate.Features.csproj" />
     <ProjectReference Include="..\Types.Abstractions\HotChocolate.Types.Abstractions.csproj" />
   </ItemGroup>

src/HotChocolate/Core/src/Execution/RequestExecutorManager.cs

@@ -226,7 +226,7 @@ private async Task UpdateRequestExecutorAsync(string schemaName, RegisteredExecu
         previousExecutor.TypeModuleChangeMonitor.Dispose();
 
         // This will hot swap the request executor.
-        await CreateRequestExecutorAsync(schemaName, false, CancellationToken.None)
+        await CreateRequestExecutorAsync(schemaName, isInitialCreation: false, CancellationToken.None)
             .ConfigureAwait(false);
 
         previousExecutor.DiagnosticEvents.ExecutorEvicted(schemaName, previousExecutor.Executor);

src/HotChocolate/Core/test/Execution.Tests/RequestExecutorProxyTests.cs

@@ -41,16 +41,14 @@ public async Task Ensure_Executor_Is_Correctly_Swapped_When_Evicted()
                 .Services
                 .BuildServiceProvider()
                 .GetRequiredService<RequestExecutorManager>();
-        var evicted = false;
         var updated = false;
 
         var proxy = new TestProxy(resolver, resolver, ISchemaDefinition.DefaultName);
-        proxy.ExecutorEvicted += () =>
+        proxy.ExecutorUpdated += () =>
         {
-            evicted = true;
+            updated = true;
             executorUpdatedResetEvent.Set();
         };
-        proxy.ExecutorUpdated += () => updated = true;
 
         // act
         var a = await proxy.GetExecutorAsync(CancellationToken.None);
@@ -60,7 +58,6 @@ public async Task Ensure_Executor_Is_Correctly_Swapped_When_Evicted()
 
         // assert
         Assert.NotSame(a, b);
-        Assert.True(evicted);
         Assert.True(updated);
     }
 
@@ -70,19 +67,13 @@ private class TestProxy(
         string schemaName)
         : RequestExecutorProxy(executorProvider, executorEvents, schemaName)
     {
-        public event Action? ExecutorEvicted;
         public event Action? ExecutorUpdated;
 
-        protected override void OnRequestExecutorUpdated(IRequestExecutor? executor)
+        protected override void OnAfterRequestExecutorSwapped(
+            IRequestExecutor newExecutor,
+            IRequestExecutor oldExecutor)
         {
-            if (executor is null)
-            {
-                ExecutorEvicted?.Invoke();
-            }
-            else
-            {
-                ExecutorUpdated?.Invoke();
-            }
+            ExecutorUpdated?.Invoke();
         }
     }
 }