Adding rewind to the sidecar

.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

src/Extensions/AzureBlobPayloads/Interceptors/AzureBlobPayloadsSideCarInterceptor.cs

@@ -48,6 +48,9 @@ protected override async Task ExternalizeRequestPayloadsAsync<TRequest>(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<Func<Task>> 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)

src/Shared/Grpc/ProtoUtils.cs

@@ -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)
@@ -500,6 +496,21 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse(
         return response;
     }
 
+    /// <summary>
+    /// Creates a protobuf trace context from an activity context.
+    /// </summary>
+    /// <param name="activityContext">The activity context to convert.</param>
+    /// <returns>The corresponding protobuf trace context.</returns>
+    internal static P.TraceContext CreateTraceContext(ActivityContext activityContext)
+    {
+        return new()
+        {
+            TraceParent =
+                $"00-{activityContext.TraceId}-{activityContext.SpanId}-0{activityContext.TraceFlags:d}",
+            TraceState = activityContext.TraceState,
+        };
+    }
+
     /// <summary>
     /// Converts a <see cref="P.OrchestrationStatus" /> to a <see cref="OrchestrationStatus" />.
     /// </summary>

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -247,16 +247,42 @@ static string GetActionsListForLogging(IReadOnlyList<P.OrchestratorAction> actio
             return failureDetails;
         }
 
-        async ValueTask<OrchestrationRuntimeState> BuildRuntimeStateAsync(
+        static OrchestrationRuntimeState BuildRuntimeState(
             P.OrchestratorRequest orchestratorRequest,
-            ProtoUtils.EntityConversionState? entityConversionState,
-            CancellationToken cancellation)
+            IReadOnlyList<P.HistoryEvent> pastEvents,
+            ProtoUtils.EntityConversionState? entityConversionState)
         {
             Func<P.HistoryEvent, HistoryEvent> converter = entityConversionState is null
                 ? ProtoUtils.ConvertHistoryEvent
                 : entityConversionState.ConvertFromProto;
 
-            List<HistoryEvent> pastEvents;
+            List<HistoryEvent> 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<IReadOnlyList<P.HistoryEvent>> GetPastEventsAsync(
+            P.OrchestratorRequest orchestratorRequest,
+            CancellationToken cancellation)
+        {
             if (orchestratorRequest.RequiresHistoryStreaming)
             {
                 // Stream the remaining events from the remote service
@@ -280,40 +306,19 @@ async ValueTask<OrchestrationRuntimeState> BuildRuntimeStateAsync(
                 // chunks (e.g. one event per chunk) that would reallocate and copy on every chunk, which is
                 // itself quadratic. List<T>.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<HistoryEvent>();
+                List<P.HistoryEvent> 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<HistoryEvent>(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<AsyncServerStreamingCall<P.WorkItem>> ConnectAsync(CancellationToken cancellation)
@@ -602,25 +607,90 @@ 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<P.HistoryEvent>? 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<P.HistoryEvent> 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)
+            {
+                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.
+            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 +700,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 +787,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)