From 1517f78f633ec547d54e923d86289b0a5ae57a6f Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 13 Jun 2026 17:56:14 +0000 Subject: [PATCH 1/3] test: e2e coverage for run-idempotency conflict-handling strategies (#2387) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test: e2e coverage for run-idempotency conflict-handling strategies Covers the patterns documented in foundations/idempotency: - claim-only hook mutex: token claimed and held with no payload data, duplicate identifies the owner, token released after completion - adopt the owner's result via conflict.returnValue - signal the owner: duplicate forwards its payload via resumeHook - supersede: duplicate cancels the owner and reclaims the token - route-side resume-or-start retry pattern reaching the started run Co-Authored-By: Claude Fable 5 * test: fix adopt-owner-result race — gate owner completion on observed conflict On slow runtimes the duplicate's first invocation could land after the owner completed and released the token, making the duplicate a fresh owner that waits forever for a payload (90s timeout across CI matrices). Poll the duplicate's event log for hook_conflict before resuming the owner, and widen the test timeout for the added gate budget. Co-Authored-By: Claude Fable 5 * review: assert superseded owner's returnValue rejection; empty changeset - Await run1.returnValue and assert WorkflowRunCancelledError so the cancellation is verified end-to-end and no rejection leaks from the supersede test. - Test-only PR: use an empty changeset. Co-Authored-By: Claude Fable 5 * ci: retrigger preview deployments (turbopack deployment for 2e9d000 wedged in esbuild hang) Co-Authored-By: Claude Fable 5 * ci: bust poisoned turbo cache entry for nextjs-turbopack build The 2e9d000 deployment's next build crashed in an esbuild hang but its task (70724907c9dd3a29) was recorded into the turbo remote cache anyway, so every subsequent build with the same input hash replays the broken artifact (missing routes-manifest). Change a build input to force a fresh execution. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 Signed-off-by: Pranay Prakash --- .changeset/idempotency-strategy-e2e.md | 2 + packages/core/e2e/e2e.test.ts | 240 ++++++++++++++++++ workbench/example/workflows/99_e2e.ts | 126 +++++++++ .../app/workflows/definitions.ts | 10 + 4 files changed, 378 insertions(+) create mode 100644 .changeset/idempotency-strategy-e2e.md diff --git a/.changeset/idempotency-strategy-e2e.md b/.changeset/idempotency-strategy-e2e.md new file mode 100644 index 0000000000..a845151cc8 --- /dev/null +++ b/.changeset/idempotency-strategy-e2e.md @@ -0,0 +1,2 @@ +--- +--- diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 59f2b504e2..0613f66176 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1620,6 +1620,246 @@ describe('e2e', () => { } ); + test( + 'hookClaimOnlyMutexWorkflow - hook works as a pure run mutex without payload data', + { timeout: 90_000 }, + async () => { + const token = Math.random().toString(36).slice(2); + + // Owner claims the token and holds it during unrelated work, + // never awaiting payload data. + const run1 = await start(await e2e('hookClaimOnlyMutexWorkflow'), [ + token, + 15_000, + ]); + await waitForHook(token, { runId: run1.runId }); + + // A duplicate started while the owner holds the token observes the + // conflict and identifies the owner. + const run2 = await start(await e2e('hookClaimOnlyMutexWorkflow'), [ + token, + 15_000, + ]); + const run2Result = await run2.returnValue; + expect(run2Result).toEqual({ + role: 'duplicate', + conflictRunId: run1.runId, + }); + + // The owner completes without ever receiving a payload. + const run1Result = await run1.returnValue; + expect(run1Result).toMatchObject({ role: 'owner' }); + + const world = await getWorld(); + const { data: events } = await world.events.list({ runId: run1.runId }); + expect( + events.some((e) => e.eventType === 'hook_received'), + 'claim-only owner should never receive hook payload data' + ).toBe(false); + + // Completion releases the token: a later run claims it cleanly. + const waitForTokenRelease = async () => { + const timeoutAt = Date.now() + 20_000; + while (Date.now() < timeoutAt) { + try { + await getHookByToken(token); + } catch (error) { + if (HookNotFoundError.is(error)) { + return; + } + throw error; + } + await sleep(1_000); + } + throw new Error(`Timed out waiting for token ${token} to be released`); + }; + await waitForTokenRelease(); + + const run3 = await start(await e2e('hookClaimOnlyMutexWorkflow'), [ + token, + 100, + ]); + const run3Result = await run3.returnValue; + expect(run3Result).toMatchObject({ role: 'owner' }); + } + ); + + test( + 'hookAdoptOwnerResultWorkflow - duplicate adopts the owner result via conflict.returnValue', + { timeout: 120_000 }, + async () => { + const token = Math.random().toString(36).slice(2); + + const run1 = await start(await e2e('hookAdoptOwnerResultWorkflow'), [ + token, + 'owner-marker', + ]); + await waitForHook(token, { runId: run1.runId }); + + // The duplicate suspends on the owner's returnValue. + const run2 = await start(await e2e('hookAdoptOwnerResultWorkflow'), [ + token, + 'duplicate-marker', + ]); + + // Wait until run2 has actually observed the conflict before letting + // the owner complete. On slow runtimes run2's first invocation can + // otherwise land after the owner finished and released the token, + // turning run2 into a fresh owner that waits forever for a payload. + const world = await getWorld(); + const conflictDeadline = Date.now() + 60_000; + let sawConflict = false; + while (Date.now() < conflictDeadline) { + const { data: run2Events } = await world.events.list({ + runId: run2.runId, + }); + if (run2Events.some((e) => e.eventType === 'hook_conflict')) { + sawConflict = true; + break; + } + await sleep(500); + } + expect( + sawConflict, + 'run2 should observe the hook conflict while the owner is active' + ).toBe(true); + + // Complete the owner. + await resumeHook(token, { value: 'adopted-value' }); + + const run1Result = await run1.returnValue; + expect(run1Result).toEqual({ + role: 'owner', + marker: 'owner-marker', + value: 'adopted-value', + }); + + // The duplicate returns the owner's exact result, so callers cannot + // tell which run did the work. + const run2Result = await run2.returnValue; + expect(run2Result).toEqual({ + role: 'duplicate', + conflictRunId: run1.runId, + adopted: run1Result, + }); + } + ); + + test( + 'hookSignalOwnerWorkflow - duplicate forwards its payload to the owner via resumeHook', + { timeout: 90_000 }, + async () => { + const token = Math.random().toString(36).slice(2); + + const run1 = await start(await e2e('hookSignalOwnerWorkflow'), [ + token, + 'owner-input', + ]); + await waitForHook(token, { runId: run1.runId }); + + // The duplicate forwards its own input into the owner's hook + // instead of doing the work itself. + const run2 = await start(await e2e('hookSignalOwnerWorkflow'), [ + token, + 'forwarded-from-duplicate', + ]); + + const run2Result = await run2.returnValue; + expect(run2Result).toEqual({ + role: 'duplicate', + forwardedTo: run1.runId, + }); + + // The owner receives the duplicate's payload and completes with it. + const run1Result = await run1.returnValue; + expect(run1Result).toEqual({ + role: 'owner', + received: 'forwarded-from-duplicate', + }); + } + ); + + test( + 'hookSupersedeOwnerWorkflow - duplicate cancels the owner and claims the released token', + { timeout: 90_000 }, + async () => { + const token = Math.random().toString(36).slice(2); + + // The first run claims the token and waits for a payload. + const run1 = await start(await e2e('hookSupersedeOwnerWorkflow'), [ + token, + ]); + await waitForHook(token, { runId: run1.runId }); + + // Newest-wins: the second run cancels the owner and claims the token. + const run2 = await start(await e2e('hookSupersedeOwnerWorkflow'), [ + token, + ]); + await waitForHook(token, { runId: run2.runId, timeoutMs: 60_000 }); + + // The superseded owner ends up cancelled — assert via both the + // rejected returnValue (also prevents an unhandled rejection from + // leaking out of this test) and the inspected run status. + const run1Error = await run1.returnValue.catch((e: unknown) => e); + expect(WorkflowRunCancelledError.is(run1Error)).toBe(true); + const { json: run1Data } = await cliInspectJson(`runs ${run1.runId}`); + expect(run1Data.status).toBe('cancelled'); + + // The new owner receives payloads on the reclaimed token. + await resumeHook(token, { message: 'post-supersede' }); + const run2Result = await run2.returnValue; + expect(run2Result).toMatchObject({ + role: 'owner', + received: 'post-supersede', + }); + } + ); + + test( + 'resume-or-start route pattern - resumeHook retried after start() reaches the new run', + { timeout: 90_000 }, + async () => { + const token = Math.random().toString(36).slice(2); + + // No active run yet: the resume half fails with HookNotFoundError. + await expect( + resumeHook(token, { message: 'too-early' }) + ).rejects.toSatisfy((e: unknown) => HookNotFoundError.is(e)); + + // Start the workflow, then retry the resume until the run registers + // its deterministic hook — the documented route-side pattern. + const run = await start(await e2e('hookSignalOwnerWorkflow'), [ + token, + 'unused-owner-input', + ]); + + const deadline = Date.now() + 30_000; + let resumed: Awaited> | undefined; + while (Date.now() < deadline) { + try { + resumed = await resumeHook(token, { message: 'delivered' }); + break; + } catch (error) { + if (!HookNotFoundError.is(error)) throw error; + await sleep(250); + } + } + expect( + resumed, + 'resume retry should reach the started run' + ).toBeDefined(); + + // The resume reached the run this request started (no concurrent + // racer in this test), and the payload was not dropped. + expect(resumed?.runId).toBe(run.runId); + const result = await run.returnValue; + expect(result).toEqual({ + role: 'owner', + received: 'delivered', + }); + } + ); + test( 'hookDisposeTestWorkflow - hook token reuse after explicit disposal while workflow still running', { diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index ff6a0f2cc0..df287c2e3f 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -682,6 +682,132 @@ export async function hookGetConflictThenStepParallelWorkflow( }; } +////////////////////////////////////////////////////////// +// Run idempotency / conflict-handling strategy workflows. +// These mirror the patterns documented in +// docs/content/docs/*/foundations/idempotency.mdx. +////////////////////////////////////////////////////////// + +/** + * Claim-only run mutex: the hook is used purely for run idempotency — + * the workflow claims the token, holds it while doing unrelated work, + * and never awaits hook payload data. Duplicates started while the + * owner holds the token observe the conflict and return early. + */ +export async function hookClaimOnlyMutexWorkflow( + token: string, + holdMs: number +) { + 'use workflow'; + + using hook = createHook({ token }); + + const conflict = await hook.getConflict(); + if (conflict) { + return { + role: 'duplicate' as const, + conflictRunId: conflict.runId, + }; + } + + // Hold the token for the duration of the work without ever awaiting + // hook payload data. + const work = await hookGetConflictTimedStep('A', holdMs); + + return { + role: 'owner' as const, + workEndedAt: work.endedAt, + }; +} + +/** + * "Adopt the owner's result" strategy: the duplicate run waits for the + * active owner to finish and returns the owner's result, so callers + * cannot tell which run did the work. + */ +export async function hookAdoptOwnerResultWorkflow( + token: string, + marker: string +) { + 'use workflow'; + + using hook = createHook<{ value: string }>({ token }); + + const conflict = await hook.getConflict(); + if (conflict) { + const adopted = await conflict.returnValue; + return { + role: 'duplicate' as const, + conflictRunId: conflict.runId, + adopted, + }; + } + + const payload = await hook; + return { + role: 'owner' as const, + marker, + value: payload.value, + }; +} + +async function forwardPayloadToOwner(token: string, message: string) { + 'use step'; + await resumeHook(token, { message }); +} + +/** + * "Signal the owner" strategy: the duplicate run forwards its input to + * the active owner's hook from a step instead of doing the work itself. + */ +export async function hookSignalOwnerWorkflow(token: string, message: string) { + 'use workflow'; + + using hook = createHook<{ message: string }>({ token }); + + const conflict = await hook.getConflict(); + if (conflict) { + await forwardPayloadToOwner(token, message); + return { + role: 'duplicate' as const, + forwardedTo: conflict.runId, + }; + } + + const payload = await hook; + return { + role: 'owner' as const, + received: payload.message, + }; +} + +/** + * "Supersede the owner" strategy (newest-wins): cancel the active owner + * and claim the released token. Cancellation disposes the owner's hooks; + * the retry loop covers the window where disposal has not propagated. + */ +export async function hookSupersedeOwnerWorkflow(token: string) { + 'use workflow'; + + for (let attempt = 0; attempt < 5; attempt++) { + using hook = createHook<{ message: string }>({ token }); + + const conflict = await hook.getConflict(); + if (!conflict) { + const payload = await hook; + return { + role: 'owner' as const, + attempt, + received: payload.message, + }; + } + + await conflict.cancel(); + } + + throw new Error(`Could not claim ${token} after cancelling the owner`); +} + ////////////////////////////////////////////////////////// /** diff --git a/workbench/nextjs-turbopack/app/workflows/definitions.ts b/workbench/nextjs-turbopack/app/workflows/definitions.ts index dcf552aa38..170a028c33 100644 --- a/workbench/nextjs-turbopack/app/workflows/definitions.ts +++ b/workbench/nextjs-turbopack/app/workflows/definitions.ts @@ -1,5 +1,8 @@ import 'server-only'; +// NOTE: comment edit busts the turbo task hash 70724907c9dd3a29, whose +// remote-cache entry was poisoned by an esbuild hang during the build of +// dpl_6UXttzGpjxbrFcpRQZGR6E9v3G2Q (missing .next/routes-manifest.json). import { allWorkflows } from '@/_workflows'; import type { WorkflowDefinition } from './types'; @@ -43,6 +46,13 @@ const DEFAULT_ARGS_MAP: Record = { RANDOM_ARG_PLACEHOLDER, RANDOM_ARG_PLACEHOLDER, ], + hookClaimOnlyMutexWorkflow: [RANDOM_ARG_PLACEHOLDER, 1000], + hookAdoptOwnerResultWorkflow: [ + RANDOM_ARG_PLACEHOLDER, + RANDOM_ARG_PLACEHOLDER, + ], + hookSignalOwnerWorkflow: [RANDOM_ARG_PLACEHOLDER, RANDOM_ARG_PLACEHOLDER], + hookSupersedeOwnerWorkflow: [RANDOM_ARG_PLACEHOLDER], closureVariableWorkflow: [7], // 100_durable_agent_e2e.ts agentBasicE2e: ['hello world'], From 7135471294f0d23b931b89870d89f03f393f3c9f Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Sat, 13 Jun 2026 14:36:06 -0700 Subject: [PATCH 2/3] fix(backport): adapt conflict-handling tests to stable's getConflict/getRun API The backport of #2387 used APIs that only exist on `main`, breaking the nextjs-turbopack/webpack builds and the e2e suite on `stable`: - `hookAdoptOwnerResultWorkflow`/`hookSupersedeOwnerWorkflow` read `conflict.returnValue`/`conflict.cancel()`, but on `stable` `getConflict()` resolves with `{ runId }`. Resolve the owning run via `getRun(conflict.runId)` inside a step (the documented stable pattern) to await its result / cancel it. - Import `resumeHook` from `workflow/api` in 99_e2e.ts (was used by `forwardPayloadToOwner` but never imported). - Convert the backported `waitForHook(token, { runId })` call sites to `waitForHookState(token, predicate)`; `waitForHook` does not exist on `stable` (#2405 standardized on `waitForHookState`). Both workbench builds and `biome check` pass locally. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/core/e2e/e2e.test.ts | 26 +++++++++++++++++++++----- workbench/example/workflows/99_e2e.ts | 26 +++++++++++++++++++++++--- 2 files changed, 44 insertions(+), 8 deletions(-) diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index d748d68d05..c2bf8b788a 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1636,7 +1636,10 @@ describe('e2e', () => { token, 15_000, ]); - await waitForHook(token, { runId: run1.runId }); + await waitForHookState( + token, + (candidate) => candidate?.runId === run1.runId + ); // A duplicate started while the owner holds the token observes the // conflict and identifies the owner. @@ -1698,7 +1701,10 @@ describe('e2e', () => { token, 'owner-marker', ]); - await waitForHook(token, { runId: run1.runId }); + await waitForHookState( + token, + (candidate) => candidate?.runId === run1.runId + ); // The duplicate suspends on the owner's returnValue. const run2 = await start(await e2e('hookAdoptOwnerResultWorkflow'), [ @@ -1759,7 +1765,10 @@ describe('e2e', () => { token, 'owner-input', ]); - await waitForHook(token, { runId: run1.runId }); + await waitForHookState( + token, + (candidate) => candidate?.runId === run1.runId + ); // The duplicate forwards its own input into the owner's hook // instead of doing the work itself. @@ -1793,13 +1802,20 @@ describe('e2e', () => { const run1 = await start(await e2e('hookSupersedeOwnerWorkflow'), [ token, ]); - await waitForHook(token, { runId: run1.runId }); + await waitForHookState( + token, + (candidate) => candidate?.runId === run1.runId + ); // Newest-wins: the second run cancels the owner and claims the token. const run2 = await start(await e2e('hookSupersedeOwnerWorkflow'), [ token, ]); - await waitForHook(token, { runId: run2.runId, timeoutMs: 60_000 }); + await waitForHookState( + token, + (candidate) => candidate?.runId === run2.runId, + 60_000 + ); // The superseded owner ends up cancelled — assert via both the // rejected returnValue (also prevents an unhandled rejection from diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index df287c2e3f..7c182c3c19 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -13,7 +13,7 @@ import { RetryableError, sleep, } from 'workflow'; -import { getRun, start } from 'workflow/api'; +import { getRun, resumeHook, start } from 'workflow/api'; import { importedStepOnly } from './_imported_step_only'; import { callThrower, stepThatThrowsFromHelper } from './helpers'; @@ -720,6 +720,16 @@ export async function hookClaimOnlyMutexWorkflow( }; } +/** + * Awaits the active owner's result via `getRun()` inside a step. On this + * channel `getConflict()` resolves with `{ runId }`, so the duplicate + * resolves the owning run by id to read its return value. + */ +async function adoptOwnerResult(runId: string) { + 'use step'; + return await getRun(runId).returnValue; +} + /** * "Adopt the owner's result" strategy: the duplicate run waits for the * active owner to finish and returns the owner's result, so callers @@ -735,7 +745,7 @@ export async function hookAdoptOwnerResultWorkflow( const conflict = await hook.getConflict(); if (conflict) { - const adopted = await conflict.returnValue; + const adopted = await adoptOwnerResult(conflict.runId); return { role: 'duplicate' as const, conflictRunId: conflict.runId, @@ -781,6 +791,16 @@ export async function hookSignalOwnerWorkflow(token: string, message: string) { }; } +/** + * Cancels the active owner via `getRun()` inside a step. `getConflict()` + * resolves with `{ runId }`, so the superseding run resolves the owner by + * id to cancel it before reclaiming the token. + */ +async function cancelOwner(runId: string) { + 'use step'; + await getRun(runId).cancel(); +} + /** * "Supersede the owner" strategy (newest-wins): cancel the active owner * and claim the released token. Cancellation disposes the owner's hooks; @@ -802,7 +822,7 @@ export async function hookSupersedeOwnerWorkflow(token: string) { }; } - await conflict.cancel(); + await cancelOwner(conflict.runId); } throw new Error(`Could not claim ${token} after cancelling the owner`); From 92d02ee6e0bd821848bca14af26db400bbfad1c2 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Sat, 13 Jun 2026 14:49:08 -0700 Subject: [PATCH 3/3] fix(backport): import HookNotFoundError in e2e test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backported conflict tests call `HookNotFoundError.is()` in `hookClaimOnlyMutexWorkflow` (token-release wait) and the resume-or-start route test, but the import was never carried into the stable test file — causing a runtime `ReferenceError: HookNotFoundError is not defined`. Import it from `@workflow/errors` (matches `main`). Verified locally against nextjs-turbopack: the two previously-failing tests plus the adopt/signal/supersede rewrites all pass (5/5). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/core/e2e/e2e.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index c2bf8b788a..79277a4695 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -2,6 +2,7 @@ import fs from 'node:fs'; import path from 'node:path'; import { setTimeout as sleep } from 'node:timers/promises'; import { + HookNotFoundError, WorkflowRunCancelledError, WorkflowRunFailedError, WorkflowWorldError,