From 85ef3090c97ac9f18bb737aa58a57c2f6eb5068e Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 4 Nov 2026 13:57:07 -0800 Subject: [PATCH 1/2] Preflight externalized payload sizes before uploads Validate eligible UTF-8 payload snapshots before dispatching any message uploads, including entity batches and nested carryover/rewind history. Preserve cancellation priority, bounded I/O, retry tokens, and entity unlocks. Fixes #820. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 1 + .../AzureBlobPayloadsSideCarInterceptor.cs | 208 ++----- .../Interceptors/PayloadInterceptor.cs | 23 +- src/Extensions/AzureBlobPayloads/README.md | 24 + .../OrchestratorPayloadFailureTests.cs | 5 + ...zureBlobPayloadsSideCarInterceptorTests.cs | 542 ++++++++++++++++++ 6 files changed, 650 insertions(+), 153 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f6ef31ef..5b9099be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Validate all eligible externalized payload sizes before uploading any field of an orchestration response or entity batch, including nested carryover and rewind history. Deterministic size failures now precede hypothetical upload failures and cause no uploads or partial token substitutions ([#820](https://github.com/microsoft/durabletask-dotnet/issues/820)). - Keep externalized payload tokens stable across completion retries and release entity locks when payload storage permanently fails an orchestration. - Add the worker-owned `SetOrchestratorResponsePreparation` infrastructure API for response preparation before final worker entity cleanup and chunking. - Restrict Azure Blob payload downloads and deletion to the configured container and storage endpoint by default, with explicit account-wide access through `AllowAllContainersInConfiguredAccount` and additional identity-authenticated account endpoints through `AdditionalAccountUris`. Preserve v1/v2 token formats. Before changing the storage location, complete or explicitly migrate existing instances and clean up their payloads, or explicitly authorize the original location. diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 2bfaf1d5..84d1994e 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -333,125 +333,78 @@ void AddOrchestrationStateResolutionOperations( } } - async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, CancellationToken cancellation) + void AddExternalizationOperation( + List> operations, + object message, + string? value, + Action assign, + CancellationToken cancellation) { - List> operations = []; - string? customStatus = r.CustomStatus; - if (this.TryGetExternalizationSize(customStatus, out int customStatusSize)) + if (this.TryGetExternalizationSize(value, out int size)) { - operations.Add( - async () => r.CustomStatus = - await this.ExternalizePayloadAsync(customStatus!, customStatusSize, cancellation)); + // Validate snapshots while collecting, before dispatch can upload or mutate any field. + this.ValidatePayloadSize(size, cancellation); + operations.Add(async () => + { + string token = await this.ExternalizePayloadAsync(value!, cancellation); + this.BeforeSharedMessageLockForTest?.Invoke(); + lock (message) + { + this.SharedMessageLockAcquiredForTest?.Invoke(message); + assign(token); + } + }); } + } + + async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, CancellationToken cancellation) + { + List> operations = []; + this.AddExternalizationOperation(operations, r, r.CustomStatus, value => r.CustomStatus = value, cancellation); foreach (P.OrchestratorAction a in r.Actions) { if (a.CompleteOrchestration is { } complete) { - string? result = complete.Result; - if (this.TryGetExternalizationSize(result, out int resultSize)) - { - operations.Add(async () => - { - string externalizedResult = - await this.ExternalizePayloadAsync(result!, resultSize, cancellation); - this.BeforeSharedMessageLockForTest?.Invoke(); - lock (complete) - { - this.SharedMessageLockAcquiredForTest?.Invoke(complete); - complete.Result = externalizedResult; - } - }); - } - - string? details = complete.Details; - if (this.TryGetExternalizationSize(details, out int detailsSize)) - { - operations.Add(async () => - { - string externalizedDetails = - await this.ExternalizePayloadAsync(details!, detailsSize, cancellation); - this.BeforeSharedMessageLockForTest?.Invoke(); - lock (complete) - { - this.SharedMessageLockAcquiredForTest?.Invoke(complete); - complete.Details = externalizedDetails; - } - }); - } + this.AddExternalizationOperation(operations, complete, complete.Result, value => complete.Result = value, cancellation); + this.AddExternalizationOperation(operations, complete, complete.Details, value => complete.Details = value, cancellation); foreach (P.HistoryEvent historyEvent in complete.CarryoverEvents) { - operations.Add(() => this.ExternalizeHistoryEventAsync(historyEvent, cancellation)); + this.AddHistoryEventExternalizationOperations(operations, historyEvent, cancellation); } } if (a.TerminateOrchestration is { } term) { - string? reason = term.Reason; - if (this.TryGetExternalizationSize(reason, out int reasonSize)) - { - operations.Add( - async () => term.Reason = - await this.ExternalizePayloadAsync(reason!, reasonSize, cancellation)); - } + this.AddExternalizationOperation(operations, term, term.Reason, value => term.Reason = value, cancellation); } if (a.ScheduleTask is { } schedule) { - string? input = schedule.Input; - if (this.TryGetExternalizationSize(input, out int inputSize)) - { - operations.Add( - async () => schedule.Input = - await this.ExternalizePayloadAsync(input!, inputSize, cancellation)); - } + this.AddExternalizationOperation(operations, schedule, schedule.Input, value => schedule.Input = value, cancellation); } if (a.CreateSubOrchestration is { } sub) { - string? input = sub.Input; - if (this.TryGetExternalizationSize(input, out int inputSize)) - { - operations.Add( - async () => sub.Input = - await this.ExternalizePayloadAsync(input!, inputSize, cancellation)); - } + this.AddExternalizationOperation(operations, sub, sub.Input, value => sub.Input = value, cancellation); } if (a.SendEvent is { } sendEvt) { - string? data = sendEvt.Data; - if (this.TryGetExternalizationSize(data, out int dataSize)) - { - operations.Add( - async () => sendEvt.Data = - await this.ExternalizePayloadAsync(data!, dataSize, cancellation)); - } + this.AddExternalizationOperation(operations, sendEvt, sendEvt.Data, value => sendEvt.Data = value, cancellation); } if (a.SendEntityMessage is { } entityMsg) { if (entityMsg.EntityOperationSignaled is { } sig) { - string? input = sig.Input; - if (this.TryGetExternalizationSize(input, out int inputSize)) - { - operations.Add( - async () => sig.Input = - await this.ExternalizePayloadAsync(input!, inputSize, cancellation)); - } + this.AddExternalizationOperation(operations, sig, sig.Input, value => sig.Input = value, cancellation); } if (entityMsg.EntityOperationCalled is { } called) { - string? input = called.Input; - if (this.TryGetExternalizationSize(input, out int inputSize)) - { - operations.Add( - async () => called.Input = - await this.ExternalizePayloadAsync(input!, inputSize, cancellation)); - } + this.AddExternalizationOperation(operations, called, called.Input, value => called.Input = value, cancellation); } } @@ -459,7 +412,7 @@ async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, Cancel { foreach (P.HistoryEvent historyEvent in rewind.NewHistory) { - operations.Add(() => this.ExternalizeHistoryEventAsync(historyEvent, cancellation)); + this.AddHistoryEventExternalizationOperations(operations, historyEvent, cancellation); } } } @@ -467,7 +420,8 @@ async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, Cancel await RunWithBoundedConcurrencyAsync(operations, cancellation); } - async Task ExternalizeHistoryEventAsync( + void AddHistoryEventExternalizationOperations( + List> operations, P.HistoryEvent historyEvent, CancellationToken cancellation) { @@ -475,66 +429,66 @@ async Task ExternalizeHistoryEventAsync( switch (historyEvent.EventTypeCase) { case P.HistoryEvent.EventTypeOneofCase.ExecutionStarted when historyEvent.ExecutionStarted is { } es: - es.Input = await this.MaybeExternalizeAsync(es.Input, cancellation); + this.AddExternalizationOperation(operations, es, es.Input, value => es.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted when historyEvent.ExecutionCompleted is { } ec: - ec.Result = await this.MaybeExternalizeAsync(ec.Result, cancellation); + this.AddExternalizationOperation(operations, ec, ec.Result, value => ec.Result = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated when historyEvent.ExecutionTerminated is { } et: - et.Input = await this.MaybeExternalizeAsync(et.Input, cancellation); + this.AddExternalizationOperation(operations, et, et.Input, value => et.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.EventRaised when historyEvent.EventRaised is { } er: - er.Input = await this.MaybeExternalizeAsync(er.Input, cancellation); + this.AddExternalizationOperation(operations, er, er.Input, value => er.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.TaskScheduled when historyEvent.TaskScheduled is { } ts: - ts.Input = await this.MaybeExternalizeAsync(ts.Input, cancellation); + this.AddExternalizationOperation(operations, ts, ts.Input, value => ts.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.TaskCompleted when historyEvent.TaskCompleted is { } tc: - tc.Result = await this.MaybeExternalizeAsync(tc.Result, cancellation); + this.AddExternalizationOperation(operations, tc, tc.Result, value => tc.Result = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated when historyEvent.SubOrchestrationInstanceCreated is { } soc: - soc.Input = await this.MaybeExternalizeAsync(soc.Input, cancellation); + this.AddExternalizationOperation(operations, soc, soc.Input, value => soc.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted when historyEvent.SubOrchestrationInstanceCompleted is { } sox: - sox.Result = await this.MaybeExternalizeAsync(sox.Result, cancellation); + this.AddExternalizationOperation(operations, sox, sox.Result, value => sox.Result = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.EventSent when historyEvent.EventSent is { } esent: - esent.Input = await this.MaybeExternalizeAsync(esent.Input, cancellation); + this.AddExternalizationOperation(operations, esent, esent.Input, value => esent.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.GenericEvent when historyEvent.GenericEvent is { } ge: - ge.Data = await this.MaybeExternalizeAsync(ge.Data, cancellation); + this.AddExternalizationOperation(operations, ge, ge.Data, value => ge.Data = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.ContinueAsNew when historyEvent.ContinueAsNew is { } can: - can.Input = await this.MaybeExternalizeAsync(can.Input, cancellation); + this.AddExternalizationOperation(operations, can, can.Input, value => can.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended when historyEvent.ExecutionSuspended is { } esus: - esus.Input = await this.MaybeExternalizeAsync(esus.Input, cancellation); + this.AddExternalizationOperation(operations, esus, esus.Input, value => esus.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.ExecutionResumed when historyEvent.ExecutionResumed is { } eres: - eres.Input = await this.MaybeExternalizeAsync(eres.Input, cancellation); + this.AddExternalizationOperation(operations, eres, eres.Input, value => eres.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound when historyEvent.ExecutionRewound is { } erew: - erew.Reason = await this.MaybeExternalizeAsync(erew.Reason, cancellation); + this.AddExternalizationOperation(operations, erew, erew.Reason, value => erew.Reason = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled when historyEvent.EntityOperationSignaled is { } eos: - eos.Input = await this.MaybeExternalizeAsync(eos.Input, cancellation); + this.AddExternalizationOperation(operations, eos, eos.Input, value => eos.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled when historyEvent.EntityOperationCalled is { } eoc: - eoc.Input = await this.MaybeExternalizeAsync(eoc.Input, cancellation); + this.AddExternalizationOperation(operations, eoc, eoc.Input, value => eoc.Input = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted when historyEvent.EntityOperationCompleted is { } ecomp: - ecomp.Output = await this.MaybeExternalizeAsync(ecomp.Output, cancellation); + this.AddExternalizationOperation(operations, ecomp, ecomp.Output, value => ecomp.Output = value, cancellation); break; case P.HistoryEvent.EventTypeOneofCase.HistoryState when historyEvent.HistoryState?.OrchestrationState is { } state: - state.Input = await this.MaybeExternalizeAsync(state.Input, cancellation); - state.Output = await this.MaybeExternalizeAsync(state.Output, cancellation); - state.CustomStatus = await this.MaybeExternalizeAsync(state.CustomStatus, cancellation); + this.AddExternalizationOperation(operations, state, state.Input, value => state.Input = value, cancellation); + this.AddExternalizationOperation(operations, state, state.Output, value => state.Output = value, cancellation); + this.AddExternalizationOperation(operations, state, state.CustomStatus, value => state.CustomStatus = value, cancellation); break; } } @@ -542,13 +496,7 @@ async Task ExternalizeHistoryEventAsync( async Task ExternalizeEntityBatchResultAsync(P.EntityBatchResult r, CancellationToken cancellation) { List> operations = []; - string? entityState = r.EntityState; - if (this.TryGetExternalizationSize(entityState, out int entityStateSize)) - { - operations.Add( - async () => r.EntityState = - await this.ExternalizePayloadAsync(entityState!, entityStateSize, cancellation)); - } + this.AddExternalizationOperation(operations, r, r.EntityState, value => r.EntityState = value, cancellation); if (r.Results != null) { @@ -556,13 +504,7 @@ async Task ExternalizeEntityBatchResultAsync(P.EntityBatchResult r, Cancellation { if (result.Success is { } success) { - string? resultValue = success.Result; - if (this.TryGetExternalizationSize(resultValue, out int resultSize)) - { - operations.Add( - async () => success.Result = - await this.ExternalizePayloadAsync(resultValue!, resultSize, cancellation)); - } + this.AddExternalizationOperation(operations, success, success.Result, value => success.Result = value, cancellation); } } } @@ -573,24 +515,12 @@ async Task ExternalizeEntityBatchResultAsync(P.EntityBatchResult r, Cancellation { if (action.SendSignal is { } sendSig) { - string? input = sendSig.Input; - if (this.TryGetExternalizationSize(input, out int inputSize)) - { - operations.Add( - async () => sendSig.Input = - await this.ExternalizePayloadAsync(input!, inputSize, cancellation)); - } + this.AddExternalizationOperation(operations, sendSig, sendSig.Input, value => sendSig.Input = value, cancellation); } if (action.StartNewOrchestration is { } start) { - string? input = start.Input; - if (this.TryGetExternalizationSize(input, out int inputSize)) - { - operations.Add( - async () => start.Input = - await this.ExternalizePayloadAsync(input!, inputSize, cancellation)); - } + this.AddExternalizationOperation(operations, start, start.Input, value => start.Input = value, cancellation); } } } @@ -601,25 +531,13 @@ async Task ExternalizeEntityBatchResultAsync(P.EntityBatchResult r, Cancellation async Task ExternalizeEntityBatchRequestAsync(P.EntityBatchRequest r, CancellationToken cancellation) { List> operations = []; - string? entityState = r.EntityState; - if (this.TryGetExternalizationSize(entityState, out int entityStateSize)) - { - operations.Add( - async () => r.EntityState = - await this.ExternalizePayloadAsync(entityState!, entityStateSize, cancellation)); - } + this.AddExternalizationOperation(operations, r, r.EntityState, value => r.EntityState = value, cancellation); if (r.Operations != null) { foreach (P.OperationRequest op in r.Operations) { - string? input = op.Input; - if (this.TryGetExternalizationSize(input, out int inputSize)) - { - operations.Add( - async () => op.Input = - await this.ExternalizePayloadAsync(input!, inputSize, cancellation)); - } + this.AddExternalizationOperation(operations, op, op.Input, value => op.Input = value, cancellation); } } diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs index 76b9fb5f..8f0ab337 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/PayloadInterceptor.cs @@ -186,7 +186,8 @@ public override AsyncServerStreamingCall AsyncServerStreamingCall @@ -206,26 +207,32 @@ public override AsyncServerStreamingCall AsyncServerStreamingCall - /// Externalizes a payload whose UTF-8 size has already been calculated. + /// Externalizes a payload whose UTF-8 size has already been validated. /// /// The value to externalize. - /// The UTF-8 size of . /// Cancellation token. /// A task that returns the externalized token. - private protected async Task ExternalizePayloadAsync( + private protected Task ExternalizePayloadAsync( string value, - int size, CancellationToken cancellation) { - // Enforce a hard cap to prevent unbounded payload sizes + return this.payloadStore.UploadAsync(value, cancellation); + } + + /// + /// Validates an eligible payload's UTF-8 size before starting storage operations. + /// + /// The UTF-8 payload size. + /// Cancellation token, honored before size validation. + private protected void ValidatePayloadSize(int size, CancellationToken cancellation) + { + cancellation.ThrowIfCancellationRequested(); if (size > this.options.MaxPayloadBytes) { throw new PayloadStorageException( $"Payload size {size / 1024} KB exceeds the configured maximum of {this.options.MaxPayloadBytes / 1024} KB. " + "Reduce the payload size or increase the max payload size limit."); } - - return await this.payloadStore.UploadAsync(value, cancellation); } /// diff --git a/src/Extensions/AzureBlobPayloads/README.md b/src/Extensions/AzureBlobPayloads/README.md index fadc3fa8..146700ca 100644 --- a/src/Extensions/AzureBlobPayloads/README.md +++ b/src/Extensions/AzureBlobPayloads/README.md @@ -146,6 +146,30 @@ the appropriate explicit scope allows a later attempt to proceed. An authorized delete still checks the blob's ownership metadata and uses its ETag in the delete condition. +## Payload size validation + +Before uploading a multi-field orchestration response or entity batch, the +interceptor snapshots eligible payload fields and validates their UTF-8 byte +sizes. This includes nested continue-as-new carryover and rewind history fields. +A deterministic `MaxPayloadBytes` violation causes zero uploads and no partial +payload-token substitutions for that message. Existing worker failure conversion +still applies, including entity unlocks on permanent orchestration failures. + +The threshold and maximum are inclusive: a nonempty value at `ThresholdBytes` +is externalized, and a value at `MaxPayloadBytes` is allowed. The maximum applies +only to fields selected for externalization. Empty, absent, below-threshold, and +recognized-token values are unchanged. Precancellation takes priority over size +validation for eligible work; a message with no eligible fields performs no +storage work, even with a canceled token. + +Because preflight starts no I/O, a deterministic size violation takes precedence +over an earlier field's hypothetical upload failure. Once preflight passes, +uploads retain bounded concurrency, drainage, and message-order selection of +actual I/O failures. This is not a cross-blob transaction or cleanup of existing +orphan blobs: real I/O failures, cancellation during uploads, and uncertain +backend submission can still leave unrecorded uploads. No catch-and-delete +rollback is performed. + ## SDK and dashboard boundaries The client and worker gRPC interceptors resolve orchestration inputs, external diff --git a/test/Extensions/AzureBlobPayloads.Tests/OrchestratorPayloadFailureTests.cs b/test/Extensions/AzureBlobPayloads.Tests/OrchestratorPayloadFailureTests.cs index 020b2ab4..f07fa3cf 100644 --- a/test/Extensions/AzureBlobPayloads.Tests/OrchestratorPayloadFailureTests.cs +++ b/test/Extensions/AzureBlobPayloads.Tests/OrchestratorPayloadFailureTests.cs @@ -157,6 +157,11 @@ static async Task RunLockedOrchestrationAsync( { IAsyncDisposable locks = await context.Entities.LockEntitiesAsync(entities); await using IAsyncDisposable? releasedLocks = explicitlyReleaseLocks ? locks : null; + if (failure == PayloadFailure.SizeLimit) + { + context.SetCustomStatus(new string('s', 1024)); + } + if (!completeOrchestration) { await context.Entities.CallEntityAsync(entities[0], "set", payload); diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 3595c414..27b35e85 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -457,6 +457,525 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Externali store.MaxObservedConcurrency.Should().BeLessOrEqualTo(8, "concurrency must be bounded to avoid Azure Storage throttling"); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_OversizedLaterPayloadDoesNotUploadOrMutateTokens( + bool unicode) + { + // Arrange + TrackingPayloadStore store = new(); + LargePayloadStorageOptions options = new() { ThresholdBytes = 1024, MaxPayloadBytes = 2048 }; + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, options); + P.OrchestratorAction action = new() + { + Id = 1, + ScheduleTask = new P.ScheduleTaskAction + { + Name = "activity", + Input = unicode ? new string('\u00e9', 1025) : new string('x', 2049), + }, + }; + P.OrchestratorAction expectedAction = action.Clone(); + P.OrchestratorResponse response = new() + { + InstanceId = "instance-1", + CompletionToken = "completion-token", + CustomStatus = new string('s', 1024), + Actions = { action }, + }; + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert + store.UploadCount.Should().Be(0); + action.Should().Be(expectedAction); + response.CustomStatus.Should().BeNull(); + response.CompletionToken.Should().Be("completion-token"); + P.CompleteOrchestrationAction failure = response.Actions.Should().ContainSingle().Subject.CompleteOrchestration; + failure.OrchestrationStatus.Should().Be(P.OrchestrationStatus.Failed); + failure.FailureDetails.ErrorType.Should().Be(typeof(PayloadStorageException).FullName); + failure.FailureDetails.IsNonRetriable.Should().BeTrue(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ExternalizeRequestPayloadsAsync_EntityBatchRequest_OversizedLaterPayloadDoesNotUploadOrMutateTokens( + bool unicode) + { + // Arrange + TrackingPayloadStore store = new(); + LargePayloadStorageOptions options = new() { ThresholdBytes = 1024, MaxPayloadBytes = 2048 }; + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, options); + P.EntityBatchRequest request = new() + { + InstanceId = "entity-1", + EntityState = new string('s', 1024), + Operations = + { + new P.OperationRequest { Operation = "first", Input = new string('x', 2048) }, + new P.OperationRequest + { + Operation = "second", + Input = unicode ? new string('\u00e9', 1025) : new string('x', 2049), + }, + }, + }; + P.EntityBatchRequest expected = request.Clone(); + + // Act + Func act = () => ExternalizeAsync(interceptor, request, CancellationToken.None); + + // Assert + await act.Should().ThrowExactlyAsync() + .WithMessage("*exceeds the configured maximum*"); + store.UploadCount.Should().Be(0); + request.Should().Be(expected); + } + + [Theory] + [InlineData("result")] + [InlineData("details")] + [InlineData("termination")] + [InlineData("sub-orchestration")] + [InlineData("event")] + [InlineData("entity-signal")] + [InlineData("entity-call")] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_OversizedActionPreflightsEntireMessage( + string field) + { + // Arrange + string oversized = new('x', 13); + P.OrchestratorAction action = field switch + { + "result" => new() { CompleteOrchestration = new P.CompleteOrchestrationAction { Result = oversized } }, + "details" => new() + { + CompleteOrchestration = new P.CompleteOrchestrationAction { Result = "result--", Details = oversized }, + }, + "termination" => new() { TerminateOrchestration = new P.TerminateOrchestrationAction { Reason = oversized } }, + "sub-orchestration" => new() { CreateSubOrchestration = new P.CreateSubOrchestrationAction { Input = oversized } }, + "event" => new() { SendEvent = new P.SendEventAction { Data = oversized } }, + "entity-signal" => new() + { + SendEntityMessage = new P.SendEntityMessageAction + { + EntityOperationSignaled = new P.EntityOperationSignaledEvent { Input = oversized }, + }, + }, + "entity-call" => new() + { + SendEntityMessage = new P.SendEntityMessageAction + { + EntityOperationCalled = new P.EntityOperationCalledEvent { Input = oversized }, + }, + }, + _ => throw new ArgumentOutOfRangeException(nameof(field)), + }; + P.OrchestratorAction expected = action.Clone(); + P.OrchestratorResponse response = new() { CustomStatus = "status--", Actions = { action } }; + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 12 }); + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert + store.UploadCount.Should().Be(0); + action.Should().Be(expected); + P.CompleteOrchestrationAction failure = response.Actions.Should().ContainSingle().Subject.CompleteOrchestration; + failure.OrchestrationStatus.Should().Be(P.OrchestrationStatus.Failed); + failure.FailureDetails.ErrorType.Should().Be(typeof(PayloadStorageException).FullName); + failure.FailureDetails.IsNonRetriable.Should().BeTrue(); + } + + [Theory] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.ExecutionStarted)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.ExecutionStarted)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.EventRaised)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.EventRaised)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.TaskScheduled)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.TaskScheduled)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.TaskCompleted)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.TaskCompleted)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.EventSent)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.EventSent)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.GenericEvent)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.GenericEvent)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.ContinueAsNew)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.ContinueAsNew)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.ExecutionResumed)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.ExecutionResumed)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.ExecutionRewound)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.ExecutionRewound)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted)] + [InlineData(false, P.HistoryEvent.EventTypeOneofCase.HistoryState)] + [InlineData(true, P.HistoryEvent.EventTypeOneofCase.HistoryState)] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_OversizedNestedPayloadPreflightsEntireMessage( + bool rewind, + P.HistoryEvent.EventTypeOneofCase eventType) + { + // Arrange + P.HistoryEvent historyEvent = CreateHistoryEventWithPayload(eventType, new string('x', 13)); + P.OrchestratorResponse response = CreateResponseWithHistory(historyEvent, rewind); + response.CustomStatus = "status--"; + P.OrchestratorAction action = response.Actions[0]; + P.OrchestratorAction expected = action.Clone(); + TrackingPayloadStore store = new(uploadAsync: (_, _) => throw new RequestFailedException(500, "Upload must not start.")); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 12 }); + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert: deterministic size failure precedes even a hypothetical earlier I/O failure. + store.UploadCount.Should().Be(0); + action.Should().Be(expected); + P.CompleteOrchestrationAction failure = response.Actions.Should().ContainSingle().Subject.CompleteOrchestration; + failure.OrchestrationStatus.Should().Be(P.OrchestrationStatus.Failed); + failure.FailureDetails.ErrorType.Should().Be(typeof(PayloadStorageException).FullName); + failure.FailureDetails.IsNonRetriable.Should().BeTrue(); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_OversizedLaterHistoryStateFieldDoesNotMutateTokens( + bool rewind, + bool oversizedStatus) + { + // Arrange + P.HistoryEvent historyEvent = new() + { + HistoryState = new P.HistoryStateEvent + { + OrchestrationState = new P.OrchestrationState + { + Input = "input---", + Output = oversizedStatus ? "output--" : new string('x', 13), + CustomStatus = oversizedStatus ? new string('x', 13) : "status--", + }, + }, + }; + P.HistoryEvent expected = historyEvent.Clone(); + P.OrchestratorResponse response = CreateResponseWithHistory(historyEvent, rewind); + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 12 }); + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert + store.UploadCount.Should().Be(0); + historyEvent.Should().Be(expected); + response.Actions.Should().ContainSingle().Which.CompleteOrchestration.OrchestrationStatus + .Should().Be(P.OrchestrationStatus.Failed); + } + + [Theory] + [InlineData("result", false)] + [InlineData("result", true)] + [InlineData("signal", false)] + [InlineData("signal", true)] + [InlineData("start", false)] + [InlineData("start", true)] + public async Task ExternalizeRequestPayloadsAsync_EntityBatchResult_OversizedLaterPayloadDoesNotUploadOrMutateTokens( + string field, + bool unicode) + { + // Arrange + string oversized = unicode ? new string('\u00e9', 7) : new string('x', 13); + P.EntityBatchResult result = new() + { + EntityState = "state---", + Results = + { + new P.OperationResult + { + Success = new P.OperationResultSuccess { Result = field == "result" ? oversized : "result--" }, + }, + }, + Actions = + { + new P.OperationAction + { + SendSignal = new P.SendSignalAction { Input = field == "signal" ? oversized : "signal--" }, + }, + new P.OperationAction + { + StartNewOrchestration = new P.StartNewOrchestrationAction { Input = field == "start" ? oversized : "start---" }, + }, + }, + }; + P.EntityBatchResult expected = result.Clone(); + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 12 }); + + // Act + Func act = () => ExternalizeAsync(interceptor, result, CancellationToken.None); + + // Assert + await act.Should().ThrowExactlyAsync(); + store.UploadCount.Should().Be(0); + result.Should().Be(expected); + } + + [Theory] + [InlineData("client")] + [InlineData("activity")] + [InlineData("orchestration")] + [InlineData("entity-request")] + [InlineData("entity-result")] + [InlineData("entity-v2")] + [InlineData("carryover")] + [InlineData("rewind")] + public async Task ExternalizeRequestPayloadsAsync_PreCancelledOversizedPayloadPreservesMessage( + string messageKind) + { + // Arrange + string oversized = new('x', 13); + IMessage request = messageKind switch + { + "client" => new P.CreateInstanceRequest { Input = oversized }, + "activity" => new P.ActivityResponse { Result = oversized }, + "orchestration" => new P.OrchestratorResponse { CustomStatus = oversized }, + "entity-request" => new P.EntityBatchRequest + { + EntityState = oversized, + Operations = { new P.OperationRequest { Input = "input---" } }, + }, + "entity-result" => new P.EntityBatchResult { EntityState = oversized }, + "entity-v2" => new P.EntityRequest { EntityState = oversized }, + "carryover" => CreateResponseWithHistory( + CreateHistoryEventWithPayload(P.HistoryEvent.EventTypeOneofCase.ExecutionRewound, oversized), + rewind: false), + "rewind" => CreateResponseWithHistory( + CreateHistoryEventWithPayload(P.HistoryEvent.EventTypeOneofCase.HistoryState, oversized), + rewind: true), + _ => throw new ArgumentOutOfRangeException(nameof(messageKind)), + }; + ByteString expected = request.ToByteString(); + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 12 }); + using CancellationTokenSource cts = new(); + cts.Cancel(); + + // Act + Func act = () => ExternalizeAsync(interceptor, request, cts.Token); + + // Assert + (await act.Should().ThrowAsync()).Which.CancellationToken.Should().Be(cts.Token); + store.UploadCount.Should().Be(0); + request.ToByteString().Should().Equal(expected); + } + + [Theory] + [InlineData("client")] + [InlineData("activity")] + [InlineData("orchestration")] + [InlineData("entity-request")] + [InlineData("entity-result")] + [InlineData("entity-v2")] + public async Task ExternalizeRequestPayloadsAsync_PreCancelledIneligiblePayloadsIgnoreMaximumAndPerformNoWork( + string messageKind) + { + // Arrange: inline values exceed the maximum but not the externalization threshold. + TrackingPayloadStore store = new(); + string token = store.Seed("original"); + IMessage request = messageKind switch + { + "client" => new P.CreateInstanceRequest { Input = "inline" }, + "activity" => new P.ActivityResponse { Result = token }, + "orchestration" => new P.OrchestratorResponse + { + CustomStatus = "inline", + Actions = + { + new P.OrchestratorAction + { + CompleteOrchestration = new P.CompleteOrchestrationAction + { + Result = token, + Details = string.Empty, + CarryoverEvents = + { + new P.HistoryEvent + { + HistoryState = new P.HistoryStateEvent + { + OrchestrationState = new P.OrchestrationState { Output = "inline", CustomStatus = token }, + }, + }, + }, + }, + }, + new P.OrchestratorAction + { + RewindOrchestration = new P.RewindOrchestrationAction + { + NewHistory = + { + new P.HistoryEvent { EventRaised = new P.EventRaisedEvent { Input = string.Empty } }, + new P.HistoryEvent { TimerFired = new P.TimerFiredEvent() }, + }, + }, + }, + }, + }, + "entity-request" => new P.EntityBatchRequest + { + EntityState = "inline", + Operations = { new P.OperationRequest { Input = token }, new P.OperationRequest() }, + }, + "entity-result" => new P.EntityBatchResult + { + EntityState = "inline", + Results = { new P.OperationResult { Success = new P.OperationResultSuccess { Result = token } } }, + }, + "entity-v2" => new P.EntityRequest(), + _ => throw new ArgumentOutOfRangeException(nameof(messageKind)), + }; + ByteString expected = request.ToByteString(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 4 }); + using CancellationTokenSource cts = new(); + cts.Cancel(); + + // Act + await ExternalizeAsync(interceptor, request, cts.Token); + + // Assert + store.UploadCount.Should().Be(0); + store.DownloadCount.Should().Be(0); + request.ToByteString().Should().Equal(expected); + } + + [Theory] + [InlineData("orchestration", false)] + [InlineData("orchestration", true)] + [InlineData("entity-request", false)] + [InlineData("entity-request", true)] + [InlineData("entity-result", false)] + [InlineData("entity-result", true)] + public async Task ExternalizeRequestPayloadsAsync_AllValidBatchHonorsUtf8BoundariesAndExistingTokens( + string messageKind, + bool unicode) + { + // Arrange + string threshold = unicode ? new string('\u00e9', 4) : new string('t', 8); + string maximum = unicode ? new string('\u00e9', 6) : new string('m', 12); + TrackingPayloadStore store = new(); + string token = store.Seed("original"); + IMessage request = messageKind switch + { + "orchestration" => new P.OrchestratorResponse + { + CustomStatus = threshold, + Actions = + { + new P.OrchestratorAction { ScheduleTask = new P.ScheduleTaskAction { Input = maximum } }, + new P.OrchestratorAction { SendEvent = new P.SendEventAction { Data = "inline" } }, + new P.OrchestratorAction { CompleteOrchestration = new P.CompleteOrchestrationAction { Result = token } }, + }, + }, + "entity-request" => new P.EntityBatchRequest + { + EntityState = threshold, + Operations = + { + new P.OperationRequest { Input = maximum }, + new P.OperationRequest { Input = "inline" }, + new P.OperationRequest { Input = token }, + new P.OperationRequest { Input = string.Empty }, + new P.OperationRequest(), + }, + }, + "entity-result" => new P.EntityBatchResult + { + EntityState = threshold, + Results = + { + new P.OperationResult { Success = new P.OperationResultSuccess { Result = maximum } }, + new P.OperationResult { Success = new P.OperationResultSuccess { Result = "inline" } }, + new P.OperationResult { Success = new P.OperationResultSuccess { Result = token } }, + new P.OperationResult { Success = new P.OperationResultSuccess { Result = string.Empty } }, + new P.OperationResult { Success = new P.OperationResultSuccess() }, + }, + }, + _ => throw new ArgumentOutOfRangeException(nameof(messageKind)), + }; + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 12 }); + + // Act + await ExternalizeAsync(interceptor, request, CancellationToken.None); + await ExternalizeAsync(interceptor, request, CancellationToken.None); + + // Assert: exactly two token substitutions, including after a retry. + (string? first, string? second, string? inline, string? known) = request switch + { + P.OrchestratorResponse r => (r.CustomStatus, r.Actions[0].ScheduleTask.Input, r.Actions[1].SendEvent.Data, r.Actions[2].CompleteOrchestration.Result), + P.EntityBatchRequest r => (r.EntityState, r.Operations[0].Input, r.Operations[1].Input, r.Operations[2].Input), + P.EntityBatchResult r => (r.EntityState, r.Results[0].Success.Result, r.Results[1].Success.Result, r.Results[2].Success.Result), + _ => throw new InvalidOperationException(), + }; + store.UploadCount.Should().Be(2); + store.GetUploadedValue(Assert.IsType(first)).Should().Be(threshold); + store.GetUploadedValue(Assert.IsType(second)).Should().Be(maximum); + inline.Should().Be("inline"); + known.Should().Be(token); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_UsesNestedFieldSnapshotsBeforeDispatch( + bool rewind) + { + // Arrange + P.OrchestrationState state = new() { Input = "input---", Output = "output--", CustomStatus = "status--" }; + P.HistoryEvent historyEvent = new() { HistoryState = new P.HistoryStateEvent { OrchestrationState = state } }; + P.OrchestratorResponse response = CreateResponseWithHistory(historyEvent, rewind); + response.CustomStatus = "first---"; + TrackingPayloadStore store = new(uploadAsync: (payload, _) => + { + if (payload == "first---") + { + state.Output = new string('x', 13); + state.CustomStatus = new string('y', 13); + } + + return Task.FromResult($"test-blob://{payload}"); + }); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 12 }); + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert + store.UploadCount.Should().Be(4); + response.CustomStatus.Should().Be("test-blob://first---"); + state.Input.Should().Be("test-blob://input---"); + state.Output.Should().Be("test-blob://output--"); + state.CustomStatus.Should().Be("test-blob://status--"); + } + [Fact] public async Task ExternalizeRequestPayloadsAsync_CreateInstanceRequest_KnownTokenBypassesSizeLimits() { @@ -1618,6 +2137,29 @@ public async Task RunWithBoundedConcurrencyAsync_CancellationAfterAllOperationsC completedOperations.Should().Be(2); } + static P.OrchestratorResponse CreateResponseWithHistory(P.HistoryEvent historyEvent, bool rewind) + { + P.OrchestratorAction action = rewind + ? new() + { + RewindOrchestration = new P.RewindOrchestrationAction { NewHistory = { historyEvent } }, + } + : new() + { + CompleteOrchestration = new P.CompleteOrchestrationAction + { + OrchestrationStatus = P.OrchestrationStatus.ContinuedAsNew, + CarryoverEvents = { historyEvent }, + }, + }; + return new() + { + InstanceId = "instance-1", + CompletionToken = "completion-token", + Actions = { action }, + }; + } + static P.HistoryEvent CreateHistoryEventWithPayload( P.HistoryEvent.EventTypeOneofCase eventType, string payload) From 08f0074ce7634615d84ff07c8117d91f706e997d Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 4 Nov 2026 23:52:21 -0500 Subject: [PATCH 2/2] Clarify payload assignment concurrency documentation Describe distinct captured field assignments and only the retained completion/hydration lock scopes, rather than promising generic message serialization. No runtime behavior changes. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../Interceptors/AzureBlobPayloadsSideCarInterceptor.cs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index acc615ec..91c9aaf5 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -747,10 +747,11 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell /// bounded concurrency, instead of either awaiting them one at a time (additive latency) or /// firing them all at once via an unbounded /// (creating a per-message burst for messages with many payloads). Each delegate is - /// expected to assign its result to a distinct protobuf field/element, so the relative + /// expected to assign its result to a distinct captured protobuf field/element, so the relative /// completion order between operations does not affect correctness. Blob I/O is concurrent, - /// but assignments targeting fields on the same protobuf message are serialized rather than - /// relying on current generated setter implementation details. + /// and independent field assignments can overlap. Only the pre-existing completion and + /// state-hydration paths retain scoped assignment locks; this helper does not serialize + /// message mutations. /// /// /// Cancellation is honored before starting any operation not already in flight. If any