fix(workflows): preserve pending result deliveries

extensions/workflows/result-delivery.test.ts

@@ -148,3 +148,195 @@ test("a receipt persistence failure retains the same delivery for at-least-once
   assert.equal(run.delivery?.id, "workflow:wf_receipt:terminal");
   assert.match(run.delivery?.lastError ?? "", /receipt persistence failed/i);
 });
+
+test("a completion queued during a failed flush is drained without another lifecycle event", async () => {
+  const first = details("wf_during_flush_a");
+  const second = details("wf_during_flush_b");
+  let idle = false;
+  let releaseFirstAttempt!: () => void;
+  let markFirstAttemptStarted!: () => void;
+  const firstAttemptStarted = new Promise<void>((resolve) => {
+    markFirstAttemptStarted = resolve;
+  });
+  const calls: string[][] = [];
+  const delivery = createWorkflowResultDelivery({
+    isIdle: () => idle,
+    persist: () => {},
+    deliver: async (envelopes) => {
+      calls.push(envelopes.map((entry) => entry.deliveryId));
+      if (calls.length === 1) {
+        markFirstAttemptStarted();
+        await new Promise<void>((resolve) => {
+          releaseFirstAttempt = resolve;
+        });
+        throw new Error("session unavailable");
+      }
+      return envelopes.map((entry) => ({
+        deliveryId: entry.deliveryId,
+        delivered: true,
+      }));
+    },
+  });
+
+  delivery.defer({
+    deliveryId: first.delivery!.id,
+    runId: first.runId,
+    details: first,
+  });
+  idle = true;
+  const flushing = delivery.parentSettled();
+  await firstAttemptStarted;
+  delivery.defer({
+    deliveryId: second.delivery!.id,
+    runId: second.runId,
+    details: second,
+  });
+  releaseFirstAttempt();
+  await flushing;
+
+  assert.equal(delivery.size(), 0);
+  assert.equal(first.delivery?.state, "delivered");
+  assert.equal(second.delivery?.state, "delivered");
+  assert.deepEqual(calls, [
+    ["workflow:wf_during_flush_a:terminal"],
+    [
+      "workflow:wf_during_flush_b:terminal",
+      "workflow:wf_during_flush_a:terminal",
+    ],
+  ]);
+});
+
+test("a persistence exception cannot drop later envelopes from a failed batch", async () => {
+  const first = details("wf_persist_a");
+  const second = details("wf_persist_b");
+  let failPersistence = false;
+  let failTransport = true;
+  const persistedAfterFailure: string[] = [];
+  const delivery = createWorkflowResultDelivery({
+    isIdle: () => false,
+    persist: (current) => {
+      if (!failPersistence) return;
+      persistedAfterFailure.push(current.runId);
+      if (current.runId === first.runId) throw new Error("disk unavailable");
+    },
+    deliver: async (envelopes) => {
+      if (failTransport) throw new Error("session unavailable");
+      return envelopes.map((entry) => ({
+        deliveryId: entry.deliveryId,
+        delivered: true,
+      }));
+    },
+  });
+
+  for (const run of [first, second]) {
+    delivery.defer({
+      deliveryId: run.delivery!.id,
+      runId: run.runId,
+      details: run,
+    });
+  }
+  failPersistence = true;
+  await delivery.parentSettled();
+
+  assert.equal(delivery.size(), 2);
+  assert.deepEqual(persistedAfterFailure, [first.runId, second.runId]);
+  assert.equal(first.delivery?.state, "pending");
+  assert.equal(second.delivery?.state, "pending");
+
+  failPersistence = false;
+  failTransport = false;
+  await delivery.parentSettled();
+  assert.equal(delivery.size(), 0);
+  assert.equal(first.delivery?.state, "delivered");
+  assert.equal(second.delivery?.state, "delivered");
+});
+
+test("a persistence exception cannot drop later unacknowledged receipts", async () => {
+  const first = details("wf_unack_a");
+  const second = details("wf_unack_b");
+  let failPersistence = false;
+  let acknowledge = false;
+  const persistedAfterFailure: string[] = [];
+  const delivery = createWorkflowResultDelivery({
+    isIdle: () => false,
+    persist: (current) => {
+      if (!failPersistence) return;
+      persistedAfterFailure.push(current.runId);
+      if (current.runId === first.runId) throw new Error("disk unavailable");
+    },
+    deliver: async (envelopes) =>
+      envelopes.map((entry) => ({
+        deliveryId: entry.deliveryId,
+        delivered: acknowledge,
+      })),
+  });
+
+  for (const run of [first, second]) {
+    delivery.defer({
+      deliveryId: run.delivery!.id,
+      runId: run.runId,
+      details: run,
+    });
+  }
+  failPersistence = true;
+  await delivery.parentSettled();
+
+  assert.equal(delivery.size(), 2);
+  assert.deepEqual(persistedAfterFailure, [first.runId, second.runId]);
+  assert.equal(first.delivery?.state, "pending");
+  assert.equal(second.delivery?.state, "pending");
+
+  failPersistence = false;
+  acknowledge = true;
+  await delivery.parentSettled();
+  assert.equal(delivery.size(), 0);
+  assert.equal(first.delivery?.state, "delivered");
+  assert.equal(second.delivery?.state, "delivered");
+});
+
+test("a held-inline restore persistence failure does not block the final idle flush", async () => {
+  const first = details("wf_restore_a");
+  const second = details("wf_restore_b");
+  first.delivery!.state = "held-for-inline";
+  second.delivery!.state = "pending";
+  let failPersistence = true;
+  const delivery = createWorkflowResultDelivery({
+    isIdle: () => true,
+    persist: (current) => {
+      if (failPersistence && current.runId === first.runId) {
+        throw new Error("disk unavailable");
+      }
+    },
+    deliver: async (envelopes) =>
+      envelopes.map((entry) => ({
+        deliveryId: entry.deliveryId,
+        delivered: true,
+      })),
+  });
+
+  assert.equal(
+    delivery.restore({
+      deliveryId: first.delivery!.id,
+      runId: first.runId,
+      details: first,
+    }),
+    true,
+  );
+  assert.equal(
+    delivery.restore({
+      deliveryId: second.delivery!.id,
+      runId: second.runId,
+      details: second,
+    }),
+    true,
+  );
+  assert.equal(delivery.size(), 2);
+  assert.equal(first.delivery?.state, "pending");
+  assert.match(first.delivery?.lastError ?? "", /persistence failed/i);
+
+  failPersistence = false;
+  await delivery.flushIfIdle();
+  assert.equal(delivery.size(), 0);
+  assert.equal(first.delivery?.state, "delivered");
+  assert.equal(second.delivery?.state, "delivered");
+});

extensions/workflows/result-delivery.ts

@@ -38,6 +38,8 @@ export function createWorkflowResultDelivery(
 ) {
   const pending = new Map<string, WorkflowCompletionEnvelope>();
   let flushing: Promise<void> | undefined;
+  let flushRequested = false;
+  let wakeRequested = false;
 
   const persistState = (
     details: WorkflowDetails,
@@ -61,63 +63,113 @@ export function createWorkflowResultDelivery(
     pending.set(envelope.deliveryId, envelope);
   };
 
+  const retainPending = (
+    envelope: WorkflowCompletionEnvelope,
+    patch: Partial<NonNullable<WorkflowDetails["delivery"]>>,
+    persistenceFailure: string,
+  ) => {
+    // Memory owns the retry before persistence is attempted. A broken disk
+    // must not make this envelope, or any sibling after it, disappear from the
+    // current process.
+    enqueue(envelope);
+    try {
+      persistState(envelope.details, "pending", patch);
+    } catch (error) {
+      const delivery = envelope.details.delivery;
+      if (delivery) {
+        envelope.details.delivery = {
+          ...delivery,
+          state: "pending",
+          updatedAt: Date.now(),
+          lastError: `${persistenceFailure}: ${errorText(error)}`,
+        };
+      }
+    }
+  };
+
   const flush = async (wake: boolean) => {
-    if (flushing) return flushing;
+    if (flushing) {
+      flushRequested = true;
+      wakeRequested ||= wake;
+      return flushing;
+    }
     if (pending.size === 0) return;
-    const envelopes = [...pending.values()];
-    for (const envelope of envelopes) pending.delete(envelope.deliveryId);
 
     flushing = (async () => {
-      let receipts: readonly WorkflowDeliveryReceipt[];
-      try {
-        receipts = await options.deliver(envelopes, wake);
-      } catch (error) {
-        const message = errorText(error);
+      let passWake = wake;
+      while (pending.size > 0) {
+        flushRequested = false;
+        wakeRequested = false;
+        const envelopes = [...pending.values()];
         for (const envelope of envelopes) {
-          enqueue(envelope);
-          persistState(envelope.details, "pending", {
-            attempts: (envelope.details.delivery?.attempts ?? 0) + 1,
-            lastError: message,
-          });
+          pending.delete(envelope.deliveryId);
+        }
+
+        let receipts: readonly WorkflowDeliveryReceipt[] | undefined;
+        try {
+          receipts = await options.deliver(envelopes, passWake);
+        } catch (error) {
+          const message = errorText(error);
+          for (const envelope of envelopes) {
+            retainPending(
+              envelope,
+              {
+                attempts: (envelope.details.delivery?.attempts ?? 0) + 1,
+                lastError: message,
+              },
+              "Pending delivery persistence failed",
+            );
+          }
         }
-        return;
-      }
 
-      const byId = new Map(
-        receipts.map((receipt) => [receipt.deliveryId, receipt] as const),
-      );
-      for (const envelope of envelopes) {
-        const receipt = byId.get(envelope.deliveryId);
-        if (receipt?.delivered) {
-          try {
-            persistState(envelope.details, "delivered", {
-              attempts: (envelope.details.delivery?.attempts ?? 0) + 1,
-              deliveredAt: Date.now(),
-              lastError: undefined,
-            });
-          } catch (error) {
-            // The transport already accepted the message, but the durable
-            // receipt did not commit. Retain it for at-least-once recovery;
-            // the visible stable id lets the parent recognize a rare replay.
-            enqueue(envelope);
-            const delivery = envelope.details.delivery;
-            if (delivery) {
-              envelope.details.delivery = {
-                ...delivery,
-                state: "pending",
-                updatedAt: Date.now(),
-                lastError: `Delivery receipt persistence failed: ${errorText(error)}`,
-              };
+        if (receipts !== undefined) {
+          const byId = new Map(
+            receipts.map((receipt) => [receipt.deliveryId, receipt] as const),
+          );
+          for (const envelope of envelopes) {
+            const receipt = byId.get(envelope.deliveryId);
+            if (receipt?.delivered) {
+              try {
+                persistState(envelope.details, "delivered", {
+                  attempts: (envelope.details.delivery?.attempts ?? 0) + 1,
+                  deliveredAt: Date.now(),
+                  lastError: undefined,
+                });
+              } catch (error) {
+                // The transport already accepted the message, but the durable
+                // receipt did not commit. Retain it for at-least-once recovery;
+                // the visible stable id lets the parent recognize a rare replay.
+                enqueue(envelope);
+                const delivery = envelope.details.delivery;
+                if (delivery) {
+                  envelope.details.delivery = {
+                    ...delivery,
+                    state: "pending",
+                    updatedAt: Date.now(),
+                    lastError: `Delivery receipt persistence failed: ${errorText(error)}`,
+                  };
+                }
+              }
+              continue;
             }
+            retainPending(
+              envelope,
+              {
+                attempts: (envelope.details.delivery?.attempts ?? 0) + 1,
+                lastError:
+                  receipt?.error ?? "Completion delivery was not acknowledged",
+              },
+              "Pending delivery persistence failed",
+            );
           }
-          continue;
         }
-        enqueue(envelope);
-        persistState(envelope.details, "pending", {
-          attempts: (envelope.details.delivery?.attempts ?? 0) + 1,
-          lastError:
-            receipt?.error ?? "Completion delivery was not acknowledged",
-        });
+
+        // A completion or lifecycle edge joined this pass while transport was
+        // in flight. Drain once more so that request is not lost. Failures
+        // retained above do not request their own immediate retry, preventing
+        // an unavailable transport from creating a busy loop.
+        if (!flushRequested) break;
+        passWake ||= wakeRequested;
       }
     })().finally(() => {
       flushing = undefined;
@@ -161,11 +213,14 @@ export function createWorkflowResultDelivery(
       // A process restart cannot still own the inline waiter. Deterministically
       // reconstruct pending delivery from the terminal artifact.
       if (state === "held-for-inline") {
-        persistState(envelope.details, "pending", {
-          lastError: "Inline waiter was not active after session restart",
-        });
+        retainPending(
+          envelope,
+          { lastError: "Inline waiter was not active after session restart" },
+          "Restored delivery state persistence failed",
+        );
+      } else {
+        enqueue(envelope);
       }
-      enqueue(envelope);
       return true;
     },