diff --git a/src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs b/src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs index f19ca532..3dfe627b 100644 --- a/src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs +++ b/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(); + var timerMessages = new List(); + var orchestratorMessages = new List(); + + // 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 activityMessages, - out List orchestratorMessages, - out List 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 activityMessages, // CA1859: Use concrete types for better performance - out List orchestratorMessages, // CA1859: Use concrete types for better performance - out List timerMessages, // CA1859: Use concrete types for better performance + List activityMessages, // CA1859: Use concrete types for better performance + List orchestratorMessages, // CA1859: Use concrete types for better performance + List 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? newActivityMessages = null; // CA1859: Use concrete types for better performance - List? newTimerMessages = null; // CA1859: Use concrete types for better performance - List? 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(); - 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(); - 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(); - 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(); - 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(); - orchestratorMessages = new List(); - timerMessages = new List(); continueAsNew = true; updatedStatus = null; return; @@ -457,8 +456,7 @@ void ApplyOrchestratorActions( completeAction.FailureDetails); } - newOrchestratorMessages ??= new List(); - 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(); - timerMessages = newTimerMessages ?? new List(); - orchestratorMessages = newOrchestratorMessages ?? new List(); - updatedStatus = new OrchestrationState { OrchestrationInstance = runtimeState.OrchestrationInstance, diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs index db803563..f7036d28 100644 --- a/src/Shared/Grpc/ProtoUtils.cs +++ b/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"); } /// diff --git a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs index b03ff679..500a1952 100644 --- a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs +++ b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs @@ -1145,7 +1145,51 @@ public ValueTask IsOrchestrationValidAsync(OrchestrationFilterParameters i } } + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task ContinueAsNewEventsArePreserved(bool injectTimers) + { + const int EventCount = 10; + async Task OrchestratorFunc(TaskOrchestrationContext ctx, int counter) + { + await ctx.WaitForExternalEvent("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(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()); + } + // 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 }