fix: preserve trailing events after continue-as-new

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.

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
 

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.

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

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<readonly [string, unknown]>,
+    ) {
+      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));
+      },
+    );
+  });
 });

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