fix: cancel pending timers on continue-as-new in InMemoryOrchestrationBackend

packages/durabletask-js/src/testing/in-memory-backend.ts

@@ -62,6 +62,7 @@ export class InMemoryOrchestrationBackend {
   private readonly activityQueue: ActivityWorkItem[] = [];
   private readonly stateWaiters: Map<string, StateWaiter[]> = new Map();
   private readonly pendingTimers: Set<ReturnType<typeof setTimeout>> = new Set();
+  private readonly instanceTimers: Map<string, Set<ReturnType<typeof setTimeout>>> = 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<typeof setTimeout>): 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<typeof setTimeout>): 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);

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