[Fusion] Add transport batching to planner level

src/HotChocolate/Fusion/benchmarks/k6/deep-recursion.js

@@ -3,7 +3,7 @@ import { check } from "k6";
 import { Rate } from "k6/metrics";
 import { textSummary } from "https://jslib.k6.io/k6-summary/0.0.1/index.js";
 
-const GRAPHQL_URL = 'http://localhost:5000/graphql';
+const GRAPHQL_URL = 'http://localhost:5220/graphql';
 const endpoint = __ENV.GATEWAY_ENDPOINT || GRAPHQL_URL;
 const mode = __ENV.MODE || "constant";
 const isConstant = mode === "constant";

src/HotChocolate/Fusion/benchmarks/k6/eShop.Gateway/Properties/launchSettings.json

@@ -4,8 +4,8 @@
       "commandName": "Project",
       "dotnetRunMessages": true,
       "launchBrowser": true,
-      "launchUrl": "http://localhost:5000/graphql",
-      "applicationUrl": "http://localhost:5000",
+      "launchUrl": "http://localhost:5220/graphql",
+      "applicationUrl": "http://localhost:5220",
       "environmentVariables": {
         "ASPNETCORE_ENVIRONMENT": "Production"
       }

src/HotChocolate/Fusion/src/Fusion.Diagnostics/Listeners/FusionActivityExecutionDiagnosticEventListener.cs

@@ -10,7 +10,8 @@ namespace HotChocolate.Fusion.Diagnostics.Listeners;
 
 internal sealed class FusionActivityExecutionDiagnosticEventListener(
     FusionActivityEnricher enricher,
-    InstrumentationOptions options) : FusionExecutionDiagnosticEventListener
+    InstrumentationOptions options)
+    : FusionExecutionDiagnosticEventListener
 {
     public override IDisposable ExecuteRequest(RequestContext context)
     {

src/HotChocolate/Fusion/src/Fusion.Diagnostics/Spans/ExecutePlanNodeSpan.cs

@@ -52,10 +52,6 @@ internal sealed class ExecutePlanNodeSpan(
         {
             SetSourceSchemaTags(activity, operationExecutionNode.Operation, schemaName);
         }
-        else if (node is OperationBatchExecutionNode batchExecutionNode)
-        {
-            SetSourceSchemaTags(activity, batchExecutionNode.Operation, schemaName);
-        }
 
         return new ExecutePlanNodeSpan(activity, context, node, schemaName, enricher);
     }

src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Clients/BatchStreamResult.cs

@@ -0,0 +1,7 @@
+namespace HotChocolate.Fusion.Execution.Clients;
+
+/// <summary>
+/// Represents a single result from a batch stream, tagged with the index of the
+/// request it belongs to.
+/// </summary>
+public readonly record struct BatchStreamResult(int RequestIndex, SourceSchemaResult Result);

src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Clients/ISourceSchemaClient.cs

@@ -27,16 +27,18 @@ ValueTask<SourceSchemaClientResponse> ExecuteAsync(
         CancellationToken cancellationToken);
 
     /// <summary>
-    /// Executes multiple GraphQL operations as a single batched transport request.
+    /// Executes multiple GraphQL operations as a single batched transport request and
+    /// streams results back as they arrive. Each result is tagged with its request index
+    /// so the caller can route it to the correct operation.
     /// </summary>
     /// <param name="context">The current operation plan execution context.</param>
     /// <param name="requests">The requests to include in the batch.</param>
     /// <param name="cancellationToken">A token to cancel the operation.</param>
     /// <returns>
-    /// A dictionary mapping each request's <see cref="SourceSchemaClientRequest.Node"/> ID
-    /// to its corresponding response.
+    /// An async stream of <see cref="BatchStreamResult"/> where each item contains
+    /// the request index and the corresponding <see cref="SourceSchemaResult"/>.
     /// </returns>
-    ValueTask<ImmutableArray<SourceSchemaClientResponse>> ExecuteBatchAsync(
+    IAsyncEnumerable<BatchStreamResult> ExecuteBatchStreamAsync(
         OperationPlanContext context,
         ImmutableArray<SourceSchemaClientRequest> requests,
         CancellationToken cancellationToken);

src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Clients/SourceSchemaClientRequest.cs

@@ -19,14 +19,6 @@ public sealed class SourceSchemaClientRequest
     /// </summary>
     public required string SchemaName { get; init; }
 
-    /// <summary>
-    /// Gets the optional batching group identifier assigned at planning time.
-    /// When set, the <see cref="ISourceSchemaScheduler"/> holds this request until
-    /// all nodes in the same group have submitted or been skipped, then dispatches
-    /// them together via <see cref="ISourceSchemaClient.ExecuteBatchAsync"/>.
-    /// </summary>
-    public int? BatchingGroupId { get; init; }
-
     /// <summary>
     /// Gets the GraphQL operation type (query, mutation, or subscription).
     /// </summary>