diff --git a/.changeset/precondition-guard-default-on.md b/.changeset/precondition-guard-default-on.md new file mode 100644 index 0000000000..c029a12c9d --- /dev/null +++ b/.changeset/precondition-guard-default-on.md @@ -0,0 +1,8 @@ +--- +"workflow": minor +"@workflow/core": minor +"@workflow/world": patch +"@workflow/errors": patch +--- + +The `WORKFLOW_PRECONDITION_GUARD` event-creation guard is now on by default; opt out with `WORKFLOW_PRECONDITION_GUARD=0`. diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 581be18c07..7a31d0c401 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -40,10 +40,10 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL ### `WORKFLOW_PRECONDITION_GUARD` -- Default: disabled -- Set `1` to enable an optimistic-concurrency guard for event creation: replay-context event creations send a `stateUpdatedAt` snapshot timestamp, and a backend that supports the guard rejects a creation with 412 ([`PreconditionFailedError`](/docs/api-reference/workflow-errors/precondition-failed-error)) when a newer out-of-band event (a received hook or a completed step) was recorded after that snapshot. +- Default: enabled +- An optimistic-concurrency guard for event creation: replay-context event creations send a `stateUpdatedAt` snapshot timestamp, and a backend that supports the guard rejects a creation with 412 ([`PreconditionFailedError`](/docs/api-reference/workflow-errors/precondition-failed-error)) when a newer out-of-band event (a received hook or a completed step) was recorded after that snapshot. - On rejection the runtime reloads the event log and retries, falling back to a queue re-invocation with a fresh replay if it cannot catch up. -- Backends that do not support the guard ignore the snapshot. +- Set `0` to disable. Backends that do not support the guard ignore the snapshot. ## Inline execution diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index b402b7a952..da295e773c 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -2446,6 +2446,17 @@ export function workflowEntrypoint( // Serialize the original thrown value so its full // type identity and custom properties round-trip // through the event log. + // + // Precondition-guard asymmetry: unlike `run_completed`, + // this terminal `run_failed` sends no `stateUpdatedAt` + // snapshot, so it is never 412-rejected even if a hook + // landed mid-replay and could have changed the path that + // threw. This is intentional and fail-open: a spurious + // failure is recoverable (the run can be re-run from the + // dashboard), whereas a spurious *completion* commits a + // wrong result. Guarding this write symmetrically would + // also need the loaded event log, which is scoped to the + // replay `try` above and not available in this catch. try { // Turbo: order the terminal write after the // backgrounded run_started so the run exists. diff --git a/packages/core/src/runtime/helpers.test.ts b/packages/core/src/runtime/helpers.test.ts index 966f03840f..faad891b41 100644 --- a/packages/core/src/runtime/helpers.test.ts +++ b/packages/core/src/runtime/helpers.test.ts @@ -510,8 +510,8 @@ describe('withPreconditionRetry', () => { } }); - it('passes no snapshot to op when the guard is not opted in', async () => { - delete process.env.WORKFLOW_PRECONDITION_GUARD; + it('passes no snapshot to op when the guard is explicitly disabled', async () => { + process.env.WORKFLOW_PRECONDITION_GUARD = '0'; const log: MutableEventLog = { events: [makeUlidEvent(1_700_000_000_000)], cursor: 'c0', @@ -528,6 +528,24 @@ describe('withPreconditionRetry', () => { expect(eventsListMock).not.toHaveBeenCalled(); }); + it('sends a snapshot by default when the guard variable is unset (on by default)', async () => { + delete process.env.WORKFLOW_PRECONDITION_GUARD; + const time = 1_700_000_000_000; + const log: MutableEventLog = { + events: [makeUlidEvent(time)], + cursor: 'c0', + }; + const op = vi.fn(async (stateUpdatedAt?: number) => { + expect(stateUpdatedAt).toBe(time); + return 'ok'; + }); + + await expect(withPreconditionRetry('wrun_test', log, op)).resolves.toBe( + 'ok' + ); + expect(op).toHaveBeenCalledTimes(1); + }); + it('passes the latest snapshot time to op and returns its result without reloading', async () => { const time = 1_700_000_000_000; const log: MutableEventLog = { diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index a4f7294c29..524c7c2846 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -589,13 +589,15 @@ export interface MutableEventLog { } /** - * Whether the optimistic-concurrency guard for event creation is enabled - * (`WORKFLOW_PRECONDITION_GUARD=1`, set where the runtime executes). Off by - * default: replay-context creates only send a `stateUpdatedAt` snapshot (and - * can therefore be rejected with 412 by the backend) when it is enabled. + * Whether the optimistic-concurrency guard for event creation is enabled. + * **On by default** where the runtime executes: replay-context creates send a + * `stateUpdatedAt` snapshot (and can be rejected with 412 by a supporting + * backend) unless `WORKFLOW_PRECONDITION_GUARD` is set to `0`. Backends without + * guard support ignore the snapshot, so enabling by default is + * backward-compatible. */ export function isPreconditionGuardEnabled(): boolean { - return process.env.WORKFLOW_PRECONDITION_GUARD === '1'; + return process.env.WORKFLOW_PRECONDITION_GUARD !== '0'; } /** @@ -620,7 +622,17 @@ export function latestEventStateUpdatedAt(events: Event[]): number | undefined { const eventId = last.eventId; const underscore = eventId.lastIndexOf('_'); const rawUlid = underscore === -1 ? eventId : eventId.slice(underscore + 1); - return ulidToDate(rawUlid)?.getTime() ?? undefined; + const time = ulidToDate(rawUlid)?.getTime(); + if (time === undefined) { + // Fail open: a non-decodable id disarms the guard for this create (no + // snapshot sent). Log so a fleet-wide silent disarm is diagnosable. + runtimeLogger.debug( + 'Precondition guard: latest event id is not a decodable ULID; sending no snapshot', + { eventId } + ); + return undefined; + } + return time; } /** @@ -678,6 +690,13 @@ export async function withPreconditionRetry( new Set(log.events.map((e) => e.eventId)), loaded.events ); + // When several creates share one `log` (e.g. hook creations under + // `Promise.all` in `handleSuspension`), concurrent 412s can reload + // concurrently. The event merge above is safe — `appendUniqueEvents` + // builds its dedup set synchronously right before appending — but this + // cursor write is last-write-wins, so an interleaved older reload can + // briefly regress the cursor. The only consequence is refetching a few + // already-deduped events on a later load; correctness is unaffected. log.cursor = loaded.cursor ?? log.cursor; } } diff --git a/packages/errors/src/index.ts b/packages/errors/src/index.ts index 573e1d7218..3d7ab5a49f 100644 --- a/packages/errors/src/index.ts +++ b/packages/errors/src/index.ts @@ -769,6 +769,10 @@ export class ThrottleError extends WorkflowWorldError { * The workflow runtime handles this automatically: it reloads the event log * and retries, ultimately re-enqueueing the run if it cannot catch up. Users * interacting with world storage backends directly may encounter it. + * + * @property retryAfter - Delay in seconds before retrying. Accepted for + * forward-compatibility; the runtime currently reloads and retries + * immediately and does not read this field. */ export class PreconditionFailedError extends WorkflowWorldError { constructor(message: string, options?: { retryAfter?: number }) { diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index cb5969ea23..f0c6f419aa 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -686,6 +686,16 @@ export interface CreateEventParams { * when a newer out-of-band event was recorded after this snapshot, enabling * an optimistic-concurrency guard. Omitted by callers without a loaded event * log. + * + * Backend contract (for World implementers who want to support the guard): + * maintain a per-run marker holding the ULID time of the most recent + * *externally-originated* event — a `hook_received` or `step_completed` + * created **without** a `stateUpdatedAt` (replay-origin events carry one and + * must not advance the marker). On a create that carries `stateUpdatedAt`, + * reject with 412 when `stateUpdatedAt < marker` (strictly older); an equal + * timestamp must pass (anti-livelock, so an up-to-date client is never + * rejected). A backend that ignores this field simply disables the guard — + * the client falls open and behaves as before. */ stateUpdatedAt?: number; /**