From 35180f269bdb43b1c71c14075c5c875326fd5fc2 Mon Sep 17 00:00:00 2001 From: Michael Staib Date: Wed, 17 Jun 2026 20:18:45 +0200 Subject: [PATCH 1/2] Remove content from result store --- .../Execution/ExecutionState.cs | 64 +++++++ .../Nodes/OperationBatchExecutionNode.cs | 56 +++++- .../Execution/Nodes/OperationExecutionNode.cs | 62 +++++-- .../Execution/OperationPlanContext.cs | 3 + .../Execution/OperationPlanExecutor.cs | 48 ++++- .../Execution/PendingMerge.cs | 173 ++++++++++++++++++ 6 files changed, 383 insertions(+), 23 deletions(-) create mode 100644 src/HotChocolate/Fusion/src/Fusion.Execution/Execution/PendingMerge.cs diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/ExecutionState.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/ExecutionState.cs index 436c46e5821..9265aef6bc6 100644 --- a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/ExecutionState.cs +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/ExecutionState.cs @@ -19,6 +19,8 @@ internal sealed class ExecutionState private readonly List _trackedNodeStateSlots = []; private readonly List _trackedDependencySlots = []; private readonly ConcurrentQueue _completedResults = new(); + private readonly ConcurrentQueue _pendingMerges = new(); + private Dictionary? _mergeFailures; private ulong[] _failedOrSkippedBitset = []; private bool _collectTelemetry; @@ -136,6 +138,8 @@ public void Reset() ResetRemainingDependencies(); _completedResults.Clear(); + ClearPendingMerges(); + _mergeFailures?.Clear(); _activeNodes = 0; Traces.Clear(); @@ -171,6 +175,52 @@ public void EnqueueForCompletion(ExecutionNodeResult result) public bool TryDequeueCompletedResult([NotNullWhen(true)] out ExecutionNodeResult? result) => _completedResults.TryDequeue(out result); + public void EnqueueMerge(PendingMerge merge) + { + _pendingMerges.Enqueue(merge); + Signal.Set(); + } + + public bool TryDequeuePendingMerge(out PendingMerge merge) + => _pendingMerges.TryDequeue(out merge); + + public void ApplyMerge(OperationPlanContext context, PendingMerge merge) + { + try + { + merge.Apply(context); + } + catch (OperationCanceledException) when (_cts.IsCancellationRequested) + { + MarkMergeFailed(merge.Node.Id, exception: null); + } + catch (Exception exception) + { + MarkMergeFailed(merge.Node.Id, exception); + context.DiagnosticEvents.SourceSchemaStoreError( + context, + merge.Node, + merge.SchemaName, + exception); + context.AddErrors(exception, merge.VariableValueSets, merge.ResultSelectionSet); + } + } + + public ExecutionNodeResult ApplyPendingMergeFailure(ExecutionNodeResult result) + { + if (_mergeFailures is not null + && _mergeFailures.Remove(result.Id, out var exception)) + { + return result with + { + Status = ExecutionStatus.Failed, + Exception = exception + }; + } + + return result; + } + public void CancelProcessing() { if (!_cts.IsCancellationRequested) @@ -581,6 +631,20 @@ internal void AddToBacklog(ExecutionNode node) } } + private void MarkMergeFailed(int nodeId, Exception? exception) + { + _mergeFailures ??= []; + _mergeFailures[nodeId] = exception; + } + + private void ClearPendingMerges() + { + while (_pendingMerges.TryDequeue(out var merge)) + { + merge.DisposeUnmerged(); + } + } + private bool RemoveFromBacklog(int nodeId, byte targetState) { if (!IsInBacklog(nodeId)) diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationBatchExecutionNode.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationBatchExecutionNode.cs index 7d11946eaa7..881c2889b54 100644 --- a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationBatchExecutionNode.cs +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationBatchExecutionNode.cs @@ -97,20 +97,42 @@ private async ValueTask ExecuteSingleAsync( hasSomeErrors = true; } + var pendingMerge = default(PendingMerge); + var hasPendingMerge = false; + try { - context.AddPartialResult( + pendingMerge = PendingMerge.Single( + this, + schemaName, operation.Source, - result, operation.ResultSelectionSet, + variables, + result, hasErrors); + hasPendingMerge = true; + context.EnqueuePendingMerge(pendingMerge); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { + if (hasPendingMerge) + { + pendingMerge.DisposeUnmerged(); + } + return ExecutionStatus.Failed; } catch (Exception exception) { + if (hasPendingMerge) + { + pendingMerge.DisposeUnmerged(); + } + else + { + result.Dispose(); + } + diagnosticEvents.SourceSchemaStoreError(context, this, schemaName, exception); context.AddErrors(exception, variables, operation.ResultSelectionSet); return ExecutionStatus.Failed; @@ -156,8 +178,8 @@ private async ValueTask ExecuteBatchAsync( var requests = requestBuilder.DrainToImmutable(); // Obtain a transport client for the source schema and stream the batch - // response. As each individual result arrives, we merge it into the - // result store so downstream nodes can consume the data. + // response. As each individual result arrives, we queue its merge for + // the executor loop so downstream nodes can consume the data after completion. var client = context.GetClient(schemaName, requests[0].OperationType); receivedResults.AsSpan(0, operationCount).Clear(); var overallStatus = ExecutionStatus.Success; @@ -172,20 +194,42 @@ private async ValueTask ExecuteBatchAsync( receivedResults[requestIndex] = true; + var pendingMerge = default(PendingMerge); + var hasPendingMerge = false; + try { - context.AddPartialResult( + pendingMerge = PendingMerge.Single( + this, + schemaName, op.Source, - result, op.ResultSelectionSet, + variablesByIndex[requestIndex], + result, hasErrors); + hasPendingMerge = true; + context.EnqueuePendingMerge(pendingMerge); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { + if (hasPendingMerge) + { + pendingMerge.DisposeUnmerged(); + } + return ExecutionStatus.Failed; } catch (Exception exception) { + if (hasPendingMerge) + { + pendingMerge.DisposeUnmerged(); + } + else + { + result.Dispose(); + } + diagnosticEvents.SourceSchemaStoreError(context, this, schemaName, exception); context.AddErrors(exception, variablesByIndex[requestIndex], op.ResultSelectionSet); overallStatus = ExecutionStatus.Failed; diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationExecutionNode.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationExecutionNode.cs index 80700dc31cb..d1c8d808028 100644 --- a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationExecutionNode.cs +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationExecutionNode.cs @@ -202,54 +202,88 @@ protected override async ValueTask OnExecuteAsync( return ExecutionStatus.Failed; } + var pendingMerge = default(PendingMerge); + var hasPendingMerge = false; + try { if (buffer is not null) { - context.AddPartialResults( + pendingMerge = PendingMerge.Multiple( + this, + schemaName, _source, - buffer.AsSpan(0, index), _resultSelectionSet, + variables, + buffer, + index, hasSomeErrors); + hasPendingMerge = true; } else if (singleResult is not null) { - var firstResult = singleResult; - context.AddPartialResults( + pendingMerge = PendingMerge.Single( + this, + schemaName, _source, - MemoryMarshal.CreateReadOnlySpan(ref firstResult, 1), _resultSelectionSet, + variables, + singleResult, hasSomeErrors); + hasPendingMerge = true; } else { - context.AddPartialResults( + pendingMerge = PendingMerge.Empty( + this, + schemaName, _source, - [], _resultSelectionSet, + variables, hasSomeErrors); + hasPendingMerge = true; } + + context.EnqueuePendingMerge(pendingMerge); + buffer = null; + singleResult = null; } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { // If the execution of the node was cancelled, either the entire request was cancelled // or the execution was halted. In both cases we do not want to produce any errors // and just exit the node as quickly as possible. + if (hasPendingMerge) + { + pendingMerge.DisposeUnmerged(); + } + return ExecutionStatus.Failed; } catch (Exception exception) { - diagnosticEvents.SourceSchemaStoreError(context, this, schemaName, exception); - context.AddErrors(exception, variables, _resultSelectionSet); - return ExecutionStatus.Failed; - } - finally - { - if (buffer is not null) + if (hasPendingMerge) + { + pendingMerge.DisposeUnmerged(); + } + else if (buffer is not null) { + foreach (var result in buffer.AsSpan(0, index)) + { + result?.Dispose(); + } + buffer.AsSpan(0, index).Clear(); ArrayPool.Shared.Return(buffer); } + else + { + singleResult?.Dispose(); + } + + diagnosticEvents.SourceSchemaStoreError(context, this, schemaName, exception); + context.AddErrors(exception, variables, _resultSelectionSet); + return ExecutionStatus.Failed; } return hasSomeErrors ? ExecutionStatus.PartialSuccess : ExecutionStatus.Success; diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/OperationPlanContext.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/OperationPlanContext.cs index 522bd7b3332..b08974817cb 100644 --- a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/OperationPlanContext.cs +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/OperationPlanContext.cs @@ -294,6 +294,9 @@ internal void TrackTransport(ExecutionNode node, Uri? uri, string? contentType) internal void CompleteNode(ExecutionNodeResult result) => _executionState.EnqueueForCompletion(result); + internal void EnqueuePendingMerge(PendingMerge merge) + => _executionState.EnqueueMerge(merge); + internal ImmutableArray CreateVariableValueSets( SelectionPath selectionSet, ReadOnlySpan forwardedVariables, diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/OperationPlanExecutor.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/OperationPlanExecutor.cs index 15f9e926b7e..18d4f00a3c4 100644 --- a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/OperationPlanExecutor.cs +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/OperationPlanExecutor.cs @@ -923,13 +923,27 @@ private static async Task ExecuteQueryAsync( while (!cancellationToken.IsCancellationRequested && executionState.IsProcessing()) { + while (executionState.TryDequeuePendingMerge(out var merge)) + { + executionState.ApplyMerge(context, merge); + } + while (executionState.TryDequeueCompletedResult(out var result)) { + while (executionState.TryDequeuePendingMerge(out var merge)) + { + executionState.ApplyMerge(context, merge); + } + + result = executionState.ApplyPendingMergeFailure(result); var node = plan.GetNodeById(result.Id); executionState.CompleteNode(plan, node, result); } - executionState.EnqueueNextNodes(context, cancellationToken); + if (!cancellationToken.IsCancellationRequested) + { + executionState.EnqueueNextNodes(context, cancellationToken); + } if (cancellationToken.IsCancellationRequested || !executionState.IsProcessing()) { @@ -975,13 +989,27 @@ private static async Task ExecuteMutationAsync( // as long as there are active nodes that result from processing the current subtree. while (!cancellationToken.IsCancellationRequested && executionState.HasActiveNodes()) { + while (executionState.TryDequeuePendingMerge(out var merge)) + { + executionState.ApplyMerge(context, merge); + } + while (executionState.TryDequeueCompletedResult(out var result)) { + while (executionState.TryDequeuePendingMerge(out var merge)) + { + executionState.ApplyMerge(context, merge); + } + + result = executionState.ApplyPendingMergeFailure(result); var node = plan.GetNodeById(result.Id); executionState.CompleteNode(plan, node, result); } - executionState.EnqueueNextNodes(context, cancellationToken); + if (!cancellationToken.IsCancellationRequested) + { + executionState.EnqueueNextNodes(context, cancellationToken); + } if (cancellationToken.IsCancellationRequested || !executionState.HasActiveNodes()) { @@ -1077,13 +1105,27 @@ private static async IAsyncEnumerable CreateResponseStream( while (!eventToken.IsCancellationRequested && executionState.IsProcessing()) { + while (executionState.TryDequeuePendingMerge(out var merge)) + { + executionState.ApplyMerge(context, merge); + } + while (executionState.TryDequeueCompletedResult(out var nodeResult)) { + while (executionState.TryDequeuePendingMerge(out var merge)) + { + executionState.ApplyMerge(context, merge); + } + + nodeResult = executionState.ApplyPendingMergeFailure(nodeResult); var node = plan.GetNodeById(nodeResult.Id); executionState.CompleteNode(plan, node, nodeResult); } - executionState.EnqueueNextNodes(context, eventToken); + if (!eventToken.IsCancellationRequested) + { + executionState.EnqueueNextNodes(context, eventToken); + } if (eventToken.IsCancellationRequested || !executionState.IsProcessing()) { diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/PendingMerge.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/PendingMerge.cs new file mode 100644 index 00000000000..a4a8ef385f2 --- /dev/null +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/PendingMerge.cs @@ -0,0 +1,173 @@ +using System.Buffers; +using System.Collections.Immutable; +using HotChocolate.Execution; +using HotChocolate.Fusion.Execution.Clients; +using HotChocolate.Fusion.Execution.Nodes; + +namespace HotChocolate.Fusion.Execution; + +internal readonly struct PendingMerge +{ + private readonly PendingMergeKind _kind; + private readonly SourceSchemaResult? _result; + private readonly SourceSchemaResult[]? _buffer; + private readonly int _count; + + private PendingMerge( + ExecutionNode node, + string schemaName, + SelectionPath sourcePath, + ResultSelectionSet resultSelectionSet, + ImmutableArray variableValueSets, + bool containsErrors, + PendingMergeKind kind, + SourceSchemaResult? result, + SourceSchemaResult[]? buffer, + int count) + { + Node = node; + SchemaName = schemaName; + SourcePath = sourcePath; + ResultSelectionSet = resultSelectionSet; + VariableValueSets = variableValueSets; + ContainsErrors = containsErrors; + _kind = kind; + _result = result; + _buffer = buffer; + _count = count; + } + + public ExecutionNode Node { get; } + + public string SchemaName { get; } + + public SelectionPath SourcePath { get; } + + public ResultSelectionSet ResultSelectionSet { get; } + + public ImmutableArray VariableValueSets { get; } + + public bool ContainsErrors { get; } + + public static PendingMerge Empty( + ExecutionNode node, + string schemaName, + SelectionPath sourcePath, + ResultSelectionSet resultSelectionSet, + ImmutableArray variableValueSets, + bool containsErrors) + => new( + node, + schemaName, + sourcePath, + resultSelectionSet, + variableValueSets, + containsErrors, + PendingMergeKind.Empty, + result: null, + buffer: null, + count: 0); + + public static PendingMerge Single( + ExecutionNode node, + string schemaName, + SelectionPath sourcePath, + ResultSelectionSet resultSelectionSet, + ImmutableArray variableValueSets, + SourceSchemaResult result, + bool containsErrors) + => new( + node, + schemaName, + sourcePath, + resultSelectionSet, + variableValueSets, + containsErrors, + PendingMergeKind.Single, + result, + buffer: null, + count: 1); + + public static PendingMerge Multiple( + ExecutionNode node, + string schemaName, + SelectionPath sourcePath, + ResultSelectionSet resultSelectionSet, + ImmutableArray variableValueSets, + SourceSchemaResult[] buffer, + int count, + bool containsErrors) + => new( + node, + schemaName, + sourcePath, + resultSelectionSet, + variableValueSets, + containsErrors, + PendingMergeKind.Multiple, + result: null, + buffer, + count); + + public void Apply(OperationPlanContext context) + { + switch (_kind) + { + case PendingMergeKind.Empty: + context.AddPartialResults(SourcePath, [], ResultSelectionSet, ContainsErrors); + break; + + case PendingMergeKind.Single: + context.AddPartialResult(SourcePath, _result!, ResultSelectionSet, ContainsErrors); + break; + + case PendingMergeKind.Multiple: + var buffer = _buffer!; + try + { + context.AddPartialResults( + SourcePath, + buffer.AsSpan(0, _count), + ResultSelectionSet, + ContainsErrors); + } + finally + { + buffer.AsSpan(0, _count).Clear(); + ArrayPool.Shared.Return(buffer); + } + break; + } + } + + public void DisposeUnmerged() + { + switch (_kind) + { + case PendingMergeKind.Single: + _result?.Dispose(); + break; + + case PendingMergeKind.Multiple: + var buffer = _buffer; + if (buffer is not null) + { + foreach (var result in buffer.AsSpan(0, _count)) + { + result?.Dispose(); + } + + buffer.AsSpan(0, _count).Clear(); + ArrayPool.Shared.Return(buffer); + } + break; + } + } + + private enum PendingMergeKind + { + Empty, + Single, + Multiple + } +} From 0b0dcff17839966bf733c40e358928c53ed4fd01 Mon Sep 17 00:00:00 2001 From: Michael Staib Date: Thu, 18 Jun 2026 16:49:05 +0200 Subject: [PATCH 2/2] edits --- .../Execution/Nodes/OperationExecutionNode.cs | 13 +--- .../Execution/PendingMerge.cs | 24 ------- .../Execution/Results/FetchResultStore.cs | 53 ++++++++++++++- .../Execution/OperationExecutionNodeTests.cs | 65 +++++++++++++++++-- .../Results/FetchResultStoreTests.cs | 62 ++++++++++++++++++ 5 files changed, 175 insertions(+), 42 deletions(-) diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationExecutionNode.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationExecutionNode.cs index d1c8d808028..8edc947dc2b 100644 --- a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationExecutionNode.cs +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Nodes/OperationExecutionNode.cs @@ -232,19 +232,12 @@ protected override async ValueTask OnExecuteAsync( hasSomeErrors); hasPendingMerge = true; } - else + + if (hasPendingMerge) { - pendingMerge = PendingMerge.Empty( - this, - schemaName, - _source, - _resultSelectionSet, - variables, - hasSomeErrors); - hasPendingMerge = true; + context.EnqueuePendingMerge(pendingMerge); } - context.EnqueuePendingMerge(pendingMerge); buffer = null; singleResult = null; } diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/PendingMerge.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/PendingMerge.cs index a4a8ef385f2..2cb7495e117 100644 --- a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/PendingMerge.cs +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/PendingMerge.cs @@ -49,25 +49,6 @@ private PendingMerge( public bool ContainsErrors { get; } - public static PendingMerge Empty( - ExecutionNode node, - string schemaName, - SelectionPath sourcePath, - ResultSelectionSet resultSelectionSet, - ImmutableArray variableValueSets, - bool containsErrors) - => new( - node, - schemaName, - sourcePath, - resultSelectionSet, - variableValueSets, - containsErrors, - PendingMergeKind.Empty, - result: null, - buffer: null, - count: 0); - public static PendingMerge Single( ExecutionNode node, string schemaName, @@ -113,10 +94,6 @@ public void Apply(OperationPlanContext context) { switch (_kind) { - case PendingMergeKind.Empty: - context.AddPartialResults(SourcePath, [], ResultSelectionSet, ContainsErrors); - break; - case PendingMergeKind.Single: context.AddPartialResult(SourcePath, _result!, ResultSelectionSet, ContainsErrors); break; @@ -166,7 +143,6 @@ public void DisposeUnmerged() private enum PendingMergeKind { - Empty, Single, Multiple } diff --git a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Results/FetchResultStore.cs b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Results/FetchResultStore.cs index 4b656c822ab..3b9c59434ed 100644 --- a/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Results/FetchResultStore.cs +++ b/src/HotChocolate/Fusion/src/Fusion.Execution/Execution/Results/FetchResultStore.cs @@ -128,6 +128,8 @@ public bool AddPartialResults( lock (_lock) { + var i = 0; + try { if (rootErrors is not null) @@ -137,8 +139,7 @@ public bool AddPartialResults( } var resultData = _result.Data; - - for (var i = 0; i < results.Length; i++) + for (i = 0; i < results.Length; i++) { var result = results[i]; _memory.Add(result); @@ -151,12 +152,18 @@ public bool AddPartialResults( errorTriesSpan[i], resultSelectionSet)) { + RegisterRemainingResults(_memory, results, i); return false; } } return true; } + catch + { + RegisterRemainingResults(_memory, results, i); + throw; + } finally { ReturnPathSegments(results); @@ -170,6 +177,22 @@ public bool AddPartialResults( ArrayPool.Shared.Return(dataElements); ArrayPool.Shared.Return(errorTries); } + + static void RegisterRemainingResults( + List _memory, + ReadOnlySpan results, + int i) + { + i++; + + if (i < results.Length) + { + for (; i < results.Length; i++) + { + _memory.Add(results[i]); + } + } + } } private bool AddPartialResultsNoErrors( @@ -189,11 +212,13 @@ private bool AddPartialResultsNoErrors( lock (_lock) { + var i = 0; + try { var resultData = _result.Data; - for (var i = 0; i < results.Length; i++) + for (i = 0; i < results.Length; i++) { var result = results[i]; _memory.Add(result); @@ -206,12 +231,18 @@ private bool AddPartialResultsNoErrors( errorTrie: null, resultSelectionSet)) { + RegisterRemainingResults(_memory, results, i); return false; } } return true; } + catch + { + RegisterRemainingResults(_memory, results, i); + throw; + } finally { ReturnPathSegments(results); @@ -223,6 +254,22 @@ private bool AddPartialResultsNoErrors( dataElementsSpan.Clear(); ArrayPool.Shared.Return(dataElements); } + + static void RegisterRemainingResults( + List _memory, + ReadOnlySpan results, + int i) + { + i++; + + if (i < results.Length) + { + for (; i < results.Length; i++) + { + _memory.Add(results[i]); + } + } + } } private bool AddSinglePartialResult( diff --git a/src/HotChocolate/Fusion/test/Fusion.Execution.Tests/Execution/OperationExecutionNodeTests.cs b/src/HotChocolate/Fusion/test/Fusion.Execution.Tests/Execution/OperationExecutionNodeTests.cs index c0e4bdff0ec..0f7409ba7bb 100644 --- a/src/HotChocolate/Fusion/test/Fusion.Execution.Tests/Execution/OperationExecutionNodeTests.cs +++ b/src/HotChocolate/Fusion/test/Fusion.Execution.Tests/Execution/OperationExecutionNodeTests.cs @@ -12,6 +12,29 @@ namespace HotChocolate.Fusion.Execution; public sealed class OperationExecutionNodeTests : FusionTestBase { + [Fact] + public async Task ExecuteAsync_Should_Succeed_When_SourceSchemaReturnsNoResults() + { + // arrange + var client = new EmptyQueryClient(); + var executor = await CreateExecutorAsync(client, SupportedOperationType.Query); + var request = CreateQueryRequest(); + + // act + await using var result = await executor.ExecuteAsync( + request, + TestContext.Current.CancellationToken); + + // assert + var operationResult = result.ExpectOperationResult(); + Assert.True(operationResult.Data.HasValue); + + if (operationResult.Errors is { } errors) + { + Assert.Empty(errors); + } + } + [Fact] public async Task MoveNextAsync_Should_DisposeEventArena_When_EventFailsAfterArenaMinted() { @@ -71,7 +94,9 @@ public async Task MoveNextAsync_Should_NotDisposePriorEventArena_When_NextEventF } } - private static async Task CreateExecutorAsync(ISourceSchemaClient client) + private static async Task CreateExecutorAsync( + ISourceSchemaClient client, + SupportedOperationType supportedOperations = SupportedOperationType.Subscription) { var services = new ServiceCollection(); services.AddHttpClient(); @@ -97,11 +122,21 @@ type Subscription { FusionSetupUtilities.Configure( builder, setup => setup.ClientConfigurationModifiers.Add( - _ => new TestSubscriptionClientConfiguration("events"))); + _ => new TestSubscriptionClientConfiguration("events", supportedOperations))); return await services.BuildGatewayAsync(TestContext.Current.CancellationToken); } + private static IOperationRequest CreateQueryRequest() + => OperationRequestBuilder.New() + .SetDocument( + """ + query { + field + } + """) + .Build(); + private static IOperationRequest CreateSubscriptionRequest() => OperationRequestBuilder.New() .SetDocument( @@ -116,7 +151,7 @@ private abstract class TestSubscriptionClient : ISourceSchemaClient { public SourceSchemaClientCapabilities Capabilities => SourceSchemaClientCapabilities.None; - public IAsyncEnumerable ExecuteAsync( + public virtual IAsyncEnumerable ExecuteAsync( OperationPlanContext context, SourceSchemaClientRequest request, CancellationToken cancellationToken) @@ -136,6 +171,24 @@ public abstract IAsyncEnumerable SubscribeAsync( public ValueTask DisposeAsync() => ValueTask.CompletedTask; } + private sealed class EmptyQueryClient : TestSubscriptionClient + { + public override async IAsyncEnumerable ExecuteAsync( + OperationPlanContext context, + SourceSchemaClientRequest request, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Yield(); + yield break; + } + + public override IAsyncEnumerable SubscribeAsync( + OperationPlanContext context, + SourceSchemaClientRequest request, + CancellationToken cancellationToken) + => throw new NotSupportedException(); + } + private sealed class ThrowingSubscriptionClient : TestSubscriptionClient { private readonly List _mintedArenas = []; @@ -196,11 +249,13 @@ public ISourceSchemaClient CreateClient( => client; } - private sealed class TestSubscriptionClientConfiguration(string name) + private sealed class TestSubscriptionClientConfiguration( + string name, + SupportedOperationType supportedOperations) : ISourceSchemaClientConfiguration { public string Name { get; } = name; - public SupportedOperationType SupportedOperations => SupportedOperationType.Subscription; + public SupportedOperationType SupportedOperations { get; } = supportedOperations; } } diff --git a/src/HotChocolate/Fusion/test/Fusion.Execution.Tests/Execution/Results/FetchResultStoreTests.cs b/src/HotChocolate/Fusion/test/Fusion.Execution.Tests/Execution/Results/FetchResultStoreTests.cs index 01961af3c68..2604b4c5c10 100644 --- a/src/HotChocolate/Fusion/test/Fusion.Execution.Tests/Execution/Results/FetchResultStoreTests.cs +++ b/src/HotChocolate/Fusion/test/Fusion.Execution.Tests/Execution/Results/FetchResultStoreTests.cs @@ -2,6 +2,7 @@ using HotChocolate.Buffers; using HotChocolate.Execution; using HotChocolate.Execution.Errors; +using HotChocolate.Fusion.Execution.Clients; using HotChocolate.Fusion.Execution.Nodes; using HotChocolate.Fusion.Language; using HotChocolate.Fusion.Text.Json; @@ -12,6 +13,59 @@ namespace HotChocolate.Fusion.Execution.Results; public sealed class FetchResultStoreTests : FusionTestBase { + private static readonly byte[] s_fieldPayload = """{"data":{"field":"value"}}"""u8.ToArray(); + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void AddPartialResults_Should_RegisterAllResults_When_MergeThrows(bool containsErrors) + { + // arrange + var schema = ComposeSchema( + """ + # name: test + type Query { + field: String + } + """); + var plan = PlanOperation(schema, "{ field }"); + var node = Assert.IsType(Assert.Single(plan.RootNodes)); + + using var resultArena = new MemoryArena(); + using var sourceArena = new MemoryArena(); + using var store = new FetchResultStore(); + store.Initialize( + resultArena, + schema, + DefaultErrorHandler.Default, + plan.Operation, + ErrorHandlingMode.Propagate, + includeFlags: 0, + deferFlags: 0, + pathSegmentLocalPoolCapacity: 16); + + var results = new[] + { + CreateSourceSchemaResult(sourceArena, CompactPath.Root), + CreateSourceSchemaResult(sourceArena, Path(~0)), + CreateSourceSchemaResult(sourceArena, CompactPath.Root) + }; + + // act + var exception = Assert.Throws( + () => store.AddPartialResults( + SelectionPath.Root, + results, + node.ResultSelectionSet, + containsErrors)); + + // assert + Assert.Contains("Expected StartArray but found StartObject.", exception.Message); + Assert.Contains(results[0], store.MemoryOwners); + Assert.Contains(results[1], store.MemoryOwners); + Assert.Contains(results[2], store.MemoryOwners); + } + [Fact] public void Reset_Should_ClearAccumulatedErrors_When_ReusedForNextEvent() { @@ -399,6 +453,14 @@ private static CompactPath Path(params int[] segments) return new CompactPath(buffer); } + private static SourceSchemaResult CreateSourceSchemaResult( + IMemoryArena arena, + CompactPath path) + { + var document = SourceResultDocument.Parse(arena, s_fieldPayload, s_fieldPayload.Length); + return new SourceSchemaResult(path, document); + } + private static string Normalize(JsonSegment segment) { using var document = JsonDocument.Parse(segment.AsSequence());