Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sophiatev merged 3 commits into from
Nov 11, 2025
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
906a8e5
Initial attempt to fix https://github.com/microsoft/durabletask-dotne…
cgillum Nov 7, 2025
86256c0
fixing the test dispatcher
Nov 11, 2025
63a5055
trying to remove the large diff for the ProtoUtils file
Nov 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 25 additions & 31 deletions src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand All @@ -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)
Expand Down Expand Up @@ -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)
{
Expand All @@ -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;

Expand Down Expand Up @@ -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,
Expand All @@ -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)
{
Expand Down Expand Up @@ -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,
Expand All @@ -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)
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down
29 changes: 24 additions & 5 deletions src/Shared/Grpc/ProtoUtils.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) },
Comment thread
cgillum marked this conversation as resolved.
Details = completeAction.Details,
NewVersion = completeAction.NewVersion,
OrchestrationStatus = completeAction.OrchestrationStatus.ToProtobuf(),
Expand Down Expand Up @@ -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>
Expand Down
46 changes: 45 additions & 1 deletion test/Grpc.IntegrationTests/OrchestrationPatterns.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Loading