diff --git a/CHANGELOG.md b/CHANGELOG.md index 787de264..1df0d96a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## 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)). + ## v1.28.0 - Keep externalized payload tokens stable across completion retries and release entity locks when payload storage permanently fails an orchestration. diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 2bfaf1d5..91c9aaf5 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -333,125 +333,92 @@ void AddOrchestrationStateResolutionOperations( } } - async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, CancellationToken cancellation) + void AddExternalizationOperation( + List> operations, + 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 () => assign(await this.ExternalizePayloadAsync(value!, cancellation))); } + } + + async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, CancellationToken cancellation) + { + List> operations = []; + this.AddExternalizationOperation(operations, 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 () => + this.AddExternalizationOperation( + operations, + complete.Result, + value => { - string externalizedResult = - await this.ExternalizePayloadAsync(result!, resultSize, cancellation); this.BeforeSharedMessageLockForTest?.Invoke(); lock (complete) { this.SharedMessageLockAcquiredForTest?.Invoke(complete); - complete.Result = externalizedResult; + complete.Result = value; } - }); - } - - string? details = complete.Details; - if (this.TryGetExternalizationSize(details, out int detailsSize)) - { - operations.Add(async () => + }, + cancellation); + this.AddExternalizationOperation( + operations, + complete.Details, + value => { - string externalizedDetails = - await this.ExternalizePayloadAsync(details!, detailsSize, cancellation); this.BeforeSharedMessageLockForTest?.Invoke(); lock (complete) { this.SharedMessageLockAcquiredForTest?.Invoke(complete); - complete.Details = externalizedDetails; + 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.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.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.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.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.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.Input, value => called.Input = value, cancellation); } } @@ -459,7 +426,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 +434,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 +443,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.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.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.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.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.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.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.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.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.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.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.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.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.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.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.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.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.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.Input, value => state.Input = value, cancellation); + this.AddExternalizationOperation(operations, state.Output, value => state.Output = value, cancellation); + this.AddExternalizationOperation(operations, state.CustomStatus, value => state.CustomStatus = value, cancellation); break; } } @@ -542,13 +510,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.EntityState, value => r.EntityState = value, cancellation); if (r.Results != null) { @@ -556,13 +518,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.Result, value => success.Result = value, cancellation); } } } @@ -573,24 +529,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.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.Input, value => start.Input = value, cancellation); } } } @@ -601,25 +545,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.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.Input, value => op.Input = value, cancellation); } } @@ -815,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 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..9419cfab 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -457,6 +457,675 @@ 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 + TaskCompletionSource firstWaveStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseUploads = new(TaskCreationOptions.RunContinuationsAsynchronously); + int uploadsStarted = 0; + 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---"; + for (int i = 0; i < 7; i++) + { + response.Actions.Insert(i, new P.OrchestratorAction + { + ScheduleTask = new P.ScheduleTaskAction { Input = $"input--{i}" }, + }); + } + + TrackingPayloadStore store = new(uploadAsync: async (payload, cancellationToken) => + { + if (Interlocked.Increment(ref uploadsStarted) == 8) + { + firstWaveStarted.TrySetResult(true); + } + + await releaseUploads.Task.WaitAsync(cancellationToken); + return $"test-blob://{payload}"; + }); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, new() { ThresholdBytes = 8, MaxPayloadBytes = 12 }); + + // Act: nested fields are queued behind the eight in-flight uploads. + Task externalizeTask = ExternalizeAsync(interceptor, response, CancellationToken.None); + try + { + await firstWaveStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); + state.Output = new string('x', 13); + state.CustomStatus = new string('y', 13); + } + finally + { + releaseUploads.TrySetResult(true); + await externalizeTask.WaitAsync(TimeSpan.FromSeconds(10)); + } + + // Assert + store.UploadCount.Should().Be(11); + store.MaxObservedConcurrency.Should().Be(8); + 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--"); + for (int i = 0; i < 7; i++) + { + response.Actions[i].ScheduleTask.Input.Should().Be($"test-blob://input--{i}"); + } + } + + [Theory] + [InlineData("orchestration")] + [InlineData("carryover")] + [InlineData("rewind")] + [InlineData("entity-request")] + [InlineData("entity-result")] + public async Task ExternalizeRequestPayloadsAsync_IndependentFieldsDoNotAcquireMessageMonitors(string messageKind) + { + // Arrange + IMessage request; + object[] messages; + switch (messageKind) + { + case "orchestration": + P.OrchestratorResponse response = new() + { + CustomStatus = "payload", + Actions = { new P.OrchestratorAction { ScheduleTask = new P.ScheduleTaskAction { Input = "payload" } } }, + }; + request = response; + messages = [response, response.Actions[0].ScheduleTask]; + break; + case "carryover": + case "rewind": + P.HistoryEvent historyEvent = CreateHistoryEventWithPayload( + P.HistoryEvent.EventTypeOneofCase.HistoryState, + "payload"); + P.OrchestratorResponse nestedResponse = CreateResponseWithHistory(historyEvent, messageKind == "rewind"); + request = nestedResponse; + messages = [nestedResponse, historyEvent.HistoryState.OrchestrationState]; + break; + case "entity-request": + P.EntityBatchRequest batch = new() + { + EntityState = "payload", + Operations = { new P.OperationRequest { Input = "payload" } }, + }; + request = batch; + messages = [batch, batch.Operations[0]]; + break; + case "entity-result": + P.EntityBatchResult result = new() + { + EntityState = "payload", + Results = { new P.OperationResult { Success = new P.OperationResultSuccess { Result = "payload" } } }, + Actions = + { + new P.OperationAction { SendSignal = new P.SendSignalAction { Input = "payload" } }, + new P.OperationAction { StartNewOrchestration = new P.StartNewOrchestrationAction { Input = "payload" } }, + }, + }; + request = result; + messages = [result, result.Results[0].Success, result.Actions[0].SendSignal, result.Actions[1].StartNewOrchestration]; + break; + default: + throw new ArgumentOutOfRangeException(nameof(messageKind)); + } + + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + TaskCompletionSource messageLocksHeld = new(TaskCreationOptions.RunContinuationsAsynchronously); + using ManualResetEventSlim releaseMessageLocks = new(); + Task holdMessageLocks = Task.Factory.StartNew( + () => + { + int acquired = 0; + try + { + foreach (object message in messages) + { + Monitor.Enter(message); + acquired++; + } + + messageLocksHeld.SetResult(true); + releaseMessageLocks.Wait(); + } + finally + { + for (int i = acquired - 1; i >= 0; i--) + { + Monitor.Exit(messages[i]); + } + } + }, + CancellationToken.None, + TaskCreationOptions.LongRunning, + TaskScheduler.Default); + Task externalizeTask = Task.CompletedTask; + + // Act: finish externalization while another thread still owns the message monitors. + try + { + await messageLocksHeld.Task.WaitAsync(TimeSpan.FromSeconds(10)); + externalizeTask = Task.Run(() => ExternalizeAsync(interceptor, request, CancellationToken.None)); + await externalizeTask.WaitAsync(TimeSpan.FromSeconds(10)); + } + finally + { + releaseMessageLocks.Set(); + await holdMessageLocks.WaitAsync(TimeSpan.FromSeconds(10)); + await externalizeTask.WaitAsync(TimeSpan.FromSeconds(10)); + } + + // Assert + IReadOnlyList payloads = request switch + { + P.OrchestratorResponse r when messageKind == "orchestration" => + [r.CustomStatus, r.Actions[0].ScheduleTask.Input], + P.OrchestratorResponse r => GetHistoryEventPayloads( + r.Actions[0].CompleteOrchestration?.CarryoverEvents[0] ?? r.Actions[0].RewindOrchestration.NewHistory[0]), + P.EntityBatchRequest r => [r.EntityState, r.Operations[0].Input], + P.EntityBatchResult r => + [r.EntityState, r.Results[0].Success.Result, r.Actions[0].SendSignal.Input, r.Actions[1].StartNewOrchestration.Input], + _ => throw new InvalidOperationException(), + }; + store.UploadCount.Should().Be(payloads.Count); + foreach (string token in payloads) + { + store.GetUploadedValue(token).Should().Be("payload"); + } + } + [Fact] public async Task ExternalizeRequestPayloadsAsync_CreateInstanceRequest_KnownTokenBypassesSizeLimits() { @@ -869,8 +1538,11 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_PreCancel response.Should().Be(expected); } - [Fact] - public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_CarryoverCancellationDrainsBoundedUploads() + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_NestedHistoryCancellationDrainsBoundedUploads( + bool rewind) { // Arrange TaskCompletionSource firstWaveStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -905,8 +1577,17 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Carryover { InstanceId = "instance-1", CompletionToken = "completion-token", - Actions = { new P.OrchestratorAction { CompleteOrchestration = complete } }, + Actions = + { + rewind + ? new P.OrchestratorAction + { + RewindOrchestration = new P.RewindOrchestrationAction { NewHistory = { complete.CarryoverEvents } }, + } + : new P.OrchestratorAction { CompleteOrchestration = complete }, + }, }; + P.OrchestratorAction originalAction = response.Actions[0]; // Act Task externalizeTask = ExternalizeAsync(interceptor, response, cts.Token); @@ -928,7 +1609,7 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Carryover store.MaxObservedConcurrency.Should().Be(8); finished.Should().Be(8); response.Actions.Should().ContainSingle(); - response.Actions[0].CompleteOrchestration.Should().BeSameAs(complete); + response.Actions[0].Should().BeSameAs(originalAction); complete.OrchestrationStatus.Should().Be(P.OrchestrationStatus.ContinuedAsNew); response.CompletionToken.Should().Be("completion-token"); for (int i = 0; i < complete.CarryoverEvents.Count; i++) @@ -938,24 +1619,35 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Carryover } [Theory] - [InlineData(403)] - [InlineData(500)] - public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_CarryoverFailureDrainsBeforeConvertingOrRethrowing( - int statusCode) + [InlineData(403, false)] + [InlineData(403, true)] + [InlineData(500, false)] + [InlineData(500, true)] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_NestedHistoryFailureDrainsBeforeConvertingOrRethrowing( + int statusCode, + bool rewind) { // Arrange TaskCompletionSource failureStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); TaskCompletionSource releaseUpload = new(TaskCreationOptions.RunContinuationsAsynchronously); + int finished = 0; TrackingPayloadStore store = new(uploadAsync: async (payload, cancellationToken) => { - if (payload == "slow") + try { - await releaseUpload.Task.WaitAsync(cancellationToken); - return "test-blob://slow"; - } + if (payload == "slow") + { + await releaseUpload.Task.WaitAsync(cancellationToken); + return "test-blob://slow"; + } - failureStarted.SetResult(true); - throw new RequestFailedException(statusCode, "Simulated carryover upload failure."); + failureStarted.SetResult(true); + throw new RequestFailedException(statusCode, "Simulated carryover upload failure."); + } + finally + { + Interlocked.Increment(ref finished); + } }); AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); P.CompleteOrchestrationAction complete = new() @@ -972,8 +1664,17 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Carryover { InstanceId = "instance-1", CompletionToken = "completion-token", - Actions = { new P.OrchestratorAction { CompleteOrchestration = complete } }, + Actions = + { + rewind + ? new P.OrchestratorAction + { + RewindOrchestration = new P.RewindOrchestrationAction { NewHistory = { complete.CarryoverEvents } }, + } + : new P.OrchestratorAction { CompleteOrchestration = complete }, + }, }; + P.OrchestratorAction originalAction = response.Actions[0]; // Act Task externalizeTask = ExternalizeAsync(interceptor, response, CancellationToken.None); @@ -981,7 +1682,7 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Carryover { await failureStarted.Task.WaitAsync(TimeSpan.FromSeconds(10)); externalizeTask.IsCompleted.Should().BeFalse(); - response.Actions[0].CompleteOrchestration.Should().BeSameAs(complete); + response.Actions[0].Should().BeSameAs(originalAction); } finally { @@ -1004,11 +1705,12 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Carryover Func act = () => externalizeTask; (await act.Should().ThrowExactlyAsync()) .WithMessage("Simulated carryover upload failure."); - response.Actions[0].CompleteOrchestration.Should().BeSameAs(complete); + response.Actions[0].Should().BeSameAs(originalAction); complete.OrchestrationStatus.Should().Be(P.OrchestrationStatus.ContinuedAsNew); } store.UploadCount.Should().Be(2, "failure must stop dispatch before the queued carryover event"); + finished.Should().Be(2, "every started upload must finish before returning or converting the response"); complete.CarryoverEvents[0].EventRaised.Input.Should().Be("test-blob://slow"); complete.CarryoverEvents[2].EventRaised.Input.Should().Be("queued"); response.CompletionToken.Should().Be("completion-token"); @@ -1618,6 +2320,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)