From 384c73c1b3209b22518e26c5b07e8953a9c36ae1 Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Mon, 24 Aug 2026 23:53:23 +0800 Subject: [PATCH 1/2] fix(workflows): preserve pending result deliveries --- extensions/workflows/result-delivery.test.ts | 136 ++++++++++++++++ extensions/workflows/result-delivery.ts | 157 +++++++++++++------ 2 files changed, 242 insertions(+), 51 deletions(-) diff --git a/extensions/workflows/result-delivery.test.ts b/extensions/workflows/result-delivery.test.ts index d1c2d45f..67283278 100644 --- a/extensions/workflows/result-delivery.test.ts +++ b/extensions/workflows/result-delivery.test.ts @@ -148,3 +148,139 @@ 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((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((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 held-inline restore persistence failure does not block sibling restoration", () => { + const first = details("wf_restore_a"); + const second = details("wf_restore_b"); + first.delivery!.state = "held-for-inline"; + second.delivery!.state = "pending"; + const delivery = createWorkflowResultDelivery({ + isIdle: () => false, + persist: (current) => { + if (current.runId === first.runId) throw new Error("disk unavailable"); + }, + deliver: async () => [], + }); + + 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); +}); diff --git a/extensions/workflows/result-delivery.ts b/extensions/workflows/result-delivery.ts index f8613999..775fa86e 100644 --- a/extensions/workflows/result-delivery.ts +++ b/extensions/workflows/result-delivery.ts @@ -38,6 +38,8 @@ export function createWorkflowResultDelivery( ) { const pending = new Map(); let flushing: Promise | 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>, + 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; }, From b8a397369c972ffcc73860228fbd529fa7640bfa Mon Sep 17 00:00:00 2001 From: tt-a1i Date: Mon, 24 Aug 2026 23:59:58 +0800 Subject: [PATCH 2/2] test(workflows): cover delivery recovery branches --- extensions/workflows/result-delivery.test.ts | 64 ++++++++++++++++++-- 1 file changed, 60 insertions(+), 4 deletions(-) diff --git a/extensions/workflows/result-delivery.test.ts b/extensions/workflows/result-delivery.test.ts index 67283278..36a2da02 100644 --- a/extensions/workflows/result-delivery.test.ts +++ b/extensions/workflows/result-delivery.test.ts @@ -251,17 +251,67 @@ test("a persistence exception cannot drop later envelopes from a failed batch", assert.equal(second.delivery?.state, "delivered"); }); -test("a held-inline restore persistence failure does not block sibling restoration", () => { +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: () => false, + isIdle: () => true, persist: (current) => { - if (current.runId === first.runId) throw new Error("disk unavailable"); + if (failPersistence && current.runId === first.runId) { + throw new Error("disk unavailable"); + } }, - deliver: async () => [], + deliver: async (envelopes) => + envelopes.map((entry) => ({ + deliveryId: entry.deliveryId, + delivered: true, + })), }); assert.equal( @@ -283,4 +333,10 @@ test("a held-inline restore persistence failure does not block sibling restorati 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"); });