Initial attempt to fix carryover events issue on continue-as-new (#496)

* Initial attempt to fix #387

* fixing the test dispatcher

* trying to remove the large diff for the ProtoUtils file

---------

Co-authored-by: Sophia Tevosyan <[email protected]>

Author: cgillum

src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs

@@ -111,8 +111,12 @@ protected override async Task ExecuteWorkItemAsync(TaskOrchestrationWorkItem wor
         {
             throw new ArgumentException($"Could not find an orchestration instance ID in the work item's runtime state.", nameof(workItem));
         }
-
-        // We loop for as long as the orchestrator does a ContinueAsNew
+
+        var activityMessages = new List<TaskMessage>();
+        var timerMessages = new List<TaskMessage>();
+        var orchestratorMessages = new List<TaskMessage>();
+
+        // We loop for as long as the orchestrator does a ContinueAsNew
         while (true)
         {
             if (this.log.IsEnabled(LogLevel.Debug))
@@ -138,9 +142,9 @@ protected override async Task ExecuteWorkItemAsync(TaskOrchestrationWorkItem wor
             this.ApplyOrchestratorActions(
                 result,
                 ref workItem.OrchestrationRuntimeState,
-                out List<TaskMessage> activityMessages,
-                out List<TaskMessage> orchestratorMessages,
-                out List<TaskMessage> timerMessages,
+                activityMessages,
+                orchestratorMessages,
+                timerMessages,
                 out OrchestrationState? updatedStatus,
                 out bool continueAsNew);
             if (continueAsNew)
@@ -247,9 +251,9 @@ static string GetShortHistoryEventDescription(HistoryEvent e)
     void ApplyOrchestratorActions(
         OrchestratorExecutionResult result,
         ref OrchestrationRuntimeState runtimeState,
-        out List<TaskMessage> activityMessages, // CA1859: Use concrete types for better performance
-        out List<TaskMessage> orchestratorMessages, // CA1859: Use concrete types for better performance
-        out List<TaskMessage> timerMessages, // CA1859: Use concrete types for better performance
+        List<TaskMessage> activityMessages, // CA1859: Use concrete types for better performance
+        List<TaskMessage> orchestratorMessages, // CA1859: Use concrete types for better performance
+        List<TaskMessage> timerMessages, // CA1859: Use concrete types for better performance
         out OrchestrationState? updatedStatus,
         out bool continueAsNew)
     {
@@ -258,9 +262,6 @@ void ApplyOrchestratorActions(
             throw new ArgumentException($"The provided {nameof(OrchestrationRuntimeState)} doesn't contain an instance ID!", nameof(runtimeState));
         }
 
-        List<TaskMessage>? newActivityMessages = null; // CA1859: Use concrete types for better performance
-        List<TaskMessage>? newTimerMessages = null; // CA1859: Use concrete types for better performance
-        List<TaskMessage>? newOrchestratorMessages = null; // CA1859: Use concrete types for better performance
         FailureDetails? failureDetails = null;
         continueAsNew = false;
 
@@ -288,8 +289,7 @@ void ApplyOrchestratorActions(
                     scheduledEvent.ParentTraceContext ??= new(grpcAction.ParentTraceContext.TraceParent, grpcAction.ParentTraceContext.TraceState);
                 }
 
-                newActivityMessages ??= new List<TaskMessage>();
-                newActivityMessages.Add(new TaskMessage
+                activityMessages.Add(new TaskMessage
                 {
                     Event = scheduledEvent,
                     OrchestrationInstance = runtimeState.OrchestrationInstance,
@@ -301,8 +301,7 @@ void ApplyOrchestratorActions(
             {
                 TimerCreatedEvent timerEvent = new(timerAction.Id, timerAction.FireAt);
 
-                newTimerMessages ??= new List<TaskMessage>();
-                newTimerMessages.Add(new TaskMessage
+                timerMessages.Add(new TaskMessage
                 {
                     Event = new TimerFiredEvent(-1, timerAction.FireAt)
                     {
@@ -346,8 +345,7 @@ void ApplyOrchestratorActions(
                     Tags = subOrchestrationAction.Tags,
                 };
 
-                newOrchestratorMessages ??= new List<TaskMessage>();
-                newOrchestratorMessages.Add(new TaskMessage
+                orchestratorMessages.Add(new TaskMessage
                 {
                     Event = startedEvent,
                     OrchestrationInstance = startedEvent.OrchestrationInstance,
@@ -367,13 +365,17 @@ void ApplyOrchestratorActions(
                     Input = sendEventAction.EventData,
                 };
 
-                runtimeState.AddEvent(sendEvent);
+                runtimeState.AddEvent(sendEvent);
+
+                EventRaisedEvent eventRaisedEvent = new(-1, sendEventAction.EventData)
+                {
+                    Name = sendEventAction.EventName
+                };
 
-                newOrchestratorMessages ??= new List<TaskMessage>();
-                newOrchestratorMessages.Add(new TaskMessage
+                orchestratorMessages.Add(new TaskMessage
                 {
-                    Event = sendEvent,
-                    OrchestrationInstance = runtimeState.OrchestrationInstance,
+                    Event = eventRaisedEvent,
+                    OrchestrationInstance = sendEventAction.Instance,
                 });
             }
             else if (action is OrchestrationCompleteOrchestratorAction completeAction)
@@ -408,9 +410,6 @@ void ApplyOrchestratorActions(
                     }
 
                     runtimeState = newRuntimeState;
-                    activityMessages = new List<TaskMessage>();
-                    orchestratorMessages = new List<TaskMessage>();
-                    timerMessages = new List<TaskMessage>();
                     continueAsNew = true;
                     updatedStatus = null;
                     return;
@@ -457,8 +456,7 @@ void ApplyOrchestratorActions(
                             completeAction.FailureDetails);
                     }
 
-                    newOrchestratorMessages ??= new List<TaskMessage>();
-                    newOrchestratorMessages.Add(new TaskMessage
+                    orchestratorMessages.Add(new TaskMessage
                     {
                         Event = subOrchestratorCompletedEvent,
                         OrchestrationInstance = runtimeState.ParentInstance.OrchestrationInstance,
@@ -475,10 +473,6 @@ void ApplyOrchestratorActions(
 
         runtimeState.AddEvent(new OrchestratorCompletedEvent(-1));
 
-        activityMessages = newActivityMessages ?? new List<TaskMessage>();
-        timerMessages = newTimerMessages ?? new List<TaskMessage>();
-        orchestratorMessages = newOrchestratorMessages ?? new List<TaskMessage>();
-
         updatedStatus = new OrchestrationState
         {
             OrchestrationInstance = runtimeState.OrchestrationInstance,

src/Shared/Grpc/ProtoUtils.cs

@@ -461,11 +461,8 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse(
 
                     var completeAction = (OrchestrationCompleteOrchestratorAction)action;
                     protoAction.CompleteOrchestration = new P.CompleteOrchestrationAction
-                    {
-                        CarryoverEvents =
-                        {
-                            // TODO
-                        },
+                    {
+                        CarryoverEvents = { completeAction.CarryoverEvents.Select(ToProtobuf) },
                         Details = completeAction.Details,
                         NewVersion = completeAction.NewVersion,
                         OrchestrationStatus = completeAction.OrchestrationStatus.ToProtobuf(),
@@ -1170,6 +1167,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");
     }
 
     /// <summary>

test/Grpc.IntegrationTests/OrchestrationPatterns.cs

@@ -1145,7 +1145,51 @@ public ValueTask<bool> IsOrchestrationValidAsync(OrchestrationFilterParameters i
         }
     }
 
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public async Task ContinueAsNewEventsArePreserved(bool injectTimers)
+    {
+        const int EventCount = 10;
+        async Task<int> OrchestratorFunc(TaskOrchestrationContext ctx, int counter)
+        {
+            await ctx.WaitForExternalEvent<string>("Event");
+            counter++;
+
+            if (injectTimers)
+            {
+                await ctx.CreateTimer(TimeSpan.FromMilliseconds(1), CancellationToken.None);
+            }
+
+            if (counter < EventCount)
+            {
+                ctx.ContinueAsNew(counter, preserveUnprocessedEvents: true);
+            }
+
+            return counter;
+        }
+
+        await using HostTestLifetime server = await this.StartWorkerAsync(
+            b => b.AddTasks(
+                tasks => tasks.AddOrchestratorFunc<int, int>(nameof(OrchestratorFunc), OrchestratorFunc)));
+
+        string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(
+            nameof(OrchestratorFunc),
+            input: 0);
+
+        for (int i = 0; i < EventCount; i++)
+        {
+            await server.Client.RaiseEventAsync(instanceId, eventName: "Event");
+            await Task.Delay(TimeSpan.FromMilliseconds(1), this.TimeoutToken);
+        }
+
+        OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync(
+            instanceId, getInputsAndOutputs: true, this.TimeoutToken);
+        Assert.NotNull(metadata);
+        Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
+        Assert.Equal(EventCount, metadata.ReadOutputAs<int>());
+    }
+
     // TODO: Test for multiple external events with the same name
-    // TODO: Test for ContinueAsNew with external events that carry over
     // TODO: Test for catching activity exceptions of specific types
 }