From d642859ba6e0464d93dcdd72edddb52a8711e6e5 Mon Sep 17 00:00:00 2001 From: Copilot Date: Sun, 29 Mar 2026 08:33:20 +0000 Subject: [PATCH] fix: cancel pending timers on continue-as-new in InMemoryOrchestrationBackend When an orchestration calls continueAsNew(), the InMemoryOrchestrationBackend resets instance state (history, status, input) but does not cancel pending setTimeout handles from the previous iteration. These stale timers continue to fire and deliver TimerFired events into the new iteration. Because the new iteration resets the sequence counter, stale timer IDs can collide with task IDs in the new iteration, potentially completing the wrong task or causing unexpected-event warnings. This commit: - Adds per-instance timer tracking (instanceTimers map) - Cancels all pending timers for an instance on continue-as-new - Fixes carryover event ordering so OrchestratorStarted and ExecutionStarted come before carryover events, matching real sidecar behavior - Adds tests for both the timer cancellation and event ordering fixes Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/testing/in-memory-backend.ts | 53 +++++++++++++-- .../test/in-memory-backend.spec.ts | 66 +++++++++++++++++++ 2 files changed, 113 insertions(+), 6 deletions(-) diff --git a/packages/durabletask-js/src/testing/in-memory-backend.ts b/packages/durabletask-js/src/testing/in-memory-backend.ts index 2c96acb1..9920cc3c 100644 --- a/packages/durabletask-js/src/testing/in-memory-backend.ts +++ b/packages/durabletask-js/src/testing/in-memory-backend.ts @@ -62,6 +62,7 @@ export class InMemoryOrchestrationBackend { private readonly activityQueue: ActivityWorkItem[] = []; private readonly stateWaiters: Map = new Map(); private readonly pendingTimers: Set> = new Set(); + private readonly instanceTimers: Map>> = new Map(); private nextCompletionToken: number = 1; private readonly maxHistorySize: number; @@ -394,6 +395,7 @@ export class InMemoryOrchestrationBackend { clearTimeout(timer); } this.pendingTimers.clear(); + this.instanceTimers.clear(); } /** @@ -477,6 +479,11 @@ export class InMemoryOrchestrationBackend { const newInput = completeAction.getResult()?.getValue(); const carryoverEvents = completeAction.getCarryovereventsList(); + // Cancel all pending timers for this instance. Timers from the previous + // iteration must not fire into the new iteration — their timer IDs could + // collide with new sequence numbers and complete the wrong tasks. + this.cancelInstanceTimers(instance.instanceId); + // Reset instance state instance.history = []; instance.input = newInput; @@ -485,14 +492,11 @@ export class InMemoryOrchestrationBackend { instance.failureDetails = undefined; instance.status = pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING; - // Add carryover events - instance.pendingEvents = [...carryoverEvents]; - - // Add new execution started events + // Build new events: OrchestratorStarted and ExecutionStarted must come + // before carryover events, matching the real sidecar's event ordering. const orchestratorStarted = pbh.newOrchestratorStartedEvent(new Date()); const executionStarted = pbh.newExecutionStartedEvent(instance.name, instance.instanceId, newInput); - instance.pendingEvents.push(orchestratorStarted); - instance.pendingEvents.push(executionStarted); + instance.pendingEvents = [orchestratorStarted, executionStarted, ...carryoverEvents]; this.enqueueOrchestration(instance.instanceId); } @@ -543,6 +547,7 @@ export class InMemoryOrchestrationBackend { const timerHandle = setTimeout(() => { this.pendingTimers.delete(timerHandle); + this.removeInstanceTimer(instance.instanceId, timerHandle); const currentInstance = this.instances.get(instance.instanceId); if (currentInstance && !this.isTerminalStatus(currentInstance.status)) { const timerFiredEvent = pbh.newTimerFiredEvent(timerId, fireAt); @@ -552,6 +557,7 @@ export class InMemoryOrchestrationBackend { } }, delay); this.pendingTimers.add(timerHandle); + this.addInstanceTimer(instance.instanceId, timerHandle); } private processCreateSubOrchestrationAction(instance: OrchestrationInstance, action: pb.OrchestratorAction): void { @@ -638,6 +644,41 @@ export class InMemoryOrchestrationBackend { } } + private addInstanceTimer(instanceId: string, timerHandle: ReturnType): void { + let timers = this.instanceTimers.get(instanceId); + if (!timers) { + timers = new Set(); + this.instanceTimers.set(instanceId, timers); + } + timers.add(timerHandle); + } + + private removeInstanceTimer(instanceId: string, timerHandle: ReturnType): void { + const timers = this.instanceTimers.get(instanceId); + if (timers) { + timers.delete(timerHandle); + if (timers.size === 0) { + this.instanceTimers.delete(instanceId); + } + } + } + + /** + * Cancels all pending timers for a specific instance. + * Used during continue-as-new to prevent stale timers from the previous + * iteration from firing into the new iteration. + */ + private cancelInstanceTimers(instanceId: string): void { + const timers = this.instanceTimers.get(instanceId); + if (timers) { + for (const timer of timers) { + clearTimeout(timer); + this.pendingTimers.delete(timer); + } + this.instanceTimers.delete(instanceId); + } + } + private notifyWaiters(instanceId: string): void { const instance = this.instances.get(instanceId); const waiters = this.stateWaiters.get(instanceId); diff --git a/packages/durabletask-js/test/in-memory-backend.spec.ts b/packages/durabletask-js/test/in-memory-backend.spec.ts index 66fe6a96..698230bd 100644 --- a/packages/durabletask-js/test/in-memory-backend.spec.ts +++ b/packages/durabletask-js/test/in-memory-backend.spec.ts @@ -377,4 +377,70 @@ describe("In-Memory Backend", () => { expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED); expect(state?.serializedOutput).toEqual(JSON.stringify(42)); }); + + it("should cancel pending timers on continue-as-new", async () => { + // This test verifies that timers from a previous iteration are cancelled + // when the orchestration does continue-as-new. Without this fix, stale + // timers fire into the new iteration and deliver events with timer IDs + // that could collide with new tasks. + let iteration2TimerFired = false; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, input: number): any { + if (input === 1) { + // First iteration: create a long timer, then immediately continue-as-new. + // The timer should be cancelled and NOT fire into the second iteration. + ctx.createTimer(60); // 60-second timer — should be cancelled + ctx.continueAsNew(2, false); + } else { + // Second iteration: create a short timer and wait for it. + // If the stale timer from iteration 1 leaks, it could complete + // the wrong task or cause unexpected events. + yield ctx.createTimer(0.05); // 50ms timer + iteration2TimerFired = true; + return "done"; + } + }; + + worker.addOrchestrator(orchestrator); + await worker.start(); + + const id = await client.scheduleNewOrchestration(orchestrator, 1); + const state = await client.waitForOrchestrationCompletion(id, true, 10); + + expect(state).toBeDefined(); + expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toEqual(JSON.stringify("done")); + expect(iteration2TimerFired).toBe(true); + }); + + it("should carry over external events in correct order after continue-as-new", async () => { + // Verifies that carryover events from continue-as-new are placed after + // OrchestratorStarted and ExecutionStarted, matching real sidecar behavior. + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, input: number): any { + if (input === 1) { + // First iteration: buffer an external event but don't consume it, + // then continue-as-new with saveEvents=true + ctx.continueAsNew(2, true); + } else { + // Second iteration: the carryover event should be available + const value = yield ctx.waitForExternalEvent("my_event"); + return value; + } + }; + + worker.addOrchestrator(orchestrator); + await worker.start(); + + const id = await client.scheduleNewOrchestration(orchestrator, 1); + + // Raise the event before orchestration processes — it will be buffered + // and carried over when continue-as-new is called with saveEvents=true + await client.raiseOrchestrationEvent(id, "my_event", "carried-over"); + + const state = await client.waitForOrchestrationCompletion(id, true, 10); + + expect(state).toBeDefined(); + expect(state?.runtimeStatus).toEqual(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toEqual(JSON.stringify("carried-over")); + }); });