[Fusion] Added mutation support to planner. (#8478)

Author: michaelstaib

src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/ExecutionStrategy.cs

@@ -0,0 +1,14 @@
+namespace HotChocolate.Fusion.Execution;
+
+internal enum ExecutionStrategy
+{
+    /// <summary>
+    /// Everything that can be executed in parallel will be executed in parallel.
+    /// </summary>
+    Parallel,
+
+    /// <summary>
+    /// Root nodes will be executed in sequence, everything else can be parallelized.
+    /// </summary>
+    SequentialRoots
+}

src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/Nodes/Operation.cs

@@ -19,6 +19,7 @@ public sealed class Operation
 
     internal Operation(
         string id,
+        string hash,
         OperationDefinitionNode definition,
         IObjectTypeDefinition rootType,
         ISchemaDefinition schema,
@@ -35,6 +36,7 @@ internal Operation(
         ArgumentNullException.ThrowIfNull(includeConditions);
 
         Id = id;
+        Hash = hash;
         Definition = definition;
         RootType = rootType;
         Schema = schema;
@@ -51,6 +53,11 @@ internal Operation(
     /// </summary>
     public string Id { get; }
 
+    /// <summary>
+    /// Gets the hash of the original operation document.
+    /// </summary>
+    public string Hash { get; }
+
     /// <summary>
     /// Gets the name of the operation.
     /// </summary>

src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/Nodes/OperationCompiler.cs

@@ -24,7 +24,7 @@ public OperationCompiler(
         _typeNameField = new TypeNameField(nonNullStringType);
     }
 
-    public Operation Compile(string id, OperationDefinitionNode operationDefinition)
+    public Operation Compile(string id, string hash, OperationDefinitionNode operationDefinition)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(id);
         ArgumentNullException.ThrowIfNull(operationDefinition);
@@ -53,6 +53,7 @@ public Operation Compile(string id, OperationDefinitionNode operationDefinition)
 
             return new Operation(
                 id,
+                hash,
                 operationDefinition,
                 rootType,
                 _schema,

src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/Nodes/Serialization/JsonOperationPlanFormatter.cs

@@ -112,6 +112,7 @@ private static void WriteOperationNode(
     {
         jsonWriter.WriteStartObject();
         jsonWriter.WriteNumber("id", node.Id);
+        jsonWriter.WriteString("type", "Operation");
         jsonWriter.WriteString("schema", node.SchemaName);
         jsonWriter.WriteString("operation", node.Operation.ToString(indented: true));
 
@@ -182,6 +183,7 @@ private static void WriteIntrospectionNode(
         {
             jsonWriter.WriteStartObject();
             jsonWriter.WriteNumber("id", selection.Id);
+            jsonWriter.WriteString("type", "Introspection");
             jsonWriter.WriteString("responseName", selection.ResponseName);
             jsonWriter.WriteString("fieldName", selection.Field.Name);
             jsonWriter.WriteEndObject();

src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/OperationPlanExecutor.cs

@@ -1,6 +1,5 @@
 using System.Collections.Immutable;
 using System.Runtime.CompilerServices;
-using System.Runtime.InteropServices;
 using HotChocolate.Execution;
 using HotChocolate.Fusion.Execution.Nodes;
 using HotChocolate.Language;
@@ -14,10 +13,18 @@ public async Task<IExecutionResult> ExecuteAsync(
         CancellationToken cancellationToken = default)
     {
         context.Begin();
-        await ExecutorSession.ExecuteAsync(context, cancellationToken);
+        var strategy = DetermineExecutionStrategy(context);
+        await ExecutorSession.ExecuteAsync(context, strategy, cancellationToken);
         return context.Complete();
     }
 
+    private static ExecutionStrategy DetermineExecutionStrategy(OperationPlanContext context)
+        => context.OperationPlan.Operation.Definition.Operation switch
+        {
+            OperationType.Mutation => ExecutionStrategy.SequentialRoots,
+            _ => ExecutionStrategy.Parallel
+        };
+
     private sealed class ExecutorSession
     {
         private readonly List<ExecutionNode> _stack = [];
@@ -29,33 +36,96 @@ private sealed class ExecutorSession
         private readonly OperationPlan _plan;
         private readonly ImmutableArray<ExecutionNodeTrace>.Builder? _traces;
         private readonly CancellationToken _cancellationToken;
+        private readonly ExecutionStrategy _strategy;
+        private int _nextRootNode;
 
-        private ExecutorSession(OperationPlanContext context, CancellationToken cancellationToken)
+        private ExecutorSession(OperationPlanContext context, ExecutionStrategy strategy, CancellationToken cancellationToken)
         {
             _context = context;
+            _strategy = strategy;
             _cancellationToken = cancellationToken;
             _plan = context.OperationPlan;
             _backlog = [.. context.OperationPlan.AllNodes];
+
+            // For sequential execution (mutations), remove root nodes from backlog initially
+            if (_strategy == ExecutionStrategy.SequentialRoots)
+            {
+                foreach (var root in context.OperationPlan.RootNodes)
+                {
+                    _backlog.Remove(root);
+                }
+            }
+
             var collectTracing = context.Schema.GetRequestOptions().CollectOperationPlanTelemetry;
             _traces = collectTracing ? ImmutableArray.CreateBuilder<ExecutionNodeTrace>() : null;
         }
 
-        public static Task ExecuteAsync(OperationPlanContext context, CancellationToken cancellationToken)
-            => new ExecutorSession(context, cancellationToken).ExecuteInternalAsync();
+        public static Task ExecuteAsync(OperationPlanContext context, ExecutionStrategy strategy, CancellationToken cancellationToken)
+            => new ExecutorSession(context, strategy, cancellationToken).ExecuteInternalAsync();
 
         private async Task ExecuteInternalAsync()
         {
-            Start();
+            if (_strategy == ExecutionStrategy.Parallel)
+            {
+                await ExecuteQueryAsync();
+            }
+            else
+            {
+                await ExecuteMutationAsync();
+            }
+
+            if (_traces is { Count: > 0 })
+            {
+                _context.Traces = [.. _traces];
+            }
+        }
+
+        private async Task ExecuteQueryAsync()
+        {
+            // Start all root nodes immediately for parallel execution
+            StartAllRootNodes();
 
+            // Process until all nodes complete
             while (IsProcessing())
             {
                 await WaitForNextCompletionAsync();
                 EnqueueNextNodes();
             }
         }
 
-        private ReadOnlySpan<ExecutionNode> WaitingToRun
-            => CollectionsMarshal.AsSpan(_backlog);
+        private async Task ExecuteMutationAsync()
+        {
+            // Sequential root processing - one root at a time
+            while (StartNextRootNode())
+            {
+                // Complete the entire subtree of current root before starting next
+                var enqueued = true;
+                while (enqueued)
+                {
+                    await WaitForNextCompletionAsync();
+                    enqueued = EnqueueNextNodes();
+                }
+            }
+        }
+
+        private void StartAllRootNodes()
+        {
+            foreach (var node in _context.OperationPlan.RootNodes)
+            {
+                StartNode(node);
+            }
+        }
+
+        private bool StartNextRootNode()
+        {
+            var roots = _context.OperationPlan.RootNodes;
+            if (_nextRootNode < roots.Length)
+            {
+                StartNode(roots[_nextRootNode++]);
+                return true;
+            }
+            return false;
+        }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private bool IsProcessing() => _backlog.Count > 0 || _activeTasks.Count > 0;
@@ -69,7 +139,7 @@ private void SkipNode(ExecutionNode node)
             {
                 _backlog.Remove(current);
 
-                foreach (var enqueuedNode in WaitingToRun)
+                foreach (var enqueuedNode in _backlog)
                 {
                     if (enqueuedNode.Dependencies.Contains(current))
                     {
@@ -79,14 +149,6 @@ private void SkipNode(ExecutionNode node)
             }
         }
 
-        private void Start()
-        {
-            foreach (var node in _context.OperationPlan.RootNodes)
-            {
-                StartNode(node);
-            }
-        }
-
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void StartNode(ExecutionNode node)
         {
@@ -109,7 +171,7 @@ private async Task WaitForNextCompletionAsync()
                     _traces?.Add(new ExecutionNodeTrace
                     {
                         Id = task.Result.Id,
-                        SpanId = task.Result.Activity?.Id,
+                        SpanId = task.Result.Activity?.SpanId.ToHexString(),
                         Status = task.Result.Status,
                         Duration = task.Result.Duration
                     });
@@ -122,7 +184,7 @@ private async Task WaitForNextCompletionAsync()
                 else if (task.IsFaulted || task.IsCanceled)
                 {
                     // execution nodes are not expected to throw as exception should be handled within.
-                    // if they do its a fatal error for the execution, so we await failed task here
+                    // if they do it's a fatal error for the execution, so we await failed task here
                     // so that they can throw and terminate the execution.
                     await task;
                 }
@@ -134,16 +196,14 @@ private async Task WaitForNextCompletionAsync()
             }
 
             _completedTasks.Clear();
-
-            if (_backlog.Count == 0 && _traces is { Count: > 0 })
-            {
-                _context.Traces = [.. _traces];
-            }
         }
 
-        private void EnqueueNextNodes()
+        private bool EnqueueNextNodes()
         {
-            foreach (var node in WaitingToRun)
+            var enqueued = false;
+            _stack.Clear();
+
+            foreach (var node in _backlog)
             {
                 var dependenciesFulfilled = true;
 
@@ -159,9 +219,17 @@ private void EnqueueNextNodes()
 
                 if (dependenciesFulfilled)
                 {
-                    StartNode(node);
+                    _stack.Push(node);
+                    enqueued = true;
                 }
             }
+
+            foreach (var node in _stack)
+            {
+                StartNode(node);
+            }
+
+            return enqueued;
         }
     }
 }

src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/Pipeline/OperationPlanMiddleware.cs

@@ -36,9 +36,11 @@ public ValueTask InvokeAsync(RequestContext context, RequestDelegate next)
 
         // Before we can plan an operation, we must defragmentize it and remove statical include conditions.
         var operationId = context.GetOperationId();
+        var operationHash = context.OperationDocumentInfo.Hash.Value;
+        var operationShortHash = operationHash[..8];
         var rewritten = _rewriter.RewriteDocument(operationDocumentInfo.Document, context.Request.OperationName);
         var operation = GetOperation(rewritten);
-        var executionPlan = _planner.CreatePlan(operationId, operation);
+        var executionPlan = _planner.CreatePlan(operationId, operationHash, operationShortHash, operation);
         context.SetOperationPlan(executionPlan);
 
         return next(context);

src/HotChocolate/Fusion-vnext/src/Fusion.Execution/Execution/Results/ValueCompletion.cs

@@ -273,6 +273,7 @@ private bool TryCompleteAbstractValue(
             depth,
             parent);
 
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
     private IObjectTypeDefinition GetType(IType type, JsonElement data)
     {
         var namedType = type.NamedType();
@@ -286,6 +287,7 @@ private IObjectTypeDefinition GetType(IType type, JsonElement data)
         return _schema.Types.GetType<IObjectTypeDefinition>(typeName);
     }
 
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
     private void AssertDepthAllowed(ref int depth)
     {
         depth++;
@@ -299,6 +301,7 @@ private void AssertDepthAllowed(ref int depth)
 
 file static class ValueCompletionExtensions
 {
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
     public static bool IsNullOrUndefined(this JsonElement element)
         => element.ValueKind is JsonValueKind.Null or JsonValueKind.Undefined;
 }