diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f2ffe8f..c5dcc995 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Isolate incoming payload hydration failures to the failed delivery, preserving the shared work-item stream and unrelated in-flight work ([#825](https://github.com/microsoft/durabletask-dotnet/issues/825)). - Externalize continue-as-new carryover event payloads using the configured UTF-8 size limits, preventing oversized completion responses ([#819](https://github.com/microsoft/durabletask-dotnet/issues/819)). ## v1.27.1 diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs index 3274229c..b2e88a84 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs @@ -4,6 +4,7 @@ using System.Text; using Grpc.Core; using Grpc.Core.Interceptors; +using P = Microsoft.DurableTask.Protobuf; namespace Microsoft.DurableTask; @@ -143,7 +144,20 @@ public override AsyncServerStreamingCall AsyncServerStreamingCall wrapped = new TransformingStreamReader(call.ResponseStream, async (msg, ct) => { - await this.ResolveResponsePayloadsAsync(msg, ct); + try + { + await this.ResolveResponsePayloadsAsync(msg, ct); + } + catch (Exception ex) when (msg is P.WorkItem workItem + && workItem.RequestCase is P.WorkItem.RequestOneofCase.ActivityRequest + or P.WorkItem.RequestOneofCase.OrchestratorRequest + or P.WorkItem.RequestOneofCase.EntityRequest + or P.WorkItem.RequestOneofCase.EntityRequestV2 + && !ct.IsCancellationRequested) + { + throw new WorkItemPayloadException(workItem, ex); + } + return msg; }); diff --git a/src/Grpc/Grpc.csproj b/src/Grpc/Grpc.csproj index e433e01e..4798b8ba 100644 --- a/src/Grpc/Grpc.csproj +++ b/src/Grpc/Grpc.csproj @@ -21,4 +21,10 @@ + + + + + + diff --git a/src/Grpc/WorkItemPayloadException.cs b/src/Grpc/WorkItemPayloadException.cs new file mode 100644 index 00000000..6386098a --- /dev/null +++ b/src/Grpc/WorkItemPayloadException.cs @@ -0,0 +1,20 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask; + +/// +/// A received delivery whose payloads could not be resolved. The item is only safe to abandon, not dispatch. +/// +/// The failed delivery, including its original completion token. +/// The payload resolution failure. +internal sealed class WorkItemPayloadException(P.WorkItem workItem, Exception innerException) + : Exception("Failed to resolve payloads for a work-item delivery.", innerException) +{ + /// + /// Gets the failed delivery. Its payload fields may be partially resolved. + /// + public P.WorkItem WorkItem { get; } = workItem; +} diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index abfba18a..96e063e3 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -279,6 +279,13 @@ static OrchestrationRuntimeState BuildRuntimeState( return runtimeState; } + static string GetWorkItemInstanceId(P.WorkItem? workItem) => + workItem?.OrchestratorRequest?.InstanceId ?? + workItem?.ActivityRequest?.OrchestrationInstance?.InstanceId ?? + workItem?.EntityRequest?.InstanceId ?? + workItem?.EntityRequestV2?.InstanceId ?? + string.Empty; + async ValueTask> GetPastEventsAsync( P.OrchestratorRequest orchestratorRequest, CancellationToken cancellation) @@ -385,16 +392,15 @@ async Task ProcessWorkItemsAsync( // window. If that fires we treat the stream as silently disconnected and reconnect. TimeSpan silentDisconnectTimeout = this.internalOptions.SilentDisconnectTimeout; - // NOTE: the consumer deliberately does NOT wrap its await foreach in an outer loop. - // The underlying IAsyncStreamReader is single-use — once the server terminates the stream - // (e.g. via a graceful HTTP/2 GOAWAY with OK trailers during a rolling upgrade), MoveNext - // returns false forever and re-entering await foreach would tight-spin with no yield. + // Read directly so a delivery-local hydration failure does not terminate a ReadAllAsync + // iterator. A false MoveNext still ends consumption permanently (no drain tight-spin). WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - ct => stream.ResponseStream.ReadAllAsync(ct), + stream.ResponseStream, silentDisconnectTimeout, workItem => this.DispatchWorkItem(workItem, cancellation), onFirstMessage, - cancellation); + cancellation, + onPayloadFailure: this.OnWorkItemPayloadFailureAsync); switch (result.Outcome) { @@ -534,90 +540,82 @@ void RunBackgroundTask(P.WorkItem? workItem, Func handler, CancellationTok } catch (Exception ex) { - string instanceId = - workItem?.OrchestratorRequest?.InstanceId ?? - workItem?.ActivityRequest?.OrchestrationInstance?.InstanceId ?? - workItem?.EntityRequest?.InstanceId ?? - workItem?.EntityRequestV2?.InstanceId ?? - string.Empty; - this.Logger.UnexpectedError(ex, instanceId); - - if (workItem?.OrchestratorRequest != null) - { - this.Logger.AbandoningOrchestratorWorkItem(instanceId, workItem.CompletionToken ?? string.Empty); - await this.ExecuteWithRetryAsync( - async () => await this.client.AbandonTaskOrchestratorWorkItemAsync( - new P.AbandonOrchestrationTaskRequest - { - CompletionToken = workItem.CompletionToken, - }, - cancellationToken: cancellation), - nameof(this.client.AbandonTaskOrchestratorWorkItemAsync), - cancellation); - this.Logger.AbandonedOrchestratorWorkItem(instanceId, workItem.CompletionToken ?? string.Empty); - } - else if (workItem?.ActivityRequest != null) - { - this.Logger.AbandoningActivityWorkItem( - instanceId, - workItem.ActivityRequest.Name, - workItem.ActivityRequest.TaskId, - workItem.CompletionToken ?? string.Empty); - await this.ExecuteWithRetryAsync( - async () => await this.client.AbandonTaskActivityWorkItemAsync( - new P.AbandonActivityTaskRequest - { - CompletionToken = workItem.CompletionToken, - }, - cancellationToken: cancellation), - nameof(this.client.AbandonTaskActivityWorkItemAsync), - cancellation); - this.Logger.AbandonedActivityWorkItem( - instanceId, - workItem.ActivityRequest.Name, - workItem.ActivityRequest.TaskId, - workItem.CompletionToken ?? string.Empty); - } - else if (workItem?.EntityRequest != null) - { - this.Logger.AbandoningEntityWorkItem( - workItem.EntityRequest.InstanceId, - workItem.CompletionToken ?? string.Empty); - await this.ExecuteWithRetryAsync( - async () => await this.client.AbandonTaskEntityWorkItemAsync( - new P.AbandonEntityTaskRequest - { - CompletionToken = workItem.CompletionToken, - }, - cancellationToken: cancellation), - nameof(this.client.AbandonTaskEntityWorkItemAsync), - cancellation); - this.Logger.AbandonedEntityWorkItem( - workItem.EntityRequest.InstanceId, - workItem.CompletionToken ?? string.Empty); - } - else if (workItem?.EntityRequestV2 != null) - { - this.Logger.AbandoningEntityWorkItem( - workItem.EntityRequestV2.InstanceId, - workItem.CompletionToken ?? string.Empty); - await this.ExecuteWithRetryAsync( - async () => await this.client.AbandonTaskEntityWorkItemAsync( - new P.AbandonEntityTaskRequest - { - CompletionToken = workItem.CompletionToken, - }, - cancellationToken: cancellation), - nameof(this.client.AbandonTaskEntityWorkItemAsync), - cancellation); - this.Logger.AbandonedEntityWorkItem( - workItem.EntityRequestV2.InstanceId, - workItem.CompletionToken ?? string.Empty); - } + this.Logger.UnexpectedError(ex, GetWorkItemInstanceId(workItem)); + await this.AbandonWorkItemAsync(workItem, cancellation); } }); } + async Task OnWorkItemPayloadFailureAsync(WorkItemPayloadException exception, CancellationToken cancellation) + { + P.WorkItem workItem = exception.WorkItem; + this.Logger.UnexpectedError(exception, GetWorkItemInstanceId(workItem)); + try + { + await this.AbandonWorkItemAsync(workItem, cancellation); + } + catch (RpcException ex) when (ex.StatusCode == StatusCode.NotFound) + { + // Ownership loss is delivery-local; other abandon failures must reach the reconnect loop. + this.Logger.WorkItemOwnershipLost(workItem.RequestCase.ToString(), workItem.CompletionToken, ex); + } + } + + async Task AbandonWorkItemAsync(P.WorkItem? workItem, CancellationToken cancellation) + { + string instanceId = GetWorkItemInstanceId(workItem); + if (workItem?.OrchestratorRequest != null) + { + this.Logger.AbandoningOrchestratorWorkItem(instanceId, workItem.CompletionToken ?? string.Empty); + await this.ExecuteWithRetryAsync( + async () => await this.client.AbandonTaskOrchestratorWorkItemAsync( + new P.AbandonOrchestrationTaskRequest + { + CompletionToken = workItem.CompletionToken, + }, + cancellationToken: cancellation), + nameof(this.client.AbandonTaskOrchestratorWorkItemAsync), + cancellation); + this.Logger.AbandonedOrchestratorWorkItem(instanceId, workItem.CompletionToken ?? string.Empty); + } + else if (workItem?.ActivityRequest != null) + { + this.Logger.AbandoningActivityWorkItem( + instanceId, + workItem.ActivityRequest.Name, + workItem.ActivityRequest.TaskId, + workItem.CompletionToken ?? string.Empty); + await this.ExecuteWithRetryAsync( + async () => await this.client.AbandonTaskActivityWorkItemAsync( + new P.AbandonActivityTaskRequest + { + CompletionToken = workItem.CompletionToken, + }, + cancellationToken: cancellation), + nameof(this.client.AbandonTaskActivityWorkItemAsync), + cancellation); + this.Logger.AbandonedActivityWorkItem( + instanceId, + workItem.ActivityRequest.Name, + workItem.ActivityRequest.TaskId, + workItem.CompletionToken ?? string.Empty); + } + else if (workItem?.EntityRequest != null || workItem?.EntityRequestV2 != null) + { + this.Logger.AbandoningEntityWorkItem(instanceId, workItem.CompletionToken ?? string.Empty); + await this.ExecuteWithRetryAsync( + async () => await this.client.AbandonTaskEntityWorkItemAsync( + new P.AbandonEntityTaskRequest + { + CompletionToken = workItem.CompletionToken, + }, + cancellationToken: cancellation), + nameof(this.client.AbandonTaskEntityWorkItemAsync), + cancellation); + this.Logger.AbandonedEntityWorkItem(instanceId, workItem.CompletionToken ?? string.Empty); + } + } + async Task OnRunOrchestratorAsync( P.OrchestratorRequest request, string completionToken, diff --git a/src/Worker/Grpc/Logs.cs b/src/Worker/Grpc/Logs.cs index f378a7fc..ea1efa34 100644 --- a/src/Worker/Grpc/Logs.cs +++ b/src/Worker/Grpc/Logs.cs @@ -107,5 +107,8 @@ static partial class Logs [LoggerMessage(EventId = 78, Level = LogLevel.Warning, Message = "Activity notification callback failed for phase '{Phase}'.")] public static partial void ActivityNotificationFailed(this ILogger logger, ActivityNotificationPhase phase, Exception exception); + + [LoggerMessage(EventId = 79, Level = LogLevel.Warning, Message = "Ownership of '{workItemType}' work item was lost before abandonment. Completion token = '{completionToken}'")] + public static partial void WorkItemOwnershipLost(this ILogger logger, string workItemType, string completionToken, Exception exception); } } diff --git a/src/Worker/Grpc/WorkItemStreamConsumer.cs b/src/Worker/Grpc/WorkItemStreamConsumer.cs index 9c39ef7b..d814f2a3 100644 --- a/src/Worker/Grpc/WorkItemStreamConsumer.cs +++ b/src/Worker/Grpc/WorkItemStreamConsumer.cs @@ -43,10 +43,7 @@ internal static class WorkItemStreamConsumer /// /// Consume a work-item stream until shutdown, silent disconnect, or graceful drain. /// - /// - /// Factory that opens the stream with the supplied linked-cancellation token. Production passes - /// ct => stream.ResponseStream.ReadAllAsync(ct); tests pass arbitrary fakes. - /// + /// The response reader. Each read receives the linked cancellation token. /// /// How long to wait between successive items (or health pings) before treating the stream as /// silently disconnected. Non-positive values disable detection entirely. @@ -68,14 +65,19 @@ internal static class WorkItemStreamConsumer /// real elapsed time. Never invoked when disables /// detection. /// + /// + /// Optional delivery-local recovery for a payload resolution failure. Exceptions from recovery + /// propagate to the caller for connection-level handling. + /// /// The classified outcome plus whether any message was observed. public static async Task ConsumeAsync( - Func> openStream, + IAsyncStreamReader stream, TimeSpan silentDisconnectTimeout, Action onItem, Action? onFirstMessage, CancellationToken cancellation, - Action? onSilentDisconnectTimerArmed = null) + Action? onSilentDisconnectTimerArmed = null, + Func? onPayloadFailure = null) { bool silentDisconnectEnabled = silentDisconnectTimeout > TimeSpan.Zero; @@ -100,19 +102,38 @@ void ArmSilentDisconnectTimer() bool firstMessageObserved = false; - try + void ObserveMessage() { - await foreach (P.WorkItem workItem in openStream(tokenSource.Token).ConfigureAwait(false)) + ArmSilentDisconnectTimer(); + + if (!firstMessageObserved) { - ArmSilentDisconnectTimer(); + firstMessageObserved = true; + onFirstMessage?.Invoke(); + } + } - if (!firstMessageObserved) + try + { + while (true) + { + try { - firstMessageObserved = true; - onFirstMessage?.Invoke(); + if (!await stream.MoveNext(tokenSource.Token).ConfigureAwait(false)) + { + break; + } } + catch (WorkItemPayloadException ex) when (onPayloadFailure is not null) + { + ObserveMessage(); + await onPayloadFailure(ex, tokenSource.Token).ConfigureAwait(false); + continue; + } + + ObserveMessage(); - onItem(workItem); + onItem(stream.Current); } } catch (OperationCanceledException) when (cancellation.IsCancellationRequested) diff --git a/src/Worker/Grpc/Worker.Grpc.csproj b/src/Worker/Grpc/Worker.Grpc.csproj index 2b5d4113..0fa7b856 100644 --- a/src/Worker/Grpc/Worker.Grpc.csproj +++ b/src/Worker/Grpc/Worker.Grpc.csproj @@ -22,6 +22,7 @@ + diff --git a/test/Grpc.IntegrationTests/PayloadHydrationFailureTests.cs b/test/Grpc.IntegrationTests/PayloadHydrationFailureTests.cs new file mode 100644 index 00000000..e3682722 --- /dev/null +++ b/test/Grpc.IntegrationTests/PayloadHydrationFailureTests.cs @@ -0,0 +1,520 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Collections.Concurrent; +using System.IO; +using System.Net; +using System.Text.Json; +using System.Threading.Channels; +using Azure; +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Grpc.Core.Interceptors; +using Grpc.Net.Client; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.Hosting.Server.Features; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.Grpc; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Moq; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Grpc.Tests; + +public class PayloadHydrationFailureTests +{ + const string TokenPrefix = "test-blob://"; + const string BadToken = TokenPrefix + "bad"; + const string FailedCompletionToken = "failed-delivery"; + static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(15); + + public static IEnumerable HydrationFailures() + { + P.WorkItem.RequestOneofCase[] kinds = + [ + P.WorkItem.RequestOneofCase.ActivityRequest, + P.WorkItem.RequestOneofCase.OrchestratorRequest, + P.WorkItem.RequestOneofCase.EntityRequest, + P.WorkItem.RequestOneofCase.EntityRequestV2, + ]; + + foreach (P.WorkItem.RequestOneofCase kind in kinds) + { + foreach (int storageStatus in new[] { 404, 403, 0 }) + { + foreach (bool ownershipLost in new[] { false, true }) + { + yield return [kind, storageStatus, ownershipLost]; + } + } + } + } + + [Theory] + [MemberData(nameof(HydrationFailures))] + public async Task HydrationFailure_AbandonsOnlyFailedDeliveryAndKeepsStreamAsync( + P.WorkItem.RequestOneofCase kind, int storageStatus, bool ownershipLost) + { + // Arrange + using CancellationTokenSource timeout = new(TestTimeout); + await using ScriptedSidecar sidecar = await ScriptedSidecar.StartAsync(); + sidecar.AbandonStatus = ownershipLost ? StatusCode.NotFound : StatusCode.OK; + Exception failure = storageStatus == 0 + ? new IOException("Transient payload download failure.") + : new RequestFailedException(storageStatus, "Payload download failed."); + Mock store = CreatePayloadStore(failure); + TaskCompletionSource running = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource release = new(TaskCreationOptions.RunContinuationsAsynchronously); + ConcurrentQueue inputs = new(); + await using ServiceProvider provider = CreateWorker(sidecar, store.Object, async input => + { + inputs.Enqueue(input); + if (input == "running") + { + running.TrySetResult(); + await release.Task.WaitAsync(timeout.Token); + } + + return input; + }); + IHostedService worker = Assert.Single(provider.GetServices()); + await worker.StartAsync(timeout.Token); + + try + { + // Act + sidecar.Send(CreateHealthyActivity("running")); + await running.Task.WaitAsync(timeout.Token); + sidecar.Send(CreateFailedWorkItem(kind)); + sidecar.Send(new P.WorkItem { HealthPing = new P.HealthPing() }); + sidecar.Send(CreateHealthyActivity("following")); + P.ActivityResponse following = await sidecar.Completions.Reader.ReadAsync(timeout.Token); + release.TrySetResult(); + P.ActivityResponse alreadyRunning = await sidecar.Completions.Reader.ReadAsync(timeout.Token); + + // Assert + Assert.Equal(1, sidecar.StreamCount); + Assert.Equal(0, sidecar.ClosedStreamCount); + Assert.Equal(1, sidecar.HelloCount); + P.WorkItem.RequestOneofCase abandonKind = kind == P.WorkItem.RequestOneofCase.EntityRequestV2 + ? P.WorkItem.RequestOneofCase.EntityRequest + : kind; + Assert.Equal((abandonKind, FailedCompletionToken), Assert.Single(sidecar.Abandons)); + Assert.Equal(new[] { "running", "following" }, inputs.ToArray()); + Assert.Equal("following", following.CompletionToken); + Assert.Equal("\"following\"", following.Result); + Assert.Null(following.FailureDetails); + Assert.Equal("running", alreadyRunning.CompletionToken); + Assert.Equal("\"running\"", alreadyRunning.Result); + Assert.Null(alreadyRunning.FailureDetails); + store.Verify(s => s.DownloadAsync(BadToken, It.IsAny()), Times.Once); + } + finally + { + release.TrySetResult(); + await worker.StopAsync(CancellationToken.None).WaitAsync(TestTimeout); + } + } + + [Theory] + [InlineData(StatusCode.Unknown, false)] + [InlineData(StatusCode.Unavailable, false)] + [InlineData(StatusCode.PermissionDenied, false)] + [InlineData(StatusCode.Internal, false)] + [InlineData(StatusCode.Unavailable, true)] + public async Task TransportOrAbandonFailure_ReconnectsAsync(StatusCode status, bool streamFailure) + { + // Arrange + using CancellationTokenSource timeout = new(TestTimeout); + await using ScriptedSidecar sidecar = await ScriptedSidecar.StartAsync(); + sidecar.AbandonStatus = status; + Mock store = CreatePayloadStore(new IOException("Payload download failed.")); + await using ServiceProvider provider = CreateWorker(sidecar, store.Object); + IHostedService worker = Assert.Single(provider.GetServices()); + await worker.StartAsync(timeout.Token); + + try + { + Assert.Equal(1, await sidecar.Streams.Reader.ReadAsync(timeout.Token)); + + // Act + if (streamFailure) + { + sidecar.FailStream(status); + } + else + { + sidecar.Send(CreateFailedWorkItem(P.WorkItem.RequestOneofCase.ActivityRequest)); + } + + Assert.Equal(2, await sidecar.Streams.Reader.ReadAsync(timeout.Token)); + sidecar.Send(CreateHealthyActivity("following")); + P.ActivityResponse response = await sidecar.Completions.Reader.ReadAsync(timeout.Token); + + // Assert + Assert.Equal(2, sidecar.StreamCount); + Assert.Equal(2, sidecar.HelloCount); + Assert.Equal("following", response.CompletionToken); + Assert.Null(response.FailureDetails); + Assert.Equal(streamFailure ? 0 : 1, sidecar.Abandons.Count); + } + finally + { + await worker.StopAsync(CancellationToken.None).WaitAsync(TestTimeout); + } + } + + [Fact] + public async Task HydrationFailure_RedeliveryCompletesAfterPayloadRecoversAsync() + { + // Arrange + using CancellationTokenSource timeout = new(TestTimeout); + await using ScriptedSidecar sidecar = await ScriptedSidecar.StartAsync(); + Mock store = CreatePayloadStore(new IOException("Payload download failed.")); + store.SetupSequence(s => s.DownloadAsync(BadToken, It.IsAny())) + .ThrowsAsync(new RequestFailedException(404, "Payload not found.")) + .ReturnsAsync("\"recovered\""); + await using ServiceProvider provider = CreateWorker(sidecar, store.Object); + IHostedService worker = Assert.Single(provider.GetServices()); + await worker.StartAsync(timeout.Token); + + try + { + // Act + sidecar.Send(CreateFailedWorkItem(P.WorkItem.RequestOneofCase.ActivityRequest)); + await sidecar.AbandonStarted.Task.WaitAsync(timeout.Token); + P.WorkItem redelivery = CreateFailedWorkItem(P.WorkItem.RequestOneofCase.ActivityRequest); + redelivery.CompletionToken = "redelivery"; + sidecar.Send(redelivery); + P.ActivityResponse response = await sidecar.Completions.Reader.ReadAsync(timeout.Token); + + // Assert + Assert.Equal(1, sidecar.StreamCount); + Assert.Equal((P.WorkItem.RequestOneofCase.ActivityRequest, FailedCompletionToken), Assert.Single(sidecar.Abandons)); + Assert.Equal("redelivery", response.CompletionToken); + Assert.Equal("\"recovered\"", response.Result); + Assert.Null(response.FailureDetails); + } + finally + { + await worker.StopAsync(CancellationToken.None).WaitAsync(TestTimeout); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ShutdownDuringHydrationOrAbandon_DoesNotReconnectAsync(bool duringAbandon) + { + // Arrange + using CancellationTokenSource timeout = new(TestTimeout); + await using ScriptedSidecar sidecar = await ScriptedSidecar.StartAsync(); + sidecar.HoldAbandon = duringAbandon; + TaskCompletionSource downloading = new(TaskCreationOptions.RunContinuationsAsynchronously); + Mock store = CreatePayloadStore(new IOException("Payload download failed.")); + if (!duringAbandon) + { + store.Setup(s => s.DownloadAsync(BadToken, It.IsAny())) + .Returns(async (string _, CancellationToken cancellation) => + { + downloading.TrySetResult(cancellation); + await Task.Delay(Timeout.Infinite, cancellation); + return string.Empty; + }); + } + + await using ServiceProvider provider = CreateWorker(sidecar, store.Object); + IHostedService worker = Assert.Single(provider.GetServices()); + await worker.StartAsync(timeout.Token); + + try + { + sidecar.Send(CreateFailedWorkItem(P.WorkItem.RequestOneofCase.ActivityRequest)); + CancellationToken downloadCancellation = default; + if (duringAbandon) + { + await sidecar.AbandonStarted.Task.WaitAsync(timeout.Token); + } + else + { + downloadCancellation = await downloading.Task.WaitAsync(timeout.Token); + } + + // Act + await worker.StopAsync(CancellationToken.None).WaitAsync(TestTimeout); + + // Assert + Assert.Equal(1, sidecar.StreamCount); + Assert.Equal(duringAbandon ? 1 : 0, sidecar.Abandons.Count); + Assert.False(sidecar.Completions.Reader.TryRead(out _)); + if (!duringAbandon) + { + Assert.True(downloadCancellation.IsCancellationRequested); + } + } + finally + { + await worker.StopAsync(CancellationToken.None).WaitAsync(TestTimeout); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task NonWorkItemHydrationFailure_PropagatesOriginalExceptionAsync(bool historyStream) + { + // Arrange + using CancellationTokenSource timeout = new(TestTimeout); + await using ScriptedSidecar sidecar = await ScriptedSidecar.StartAsync(); + RequestFailedException failure = new(403, "Payload download forbidden."); + Mock store = CreatePayloadStore(failure); + using GrpcChannel channel = GrpcChannel.ForAddress(sidecar.Address); + P.TaskHubSidecarService.TaskHubSidecarServiceClient client = new(channel.Intercept( + new AzureBlobPayloadsSideCarInterceptor(store.Object, new LargePayloadStorageOptions()))); + + // Act + Exception? observed; + if (historyStream) + { + using AsyncServerStreamingCall call = client.StreamInstanceHistory( + new P.StreamInstanceHistoryRequest { InstanceId = "instance" }, cancellationToken: timeout.Token); + observed = await Record.ExceptionAsync(() => call.ResponseStream.MoveNext(timeout.Token)); + } + else + { + using AsyncUnaryCall call = client.GetInstanceAsync( + new P.GetInstanceRequest { InstanceId = "instance" }, cancellationToken: timeout.Token); + observed = await Record.ExceptionAsync(async () => await call.ResponseAsync); + } + + // Assert + Assert.Same(failure, observed); + Assert.Empty(sidecar.Abandons); + } + + static Mock CreatePayloadStore(Exception failure) + { + Mock store = new(); + store.Setup(s => s.IsKnownPayloadToken(It.IsAny())) + .Returns((string value) => value.StartsWith(TokenPrefix, StringComparison.Ordinal)); + store.Setup(s => s.DownloadAsync(It.IsAny(), It.IsAny())) + .Returns((string token, CancellationToken _) => token == BadToken + ? Task.FromException(failure) + : Task.FromResult(JsonSerializer.Serialize(token[TokenPrefix.Length..]))); + return store; + } + + static ServiceProvider CreateWorker(ScriptedSidecar sidecar, PayloadStore store, Func>? runActivity = null) + { + ServiceCollection services = new(); + services.AddLogging(logging => logging.ClearProviders()); + services.AddSingleton(store); + services.AddDurableTaskWorker(builder => + { + builder.UseGrpc(options => + { + options.Address = sidecar.Address; + options.Internal.TransientRetryMaxAttempts = 1; + options.Internal.ReconnectBackoffBase = TimeSpan.Zero; + options.Internal.ReconnectBackoffCap = TimeSpan.Zero; + }); + builder.UseExternalizedPayloads(); + builder.AddTasks(tasks => tasks.AddActivityFunc( + "Healthy", (_, input) => runActivity?.Invoke(input!) ?? Task.FromResult(input!))); + }); + return services.BuildServiceProvider(); + } + + static P.WorkItem CreateHealthyActivity(string name) => new() + { + CompletionToken = name, + ActivityRequest = new P.ActivityRequest + { + Name = "Healthy", + TaskId = 1, + Input = TokenPrefix + name, + OrchestrationInstance = new P.OrchestrationInstance { InstanceId = name, ExecutionId = "execution" }, + }, + }; + + static P.WorkItem CreateFailedWorkItem(P.WorkItem.RequestOneofCase kind) + { + P.WorkItem item = new() { CompletionToken = FailedCompletionToken }; + switch (kind) + { + case P.WorkItem.RequestOneofCase.ActivityRequest: + item.ActivityRequest = CreateHealthyActivity("failed").ActivityRequest; + item.ActivityRequest.Input = BadToken; + break; + case P.WorkItem.RequestOneofCase.OrchestratorRequest: + item.OrchestratorRequest = new P.OrchestratorRequest + { + InstanceId = "failed", + PastEvents = { new P.HistoryEvent { ExecutionStarted = new P.ExecutionStartedEvent { Input = TokenPrefix + "state" } } }, + NewEvents = { new P.HistoryEvent { EventRaised = new P.EventRaisedEvent { Name = "event", Input = BadToken } } }, + }; + break; + case P.WorkItem.RequestOneofCase.EntityRequest: + item.EntityRequest = new P.EntityBatchRequest + { + InstanceId = "@entity@failed", + EntityState = TokenPrefix + "state", + Operations = { new P.OperationRequest { Operation = "operation", Input = BadToken } }, + }; + break; + case P.WorkItem.RequestOneofCase.EntityRequestV2: + item.EntityRequestV2 = new P.EntityRequest + { + InstanceId = "@entity@failed", + EntityState = TokenPrefix + "state", + OperationRequests = { new P.HistoryEvent { EntityOperationSignaled = new P.EntityOperationSignaledEvent { Input = BadToken } } }, + }; + break; + default: + throw new ArgumentOutOfRangeException(nameof(kind), kind, "Unsupported work item kind."); + } + + return item; + } + + sealed class ScriptedSidecar : P.TaskHubSidecarService.TaskHubSidecarServiceBase, IAsyncDisposable + { + readonly Channel, Task>> deliveries = + Channel.CreateUnbounded, Task>>(); + IHost host = null!; + int streamCount; + int closedStreamCount; + int helloCount; + + public string Address { get; private set; } = string.Empty; + + public StatusCode AbandonStatus { get; set; } + + public bool HoldAbandon { get; set; } + + public int StreamCount => Volatile.Read(ref this.streamCount); + + public int ClosedStreamCount => Volatile.Read(ref this.closedStreamCount); + + public int HelloCount => Volatile.Read(ref this.helloCount); + + public Channel Streams { get; } = Channel.CreateUnbounded(); + + public Channel Completions { get; } = Channel.CreateUnbounded(); + + public ConcurrentQueue<(P.WorkItem.RequestOneofCase Kind, string Token)> Abandons { get; } = new(); + + public TaskCompletionSource AbandonStarted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public static async Task StartAsync() + { + ScriptedSidecar sidecar = new(); + sidecar.host = Host.CreateDefaultBuilder() + .ConfigureLogging(logging => logging.ClearProviders()) + .ConfigureWebHostDefaults(web => web + .UseKestrel(options => options.Listen(IPAddress.Loopback, 0, endpoint => endpoint.Protocols = HttpProtocols.Http2)) + .ConfigureServices(services => + { + services.AddGrpc(); + services.AddSingleton(sidecar); + }) + .Configure(app => + { + app.UseRouting(); + app.UseEndpoints(endpoints => endpoints.MapGrpcService()); + })) + .Build(); + await sidecar.host.StartAsync().WaitAsync(TestTimeout); + sidecar.Address = Assert.Single(sidecar.host.Services.GetRequiredService() + .Features.Get()!.Addresses); + return sidecar; + } + + public void Send(P.WorkItem item) => Assert.True(this.deliveries.Writer.TryWrite(stream => stream.WriteAsync(item))); + + public void FailStream(StatusCode status) => Assert.True(this.deliveries.Writer.TryWrite( + _ => Task.FromException(new RpcException(new Status(status, "Stream transport failed."))))); + + public override Task Hello(Empty request, ServerCallContext context) + { + Interlocked.Increment(ref this.helloCount); + return Task.FromResult(new Empty()); + } + + public override async Task GetWorkItems(P.GetWorkItemsRequest request, IServerStreamWriter responseStream, ServerCallContext context) + { + await context.WriteResponseHeadersAsync(new Metadata()); + this.Streams.Writer.TryWrite(Interlocked.Increment(ref this.streamCount)); + try + { + await foreach (Func, Task> deliver in this.deliveries.Reader.ReadAllAsync(context.CancellationToken)) + { + await deliver(responseStream); + } + } + finally + { + Interlocked.Increment(ref this.closedStreamCount); + } + } + + public override Task CompleteActivityTask(P.ActivityResponse request, ServerCallContext context) + { + this.Completions.Writer.TryWrite(request); + return Task.FromResult(new P.CompleteTaskResponse()); + } + + public override async Task AbandonTaskActivityWorkItem(P.AbandonActivityTaskRequest request, ServerCallContext context) + { + await this.AbandonAsync(P.WorkItem.RequestOneofCase.ActivityRequest, request.CompletionToken, context); + return new P.AbandonActivityTaskResponse(); + } + + public override async Task AbandonTaskOrchestratorWorkItem(P.AbandonOrchestrationTaskRequest request, ServerCallContext context) + { + await this.AbandonAsync(P.WorkItem.RequestOneofCase.OrchestratorRequest, request.CompletionToken, context); + return new P.AbandonOrchestrationTaskResponse(); + } + + public override async Task AbandonTaskEntityWorkItem(P.AbandonEntityTaskRequest request, ServerCallContext context) + { + await this.AbandonAsync(P.WorkItem.RequestOneofCase.EntityRequest, request.CompletionToken, context); + return new P.AbandonEntityTaskResponse(); + } + + public override Task GetInstance(P.GetInstanceRequest request, ServerCallContext context) + => Task.FromResult(new P.GetInstanceResponse { OrchestrationState = new P.OrchestrationState { Input = BadToken } }); + + public override Task StreamInstanceHistory(P.StreamInstanceHistoryRequest request, IServerStreamWriter responseStream, ServerCallContext context) + => responseStream.WriteAsync(new P.HistoryChunk + { + Events = { new P.HistoryEvent { ExecutionStarted = new P.ExecutionStartedEvent { Input = BadToken } } }, + }); + + public async ValueTask DisposeAsync() + { + using IHost host = this.host; + await host.StopAsync().WaitAsync(TestTimeout); + } + + async Task AbandonAsync(P.WorkItem.RequestOneofCase kind, string token, ServerCallContext context) + { + this.Abandons.Enqueue((kind, token)); + this.AbandonStarted.TrySetResult(); + if (this.HoldAbandon) + { + await Task.Delay(Timeout.Infinite, context.CancellationToken); + } + + if (this.AbandonStatus != StatusCode.OK) + { + throw new RpcException(new Status(this.AbandonStatus, "Abandon failed.")); + } + } + } +} diff --git a/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs b/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs index ba9104c1..330af913 100644 --- a/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs +++ b/test/Worker/Grpc.Tests/WorkItemStreamConsumerTests.cs @@ -1,7 +1,6 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -using System.Runtime.CompilerServices; using Grpc.Core; using Microsoft.DurableTask.Worker.Grpc; using P = Microsoft.DurableTask.Protobuf; @@ -16,7 +15,7 @@ public class WorkItemStreamConsumerTests public async Task EmptyStream_ReturnsGracefulDrain() { WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - openStream: _ => EmptyStream(), + stream: EmptyStream(), silentDisconnectTimeout: TimeSpan.FromSeconds(5), onItem: _ => throw new InvalidOperationException("onItem should not be invoked"), onFirstMessage: () => throw new InvalidOperationException("onFirstMessage should not be invoked"), @@ -35,7 +34,7 @@ public async Task StreamWithItems_ReturnsGracefulDrain_AndFiresCallbacks() int firstMessageCount = 0; WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - openStream: _ => StreamOf(item1, item2), + stream: StreamOf(item1, item2), silentDisconnectTimeout: TimeSpan.FromSeconds(5), onItem: received.Add, onFirstMessage: () => firstMessageCount++, @@ -51,7 +50,7 @@ public async Task StreamWithItems_ReturnsGracefulDrain_AndFiresCallbacks() public async Task VeryLargeSilentDisconnectTimeout_IsClamped_AndStreamCanStillComplete() { WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - openStream: _ => EmptyStream(), + stream: EmptyStream(), silentDisconnectTimeout: TimeSpan.FromDays(365), onItem: _ => throw new InvalidOperationException("onItem should not be invoked"), onFirstMessage: null, @@ -65,7 +64,7 @@ public async Task VeryLargeSilentDisconnectTimeout_IsClamped_AndStreamCanStillCo public async Task HangingStream_SurfacingOce_ReturnsSilentDisconnect() { WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - openStream: ct => HangingStream(ct, throwAsRpc: false), + stream: new HangingStreamReader(throwAsRpc: false), silentDisconnectTimeout: ShortTimeout, onItem: _ => { }, onFirstMessage: null, @@ -85,7 +84,7 @@ public async Task HangingStream_SurfacingOce_ReturnsSilentDisconnect() public async Task HangingStream_SurfacingRpcCancelled_ReturnsSilentDisconnect() { WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - openStream: ct => HangingStream(ct, throwAsRpc: true), + stream: new HangingStreamReader(throwAsRpc: true), silentDisconnectTimeout: ShortTimeout, onItem: _ => { }, onFirstMessage: null, @@ -104,7 +103,7 @@ public async Task OuterCancellation_WithOceFromStream_ReturnsShutdown() outer.CancelAfter(ShortTimeout); WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - openStream: ct => HangingStream(ct, throwAsRpc: false), + stream: new HangingStreamReader(throwAsRpc: false), silentDisconnectTimeout: TimeSpan.FromSeconds(30), onItem: _ => { }, onFirstMessage: null, @@ -126,7 +125,7 @@ public async Task OuterCancellation_WithRpcCancelledFromStream_PropagatesExcepti outer.CancelAfter(ShortTimeout); Func act = () => WorkItemStreamConsumer.ConsumeAsync( - openStream: ct => HangingStream(ct, throwAsRpc: true), + stream: new HangingStreamReader(throwAsRpc: true), silentDisconnectTimeout: TimeSpan.FromSeconds(30), onItem: _ => { }, onFirstMessage: null, @@ -165,7 +164,7 @@ public async Task PerItem_HeartbeatReset_KeepsTimerAlive() } WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - openStream: _ => StreamOf(items), + stream: StreamOf(items), silentDisconnectTimeout: TimeSpan.FromMilliseconds(500), onItem: _ => events.Add($"item{itemIndex++}"), onFirstMessage: null, @@ -193,7 +192,7 @@ public async Task PerItem_HeartbeatReset_KeepsTimerAlive() public async Task UnrelatedRpcException_Propagates() { Func act = () => WorkItemStreamConsumer.ConsumeAsync( - openStream: _ => ThrowingStream(new RpcException(new Status(StatusCode.Unavailable, "boom"))), + stream: ThrowingStream(new RpcException(new Status(StatusCode.Unavailable, "boom"))), silentDisconnectTimeout: TimeSpan.FromSeconds(5), onItem: _ => { }, onFirstMessage: null, @@ -213,7 +212,7 @@ public async Task NonPositiveSilentDisconnectTimeout_OnlyShutdownEndsLoop(int ti // Act WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( - openStream: ct => HangingStream(ct, throwAsRpc: false), + stream: new HangingStreamReader(throwAsRpc: false), silentDisconnectTimeout: TimeSpan.FromMilliseconds(timeoutMilliseconds), onItem: _ => { }, onFirstMessage: null, @@ -223,55 +222,138 @@ public async Task NonPositiveSilentDisconnectTimeout_OnlyShutdownEndsLoop(int ti result.Outcome.Should().Be(WorkItemStreamOutcome.Shutdown); } -#pragma warning disable CS1998 // Async method lacks 'await' operators - static async IAsyncEnumerable EmptyStream() + [Fact] + public async Task HydrationFailure_RearmsTimerAndRecoversBeforeReadingNextItemAsync() { - yield break; + // Arrange + P.WorkItem failed = new() + { + CompletionToken = "failed-token", + OrchestratorRequest = new P.OrchestratorRequest + { + NewEvents = { new P.HistoryEvent { EventRaised = new P.EventRaisedEvent { Input = "partially resolved" } } }, + }, + }; + WorkItemPayloadException failure = new(failed, new InvalidOperationException("Download failed.")); + P.WorkItem healthy = new() { HealthPing = new P.HealthPing() }; + Mock> reader = new(MockBehavior.Strict); + reader.SetupSequence(r => r.MoveNext(It.IsAny())) + .ThrowsAsync(failure) + .ReturnsAsync(true) + .ReturnsAsync(false); + reader.SetupGet(r => r.Current).Returns(healthy); + List events = new(); + List dispatched = new(); + + // Act + WorkItemStreamResult result = await WorkItemStreamConsumer.ConsumeAsync( + stream: reader.Object, + silentDisconnectTimeout: TimeSpan.FromSeconds(5), + onItem: item => + { + events.Add("dispatch"); + dispatched.Add(item); + }, + onFirstMessage: () => events.Add("first"), + cancellation: CancellationToken.None, + onSilentDisconnectTimerArmed: () => events.Add("armed"), + onPayloadFailure: (exception, cancellation) => + { + Assert.Same(failure, exception); + Assert.Same(failed, exception.WorkItem); + Assert.True(cancellation.CanBeCanceled); + events.Add("abandon"); + return Task.CompletedTask; + }); + + // Assert + Assert.Equal(WorkItemStreamOutcome.GracefulDrain, result.Outcome); + Assert.True(result.FirstMessageObserved); + Assert.Equal(new[] { "armed", "armed", "first", "abandon", "armed", "dispatch" }, events); + Assert.Same(healthy, Assert.Single(dispatched)); + reader.VerifyGet(r => r.Current, Times.Once); + reader.Verify(r => r.MoveNext(It.IsAny()), Times.Exactly(3)); } - static async IAsyncEnumerable StreamOf(params P.WorkItem[] items) + [Theory] + [InlineData(StatusCode.Unavailable)] + [InlineData(StatusCode.PermissionDenied)] + public async Task HydrationRecoveryFailure_PropagatesForConnectionRecoveryAsync(StatusCode status) { - foreach (P.WorkItem item in items) - { - yield return item; - } + // Arrange + WorkItemPayloadException failure = new(new P.WorkItem(), new InvalidOperationException("Download failed.")); + RpcException abandonFailure = new(new Status(status, "Abandon failed.")); + + // Act + Func act = () => WorkItemStreamConsumer.ConsumeAsync( + stream: ThrowingStream(failure), + silentDisconnectTimeout: TimeSpan.FromSeconds(5), + onItem: _ => throw new InvalidOperationException("Failed delivery must not be dispatched."), + onFirstMessage: null, + cancellation: CancellationToken.None, + onPayloadFailure: (_, _) => Task.FromException(abandonFailure)); + + // Assert + (await act.Should().ThrowAsync()).Which.Should().BeSameAs(abandonFailure); } - static IAsyncEnumerable ThrowingStream(Exception ex) => new ThrowingAsyncEnumerable(ex); -#pragma warning restore CS1998 + [Fact] + public async Task HydrationFailure_WithoutRecoveryHandler_PropagatesAsync() + { + // Arrange + WorkItemPayloadException failure = new(new P.WorkItem(), new InvalidOperationException("Download failed.")); + + // Act + Func act = () => WorkItemStreamConsumer.ConsumeAsync( + stream: ThrowingStream(failure), + silentDisconnectTimeout: TimeSpan.FromSeconds(5), + onItem: _ => throw new InvalidOperationException("Failed delivery must not be dispatched."), + onFirstMessage: null, + cancellation: CancellationToken.None); + + // Assert + (await act.Should().ThrowAsync()).Which.Should().BeSameAs(failure); + } + + static IAsyncStreamReader EmptyStream() => StreamOf(); - static async IAsyncEnumerable HangingStream( - [EnumeratorCancellation] CancellationToken ct, - bool throwAsRpc) + static IAsyncStreamReader StreamOf(params P.WorkItem[] items) => new SequenceStreamReader(items); + + static IAsyncStreamReader ThrowingStream(Exception ex) => new ThrowingStreamReader(ex); + + sealed class SequenceStreamReader(P.WorkItem[] items) : IAsyncStreamReader { - try - { - await Task.Delay(Timeout.Infinite, ct); - } - catch (OperationCanceledException) when (throwAsRpc) - { - // Mimic grpc-dotnet's default surface shape for linked-token cancellation. - throw new RpcException(new Status(StatusCode.Cancelled, "stream cancelled")); - } + int index = -1; - yield break; + public P.WorkItem Current => items[this.index]; + + public Task MoveNext(CancellationToken cancellationToken) => Task.FromResult(++this.index < items.Length); } - sealed class ThrowingAsyncEnumerable : IAsyncEnumerable, IAsyncEnumerator + sealed class HangingStreamReader(bool throwAsRpc) : IAsyncStreamReader { - readonly Exception exception; + public P.WorkItem Current => throw new InvalidOperationException("No item is available for a hanging stream."); - public ThrowingAsyncEnumerable(Exception exception) + public async Task MoveNext(CancellationToken cancellationToken) { - this.exception = exception; + try + { + await Task.Delay(Timeout.Infinite, cancellationToken); + } + catch (OperationCanceledException) when (throwAsRpc) + { + // Mimic grpc-dotnet's default surface shape for linked-token cancellation. + throw new RpcException(new Status(StatusCode.Cancelled, "stream cancelled")); + } + + return false; } + } + sealed class ThrowingStreamReader(Exception exception) : IAsyncStreamReader + { public P.WorkItem Current => throw new InvalidOperationException("No current item is available for a throwing stream."); - public IAsyncEnumerator GetAsyncEnumerator(CancellationToken cancellationToken = default) => this; - - public ValueTask DisposeAsync() => default; - - public ValueTask MoveNextAsync() => ValueTask.FromException(this.exception); + public Task MoveNext(CancellationToken cancellationToken) => Task.FromException(exception); } }