From 537a330203c840f8267628c8433c36accf439657 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Wed, 25 Mar 2026 08:40:12 +0000 Subject: [PATCH] fix: prevent infinite loop in suspend/resume event buffer iteration handleExecutionResumed iterates _suspendedEvents using for..of while processEvent can push new entries onto the same array. If a buffered ExecutionSuspended event sets _isSuspended=true during iteration, subsequent suspendable events are re-pushed to the array being iterated, creating an infinite loop that hangs the worker process. Two fixes: 1. Snapshot _suspendedEvents before iterating and clear it first. Events buffered during processing land in the new (empty) array and are correctly held for the next Resume event. 2. Add ExecutionSuspended to the isSuspendable exclusion list so suspend events are processed immediately rather than buffered. Also uses strict equality (===) instead of loose (==) for the indexOf check. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- packages/durabletask-js/src/worker/index.ts | 8 +- .../src/worker/orchestration-executor.ts | 12 ++- .../test/orchestration_executor.spec.ts | 77 +++++++++++++++++++ 3 files changed, 91 insertions(+), 6 deletions(-) diff --git a/packages/durabletask-js/src/worker/index.ts b/packages/durabletask-js/src/worker/index.ts index 36f8c0b..2afac86 100644 --- a/packages/durabletask-js/src/worker/index.ts +++ b/packages/durabletask-js/src/worker/index.ts @@ -110,8 +110,10 @@ export function getActionSummary(newActions: pb.OrchestratorAction[]): string { */ export function isSuspendable(event: pb.HistoryEvent): boolean { return ( - [pb.HistoryEvent.EventtypeCase.EXECUTIONRESUMED, pb.HistoryEvent.EventtypeCase.EXECUTIONTERMINATED].indexOf( - event.getEventtypeCase(), - ) == -1 + [ + pb.HistoryEvent.EventtypeCase.EXECUTIONRESUMED, + pb.HistoryEvent.EventtypeCase.EXECUTIONTERMINATED, + pb.HistoryEvent.EventtypeCase.EXECUTIONSUSPENDED, + ].indexOf(event.getEventtypeCase()) === -1 ); } diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index e081d32..6df108c 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -470,11 +470,17 @@ export class OrchestrationExecutor { this._isSuspended = false; - for (const e of this._suspendedEvents) { + // Snapshot and clear before iterating. If any buffered event (e.g. a + // second ExecutionSuspended) sets _isSuspended back to true during + // processing, subsequent suspendable events will be pushed into the + // now-empty _suspendedEvents array instead of the array we are + // iterating, preventing an infinite loop. + const eventsToProcess = this._suspendedEvents; + this._suspendedEvents = []; + + for (const e of eventsToProcess) { await this.processEvent(ctx, e); } - - this._suspendedEvents = []; } private async handleExecutionTerminated(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { diff --git a/packages/durabletask-js/test/orchestration_executor.spec.ts b/packages/durabletask-js/test/orchestration_executor.spec.ts index 741d231..6b9e2d0 100644 --- a/packages/durabletask-js/test/orchestration_executor.spec.ts +++ b/packages/durabletask-js/test/orchestration_executor.spec.ts @@ -717,6 +717,83 @@ describe("Orchestration Executor", () => { expect(completeAction?.getResult()?.getValue()).toEqual("42"); }); + it("should not enter an infinite loop when buffered events include a second suspend", async () => { + // This test verifies two fixes working together: + // 1. isSuspendable now excludes ExecutionSuspended, so a second suspend + // is processed immediately (as a no-op) rather than being buffered. + // 2. handleExecutionResumed snapshots the buffer before iterating, + // preventing any re-entrant modification from causing an infinite loop. + // + // Without these fixes, a buffered SUSPEND event would set _isSuspended=true + // during iteration, causing subsequent events to be re-pushed onto the same + // array, creating an infinite loop that hangs the worker. + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, _: any): any { + const res1 = yield ctx.waitForExternalEvent("event1"); + const res2 = yield ctx.waitForExternalEvent("event2"); + return [res1, res2]; + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + + const oldEvents = [newOrchestratorStartedEvent(), newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID)]; + + // Suspend, buffer an event, suspend again, buffer another event + let newEvents = [ + newSuspendEvent(), + newEventRaisedEvent("event1", JSON.stringify("val1")), + newSuspendEvent(), + newEventRaisedEvent("event2", JSON.stringify("val2")), + ]; + + let executor = new OrchestrationExecutor(registry, testLogger); + let result = await executor.execute(TEST_INSTANCE_ID, oldEvents, newEvents); + // Both events are buffered; second suspend is processed immediately (no-op) + expect(result.actions.length).toBe(0); + + // Resume: both buffered events are delivered, orchestration completes + oldEvents.push(...newEvents); + newEvents = [newResumeEvent()]; + + executor = new OrchestrationExecutor(registry, testLogger); + result = await executor.execute(TEST_INSTANCE_ID, oldEvents, newEvents); + + const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); + expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(completeAction?.getResult()?.getValue()).toEqual(JSON.stringify(["val1", "val2"])); + }); + + it("should treat ExecutionSuspended as non-suspendable so it is never buffered", async () => { + // Verify that the isSuspendable function correctly excludes + // ExecutionSuspended events from being buffered, providing + // defense-in-depth against the infinite loop scenario. + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, _: any): any { + const res = yield ctx.waitForExternalEvent("my_event"); + return res; + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + + const oldEvents = [newOrchestratorStartedEvent(), newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID)]; + + // Suspend, then immediately suspend again (the second suspend should be processed immediately, not buffered) + // Then an event arrives while suspended, followed by resume + const newEvents = [ + newSuspendEvent(), + newSuspendEvent(), + newEventRaisedEvent("my_event", JSON.stringify("hello")), + newResumeEvent(), + ]; + + const executor = new OrchestrationExecutor(registry, testLogger); + const result = await executor.execute(TEST_INSTANCE_ID, oldEvents, newEvents); + + const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); + expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(completeAction?.getResult()?.getValue()).toEqual(JSON.stringify("hello")); + }); + it("should test that an orchestration can be terminated before it completes", async () => { const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, _: any): any { const res = yield ctx.waitForExternalEvent("my_event");