Fixed Fusion Subscription Error Handling.

src/HotChocolate/Fusion/src/Core/Execution/ExecutionUtils.cs

@@ -543,6 +543,27 @@ public static void ExtractErrors(
         }
     }
 
+    public static void ExtractErrors(
+        DocumentNode document,
+        OperationDefinitionNode operation,
+        ResultBuilder resultBuilder,
+        IErrorHandler errorHandler,
+        JsonElement errors,
+        Path parentPath,
+        int pathDepth,
+        bool addDebugInfo)
+    {
+        if (errors.ValueKind is not JsonValueKind.Array)
+        {
+            return;
+        }
+
+        foreach (var error in errors.EnumerateArray())
+        {
+            ExtractError(document, operation, resultBuilder, errorHandler, error, parentPath, pathDepth, addDebugInfo);
+        }
+    }
+
     private static void ExtractError(
         DocumentNode document,
         OperationDefinitionNode operation,

src/HotChocolate/Fusion/src/Core/Execution/Nodes/Subscribe.cs

@@ -1,5 +1,6 @@
 using System.Buffers;
 using System.Runtime.CompilerServices;
+using System.Text.Json;
 using HotChocolate.Execution;
 using HotChocolate.Execution.DependencyInjection;
 using HotChocolate.Execution.Processing;
@@ -58,7 +59,7 @@ internal async IAsyncEnumerable<IOperationResult> SubscribeAsync(
             var context = rootContext.Clone();
             var operationContext = context.OperationContext;
 
-                // we ensure that the query plan is only included once per stream
+            // we ensure that the query plan is only included once per stream
             // in order to not inflate response sizes.
             if (initialResponse)
             {
@@ -80,6 +81,33 @@ internal async IAsyncEnumerable<IOperationResult> SubscribeAsync(
             }
             initialResponse = false;
 
+            if(response.Data.ValueKind is JsonValueKind.Null or JsonValueKind.Undefined
+                && response.Errors.ValueKind is JsonValueKind.Array)
+            {
+                ExtractErrors(
+                    operationContext.Operation.Document,
+                    operationContext.Operation.Definition,
+                    operationContext.Result,
+                    operationContext.ErrorHandler,
+                    response.Errors,
+                    HotChocolate.Path.Root,
+                    pathDepth: 0,
+                    addDebugInfo: context.ShowDebugInfo);
+
+                // Before we yield back the result we register with it the rented operation context.
+                // When the result is disposed in the transport after usage
+                // so will the operation context.
+                context.Result.RegisterForCleanup(
+                    () =>
+                    {
+                        context.Dispose();
+                        return default;
+                    });
+
+                yield return context.Result.BuildResult();
+                continue;
+            }
+
             // Before we can start building requests we need to rent state for the execution result.
             var rootSelectionSet = Unsafe.As<SelectionSet>(context.Operation.RootSelectionSet);
             var rootResult = context.Result.RentObject(rootSelectionSet.Selections.Count);

src/HotChocolate/Fusion/test/CommandLine.Tests/__snapshots__/ComposeCommandTests.Compose_Fusion_Graph_Append_Subgraph.md

@@ -28,6 +28,7 @@ type Mutation {
 
 type Subscription {
   onNewReview: Review!
+  onNewReviewError: Review!
 }
 
 type AddReviewPayload {
@@ -123,6 +124,7 @@ type Mutation {
 
 type Subscription {
   onNewReview: Review! @resolver(subgraph: "Reviews2", select: "{ onNewReview }", kind: "SUBSCRIBE")
+  onNewReviewError: Review! @resolver(subgraph: "Reviews2", select: "{ onNewReviewError }", kind: "SUBSCRIBE")
 }
 
 type AddReviewPayload {
@@ -218,7 +220,7 @@ scalar Date
 ```json
 {
   "Name": "Reviews2",
-  "Schema": "schema {\n  query: Query\n  mutation: Mutation\n  subscription: Subscription\n}\n\n\"The node interface is implemented by entities that have a global unique identifier.\"\ninterface Node {\n  id: ID!\n}\n\ntype Query {\n  \"Fetches an object given its ID.\"\n  node(\"ID of the object.\" id: ID!): Node\n  \"Lookup nodes by a list of IDs.\"\n  nodes(\"The list of node IDs.\" ids: [ID!]!): [Node]!\n  reviews: [Review!]!\n  reviewById(id: ID!): Review\n  authorById(id: ID!): User\n  productById(id: ID!): Product\n  reviewOrAuthor: ReviewOrAuthor!\n  viewer: Viewer!\n}\n\ntype Mutation {\n  addReview(input: AddReviewInput!): AddReviewPayload!\n}\n\ntype Subscription {\n  onNewReview: Review!\n}\n\ntype Product {\n  reviews: [Review!]!\n  id: ID!\n}\n\n\"The user who wrote the review.\"\ntype User implements Node {\n  reviews: [Review!]!\n  id: ID!\n  name: String!\n}\n\ntype Review implements Node {\n  errorField: String\n  id: ID!\n  author: User!\n  product: Product!\n  body: String!\n}\n\nunion ReviewOrAuthor = User | Review\n\ntype Viewer {\n  latestReview: Review\n  data: SomeData!\n}\n\ntype SomeData {\n  reviewsValue: String!\n}\n\ninput AddReviewInput {\n  body: String!\n  authorId: Int!\n  upc: Int!\n}\n\ntype AddReviewPayload {\n  review: Review\n}",
+  "Schema": "schema {\n  query: Query\n  mutation: Mutation\n  subscription: Subscription\n}\n\n\"The node interface is implemented by entities that have a global unique identifier.\"\ninterface Node {\n  id: ID!\n}\n\ntype Query {\n  \"Fetches an object given its ID.\"\n  node(\"ID of the object.\" id: ID!): Node\n  \"Lookup nodes by a list of IDs.\"\n  nodes(\"The list of node IDs.\" ids: [ID!]!): [Node]!\n  reviews: [Review!]!\n  reviewById(id: ID!): Review\n  authorById(id: ID!): User\n  productById(id: ID!): Product\n  reviewOrAuthor: ReviewOrAuthor!\n  viewer: Viewer!\n}\n\ntype Mutation {\n  addReview(input: AddReviewInput!): AddReviewPayload!\n}\n\ntype Subscription {\n  onNewReview: Review!\n  onNewReviewError: Review!\n}\n\ntype Product {\n  reviews: [Review!]!\n  id: ID!\n}\n\n\"The user who wrote the review.\"\ntype User implements Node {\n  reviews: [Review!]!\n  id: ID!\n  name: String!\n}\n\ntype Review implements Node {\n  errorField: String\n  id: ID!\n  author: User!\n  product: Product!\n  body: String!\n}\n\nunion ReviewOrAuthor = User | Review\n\ntype Viewer {\n  latestReview: Review\n  data: SomeData!\n}\n\ntype SomeData {\n  reviewsValue: String!\n}\n\ninput AddReviewInput {\n  body: String!\n  authorId: Int!\n  upc: Int!\n}\n\ntype AddReviewPayload {\n  review: Review\n}",
   "Extensions": [
     "extend type Query {\n  authorById(id: ID!\n    @is(field: \"id\")): User\n  productById(id: ID!\n    @is(field: \"id\")): Product\n}\n\nschema\n  @rename(coordinate: \"Query.authorById\", newName: \"userById\") {\n\n}"
   ],

src/HotChocolate/Fusion/test/Composition.Tests/__snapshots__/DemoIntegrationTests.Accounts_And_Reviews2_Products_With_Nodes.graphql

@@ -79,6 +79,8 @@ type Mutation {
 type Subscription {
   onNewReview: Review!
     @resolver(subgraph: "Reviews2", select: "{ onNewReview }", kind: "SUBSCRIBE")
+  onNewReviewError: Review!
+    @resolver(subgraph: "Reviews2", select: "{ onNewReviewError }", kind: "SUBSCRIBE")
 }
 
 type AddReviewPayload {

src/HotChocolate/Fusion/test/Core.Tests/DemoIntegrationTests.cs

@@ -362,12 +362,11 @@ public async Task Authors_And_Reviews_Subscription_OnNewReview()
 
         // act
         var fusionGraph = await new FusionGraphComposer(logFactory: _logFactory).ComposeAsync(
-            new[]
-            {
+            [
                 demoProject.Reviews2.ToConfiguration(Reviews2ExtensionSdl),
-                demoProject.Accounts.ToConfiguration(AccountsExtensionSdl),
-            },
-            default,
+                demoProject.Accounts.ToConfiguration(AccountsExtensionSdl)
+            ],
+            null,
             cts.Token);
 
         var executor = await new ServiceCollection()
@@ -403,6 +402,55 @@ subscription OnNewReview {
         await snapshot.MatchMarkdownAsync(cts.Token);
     }
 
+    [Fact]
+    public async Task Authors_And_Reviews_Subscription_OnNewReviewError()
+    {
+        // arrange
+        using var cts = TestEnvironment.CreateCancellationTokenSource();
+        using var demoProject = await DemoProject.CreateAsync(cts.Token);
+
+        // act
+        var fusionGraph = await new FusionGraphComposer(logFactory: _logFactory).ComposeAsync(
+            [
+                demoProject.Reviews2.ToConfiguration(Reviews2ExtensionSdl),
+                demoProject.Accounts.ToConfiguration(AccountsExtensionSdl)
+            ],
+            null,
+            cts.Token);
+
+        var executor = await new ServiceCollection()
+            .AddSingleton(demoProject.HttpClientFactory)
+            .AddSingleton(demoProject.WebSocketConnectionFactory)
+            .AddFusionGatewayServer()
+            .ConfigureFromDocument(SchemaFormatter.FormatAsDocument(fusionGraph))
+            .BuildRequestExecutorAsync(cancellationToken: cts.Token);
+
+        var request = Parse(
+            """
+            subscription OnNewReview {
+                onNewReviewError {
+                    body
+                    author {
+                        name
+                    }
+                }
+            }
+            """);
+
+        // act
+        await using var result = await executor.ExecuteAsync(
+            OperationRequestBuilder
+                .New()
+                .SetDocument(request)
+                .Build(),
+            cts.Token);
+
+        // assert
+        var snapshot = new Snapshot();
+        await CollectStreamSnapshotData(snapshot, request, result, cts.Token);
+        await snapshot.MatchMarkdownAsync(cts.Token);
+    }
+
     [Fact]
     public async Task Authors_And_Reviews_Subscription_OnError()
     {
@@ -1913,60 +1961,6 @@ query TopProducts {
         Assert.Null(result.ExpectOperationResult().Errors);
     }
 
-    // TODO : FIX THIS TEST
-    [Fact(Skip = "This test does not work anymore as it uses CCN which we removed ... ")]
-    public async Task ResolveByKey_Handles_Null_Item_Correctly()
-    {
-        // arrange
-        using var demoProject = await DemoProject.CreateAsync();
-
-        // act
-        var fusionGraph = await new FusionGraphComposer(logFactory: _logFactory).ComposeAsync(
-            new[]
-            {
-                demoProject.Products.ToConfiguration(),
-                demoProject.Resale.ToConfiguration(),
-            }, new FusionFeatureCollection(FusionFeatures.NodeField));
-
-        var executor = await new ServiceCollection()
-            .AddSingleton(demoProject.HttpClientFactory)
-            .AddSingleton(demoProject.WebSocketConnectionFactory)
-            .AddFusionGatewayServer()
-            .ConfigureFromDocument(SchemaFormatter.FormatAsDocument(fusionGraph))
-            .BuildRequestExecutorAsync();
-
-        var request = Parse(
-            """
-            {
-              viewer {
-                # The second product does not exist in the products subgraph
-                recommendedResalableProducts {
-                  edges {
-                    node {
-                      product? {
-                        id
-                        name
-                      }
-                    }
-                  }
-                }
-              }
-            }
-            """);
-
-        // act
-        await using var result = await executor.ExecuteAsync(
-            OperationRequestBuilder
-                .New()
-                .SetDocument(request)
-                .Build());
-
-        // assert
-        var snapshot = new Snapshot();
-        CollectSnapshotData(snapshot, request, result);
-        await snapshot.MatchMarkdownAsync();
-    }
-
     public sealed class HotReloadConfiguration : IObservable<GatewayConfiguration>
     {
         private GatewayConfiguration _configuration;

src/HotChocolate/Fusion/test/Core.Tests/__snapshots__/ConfigurationRewriterTests.Rewrite_HttpClient_Configuration.md

@@ -28,6 +28,7 @@ type Mutation {
 
 type Subscription {
   onNewReview: Review! @resolver(subgraph: "Reviews2", select: "{ onNewReview }", kind: "SUBSCRIBE")
+  onNewReviewError: Review! @resolver(subgraph: "Reviews2", select: "{ onNewReviewError }", kind: "SUBSCRIBE")
 }
 
 type AddReviewPayload {
@@ -123,6 +124,7 @@ type Mutation {
 
 type Subscription {
   onNewReview: Review! @resolver(subgraph: "Reviews2", select: "{ onNewReview }", kind: "SUBSCRIBE")
+  onNewReviewError: Review! @resolver(subgraph: "Reviews2", select: "{ onNewReviewError }", kind: "SUBSCRIBE")
 }
 
 type AddReviewPayload {

src/HotChocolate/Fusion/test/Core.Tests/__snapshots__/DemoIntegrationTests.Authors_And_Reviews_And_Products_AutoCompose.graphql

@@ -68,6 +68,8 @@ type Mutation {
 type Subscription {
   onNewReview: Review!
     @resolver(subgraph: "Reviews2", select: "{ onNewReview }", kind: "SUBSCRIBE")
+  onNewReviewError: Review!
+    @resolver(subgraph: "Reviews2", select: "{ onNewReviewError }", kind: "SUBSCRIBE")
 }
 
 type AddReviewPayload {

src/HotChocolate/Fusion/test/Core.Tests/__snapshots__/DemoIntegrationTests.Authors_And_Reviews_And_Products_Introspection.md

@@ -458,6 +458,13 @@
                 "name": null,
                 "kind": "NON_NULL"
               }
+            },
+            {
+              "name": "onNewReviewError",
+              "type": {
+                "name": null,
+                "kind": "NON_NULL"
+              }
             }
           ]
         },

src/HotChocolate/Fusion/test/Core.Tests/__snapshots__/DemoIntegrationTests.Authors_And_Reviews_AutoCompose.graphql

@@ -49,6 +49,8 @@ type Mutation {
 type Subscription {
   onNewReview: Review!
     @resolver(subgraph: "Reviews2", select: "{ onNewReview }", kind: "SUBSCRIBE")
+  onNewReviewError: Review!
+    @resolver(subgraph: "Reviews2", select: "{ onNewReviewError }", kind: "SUBSCRIBE")
 }
 
 type AddReviewPayload {

src/HotChocolate/Fusion/test/Core.Tests/__snapshots__/DemoIntegrationTests.Authors_And_Reviews_Subscription_OnError.md

@@ -21,19 +21,7 @@
 {
   "errors": [
     {
-      "message": "Cannot return null for non-nullable field.",
-      "locations": [
-        {
-          "line": 2,
-          "column": 5
-        }
-      ],
-      "path": [
-        "onError"
-      ],
-      "extensions": {
-        "code": "HC0018"
-      }
+      "message": "Unexpected Execution Error"
     }
   ],
   "data": null

src/HotChocolate/Fusion/test/Core.Tests/__snapshots__/DemoIntegrationTests.Authors_And_Reviews_Subscription_OnError_SSE.md

@@ -21,19 +21,7 @@
 {
   "errors": [
     {
-      "message": "Cannot return null for non-nullable field.",
-      "locations": [
-        {
-          "line": 2,
-          "column": 5
-        }
-      ],
-      "path": [
-        "onError"
-      ],
-      "extensions": {
-        "code": "HC0018"
-      }
+      "message": "Unexpected Execution Error"
     }
   ],
   "data": null

src/HotChocolate/Fusion/test/Core.Tests/__snapshots__/DemoIntegrationTests.Authors_And_Reviews_Subscription_OnNewReviewError.md

@@ -0,0 +1,62 @@
+# Authors_And_Reviews_Subscription_OnNewReviewError
+
+## Result 1
+
+```text
+{
+  "errors": [
+    {
+      "message": "Unexpected Execution Error"
+    }
+  ],
+  "data": null
+}
+```
+
+## Request
+
+```graphql
+subscription OnNewReview {
+  onNewReviewError {
+    body
+    author {
+      name
+    }
+  }
+}
+```
+
+## QueryPlan Hash
+
+```text
+0A82A531BB6C8D6C44B690B4509A1667A89A3C2C
+```
+
+## QueryPlan
+
+```json
+{
+  "document": "subscription OnNewReview { onNewReviewError { body author { name } } }",
+  "operation": "OnNewReview",
+  "rootNode": {
+    "type": "Subscribe",
+    "subgraph": "Reviews2",
+    "document": "subscription OnNewReview_1 { onNewReviewError { body author { name } } }",
+    "selectionSetId": 0,
+    "nodes": [
+      {
+        "type": "Sequence",
+        "nodes": [
+          {
+            "type": "Compose",
+            "selectionSetIds": [
+              0
+            ]
+          }
+        ]
+      }
+    ]
+  }
+}
+```
+