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