diff --git a/.changeset/world-local-resume-redelivery-converge.md b/.changeset/world-local-resume-redelivery-converge.md new file mode 100644 index 0000000000..4954e25a43 --- /dev/null +++ b/.changeset/world-local-resume-redelivery-converge.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-local': patch +--- + +Converge a redelivered `hook_received` re-ensure on its already-committed `(runId, resumeId)` claim instead of rejecting with `HookNotFoundError` when the hook was disposed after the resume was recorded. The rejection made the queue consumer ack the redelivery as "nothing left to resume", silently dropping any continuation the message carried. diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index d072eed8f2..d161d1ef8d 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -1026,6 +1026,43 @@ export function createEventsStorage( isHookEventRequiringExistence(data.eventType) && data.correlationId ) { + // Redelivery convergence — checked BEFORE the disposal/existence + // rejections below: if this resume's `(runId, resumeId)` claim is + // already committed AND its pinned event is journaled, return that + // event as success. The claim proves this exact resume was accepted + // while the hook was alive, and the event is already in the log, so + // replay observes it either way. Without this, a queue redelivery + // of the consumer's re-ensure after the workflow disposed the hook + // (dispose → sleep) is rejected with HookNotFound — which the + // consumer treats as "nothing left to resume" and acks, losing + // whatever continuation the message carried. A claim with a + // mismatched hookId or payload digest is NOT converged here; it + // falls through to the full validation below, which rejects it the + // same way it always has. + if (data.eventType === 'hook_received' && params?.resumeId) { + const committedClaim = await readJSON( + hookResumeClaimPath(basedir, effectiveRunId, params.resumeId), + HookResumeClaimSchema + ); + if ( + committedClaim && + committedClaim.hookId === data.correlationId && + (!params.resumePayloadDigest || + !committedClaim.payloadDigest || + committedClaim.payloadDigest === params.resumePayloadDigest) + ) { + const committedEvent = await readJSONWithFallback( + basedir, + 'events', + `${effectiveRunId}-${committedClaim.eventId}`, + EventSchema, + tag + ); + if (committedEvent) { + return { event: committedEvent }; + } + } + } // A resume must never be journaled after the hook's disposal. // The disposer's durable order is: dispose lock → claim/entity // delete → `hook_disposed` append, so the hook entity can still diff --git a/packages/world-local/src/storage/hook-resume-dedup.test.ts b/packages/world-local/src/storage/hook-resume-dedup.test.ts index 867a4acb6a..48f545033b 100644 --- a/packages/world-local/src/storage/hook-resume-dedup.test.ts +++ b/packages/world-local/src/storage/hook-resume-dedup.test.ts @@ -3,7 +3,7 @@ import os from 'node:os'; import path from 'node:path'; import { SPEC_VERSION_CURRENT, type Storage } from '@workflow/world'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { createHook, createRun } from '../test-helpers.js'; +import { createHook, createRun, disposeHook } from '../test-helpers.js'; import { createStorage } from './index.js'; // When a run carries the `hookResumeInputVersion` marker, the parallel resume @@ -118,6 +118,36 @@ describe('world-local hook_received resume dedup', () => { expect(await countHookReceived(runId)).toBe(1); }); + it('converges a committed resume re-ensured AFTER the hook was disposed', async () => { + const { runId, hook } = await setup(); + const payload = new Uint8Array([1, 2, 3]); + + // Delivery 1: the consumer's re-ensure commits the resume while the + // hook is alive. + const first = await resume(runId, hook, 'resume_1', 'digest_1', payload); + + // The workflow receives the payload and disposes the hook (e.g. the + // dispose → sleep pattern), releasing its token. + await disposeHook(storage, runId, hook.hookId); + + // Delivery 2: the SAME message is redelivered (queue retry, or a + // visibility-timeout continuation riding the resume message) and + // re-ensures the same (resumeId, digest). The resume is already + // committed — this must converge on the existing event as success, NOT + // reject with HookNotFound. A rejection makes the consumer ack the + // message as "nothing left to resume", silently dropping whatever + // continuation it carried and wedging the run. + const second = await resume(runId, hook, 'resume_1', 'digest_1', payload); + + expect(second.event.eventId).toBe(first.event.eventId); + expect(await countHookReceived(runId)).toBe(1); + + // A genuinely NEW resume after disposal is still rejected. + await expect( + resume(runId, hook, 'resume_2', 'digest_2', new Uint8Array([9])) + ).rejects.toThrow(); + }); + it('rejects a reused resumeId + digest that belongs to a DIFFERENT hook', async () => { const { runId, hook } = await setup(); const otherHook = await createHook(storage, runId, {