[Fusion] Properly update RequestExecutorProxy (#8813)

tobias-tengler · web-flow · commit 5a868356f04c · 2025-10-15T14:17:29.000+02:00
diff --git a/src/HotChocolate/Core/src/Execution/RequestExecutorManager.cs b/src/HotChocolate/Core/src/Execution/RequestExecutorManager.cs
@@ -197,10 +197,7 @@ await OnRequestExecutorCreatedAsync(context, executor, setup, cancellationToken)
             registeredExecutor.Executor);
 
         _events.RaiseEvent(
-            new RequestExecutorEvent(
-                RequestExecutorEventType.Created,
-                schemaName,
-                registeredExecutor.Executor));
+            RequestExecutorEvent.Created(registeredExecutor.Executor));
 
         return registeredExecutor;
     }
@@ -219,11 +216,7 @@ await CreateRequestExecutorAsync(schemaName, isInitialCreation: false, Cancellat
 
         try
         {
-            _events.RaiseEvent(
-                new RequestExecutorEvent(
-                    RequestExecutorEventType.Evicted,
-                    schemaName,
-                    previousExecutor.Executor));
+            _events.RaiseEvent(RequestExecutorEvent.Evicted(previousExecutor.Executor));
         }
         finally
         {
diff --git a/src/HotChocolate/Core/test/Execution.Tests/RequestExecutorProxyTests.cs b/src/HotChocolate/Core/test/Execution.Tests/RequestExecutorProxyTests.cs
@@ -9,7 +9,7 @@ public class RequestExecutorProxyTests
     public async Task Ensure_Executor_Is_Cached()
     {
         // arrange
-        var resolver =
+        var manager =
             new ServiceCollection()
                 .AddGraphQL()
                 .AddStarWarsRepositories()
@@ -19,7 +19,7 @@ public async Task Ensure_Executor_Is_Cached()
                 .GetRequiredService<RequestExecutorManager>();
 
         // act
-        var proxy = new RequestExecutorProxy(resolver, resolver, ISchemaDefinition.DefaultName);
+        var proxy = new RequestExecutorProxy(manager, manager, ISchemaDefinition.DefaultName);
         var a = await proxy.GetExecutorAsync(CancellationToken.None);
         var b = await proxy.GetExecutorAsync(CancellationToken.None);
 
@@ -33,7 +33,7 @@ public async Task Ensure_Executor_Is_Correctly_Swapped_When_Evicted()
         // arrange
         var executorUpdatedResetEvent = new ManualResetEventSlim(false);
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
-        var resolver =
+        var manager =
             new ServiceCollection()
                 .AddGraphQL()
                 .AddStarWarsRepositories()
@@ -43,7 +43,7 @@ public async Task Ensure_Executor_Is_Correctly_Swapped_When_Evicted()
                 .GetRequiredService<RequestExecutorManager>();
         var updated = false;
 
-        var proxy = new TestProxy(resolver, resolver, ISchemaDefinition.DefaultName);
+        var proxy = new TestProxy(manager, manager, ISchemaDefinition.DefaultName);
         proxy.ExecutorUpdated += () =>
         {
             updated = true;
@@ -52,7 +52,7 @@ public async Task Ensure_Executor_Is_Correctly_Swapped_When_Evicted()
 
         // act
         var a = await proxy.GetExecutorAsync(CancellationToken.None);
-        resolver.EvictExecutor();
+        manager.EvictExecutor();
         executorUpdatedResetEvent.Wait(cts.Token);
         var b = await proxy.GetExecutorAsync(CancellationToken.None);
 
diff --git a/src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/FusionRequestExecutorManager.cs b/src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/FusionRequestExecutorManager.cs
@@ -50,6 +50,7 @@ internal sealed class FusionRequestExecutorManager
     private ImmutableArray<ObserverSession> _observers = [];
 
     private bool _disposed;
+    private ulong _version;
 
     public FusionRequestExecutorManager(
         IOptionsMonitor<FusionGatewaySetup> optionsMonitor,
@@ -107,7 +108,6 @@ private async ValueTask<IRequestExecutor> GetOrCreateRequestExecutorAsync(
 
             registration = await CreateInitialRegistrationAsync(schemaName, cancellationToken).ConfigureAwait(false);
             _registry.TryAdd(schemaName, registration);
-            await _executorEvents.WriteCreatedAsync(registration.Executor, cancellationToken).ConfigureAwait(false);
             return registration.Executor;
         }
         finally
@@ -121,9 +121,14 @@ private SemaphoreSlim GetSemaphoreForSchema(string schemaName)
 
     private async ValueTask EvictExecutorAsync(FusionRequestExecutor executor, CancellationToken cancellationToken)
     {
-        await _executorEvents.WriteEvictedAsync(executor, cancellationToken);
-
-        EvictRequestExecutorAsync(executor).FireAndForget();
+        try
+        {
+            await _executorEvents.WriteEvictedAsync(executor, cancellationToken);
+        }
+        finally
+        {
+            EvictRequestExecutorAsync(executor).FireAndForget();
+        }
     }
 
     private static async Task EvictRequestExecutorAsync(FusionRequestExecutor previousExecutor)
@@ -150,6 +155,8 @@ private async ValueTask<RequestExecutorRegistration> CreateInitialRegistrationAs
 
         await WarmupExecutorAsync(executor, true, cancellationToken).ConfigureAwait(false);
 
+        await _executorEvents.WriteCreatedAsync(executor, cancellationToken).ConfigureAwait(false);
+
         return new RequestExecutorRegistration(
             this,
             documentProvider,
@@ -161,6 +168,13 @@ private FusionRequestExecutor CreateRequestExecutor(
         string schemaName,
         FusionConfiguration configuration)
     {
+        ulong version;
+
+        unchecked
+        {
+            version = ++_version;
+        }
+
         var setup = _optionsMonitor.Get(schemaName);
 
         var options = CreateOptions(setup);
@@ -179,7 +193,7 @@ private FusionRequestExecutor CreateRequestExecutor(
         var pipeline = CreatePipeline(setup, schema, schemaServices, requestOptions);
 
         var contextPool = schemaServices.GetRequiredService<ObjectPool<PooledRequestContext>>();
-        var executor = new FusionRequestExecutor(schema, _applicationServices, pipeline, contextPool, 0);
+        var executor = new FusionRequestExecutor(schema, _applicationServices, pipeline, contextPool, version);
         var requestExecutorAccessor = schemaServices.GetRequiredService<RequestExecutorAccessor>();
         requestExecutorAccessor.RequestExecutor = executor;
 
@@ -646,6 +660,8 @@ private async Task WaitForUpdatesAsync()
 
                 await _manager.WarmupExecutorAsync(nextExecutor, false, _cancellationToken).ConfigureAwait(false);
 
+                await _manager._executorEvents.WriteCreatedAsync(nextExecutor, _cancellationToken).ConfigureAwait(false);
+
                 Executor = nextExecutor;
 
                 await _manager.EvictExecutorAsync(previousExecutor, _cancellationToken);
diff --git a/src/HotChocolate/Fusion-vnext/test/Fusion.Execution.Tests/Execution/FusionRequestExecutorManagerTests.cs b/src/HotChocolate/Fusion-vnext/test/Fusion.Execution.Tests/Execution/FusionRequestExecutorManagerTests.cs
@@ -1,15 +1,10 @@
-using System.Buffers;
-using System.Net.Security;
-using System.Text.Json;
-using HotChocolate.Buffers;
 using HotChocolate.Caching.Memory;
 using HotChocolate.Execution;
 using HotChocolate.Fusion.Configuration;
 using HotChocolate.Fusion.Execution.Nodes;
 using HotChocolate.Fusion.Execution.Pipeline;
 using HotChocolate.Language;
 using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Hosting;
 
 namespace HotChocolate.Fusion.Execution;
 
@@ -22,10 +17,6 @@ public async Task GetExecutorAsync_Throws_If_Schema_Does_Not_Exist()
         var schemaDocument =
             ComposeSchemaDocument(
                 """
-                schema {
-                    query: Query
-                }
-
                 type Query {
                     foo: String
                 }
@@ -54,10 +45,6 @@ public async Task CreateExecutor()
         var schemaDocument =
             ComposeSchemaDocument(
                 """
-                schema {
-                    query: Query
-                }
-
                 type Query {
                     foo: String
                 }
@@ -86,10 +73,6 @@ public async Task GetOperationPlanFromExecution()
         var schemaDocument =
             ComposeSchemaDocument(
                 """
-                schema {
-                    query: Query
-                }
-
                 type Query {
                     foo: String
                 }
@@ -181,10 +164,6 @@ public async Task Plan_Cache_Should_Be_Scoped_To_Executor()
         configProvider.UpdateConfiguration(
             CreateConfiguration(
                 """
-                schema {
-                  query: Query
-                }
-
                 type Query {
                   field2: String!
                 }
@@ -241,10 +220,6 @@ public async Task Executor_Should_Only_Be_Switched_Once_It_Is_Warmed_Up()
         configProvider.UpdateConfiguration(
             CreateConfiguration(
                 """
-                schema {
-                  query: Query
-                }
-
                 type Query {
                   field2: String!
                 }
@@ -304,10 +279,6 @@ public async Task WarmupTasks_Are_Applied_Correct_Number_Of_Times()
         configProvider.UpdateConfiguration(
             CreateConfiguration(
                 """
-                schema {
-                  query: Query
-                }
-
                 type Query {
                   field2: String!
                 }
@@ -426,66 +397,11 @@ private static FusionConfiguration CreateConfiguration(string? sourceSchemaText
     {
         sourceSchemaText ??=
             """
-            schema {
-              query: Query
-            }
-
             type Query {
               field: String!
             }
             """;
 
-        var schema = ComposeSchemaDocument(sourceSchemaText);
-
-        return new FusionConfiguration(
-            schema,
-            new JsonDocumentOwner(
-                JsonDocument.Parse("{ }"),
-                new EmptyMemoryOwner()));
-    }
-
-    private sealed class TestFusionConfigurationProvider(FusionConfiguration initialConfig) : IFusionConfigurationProvider
-    {
-        private List<IObserver<FusionConfiguration>> _observers = [];
-
-        public IDisposable Subscribe(IObserver<FusionConfiguration> observer)
-        {
-            if (Configuration is not null)
-            {
-                observer.OnNext(Configuration);
-            }
-
-            _observers.Add(observer);
-
-            return new Observer();
-        }
-
-        public ValueTask DisposeAsync() => ValueTask.CompletedTask;
-
-        public FusionConfiguration? Configuration { get; private set; } = initialConfig;
-
-        public void UpdateConfiguration(FusionConfiguration configuration)
-        {
-            Configuration = configuration;
-
-            foreach (var observer in _observers)
-            {
-                observer.OnNext(Configuration);
-            }
-        }
-
-        private sealed class Observer : IDisposable
-        {
-            public void Dispose()
-            {
-            }
-        }
-    }
-
-    private class EmptyMemoryOwner : IMemoryOwner<byte>
-    {
-        public Memory<byte> Memory => default;
-
-        public void Dispose() { }
+        return CreateFusionConfiguration(sourceSchemaText);
     }
 }
diff --git a/src/HotChocolate/Fusion-vnext/test/Fusion.Execution.Tests/Execution/FusionRequestExecutorProxyTests.cs b/src/HotChocolate/Fusion-vnext/test/Fusion.Execution.Tests/Execution/FusionRequestExecutorProxyTests.cs
@@ -0,0 +1,100 @@
+using HotChocolate.Execution;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace HotChocolate.Fusion.Execution;
+
+public class FusionRequestExecutorProxyTests : FusionTestBase
+{
+    [Fact]
+    public async Task Ensure_Executor_Is_Cached()
+    {
+        // arrange
+        var manager =
+            new ServiceCollection()
+                .AddGraphQLGateway()
+                .AddInMemoryConfiguration(
+                    ComposeSchemaDocument(
+                        """
+                        type Query {
+                          field: String!
+                        }
+                        """))
+                .Services
+                .BuildServiceProvider()
+                .GetRequiredService<FusionRequestExecutorManager>();
+
+        // act
+        var proxy = new RequestExecutorProxy(manager, manager, ISchemaDefinition.DefaultName);
+        var a = await proxy.GetExecutorAsync(CancellationToken.None);
+        var b = await proxy.GetExecutorAsync(CancellationToken.None);
+
+        // assert
+        Assert.Same(a, b);
+    }
+
+    [Fact]
+    public async Task Ensure_Executor_Is_Correctly_Swapped_When_Evicted()
+    {
+        // arrange
+        var executorUpdatedResetEvent = new ManualResetEventSlim(false);
+        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+
+        var configProvider = new TestFusionConfigurationProvider(
+            CreateFusionConfiguration(
+                """
+                type Query {
+                  field: String!
+                }
+                """));
+
+        var manager =
+            new ServiceCollection()
+                .AddGraphQLGateway()
+                .AddConfigurationProvider(_ => configProvider)
+                .Services
+                .BuildServiceProvider()
+                .GetRequiredService<FusionRequestExecutorManager>();
+        var updated = false;
+
+        var proxy = new TestProxy(manager, manager, ISchemaDefinition.DefaultName);
+        proxy.ExecutorUpdated += () =>
+        {
+            updated = true;
+            executorUpdatedResetEvent.Set();
+        };
+
+        // act
+        var a = await proxy.GetExecutorAsync(CancellationToken.None);
+
+        configProvider.UpdateConfiguration(
+            CreateFusionConfiguration(
+                """
+                type Query {
+                  field2: String!
+                }
+                """));
+
+        executorUpdatedResetEvent.Wait(cts.Token);
+        var b = await proxy.GetExecutorAsync(CancellationToken.None);
+
+        // assert
+        Assert.NotSame(a, b);
+        Assert.True(updated);
+    }
+
+    private class TestProxy(
+        IRequestExecutorProvider executorProvider,
+        IRequestExecutorEvents executorEvents,
+        string schemaName)
+        : RequestExecutorProxy(executorProvider, executorEvents, schemaName)
+    {
+        public event Action? ExecutorUpdated;
+
+        protected override void OnAfterRequestExecutorSwapped(
+            IRequestExecutor newExecutor,
+            IRequestExecutor oldExecutor)
+        {
+            ExecutorUpdated?.Invoke();
+        }
+    }
+}
diff --git a/src/HotChocolate/Fusion-vnext/test/Fusion.Execution.Tests/FusionTestBase.cs b/src/HotChocolate/Fusion-vnext/test/Fusion.Execution.Tests/FusionTestBase.cs

Original file line number	Diff line number	Diff line change
`@@ -197,10 +197,7 @@ await OnRequestExecutorCreatedAsync(context, executor, setup, cancellationToken)`
`197`	`197`	`registeredExecutor.Executor);`
`198`	`198`
`199`	`199`	`_events.RaiseEvent(`
`200`		`- new RequestExecutorEvent(`
`201`		`- RequestExecutorEventType.Created,`
`202`		`- schemaName,`
`203`		`- registeredExecutor.Executor));`
	`200`	`+ RequestExecutorEvent.Created(registeredExecutor.Executor));`
`204`	`201`
`205`	`202`	`return registeredExecutor;`
`206`	`203`	`}`
`@@ -219,11 +216,7 @@ await CreateRequestExecutorAsync(schemaName, isInitialCreation: false, Cancellat`
`219`	`216`
`220`	`217`	`try`
`221`	`218`	`{`
`222`		`- _events.RaiseEvent(`
`223`		`- new RequestExecutorEvent(`
`224`		`- RequestExecutorEventType.Evicted,`
`225`		`- schemaName,`
`226`		`- previousExecutor.Executor));`
	`219`	`+ _events.RaiseEvent(RequestExecutorEvent.Evicted(previousExecutor.Executor));`
`227`	`220`	`}`
`228`	`221`	`finally`
`229`	`222`	`{`