From 86d24351ef052e9f8f4a894835f1caeb882868d9 Mon Sep 17 00:00:00 2001 From: wangbill Date: Fri, 16 Oct 2026 14:58:10 -0400 Subject: [PATCH 1/2] Fix payload retries and entity cleanup after storage failures Skip recognized payload tokens before size checks so completion retries remain idempotent. Prepare orchestration payloads before terminal cleanup and chunking, preserve entity release actions, and reuse outstanding unlock obligations. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 78dff439-4d84-4d70-921c-f57421cbc383 --- CHANGELOG.md | 1 + ...rkerBuilderExtensions.AzureBlobPayloads.cs | 7 +- .../AzureBlobPayloadsSideCarInterceptor.cs | 76 +++-- .../Interceptors/PayloadInterceptor.cs | 8 +- src/Shared/Grpc/ProtoUtils.cs | 53 ++-- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 7 + .../Grpc/GrpcDurableTaskWorkerOptions.cs | 6 + src/Worker/Grpc/Worker.Grpc.csproj | 1 + .../OrchestratorPayloadFailureTests.cs | 270 ++++++++++++++++++ ...zureBlobPayloadsSideCarInterceptorTests.cs | 98 +++++++ .../LargePayloadTests.cs | 99 ++++++- 11 files changed, 565 insertions(+), 61 deletions(-) create mode 100644 test/Extensions/AzureBlobPayloads.Tests/OrchestratorPayloadFailureTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 6900d0788..2e26415f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Keep externalized payload tokens stable across completion retries and release entity locks when payload storage permanently fails an orchestration. - Preserve the original orchestration version when restarting through the orchestration-service client shim ([#463](https://github.com/microsoft/durabletask-dotnet/issues/463)). - Support configurable scheduler token audiences and government defaults ([#806](https://github.com/microsoft/durabletask-dotnet/pull/806)) diff --git a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs index dafe5784f..2777e360f 100644 --- a/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs +++ b/src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs @@ -100,7 +100,12 @@ static IDurableTaskWorkerBuilder UseExternalizedPayloadsCore(IDurableTaskWorkerB // Register an interceptor rather than moving Channel onto an intercepted CallInvoker. // Clearing Channel would disable the worker's gRPC channel recreation, and requiring a // pre-built Channel/CallInvoker would rule out the Address-only configuration. - opt.Interceptors.Add(new AzureBlobPayloadsSideCarInterceptor(store, opts)); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, opts); + opt.Interceptors.Add(interceptor); + + // Storage failures must become terminal responses before the worker releases entity locks. + // Unary interception still handles other RPCs; known tokens make the second pass a no-op. + opt.Internal.PrepareOrchestratorResponseAsync = interceptor.PrepareOrchestratorResponseAsync; opt.Capabilities.Add(P.WorkerCapability.LargePayloads); opt.ConfigureVersioningExemptions([nameof(BlobPurgeJobOrchestrator)], AutoPurgeActivityNames); diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 29fd76f59..da297c995 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -76,37 +76,7 @@ protected override async Task ExternalizeRequestPayloadsAsync(TRequest break; case P.OrchestratorResponse r: - try - { - await this.ExternalizeOrchestratorResponseAsync(r, cancellation); - } - catch (Exception ex) when (IsPermanentStorageFailure(ex)) - { - // Permanent failure during orchestration response externalization. - // Replace all actions with a single Failed completion so the orchestration - // terminates instead of being abandoned and redelivered indefinitely. - r.Actions.Clear(); - r.CustomStatus = null; -#pragma warning disable CS0612 // isPartial/chunkIndex are deprecated but still required for chunked response wire compatibility. - r.IsPartial = false; - r.ChunkIndex = null; -#pragma warning restore CS0612 - r.Actions.Add(new P.OrchestratorAction - { - CompleteOrchestration = new P.CompleteOrchestrationAction - { - OrchestrationStatus = P.OrchestrationStatus.Failed, - FailureDetails = new P.TaskFailureDetails - { - ErrorType = ex.GetType().FullName, - ErrorMessage = ex.Message, - StackTrace = ex.StackTrace, - IsNonRetriable = true, - }, - }, - }); - } - + await this.PrepareOrchestratorResponseAsync(r, cancellation); break; case P.EntityBatchResult r: await this.ExternalizeEntityBatchResultAsync(r, cancellation); @@ -264,6 +234,50 @@ protected override async Task ResolveResponsePayloadsAsync(TResponse } } + /// + /// Externalizes a response before the worker finalizes terminal entity cleanup. + /// + /// The orchestration response. + /// Cancellation token. + /// A task representing response preparation. + internal async Task PrepareOrchestratorResponseAsync(P.OrchestratorResponse response, CancellationToken cancellation) + { + try + { + await this.ExternalizeOrchestratorResponseAsync(response, cancellation); + } + catch (Exception ex) when (IsPermanentStorageFailure(ex)) + { + // Retain entity releases when replacing actions with a Failed completion so + // the orchestration terminates without leaving its entities locked. + P.OrchestratorAction[] unlocks = response.Actions.Where(action => + action.SendEntityMessage?.EntityUnlockSent is not null + || (action.SendEvent is { Name: "release", Instance: { } instance } + && global::DurableTask.Core.Common.Entities.IsEntityInstance(instance.InstanceId))).ToArray(); + response.Actions.Clear(); + response.Actions.AddRange(unlocks); + response.CustomStatus = null; +#pragma warning disable CS0612 // isPartial/chunkIndex are deprecated but still required for chunked response wire compatibility. + response.IsPartial = false; + response.ChunkIndex = null; +#pragma warning restore CS0612 + response.Actions.Add(new P.OrchestratorAction + { + CompleteOrchestration = new P.CompleteOrchestrationAction + { + OrchestrationStatus = P.OrchestrationStatus.Failed, + FailureDetails = new P.TaskFailureDetails + { + ErrorType = ex.GetType().FullName, + ErrorMessage = ex.Message, + StackTrace = ex.StackTrace, + IsNonRetriable = true, + }, + }, + }); + } + } + void AddOrchestrationStateResolutionOperations( List> operations, P.OrchestrationState state, diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs index 3274229cb..76b9fb5f0 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs @@ -174,7 +174,7 @@ public override AsyncServerStreamingCall AsyncServerStreamingCall(TResponse response, CancellationToken cancellation); /// - /// Externalizes a payload if it exceeds the threshold. + /// Externalizes a payload if it exceeds the threshold and is not already a known payload token. /// /// The value to potentially externalize. /// Cancellation token. @@ -232,11 +232,11 @@ private protected async Task ExternalizePayloadAsync( /// Calculates a value's UTF-8 size and determines whether it requires externalization. /// /// The value to inspect. - /// The calculated UTF-8 size, or zero when is empty. - /// true when the value meets the configured externalization threshold. + /// The calculated UTF-8 size, or zero when is empty or a known payload token. + /// true when a non-token value meets the configured externalization threshold. private protected bool TryGetExternalizationSize(string? value, out int size) { - if (string.IsNullOrEmpty(value)) + if (string.IsNullOrEmpty(value) || this.payloadStore.IsKnownPayloadToken(value!)) { size = 0; return false; diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs index e3691664d..02b135f6c 100644 --- a/src/Shared/Grpc/ProtoUtils.cs +++ b/src/Shared/Grpc/ProtoUtils.cs @@ -442,29 +442,6 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( break; case OrchestratorActionType.OrchestrationComplete: - - if (entityConversionState is not null) - { - // as a precaution, unlock any entities that were not unlocked for some reason, before - // completing the orchestration. - foreach ((string target, string criticalSectionId) in entityConversionState.ResetObligations()) - { - response.Actions.Add(new P.OrchestratorAction - { - Id = action.Id, - SendEntityMessage = new P.SendEntityMessageAction - { - EntityUnlockSent = new P.EntityUnlockSentEvent - { - CriticalSectionId = criticalSectionId, - TargetInstanceId = target, - ParentInstanceId = entityConversionState.CurrentInstance?.InstanceId, - }, - }, - }); - } - } - var completeAction = (OrchestrationCompleteOrchestratorAction)action; protoAction.CompleteOrchestration = new P.CompleteOrchestrationAction { @@ -491,6 +468,10 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( } response.Actions.Add(protoAction); + if (protoAction.CompleteOrchestration is not null) + { + entityConversionState?.InsertMissingUnlocks(response, protoAction); + } } return response; @@ -1291,6 +1272,32 @@ public void RemoveUnlockObligation(string target) this.unlockObligations?.Remove(target); } + /// + /// Inserts any outstanding entity releases before a terminal completion. + /// + /// The response containing the completion. + /// The completion action already added to the response. + public void InsertMissingUnlocks(P.OrchestratorResponse response, P.OrchestratorAction completionAction) + { + int index = response.Actions.IndexOf(completionAction); + foreach ((string target, string criticalSectionId) in this.ResetObligations()) + { + response.Actions.Insert(index++, new P.OrchestratorAction + { + Id = completionAction.Id, + SendEntityMessage = new P.SendEntityMessageAction + { + EntityUnlockSent = new P.EntityUnlockSentEvent + { + CriticalSectionId = criticalSectionId, + TargetInstanceId = target, + ParentInstanceId = this.CurrentInstance?.InstanceId, + }, + }, + }); + } + } + /// /// Returns the remaining unlock obligations, and clears the list. /// diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index abfba18a4..5eba66c52 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -979,11 +979,18 @@ await this.ExecuteWithRetryAsync( }; } + if (this.internalOptions.PrepareOrchestratorResponseAsync is { } prepareResponse) + { + await prepareResponse(response, cancellationToken); + } + var completeOrchestrationAction = response.Actions.FirstOrDefault( a => a.CompleteOrchestration is not null); if (completeOrchestrationAction is not null) { + entityConversionState?.InsertMissingUnlocks(response, completeOrchestrationAction); + if (completeOrchestrationAction.CompleteOrchestration.OrchestrationStatus == P.OrchestrationStatus.Failed) { traceActivity?.SetStatus( diff --git a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs index 0134a9597..0b026a0f0 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs @@ -135,6 +135,12 @@ internal class InternalOptions /// public bool InsertEntityUnlocksOnCompletion { get; set; } + /// + /// Gets or sets optional payload preparation that runs before terminal entity cleanup and chunking. + /// Permanent payload failures can turn a pending response into a terminal completion. + /// + public Func? PrepareOrchestratorResponseAsync { get; set; } + /// /// Gets or sets the maximum amount of time to wait for the Hello handshake with the backend before /// treating the connect attempt as failed and retrying. A non-positive value disables the deadline. diff --git a/src/Worker/Grpc/Worker.Grpc.csproj b/src/Worker/Grpc/Worker.Grpc.csproj index 2b5d41138..0ffbe891d 100644 --- a/src/Worker/Grpc/Worker.Grpc.csproj +++ b/src/Worker/Grpc/Worker.Grpc.csproj @@ -22,6 +22,7 @@ + diff --git a/test/Extensions/AzureBlobPayloads.Tests/OrchestratorPayloadFailureTests.cs b/test/Extensions/AzureBlobPayloads.Tests/OrchestratorPayloadFailureTests.cs new file mode 100644 index 000000000..53d7c96ef --- /dev/null +++ b/test/Extensions/AzureBlobPayloads.Tests/OrchestratorPayloadFailureTests.cs @@ -0,0 +1,270 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Grpc.Core; +using Microsoft.DurableTask.Entities; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.Grpc; +using Microsoft.DurableTask.Worker.Grpc.Internal; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Extensions.AzureBlobPayloads.Tests; + +public sealed class OrchestratorPayloadFailureTests +{ + [Theory] + [InlineData(false, false)] + [InlineData(true, false)] + [InlineData(false, true)] + [InlineData(true, true)] + public async Task PermanentPayloadFailure_ReleasesEntitiesBeforeSendingFailedCompletion( + bool completeOrchestration, + bool streamHistory) + { + // Arrange + const string InstanceId = "locked-payload-instance"; + const string ExecutionId = "execution-1"; + const string OrchestratorName = "LockedPayloadOrchestrator"; + EntityInstanceId[] entities = [new("counter", "one"), new("counter", "two")]; + string oversizedPayload = new('x', GrpcDurableTaskWorkerOptions.MinCompleteOrchestrationWorkItemChunkSizeInBytes + 1024); + Channel workItems = Channel.CreateUnbounded(); + Channel history = Channel.CreateUnbounded(); + Channel completions = Channel.CreateUnbounded(); + Mock invoker = new(MockBehavior.Strict); + invoker.Setup(value => value.AsyncUnaryCall( + It.IsAny>(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(() => CreateUnaryCall(new Empty())); + invoker.Setup(value => value.AsyncServerStreamingCall( + It.IsAny>(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(() => new AsyncServerStreamingCall( + new ChannelStreamReader(workItems.Reader), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { })); + invoker.Setup(value => value.AsyncServerStreamingCall( + It.IsAny>(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(() => new AsyncServerStreamingCall( + new ChannelStreamReader(history.Reader), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { })); + invoker.Setup(value => value.AsyncUnaryCall( + It.IsAny>(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback, string, CallOptions, P.OrchestratorResponse>( + (_, _, _, response) => completions.Writer.TryWrite(response.Clone())) + .Returns(() => CreateUnaryCall(new P.CompleteTaskResponse())); + Mock store = new(MockBehavior.Strict); + store.Setup(value => value.IsKnownPayloadToken(It.IsAny())).Returns(false); + ServiceCollection services = new(); + services.AddSingleton(NullLoggerFactory.Instance); + services.AddSingleton(typeof(ILogger<>), typeof(NullLogger<>)); + services.AddSingleton(store.Object); + services.AddDurableTaskWorker(builder => + { + builder.Configure(options => options.EnableEntitySupport = true); + builder.UseGrpc(options => + { + options.CallInvoker = invoker.Object; + options.ConfigureForAzureManaged(); + options.CompleteOrchestrationWorkItemChunkSizeInBytes = + GrpcDurableTaskWorkerOptions.MinCompleteOrchestrationWorkItemChunkSizeInBytes; + }); + builder.UseExternalizedPayloads(options => + { + options.ThresholdBytes = 1024; + options.MaxPayloadBytes = 2048; + }); + builder.AddTasks(tasks => tasks.AddOrchestratorFunc( + OrchestratorName, + async (context, _) => + { + await using IAsyncDisposable locks = await context.Entities.LockEntitiesAsync(entities); + if (!completeOrchestration) + { + await context.Entities.CallEntityAsync(entities[0], "set", oversizedPayload); + } + + return oversizedPayload; + })); + }); + await using ServiceProvider provider = services.BuildServiceProvider(); + IHostedService worker = Assert.Single(provider.GetServices()); + using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(30)); + DateTime timestamp = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + P.OrchestratorRequest initialRequest = new() + { + InstanceId = InstanceId, + ExecutionId = ExecutionId, + EntityParameters = new P.OrchestratorEntityParameters + { + EntityMessageReorderWindow = Duration.FromTimeSpan(TimeSpan.Zero), + }, + NewEvents = + { + new P.HistoryEvent + { + EventId = -1, + Timestamp = Timestamp.FromDateTime(timestamp), + OrchestratorStarted = new P.OrchestratorStartedEvent(), + }, + new P.HistoryEvent + { + EventId = -1, + Timestamp = Timestamp.FromDateTime(timestamp), + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = OrchestratorName, + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = InstanceId, + ExecutionId = ExecutionId, + }, + }, + }, + }, + }; + await worker.StartAsync(timeout.Token); + try + { + await workItems.Writer.WriteAsync(new P.WorkItem + { + OrchestratorRequest = initialRequest, + CompletionToken = "acquire-token", + }, timeout.Token); + P.OrchestratorResponse acquiring = await completions.Reader.ReadAsync(timeout.Token); + P.OrchestratorAction acquireAction = Assert.Single(acquiring.Actions); + Assert.True(acquireAction.SendEntityMessage is not null, acquiring.ToString()); + P.EntityLockRequestedEvent acquire = acquireAction.SendEntityMessage.EntityLockRequested; + Assert.NotNull(acquire); + P.OrchestratorRequest resumedRequest = new() + { + InstanceId = InstanceId, + ExecutionId = ExecutionId, + EntityParameters = initialRequest.EntityParameters.Clone(), + PastEvents = + { + initialRequest.NewEvents, + new P.HistoryEvent + { + EventId = acquireAction.Id, + Timestamp = Timestamp.FromDateTime(timestamp), + EntityLockRequested = acquire.Clone(), + }, + new P.HistoryEvent + { + EventId = -1, + Timestamp = Timestamp.FromDateTime(timestamp), + OrchestratorCompleted = new P.OrchestratorCompletedEvent(), + }, + }, + NewEvents = + { + new P.HistoryEvent + { + EventId = -1, + Timestamp = Timestamp.FromDateTime(timestamp.AddSeconds(1)), + OrchestratorStarted = new P.OrchestratorStartedEvent(), + }, + new P.HistoryEvent + { + EventId = -1, + Timestamp = Timestamp.FromDateTime(timestamp.AddSeconds(1)), + EntityLockGranted = new P.EntityLockGrantedEvent + { + CriticalSectionId = acquire.CriticalSectionId, + }, + }, + }, + }; + if (streamHistory) + { + await history.Writer.WriteAsync( + new P.HistoryChunk { Events = { resumedRequest.PastEvents } }, timeout.Token); + history.Writer.Complete(); + resumedRequest.PastEvents.Clear(); + resumedRequest.RequiresHistoryStreaming = true; + } + + // Act + await workItems.Writer.WriteAsync(new P.WorkItem + { + OrchestratorRequest = resumedRequest, + CompletionToken = "resumed-token", + }, timeout.Token); + P.OrchestratorResponse failed = await completions.Reader.ReadAsync(timeout.Token); + + // Assert + P.CompleteOrchestrationAction completion = failed.Actions.Last().CompleteOrchestration; + Assert.NotNull(completion); + Assert.Equal(P.OrchestrationStatus.Failed, completion.OrchestrationStatus); + Assert.Equal(typeof(PayloadStorageException).FullName, completion.FailureDetails.ErrorType); + Assert.True(completion.FailureDetails.IsNonRetriable); + Assert.Equal(entities.Length + 1, failed.Actions.Count); + P.EntityUnlockSentEvent[] unlocks = failed.Actions.Take(entities.Length) + .Select(action => action.SendEntityMessage.EntityUnlockSent).ToArray(); + Assert.Equal(entities.Select(entity => entity.ToString()), unlocks.Select(unlock => unlock.TargetInstanceId)); + Assert.All(unlocks, unlock => + { + Assert.Equal(acquire.CriticalSectionId, unlock.CriticalSectionId); + Assert.Equal(InstanceId, unlock.ParentInstanceId); + }); + Assert.Equal("resumed-token", failed.CompletionToken); + invoker.Verify(value => value.AsyncUnaryCall( + It.IsAny>(), + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Exactly(2)); + store.Verify(value => value.UploadAsync(It.IsAny(), It.IsAny()), Times.Never); + } + finally + { + await worker.StopAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(10)); + } + } + + static AsyncUnaryCall CreateUnaryCall(T response) + => new( + Task.FromResult(response), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { }); + + sealed class ChannelStreamReader(ChannelReader reader) : IAsyncStreamReader + { + public T Current { get; private set; } = default!; + + public async Task MoveNext(CancellationToken cancellationToken) + { + if (await reader.WaitToReadAsync(cancellationToken)) + { + this.Current = await reader.ReadAsync(cancellationToken); + return true; + } + + return false; + } + } +} diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index c8135ef6e..9c6a765ec 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -456,6 +456,24 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Externali store.MaxObservedConcurrency.Should().BeLessOrEqualTo(8, "concurrency must be bounded to avoid Azure Storage throttling"); } + [Fact] + public async Task ExternalizeRequestPayloadsAsync_CreateInstanceRequest_KnownTokenBypassesSizeLimits() + { + // Arrange + TrackingPayloadStore store = new(); + string token = store.Seed("original input"); + LargePayloadStorageOptions options = new() { ThresholdBytes = 1, MaxPayloadBytes = 1 }; + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, options); + P.CreateInstanceRequest request = new() { Input = token }; + + // Act + await ExternalizeAsync(interceptor, request, CancellationToken.None); + + // Assert + request.Input.Should().Be(token); + store.UploadCount.Should().Be(0); + } + [Fact] public async Task ExternalizeRequestPayloadsAsync_RewindInstanceRequest_ExternalizesReason() { @@ -520,6 +538,7 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Externali // Act await ExternalizeAsync(interceptor, response, CancellationToken.None); + await ExternalizeAsync(interceptor, response, CancellationToken.None); // Assert IReadOnlyList tokens = GetHistoryEventPayloads(historyEvent); @@ -587,6 +606,7 @@ public async Task ExternalizeRequestPayloadsAsync_EntityBatchRequest_Externalize // Act await ExternalizeAsync(interceptor, request, CancellationToken.None); + await ExternalizeAsync(interceptor, request, CancellationToken.None); // Assert store.GetUploadedValue(request.EntityState).Should().Be("state-0"); @@ -596,6 +616,8 @@ public async Task ExternalizeRequestPayloadsAsync_EntityBatchRequest_Externalize token.Should().NotBe(expectedInputs[i]); store.GetUploadedValue(token).Should().Be(expectedInputs[i]); } + + store.UploadCount.Should().Be(operationCount + 1); } [Fact] @@ -628,6 +650,7 @@ public async Task ExternalizeRequestPayloadsAsync_EntityBatchResult_Externalizes // Act await ExternalizeAsync(interceptor, result, CancellationToken.None); + await ExternalizeAsync(interceptor, result, CancellationToken.None); // Assert store.GetUploadedValue(result.EntityState).Should().Be("state-final"); @@ -638,6 +661,7 @@ public async Task ExternalizeRequestPayloadsAsync_EntityBatchResult_Externalizes store.GetUploadedValue(result.Actions[0].SendSignal.Input).Should().Be("signal-input"); store.GetUploadedValue(result.Actions[1].StartNewOrchestration.Input).Should().Be("start-input"); + store.UploadCount.Should().Be(resultCount + 3); } [Fact] @@ -699,6 +723,80 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Permanent #pragma warning restore CS0612 } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_PermanentFailure_PreservesEntityUnlocks( + bool useLegacyEntityEvents) + { + // Arrange + TrackingPayloadStore store = new(); + LargePayloadStorageOptions options = new() { ThresholdBytes = 1024, MaxPayloadBytes = 2048 }; + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, options); + P.OrchestratorAction unlock = new() { Id = 7 }; + if (useLegacyEntityEvents) + { + unlock.SendEvent = new P.SendEventAction + { + Instance = new P.OrchestrationInstance { InstanceId = "@counter@one" }, + Name = "release", + Data = """{"id":"11111111-1111-1111-1111-111111111111","parent":"instance-1"}""", + }; + } + else + { + unlock.SendEntityMessage = new P.SendEntityMessageAction + { + EntityUnlockSent = new P.EntityUnlockSentEvent + { + CriticalSectionId = "11111111-1111-1111-1111-111111111111", + ParentInstanceId = "instance-1", + TargetInstanceId = "@counter@one", + }, + }; + } + + P.OrchestratorResponse response = new() + { + InstanceId = "instance-1", + CompletionToken = "completion-token", + Actions = + { + unlock, + new P.OrchestratorAction + { + Id = 8, + SendEvent = new P.SendEventAction + { + Instance = new P.OrchestrationInstance { InstanceId = "another-orchestration" }, + Name = "release", + Data = "discarded", + }, + }, + new P.OrchestratorAction + { + Id = 9, + CompleteOrchestration = new P.CompleteOrchestrationAction + { + OrchestrationStatus = P.OrchestrationStatus.Completed, + Result = new string('x', 4096), + }, + }, + }, + }; + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert + response.Actions.Should().HaveCount(2); + response.Actions[0].Should().BeSameAs(unlock); + response.Actions[1].CompleteOrchestration.OrchestrationStatus.Should().Be(P.OrchestrationStatus.Failed); + response.Actions[1].CompleteOrchestration.FailureDetails.IsNonRetriable.Should().BeTrue(); + response.CompletionToken.Should().Be("completion-token"); + store.UploadCount.Should().Be(0); + } + [Fact] public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_EarlierRetriableFailureWinsOverLaterPermanentFailure() { diff --git a/test/Grpc.IntegrationTests/LargePayloadTests.cs b/test/Grpc.IntegrationTests/LargePayloadTests.cs index 8502ab562..95049c6b3 100644 --- a/test/Grpc.IntegrationTests/LargePayloadTests.cs +++ b/test/Grpc.IntegrationTests/LargePayloadTests.cs @@ -2,17 +2,76 @@ // Licensed under the MIT License. using System.Text.Json; +using FluentAssertions; +using Grpc.Core; +using Grpc.Core.Interceptors; using Microsoft.DurableTask.Client; using Microsoft.DurableTask; using Microsoft.DurableTask.Converters; using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.Grpc; using Microsoft.Extensions.DependencyInjection; using Xunit.Abstractions; +using P = Microsoft.DurableTask.Protobuf; namespace Microsoft.DurableTask.Grpc.Tests; public class LargePayloadTests(ITestOutputHelper output, GrpcSidecarFixture sidecarFixture) : IntegrationTestBase(output, sidecarFixture) { + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task CompletionRetry_DoesNotExternalizePayloadTokensAgain(bool retryActivity) + { + // Arrange + string payload = new('R', 1024); + TaskName orchestratorName = nameof(CompletionRetry_DoesNotExternalizePayloadTokensAgain); + TaskName activityName = "RetryPayloadActivity"; + InMemoryPayloadStore store = new(); + RetryCompletionInterceptor retry = new(retryActivity); + await using HostTestLifetime server = await this.StartWorkerAsync( + worker => + { + worker.AddTasks(tasks => tasks + .AddOrchestratorFunc(orchestratorName, async (context, _) => + { + string result = await context.CallActivityAsync(activityName); + context.SetCustomStatus(result); + return result; + }) + .AddActivityFunc(activityName, _ => Task.FromResult(payload))); + worker.UseExternalizedPayloads(); + worker.Services.AddSingleton(store); + worker.Services.PostConfigure( + options => options.Interceptors.Add(retry)); + }, + client => + { + client.UseExternalizedPayloads(); + client.Services.AddSingleton(store); + }, + services => services.AddExternalizedPayloadStore(options => + { + options.ThresholdBytes = 16; + options.ConnectionString = "UseDevelopmentStorage=true"; + })); + + // Act + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName); + OrchestrationMetadata result = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + + // Assert + retry.Payloads.Should().HaveCount(2); + retry.Payloads[0].Length.Should().BeGreaterThan(16); + retry.Payloads[1].Should().Be(retry.Payloads[0]); + store.UploadCount.Should().Be(3, "only the activity result, orchestration result, and custom status need uploading"); + store.uploadedPayloads.Should().NotContain(value => store.IsKnownPayloadToken(value)); + result.RuntimeStatus.Should().Be(OrchestrationRuntimeStatus.Completed); + result.ReadOutputAs().Should().Be(payload); + result.ReadCustomStatusAs().Should().Be(payload); + } + // Validates client externalizes a large orchestration input and worker resolves it. [Fact] public async Task LargeOrchestrationInputAndOutputAndCustomStatus() @@ -792,9 +851,45 @@ public async Task LargeActivityOutput_ExceedsChunkSize_ExternalizesWithLP() Assert.Equal(largeOutput.Length, output?.Length); } + sealed class RetryCompletionInterceptor(bool retryActivity) : Interceptor + { + public List Payloads { get; } = []; + + public override AsyncUnaryCall AsyncUnaryCall( + TRequest request, + ClientInterceptorContext context, + AsyncUnaryCallContinuation continuation) + { + string? payload = request switch + { + P.ActivityResponse response when retryActivity => response.Result, + P.OrchestratorResponse response when !retryActivity => + response.Actions.FirstOrDefault(action => action.CompleteOrchestration is not null) + ?.CompleteOrchestration.Result, + _ => null, + }; + if (payload is not null) + { + this.Payloads.Add(payload); + if (this.Payloads.Count == 1) + { + Status status = new(StatusCode.Unavailable, "Simulated completion transport failure."); + return new AsyncUnaryCall( + Task.FromException(new RpcException(status)), + Task.FromResult(new Metadata()), + () => status, + () => new Metadata(), + () => { }); + } + } + + return continuation(request, context); + } + } + class InMemoryPayloadStore : PayloadStore { - const string TokenPrefix = "blob:v1:"; + const string TokenPrefix = "blob:v2:"; readonly Dictionary tokenToPayload; public readonly HashSet uploadedPayloads = new(); @@ -816,7 +911,7 @@ public InMemoryPayloadStore(Dictionary shared) public override Task UploadAsync(string payLoad, CancellationToken cancellationToken) { Interlocked.Increment(ref this.uploadCount); - string token = $"blob:v1:test:{Guid.NewGuid():N}"; + string token = $"{TokenPrefix}https://payloads.example.invalid/payloads/{Guid.NewGuid():N}"; this.tokenToPayload[token] = payLoad; this.uploadedPayloads.Add(payLoad); return Task.FromResult(token); From f9baad3a1164739c77fad1d6e9fc0ffe71f6e43e Mon Sep 17 00:00:00 2001 From: wangbill Date: Fri, 16 Oct 2026 15:19:41 -0400 Subject: [PATCH 2/2] Keep payload preparation wiring free of new warnings Reuse the worker's friend-visible validation helpers without duplicating Check and Verify. Retain assembly-local compiler compatibility types for the public record, order the internal preparation member consistently, and remove changelog whitespace. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 78dff439-4d84-4d70-921c-f57421cbc383 --- CHANGELOG.md | 2 +- .../AzureBlobPayloads.csproj | 8 ++ .../AzureBlobPayloadsSideCarInterceptor.cs | 88 +++++++++---------- 3 files changed, 53 insertions(+), 45 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e26415f7..27489c444 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## Unreleased -- Keep externalized payload tokens stable across completion retries and release entity locks when payload storage permanently fails an orchestration. +- Keep externalized payload tokens stable across completion retries and release entity locks when payload storage permanently fails an orchestration. - Preserve the original orchestration version when restarting through the orchestration-service client shim ([#463](https://github.com/microsoft/durabletask-dotnet/issues/463)). - Support configurable scheduler token audiences and government defaults ([#806](https://github.com/microsoft/durabletask-dotnet/pull/806)) diff --git a/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj index 78d80fd7a..5579f6162 100644 --- a/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj +++ b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj @@ -30,4 +30,12 @@ + + + + + + + + \ No newline at end of file diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index da297c995..21b65099e 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -27,6 +27,50 @@ public sealed class AzureBlobPayloadsSideCarInterceptor(PayloadStore payloadStor Action? SharedMessageLockAcquiredForTest { get; set; } + /// + /// Externalizes a response before the worker finalizes terminal entity cleanup. + /// + /// The orchestration response. + /// Cancellation token. + /// A task representing response preparation. + internal async Task PrepareOrchestratorResponseAsync(P.OrchestratorResponse response, CancellationToken cancellation) + { + try + { + await this.ExternalizeOrchestratorResponseAsync(response, cancellation); + } + catch (Exception ex) when (IsPermanentStorageFailure(ex)) + { + // Retain entity releases when replacing actions with a Failed completion so + // the orchestration terminates without leaving its entities locked. + P.OrchestratorAction[] unlocks = response.Actions.Where(action => + action.SendEntityMessage?.EntityUnlockSent is not null + || (action.SendEvent is { Name: "release", Instance: { } instance } + && global::DurableTask.Core.Common.Entities.IsEntityInstance(instance.InstanceId))).ToArray(); + response.Actions.Clear(); + response.Actions.AddRange(unlocks); + response.CustomStatus = null; +#pragma warning disable CS0612 // isPartial/chunkIndex are deprecated but still required for chunked response wire compatibility. + response.IsPartial = false; + response.ChunkIndex = null; +#pragma warning restore CS0612 + response.Actions.Add(new P.OrchestratorAction + { + CompleteOrchestration = new P.CompleteOrchestrationAction + { + OrchestrationStatus = P.OrchestrationStatus.Failed, + FailureDetails = new P.TaskFailureDetails + { + ErrorType = ex.GetType().FullName, + ErrorMessage = ex.Message, + StackTrace = ex.StackTrace, + IsNonRetriable = true, + }, + }, + }); + } + } + /// protected override async Task ExternalizeRequestPayloadsAsync(TRequest request, CancellationToken cancellation) { @@ -234,50 +278,6 @@ protected override async Task ResolveResponsePayloadsAsync(TResponse } } - /// - /// Externalizes a response before the worker finalizes terminal entity cleanup. - /// - /// The orchestration response. - /// Cancellation token. - /// A task representing response preparation. - internal async Task PrepareOrchestratorResponseAsync(P.OrchestratorResponse response, CancellationToken cancellation) - { - try - { - await this.ExternalizeOrchestratorResponseAsync(response, cancellation); - } - catch (Exception ex) when (IsPermanentStorageFailure(ex)) - { - // Retain entity releases when replacing actions with a Failed completion so - // the orchestration terminates without leaving its entities locked. - P.OrchestratorAction[] unlocks = response.Actions.Where(action => - action.SendEntityMessage?.EntityUnlockSent is not null - || (action.SendEvent is { Name: "release", Instance: { } instance } - && global::DurableTask.Core.Common.Entities.IsEntityInstance(instance.InstanceId))).ToArray(); - response.Actions.Clear(); - response.Actions.AddRange(unlocks); - response.CustomStatus = null; -#pragma warning disable CS0612 // isPartial/chunkIndex are deprecated but still required for chunked response wire compatibility. - response.IsPartial = false; - response.ChunkIndex = null; -#pragma warning restore CS0612 - response.Actions.Add(new P.OrchestratorAction - { - CompleteOrchestration = new P.CompleteOrchestrationAction - { - OrchestrationStatus = P.OrchestrationStatus.Failed, - FailureDetails = new P.TaskFailureDetails - { - ErrorType = ex.GetType().FullName, - ErrorMessage = ex.Message, - StackTrace = ex.StackTrace, - IsNonRetriable = true, - }, - }, - }); - } - } - void AddOrchestrationStateResolutionOperations( List> operations, P.OrchestrationState state,