From 5489022e92e3c1b153817238c08b31d83e21130f Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 11 Sep 2026 10:18:20 -0700 Subject: [PATCH 01/11] added rewind to the sidecar --- .github/workflows/validate-build.yml | 16 + src/Shared/Grpc/ProtoUtils.cs | 153 ++-- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 143 +++- src/Worker/Grpc/RewindOrchestrationHandler.cs | 146 ++++ .../DtsEmulatorFactAttribute.cs | 19 + .../DtsRewindIntegrationTests.cs | 538 ++++++++++++++ .../Grpc.IntegrationTests.csproj | 2 + .../run-dts-emulator-tests.ps1 | 85 +++ .../Grpc.Tests/GrpcDurableTaskWorkerTests.cs | 672 +++++++++++++++++- .../RewindOrchestrationHandlerTests.cs | 328 +++++++++ 10 files changed, 1988 insertions(+), 114 deletions(-) create mode 100644 src/Worker/Grpc/RewindOrchestrationHandler.cs create mode 100644 test/Grpc.IntegrationTests/DtsEmulatorFactAttribute.cs create mode 100644 test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs create mode 100644 test/Grpc.IntegrationTests/run-dts-emulator-tests.ps1 create mode 100644 test/Worker/Grpc.Tests/RewindOrchestrationHandlerTests.cs diff --git a/.github/workflows/validate-build.yml b/.github/workflows/validate-build.yml index 15d6f920b..366db01d2 100644 --- a/.github/workflows/validate-build.yml +++ b/.github/workflows/validate-build.yml @@ -63,3 +63,19 @@ jobs: with: name: pkg path: out/pkg + + dts-emulator-tests: + runs-on: ubuntu-latest + timeout-minutes: 20 + + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Setup .NET from global.json + uses: actions/setup-dotnet@a98b56852c35b8e3190ac28c8c2271da59106c68 # v6.0.0 + with: + global-json-file: global.json + + - name: Run DTS emulator rewind tests + shell: pwsh + run: ./test/Grpc.IntegrationTests/run-dts-emulator-tests.ps1 -Configuration release diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs index 2412fac37..f6d6a1f7c 100644 --- a/src/Shared/Grpc/ProtoUtils.cs +++ b/src/Shared/Grpc/ProtoUtils.cs @@ -80,7 +80,7 @@ internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto, EntityCon historyEvent = new ExecutionCompletedEvent( proto.EventId, proto.ExecutionCompleted.Result, - proto.ExecutionCompleted.OrchestrationStatus.ToCore(), + proto.ExecutionCompleted.OrchestrationStatus.ToCore(), proto.ExecutionCompleted.FailureDetails.ToCore()); break; case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated: @@ -220,9 +220,9 @@ internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto, EntityCon Tags = proto.HistoryState.OrchestrationState.Tags, }); break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound: - historyEvent = new ExecutionRewoundEvent(proto.EventId); - break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound: + historyEvent = new ExecutionRewoundEvent(proto.EventId); + break; default: throw new NotSupportedException($"Deserialization of {proto.EventTypeCase} is not supported."); } @@ -335,11 +335,7 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( ActivitySpanId clientSpanId = ActivitySpanId.CreateRandom(); ActivityContext clientActivityContext = new(orchestrationActivity.TraceId, clientSpanId, orchestrationActivity.ActivityTraceFlags, orchestrationActivity.TraceStateString); - return new P.TraceContext - { - TraceParent = $"00-{clientActivityContext.TraceId}-{clientActivityContext.SpanId}-0{clientActivityContext.TraceFlags:d}", - TraceState = clientActivityContext.TraceState, - }; + return ProtoUtils.CreateTraceContext(clientActivityContext); } switch (action.OrchestratorActionType) @@ -373,15 +369,15 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( Name = subOrchestrationAction.Name, Version = subOrchestrationAction.Version, ParentTraceContext = CreateTraceContext(), - }; - + }; + if (subOrchestrationAction.Tags != null) { foreach (KeyValuePair tag in subOrchestrationAction.Tags) { protoAction.CreateSubOrchestration.Tags[tag.Key] = tag.Value; } - } + } break; case OrchestratorActionType.CreateTimer: @@ -471,17 +467,17 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( var completeAction = (OrchestrationCompleteOrchestratorAction)action; protoAction.CompleteOrchestration = new P.CompleteOrchestrationAction - { + { CarryoverEvents = { completeAction.CarryoverEvents.Select(ToProtobuf) }, Details = completeAction.Details, NewVersion = completeAction.NewVersion, OrchestrationStatus = completeAction.OrchestrationStatus.ToProtobuf(), Result = completeAction.Result, - }; - - foreach (KeyValuePair tag in completeAction.Tags) - { - protoAction.CompleteOrchestration.Tags[tag.Key] = tag.Value; + }; + + foreach (KeyValuePair tag in completeAction.Tags) + { + protoAction.CompleteOrchestration.Tags[tag.Key] = tag.Value; } if (completeAction.OrchestrationStatus == OrchestrationStatus.Failed) @@ -500,6 +496,21 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( return response; } + /// + /// Creates a protobuf trace context from an activity context. + /// + /// The activity context to convert. + /// The corresponding protobuf trace context. + internal static P.TraceContext CreateTraceContext(ActivityContext activityContext) + { + return new() + { + TraceParent = + $"00-{activityContext.TraceId}-{activityContext.SpanId}-0{activityContext.TraceFlags:d}", + TraceState = activityContext.TraceState, + }; + } + /// /// Converts a to a . /// @@ -1052,27 +1063,27 @@ internal static T Base64Decode(this MessageParser parser, string encodedMessa case Google.Protobuf.WellKnownTypes.Value.KindOneofCase.NumberValue: return value.NumberValue; case Google.Protobuf.WellKnownTypes.Value.KindOneofCase.StringValue: - string stringValue = value.StringValue; - - // If the value starts with the 'dt:' prefix, it may represent a DateTime value — attempt to parse it. - if (stringValue.StartsWith("dt:", StringComparison.Ordinal)) - { - if (DateTime.TryParse(stringValue[3..], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTime date)) - { - return date; - } - } - - // If the value starts with the 'dto:' prefix, it may represent a DateTime value — attempt to parse it. - if (stringValue.StartsWith("dto:", StringComparison.Ordinal)) - { - if (DateTimeOffset.TryParse(stringValue[4..], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTimeOffset date)) - { - return date; - } - } - - // Otherwise just return as string + string stringValue = value.StringValue; + + // If the value starts with the 'dt:' prefix, it may represent a DateTime value � attempt to parse it. + if (stringValue.StartsWith("dt:", StringComparison.Ordinal)) + { + if (DateTime.TryParse(stringValue[3..], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTime date)) + { + return date; + } + } + + // If the value starts with the 'dto:' prefix, it may represent a DateTime value � attempt to parse it. + if (stringValue.StartsWith("dto:", StringComparison.Ordinal)) + { + if (DateTimeOffset.TryParse(stringValue[4..], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTimeOffset date)) + { + return date; + } + } + + // Otherwise just return as string return stringValue; case Google.Protobuf.WellKnownTypes.Value.KindOneofCase.BoolValue: return value.BoolValue; @@ -1082,16 +1093,16 @@ internal static T Base64Decode(this MessageParser parser, string encodedMessa pair => ConvertValueToObject(pair.Value)); case Google.Protobuf.WellKnownTypes.Value.KindOneofCase.ListValue: return value.ListValue.Values.Select(ConvertValueToObject).ToList(); - default: - // Fallback: serialize the whole value to JSON string + default: + // Fallback: serialize the whole value to JSON string return JsonSerializer.Serialize(value); } - } - + } + /// /// Converts a MapFieldinto a IDictionary. /// - /// The map to convert. + /// The map to convert. /// Dictionary contains the converted obejct. internal static IDictionary ConvertProperties(MapField properties) { @@ -1116,8 +1127,8 @@ internal static Value ConvertObjectToValue(object? obj) long l => Value.ForNumber(l), float f => Value.ForNumber(f), double d => Value.ForNumber(d), - decimal dec => Value.ForNumber((double)dec), - + decimal dec => Value.ForNumber((double)dec), + // For DateTime and DateTimeOffset, add prefix to distinguish from normal string. DateTime dt => Value.ForString($"dt:{dt.ToString("O")}"), DateTimeOffset dto => Value.ForString($"dto:{dto.ToString("O")}"), @@ -1125,9 +1136,9 @@ internal static Value ConvertObjectToValue(object? obj) { Fields = { dict.ToDictionary(kvp => kvp.Key, kvp => ConvertObjectToValue(kvp.Value)) }, }), - IEnumerable e => Value.ForList(e.Cast().Select(ConvertObjectToValue).ToArray()), - - // Fallback: convert unlisted type to string. + IEnumerable e => Value.ForList(e.Cast().Select(ConvertObjectToValue).ToArray()), + + // Fallback: convert unlisted type to string. _ => Value.ForString(obj.ToString() ?? string.Empty), }; } @@ -1177,28 +1188,28 @@ static P.OrchestrationInstance ToProtobuf(this OrchestrationInstance instance) InstanceId = instance.InstanceId, ExecutionId = instance.ExecutionId, }; - } - - static P.HistoryEvent ToProtobuf(HistoryEvent e) - { - var payload = new P.HistoryEvent() - { - EventId = e.EventId, - Timestamp = Timestamp.FromDateTime(e.Timestamp), - }; - - if (e.EventType == EventType.EventRaised) - { - var eventRaised = (EventRaisedEvent)e; - payload.EventRaised = new P.EventRaisedEvent - { - Name = eventRaised.Name, - Input = eventRaised.Input, - }; - return payload; - } - - throw new ArgumentException("Unsupported event type"); + } + + static P.HistoryEvent ToProtobuf(HistoryEvent e) + { + var payload = new P.HistoryEvent() + { + EventId = e.EventId, + Timestamp = Timestamp.FromDateTime(e.Timestamp), + }; + + if (e.EventType == EventType.EventRaised) + { + var eventRaised = (EventRaisedEvent)e; + payload.EventRaised = new P.EventRaisedEvent + { + Name = eventRaised.Name, + Input = eventRaised.Input, + }; + return payload; + } + + throw new ArgumentException("Unsupported event type"); } /// diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 5dd18d522..266514af8 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -247,16 +247,42 @@ static string GetActionsListForLogging(IReadOnlyList actio return failureDetails; } - async ValueTask BuildRuntimeStateAsync( + static OrchestrationRuntimeState BuildRuntimeState( P.OrchestratorRequest orchestratorRequest, - ProtoUtils.EntityConversionState? entityConversionState, - CancellationToken cancellation) + IReadOnlyList pastEvents, + ProtoUtils.EntityConversionState? entityConversionState) { Func converter = entityConversionState is null ? ProtoUtils.ConvertHistoryEvent : entityConversionState.ConvertFromProto; - List pastEvents; + List convertedPastEvents = new(pastEvents.Count); + foreach (P.HistoryEvent protoEvent in pastEvents) + { + convertedPastEvents.Add(converter(protoEvent)); + } + + // Reconstruct the orchestration state in a way that correctly distinguishes new events from past events + var runtimeState = new OrchestrationRuntimeState(convertedPastEvents); + foreach (P.HistoryEvent protoEvent in orchestratorRequest.NewEvents) + { + // AddEvent() puts events into the NewEvents list. + runtimeState.AddEvent(converter(protoEvent)); + } + + if (runtimeState.ExecutionStartedEvent == null) + { + // TODO: What's the right way to handle this? Callback to the sidecar with a retriable error request? + throw new InvalidOperationException("The provided orchestration history was incomplete"); + } + + return runtimeState; + } + + async ValueTask> GetPastEventsAsync( + P.OrchestratorRequest orchestratorRequest, + CancellationToken cancellation) + { if (orchestratorRequest.RequiresHistoryStreaming) { // Stream the remaining events from the remote service @@ -280,40 +306,19 @@ async ValueTask BuildRuntimeStateAsync( // chunks (e.g. one event per chunk) that would reallocate and copy on every chunk, which is // itself quadratic. List.Add's built-in geometric (doubling) growth already gives // amortized O(1) appends, so we let it manage capacity on its own. - pastEvents = new List(); + List pastEvents = new(); await foreach (P.HistoryChunk chunk in streamResponse.ResponseStream.ReadAllAsync(cancellation)) { foreach (P.HistoryEvent protoEvent in chunk.Events) { - pastEvents.Add(converter(protoEvent)); + pastEvents.Add(protoEvent); } } - } - else - { - // The history was already provided in the work item request - pastEvents = new List(orchestratorRequest.PastEvents.Count); - foreach (P.HistoryEvent protoEvent in orchestratorRequest.PastEvents) - { - pastEvents.Add(converter(protoEvent)); - } - } - // Reconstruct the orchestration state in a way that correctly distinguishes new events from past events - var runtimeState = new OrchestrationRuntimeState(pastEvents); - foreach (P.HistoryEvent protoEvent in orchestratorRequest.NewEvents) - { - // AddEvent() puts events into the NewEvents list. - runtimeState.AddEvent(converter(protoEvent)); + return pastEvents; } - if (runtimeState.ExecutionStartedEvent == null) - { - // TODO: What's the right way to handle this? Callback to the sidecar with a retriable error request? - throw new InvalidOperationException("The provided orchestration history was incomplete"); - } - - return runtimeState; + return orchestratorRequest.PastEvents; } async Task> ConnectAsync(CancellationToken cancellation) @@ -602,25 +607,88 @@ async Task OnRunOrchestratorAsync( string completionToken, CancellationToken cancellationToken) { + P.ExecutionRewoundEvent? rewindEvent = request + .NewEvents + .Where(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionRewound) + .Select(e => e.ExecutionRewound) + .LastOrDefault(); + + IReadOnlyList? materializedPastEvents = null; + bool isInitialRewind = false; + if (rewindEvent is not null || request.RequiresHistoryStreaming) + { + materializedPastEvents = await this.GetPastEventsAsync(request, cancellationToken); + } + + if (rewindEvent is not null) + { + // The initial request still has the terminal event. After the history is rewritten, + // a second rewind event is used only to jump-start normal orchestration execution. + P.ExecutionCompletedEvent? completedEvent = materializedPastEvents! + .Where(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted) + .Select(e => e.ExecutionCompleted) + .LastOrDefault(); + + if (completedEvent is not null) + { + if (completedEvent.OrchestrationStatus != P.OrchestrationStatus.Failed) + { + throw new InvalidOperationException( + "Expected a rewind request's ExecutionCompleted event to have status Failed, " + + $"but found '{completedEvent.OrchestrationStatus}'."); + } + + isInitialRewind = true; + } + } + + IReadOnlyList pastEvents = materializedPastEvents ?? request.PastEvents; var executionStartedEvent = request .NewEvents - .Concat(request.PastEvents) + .Concat(pastEvents) .Where(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) .Select(e => e.ExecutionStarted) .FirstOrDefault(); + if (isInitialRewind + && rewindEvent!.ParentTraceContext is not null) + { + if (executionStartedEvent is null) + { + throw new InvalidOperationException("Rewinding orchestration has no ExecutionStartedEvent in its history"); + } + + executionStartedEvent = executionStartedEvent.Clone(); + executionStartedEvent.ParentTraceContext = rewindEvent.ParentTraceContext; + } + + // A rewind starts a new orchestration span instead of continuing the failed execution's stored span. + P.OrchestrationTraceContext? orchestrationTraceContext = + isInitialRewind ? null : request.OrchestrationTraceContext; Activity? traceActivity = TraceHelper.StartTraceActivityForOrchestrationExecution( executionStartedEvent, - request.OrchestrationTraceContext); + orchestrationTraceContext); + + if (isInitialRewind) + { + await this.CompleteOrchestratorTaskWithChunkingAsync( + RewindOrchestrationHandler.CreateResponse( + request, + pastEvents, + completionToken, + traceActivity), + this.worker.grpcOptions.CompleteOrchestrationWorkItemChunkSizeInBytes, + cancellationToken); + return; + } if (executionStartedEvent is not null) { P.HistoryEvent? GetSuborchestrationInstanceCreatedEvent(int eventId) { var subOrchestrationEvent = - request - .PastEvents + pastEvents .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated) .FirstOrDefault(x => x.EventId == eventId); @@ -630,8 +698,7 @@ async Task OnRunOrchestratorAsync( P.HistoryEvent? GetTaskScheduledEvent(int eventId) { var taskScheduledEvent = - request - .PastEvents + pastEvents .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled) .LastOrDefault(x => x.EventId == eventId); @@ -718,10 +785,10 @@ async Task OnRunOrchestratorAsync( bool versionFailure = false; try { - OrchestrationRuntimeState runtimeState = await this.BuildRuntimeStateAsync( + OrchestrationRuntimeState runtimeState = BuildRuntimeState( request, - entityConversionState, - cancellationToken); + pastEvents, + entityConversionState); bool filterPassed = true; if (this.orchestrationFilter != null) diff --git a/src/Worker/Grpc/RewindOrchestrationHandler.cs b/src/Worker/Grpc/RewindOrchestrationHandler.cs new file mode 100644 index 000000000..b6f1cb134 --- /dev/null +++ b/src/Worker/Grpc/RewindOrchestrationHandler.cs @@ -0,0 +1,146 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Diagnostics; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc; + +/// +/// Builds the worker response for an orchestration rewind request. +/// +static class RewindOrchestrationHandler +{ + /// + /// Creates an orchestrator response containing the replacement history for a rewind. + /// + /// The orchestrator request. + /// The complete committed orchestration history. + /// The completion token for the work item. + /// The current orchestration trace activity. + /// The response containing a single rewind action. + internal static P.OrchestratorResponse CreateResponse( + P.OrchestratorRequest request, + IReadOnlyList pastEvents, + string completionToken, + Activity? orchestrationActivity) + { + Check.NotNull(request); + Check.NotNull(pastEvents); + + if (request.NewEvents.Count != 2 + || request.NewEvents[1].EventTypeCase != P.HistoryEvent.EventTypeOneofCase.ExecutionRewound) + { + throw new InvalidOperationException( + "When rewinding an orchestration, the new events list must contain exactly two events: " + + "OrchestratorStarted and ExecutionRewound."); + } + + P.ExecutionRewoundEvent rewindEvent = request.NewEvents[1].ExecutionRewound; + List allEvents = [.. pastEvents, .. request.NewEvents]; + P.HistoryEvent? executionStartedEvent = allEvents.FirstOrDefault( + e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted); + + P.TraceContext? orchestrationParentTraceContext = rewindEvent.ParentTraceContext + ?? executionStartedEvent?.ExecutionStarted.ParentTraceContext; + ActivityContext orchestrationParentContext = default; + bool hasOrchestrationParentContext = false; + + HashSet failedTaskIds = []; + foreach (P.HistoryEvent historyEvent in allEvents) + { + if (historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskFailed) + { + failedTaskIds.Add(historyEvent.TaskFailed.TaskScheduledId); + } + else if (historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed) + { + failedTaskIds.Add(historyEvent.SubOrchestrationInstanceFailed.TaskScheduledId); + } + } + + string newExecutionId = Guid.NewGuid().ToString("N"); + P.RewindOrchestrationAction rewindAction = new(); + + // Retry timers are retained to match the existing rewind protocol. Rewinding failed activities + // that were scheduled with retry policies is not currently supported. + foreach (P.HistoryEvent historyEvent in allEvents) + { + // Do not add any failed tasks or the failed execution completed event to the new history + if (historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskFailed + || historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed + || historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted + || (historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled + && failedTaskIds.Contains(historyEvent.EventId))) + { + continue; + } + + // Modify the execution started event to reflect the new execution ID and new parent execution ID (if applicable) + if (historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) + { + P.HistoryEvent eventCopy = historyEvent.Clone(); + eventCopy.ExecutionStarted.OrchestrationInstance.ExecutionId = newExecutionId; + + if (!string.IsNullOrEmpty(rewindEvent.ParentExecutionId) + && eventCopy.ExecutionStarted.ParentInstance?.OrchestrationInstance is { } parentInstance) + { + parentInstance.ExecutionId = rewindEvent.ParentExecutionId; + } + + if (rewindEvent.ParentTraceContext is not null) + { + eventCopy.ExecutionStarted.ParentTraceContext = rewindEvent.ParentTraceContext.Clone(); + } + + hasOrchestrationParentContext = ActivityContext.TryParse( + eventCopy.ExecutionStarted.ParentTraceContext?.TraceParent, + eventCopy.ExecutionStarted.ParentTraceContext?.TraceState, + out orchestrationParentContext); + + rewindAction.NewHistory.Add(eventCopy); + continue; + } + + if (historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated + && failedTaskIds.Contains(historyEvent.EventId) + && hasOrchestrationParentContext) + { + // We set a new client span ID here so that the execution of the rewound suborchestration is not tied to the + // old parent. + ActivityContext newParentTraceContext = new( + orchestrationParentContext.TraceId, + ActivitySpanId.CreateRandom(), + orchestrationParentContext.TraceFlags, + orchestrationParentContext.TraceState); + + P.HistoryEvent eventCopy = historyEvent.Clone(); + eventCopy.SubOrchestrationInstanceCreated.ParentTraceContext = + ProtoUtils.CreateTraceContext(newParentTraceContext); + rewindAction.NewHistory.Add(eventCopy); + continue; + } + + rewindAction.NewHistory.Add(historyEvent); + } + + return new P.OrchestratorResponse + { + InstanceId = request.InstanceId, + CompletionToken = completionToken, + OrchestrationTraceContext = new() + { + SpanID = orchestrationActivity?.SpanId.ToString(), + SpanStartTime = orchestrationActivity?.StartTimeUtc.ToTimestamp(), + }, + Actions = + { + new P.OrchestratorAction + { + Id = -1, + RewindOrchestration = rewindAction, + }, + }, + }; + } +} diff --git a/test/Grpc.IntegrationTests/DtsEmulatorFactAttribute.cs b/test/Grpc.IntegrationTests/DtsEmulatorFactAttribute.cs new file mode 100644 index 000000000..0a42226f9 --- /dev/null +++ b/test/Grpc.IntegrationTests/DtsEmulatorFactAttribute.cs @@ -0,0 +1,19 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace Microsoft.DurableTask.Grpc.Tests; + +sealed class DtsEmulatorFactAttribute : FactAttribute +{ + internal const string ConnectionStringEnvironmentVariable = "DTS_EMULATOR_CONNECTION_STRING"; + + public DtsEmulatorFactAttribute() + { + if (string.IsNullOrWhiteSpace( + Environment.GetEnvironmentVariable(ConnectionStringEnvironmentVariable))) + { + this.Skip = + $"Set {ConnectionStringEnvironmentVariable} to run tests against a DTS emulator."; + } + } +} diff --git a/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs b/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs new file mode 100644 index 000000000..d5f245d52 --- /dev/null +++ b/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs @@ -0,0 +1,538 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Collections.Concurrent; +using System.Diagnostics; +using Microsoft.DurableTask.Client; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Tests.Logging; +using Microsoft.DurableTask.Worker; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Xunit.Abstractions; + +namespace Microsoft.DurableTask.Grpc.Tests; + +/// +/// End-to-end rewind tests that run against the DTS emulator. +/// +[Trait("Category", "DtsEmulator")] +public sealed class DtsRewindIntegrationTests : IDisposable +{ + readonly CancellationTokenSource testTimeoutSource = + new(Debugger.IsAttached ? TimeSpan.FromMinutes(5) : TimeSpan.FromSeconds(60)); + readonly TestLogProvider logProvider; + + /// + /// Initializes a new instance of the class. + /// + /// The test output helper. + public DtsRewindIntegrationTests(ITestOutputHelper output) + { + this.logProvider = new(output); + } + + CancellationToken TimeoutToken => this.testTimeoutSource.Token; + + static string ConnectionString => + Environment.GetEnvironmentVariable(DtsEmulatorFactAttribute.ConnectionStringEnvironmentVariable) + ?? throw new InvalidOperationException( + $"{DtsEmulatorFactAttribute.ConnectionStringEnvironmentVariable} is not set."); + + /// + /// Verifies that rewinding re-executes a failed activity. + /// + [DtsEmulatorFact] + public async Task RewindFailedActivityAsync() + { + // Arrange + TaskName orchestratorName = nameof(RewindFailedActivityAsync); + TaskName activityName = $"{nameof(RewindFailedActivityAsync)}_Activity"; + int activityCallCount = 0; + int shouldFail = 1; + + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks + .AddOrchestratorFunc( + orchestratorName, + async (context, input) => + await context.CallActivityAsync(activityName, input)) + .AddActivityFunc(activityName, (context, input) => + { + Interlocked.Increment(ref activityCallCount); + if (Volatile.Read(ref shouldFail) == 1) + { + throw new InvalidOperationException("Simulated failure."); + } + + return Task.FromResult($"Hello, {input}!"); + })); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + orchestratorName, + "World", + cancellation: this.TimeoutToken); + OrchestrationMetadata failed = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); + + // Act + Volatile.Write(ref shouldFail, 0); + await server.Client.RewindInstanceAsync(instanceId, "retry after fix", this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal("Hello, World!", completed.ReadOutputAs()); + Assert.Null(completed.FailureDetails); + Assert.Equal(2, Volatile.Read(ref activityCallCount)); + } + + /// + /// Verifies that rewind replays successful activity results and re-executes only the failed activity. + /// + [DtsEmulatorFact] + public async Task RewindPreservesSuccessfulResultsAsync() + { + // Arrange + TaskName orchestratorName = nameof(RewindPreservesSuccessfulResultsAsync); + TaskName firstActivityName = $"{nameof(RewindPreservesSuccessfulResultsAsync)}_First"; + TaskName secondActivityName = $"{nameof(RewindPreservesSuccessfulResultsAsync)}_Second"; + ConcurrentDictionary callCounts = []; + int shouldFailSecond = 1; + + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks + .AddOrchestratorFunc( + orchestratorName, + async (context, input) => + { + string first = await context.CallActivityAsync(firstActivityName, input); + string second = await context.CallActivityAsync(secondActivityName, input); + return [first, second]; + }) + .AddActivityFunc(firstActivityName, (context, input) => + { + callCounts.AddOrUpdate("first", 1, (_, count) => count + 1); + return Task.FromResult($"first:{input}"); + }) + .AddActivityFunc(secondActivityName, (context, input) => + { + callCounts.AddOrUpdate("second", 1, (_, count) => count + 1); + if (Volatile.Read(ref shouldFailSecond) == 1) + { + throw new InvalidOperationException("Temporary failure."); + } + + return Task.FromResult($"second:{input}"); + })); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + orchestratorName, + "test", + cancellation: this.TimeoutToken); + OrchestrationMetadata failed = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); + + // Act + Volatile.Write(ref shouldFailSecond, 0); + await server.Client.RewindInstanceAsync(instanceId, "retry", this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + string[] output = completed.ReadOutputAs() + ?? throw new InvalidOperationException("The orchestration output was missing."); + Assert.Equal(["first:test", "second:test"], output); + Assert.Null(completed.FailureDetails); + Assert.Equal(1, callCounts["first"]); + Assert.Equal(2, callCounts["second"]); + } + + /// + /// Verifies that rewinding a nonexistent orchestration fails. + /// + [DtsEmulatorFact] + public async Task RewindNonexistentOrchestrationThrowsAsync() + { + // Arrange + await using HostTestLifetime server = await this.StartClientAsync(); + string instanceId = $"nonexistent-{Guid.NewGuid():N}"; + + // Act + Func act = () => server.Client.RewindInstanceAsync( + instanceId, + "should fail", + this.TimeoutToken); + + // Assert + await Assert.ThrowsAsync(act); + } + + /// + /// Verifies that rewinding a completed orchestration fails. + /// + [DtsEmulatorFact] + public async Task RewindCompletedOrchestrationThrowsAsync() + { + // Arrange + TaskName orchestratorName = nameof(RewindCompletedOrchestrationThrowsAsync); + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks.AddOrchestratorFunc( + orchestratorName, + context => Task.FromResult("done"))); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + orchestratorName, + cancellation: this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + + // Act + Func act = () => server.Client.RewindInstanceAsync( + instanceId, + "should fail", + this.TimeoutToken); + + // Assert + await Assert.ThrowsAsync(act); + } + + /// + /// Verifies that rewind recursively re-executes a failed sub-orchestration. + /// + [DtsEmulatorFact] + public async Task RewindFailedSubOrchestrationAsync() + { + // Arrange + TaskName parentOrchestratorName = $"{nameof(RewindFailedSubOrchestrationAsync)}_Parent"; + TaskName childOrchestratorName = $"{nameof(RewindFailedSubOrchestrationAsync)}_Child"; + TaskName activityName = $"{nameof(RewindFailedSubOrchestrationAsync)}_Activity"; + int activityCallCount = 0; + + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks + .AddOrchestratorFunc( + parentOrchestratorName, + async (context, input) => + { + string result = await context.CallSubOrchestratorAsync( + childOrchestratorName, + input); + return $"parent:{result}"; + }) + .AddOrchestratorFunc( + childOrchestratorName, + async (context, input) => + await context.CallActivityAsync(activityName, input)) + .AddActivityFunc(activityName, (context, input) => + { + int count = Interlocked.Increment(ref activityCallCount); + if (count == 1) + { + throw new InvalidOperationException("Child failure."); + } + + return Task.FromResult($"child:{input}"); + })); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + parentOrchestratorName, + "data", + cancellation: this.TimeoutToken); + OrchestrationMetadata failed = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); + + // Act + await server.Client.RewindInstanceAsync(instanceId, "sub-orchestration fix", this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal("parent:child:data", completed.ReadOutputAs()); + Assert.Equal(2, Volatile.Read(ref activityCallCount)); + } + + /// + /// Verifies that a purged failed sub-orchestration is recreated when its parent is rewound. + /// + [DtsEmulatorFact] + public async Task RewindPurgedSubOrchestrationAsync() + { + // Arrange + TaskName parentOrchestratorName = $"{nameof(RewindPurgedSubOrchestrationAsync)}_Parent"; + TaskName childOrchestratorName = $"{nameof(RewindPurgedSubOrchestrationAsync)}_Child"; + TaskName activityName = $"{nameof(RewindPurgedSubOrchestrationAsync)}_Activity"; + string childInstanceId = $"child-{Guid.NewGuid():N}"; + const string ChildVersion = "v1"; + SubOrchestrationOptions childOptions = new(instanceId: childInstanceId) + { + Version = ChildVersion, + Tags = new Dictionary + { + ["scenario"] = "purged-sub-orchestration", + ["preserve"] = "true", + }, + }; + int activityCallCount = 0; + + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks + .AddOrchestratorFunc( + parentOrchestratorName, + async (context, input) => + { + string result = await context.CallSubOrchestratorAsync( + childOrchestratorName, + input, + childOptions); + return $"parent:{result}"; + }) + .AddOrchestratorFunc( + childOrchestratorName, + async (context, input) => + { + string result = await context.CallActivityAsync(activityName, input); + return $"{context.Version}:{result}"; + }) + .AddActivityFunc(activityName, (context, input) => + { + int count = Interlocked.Increment(ref activityCallCount); + if (count == 1) + { + throw new InvalidOperationException("Child failure."); + } + + return Task.FromResult($"child:{input}"); + })); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + parentOrchestratorName, + "data", + cancellation: this.TimeoutToken); + OrchestrationMetadata failed = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); + + PurgeResult purgeResult = await server.Client.PurgeInstanceAsync( + childInstanceId, + this.TimeoutToken); + Assert.Equal(1, purgeResult.PurgedInstanceCount); + + // Act + await server.Client.RewindInstanceAsync(instanceId, "purge and retry", this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal($"parent:{ChildVersion}:child:data", completed.ReadOutputAs()); + Assert.Equal(2, Volatile.Read(ref activityCallCount)); + + OrchestrationMetadata recreatedChild = await server.Client.GetInstanceAsync( + childInstanceId, + getInputsAndOutputs: true, + this.TimeoutToken) + ?? throw new InvalidOperationException("The recreated child orchestration was not found."); + Assert.Equal(childOrchestratorName.Name, recreatedChild.Name); + Assert.Equal(childInstanceId, recreatedChild.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, recreatedChild.RuntimeStatus); + Assert.Equal("data", recreatedChild.ReadInputAs()); + Assert.Equal($"{ChildVersion}:child:data", recreatedChild.ReadOutputAs()); + Assert.Equal(childOptions.Tags, recreatedChild.Tags); + } + + /// + /// Verifies that rewind accepts an empty reason. + /// + [DtsEmulatorFact] + public async Task RewindWithoutReasonAsync() + { + // Arrange + TaskName orchestratorName = nameof(RewindWithoutReasonAsync); + TaskName activityName = $"{nameof(RewindWithoutReasonAsync)}_Activity"; + int activityCallCount = 0; + + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks + .AddOrchestratorFunc( + orchestratorName, + async context => await context.CallActivityAsync(activityName)) + .AddActivityFunc(activityName, (TaskActivityContext context) => + { + int count = Interlocked.Increment(ref activityCallCount); + if (count == 1) + { + throw new InvalidOperationException("Simulated failure."); + } + + return Task.FromResult("ok"); + })); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + orchestratorName, + cancellation: this.TimeoutToken); + OrchestrationMetadata failed = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); + + // Act + await server.Client.RewindInstanceAsync(instanceId, string.Empty, this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal("ok", completed.ReadOutputAs()); + } + + /// + /// Verifies that the same orchestration can be rewound more than once. + /// + [DtsEmulatorFact] + public async Task RewindTwiceAsync() + { + // Arrange + TaskName orchestratorName = nameof(RewindTwiceAsync); + TaskName activityName = $"{nameof(RewindTwiceAsync)}_Activity"; + int activityCallCount = 0; + + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks + .AddOrchestratorFunc( + orchestratorName, + async (context, input) => + await context.CallActivityAsync(activityName, input)) + .AddActivityFunc(activityName, (context, input) => + { + int count = Interlocked.Increment(ref activityCallCount); + if (count <= 2) + { + throw new InvalidOperationException($"Failure #{count}."); + } + + return Task.FromResult($"Hello, {input}!"); + })); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + orchestratorName, + "World", + cancellation: this.TimeoutToken); + OrchestrationMetadata firstFailure = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, firstFailure.RuntimeStatus); + + await server.Client.RewindInstanceAsync(instanceId, "first rewind", this.TimeoutToken); + OrchestrationMetadata secondFailure = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, secondFailure.RuntimeStatus); + + // Act + await server.Client.RewindInstanceAsync(instanceId, "second rewind", this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal("Hello, World!", completed.ReadOutputAs()); + Assert.Equal(3, Volatile.Read(ref activityCallCount)); + } + + /// + public void Dispose() + { + this.testTimeoutSource.Dispose(); + } + + async Task StartClientAsync() + { + return await this.StartHostAsync(configureWorker: null); + } + + async Task StartWorkerAsync(Action configureWorker) + { + return await this.StartHostAsync(configureWorker); + } + + async Task StartHostAsync(Action? configureWorker) + { + IHost host = Host.CreateDefaultBuilder() + .ConfigureLogging(logging => + { + logging.ClearProviders(); + logging.AddProvider(this.logProvider); + logging.SetMinimumLevel(LogLevel.Warning); + }) + .ConfigureServices(services => + { + if (configureWorker is not null) + { + services.AddDurableTaskWorker(builder => + { + configureWorker(builder); + builder.UseDurableTaskScheduler(ConnectionString); + }); + } + + services.AddDurableTaskClient(builder => + builder.UseDurableTaskScheduler(ConnectionString)); + }) + .Build(); + + try + { + await host.StartAsync(this.TimeoutToken); + return new HostTestLifetime(host, this.TimeoutToken); + } + catch + { + host.Dispose(); + throw; + } + } + + async Task WaitForCompletionAsync( + DurableTaskClient client, + string instanceId) + { + return await client.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + this.TimeoutToken); + } + + sealed class HostTestLifetime : IAsyncDisposable + { + readonly IHost host; + readonly CancellationToken cancellation; + + public HostTestLifetime(IHost host, CancellationToken cancellation) + { + this.host = host; + this.cancellation = cancellation; + this.Client = host.Services.GetRequiredService(); + } + + public DurableTaskClient Client { get; } + + public async ValueTask DisposeAsync() + { + try + { + await this.host.StopAsync(this.cancellation); + } + finally + { + this.host.Dispose(); + } + } + } +} diff --git a/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj b/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj index f891cc73a..3007a1014 100644 --- a/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj +++ b/test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj @@ -5,8 +5,10 @@ + + diff --git a/test/Grpc.IntegrationTests/run-dts-emulator-tests.ps1 b/test/Grpc.IntegrationTests/run-dts-emulator-tests.ps1 new file mode 100644 index 000000000..ec9fcc7d1 --- /dev/null +++ b/test/Grpc.IntegrationTests/run-dts-emulator-tests.ps1 @@ -0,0 +1,85 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +param( + [string]$Configuration = "Debug", + [string]$Image = "mcr.microsoft.com/dts/dts-emulator:latest" +) + +$ErrorActionPreference = "Stop" +$containerName = "durabletask-dotnet-dts-$([Guid]::NewGuid().ToString('N'))" +$previousConnectionString = $env:DTS_EMULATOR_CONNECTION_STRING +$containerStarted = $false +$httpClient = $null + +try { + docker info | Out-Null + if ($LASTEXITCODE -ne 0) { + throw "Docker is not available." + } + + docker run ` + --name $containerName ` + --rm ` + --detach ` + --pull always ` + --publish "127.0.0.1::8080" ` + $Image | Out-Null + if ($LASTEXITCODE -ne 0) { + throw "Failed to start the DTS emulator container." + } + + $containerStarted = $true + $publishedPort = docker port $containerName "8080/tcp" + if ($LASTEXITCODE -ne 0 -or $publishedPort -notmatch ':(\d+)$') { + throw "Failed to determine the DTS emulator port." + } + + $port = [int]$Matches[1] + $ready = $false + $deadline = [DateTime]::UtcNow.AddSeconds(30) + $httpClient = [System.Net.Http.HttpClient]::new() + $httpClient.Timeout = [TimeSpan]::FromSeconds(1) + while ([DateTime]::UtcNow -lt $deadline) { + try { + $response = $httpClient.GetAsync( + "http://127.0.0.1:$port/").GetAwaiter().GetResult() + $response.Dispose() + $ready = $true + break + } + catch [System.Net.Http.HttpRequestException] { + } + catch [System.Threading.Tasks.TaskCanceledException] { + } + + Start-Sleep -Milliseconds 250 + } + + if (-not $ready) { + docker logs $containerName + throw "The DTS emulator did not become ready within 30 seconds." + } + + $env:DTS_EMULATOR_CONNECTION_STRING = + "Endpoint=http://127.0.0.1:$port;TaskHub=default;Authentication=None" + + dotnet test ` + "$PSScriptRoot\Grpc.IntegrationTests.csproj" ` + --configuration $Configuration ` + --filter "Category=DtsEmulator" + if ($LASTEXITCODE -ne 0) { + docker logs $containerName + throw "The DTS emulator integration tests failed." + } +} +finally { + if ($null -ne $httpClient) { + $httpClient.Dispose() + } + + $env:DTS_EMULATOR_CONNECTION_STRING = $previousConnectionString + if ($containerStarted) { + docker rm --force $containerName | Out-Null + } +} diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index 816dfb874..41e80de46 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -2,6 +2,7 @@ // Licensed under the MIT License. using System.Collections.Concurrent; +using System.Diagnostics; using System.IO; using System.Reflection; using DurableTask.Core; @@ -12,6 +13,7 @@ using Microsoft.DurableTask.Tests.Logging; using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc.Internal; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using P = Microsoft.DurableTask.Protobuf; @@ -36,9 +38,15 @@ public class GrpcDurableTaskWorkerTests .GetMethod("DispatchWorkItem", BindingFlags.Instance | BindingFlags.NonPublic)!; static readonly MethodInfo TryRecreateChannelAsyncMethod = typeof(GrpcDurableTaskWorker) .GetMethod("TryRecreateChannelAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; - static readonly MethodInfo BuildRuntimeStateAsyncMethod = typeof(GrpcDurableTaskWorker) + static readonly MethodInfo BuildRuntimeStateMethod = typeof(GrpcDurableTaskWorker) .GetNestedType("Processor", BindingFlags.NonPublic)! - .GetMethod("BuildRuntimeStateAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + .GetMethod("BuildRuntimeState", BindingFlags.Static | BindingFlags.NonPublic)!; + static readonly MethodInfo GetPastEventsAsyncMethod = typeof(GrpcDurableTaskWorker) + .GetNestedType("Processor", BindingFlags.NonPublic)! + .GetMethod("GetPastEventsAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; + static readonly MethodInfo OnRunOrchestratorAsyncMethod = typeof(GrpcDurableTaskWorker) + .GetNestedType("Processor", BindingFlags.NonPublic)! + .GetMethod("OnRunOrchestratorAsync", BindingFlags.Instance | BindingFlags.NonPublic)!; [Fact] public async Task ExecuteAsync_ConnectFailureThreshold_RecreatesConfiguredChannel() @@ -664,6 +672,645 @@ public async Task BuildRuntimeStateAsync_MultiChunkHistoryStream_MaterializesEve runtimeState.ExecutionStartedEvent!.Name.Should().Be("TestOrchestration"); } + [Fact] + public async Task OnRunOrchestratorAsync_StreamingHistory_StartsTraceFromExecutionStarted() + { + // Arrange + const string ParentTraceId = "11111111111111111111111111111111"; + const string ParentSpanId = "2222222222222222"; + SequenceAsyncStreamReader historyReader = new( + new P.HistoryChunk + { + Events = + { + new P.HistoryEvent + { + Timestamp = Timestamp.FromDateTime( + new DateTime(2025, 1, 1, 0, 0, 0, DateTimeKind.Utc)), + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = "TestOrchestration", + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + }, + ParentTraceContext = new P.TraceContext + { + TraceParent = $"00-{ParentTraceId}-{ParentSpanId}-01", + }, + }, + }, + }, + }); + + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + clientMock + .Setup(client => client.StreamInstanceHistory( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(CreateServerStreamingCall(historyReader)); + clientMock + .Setup(client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(CreateUnaryCall(Task.FromResult(new P.CompleteTaskResponse()))); + + Mock factoryMock = new(MockBehavior.Strict); + factoryMock + .Setup(factory => factory.TryCreateOrchestrator( + It.IsAny(), + It.IsAny(), + out It.Ref.IsAny)) + .Returns(false); + using ServiceProvider services = new ServiceCollection().BuildServiceProvider(); + GrpcDurableTaskWorker worker = CreateWorker( + new GrpcDurableTaskWorkerOptions(), + new DurableTaskWorkerOptions(), + NullLoggerFactory.Instance, + factoryMock.Object, + services); + + P.OrchestratorRequest request = new() + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + RequiresHistoryStreaming = true, + NewEvents = + { + new P.HistoryEvent + { + Timestamp = Timestamp.FromDateTime( + new DateTime(2025, 1, 1, 0, 0, 1, DateTimeKind.Utc)), + OrchestratorStarted = new P.OrchestratorStartedEvent(), + }, + }, + }; + + Activity? orchestrationActivity = null; + using ActivityListener listener = new() + { + ShouldListenTo = source => source.Name == "Microsoft.DurableTask", + Sample = (ref ActivityCreationOptions options) => + ActivitySamplingResult.AllDataAndRecorded, + ActivityStarted = activity => orchestrationActivity = activity, + }; + ActivitySource.AddActivityListener(listener); + + // Act + await InvokeOnRunOrchestratorAsync( + CreateProcessor(worker, clientMock.Object), + request, + "completion-token", + CancellationToken.None); + + // Assert + orchestrationActivity.Should().NotBeNull(); + orchestrationActivity!.TraceId.Should().Be(ActivityTraceId.CreateFromString(ParentTraceId.AsSpan())); + orchestrationActivity.ParentSpanId.Should().Be(ActivitySpanId.CreateFromString(ParentSpanId.AsSpan())); + clientMock.VerifyAll(); + factoryMock.VerifyAll(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task OnRunOrchestratorAsync_RewindRequest_CreatesFreshTraceContext( + bool rewindHasParentTraceContext) + { + // Arrange + const string ExecutionTraceId = "11111111111111111111111111111111"; + const string ExecutionParentSpanId = "2222222222222222"; + const string ExecutionTraceState = "execution=value"; + const string PreviousOrchestrationSpanId = "3333333333333333"; + const string RewindTraceId = "44444444444444444444444444444444"; + const string RewindParentSpanId = "5555555555555555"; + const string RewindTraceState = "rewind=value"; + + P.ExecutionRewoundEvent rewindEvent = new() { Reason = "fixed" }; + if (rewindHasParentTraceContext) + { + rewindEvent.ParentTraceContext = new P.TraceContext + { + TraceParent = $"00-{RewindTraceId}-{RewindParentSpanId}-01", + TraceState = RewindTraceState, + }; + } + + P.OrchestratorRequest request = new() + { + InstanceId = "instance-1", + OrchestrationTraceContext = new P.OrchestrationTraceContext + { + SpanID = PreviousOrchestrationSpanId, + SpanStartTime = Timestamp.FromDateTime( + new DateTime(2025, 1, 1, 0, 0, 0, DateTimeKind.Utc)), + }, + PastEvents = + { + new P.HistoryEvent + { + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = "TestOrchestration", + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + }, + ParentTraceContext = new P.TraceContext + { + TraceParent = $"00-{ExecutionTraceId}-{ExecutionParentSpanId}-01", + TraceState = ExecutionTraceState, + }, + }, + }, + new P.HistoryEvent + { + ExecutionCompleted = new P.ExecutionCompletedEvent + { + OrchestrationStatus = P.OrchestrationStatus.Failed, + }, + }, + }, + NewEvents = + { + new P.HistoryEvent + { + OrchestratorStarted = new P.OrchestratorStartedEvent(), + }, + new P.HistoryEvent + { + ExecutionRewound = rewindEvent, + }, + }, + }; + + P.OrchestratorResponse? completedResponse = null; + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + clientMock + .Setup(client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback( + (response, _, _, _) => completedResponse = response) + .Returns(CreateUnaryCall(Task.FromResult(new P.CompleteTaskResponse()))); + + Mock factoryMock = new(MockBehavior.Strict); + GrpcDurableTaskWorker worker = CreateWorker( + new GrpcDurableTaskWorkerOptions(), + new DurableTaskWorkerOptions(), + NullLoggerFactory.Instance, + factoryMock.Object); + + Activity? rewindActivity = null; + using ActivityListener listener = new() + { + ShouldListenTo = source => source.Name == "Microsoft.DurableTask", + Sample = (ref ActivityCreationOptions options) => + ActivitySamplingResult.AllDataAndRecorded, + ActivityStarted = activity => rewindActivity = activity, + }; + ActivitySource.AddActivityListener(listener); + + try + { + // Act + await InvokeOnRunOrchestratorAsync( + CreateProcessor(worker, clientMock.Object), + request, + "completion-token", + CancellationToken.None); + + // Assert + factoryMock.VerifyNoOtherCalls(); + completedResponse.Should().NotBeNull(); + rewindActivity.Should().NotBeNull(); + P.OrchestratorResponse response = completedResponse!; + Activity activity = rewindActivity!; + response.OrchestrationTraceContext.SpanID.Should().NotBeNullOrEmpty(); + response.OrchestrationTraceContext.SpanID.Should().NotBe(PreviousOrchestrationSpanId); + response.OrchestrationTraceContext.SpanID.Should().Be(activity.SpanId.ToString()); + + string expectedTraceId = rewindHasParentTraceContext ? RewindTraceId : ExecutionTraceId; + string expectedParentSpanId = + rewindHasParentTraceContext ? RewindParentSpanId : ExecutionParentSpanId; + string expectedTraceState = + rewindHasParentTraceContext ? RewindTraceState : ExecutionTraceState; + activity.TraceId.Should().Be(ActivityTraceId.CreateFromString(expectedTraceId.AsSpan())); + activity.ParentSpanId.Should().Be( + ActivitySpanId.CreateFromString(expectedParentSpanId.AsSpan())); + activity.TraceStateString.Should().Be(expectedTraceState); + clientMock.VerifyAll(); + } + finally + { + rewindActivity?.Stop(); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task OnRunOrchestratorAsync_RewindRequest_ReturnsRewrittenHistory( + bool rewindHasParentTraceContext) + { + // Arrange + const string ExecutionTraceId = "11111111111111111111111111111111"; + const string ExecutionParentSpanId = "2222222222222222"; + const string ExecutionTraceState = "execution=value"; + const string SuccessfulChildSpanId = "3333333333333333"; + const string FailedChildSpanId = "4444444444444444"; + const string RewindTraceId = "55555555555555555555555555555555"; + const string RewindParentSpanId = "6666666666666666"; + const string RewindTraceState = "rewind=value"; + + P.TraceContext? rewindParentTraceContext = rewindHasParentTraceContext + ? new P.TraceContext + { + TraceParent = $"00-{RewindTraceId}-{RewindParentSpanId}-01", + TraceState = RewindTraceState, + } + : null; + + P.HistoryEvent executionStarted = new() + { + EventId = -1, + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = "TestOrchestration", + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "instance-1", + ExecutionId = "old-execution", + }, + ParentInstance = new P.ParentInstanceInfo + { + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "parent-instance", + ExecutionId = "old-parent-execution", + }, + }, + ParentTraceContext = new P.TraceContext + { + TraceParent = $"00-{ExecutionTraceId}-{ExecutionParentSpanId}-01", + TraceState = ExecutionTraceState, + }, + }, + }; + P.HistoryEvent successfulChildCreated = new() + { + EventId = 4, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = "successful-child", + ParentTraceContext = new P.TraceContext + { + TraceParent = $"00-{ExecutionTraceId}-{SuccessfulChildSpanId}-01", + }, + }, + }; + P.HistoryEvent failedChildCreated = new() + { + EventId = 6, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = "failed-child", + ParentTraceContext = new P.TraceContext + { + TraceParent = $"00-{ExecutionTraceId}-{FailedChildSpanId}-01", + }, + }, + }; + P.OrchestratorRequest request = new() + { + InstanceId = "instance-1", + PastEvents = + { + executionStarted, + new P.HistoryEvent + { + EventId = 0, + TaskScheduled = new P.TaskScheduledEvent { Name = "SuccessfulActivity" }, + }, + new P.HistoryEvent + { + EventId = 1, + TaskCompleted = new P.TaskCompletedEvent { TaskScheduledId = 0 }, + }, + new P.HistoryEvent + { + EventId = 2, + TaskScheduled = new P.TaskScheduledEvent { Name = "FailedActivity" }, + }, + new P.HistoryEvent + { + EventId = 3, + TaskFailed = new P.TaskFailedEvent { TaskScheduledId = 2 }, + }, + successfulChildCreated, + new P.HistoryEvent + { + EventId = 5, + SubOrchestrationInstanceCompleted = + new P.SubOrchestrationInstanceCompletedEvent { TaskScheduledId = 4 }, + }, + failedChildCreated, + new P.HistoryEvent + { + EventId = 7, + SubOrchestrationInstanceFailed = + new P.SubOrchestrationInstanceFailedEvent { TaskScheduledId = 6 }, + }, + new P.HistoryEvent + { + EventId = 8, + ExecutionCompleted = new P.ExecutionCompletedEvent + { + OrchestrationStatus = P.OrchestrationStatus.Failed, + }, + }, + }, + NewEvents = + { + new P.HistoryEvent + { + EventId = 9, + OrchestratorStarted = new P.OrchestratorStartedEvent(), + }, + new P.HistoryEvent + { + EventId = 10, + ExecutionRewound = new P.ExecutionRewoundEvent + { + Reason = "fixed", + ParentExecutionId = "new-parent-execution", + ParentTraceContext = rewindParentTraceContext, + }, + }, + }, + }; + + P.OrchestratorResponse? completedResponse = null; + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + clientMock + .Setup(client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback( + (response, _, _, _) => completedResponse = response) + .Returns(CreateUnaryCall(Task.FromResult(new P.CompleteTaskResponse()))); + + Mock factoryMock = new(MockBehavior.Strict); + GrpcDurableTaskWorker worker = CreateWorker( + new GrpcDurableTaskWorkerOptions(), + new DurableTaskWorkerOptions(), + NullLoggerFactory.Instance, + factoryMock.Object); + + // Act + await InvokeOnRunOrchestratorAsync( + CreateProcessor(worker, clientMock.Object), + request, + "completion-token", + CancellationToken.None); + + // Assert + factoryMock.VerifyNoOtherCalls(); + P.OrchestratorAction action = completedResponse!.Actions.Should().ContainSingle().Subject; + action.Id.Should().Be(-1); + action.OrchestratorActionTypeCase.Should().Be( + P.OrchestratorAction.OrchestratorActionTypeOneofCase.RewindOrchestration); + P.HistoryEvent[] newHistory = action.RewindOrchestration.NewHistory.ToArray(); + + // Confirm the shape of the new history (failed tasks removed, etc.) + newHistory.Should().HaveCount(8); + newHistory.Should().NotContain(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskFailed + || e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed + || e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted + || (e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled && e.EventId == 2)); + newHistory.Should().ContainSingle(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled && e.EventId == 0); + newHistory.Should().ContainSingle(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskCompleted && e.EventId == 1); + newHistory.Should().ContainSingle(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated + && e.EventId == 4); + newHistory.Should().ContainSingle(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted + && e.EventId == 5); + newHistory.Should().ContainSingle(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.OrchestratorStarted + && e.EventId == 9); + + // Confirm the new execution ID and trace state of the rewritten ExecutionStartedEvent + P.ExecutionStartedEvent rewrittenStart = newHistory + .Single(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) + .ExecutionStarted; + Guid.TryParseExact( + rewrittenStart.OrchestrationInstance.ExecutionId, + "N", + out _).Should().BeTrue(); + rewrittenStart.OrchestrationInstance.ExecutionId.Should().NotBe("old-execution"); + rewrittenStart.ParentInstance.OrchestrationInstance.ExecutionId + .Should().Be("new-parent-execution"); + rewrittenStart.ParentTraceContext.TraceParent.Should().Be( + rewindParentTraceContext?.TraceParent + ?? executionStarted.ExecutionStarted.ParentTraceContext.TraceParent); + rewrittenStart.ParentTraceContext.TraceState.Should().Be( + rewindParentTraceContext?.TraceState + ?? executionStarted.ExecutionStarted.ParentTraceContext.TraceState); + + // Confirm the new trace information of the failed suborchestration + P.SubOrchestrationInstanceCreatedEvent rewrittenFailedChild = newHistory + .Single(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated + && e.EventId == 6) + .SubOrchestrationInstanceCreated; + ActivityContext.TryParse( + rewrittenFailedChild.ParentTraceContext.TraceParent, + rewrittenFailedChild.ParentTraceContext.TraceState, + out ActivityContext failedChildTraceContext).Should().BeTrue(); + string expectedTraceId = rewindHasParentTraceContext ? RewindTraceId : ExecutionTraceId; + string expectedTraceState = rewindHasParentTraceContext ? RewindTraceState : ExecutionTraceState; + failedChildTraceContext.TraceId.Should().Be( + ActivityTraceId.CreateFromString(expectedTraceId.AsSpan())); + // A new span ID should be generated for the failed suborchestration + failedChildTraceContext.SpanId.ToString().Should().NotBe(FailedChildSpanId); + failedChildTraceContext.SpanId.ToString().Should().NotBe(ExecutionParentSpanId); + failedChildTraceContext.SpanId.ToString().Should().NotBe(RewindParentSpanId); + failedChildTraceContext.TraceFlags.Should().Be(ActivityTraceFlags.Recorded); + failedChildTraceContext.TraceState.Should().Be(expectedTraceState); + + // For successful suborchestrations, their trace information should not be altered + P.SubOrchestrationInstanceCreatedEvent rewrittenSuccessfulChild = newHistory + .Single(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated + && e.EventId == 4) + .SubOrchestrationInstanceCreated; + rewrittenSuccessfulChild.ParentTraceContext.TraceParent + .Should().Be(successfulChildCreated.SubOrchestrationInstanceCreated.ParentTraceContext.TraceParent); + + // Confirm the presence of the rewind event and its reason + P.HistoryEvent rewindEvent = newHistory.Should().ContainSingle(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionRewound + && e.EventId == 10).Subject; + rewindEvent.ExecutionRewound.Reason.Should().Be("fixed"); + + // The original events should remain unaltered + executionStarted.ExecutionStarted.OrchestrationInstance.ExecutionId.Should().Be("old-execution"); + executionStarted.ExecutionStarted.ParentInstance.OrchestrationInstance.ExecutionId + .Should().Be("old-parent-execution"); + failedChildCreated.SubOrchestrationInstanceCreated.ParentTraceContext.TraceParent + .Should().Be($"00-{ExecutionTraceId}-{FailedChildSpanId}-01"); + clientMock.VerifyAll(); + } + + [Fact] + public async Task OnRunOrchestratorAsync_RewindRequestForNonFailedOrchestration_Throws() + { + // Arrange + P.OrchestratorRequest request = new() + { + InstanceId = "instance-1", + PastEvents = + { + new P.HistoryEvent + { + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = "TestOrchestration", + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "instance-1", + ExecutionId = "execution-1", + }, + }, + }, + new P.HistoryEvent + { + ExecutionCompleted = new P.ExecutionCompletedEvent + { + OrchestrationStatus = P.OrchestrationStatus.Completed, + }, + }, + }, + NewEvents = + { + new P.HistoryEvent + { + OrchestratorStarted = new P.OrchestratorStartedEvent(), + }, + new P.HistoryEvent + { + ExecutionRewound = new P.ExecutionRewoundEvent + { + Reason = "invalid rewind", + }, + }, + }, + }; + + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + Mock factoryMock = new(MockBehavior.Strict); + GrpcDurableTaskWorker worker = CreateWorker( + new GrpcDurableTaskWorkerOptions(), + new DurableTaskWorkerOptions(), + NullLoggerFactory.Instance, + factoryMock.Object); + + // Act + Func act = () => InvokeOnRunOrchestratorAsync( + CreateProcessor(worker, clientMock.Object), + request, + "completion-token", + CancellationToken.None); + + // Assert + await act.Should().ThrowAsync() + .WithMessage("*ExecutionCompleted event to have status Failed*Completed*"); + factoryMock.VerifyNoOtherCalls(); + } + + [Fact] + public async Task OnRunOrchestratorAsync_RewindRequestWithoutExecutionStarted_Throws() + { + // Arrange + P.OrchestratorRequest request = new() + { + InstanceId = "instance-1", + PastEvents = + { + new P.HistoryEvent + { + ExecutionCompleted = new P.ExecutionCompletedEvent + { + OrchestrationStatus = P.OrchestrationStatus.Failed, + }, + }, + }, + NewEvents = + { + new P.HistoryEvent + { + OrchestratorStarted = new P.OrchestratorStartedEvent(), + }, + new P.HistoryEvent + { + ExecutionRewound = new P.ExecutionRewoundEvent + { + Reason = "invalid rewind", + ParentTraceContext = new P.TraceContext + { + TraceParent = + "00-11111111111111111111111111111111-2222222222222222-01", + }, + }, + }, + }, + }; + + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + Mock factoryMock = new(MockBehavior.Strict); + GrpcDurableTaskWorker worker = CreateWorker( + new GrpcDurableTaskWorkerOptions(), + new DurableTaskWorkerOptions(), + NullLoggerFactory.Instance, + factoryMock.Object); + + // Act + Func act = () => InvokeOnRunOrchestratorAsync( + CreateProcessor(worker, clientMock.Object), + request, + "completion-token", + CancellationToken.None); + + // Assert + await act.Should().ThrowAsync() + .WithMessage("*no ExecutionStartedEvent*"); + factoryMock.VerifyNoOtherCalls(); + } + [Fact] public async Task BuildRuntimeStateAsync_HistoryStreamCanceledMidStream_PropagatesCancellation() { @@ -1013,9 +1660,24 @@ static async Task InvokeBuildRuntimeStateAsync( object? entityConversionState, CancellationToken cancellationToken) { - object result = BuildRuntimeStateAsyncMethod.Invoke( - processor, new object?[] { orchestratorRequest, entityConversionState, cancellationToken })!; - return await (ValueTask)result; + object result = GetPastEventsAsyncMethod.Invoke( + processor, new object?[] { orchestratorRequest, cancellationToken })!; + IReadOnlyList pastEvents = + await (ValueTask>)result; + + return (OrchestrationRuntimeState)BuildRuntimeStateMethod.Invoke( + null, new object?[] { orchestratorRequest, pastEvents, entityConversionState })!; + } + + static async Task InvokeOnRunOrchestratorAsync( + object processor, + P.OrchestratorRequest orchestratorRequest, + string completionToken, + CancellationToken cancellationToken) + { + Task task = (Task)OnRunOrchestratorAsyncMethod.Invoke( + processor, new object?[] { orchestratorRequest, completionToken, cancellationToken })!; + await task; } static T GetResultProperty(object result, string propertyName) diff --git a/test/Worker/Grpc.Tests/RewindOrchestrationHandlerTests.cs b/test/Worker/Grpc.Tests/RewindOrchestrationHandlerTests.cs new file mode 100644 index 000000000..a55085590 --- /dev/null +++ b/test/Worker/Grpc.Tests/RewindOrchestrationHandlerTests.cs @@ -0,0 +1,328 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Diagnostics; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +public class RewindOrchestrationHandlerTests +{ + [Fact] + public void CreateResponse_RewritesFailedHistory() + { + // Arrange + P.HistoryEvent executionStarted = CreateExecutionStarted("old-execution"); + P.HistoryEvent[] pastEvents = + [ + executionStarted, + new() + { + EventId = 0, + TaskScheduled = new P.TaskScheduledEvent { Name = "SuccessfulActivity" }, + }, + new() + { + EventId = 1, + TaskCompleted = new P.TaskCompletedEvent { TaskScheduledId = 0 }, + }, + new() + { + EventId = 2, + TaskScheduled = new P.TaskScheduledEvent { Name = "FailedActivity" }, + }, + new() + { + EventId = 3, + TaskFailed = new P.TaskFailedEvent { TaskScheduledId = 2 }, + }, + new() + { + EventId = 4, + SubOrchestrationInstanceCreated = + new P.SubOrchestrationInstanceCreatedEvent { InstanceId = "failed-child" }, + }, + new() + { + EventId = 5, + SubOrchestrationInstanceFailed = + new P.SubOrchestrationInstanceFailedEvent { TaskScheduledId = 4 }, + }, + new() + { + EventId = 6, + ExecutionCompleted = new P.ExecutionCompletedEvent + { + OrchestrationStatus = P.OrchestrationStatus.Failed, + }, + }, + ]; + + // Act + P.OrchestratorResponse response = RewindOrchestrationHandler.CreateResponse( + CreateRewindRequest(), + pastEvents, + "completion-token", + orchestrationActivity: null); + + // Assert + P.OrchestratorAction action = response.Actions.Should().ContainSingle().Subject; + action.Id.Should().Be(-1); + P.HistoryEvent[] newHistory = action.RewindOrchestration.NewHistory.ToArray(); + + newHistory.Should().NotContain(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskFailed + || e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed + || e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted + || (e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled && e.EventId == 2)); + newHistory.Should().Contain(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled && e.EventId == 0); + newHistory.Should().Contain(e => + e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated + && e.SubOrchestrationInstanceCreated.InstanceId == "failed-child"); + newHistory.Last().ExecutionRewound.Reason.Should().Be("rewind reason"); + + string newExecutionId = newHistory[0].ExecutionStarted.OrchestrationInstance.ExecutionId; + Guid.TryParseExact(newExecutionId, "N", out _).Should().BeTrue(); + executionStarted.ExecutionStarted.OrchestrationInstance.ExecutionId.Should().Be("old-execution"); + } + + [Fact] + public void CreateResponse_UpdatesParentExecutionId() + { + // Arrange + P.HistoryEvent executionStarted = CreateExecutionStarted( + "old-execution", + parentExecutionId: "old-parent-execution"); + P.OrchestratorRequest request = CreateRewindRequest(parentExecutionId: "new-parent-execution"); + + // Act + P.OrchestratorResponse response = RewindOrchestrationHandler.CreateResponse( + request, + [executionStarted], + "completion-token", + orchestrationActivity: null); + + // Assert + P.ExecutionStartedEvent rewrittenStart = + response.Actions[0].RewindOrchestration.NewHistory[0].ExecutionStarted; + rewrittenStart.ParentInstance.OrchestrationInstance.ExecutionId.Should().Be("new-parent-execution"); + executionStarted.ExecutionStarted.ParentInstance.OrchestrationInstance.ExecutionId + .Should().Be("old-parent-execution"); + } + + [Fact] + public void CreateResponse_UpdatesParentTraceContext() + { + // Arrange + P.TraceContext originalParentTraceContext = CreateTraceContext( + "11111111111111111111111111111111", + "2222222222222222"); + P.TraceContext rewindParentTraceContext = CreateTraceContext( + "33333333333333333333333333333333", + "4444444444444444", + "vendor=value"); + P.HistoryEvent executionStarted = CreateExecutionStarted( + "old-execution", + parentExecutionId: "old-parent-execution", + parentTraceContext: originalParentTraceContext); + P.OrchestratorRequest request = CreateRewindRequest( + parentExecutionId: "new-parent-execution", + parentTraceContext: rewindParentTraceContext); + + // Act + P.OrchestratorResponse response = RewindOrchestrationHandler.CreateResponse( + request, + [executionStarted], + "completion-token", + orchestrationActivity: null); + + // Assert + P.ExecutionStartedEvent rewrittenStart = + response.Actions[0].RewindOrchestration.NewHistory[0].ExecutionStarted; + rewrittenStart.ParentTraceContext.TraceParent.Should().Be(rewindParentTraceContext.TraceParent); + rewrittenStart.ParentTraceContext.TraceState.Should().Be(rewindParentTraceContext.TraceState); + + // The original trace context should not be modified + executionStarted.ExecutionStarted.ParentTraceContext.TraceParent + .Should().Be(originalParentTraceContext.TraceParent); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void CreateResponse_ReplacesFailedSubOrchestrationTraceContext( + bool rewindHasParentTraceContext) + { + // Arrange + const string ExecutionTraceId = "11111111111111111111111111111111"; + const string ExecutionParentSpanId = "2222222222222222"; + const string ExecutionTraceState = "execution=value"; + const string OldChildSpanId = "3333333333333333"; + const string RewindTraceId = "44444444444444444444444444444444"; + const string RewindParentSpanId = "5555555555555555"; + const string RewindTraceState = "rewind=value"; + + P.HistoryEvent failedChildCreated = new() + { + EventId = 4, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = "failed-child", + ParentTraceContext = CreateTraceContext(ExecutionTraceId, OldChildSpanId), + }, + }; + P.HistoryEvent[] pastEvents = + [ + CreateExecutionStarted( + "old-execution", + parentTraceContext: CreateTraceContext( + ExecutionTraceId, + ExecutionParentSpanId, + ExecutionTraceState)), + failedChildCreated, + new() + { + EventId = 5, + SubOrchestrationInstanceFailed = + new P.SubOrchestrationInstanceFailedEvent { TaskScheduledId = 4 }, + }, + ]; + P.TraceContext? rewindParentTraceContext = rewindHasParentTraceContext + ? CreateTraceContext(RewindTraceId, RewindParentSpanId, RewindTraceState) + : null; + + // Act + P.OrchestratorResponse response = RewindOrchestrationHandler.CreateResponse( + CreateRewindRequest(parentTraceContext: rewindParentTraceContext), + pastEvents, + "completion-token", + orchestrationActivity: null); + + // Assert + P.SubOrchestrationInstanceCreatedEvent rewrittenChild = response + .Actions[0] + .RewindOrchestration + .NewHistory + .Single(e => e.EventId == 4) + .SubOrchestrationInstanceCreated; + ActivityContext.TryParse( + rewrittenChild.ParentTraceContext.TraceParent, + rewrittenChild.ParentTraceContext.TraceState, + out ActivityContext childTraceContext).Should().BeTrue(); + + string expectedTraceId = rewindHasParentTraceContext ? RewindTraceId : ExecutionTraceId; + childTraceContext.TraceId.Should().Be(ActivityTraceId.CreateFromString(expectedTraceId.AsSpan())); + + // A new span ID should be generated + childTraceContext.SpanId.ToString().Should().NotBe(OldChildSpanId); + childTraceContext.SpanId.ToString().Should().NotBe(ExecutionParentSpanId); + childTraceContext.SpanId.ToString().Should().NotBe(RewindParentSpanId); + + childTraceContext.TraceFlags.Should().Be(ActivityTraceFlags.Recorded); + childTraceContext.TraceState.Should().Be( + rewindHasParentTraceContext ? RewindTraceState : ExecutionTraceState); + + // The original trace context should not be modified + failedChildCreated.SubOrchestrationInstanceCreated.ParentTraceContext.TraceParent + .Should().Be(CreateTraceContext(ExecutionTraceId, OldChildSpanId).TraceParent); + } + + [Fact] + public void CreateResponse_RejectsUnexpectedNewEvents() + { + // Arrange + P.OrchestratorRequest request = new() + { + NewEvents = + { + new P.HistoryEvent + { + ExecutionRewound = new P.ExecutionRewoundEvent(), + }, + }, + }; + + // Act + Action act = () => RewindOrchestrationHandler.CreateResponse( + request, + [], + "completion-token", + orchestrationActivity: null); + + // Assert + act.Should().Throw() + .WithMessage("*exactly two events*"); + } + + static P.HistoryEvent CreateExecutionStarted( + string executionId, + string? parentExecutionId = null, + P.TraceContext? parentTraceContext = null) + { + P.ExecutionStartedEvent executionStarted = new() + { + Name = "TestOrchestration", + Input = "input", + ParentTraceContext = parentTraceContext, + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "instance", + ExecutionId = executionId, + }, + }; + + if (parentExecutionId is not null) + { + executionStarted.ParentInstance = new P.ParentInstanceInfo + { + OrchestrationInstance = new P.OrchestrationInstance + { + InstanceId = "parent", + ExecutionId = parentExecutionId, + }, + }; + } + + return new P.HistoryEvent + { + EventId = -1, + ExecutionStarted = executionStarted, + }; + } + + static P.OrchestratorRequest CreateRewindRequest( + string? parentExecutionId = null, + P.TraceContext? parentTraceContext = null) + { + return new P.OrchestratorRequest + { + InstanceId = "instance", + NewEvents = + { + new P.HistoryEvent + { + OrchestratorStarted = new P.OrchestratorStartedEvent(), + }, + new P.HistoryEvent + { + ExecutionRewound = new P.ExecutionRewoundEvent + { + Reason = "rewind reason", + ParentExecutionId = parentExecutionId, + ParentTraceContext = parentTraceContext, + }, + }, + }, + }; + } + + static P.TraceContext CreateTraceContext(string traceId, string spanId, string? traceState = null) + { + return new() + { + TraceParent = $"00-{traceId}-{spanId}-01", + TraceState = traceState, + }; + } +} From 289719711543172fb6e0c8af8e766e02e81e86e7 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 11 Sep 2026 10:28:17 -0700 Subject: [PATCH 02/11] remove whitespace changes --- src/Shared/Grpc/ProtoUtils.cs | 132 +++++++++++++++++----------------- 1 file changed, 66 insertions(+), 66 deletions(-) diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs index f6d6a1f7c..e3691664d 100644 --- a/src/Shared/Grpc/ProtoUtils.cs +++ b/src/Shared/Grpc/ProtoUtils.cs @@ -80,7 +80,7 @@ internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto, EntityCon historyEvent = new ExecutionCompletedEvent( proto.EventId, proto.ExecutionCompleted.Result, - proto.ExecutionCompleted.OrchestrationStatus.ToCore(), + proto.ExecutionCompleted.OrchestrationStatus.ToCore(), proto.ExecutionCompleted.FailureDetails.ToCore()); break; case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated: @@ -220,9 +220,9 @@ internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto, EntityCon Tags = proto.HistoryState.OrchestrationState.Tags, }); break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound: - historyEvent = new ExecutionRewoundEvent(proto.EventId); - break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound: + historyEvent = new ExecutionRewoundEvent(proto.EventId); + break; default: throw new NotSupportedException($"Deserialization of {proto.EventTypeCase} is not supported."); } @@ -369,15 +369,15 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( Name = subOrchestrationAction.Name, Version = subOrchestrationAction.Version, ParentTraceContext = CreateTraceContext(), - }; - + }; + if (subOrchestrationAction.Tags != null) { foreach (KeyValuePair tag in subOrchestrationAction.Tags) { protoAction.CreateSubOrchestration.Tags[tag.Key] = tag.Value; } - } + } break; case OrchestratorActionType.CreateTimer: @@ -467,17 +467,17 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( var completeAction = (OrchestrationCompleteOrchestratorAction)action; protoAction.CompleteOrchestration = new P.CompleteOrchestrationAction - { + { CarryoverEvents = { completeAction.CarryoverEvents.Select(ToProtobuf) }, Details = completeAction.Details, NewVersion = completeAction.NewVersion, OrchestrationStatus = completeAction.OrchestrationStatus.ToProtobuf(), Result = completeAction.Result, - }; - - foreach (KeyValuePair tag in completeAction.Tags) - { - protoAction.CompleteOrchestration.Tags[tag.Key] = tag.Value; + }; + + foreach (KeyValuePair tag in completeAction.Tags) + { + protoAction.CompleteOrchestration.Tags[tag.Key] = tag.Value; } if (completeAction.OrchestrationStatus == OrchestrationStatus.Failed) @@ -1063,27 +1063,27 @@ internal static T Base64Decode(this MessageParser parser, string encodedMessa case Google.Protobuf.WellKnownTypes.Value.KindOneofCase.NumberValue: return value.NumberValue; case Google.Protobuf.WellKnownTypes.Value.KindOneofCase.StringValue: - string stringValue = value.StringValue; - - // If the value starts with the 'dt:' prefix, it may represent a DateTime value � attempt to parse it. - if (stringValue.StartsWith("dt:", StringComparison.Ordinal)) - { - if (DateTime.TryParse(stringValue[3..], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTime date)) - { - return date; - } - } - - // If the value starts with the 'dto:' prefix, it may represent a DateTime value � attempt to parse it. - if (stringValue.StartsWith("dto:", StringComparison.Ordinal)) - { - if (DateTimeOffset.TryParse(stringValue[4..], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTimeOffset date)) - { - return date; - } - } - - // Otherwise just return as string + string stringValue = value.StringValue; + + // If the value starts with the 'dt:' prefix, it may represent a DateTime value — attempt to parse it. + if (stringValue.StartsWith("dt:", StringComparison.Ordinal)) + { + if (DateTime.TryParse(stringValue[3..], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTime date)) + { + return date; + } + } + + // If the value starts with the 'dto:' prefix, it may represent a DateTime value — attempt to parse it. + if (stringValue.StartsWith("dto:", StringComparison.Ordinal)) + { + if (DateTimeOffset.TryParse(stringValue[4..], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTimeOffset date)) + { + return date; + } + } + + // Otherwise just return as string return stringValue; case Google.Protobuf.WellKnownTypes.Value.KindOneofCase.BoolValue: return value.BoolValue; @@ -1093,16 +1093,16 @@ internal static T Base64Decode(this MessageParser parser, string encodedMessa pair => ConvertValueToObject(pair.Value)); case Google.Protobuf.WellKnownTypes.Value.KindOneofCase.ListValue: return value.ListValue.Values.Select(ConvertValueToObject).ToList(); - default: - // Fallback: serialize the whole value to JSON string + default: + // Fallback: serialize the whole value to JSON string return JsonSerializer.Serialize(value); } - } - + } + /// /// Converts a MapFieldinto a IDictionary. /// - /// The map to convert. + /// The map to convert. /// Dictionary contains the converted obejct. internal static IDictionary ConvertProperties(MapField properties) { @@ -1127,8 +1127,8 @@ internal static Value ConvertObjectToValue(object? obj) long l => Value.ForNumber(l), float f => Value.ForNumber(f), double d => Value.ForNumber(d), - decimal dec => Value.ForNumber((double)dec), - + decimal dec => Value.ForNumber((double)dec), + // For DateTime and DateTimeOffset, add prefix to distinguish from normal string. DateTime dt => Value.ForString($"dt:{dt.ToString("O")}"), DateTimeOffset dto => Value.ForString($"dto:{dto.ToString("O")}"), @@ -1136,9 +1136,9 @@ internal static Value ConvertObjectToValue(object? obj) { Fields = { dict.ToDictionary(kvp => kvp.Key, kvp => ConvertObjectToValue(kvp.Value)) }, }), - IEnumerable e => Value.ForList(e.Cast().Select(ConvertObjectToValue).ToArray()), - - // Fallback: convert unlisted type to string. + IEnumerable e => Value.ForList(e.Cast().Select(ConvertObjectToValue).ToArray()), + + // Fallback: convert unlisted type to string. _ => Value.ForString(obj.ToString() ?? string.Empty), }; } @@ -1188,28 +1188,28 @@ static P.OrchestrationInstance ToProtobuf(this OrchestrationInstance instance) InstanceId = instance.InstanceId, ExecutionId = instance.ExecutionId, }; - } - - static P.HistoryEvent ToProtobuf(HistoryEvent e) - { - var payload = new P.HistoryEvent() - { - EventId = e.EventId, - Timestamp = Timestamp.FromDateTime(e.Timestamp), - }; - - if (e.EventType == EventType.EventRaised) - { - var eventRaised = (EventRaisedEvent)e; - payload.EventRaised = new P.EventRaisedEvent - { - Name = eventRaised.Name, - Input = eventRaised.Input, - }; - return payload; - } - - throw new ArgumentException("Unsupported event type"); + } + + static P.HistoryEvent ToProtobuf(HistoryEvent e) + { + var payload = new P.HistoryEvent() + { + EventId = e.EventId, + Timestamp = Timestamp.FromDateTime(e.Timestamp), + }; + + if (e.EventType == EventType.EventRaised) + { + var eventRaised = (EventRaisedEvent)e; + payload.EventRaised = new P.EventRaisedEvent + { + Name = eventRaised.Name, + Input = eventRaised.Input, + }; + return payload; + } + + throw new ArgumentException("Unsupported event type"); } /// From 00b2c9ebb1317118fd3920cc9039b3375b681324 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 11 Sep 2026 13:19:21 -0700 Subject: [PATCH 03/11] updated to include more tests, including one that caught a regression in DTS --- .../DtsRewindIntegrationTests.cs | 328 ++++++++++++++++-- 1 file changed, 297 insertions(+), 31 deletions(-) diff --git a/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs b/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs index d5f245d52..ebdf3db63 100644 --- a/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs +++ b/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs @@ -206,15 +206,96 @@ public async Task RewindCompletedOrchestrationThrowsAsync() } /// - /// Verifies that rewind recursively re-executes a failed sub-orchestration. + /// Verifies that rewind recursively re-executes failed sub-orchestrations without restarting them. + /// + [DtsEmulatorFact(Skip = "Requires a DTS emulator image containing the sidecar rewind history fix.")] + public async Task CanRewindFailedSubOrchestrationsAsync() + { + // Arrange + TaskName parentOrchestratorName = + $"{nameof(CanRewindFailedSubOrchestrationsAsync)}_Parent"; + TaskName childOrchestratorName = + $"{nameof(CanRewindFailedSubOrchestrationsAsync)}_Child"; + TaskName activityName = + $"{nameof(CanRewindFailedSubOrchestrationsAsync)}_Activity"; + ConcurrentDictionary activityCallCounts = []; + int shouldFail = 1; + + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks + .AddOrchestratorFunc( + parentOrchestratorName, + async context => + { + List> children = []; + for (int childIndex = 0; childIndex < 3; childIndex++) + { + children.Add(context.CallSubOrchestratorAsync( + childOrchestratorName, + childIndex, + new SubOrchestrationOptions( + instanceId: $"{context.InstanceId}-child-{childIndex}"))); + } + + return await Task.WhenAll(children); + }) + .AddOrchestratorFunc( + childOrchestratorName, + async (context, childIndex) => + await context.CallActivityAsync(activityName, childIndex)) + .AddActivityFunc(activityName, (context, childIndex) => + { + activityCallCounts.AddOrUpdate( + childIndex, + 1, + (_, current) => current + 1); + + if (childIndex < 2 && Volatile.Read(ref shouldFail) == 1) + { + throw new InvalidOperationException($"Child {childIndex} failed."); + } + + return Task.FromResult(childIndex); + })); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + parentOrchestratorName, + cancellation: this.TimeoutToken); + OrchestrationMetadata failed = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); + Assert.Equal(1, activityCallCounts[0]); + Assert.Equal(1, activityCallCounts[1]); + Assert.Equal(1, activityCallCounts[2]); + + // Act + Volatile.Write(ref shouldFail, 0); + await server.Client.RewindInstanceAsync(instanceId, "retry failed children", this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + await Task.Delay(TimeSpan.FromSeconds(2)); + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + int[] output = completed.ReadOutputAs() + ?? throw new InvalidOperationException("The parent output was missing."); + Assert.Equal([0, 1, 2], output); + Assert.Equal(2, activityCallCounts[0]); + Assert.Equal(2, activityCallCounts[1]); + Assert.Equal(1, activityCallCounts[2]); + } + + /// + /// Verifies that rewind recursively reaches a failed nested sub-orchestration. /// [DtsEmulatorFact] - public async Task RewindFailedSubOrchestrationAsync() + public async Task CanRewindNestedFailedSubOrchestrationAsync() { // Arrange - TaskName parentOrchestratorName = $"{nameof(RewindFailedSubOrchestrationAsync)}_Parent"; - TaskName childOrchestratorName = $"{nameof(RewindFailedSubOrchestrationAsync)}_Child"; - TaskName activityName = $"{nameof(RewindFailedSubOrchestrationAsync)}_Activity"; + TaskName parentOrchestratorName = $"{nameof(CanRewindNestedFailedSubOrchestrationAsync)}_Parent"; + TaskName childOrchestratorName = $"{nameof(CanRewindNestedFailedSubOrchestrationAsync)}_Child"; + TaskName nestedOrchestratorName = $"{nameof(CanRewindNestedFailedSubOrchestrationAsync)}_Nested"; + TaskName activityName = $"{nameof(CanRewindNestedFailedSubOrchestrationAsync)}_Activity"; int activityCallCount = 0; await using HostTestLifetime server = await this.StartWorkerAsync(builder => @@ -226,22 +307,33 @@ public async Task RewindFailedSubOrchestrationAsync() { string result = await context.CallSubOrchestratorAsync( childOrchestratorName, - input); + input, + new SubOrchestrationOptions(instanceId: $"{context.InstanceId}-child")); return $"parent:{result}"; }) .AddOrchestratorFunc( childOrchestratorName, async (context, input) => - await context.CallActivityAsync(activityName, input)) + { + string result = await context.CallSubOrchestratorAsync( + nestedOrchestratorName, + input, + new SubOrchestrationOptions(instanceId: $"{context.InstanceId}-nested")); + return $"child:{result}"; + }) + .AddOrchestratorFunc( + nestedOrchestratorName, + async (context, input) => + $"nested:{await context.CallActivityAsync(activityName, input)}") .AddActivityFunc(activityName, (context, input) => { int count = Interlocked.Increment(ref activityCallCount); if (count == 1) { - throw new InvalidOperationException("Child failure."); + throw new InvalidOperationException("Nested child failure."); } - return Task.FromResult($"child:{input}"); + return Task.FromResult($"activity:{input}"); })); }); @@ -253,33 +345,47 @@ await context.CallActivityAsync(activityName, input)) Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); // Act - await server.Client.RewindInstanceAsync(instanceId, "sub-orchestration fix", this.TimeoutToken); + await server.Client.RewindInstanceAsync(instanceId, "retry nested child", this.TimeoutToken); OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); // Assert Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); - Assert.Equal("parent:child:data", completed.ReadOutputAs()); + Assert.Equal("parent:child:nested:activity:data", completed.ReadOutputAs()); Assert.Equal(2, Volatile.Read(ref activityCallCount)); } /// - /// Verifies that a purged failed sub-orchestration is recreated when its parent is rewound. + /// Verifies that a purged failed sub-orchestration and its purged failed child are recreated on rewind. /// [DtsEmulatorFact] - public async Task RewindPurgedSubOrchestrationAsync() + public async Task RewindPurgedNestedSubOrchestrationsAsync() { // Arrange - TaskName parentOrchestratorName = $"{nameof(RewindPurgedSubOrchestrationAsync)}_Parent"; - TaskName childOrchestratorName = $"{nameof(RewindPurgedSubOrchestrationAsync)}_Child"; - TaskName activityName = $"{nameof(RewindPurgedSubOrchestrationAsync)}_Activity"; + TaskName parentOrchestratorName = $"{nameof(RewindPurgedNestedSubOrchestrationsAsync)}_Parent"; + TaskName childOrchestratorName = $"{nameof(RewindPurgedNestedSubOrchestrationsAsync)}_Child"; + TaskName nestedOrchestratorName = $"{nameof(RewindPurgedNestedSubOrchestrationsAsync)}_Nested"; + TaskName activityName = $"{nameof(RewindPurgedNestedSubOrchestrationsAsync)}_Activity"; string childInstanceId = $"child-{Guid.NewGuid():N}"; - const string ChildVersion = "v1"; + string nestedInstanceId = $"nested-{Guid.NewGuid():N}"; + const string ChildVersion = "1.0"; + const string NestedVersion = "2.0"; SubOrchestrationOptions childOptions = new(instanceId: childInstanceId) { Version = ChildVersion, Tags = new Dictionary { - ["scenario"] = "purged-sub-orchestration", + ["scenario"] = "purged-nested-sub-orchestration", + ["level"] = "child", + ["preserve"] = "true", + }, + }; + SubOrchestrationOptions nestedOptions = new(instanceId: nestedInstanceId) + { + Version = NestedVersion, + Tags = new Dictionary + { + ["scenario"] = "purged-nested-sub-orchestration", + ["level"] = "nested", ["preserve"] = "true", }, }; @@ -302,18 +408,25 @@ public async Task RewindPurgedSubOrchestrationAsync() childOrchestratorName, async (context, input) => { - string result = await context.CallActivityAsync(activityName, input); - return $"{context.Version}:{result}"; + string result = await context.CallSubOrchestratorAsync( + nestedOrchestratorName, + input, + nestedOptions); + return $"{context.Version}:child:{result}"; }) + .AddOrchestratorFunc( + nestedOrchestratorName, + async (context, input) => + $"{context.Version}:nested:{await context.CallActivityAsync(activityName, input)}") .AddActivityFunc(activityName, (context, input) => { int count = Interlocked.Increment(ref activityCallCount); if (count == 1) { - throw new InvalidOperationException("Child failure."); + throw new InvalidOperationException("Nested child failure."); } - return Task.FromResult($"child:{input}"); + return Task.FromResult($"activity:{input}"); })); }); @@ -323,32 +436,185 @@ public async Task RewindPurgedSubOrchestrationAsync() cancellation: this.TimeoutToken); OrchestrationMetadata failed = await this.WaitForCompletionAsync(server.Client, instanceId); Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); + Assert.Equal(1, Volatile.Read(ref activityCallCount)); + + PurgeResult nestedPurgeResult = await server.Client.PurgeInstanceAsync( + nestedInstanceId, + this.TimeoutToken); + Assert.Equal(1, nestedPurgeResult.PurgedInstanceCount); - PurgeResult purgeResult = await server.Client.PurgeInstanceAsync( + PurgeResult childPurgeResult = await server.Client.PurgeInstanceAsync( childInstanceId, this.TimeoutToken); - Assert.Equal(1, purgeResult.PurgedInstanceCount); + Assert.Equal(1, childPurgeResult.PurgedInstanceCount); // Act - await server.Client.RewindInstanceAsync(instanceId, "purge and retry", this.TimeoutToken); + await server.Client.RewindInstanceAsync(instanceId, "purge and retry nested child", this.TimeoutToken); OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); - - // Assert - Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); - Assert.Equal($"parent:{ChildVersion}:child:data", completed.ReadOutputAs()); - Assert.Equal(2, Volatile.Read(ref activityCallCount)); + await Task.Delay(TimeSpan.FromSeconds(2)); OrchestrationMetadata recreatedChild = await server.Client.GetInstanceAsync( childInstanceId, getInputsAndOutputs: true, this.TimeoutToken) ?? throw new InvalidOperationException("The recreated child orchestration was not found."); + OrchestrationMetadata recreatedNestedChild = await server.Client.GetInstanceAsync( + nestedInstanceId, + getInputsAndOutputs: true, + this.TimeoutToken) + ?? throw new InvalidOperationException("The recreated nested child orchestration was not found."); + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + Assert.Equal( + $"parent:{ChildVersion}:child:{NestedVersion}:nested:activity:data", + completed.ReadOutputAs()); + Assert.Equal(2, Volatile.Read(ref activityCallCount)); + Assert.Equal(childOrchestratorName.Name, recreatedChild.Name); Assert.Equal(childInstanceId, recreatedChild.InstanceId); Assert.Equal(OrchestrationRuntimeStatus.Completed, recreatedChild.RuntimeStatus); Assert.Equal("data", recreatedChild.ReadInputAs()); - Assert.Equal($"{ChildVersion}:child:data", recreatedChild.ReadOutputAs()); + Assert.Equal( + $"{ChildVersion}:child:{NestedVersion}:nested:activity:data", + recreatedChild.ReadOutputAs()); Assert.Equal(childOptions.Tags, recreatedChild.Tags); + + Assert.Equal(nestedOrchestratorName.Name, recreatedNestedChild.Name); + Assert.Equal(nestedInstanceId, recreatedNestedChild.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, recreatedNestedChild.RuntimeStatus); + Assert.Equal("data", recreatedNestedChild.ReadInputAs()); + Assert.Equal( + $"{NestedVersion}:nested:activity:data", + recreatedNestedChild.ReadOutputAs()); + Assert.Equal(nestedOptions.Tags, recreatedNestedChild.Tags); + } + + /// + /// Verifies that purged failed sub-orchestrations are recreated when their parent is rewound. + /// + [DtsEmulatorFact] + public async Task RewindPurgedSubOrchestrationsAsync() + { + // Arrange + TaskName parentOrchestratorName = $"{nameof(RewindPurgedSubOrchestrationsAsync)}_Parent"; + TaskName childOrchestratorName = $"{nameof(RewindPurgedSubOrchestrationsAsync)}_Child"; + TaskName activityName = $"{nameof(RewindPurgedSubOrchestrationsAsync)}_Activity"; + const int ChildCount = 2; + string[] childInstanceIds = Enumerable.Range(0, ChildCount) + .Select(childIndex => $"child-{childIndex}-{Guid.NewGuid():N}") + .ToArray(); + const string ChildVersion = "1.0"; + SubOrchestrationOptions[] childOptions = childInstanceIds + .Select((instanceId, childIndex) => new SubOrchestrationOptions(instanceId: instanceId) + { + Version = ChildVersion, + Tags = new Dictionary + { + ["scenario"] = "purged-sub-orchestration", + ["child-index"] = childIndex.ToString(), + ["preserve"] = "true", + }, + }) + .ToArray(); + ConcurrentDictionary activityCallCounts = []; + + await using HostTestLifetime server = await this.StartWorkerAsync(builder => + { + builder.AddTasks(tasks => tasks + .AddOrchestratorFunc( + parentOrchestratorName, + async context => + { + List> children = []; + for (int childIndex = 0; childIndex < childOptions.Length; childIndex++) + { + children.Add(context.CallSubOrchestratorAsync( + childOrchestratorName, + $"data-{childIndex}", + childOptions[childIndex])); + } + + return await Task.WhenAll(children); + }) + .AddOrchestratorFunc( + childOrchestratorName, + async (context, input) => + { + string result = await context.CallActivityAsync(activityName, input); + return $"{context.Version}:{result}"; + }) + .AddActivityFunc(activityName, (context, input) => + { + int count = activityCallCounts.AddOrUpdate(input, 1, (_, current) => current + 1); + if (count == 1) + { + throw new InvalidOperationException($"Child {input} failed."); + } + + return Task.FromResult($"child:{input}"); + })); + }); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + parentOrchestratorName, + cancellation: this.TimeoutToken); + OrchestrationMetadata failed = await this.WaitForCompletionAsync(server.Client, instanceId); + Assert.Equal(OrchestrationRuntimeStatus.Failed, failed.RuntimeStatus); + for (int childIndex = 0; childIndex < ChildCount; childIndex++) + { + Assert.Equal(1, activityCallCounts[$"data-{childIndex}"]); + } + + foreach (string childInstanceId in childInstanceIds) + { + PurgeResult purgeResult = await server.Client.PurgeInstanceAsync( + childInstanceId, + this.TimeoutToken); + Assert.Equal(1, purgeResult.PurgedInstanceCount); + } + + // Act + await server.Client.RewindInstanceAsync(instanceId, "purge and retry", this.TimeoutToken); + OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + await Task.Delay(TimeSpan.FromSeconds(2)); + OrchestrationMetadata?[] recreatedChildren = new OrchestrationMetadata?[childInstanceIds.Length]; + for (int childIndex = 0; childIndex < childInstanceIds.Length; childIndex++) + { + recreatedChildren[childIndex] = await server.Client.GetInstanceAsync( + childInstanceIds[childIndex], + getInputsAndOutputs: true, + this.TimeoutToken); + } + + // Assert + Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); + string[] output = completed.ReadOutputAs() + ?? throw new InvalidOperationException("The parent output was missing."); + Assert.Equal( + Enumerable.Range(0, ChildCount) + .Select(childIndex => $"{ChildVersion}:child:data-{childIndex}") + .ToArray(), + output); + for (int childIndex = 0; childIndex < ChildCount; childIndex++) + { + Assert.Equal(2, activityCallCounts[$"data-{childIndex}"]); + } + + for (int childIndex = 0; childIndex < childInstanceIds.Length; childIndex++) + { + OrchestrationMetadata recreatedChild = recreatedChildren[childIndex] + ?? throw new InvalidOperationException( + $"The recreated child orchestration {childIndex} was not found."); + Assert.Equal(childOrchestratorName.Name, recreatedChild.Name); + Assert.Equal(childInstanceIds[childIndex], recreatedChild.InstanceId); + Assert.Equal(OrchestrationRuntimeStatus.Completed, recreatedChild.RuntimeStatus); + Assert.Equal($"data-{childIndex}", recreatedChild.ReadInputAs()); + Assert.Equal( + $"{ChildVersion}:child:data-{childIndex}", + recreatedChild.ReadOutputAs()); + Assert.Equal(childOptions[childIndex].Tags, recreatedChild.Tags); + } } /// From 1bc4f6711a429b66b51f55bdb00c159414c00601 Mon Sep 17 00:00:00 2001 From: sophiatev <38052607+sophiatev@users.noreply.github.com> Date: Fri, 11 Sep 2026 13:27:29 -0700 Subject: [PATCH 04/11] Validate NewEvents count and event types Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/Worker/Grpc/RewindOrchestrationHandler.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Worker/Grpc/RewindOrchestrationHandler.cs b/src/Worker/Grpc/RewindOrchestrationHandler.cs index b6f1cb134..2707dd125 100644 --- a/src/Worker/Grpc/RewindOrchestrationHandler.cs +++ b/src/Worker/Grpc/RewindOrchestrationHandler.cs @@ -29,6 +29,7 @@ internal static P.OrchestratorResponse CreateResponse( Check.NotNull(pastEvents); if (request.NewEvents.Count != 2 + || request.NewEvents[0].EventTypeCase != P.HistoryEvent.EventTypeOneofCase.OrchestratorStarted || request.NewEvents[1].EventTypeCase != P.HistoryEvent.EventTypeOneofCase.ExecutionRewound) { throw new InvalidOperationException( From a4d9be344f35cafdc433c1b8ac853f4c69798249 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 11 Sep 2026 13:30:29 -0700 Subject: [PATCH 05/11] add skip to the other multi-sub test --- test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs b/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs index ebdf3db63..6382fcde0 100644 --- a/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs +++ b/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs @@ -208,7 +208,7 @@ public async Task RewindCompletedOrchestrationThrowsAsync() /// /// Verifies that rewind recursively re-executes failed sub-orchestrations without restarting them. /// - [DtsEmulatorFact(Skip = "Requires a DTS emulator image containing the sidecar rewind history fix.")] + [DtsEmulatorFact(Skip = "Requires a DTS emulator image containing the fix for rewinding multiple failed suborchestrations.")] public async Task CanRewindFailedSubOrchestrationsAsync() { // Arrange @@ -493,7 +493,7 @@ public async Task RewindPurgedNestedSubOrchestrationsAsync() /// /// Verifies that purged failed sub-orchestrations are recreated when their parent is rewound. /// - [DtsEmulatorFact] + [DtsEmulatorFact(Skip = "Requires a DTS emulator image containing the fix for rewinding multiple failed suborchestrations.")] public async Task RewindPurgedSubOrchestrationsAsync() { // Arrange From 7d704af4f0291771992da7e7b7af7737342ad111 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 11 Sep 2026 13:41:23 -0700 Subject: [PATCH 06/11] improve error detection for no ExecutionStartedEvent in history for rewind --- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 22 ++++++++++--------- .../Grpc.Tests/GrpcDurableTaskWorkerTests.cs | 5 ----- 2 files changed, 12 insertions(+), 15 deletions(-) diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 266514af8..0aecd5ed1 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -651,16 +651,18 @@ async Task OnRunOrchestratorAsync( .Select(e => e.ExecutionStarted) .FirstOrDefault(); - if (isInitialRewind - && rewindEvent!.ParentTraceContext is not null) - { - if (executionStartedEvent is null) - { - throw new InvalidOperationException("Rewinding orchestration has no ExecutionStartedEvent in its history"); - } - - executionStartedEvent = executionStartedEvent.Clone(); - executionStartedEvent.ParentTraceContext = rewindEvent.ParentTraceContext; + if (isInitialRewind) + { + if (executionStartedEvent is null) + { + throw new InvalidOperationException("Rewinding orchestration has no ExecutionStartedEvent in its history"); + } + + if (rewindEvent!.ParentTraceContext is not null) + { + executionStartedEvent = executionStartedEvent.Clone(); + executionStartedEvent.ParentTraceContext = rewindEvent.ParentTraceContext; + } } // A rewind starts a new orchestration span instead of continuing the failed execution's stored span. diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index 41e80de46..61c3cdf83 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -1278,11 +1278,6 @@ public async Task OnRunOrchestratorAsync_RewindRequestWithoutExecutionStarted_Th ExecutionRewound = new P.ExecutionRewoundEvent { Reason = "invalid rewind", - ParentTraceContext = new P.TraceContext - { - TraceParent = - "00-11111111111111111111111111111111-2222222222222222-01", - }, }, }, }, From 2cee8de3a6da3e16f101ec87500019f01d06dd84 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 11 Sep 2026 13:50:57 -0700 Subject: [PATCH 07/11] hardened the rewind tests --- .../DtsRewindIntegrationTests.cs | 123 +++++++++++++----- 1 file changed, 90 insertions(+), 33 deletions(-) diff --git a/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs b/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs index 6382fcde0..0020ae54b 100644 --- a/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs +++ b/test/Grpc.IntegrationTests/DtsRewindIntegrationTests.cs @@ -82,7 +82,10 @@ await context.CallActivityAsync(activityName, input)) // Act Volatile.Write(ref shouldFail, 0); await server.Client.RewindInstanceAsync(instanceId, "retry after fix", this.TimeoutToken); - OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + OrchestrationMetadata completed = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + failed); // Assert Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); @@ -142,7 +145,10 @@ public async Task RewindPreservesSuccessfulResultsAsync() // Act Volatile.Write(ref shouldFailSecond, 0); await server.Client.RewindInstanceAsync(instanceId, "retry", this.TimeoutToken); - OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + OrchestrationMetadata completed = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + failed); // Assert Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); @@ -272,7 +278,10 @@ await context.CallActivityAsync(activityName, childIndex)) // Act Volatile.Write(ref shouldFail, 0); await server.Client.RewindInstanceAsync(instanceId, "retry failed children", this.TimeoutToken); - OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + OrchestrationMetadata completed = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + failed); await Task.Delay(TimeSpan.FromSeconds(2)); // Assert @@ -346,7 +355,10 @@ public async Task CanRewindNestedFailedSubOrchestrationAsync() // Act await server.Client.RewindInstanceAsync(instanceId, "retry nested child", this.TimeoutToken); - OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + OrchestrationMetadata completed = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + failed); // Assert Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); @@ -450,19 +462,15 @@ public async Task RewindPurgedNestedSubOrchestrationsAsync() // Act await server.Client.RewindInstanceAsync(instanceId, "purge and retry nested child", this.TimeoutToken); - OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); - await Task.Delay(TimeSpan.FromSeconds(2)); - - OrchestrationMetadata recreatedChild = await server.Client.GetInstanceAsync( - childInstanceId, - getInputsAndOutputs: true, - this.TimeoutToken) - ?? throw new InvalidOperationException("The recreated child orchestration was not found."); - OrchestrationMetadata recreatedNestedChild = await server.Client.GetInstanceAsync( - nestedInstanceId, - getInputsAndOutputs: true, - this.TimeoutToken) - ?? throw new InvalidOperationException("The recreated nested child orchestration was not found."); + OrchestrationMetadata completed = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + failed); + OrchestrationMetadata[] recreatedInstances = await Task.WhenAll( + this.WaitForRecreatedInstanceCompletionAsync(server.Client, childInstanceId), + this.WaitForRecreatedInstanceCompletionAsync(server.Client, nestedInstanceId)); + OrchestrationMetadata recreatedChild = recreatedInstances[0]; + OrchestrationMetadata recreatedNestedChild = recreatedInstances[1]; // Assert Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); @@ -576,16 +584,13 @@ public async Task RewindPurgedSubOrchestrationsAsync() // Act await server.Client.RewindInstanceAsync(instanceId, "purge and retry", this.TimeoutToken); - OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); - await Task.Delay(TimeSpan.FromSeconds(2)); - OrchestrationMetadata?[] recreatedChildren = new OrchestrationMetadata?[childInstanceIds.Length]; - for (int childIndex = 0; childIndex < childInstanceIds.Length; childIndex++) - { - recreatedChildren[childIndex] = await server.Client.GetInstanceAsync( - childInstanceIds[childIndex], - getInputsAndOutputs: true, - this.TimeoutToken); - } + OrchestrationMetadata completed = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + failed); + OrchestrationMetadata[] recreatedChildren = await Task.WhenAll( + childInstanceIds.Select(childInstanceId => + this.WaitForRecreatedInstanceCompletionAsync(server.Client, childInstanceId))); // Assert Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); @@ -603,9 +608,7 @@ public async Task RewindPurgedSubOrchestrationsAsync() for (int childIndex = 0; childIndex < childInstanceIds.Length; childIndex++) { - OrchestrationMetadata recreatedChild = recreatedChildren[childIndex] - ?? throw new InvalidOperationException( - $"The recreated child orchestration {childIndex} was not found."); + OrchestrationMetadata recreatedChild = recreatedChildren[childIndex]; Assert.Equal(childOrchestratorName.Name, recreatedChild.Name); Assert.Equal(childInstanceIds[childIndex], recreatedChild.InstanceId); Assert.Equal(OrchestrationRuntimeStatus.Completed, recreatedChild.RuntimeStatus); @@ -654,7 +657,10 @@ public async Task RewindWithoutReasonAsync() // Act await server.Client.RewindInstanceAsync(instanceId, string.Empty, this.TimeoutToken); - OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + OrchestrationMetadata completed = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + failed); // Assert Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); @@ -699,12 +705,18 @@ await context.CallActivityAsync(activityName, input)) Assert.Equal(OrchestrationRuntimeStatus.Failed, firstFailure.RuntimeStatus); await server.Client.RewindInstanceAsync(instanceId, "first rewind", this.TimeoutToken); - OrchestrationMetadata secondFailure = await this.WaitForCompletionAsync(server.Client, instanceId); + OrchestrationMetadata secondFailure = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + firstFailure); Assert.Equal(OrchestrationRuntimeStatus.Failed, secondFailure.RuntimeStatus); // Act await server.Client.RewindInstanceAsync(instanceId, "second rewind", this.TimeoutToken); - OrchestrationMetadata completed = await this.WaitForCompletionAsync(server.Client, instanceId); + OrchestrationMetadata completed = await this.WaitForRewindCompletionAsync( + server.Client, + instanceId, + secondFailure); // Assert Assert.Equal(OrchestrationRuntimeStatus.Completed, completed.RuntimeStatus); @@ -765,6 +777,51 @@ async Task StartHostAsync(Action? c } } + async Task WaitForRecreatedInstanceCompletionAsync( + DurableTaskClient client, + string instanceId) + { + while (true) + { + OrchestrationMetadata? metadata = await client.GetInstanceAsync( + instanceId, + getInputsAndOutputs: true, + this.TimeoutToken); + if (metadata is not null) + { + return metadata.IsCompleted + ? metadata + : await this.WaitForCompletionAsync(client, instanceId); + } + + await Task.Delay(TimeSpan.FromMilliseconds(100), this.TimeoutToken); + } + } + + async Task WaitForRewindCompletionAsync( + DurableTaskClient client, + string instanceId, + OrchestrationMetadata previousState) + { + while (true) + { + OrchestrationMetadata? currentState = await client.GetInstanceAsync( + instanceId, + getInputsAndOutputs: true, + this.TimeoutToken); + if (currentState is not null + && (currentState.LastUpdatedAt > previousState.LastUpdatedAt + || !currentState.IsCompleted)) + { + return currentState.IsCompleted + ? currentState + : await this.WaitForCompletionAsync(client, instanceId); + } + + await Task.Delay(TimeSpan.FromMilliseconds(100), this.TimeoutToken); + } + } + async Task WaitForCompletionAsync( DurableTaskClient client, string instanceId) From 8418ba312494689aac7f34e95a950e6bf5f75ec6 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Fri, 11 Sep 2026 15:26:03 -0700 Subject: [PATCH 08/11] added large payload support to rewind --- .../AzureBlobPayloadsSideCarInterceptor.cs | 92 ++++++ ...zureBlobPayloadsSideCarInterceptorTests.cs | 261 ++++++++++++++++++ 2 files changed, 353 insertions(+) diff --git a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs index 4b0ba495a..29fd76f59 100644 --- a/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs +++ b/src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs @@ -48,6 +48,9 @@ protected override async Task ExternalizeRequestPayloadsAsync(TRequest case P.ResumeRequest r: r.Reason = await this.MaybeExternalizeAsync(r.Reason, cancellation); break; + case P.RewindInstanceRequest r: + r.Reason = await this.MaybeExternalizeAsync(r.Reason, cancellation); + break; case P.SignalEntityRequest r: r.Input = await this.MaybeExternalizeAsync(r.Input, cancellation); break; @@ -432,11 +435,91 @@ async Task ExternalizeOrchestratorResponseAsync(P.OrchestratorResponse r, Cancel } } } + + if (a.RewindOrchestration is { } rewind) + { + foreach (P.HistoryEvent historyEvent in rewind.NewHistory) + { + operations.Add(() => this.ExternalizeHistoryEventAsync(historyEvent, cancellation)); + } + } } await RunWithBoundedConcurrencyAsync(operations, cancellation); } + async Task ExternalizeHistoryEventAsync( + P.HistoryEvent historyEvent, + CancellationToken cancellation) + { + // Keep these fields aligned with ResolveEventPayloadsAsync + switch (historyEvent.EventTypeCase) + { + case P.HistoryEvent.EventTypeOneofCase.ExecutionStarted when historyEvent.ExecutionStarted is { } es: + es.Input = await this.MaybeExternalizeAsync(es.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted when historyEvent.ExecutionCompleted is { } ec: + ec.Result = await this.MaybeExternalizeAsync(ec.Result, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated when historyEvent.ExecutionTerminated is { } et: + et.Input = await this.MaybeExternalizeAsync(et.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.EventRaised when historyEvent.EventRaised is { } er: + er.Input = await this.MaybeExternalizeAsync(er.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.TaskScheduled when historyEvent.TaskScheduled is { } ts: + ts.Input = await this.MaybeExternalizeAsync(ts.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.TaskCompleted when historyEvent.TaskCompleted is { } tc: + tc.Result = await this.MaybeExternalizeAsync(tc.Result, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated + when historyEvent.SubOrchestrationInstanceCreated is { } soc: + soc.Input = await this.MaybeExternalizeAsync(soc.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted + when historyEvent.SubOrchestrationInstanceCompleted is { } sox: + sox.Result = await this.MaybeExternalizeAsync(sox.Result, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.EventSent when historyEvent.EventSent is { } esent: + esent.Input = await this.MaybeExternalizeAsync(esent.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.GenericEvent when historyEvent.GenericEvent is { } ge: + ge.Data = await this.MaybeExternalizeAsync(ge.Data, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.ContinueAsNew when historyEvent.ContinueAsNew is { } can: + can.Input = await this.MaybeExternalizeAsync(can.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended when historyEvent.ExecutionSuspended is { } esus: + esus.Input = await this.MaybeExternalizeAsync(esus.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionResumed when historyEvent.ExecutionResumed is { } eres: + eres.Input = await this.MaybeExternalizeAsync(eres.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound when historyEvent.ExecutionRewound is { } erew: + erew.Reason = await this.MaybeExternalizeAsync(erew.Reason, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled + when historyEvent.EntityOperationSignaled is { } eos: + eos.Input = await this.MaybeExternalizeAsync(eos.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled + when historyEvent.EntityOperationCalled is { } eoc: + eoc.Input = await this.MaybeExternalizeAsync(eoc.Input, cancellation); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted + when historyEvent.EntityOperationCompleted is { } ecomp: + ecomp.Output = await this.MaybeExternalizeAsync(ecomp.Output, 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); + break; + } + } + async Task ExternalizeEntityBatchResultAsync(P.EntityBatchResult r, CancellationToken cancellation) { List> operations = []; @@ -554,6 +637,8 @@ bool RequiresEventPayloadResolution(P.HistoryEvent e) this.RequiresResolution(e.ExecutionSuspended?.Input), P.HistoryEvent.EventTypeOneofCase.ExecutionResumed => this.RequiresResolution(e.ExecutionResumed?.Input), + P.HistoryEvent.EventTypeOneofCase.ExecutionRewound => + this.RequiresResolution(e.ExecutionRewound?.Reason), P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled => this.RequiresResolution(e.EntityOperationSignaled?.Input), P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled => @@ -663,6 +748,13 @@ async Task ResolveEventPayloadsAsync(P.HistoryEvent e, CancellationToken cancell eres.Input = await this.MaybeResolveAsync(eres.Input, cancellation); } + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound: + if (e.ExecutionRewound is { } erew) + { + erew.Reason = await this.MaybeResolveAsync(erew.Reason, cancellation); + } + break; case P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled: if (e.EntityOperationSignaled is { } eos) diff --git a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs index 4f92ef4a9..c8135ef6e 100644 --- a/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs +++ b/test/Grpc.IntegrationTests/AzureBlobPayloadsSideCarInterceptorTests.cs @@ -340,6 +340,50 @@ public async Task ResolveResponsePayloadsAsync_WorkItemOrchestratorRequest_Resol } } + [Fact] + public async Task ResolveResponsePayloadsAsync_WorkItemOrchestratorRequest_ResolvesExecutionRewoundReasons() + { + // Arrange + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + P.WorkItem workItem = new() + { + OrchestratorRequest = new P.OrchestratorRequest + { + InstanceId = "instance-1", + PastEvents = + { + new P.HistoryEvent + { + ExecutionRewound = new P.ExecutionRewoundEvent + { + Reason = store.Seed("previous rewind reason"), + }, + }, + }, + NewEvents = + { + new P.HistoryEvent + { + ExecutionRewound = new P.ExecutionRewoundEvent + { + Reason = store.Seed("current rewind reason"), + }, + }, + }, + }, + }; + + // Act + await ResolveAsync(interceptor, workItem, CancellationToken.None); + + // Assert + workItem.OrchestratorRequest.PastEvents[0].ExecutionRewound.Reason + .Should().Be("previous rewind reason"); + workItem.OrchestratorRequest.NewEvents[0].ExecutionRewound.Reason + .Should().Be("current rewind reason"); + } + [Fact] public async Task ResolveResponsePayloadsAsync_WorkItemEntityRequestV1_ResolvesEntityStateAndOperationsInOrder() { @@ -412,6 +456,119 @@ public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_Externali store.MaxObservedConcurrency.Should().BeLessOrEqualTo(8, "concurrency must be bounded to avoid Azure Storage throttling"); } + [Fact] + public async Task ExternalizeRequestPayloadsAsync_RewindInstanceRequest_ExternalizesReason() + { + // Arrange + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + P.RewindInstanceRequest request = new() + { + InstanceId = "instance-1", + Reason = "rewind reason", + }; + + // Act + await ExternalizeAsync(interceptor, request, CancellationToken.None); + + // Assert + store.GetUploadedValue(request.Reason).Should().Be("rewind reason"); + } + + [Theory] + [InlineData(P.HistoryEvent.EventTypeOneofCase.ExecutionStarted)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.EventRaised)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.TaskScheduled)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.TaskCompleted)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.EventSent)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.GenericEvent)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.ContinueAsNew)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.ExecutionResumed)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.ExecutionRewound)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted)] + [InlineData(P.HistoryEvent.EventTypeOneofCase.HistoryState)] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_ExternalizesRewindHistoryPayloads( + P.HistoryEvent.EventTypeOneofCase eventType) + { + // Arrange + string payload = new('x', 64); + P.HistoryEvent historyEvent = CreateHistoryEventWithPayload(eventType, payload); + P.OrchestratorResponse response = new() + { + InstanceId = "instance-1", + Actions = + { + new P.OrchestratorAction + { + RewindOrchestration = new P.RewindOrchestrationAction + { + NewHistory = { historyEvent }, + }, + }, + }, + }; + + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert + IReadOnlyList tokens = GetHistoryEventPayloads(historyEvent); + foreach (string token in tokens) + { + token.Should().NotBe(payload); + store.GetUploadedValue(token).Should().Be(payload); + } + + store.UploadCount.Should().Be(tokens.Count); + } + + [Fact] + public async Task ExternalizeRequestPayloadsAsync_OrchestratorResponse_ShrinksOversizedRewindHistory() + { + // Arrange: ExecutionStarted is representative because every replacement history contains + // one; response-size handling is otherwise independent of the history event type. + const int MaxInlineResponseSize = 4_089_446; + string payload = new('x', MaxInlineResponseSize + 1024); + P.HistoryEvent historyEvent = CreateHistoryEventWithPayload( + P.HistoryEvent.EventTypeOneofCase.ExecutionStarted, + payload); + P.OrchestratorResponse response = new() + { + InstanceId = "instance-1", + Actions = + { + new P.OrchestratorAction + { + RewindOrchestration = new P.RewindOrchestrationAction + { + NewHistory = { historyEvent }, + }, + }, + }, + }; + response.CalculateSize().Should().BeGreaterThan(MaxInlineResponseSize); + + TrackingPayloadStore store = new(); + AzureBlobPayloadsSideCarInterceptor interceptor = new(store, CreateOptions()); + + // Act + await ExternalizeAsync(interceptor, response, CancellationToken.None); + + // Assert + store.GetUploadedValue(historyEvent.ExecutionStarted.Input).Should().Be(payload); + response.CalculateSize().Should().BeLessThan(MaxInlineResponseSize); + } + [Fact] public async Task ExternalizeRequestPayloadsAsync_EntityBatchRequest_ExternalizesOperationsInOrder() { @@ -937,6 +1094,110 @@ public async Task RunWithBoundedConcurrencyAsync_CancellationAfterAllOperationsC completedOperations.Should().Be(2); } + static P.HistoryEvent CreateHistoryEventWithPayload( + P.HistoryEvent.EventTypeOneofCase eventType, + string payload) + { + return eventType switch + { + P.HistoryEvent.EventTypeOneofCase.ExecutionStarted => + new() { ExecutionStarted = new P.ExecutionStartedEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted => + new() { ExecutionCompleted = new P.ExecutionCompletedEvent { Result = payload } }, + P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated => + new() { ExecutionTerminated = new P.ExecutionTerminatedEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.EventRaised => + new() { EventRaised = new P.EventRaisedEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.TaskScheduled => + new() { TaskScheduled = new P.TaskScheduledEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.TaskCompleted => + new() { TaskCompleted = new P.TaskCompletedEvent { Result = payload } }, + P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated => + new() + { + SubOrchestrationInstanceCreated = + new P.SubOrchestrationInstanceCreatedEvent { Input = payload }, + }, + P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted => + new() + { + SubOrchestrationInstanceCompleted = + new P.SubOrchestrationInstanceCompletedEvent { Result = payload }, + }, + P.HistoryEvent.EventTypeOneofCase.EventSent => + new() { EventSent = new P.EventSentEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.GenericEvent => + new() { GenericEvent = new P.GenericEvent { Data = payload } }, + P.HistoryEvent.EventTypeOneofCase.ContinueAsNew => + new() { ContinueAsNew = new P.ContinueAsNewEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended => + new() { ExecutionSuspended = new P.ExecutionSuspendedEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.ExecutionResumed => + new() { ExecutionResumed = new P.ExecutionResumedEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.ExecutionRewound => + new() { ExecutionRewound = new P.ExecutionRewoundEvent { Reason = payload } }, + P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled => + new() { EntityOperationSignaled = new P.EntityOperationSignaledEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled => + new() { EntityOperationCalled = new P.EntityOperationCalledEvent { Input = payload } }, + P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted => + new() { EntityOperationCompleted = new P.EntityOperationCompletedEvent { Output = payload } }, + P.HistoryEvent.EventTypeOneofCase.HistoryState => + new() + { + HistoryState = new P.HistoryStateEvent + { + OrchestrationState = new P.OrchestrationState + { + Input = payload, + Output = payload, + CustomStatus = payload, + }, + }, + }, + _ => throw new ArgumentOutOfRangeException(nameof(eventType), eventType, null), + }; + } + + static IReadOnlyList GetHistoryEventPayloads(P.HistoryEvent historyEvent) + { + return historyEvent.EventTypeCase switch + { + P.HistoryEvent.EventTypeOneofCase.ExecutionStarted => [historyEvent.ExecutionStarted.Input], + P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted => [historyEvent.ExecutionCompleted.Result], + P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated => [historyEvent.ExecutionTerminated.Input], + P.HistoryEvent.EventTypeOneofCase.EventRaised => [historyEvent.EventRaised.Input], + P.HistoryEvent.EventTypeOneofCase.TaskScheduled => [historyEvent.TaskScheduled.Input], + P.HistoryEvent.EventTypeOneofCase.TaskCompleted => [historyEvent.TaskCompleted.Result], + P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated => + [historyEvent.SubOrchestrationInstanceCreated.Input], + P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted => + [historyEvent.SubOrchestrationInstanceCompleted.Result], + P.HistoryEvent.EventTypeOneofCase.EventSent => [historyEvent.EventSent.Input], + P.HistoryEvent.EventTypeOneofCase.GenericEvent => [historyEvent.GenericEvent.Data], + P.HistoryEvent.EventTypeOneofCase.ContinueAsNew => [historyEvent.ContinueAsNew.Input], + P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended => [historyEvent.ExecutionSuspended.Input], + P.HistoryEvent.EventTypeOneofCase.ExecutionResumed => [historyEvent.ExecutionResumed.Input], + P.HistoryEvent.EventTypeOneofCase.ExecutionRewound => [historyEvent.ExecutionRewound.Reason], + P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled => + [historyEvent.EntityOperationSignaled.Input], + P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled => + [historyEvent.EntityOperationCalled.Input], + P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted => + [historyEvent.EntityOperationCompleted.Output], + P.HistoryEvent.EventTypeOneofCase.HistoryState => + [ + historyEvent.HistoryState.OrchestrationState.Input, + historyEvent.HistoryState.OrchestrationState.Output, + historyEvent.HistoryState.OrchestrationState.CustomStatus, + ], + _ => throw new ArgumentOutOfRangeException( + nameof(historyEvent), + historyEvent.EventTypeCase, + null), + }; + } + static LargePayloadStorageOptions CreateOptions() => new() { ThresholdBytes = 1 }; static Task ExternalizeAsync(AzureBlobPayloadsSideCarInterceptor interceptor, TRequest request, CancellationToken cancellation) From 30db89ba1c9ae48fe8dfe539dd8c7dd5190b68c7 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Mon, 14 Sep 2026 14:28:45 -0700 Subject: [PATCH 09/11] removed unused variable --- src/Worker/Grpc/RewindOrchestrationHandler.cs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/Worker/Grpc/RewindOrchestrationHandler.cs b/src/Worker/Grpc/RewindOrchestrationHandler.cs index 2707dd125..40c5e6456 100644 --- a/src/Worker/Grpc/RewindOrchestrationHandler.cs +++ b/src/Worker/Grpc/RewindOrchestrationHandler.cs @@ -42,8 +42,6 @@ internal static P.OrchestratorResponse CreateResponse( P.HistoryEvent? executionStartedEvent = allEvents.FirstOrDefault( e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted); - P.TraceContext? orchestrationParentTraceContext = rewindEvent.ParentTraceContext - ?? executionStartedEvent?.ExecutionStarted.ParentTraceContext; ActivityContext orchestrationParentContext = default; bool hasOrchestrationParentContext = false; From 6021f47eb2a9bb44db8f02e9a8d957968b734492 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Mon, 14 Sep 2026 15:55:48 -0700 Subject: [PATCH 10/11] fixing tracing --- src/Worker/Grpc/RewindOrchestrationHandler.cs | 25 ++++--- .../RewindOrchestrationHandlerTests.cs | 73 +++++++++++++++++++ 2 files changed, 86 insertions(+), 12 deletions(-) diff --git a/src/Worker/Grpc/RewindOrchestrationHandler.cs b/src/Worker/Grpc/RewindOrchestrationHandler.cs index 40c5e6456..09fa3d357 100644 --- a/src/Worker/Grpc/RewindOrchestrationHandler.cs +++ b/src/Worker/Grpc/RewindOrchestrationHandler.cs @@ -42,8 +42,8 @@ internal static P.OrchestratorResponse CreateResponse( P.HistoryEvent? executionStartedEvent = allEvents.FirstOrDefault( e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted); - ActivityContext orchestrationParentContext = default; - bool hasOrchestrationParentContext = false; + ActivityContext childTraceContextSource = orchestrationActivity?.Context ?? default; + bool hasChildTraceContextSource = orchestrationActivity is not null; HashSet failedTaskIds = []; foreach (P.HistoryEvent historyEvent in allEvents) @@ -92,10 +92,13 @@ internal static P.OrchestratorResponse CreateResponse( eventCopy.ExecutionStarted.ParentTraceContext = rewindEvent.ParentTraceContext.Clone(); } - hasOrchestrationParentContext = ActivityContext.TryParse( - eventCopy.ExecutionStarted.ParentTraceContext?.TraceParent, - eventCopy.ExecutionStarted.ParentTraceContext?.TraceState, - out orchestrationParentContext); + if (!hasChildTraceContextSource) + { + hasChildTraceContextSource = ActivityContext.TryParse( + eventCopy.ExecutionStarted.ParentTraceContext?.TraceParent, + eventCopy.ExecutionStarted.ParentTraceContext?.TraceState, + out childTraceContextSource); + } rewindAction.NewHistory.Add(eventCopy); continue; @@ -103,15 +106,13 @@ internal static P.OrchestratorResponse CreateResponse( if (historyEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated && failedTaskIds.Contains(historyEvent.EventId) - && hasOrchestrationParentContext) + && hasChildTraceContextSource) { - // We set a new client span ID here so that the execution of the rewound suborchestration is not tied to the - // old parent. ActivityContext newParentTraceContext = new( - orchestrationParentContext.TraceId, + childTraceContextSource.TraceId, ActivitySpanId.CreateRandom(), - orchestrationParentContext.TraceFlags, - orchestrationParentContext.TraceState); + childTraceContextSource.TraceFlags, + childTraceContextSource.TraceState); P.HistoryEvent eventCopy = historyEvent.Clone(); eventCopy.SubOrchestrationInstanceCreated.ParentTraceContext = diff --git a/test/Worker/Grpc.Tests/RewindOrchestrationHandlerTests.cs b/test/Worker/Grpc.Tests/RewindOrchestrationHandlerTests.cs index a55085590..57396e88f 100644 --- a/test/Worker/Grpc.Tests/RewindOrchestrationHandlerTests.cs +++ b/test/Worker/Grpc.Tests/RewindOrchestrationHandlerTests.cs @@ -228,6 +228,79 @@ public void CreateResponse_ReplacesFailedSubOrchestrationTraceContext( .Should().Be(CreateTraceContext(ExecutionTraceId, OldChildSpanId).TraceParent); } + [Fact] + public void CreateResponse_ReplacesFailedSubOrchestrationTraceContext_UsesCurrentActivity() + { + // Arrange + const string TraceId = "11111111111111111111111111111111"; + const string StoredParentSpanId = "2222222222222222"; + const string StoredTraceState = "stored=value"; + const string ActivityTraceState = "activity=value"; + const string OldChildSpanId = "3333333333333333"; + + P.HistoryEvent failedChildCreated = new() + { + EventId = 4, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = "failed-child", + ParentTraceContext = CreateTraceContext(TraceId, OldChildSpanId), + }, + }; + P.HistoryEvent[] pastEvents = + [ + CreateExecutionStarted( + "old-execution", + parentTraceContext: CreateTraceContext(TraceId, StoredParentSpanId, StoredTraceState)), + failedChildCreated, + new() + { + EventId = 5, + SubOrchestrationInstanceFailed = + new P.SubOrchestrationInstanceFailedEvent { TaskScheduledId = 4 }, + }, + ]; + + using Activity orchestrationActivity = new("rewind"); + orchestrationActivity.SetIdFormat(ActivityIdFormat.W3C); + orchestrationActivity.SetParentId( + ActivityTraceId.CreateFromString(TraceId.AsSpan()), + ActivitySpanId.CreateFromString(StoredParentSpanId.AsSpan()), + ActivityTraceFlags.Recorded); + orchestrationActivity.TraceStateString = StoredTraceState; + orchestrationActivity.Start(); + orchestrationActivity.ActivityTraceFlags = ActivityTraceFlags.None; + orchestrationActivity.TraceStateString = ActivityTraceState; + + // Act + P.OrchestratorResponse response = RewindOrchestrationHandler.CreateResponse( + CreateRewindRequest(), + pastEvents, + "completion-token", + orchestrationActivity); + + // Assert + P.SubOrchestrationInstanceCreatedEvent rewrittenChild = response + .Actions[0] + .RewindOrchestration + .NewHistory + .Single(e => e.EventId == 4) + .SubOrchestrationInstanceCreated; + ActivityContext.TryParse( + rewrittenChild.ParentTraceContext.TraceParent, + rewrittenChild.ParentTraceContext.TraceState, + out ActivityContext childTraceContext).Should().BeTrue(); + + childTraceContext.TraceId.Should().Be(orchestrationActivity.TraceId); + childTraceContext.SpanId.Should().NotBe(orchestrationActivity.SpanId); + childTraceContext.SpanId.ToString().Should().NotBe(OldChildSpanId); + childTraceContext.TraceFlags.Should().Be(ActivityTraceFlags.None); + childTraceContext.TraceState.Should().Be(ActivityTraceState); + + // The stored parent context should remain unchanged. + pastEvents[0].ExecutionStarted.ParentTraceContext.TraceState.Should().Be(StoredTraceState); + } + [Fact] public void CreateResponse_RejectsUnexpectedNewEvents() { From 490b9ae0a5f41a3221f2344ebc601aa48d6b9915 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Mon, 14 Sep 2026 16:12:19 -0700 Subject: [PATCH 11/11] comment update --- src/Worker/Grpc/RewindOrchestrationHandler.cs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/Worker/Grpc/RewindOrchestrationHandler.cs b/src/Worker/Grpc/RewindOrchestrationHandler.cs index 09fa3d357..78c320807 100644 --- a/src/Worker/Grpc/RewindOrchestrationHandler.cs +++ b/src/Worker/Grpc/RewindOrchestrationHandler.cs @@ -61,8 +61,6 @@ internal static P.OrchestratorResponse CreateResponse( string newExecutionId = Guid.NewGuid().ToString("N"); P.RewindOrchestrationAction rewindAction = new(); - // Retry timers are retained to match the existing rewind protocol. Rewinding failed activities - // that were scheduled with retry policies is not currently supported. foreach (P.HistoryEvent historyEvent in allEvents) { // Do not add any failed tasks or the failed execution completed event to the new history