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")); + }); });