From 565d49bcb611ea18484cf4f8beaff1b719be51f4 Mon Sep 17 00:00:00 2001 From: Dylan Slack Date: Fri, 23 Oct 2026 10:13:44 +0200 Subject: [PATCH 1/3] feat(runtime)!: report worker callback outcome instead of run state BREAKING CHANGE: POST /runs/:runId/effects/:effectId/:attempt/:action returns { status, cancelRequested } instead of the run state. createLooms().workerCallback and the HTTP client return the same shape. Heartbeats may carry details that the next attempt's task includes. failEffect and worker fail callbacks can mark an attempt non-retryable or set retryAfterMs. Co-authored-by: Cursor --- .changeset/effect-reliability.md | 9 + apps/docs/src/routes/docs.reliability.tsx | 8 +- docs/production.md | 19 +- packages/client/src/client.ts | 62 ++++++- packages/core/src/effects.ts | 65 ++++++- packages/core/src/fold.ts | 25 +++ packages/core/src/protocol.ts | 4 + packages/core/src/state.ts | 5 + packages/core/src/worker-callback.ts | 22 ++- packages/runtime/src/dispatch-effect.ts | 66 +++++-- packages/runtime/src/helpers.ts | 9 +- packages/runtime/src/index.ts | 1 + packages/runtime/src/local-effect-executor.ts | 3 + packages/runtime/src/looms.ts | 3 +- packages/runtime/src/planners.test.ts | 6 + packages/runtime/src/retry-policy.ts | 18 +- packages/runtime/src/runtime.ts | 2 +- packages/runtime/src/server/routes/effects.ts | 73 +++++++- packages/runtime/src/types.ts | 14 +- packages/runtime/src/worker-callbacks.ts | 170 +++++++++++------- .../runtime/src/worker-effect-dispatch.ts | 2 +- packages/runtime/src/worker-protocol.test.ts | 118 +++++++++++- 22 files changed, 593 insertions(+), 111 deletions(-) create mode 100644 .changeset/effect-reliability.md diff --git a/.changeset/effect-reliability.md b/.changeset/effect-reliability.md new file mode 100644 index 0000000..5e278e9 --- /dev/null +++ b/.changeset/effect-reliability.md @@ -0,0 +1,9 @@ +--- +'@swirls/looms': minor +--- + +Worker callbacks now answer `{ status: 'accepted' | 'stale' | 'cancelled', cancelRequested }` instead of the run state. `createLooms().workerCallback` and `createLoomsClient().workerCallback` return that result. `cancelRequested` tells an attempt that is already running to stop, because a dispatch that has left the host cannot be recalled. A stale report records nothing. + +`failEffect({ error, nonRetryable, retryAfterMs })` fails the current attempt. `nonRetryable` ends the effect even when the retry policy has attempts left. `retryAfterMs` waits at least that long before the next attempt and never shortens the policy backoff. Worker `fail` callbacks take the same two fields. + +A heartbeat may include `details`. The latest details are kept across retries and included on the next `EffectWorkerTask`, so a resumed attempt can continue from the recorded progress. diff --git a/apps/docs/src/routes/docs.reliability.tsx b/apps/docs/src/routes/docs.reliability.tsx index 243e287..4ccc211 100644 --- a/apps/docs/src/routes/docs.reliability.tsx +++ b/apps/docs/src/routes/docs.reliability.tsx @@ -67,7 +67,13 @@ function Page() { a restart preserves the retry budget and backoff. Configure start, heartbeat, and execution timeouts for work delivered to external workers. Worker callbacks include the effect ID and attempt number; stale callbacks cannot complete a newer attempt. Exhaustion records{' '} - runtime.effect.failed. + runtime.effect.failed. A handler can stop that budget early with{' '} + failEffect and nonRetryable: true, or ask the next attempt to wait + at least retryAfterMs without shortening the policy backoff. A worker heartbeat + can carry details; the next dispatch of that effect includes the latest details + so the attempt can resume. The callback response is{' '} + {'{ status, cancelRequested }'}, not the run state.{' '} + cancelRequested tells an attempt that is already running to stop.

Signals and webhook delivery diff --git a/docs/production.md b/docs/production.md index 866dee0..b84ea45 100644 --- a/docs/production.md +++ b/docs/production.md @@ -19,13 +19,24 @@ are `EventPayloadTooLargeError` and `SnapshotPayloadTooLargeError`. ## Worker effect protocol -1. The runtime dispatches `{ runId, effectId, attempt, type, input }`. +1. The runtime dispatches `{ runId, effectId, attempt, type, input, details? }`. + `details` is the progress carried by the previous attempt's last heartbeat. 2. The worker calls `started` before the schedule-to-start deadline. -3. Long work calls `heartbeat`; each accepted heartbeat extends the heartbeat - deadline without exceeding start-to-close. -4. The worker calls exactly one of `complete` or `fail`. +3. Long work calls `heartbeat`. Each accepted heartbeat extends the heartbeat + deadline without exceeding start-to-close. A heartbeat may include `details`; + the latest details are kept across retries and sent on the next dispatch. +4. The worker calls exactly one of `complete` or `fail`. `fail` may set + `nonRetryable` to stop the retry policy, or `retryAfterMs` to wait at least + that long before the next attempt (the policy backoff is never shortened). + A local handler does the same by failing with `failEffect`. 5. On cancellation, stop side effects and call `cancelled`. +Each callback response is `{ status: 'accepted' | 'stale' | 'cancelled', cancelRequested }`. +The host no longer returns the run state. `cancelRequested` is how a worker learns +that an attempt already in progress should stop: a queued dispatch cannot be recalled. +`stale` means the report was for another attempt, or the attempt had already settled, +and the host recorded nothing. + Callbacks include the same `effectId` and `attempt`. Stale callbacks are ignored. A dispatch transport failure is recorded as `ambiguous`: an operator must first verify the destination's idempotency record, then use `retryEffect` or diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index 413b98e..ae384f0 100644 --- a/packages/client/src/client.ts +++ b/packages/client/src/client.ts @@ -15,6 +15,7 @@ import { type JsonValue, type RunState, type WorkerCallbackInput, + type WorkerCallbackResult, } from '@looms/core' import { consumeSseStreamEffect, eventsFromSseData, sseFrames } from './sse' @@ -41,6 +42,53 @@ const StartResultSchema = Schema.Struct({ const RunResultSchema = Schema.Struct({ runId: Schema.String, state: RunStateSchema }) +interface WorkerFailRequestBody { + error: string + nonRetryable?: boolean + retryAfterMs?: number +} + +function workerCallbackBody(input: WorkerCallbackInput) { + switch (input.kind) { + case 'complete': + return { events: input.events } + + case 'fail': { + const body: WorkerFailRequestBody = { error: input.error } + + if (input.nonRetryable !== undefined) { + body.nonRetryable = input.nonRetryable + } + + if (input.retryAfterMs !== undefined) { + body.retryAfterMs = input.retryAfterMs + } + + return body + } + + case 'heartbeat': + return input.details === undefined ? {} : { details: input.details } + case 'started': + case 'cancelled': + return {} + + default: { + const exhaustive: never = input + return exhaustive + } + } +} + +const WorkerCallbackResultSchema = Schema.Struct({ + status: Schema.Union([ + Schema.Literal('accepted'), + Schema.Literal('stale'), + Schema.Literal('cancelled'), + ]), + cancelRequested: Schema.Boolean, +}) + /** `appended` counts the events the run gained while the wake ran. */ const WakeResultSchema = Schema.Struct({ runId: Schema.String, @@ -181,17 +229,15 @@ export function createLoomsClient(options: LoomsClientOptions = {}) { runId?: string }) => request('/runs', StartResultSchema, { method: 'POST', body: stringifyJson(args) }) - const workerCallback = (runId: string, input: WorkerCallbackInput) => { - const body = - input.kind === 'complete' - ? { events: input.events } - : input.kind === 'fail' - ? { error: input.error } - : {} + const workerCallback = ( + runId: string, + input: WorkerCallbackInput, + ): Promise => { + const body = workerCallbackBody(input) return request( `/runs/${encodeURIComponent(runId)}/effects/${encodeURIComponent(input.effectId)}/${input.attempt}/${input.kind}`, - RunResultSchema, + WorkerCallbackResultSchema, { method: 'POST', headers: options.workerCallbackToken diff --git a/packages/core/src/effects.ts b/packages/core/src/effects.ts index 57035c6..40f3c88 100644 --- a/packages/core/src/effects.ts +++ b/packages/core/src/effects.ts @@ -215,6 +215,52 @@ export class EffectHandlerError extends Data.TaggedError('EffectHandlerError')<{ } } +export interface EffectFailureOptions { + readonly error: string + /** Fail the attempt even when the retry policy has attempts remaining. */ + readonly nonRetryable?: boolean + /** Wait at least this long before the next attempt. Does not shorten the policy backoff. */ + readonly retryAfterMs?: number +} + +/** A handler's deliberate attempt failure, distinct from an unexpected thrown error. */ +export class EffectFailure extends Data.TaggedError('EffectFailure')<{ + readonly error: string + readonly nonRetryable: boolean + readonly retryAfterMs?: number +}> { + constructor(options: EffectFailureOptions) { + const error = options.error + const nonRetryable = options.nonRetryable === true + + if (options.retryAfterMs === undefined) { + super({ error, nonRetryable }) + } else { + super({ error, nonRetryable, retryAfterMs: options.retryAfterMs }) + } + + this.name = 'EffectFailure' + } +} + +/** Fail the current attempt. The retry policy decides whether another attempt is scheduled. */ +export function failEffect(options: EffectFailureOptions): Effect.Effect { + return Effect.fail(new EffectFailure(options)) +} + +/** Unwrap a handler failure, including one wrapped by `EffectHandlerError`. */ +export function readEffectFailure(cause: unknown): EffectFailure | null { + if (cause instanceof EffectFailure) { + return cause + } + + if (cause instanceof EffectHandlerError) { + return readEffectFailure(cause.cause) + } + + return null +} + export type EffectHandlerResult< R = never, E extends EventInput = EventInput, @@ -266,6 +312,18 @@ function liftHandlerResult( return Effect.succeed(result) } +function invokeHandler( + handler: (input: JsonValue, ctx: EffectContext) => EffectHandlerResult, + input: JsonValue, + ctx: EffectContext, +): Effect.Effect, EffectHandlerError, R> { + try { + return liftHandlerResult(handler(input, ctx)) + } catch (cause) { + return Effect.fail(new EffectHandlerError(cause)) + } +} + /** * Define an effect. With `execute`, a run's actor cell runs the handler at * most once while it is alive; if the cell dies mid-execution, the replacement @@ -327,7 +385,12 @@ export function defineEffect(def: { Effect.gen(function* () { const input = schema ? yield* validateInputEffect(schema, raw) : raw - return handler ? yield* liftHandlerResult(handler(input, ctx)) : noOutput + if (!handler) { + return noOutput + } + + // invokeHandler turns a synchronous throw into EffectHandlerError. + return yield* invokeHandler(handler, input, ctx) }), } } diff --git a/packages/core/src/fold.ts b/packages/core/src/fold.ts index 535df71..9f5d9d9 100644 --- a/packages/core/src/fold.ts +++ b/packages/core/src/fold.ts @@ -35,6 +35,28 @@ function readString(obj: { [key: string]: JsonValue }, key: string): string | un return Predicate.isString(value) ? value : undefined } +/** Heartbeat progress already recorded for this effect, if any. */ +function priorHeartbeatDetails(state: RunState, effectId: string): JsonValue | null { + return state.effectExecutions[effectId]?.heartbeatDetails ?? null +} + +/** + * A heartbeat that carries `details` replaces the saved progress. Every other execution event + * keeps it, including the retry that starts the next attempt. + */ +function nextHeartbeatDetails( + state: RunState, + effectId: string, + event: EventEnvelope, + payload: { [key: string]: JsonValue }, +): JsonValue | null { + if (event.type === 'runtime.effect.heartbeat' && Object.hasOwn(payload, 'details')) { + return payload.details ?? null + } + + return priorHeartbeatDetails(state, effectId) +} + function cloneState(state: RunState): RunState { return { runId: state.runId, @@ -290,6 +312,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi deadlineAt: null, startedAt: event.ts, lastHeartbeatAt: null, + heartbeatDetails: priorHeartbeatDetails(state, effectId), lastError: null, }, }, @@ -352,6 +375,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi : sameAttempt ? previous.lastHeartbeatAt : null, + heartbeatDetails: nextHeartbeatDetails(state, effectId, event, payload), lastError: readString(payload, 'error') ?? null, }, }, @@ -384,6 +408,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi deadlineAt: nextAttemptAt, startedAt: null, lastHeartbeatAt: null, + heartbeatDetails: priorHeartbeatDetails(state, effectId), lastError: readString(payload, 'error') ?? 'effect failed', }, }, diff --git a/packages/core/src/protocol.ts b/packages/core/src/protocol.ts index 527ceaf..2ca4b74 100644 --- a/packages/core/src/protocol.ts +++ b/packages/core/src/protocol.ts @@ -94,6 +94,8 @@ export const protocolCatalog = defineEventCatalog('runtime', { 'effect.failed': Schema.Struct({ effectId: Schema.String, error: Schema.String, + nonRetryable: Schema.optional(Schema.Boolean), + retryAfterMs: Schema.optional(Schema.Finite), }), 'effect.attempt.started': Schema.Struct({ effectId: Schema.String, @@ -119,6 +121,8 @@ export const protocolCatalog = defineEventCatalog('runtime', { attempt: Schema.Finite, heartbeatAt: Schema.Finite, deadlineAt: Schema.NullOr(Schema.Finite), + /** Progress the next attempt should resume from. Absent on heartbeats that only extend the deadline. */ + details: Schema.optional(JsonValueSchema), }), 'effect.cancel.requested': Schema.Struct({ effectId: Schema.String, diff --git a/packages/core/src/state.ts b/packages/core/src/state.ts index 58d370b..94115ae 100644 --- a/packages/core/src/state.ts +++ b/packages/core/src/state.ts @@ -58,6 +58,8 @@ export interface EffectExecutionRecord { /** When the current attempt started executing, locally or on a worker. */ startedAt: number | null lastHeartbeatAt: number | null + /** Progress reported by the latest heartbeat. Survives retries so the next attempt can resume. */ + heartbeatDetails: JsonValue | null lastError: string | null } @@ -189,6 +191,9 @@ export const RunStateSchema = Schema.Struct({ lastHeartbeatAt: Schema.NullOr(Schema.Finite).pipe( Schema.withDecodingDefaultKey(Effect.succeed(null)), ), + heartbeatDetails: Schema.NullOr(JsonValueSchema).pipe( + Schema.withDecodingDefaultKey(Effect.succeed(null)), + ), lastError: Schema.NullOr(Schema.String), }), ).pipe(Schema.withDecodingDefaultKey(Effect.succeed({}))), diff --git a/packages/core/src/worker-callback.ts b/packages/core/src/worker-callback.ts index f93722c..2badf51 100644 --- a/packages/core/src/worker-callback.ts +++ b/packages/core/src/worker-callback.ts @@ -1,4 +1,5 @@ import type { EventInput } from './envelope' +import type { JsonValue } from './types' /** Identifies one attempt of an effect executed by an external worker. */ export interface WorkerCallback { @@ -8,8 +9,25 @@ export interface WorkerCallback { /** What an external worker reports about an effect attempt. */ export type WorkerCallbackInput = - | (WorkerCallback & { readonly kind: 'started' | 'heartbeat' | 'cancelled' }) + | (WorkerCallback & { readonly kind: 'started' | 'cancelled' }) + | (WorkerCallback & { readonly kind: 'heartbeat'; readonly details?: JsonValue }) | (WorkerCallback & { readonly kind: 'complete'; readonly events: ReadonlyArray }) - | (WorkerCallback & { readonly kind: 'fail'; readonly error: string }) + | (WorkerCallback & { + readonly kind: 'fail' + readonly error: string + /** When set, the attempt is terminal even if the retry policy has attempts left. */ + readonly nonRetryable?: boolean + /** Extra delay before the next attempt. The policy backoff is never shortened. */ + readonly retryAfterMs?: number + }) export type WorkerCallbackKind = WorkerCallbackInput['kind'] + +/** + * What the host tells the worker about the callback it just sent. + * A queued message cannot be recalled, so `cancelRequested` is how an in-flight attempt learns to stop. + */ +export interface WorkerCallbackResult { + readonly status: 'accepted' | 'stale' | 'cancelled' + readonly cancelRequested: boolean +} diff --git a/packages/runtime/src/dispatch-effect.ts b/packages/runtime/src/dispatch-effect.ts index 93a533d..db43c8f 100644 --- a/packages/runtime/src/dispatch-effect.ts +++ b/packages/runtime/src/dispatch-effect.ts @@ -1,10 +1,11 @@ import type { Context, ManagedRuntime } from 'effect' -import { Effect } from 'effect' +import { Effect, Predicate } from 'effect' import { asJson, isPrimitiveEffect, isWaitOnTimer, + readEffectFailure, type ComposedRegistry, type EffectContext, type EventInput, @@ -16,6 +17,50 @@ import { import { threadStartedEvents } from './helpers' +function causeMessage(cause: unknown): string { + if (Predicate.isString(cause)) { + return cause + } + + if (cause instanceof Error) { + return cause.message + } + + return String(cause) +} + +function failedAttempt(effectId: string, threadId: string, cause: unknown): EventInput { + const failure = readEffectFailure(cause) + const error = failure?.error ?? causeMessage(cause) + const retryAfterMs = failure?.retryAfterMs + + if (failure?.nonRetryable && retryAfterMs !== undefined) { + return { + type: 'runtime.effect.failed', + payload: { effectId, error, nonRetryable: true, retryAfterMs }, + threadId, + } + } + + if (failure?.nonRetryable) { + return { + type: 'runtime.effect.failed', + payload: { effectId, error, nonRetryable: true }, + threadId, + } + } + + if (retryAfterMs !== undefined) { + return { + type: 'runtime.effect.failed', + payload: { effectId, error, retryAfterMs }, + threadId, + } + } + + return { type: 'runtime.effect.failed', payload: { effectId, error }, threadId } +} + export function dispatchEffect( registry: ComposedRegistry, services: ManagedRuntime.ManagedRuntime, @@ -110,11 +155,7 @@ export function dispatchEffect( if (!handler) { return Effect.succeed([ - { - type: 'runtime.effect.failed', - payload: { effectId: effCtx.effectId, error: `No handler for ${eff.type}` }, - threadId: currentThreadId, - }, + failedAttempt(effCtx.effectId, currentThreadId, `No handler for ${eff.type}`), ]) } @@ -129,18 +170,7 @@ export function dispatchEffect( }) return execution.pipe( - Effect.catch((err) => - Effect.succeed([ - { - type: 'runtime.effect.failed', - payload: { - effectId: effCtx.effectId, - error: err instanceof Error ? err.message : String(err), - }, - threadId: currentThreadId, - }, - ]), - ), + Effect.catch((err) => Effect.succeed([failedAttempt(effCtx.effectId, currentThreadId, err)])), ) } diff --git a/packages/runtime/src/helpers.ts b/packages/runtime/src/helpers.ts index c3d4961..ae05828 100644 --- a/packages/runtime/src/helpers.ts +++ b/packages/runtime/src/helpers.ts @@ -35,14 +35,21 @@ export function workerTask( runId: string, item: OutstandingEffect, attempt: number, + details?: JsonValue | null, ): EffectWorkerTask { - return { + const task = { runId, effectId: item.effectId, attempt, type: item.effect.type, input: 'input' in item.effect ? item.effect.input : null, } + + if (details != null) { + return { ...task, details } + } + + return task } export class UnknownDefinitionError extends Data.TaggedError('UnknownDefinitionError')<{ diff --git a/packages/runtime/src/index.ts b/packages/runtime/src/index.ts index 1f041f8..aca9628 100644 --- a/packages/runtime/src/index.ts +++ b/packages/runtime/src/index.ts @@ -20,6 +20,7 @@ export { type WakeScheduler, type WorkerCallback, type WorkerCallbackInput, + type WorkerCallbackResult, } from './runtime' export { createLooms, type Looms, type CreateLoomsOptions, type StartResult } from './looms' export { diff --git a/packages/runtime/src/local-effect-executor.ts b/packages/runtime/src/local-effect-executor.ts index e611713..dedf602 100644 --- a/packages/runtime/src/local-effect-executor.ts +++ b/packages/runtime/src/local-effect-executor.ts @@ -273,6 +273,7 @@ export function executeLocalEffects( const attempt = input.attempts.get(item.effectId) ?? 1 const payload = Predicate.isObject(outcome.payload) ? outcome.payload : {} const error = Predicate.isString(payload.error) ? payload.error : 'effect failed' + const retryAfterMs = payload.retryAfterMs const failure = planAttemptFailure({ runId: input.runId, @@ -282,6 +283,8 @@ export function executeLocalEffects( now: input.now, retry, origin: { type: 'system' }, + nonRetryable: payload.nonRetryable === true, + retryAfterMs: Predicate.isNumber(retryAfterMs) ? retryAfterMs : undefined, }) if (failure.retrying) { diff --git a/packages/runtime/src/looms.ts b/packages/runtime/src/looms.ts index a1fc06d..74922d6 100644 --- a/packages/runtime/src/looms.ts +++ b/packages/runtime/src/looms.ts @@ -29,6 +29,7 @@ import { type RunOperationalStatus, type WakeScheduler, type WorkerCallbackInput, + type WorkerCallbackResult, } from './runtime' import { createFetchHandler, @@ -135,7 +136,7 @@ export interface Looms { cancel(runId: string, threadId?: string): Promise /** Start a new run that continues where a finished run ended. */ fork(sourceRunId: string, options?: { runId?: string }): Promise - workerCallback(runId: string, input: WorkerCallbackInput): Promise + workerCallback(runId: string, input: WorkerCallbackInput): Promise fetch(req: Request): Promise serve(options?: { port?: number; hostname?: string }): RunningServer stop(): Promise diff --git a/packages/runtime/src/planners.test.ts b/packages/runtime/src/planners.test.ts index 53eefd5..1d667aa 100644 --- a/packages/runtime/src/planners.test.ts +++ b/packages/runtime/src/planners.test.ts @@ -63,6 +63,7 @@ describe('deadline transition planner', () => { nextAttemptAt: null, deadlineAt: 1_001, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: null, }, }, @@ -95,6 +96,7 @@ describe('deadline transition planner', () => { nextAttemptAt: null, deadlineAt: 900, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: null, }, future: { @@ -104,6 +106,7 @@ describe('deadline transition planner', () => { nextAttemptAt: null, deadlineAt: 1_200, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: null, }, }, @@ -143,6 +146,7 @@ describe('deadline transition planner', () => { deadlineAt: 9_000, startedAt: null, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: 'failed', }, }, @@ -186,6 +190,7 @@ describe('worker effect planner', () => { nextAttemptAt: 1_000, deadlineAt: null, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: 'temporary', }, }, @@ -249,6 +254,7 @@ describe('worker effect routing', () => { deadlineAt: null, startedAt: null, lastHeartbeatAt: null, + heartbeatDetails: null, lastError: null, } : undefined, diff --git a/packages/runtime/src/retry-policy.ts b/packages/runtime/src/retry-policy.ts index eadfe3e..2178dc3 100644 --- a/packages/runtime/src/retry-policy.ts +++ b/packages/runtime/src/retry-policy.ts @@ -18,6 +18,17 @@ export function retryBackoff(retry: RetryPolicy, attempt: number): number { ) } +/** Policy backoff, extended when the failed attempt asked to wait longer. */ +export function retryDelay(retry: RetryPolicy, attempt: number, retryAfterMs?: number): number { + const backoff = retryBackoff(retry, attempt) + + if (retryAfterMs === undefined || !Number.isFinite(retryAfterMs) || retryAfterMs < 0) { + return backoff + } + + return Math.max(backoff, retryAfterMs) +} + export interface AttemptFailurePlan { readonly event: EventEnvelope readonly retrying: boolean @@ -32,10 +43,13 @@ export function planAttemptFailure(input: { readonly now: number readonly retry: RetryPolicy | undefined readonly origin: EventOrigin + /** Customer/domain failures must not be retried even when attempts remain. */ + readonly nonRetryable?: boolean + readonly retryAfterMs?: number }): AttemptFailurePlan { const { runId, item, attempt, error, retry } = input - if (!retry || attempt >= retry.maxAttempts) { + if (input.nonRetryable || !retry || attempt >= retry.maxAttempts) { return { event: createEffectFailedEvent(runId, item, error), retrying: false } } @@ -46,7 +60,7 @@ export function planAttemptFailure(input: { payload: { effectId: item.effectId, attempt, - nextAttemptAt: input.now + retryBackoff(retry, attempt), + nextAttemptAt: input.now + retryDelay(retry, attempt, input.retryAfterMs), error, }, threadId: item.threadId, diff --git a/packages/runtime/src/runtime.ts b/packages/runtime/src/runtime.ts index ac0b33b..4fd29df 100644 --- a/packages/runtime/src/runtime.ts +++ b/packages/runtime/src/runtime.ts @@ -79,7 +79,7 @@ export type { StartRunArgs, WakeError, } from './types' -export type { WorkerCallback, WorkerCallbackInput } from '@looms/core' +export type { WorkerCallback, WorkerCallbackInput, WorkerCallbackResult } from '@looms/core' import type { CreateRuntimeOptions, LoomsRuntime, StartRunArgs } from './types' function currentTimeMillis(): number { diff --git a/packages/runtime/src/server/routes/effects.ts b/packages/runtime/src/server/routes/effects.ts index a63c6e5..402f65b 100644 --- a/packages/runtime/src/server/routes/effects.ts +++ b/packages/runtime/src/server/routes/effects.ts @@ -1,6 +1,11 @@ import { Effect, Schema } from 'effect' -import type { WorkerCallbackInput, WorkerCallbackKind } from '@looms/core' +import { + JsonValueSchema, + type JsonValue, + type WorkerCallbackInput, + type WorkerCallbackKind, +} from '@looms/core' import { EventInputBodySchema, provideStore, readJson, type RouteOptions } from '../http' import { integer, oneOf, route, type Route } from '../route-table' @@ -13,9 +18,28 @@ const WORKER_CALLBACK_KINDS = [ 'cancelled', ] as const satisfies readonly WorkerCallbackKind[] +interface FailWorkerCallback { + effectId: string + attempt: number + kind: 'fail' + error: string + nonRetryable?: boolean + retryAfterMs?: number +} + +interface HeartbeatWorkerCallback { + effectId: string + attempt: number + kind: 'heartbeat' + details?: JsonValue +} + const WorkerCallbackBodySchema = Schema.Struct({ events: Schema.optional(Schema.Array(EventInputBodySchema)), error: Schema.optional(Schema.String), + nonRetryable: Schema.optional(Schema.Boolean), + retryAfterMs: Schema.optional(Schema.Finite), + details: Schema.optional(JsonValueSchema), }) function workerCallbackInput( @@ -31,10 +55,47 @@ function workerCallbackInput( switch (params.action) { case 'complete': return { ...callback, kind: 'complete', events: body.events ?? [] } - case 'fail': - return body.error ? { ...callback, kind: 'fail', error: body.error } : null - default: + + case 'fail': { + if (body.error === undefined) { + return null + } + + const failure: FailWorkerCallback = { + ...callback, + kind: 'fail', + error: body.error, + } + + if (body.nonRetryable !== undefined) { + failure.nonRetryable = body.nonRetryable + } + + if (body.retryAfterMs !== undefined) { + failure.retryAfterMs = body.retryAfterMs + } + + return failure + } + + case 'heartbeat': { + const heartbeat: HeartbeatWorkerCallback = { ...callback, kind: 'heartbeat' } + + if (body.details !== undefined) { + heartbeat.details = body.details + } + + return heartbeat + } + + case 'started': + case 'cancelled': return { ...callback, kind: params.action } + + default: { + const exhaustive: never = params.action + return exhaustive + } } } @@ -80,8 +141,8 @@ export function effectRoutes({ runtime, store }: RouteOptions): readonly Route[] return Response.json({ error: 'error required' }, { status: 400 }) } - const state = yield* provideStore(runtime.workerCallback(params.runId, input), store) - return Response.json({ runId: params.runId, state }) + const result = yield* provideStore(runtime.workerCallback(params.runId, input), store) + return Response.json(result) }), }), ] diff --git a/packages/runtime/src/types.ts b/packages/runtime/src/types.ts index 850756b..0654bb9 100644 --- a/packages/runtime/src/types.ts +++ b/packages/runtime/src/types.ts @@ -6,6 +6,7 @@ import type { EventEnvelope, EventInput, WorkerCallbackInput, + WorkerCallbackResult, EventStore, EventStoreAppendError, EventStoreError, @@ -92,6 +93,8 @@ export interface EffectWorkerTask { readonly attempt: number readonly type: string readonly input: JsonValue + /** Progress from the previous attempt's last heartbeat, when it reported any. */ + readonly details?: JsonValue } export interface EffectWorker { @@ -203,11 +206,18 @@ export interface LoomsRuntime< sourceRunId: string, options?: { runId?: string }, ): Effect.Effect - /** Applies a worker's report about one effect attempt; stale or foreign reports are ignored. */ + /** + * Applies a worker's report about one effect attempt. A report for another attempt, or one that + * arrives after the attempt settled, is `stale` and changes nothing. + */ workerCallback( runId: string, input: WorkerCallbackInput, - ): Effect.Effect + ): Effect.Effect< + WorkerCallbackResult, + Error | EventStoreAppendError | EventStoreError, + EventStoreTag + > readonly listRuns: Effect.Effect /** Indexed run rows. Folds a run only when its header has not been written yet. */ readonly listRunSummaries: Effect.Effect< diff --git a/packages/runtime/src/worker-callbacks.ts b/packages/runtime/src/worker-callbacks.ts index b63c9a0..882618e 100644 --- a/packages/runtime/src/worker-callbacks.ts +++ b/packages/runtime/src/worker-callbacks.ts @@ -10,6 +10,7 @@ import { type RetryPolicy, type RunState, type WorkerCallbackInput, + type WorkerCallbackResult, } from '@looms/core' import { materializeEffectOutcome } from './helpers' @@ -61,9 +62,28 @@ export interface WorkerCallbacksOptions { interface WorkerCallbackPlan { readonly events: ReadonlyArray readonly notices: ReadonlyArray + readonly result: WorkerCallbackResult } -const IGNORED: WorkerCallbackPlan = { events: [], notices: [] } +const STALE: WorkerCallbackPlan = { + events: [], + notices: [], + result: { status: 'stale', cancelRequested: false }, +} + +const CANCELLED: WorkerCallbackPlan = { + events: [], + notices: [], + result: { status: 'cancelled', cancelRequested: true }, +} + +function accepted( + events: ReadonlyArray, + notices: ReadonlyArray = [], + cancelRequested = false, +): WorkerCallbackPlan { + return { events, notices, result: { status: 'accepted', cancelRequested } } +} /** * Turn a worker callback into events against the current durable state. Callbacks for another @@ -88,6 +108,15 @@ function planWorkerCallback(input: { execution?.attempt === callback.attempt && execution.status === 'cancel_requested' + // A message already handed to a worker cannot be pulled back. Tell it to stop instead. + if ( + execution?.attempt === callback.attempt && + execution.status === 'cancel_requested' && + callback.kind !== 'cancelled' + ) { + return Effect.succeed(CANCELLED) + } + if ( !execution || execution.attempt !== callback.attempt || @@ -97,10 +126,9 @@ function planWorkerCallback(input: { execution.status === 'retry_wait' || execution.status === 'timed_out' || execution.status === 'ambiguous' || - (callback.kind === 'cancelled' && !isCancellationAcknowledgement) || - (execution.status === 'cancel_requested' && callback.kind !== 'cancelled') + (callback.kind === 'cancelled' && !isCancellationAcknowledgement) ) { - return Effect.succeed(IGNORED) + return Effect.succeed(STALE) } const retry = item ? registry.effects.get(item.effect.type)?.retry : undefined @@ -117,8 +145,8 @@ function planWorkerCallback(input: { case 'started': { return Effect.succeed( item && execution.status === 'dispatched' - ? { events: [workerStartedEvent(runId, item, callback.attempt, retry, now)], notices: [] } - : IGNORED, + ? accepted([workerStartedEvent(runId, item, callback.attempt, retry, now)]) + : STALE, ) } @@ -129,38 +157,47 @@ function planWorkerCallback(input: { !item || (!implicitStart && execution.status !== 'started' && execution.status !== 'heartbeat') ) { - return Effect.succeed(IGNORED) + return Effect.succeed(STALE) } - return Effect.succeed({ - events: [ - ...(implicitStart ? [workerStartedEvent(runId, item, callback.attempt, retry, now)] : []), - createEvent(runId, { - type: 'runtime.effect.heartbeat', - payload: { - effectId: item.effectId, - attempt: callback.attempt, - heartbeatAt: now, - deadlineAt: workerDeadline(retry, execution.startedAt ?? now, now), - }, - threadId: item.threadId, - origin: WORKER_ORIGIN, - }), - ], - notices: [notice('effect.heartbeat')], - }) + const recorded = { + effectId: item.effectId, + attempt: callback.attempt, + heartbeatAt: now, + deadlineAt: workerDeadline(retry, execution.startedAt ?? now, now), + } + + const payload = + callback.details === undefined ? recorded : { ...recorded, details: callback.details } + + return Effect.succeed( + accepted( + [ + ...(implicitStart + ? [workerStartedEvent(runId, item, callback.attempt, retry, now)] + : []), + createEvent(runId, { + type: 'runtime.effect.heartbeat', + payload, + threadId: item.threadId, + origin: WORKER_ORIGIN, + }), + ], + [notice('effect.heartbeat')], + ), + ) } case 'complete': { if (!item) { - return Effect.succeed(IGNORED) + return Effect.succeed(STALE) } return Effect.forEach(callback.events, (event) => materializeEffectOutcome(registry.catalogs, runId, item, event), ).pipe( - Effect.map((outcomes) => ({ - events: [ + Effect.map((outcomes) => + accepted([ ...outcomes, createEvent(runId, { type: 'runtime.effect.completed', @@ -170,15 +207,14 @@ function planWorkerCallback(input: { causationId: item.causingEventId, origin: WORKER_ORIGIN, }), - ], - notices: [], - })), + ]), + ), ) } case 'fail': { if (!item) { - return Effect.succeed(IGNORED) + return Effect.succeed(STALE) } const failure = planAttemptFailure({ @@ -189,38 +225,45 @@ function planWorkerCallback(input: { now, retry, origin: WORKER_ORIGIN, + nonRetryable: callback.nonRetryable, + retryAfterMs: callback.retryAfterMs, }) - return Effect.succeed({ - events: [failure.event], - notices: failure.retrying - ? [ - { - type: 'effect.retry', - runId, - at: now, - effectId: item.effectId, - attempt: callback.attempt, - detail: callback.error, - }, - ] - : [], - }) + return Effect.succeed( + accepted( + [failure.event], + failure.retrying + ? [ + { + type: 'effect.retry', + runId, + at: now, + effectId: item.effectId, + attempt: callback.attempt, + detail: callback.error, + }, + ] + : [], + ), + ) } case 'cancelled': { - return Effect.succeed({ - events: [ - createEvent(runId, { - type: 'runtime.effect.cancelled', - payload: { effectId: callback.effectId, attempt: callback.attempt }, - threadId: item?.threadId ?? null, - effectId: callback.effectId, - origin: WORKER_ORIGIN, - }), - ], - notices: [notice('effect.cancellation')], - }) + return Effect.succeed( + accepted( + [ + createEvent(runId, { + type: 'runtime.effect.cancelled', + payload: { effectId: callback.effectId, attempt: callback.attempt }, + threadId: item?.threadId ?? null, + effectId: callback.effectId, + origin: WORKER_ORIGIN, + }), + ], + [notice('effect.cancellation')], + true, + ), + ) } default: { @@ -242,19 +285,22 @@ export function createWorkerCallbacks(options: WorkerCallbacksOptions) { const committed = yield* repository.commit(store, runId, (cursor) => planWorkerCallback({ runId, state: cursor.state, callback, registry, now }).pipe( - Effect.map((plan) => ({ events: plan.events, value: plan.notices })), + Effect.map((plan) => ({ + events: plan.events, + value: { notices: plan.notices, result: plan.result }, + })), ), ) if (committed.append) { - for (const notice of committed.value) { + for (const notice of committed.value.notices) { notifyObserver(options.observer, notice) } } - return committed.cursor.state + return committed.value.result }), - ).pipe(Effect.andThen(options.wake(runId))) + ).pipe(Effect.tap((result) => (result.status === 'stale' ? Effect.void : options.wake(runId)))) return { workerCallback: applyWorkerCallback } } diff --git a/packages/runtime/src/worker-effect-dispatch.ts b/packages/runtime/src/worker-effect-dispatch.ts index 478a88d..6501d1d 100644 --- a/packages/runtime/src/worker-effect-dispatch.ts +++ b/packages/runtime/src/worker-effect-dispatch.ts @@ -147,7 +147,7 @@ export function planWorkerEffects(input: { causationId: item.causingEventId, origin: { type: 'system' }, }), - task: workerTask(input.runId, item, attempt), + task: workerTask(input.runId, item, attempt, existing?.heartbeatDetails), scheduleToStartDeadlineAt: retry?.scheduleToStartTimeoutMs ? input.now + retry.scheduleToStartTimeoutMs : null, diff --git a/packages/runtime/src/worker-protocol.test.ts b/packages/runtime/src/worker-protocol.test.ts index fe4f539..3f3f8dc 100644 --- a/packages/runtime/src/worker-protocol.test.ts +++ b/packages/runtime/src/worker-protocol.test.ts @@ -7,6 +7,7 @@ import { defineEffect, defineRuntimeModule, defineThread, + failEffect, invoke, makeMemoryEventStore, type EventEnvelope, @@ -22,7 +23,10 @@ const workerDefinition = { value: { kind: 'worker-test', name: 'worker-test' }, } -function workerModule(retry: RetryPolicy, execute?: () => ReadonlyArray) { +function workerModule( + retry: RetryPolicy, + execute?: () => ReadonlyArray | Effect.Effect, unknown>, +) { const effect = defineEffect({ type: 'worker.do', retry, @@ -418,7 +422,9 @@ describe('durable external effect workers', () => { expect(unauthorized?.status).toBe(401) expect(unknownAction?.status).toBe(404) expect(startedResponse?.status).toBe(200) + expect(await startedResponse?.json()).toEqual({ status: 'accepted', cancelRequested: false }) expect(completedResponse?.status).toBe(200) + expect(await completedResponse?.json()).toEqual({ status: 'accepted', cancelRequested: false }) expect( (await looms.getEvents(started.runId)).some((event) => event.type === 'worker.done'), @@ -703,4 +709,114 @@ describe('durable external effect workers', () => { expect((await looms.getRun(started.runId)).status).toBe('cancelled') await looms.stop() }) + + test('a non-retryable failure spends no further attempts', async () => { + const looms = createLooms({ + modules: [ + workerModule({ maxAttempts: 3, backoffMs: 1 }, () => + failEffect({ error: 'bad input', nonRetryable: true }), + ), + ], + }) + + const started = await looms.start(workerDefinition, {}) + const types = (await looms.getEvents(started.runId)).map((event) => event.type) + + expect(types).toContain('runtime.effect.failed') + expect(types).not.toContain('runtime.effect.retry.scheduled') + expect((await looms.getRun(started.runId)).status).toBe('failed') + await looms.stop() + }) + + test('retryAfterMs extends the policy backoff', async () => { + const looms = createLooms({ + modules: [ + workerModule({ maxAttempts: 3, backoffMs: 10 }, () => + failEffect({ error: 'later', retryAfterMs: 60_000 }), + ), + ], + }) + + const started = await looms.start(workerDefinition, {}) + + const scheduled = (await looms.getEvents(started.runId)).find( + (event) => event.type === 'runtime.effect.retry.scheduled', + ) + + const payload = scheduled?.payload + + expect(scheduled).toBeDefined() + + if (payload && !Array.isArray(payload) && Predicate.isReadonlyObject(payload)) { + expect(payload.nextAttemptAt).toBeGreaterThanOrEqual((scheduled?.ts ?? 0) + 60_000) + } + + await looms.stop() + }) + + test('heartbeat details resume on the next worker attempt', async () => { + const dispatched: EffectWorkerTask[] = [] + + const looms = createLooms({ + modules: [workerModule({ maxAttempts: 3, backoffMs: 5 })], + worker: { dispatch: (task) => dispatched.push(task) }, + }) + + const started = await looms.start(workerDefinition, {}) + const identity = effectIdentity(await looms.getEvents(started.runId)) + + const heartbeat = await looms.workerCallback(started.runId, { + ...identity, + kind: 'heartbeat', + details: { page: 2 }, + }) + + expect(heartbeat).toEqual({ status: 'accepted', cancelRequested: false }) + + const failed = await looms.workerCallback(started.runId, { + ...identity, + kind: 'fail', + error: 'transient', + }) + + expect(failed).toEqual({ status: 'accepted', cancelRequested: false }) + await Bun.sleep(40) + + expect(dispatched.at(-1)?.attempt).toBe(2) + expect(dispatched.at(-1)?.details).toEqual({ page: 2 }) + + const stale = await looms.workerCallback(started.runId, { + ...identity, + kind: 'heartbeat', + details: { page: 9 }, + }) + + expect(stale).toEqual({ status: 'stale', cancelRequested: false }) + await looms.stop() + }) + + test('a heartbeat during cancellation tells the worker to stop', async () => { + const looms = createLooms({ + modules: [workerModule({ maxAttempts: 1, cancellationTimeoutMs: 60_000 })], + worker: { dispatch: () => undefined, cancel: () => undefined }, + }) + + const started = await looms.start(workerDefinition, {}) + const identity = effectIdentity(await looms.getEvents(started.runId)) + await looms.workerCallback(started.runId, { ...identity, kind: 'started' }) + await looms.cancel(started.runId) + + const heartbeat = await looms.workerCallback(started.runId, { + ...identity, + kind: 'heartbeat', + details: { page: 1 }, + }) + + expect(heartbeat).toEqual({ status: 'cancelled', cancelRequested: true }) + + const events = await looms.getEvents(started.runId) + const heartbeats = events.filter((event) => event.type === 'runtime.effect.heartbeat') + expect(heartbeats).toHaveLength(0) + await looms.stop() + }) }) From f43390ea24166b19f477a677fa7b267f77fd353b Mon Sep 17 00:00:00 2001 From: Dylan Slack Date: Fri, 23 Oct 2026 14:13:13 +0200 Subject: [PATCH 2/3] refactor(runtime): build effect failure payloads from the protocol type Co-authored-by: Cursor --- packages/runtime/src/dispatch-effect.ts | 29 ++++++------------------- 1 file changed, 7 insertions(+), 22 deletions(-) diff --git a/packages/runtime/src/dispatch-effect.ts b/packages/runtime/src/dispatch-effect.ts index db43c8f..21a8b2e 100644 --- a/packages/runtime/src/dispatch-effect.ts +++ b/packages/runtime/src/dispatch-effect.ts @@ -1,4 +1,4 @@ -import type { Context, ManagedRuntime } from 'effect' +import type { Context, ManagedRuntime, Types } from 'effect' import { Effect, Predicate } from 'effect' import { @@ -8,6 +8,7 @@ import { readEffectFailure, type ComposedRegistry, type EffectContext, + type EffectFailedPayload, type EventInput, RunLauncherTag, type RegisteredDefinition, @@ -32,33 +33,17 @@ function causeMessage(cause: unknown): string { function failedAttempt(effectId: string, threadId: string, cause: unknown): EventInput { const failure = readEffectFailure(cause) const error = failure?.error ?? causeMessage(cause) - const retryAfterMs = failure?.retryAfterMs - - if (failure?.nonRetryable && retryAfterMs !== undefined) { - return { - type: 'runtime.effect.failed', - payload: { effectId, error, nonRetryable: true, retryAfterMs }, - threadId, - } - } + const payload: Types.Mutable = { effectId, error } if (failure?.nonRetryable) { - return { - type: 'runtime.effect.failed', - payload: { effectId, error, nonRetryable: true }, - threadId, - } + payload.nonRetryable = true } - if (retryAfterMs !== undefined) { - return { - type: 'runtime.effect.failed', - payload: { effectId, error, retryAfterMs }, - threadId, - } + if (failure?.retryAfterMs !== undefined) { + payload.retryAfterMs = failure.retryAfterMs } - return { type: 'runtime.effect.failed', payload: { effectId, error }, threadId } + return { type: 'runtime.effect.failed', payload, threadId } } export function dispatchEffect( From c66540e2918443dbb0d5cb4944581593708d6aeb Mon Sep 17 00:00:00 2001 From: Dylan Slack Date: Fri, 23 Oct 2026 14:39:37 +0200 Subject: [PATCH 3/3] test(runtime): measure retryAfterMs from before the run starts The retry is scheduled from the wake's clock, which is read before the event timestamp, so comparing against the event could fail by a millisecond. The assertion also no longer skips when the payload is missing. Co-authored-by: Cursor --- packages/runtime/src/worker-protocol.test.ts | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/packages/runtime/src/worker-protocol.test.ts b/packages/runtime/src/worker-protocol.test.ts index 3f3f8dc..7bf30c8 100644 --- a/packages/runtime/src/worker-protocol.test.ts +++ b/packages/runtime/src/worker-protocol.test.ts @@ -737,19 +737,19 @@ describe('durable external effect workers', () => { ], }) + const before = Date.now() const started = await looms.start(workerDefinition, {}) - const scheduled = (await looms.getEvents(started.runId)).find( + const payload = (await looms.getEvents(started.runId)).find( (event) => event.type === 'runtime.effect.retry.scheduled', - ) - - const payload = scheduled?.payload + )?.payload - expect(scheduled).toBeDefined() + const nextAttemptAt = + Predicate.isReadonlyObject(payload) && !Array.isArray(payload) + ? payload.nextAttemptAt + : undefined - if (payload && !Array.isArray(payload) && Predicate.isReadonlyObject(payload)) { - expect(payload.nextAttemptAt).toBeGreaterThanOrEqual((scheduled?.ts ?? 0) + 60_000) - } + expect(nextAttemptAt).toBeGreaterThanOrEqual(before + 60_000) await looms.stop() })