From 80af1b0994ce818fa74bb21415b90385f780de6d Mon Sep 17 00:00:00 2001 From: wangbill Date: Fri, 9 Oct 2026 09:45:22 -0400 Subject: [PATCH] fix: preserve trailing events after continue-as-new Buffer external events arriving after continue-as-new instead of completing abandoned listeners. Add real-batch public regressions and document opt-out and consumed-event behavior. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 03e10c9a-ff91-410a-bf7c-84b1ca56e0f6 --- CHANGELOG.md | 4 + README.md | 12 +- .../src/task/context/orchestration-context.ts | 4 + .../src/worker/orchestration-executor.ts | 3 +- .../test/external-event-routing.spec.ts | 222 ++++++++++++++++++ .../test/orchestration_executor.spec.ts | 60 +++++ 6 files changed, 301 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c90183..b2d6d9f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,10 @@ ### Fixes +- Preserve trailing external events in the same work item after `continueAsNew(..., true)`, + rather than delivering them to abandoned listeners in the previous execution. Already-consumed + events are not duplicated; `saveEvents=false` still discards unprocessed events. Case-insensitive + event-name matching, per-name FIFO ordering, and payload values are unchanged. - Handle external event names such as `constructor` and `__proto__` without failing the orchestration. Case-insensitive matching, per-name FIFO ordering, and unprocessed-event carryover through continue-as-new are preserved. diff --git a/README.md b/README.md index eb32839..0ebb0a4 100644 --- a/README.md +++ b/README.md @@ -434,9 +434,15 @@ const eternalOrchestrator: TOrchestrator = async function* ( }; ``` -The second argument controls whether unprocessed external events carry over. The optional third -argument becomes the restarted orchestration's `ctx.version`; omit it to retain the existing -continue-as-new behavior. +The second argument controls whether unprocessed external events carry over. With `true`, events +delivered later in the same work item after `continueAsNew()` are retained for the new execution, +even if an abandoned listener in the previous execution was waiting for them. Events already +consumed before the call are not carried over; `false` explicitly discards unprocessed events. +Case-insensitive event-name matching, per-name FIFO ordering, and payload values are preserved. +This corrects previously lost trailing events but cannot recover events discarded by an earlier worker. + +The optional third argument becomes the restarted orchestration's `ctx.version`; omit it to retain +the existing continue-as-new version behavior. ### Durable entities diff --git a/packages/durabletask-js/src/task/context/orchestration-context.ts b/packages/durabletask-js/src/task/context/orchestration-context.ts index fb4bc29..174df24 100644 --- a/packages/durabletask-js/src/task/context/orchestration-context.ts +++ b/packages/durabletask-js/src/task/context/orchestration-context.ts @@ -171,6 +171,10 @@ export abstract class OrchestrationContext { /** * Continue the orchestration execution as a new instance * + * With saveEvents=true, unprocessed external events later in the same work item + * are carried over even if the previous execution registered a listener. + * Events consumed before this call are not carried over. + * * @param newInput {any} The new input to use for the new orchestration instance. * @param saveEvents {boolean} A flag indicating whether to add any unprocessed external events in the new orchestration history. * @param newVersion {string} The optional version to use for the new orchestration instance. diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index 8c47c81..9e1aaa5 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -492,7 +492,8 @@ export class OrchestrationExecutor { let decodedResult; - if (taskList) { + // After continue-as-new, old listeners must not consume events for the next execution. + if (taskList && !ctx._isComplete) { const eventTask = taskList.shift(); decodedResult = parseJsonField(event.getEventraised()?.getInput()); diff --git a/packages/durabletask-js/test/external-event-routing.spec.ts b/packages/durabletask-js/test/external-event-routing.spec.ts index 772322e..1e4bddb 100644 --- a/packages/durabletask-js/test/external-event-routing.spec.ts +++ b/packages/durabletask-js/test/external-event-routing.spec.ts @@ -8,7 +8,9 @@ import { OrchestrationContext, OrchestrationStatus, TOrchestrator, + whenAny, } from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; describe("External event name routing", () => { let backend: InMemoryOrchestrationBackend; @@ -178,4 +180,224 @@ describe("External event name routing", () => { expect(backend.getInstance(id)?.history.filter((event) => event.hasEventraised())).toHaveLength(3); }, ); + + describe("continue-as-new trailing events", () => { + async function continueWithBatch( + orchestrator: TOrchestrator, + input: unknown, + events: ReadonlyArray, + ) { + worker.addOrchestrator(orchestrator); + await worker.start(); + const id = await client.scheduleNewOrchestration(orchestrator, input); + await expectRunning(id); + const initialExecutionId = backend.getInstance(id)?.executionId; + expect(initialExecutionId).toBeDefined(); + await worker.stop(); + expect(backend.hasPendingWork()).toBe(false); + + // With no worker polling, these public events form one real backend work item. + for (const [name, payload] of events) { + await client.raiseOrchestrationEvent(id, name, payload); + } + expect( + backend + .getInstance(id) + ?.pendingEvents.map((event) => [ + event.getEventraised()?.getName(), + event.getEventraised()?.getInput()?.getValue(), + ]), + ).toEqual(events.map(([name, payload]) => [name, JSON.stringify(payload)])); + + let carryover: pb.HistoryEvent[] = []; + const continued = backend.waitForState( + id, + (instance) => { + if (instance.executionId === initialExecutionId) { + return false; + } + carryover = instance.pendingEvents.filter((event) => event.hasEventraised()); + return true; + }, + 5000, + ); + worker = new TestOrchestrationWorker(backend); + worker.addOrchestrator(orchestrator); + await worker.start(); + await continued; + await backend.waitForState( + id, + (instance) => + instance.executionId !== initialExecutionId && + instance.status !== pb.OrchestrationStatus.ORCHESTRATION_STATUS_PENDING && + instance.pendingEvents.length === 0, + 5000, + ); + const instance = backend.getInstance(id); + expect( + instance?.history + .find((event) => event.hasExecutionstarted()) + ?.getExecutionstarted() + ?.getOrchestrationinstance() + ?.getExecutionid() + ?.getValue(), + ).toBe(instance?.executionId); + expect(instance?.executionId).not.toBe(initialExecutionId); + expect(backend.hasPendingWork()).toBe(false); + expect((await client.getOrchestrationState(id))?.failureDetails).toBeUndefined(); + return { id, carryover }; + } + + it.each([true, false])("retains the same-batch message with an old listener=%s", async (oldListener) => { + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, iteration: number): any { + if (iteration === 1) { + const rotate = ctx.waitForExternalEvent("rotate"); + if (oldListener) { + const message = ctx.waitForExternalEvent("message"); + yield whenAny([rotate, message]); + } else { + yield rotate; + } + ctx.continueAsNew(2, true); + return; + } + return yield ctx.waitForExternalEvent("message"); + }; + const payload = { kind: "original", value: 42 }; + const { id, carryover } = await continueWithBatch(orchestrator, 1, [ + ["rotate", null], + ["message", payload], + ]); + + expect(carryover.map((event) => event.getEventraised()?.getInput()?.getValue())).toEqual([ + JSON.stringify(payload), + ]); + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + expect(state?.serializedInput).toBe("2"); + expect(state?.serializedOutput).toBe(JSON.stringify(payload)); + expect(backend.getInstance(id)?.history.filter((event) => event.hasEventraised())).toHaveLength(1); + }); + + it("preserves case-insensitive FIFO and falsy trailing payloads exactly once across replay", async () => { + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, iteration: number): any { + if (iteration === 1) { + yield whenAny([ctx.waitForExternalEvent("rotate"), ctx.waitForExternalEvent("MeSsAgE")]); + ctx.continueAsNew(2, true); + return; + } + const values: unknown[] = []; + for (let i = 0; i < 4; i++) { + values.push(yield ctx.waitForExternalEvent("MESSAGE")); + } + ctx.setCustomStatus(values); + values.push(yield ctx.waitForExternalEvent("message")); + return values; + }; + const payloads = [false, 0, "", null]; + const { id, carryover } = await continueWithBatch(orchestrator, 1, [ + ["rotate", null], + ["MeSsAgE", false], + ["MESSAGE", 0], + ["message", ""], + ["mEsSaGe", null], + ]); + expect(carryover.map((event) => event.getEventraised()?.getName())).toEqual(Array(4).fill("message")); + expect(carryover.map((event) => event.getEventraised()?.getInput()?.getValue())).toEqual( + payloads.map((payload) => JSON.stringify(payload)), + ); + const waiting = await client.getOrchestrationState(id); + expect(waiting?.runtimeStatus).toBe(OrchestrationStatus.RUNNING); + expect(waiting?.serializedCustomStatus).toBe(JSON.stringify(payloads)); + const fresh = { kind: "fresh" }; + await raiseAndWaitForCommit(id, "MESSAGE", fresh); + + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toBe(JSON.stringify([...payloads, fresh])); + expect(backend.getInstance(id)?.history.filter((event) => event.hasEventraised())).toHaveLength(5); + }); + + it("delivers a message submitted after the new generation is committed", async () => { + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, iteration: number): any { + if (iteration === 1) { + yield whenAny([ctx.waitForExternalEvent("rotate"), ctx.waitForExternalEvent("message")]); + ctx.continueAsNew(2, true); + return; + } + return yield ctx.waitForExternalEvent("message"); + }; + const { id, carryover } = await continueWithBatch(orchestrator, 1, [["rotate", null]]); + expect(carryover).toHaveLength(0); + expect((await client.getOrchestrationState(id))?.runtimeStatus).toBe(OrchestrationStatus.RUNNING); + const payload = { kind: "original" }; + await raiseAndWaitForCommit(id, "message", payload); + + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toBe(JSON.stringify(payload)); + expect(backend.getInstance(id)?.history.filter((event) => event.hasEventraised())).toHaveLength(1); + }); + + it("does not carry a message actually consumed before rotation", async () => { + const orchestrator: TOrchestrator = async function* ( + ctx: OrchestrationContext, + input: { iteration: number; consumed?: unknown }, + ): any { + if (input.iteration === 1) { + const rotate = ctx.waitForExternalEvent("rotate"); + const message = ctx.waitForExternalEvent("message"); + yield whenAny([rotate, message]); + const consumed = message.getResult(); + yield rotate; + ctx.continueAsNew({ iteration: 2, consumed }, true); + return; + } + const fresh = yield ctx.waitForExternalEvent("message"); + return { consumed: input.consumed, fresh }; + }; + const consumed = { kind: "consumed" }; + const { id, carryover } = await continueWithBatch(orchestrator, { iteration: 1 }, [ + ["message", consumed], + ["rotate", null], + ]); + expect(carryover).toHaveLength(0); + expect((await client.getOrchestrationState(id))?.runtimeStatus).toBe(OrchestrationStatus.RUNNING); + expect(backend.getInstance(id)?.history.filter((event) => event.hasEventraised())).toHaveLength(0); + const fresh = { kind: "fresh" }; + await raiseAndWaitForCommit(id, "message", fresh); + + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toBe(JSON.stringify({ consumed, fresh })); + }); + + it.each([true, false])( + "drops trailing messages with saveEvents=false and an old listener=%s", + async (oldListener) => { + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, iteration: number): any { + if (iteration === 1) { + const rotate = ctx.waitForExternalEvent("rotate"); + yield oldListener ? whenAny([rotate, ctx.waitForExternalEvent("message")]) : rotate; + ctx.continueAsNew(2, false); + return; + } + return yield ctx.waitForExternalEvent("message"); + }; + const { id, carryover } = await continueWithBatch(orchestrator, 1, [ + ["rotate", null], + ["message", { kind: "discarded" }], + ]); + expect(carryover).toHaveLength(0); + expect((await client.getOrchestrationState(id))?.runtimeStatus).toBe(OrchestrationStatus.RUNNING); + expect(backend.getInstance(id)?.history.filter((event) => event.hasEventraised())).toHaveLength(0); + const fresh = { kind: "fresh" }; + await raiseAndWaitForCommit(id, "message", fresh); + + const state = await client.waitForOrchestrationCompletion(id, true, 5); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + expect(state?.serializedOutput).toBe(JSON.stringify(fresh)); + }, + ); + }); }); diff --git a/packages/durabletask-js/test/orchestration_executor.spec.ts b/packages/durabletask-js/test/orchestration_executor.spec.ts index 421238e..ca214c3 100644 --- a/packages/durabletask-js/test/orchestration_executor.spec.ts +++ b/packages/durabletask-js/test/orchestration_executor.spec.ts @@ -1079,6 +1079,66 @@ describe("Orchestration Executor", () => { } }); + it.each([true, false])( + "routes trailing events into the continue-as-new action with saveEvents=%s", + async (saveEvents) => { + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + yield whenAny([ctx.waitForExternalEvent("rotate"), ctx.waitForExternalEvent("message")]); + ctx.continueAsNew(2, saveEvents); + }; + const registry = new Registry(); + const name = registry.addOrchestrator(orchestrator); + const oldEvents = [newOrchestratorStartedEvent(), newExecutionStartedEvent(name, TEST_INSTANCE_ID)]; + const newEvents = [ + newEventRaisedEvent("rotate", "null"), + newEventRaisedEvent("MESSAGE", "false"), + newEventRaisedEvent("message", "0"), + ]; + const result = await new OrchestrationExecutor(registry, testLogger).execute( + TEST_INSTANCE_ID, + oldEvents, + newEvents, + ); + const complete = getAndValidateSingleCompleteOrchestrationAction(result); + expect(complete?.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW); + expect(complete?.getResult()?.getValue()).toBe("2"); + expect( + complete + ?.getCarryovereventsList() + .map((event) => [event.getEventraised()?.getName(), event.getEventraised()?.getInput()?.getValue()]), + ).toEqual( + saveEvents + ? [ + ["message", "false"], + ["message", "0"], + ] + : [], + ); + }, + ); + + it.each([true, false])( + "fails on a malformed trailing event after continue-as-new with saveEvents=%s", + async (saveEvents) => { + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + yield whenAny([ctx.waitForExternalEvent("rotate"), ctx.waitForExternalEvent("message")]); + ctx.continueAsNew(2, saveEvents); + }; + const registry = new Registry(); + const name = registry.addOrchestrator(orchestrator); + const result = await new OrchestrationExecutor(registry, testLogger).execute( + TEST_INSTANCE_ID, + [newOrchestratorStartedEvent(), newExecutionStartedEvent(name, TEST_INSTANCE_ID)], + [newEventRaisedEvent("rotate", "null"), newEventRaisedEvent("message", "not-json")], + ); + const complete = getAndValidateSingleCompleteOrchestrationAction(result); + expect(complete?.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(complete?.getFailuredetails()?.getErrortype()).toBe("Error"); + expect(complete?.getFailuredetails()?.getErrormessage()).toContain("Failed to parse JSON from StringValue"); + expect(complete?.getCarryovereventsList()).toEqual([]); + }, + ); + it("should set the new version on a continue-as-new action", async () => { const orchestrator: TOrchestrator = async (ctx: OrchestrationContext, input: number) => { ctx.continueAsNew(input + 1, false, "2.0.0");