From a01b3eed334368f58845157c386c76a9e2e1f9fc Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 11:33:03 -0700 Subject: [PATCH 1/7] feat: Coordinate abandoned effect recovery Return stable emit handles and retire abandoned effects atomically with durable recovery callbacks. Preserve owner heartbeat grace across cleanup and fence late completion using instance-before-effect locks. --- CHANGELOG.md | 11 + docs/api.md | 134 +++++++ docs/effect-recovery.md | 54 +++ docs/parity.md | 15 + package.json | 4 +- src/actor.ts | 81 ++++- src/cloudflare/engine.ts | 15 +- src/core.ts | 1 + src/doctor.ts | 12 +- src/effect-recovery-coordinator.ts | 236 +++++++++++++ src/effect-recovery.ts | 42 +++ src/index.ts | 2 + src/platform/uuid.ts | 3 +- src/repository.ts | 55 ++- src/schema.ts | 28 +- src/types.ts | 2 + test/actor-operations.types.ts | 2 +- test/cloudflare/effect-payloads.test.ts | 18 + test/cloudflare/worker.ts | 20 +- test/dead-letters.test.ts | 4 +- test/doctor.test.ts | 2 +- test/effect-recovery.test.ts | 404 ++++++++++++++++++++++ test/effect-recovery.types.ts | 70 ++++ test/fixtures/effect-payload-consumer.mts | 29 ++ test/runtime.test.ts | 4 +- 25 files changed, 1222 insertions(+), 26 deletions(-) create mode 100644 docs/effect-recovery.md create mode 100644 src/effect-recovery-coordinator.ts create mode 100644 src/effect-recovery.ts create mode 100644 test/effect-recovery.test.ts create mode 100644 test/effect-recovery.types.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 8365c93..ce29789 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## Unreleased + +- Return a stable `EffectHandle` from every `emit`. Overrides/wrappers must + return the handle; code expecting `void`/`undefined` needs migration. +- Add SQL effect recovery coordination with `onRecovery`, optional `onStatus`, + staged `requestEffectRecovery`, and an extending `recoveryTimeoutMilliseconds`. + Retirement and durable callbacks share the claim-locking transaction. Export + typed outcome constants and callback envelopes. Install schema migration 9 + before running this version. Cloudflare returns emit handles and rejects the + unsupported process-heartbeat recovery options before commit. + ## 0.14.9 - 2026-09-14 - Export effect failure/success payload types and `SerializedError` from the diff --git a/docs/api.md b/docs/api.md index b44f247..da4b66d 100644 --- a/docs/api.md +++ b/docs/api.md @@ -162,6 +162,140 @@ row per item. It also cannot strand an entry when the runtime coalesces an occurrence. Prefer it for a large queue of interchangeable items. Prefer `key` when one item needs an alarm that you can move on its own. +### Recovering abandoned effects + +`emit` returns an `EffectHandle` (`{ id: string }`) on every successful call. Save +it in actor state to identify that exact persisted effect. The handle, state, +effect, and callback bindings commit together; a rejected turn persists none of +them. Existing callers can ignore the handle. Overrides and wrappers that +previously returned `void` must return `super.emit(...)`; explicit +`return this.emit(...)` and code expecting `undefined` need updating. + +`onRecovery` opts a SQL effect into automatic retirement when its processing +owner stops heartbeating. `onStatus` is independent and optional: it only receives +responses to `requestEffectRecovery(handle)`, which requires both bindings. +Register both callbacks in `emit`; requests cannot rebind them. Requests stage an +intent in the current actor's fenced commit and perform no synchronous database +lookup inside the actor method. Only the originating instance can use its handle. + +`recoveryTimeoutMilliseconds` is an optional positive safe integer requiring +`onRecovery`. The effective freshness window is the greater of that persisted +override and the runtime's current `processAliveThresholdMilliseconds` (default +60,000). It can extend the window, never shorten it. It measures time since the +owner's database heartbeat, not effect duration or progress. An owner that keeps +heartbeating protects its effect indefinitely. + +`EffectRetiredPayload` is the `onRecovery` envelope: `effectId`, original +`arguments`, and `outcome: EffectRecoveryOutcome.Retired`. No outcome guard is +needed in that callback. `EffectRecoveryPayload` is the +discriminated union received by `onStatus`. `EffectRecoveryOutcome` is a frozen +constant object and a derived string-union type, exported from root and core. + +| Constant | Outcome | Meaning | +| ---------------- | ------------------ | ------------------------------------------------------------------ | +| `Retired` | `"retired"` | This check retired the abandoned processing effect. | +| `Deferred` | `"deferred"` | Owner is fresh; preserve its claim and attempts. | +| `Pending` | `"pending"` | Initial execution or retry remains with the scheduler. | +| `Completed` | `"completed"` | Includes original arguments and recorded result, including `null`. | +| `Dead` | `"dead"` | Preserve the existing terminal failure and failure callback. | +| `AlreadyRetired` | `"alreadyRetired"` | An earlier decision retired it; no new recovery notification. | +| `Missing` | `"missing"` | Owned routing metadata remains but the effect was pruned. | + +Every outcome includes `effectId`. Retired and completed require original +arguments; other outcomes may include retained arguments. Only completed has a +successful `result`. Database errors propagate as errors, never as missing or +abandoned outcomes. Unknown, foreign, and expired handles fail without exposing +another actor's effects or recreating a destroyed actor. + +Automatic retirement sends only `onRecovery`. A winning explicit check enqueues +`onRecovery` first and its separate `onStatus` response second in one transaction. +Only `onRecovery` should emit replacement work. A completed status can repair an +outcome notification using the same guarded helper as `onSuccess`: + +```ts +import { + Actor, + EffectRecoveryOutcome, + type EffectHandle, + type EffectRetiredPayload, + type EffectRecoveryPayload, + type EffectSuccessPayload, + type JsonValue, +} from "solid-objects" + +type ReportArguments = { revision: number } +type ReportResult = { artifactKey: string } + +class ReportExport extends Actor { + static override readonly actorType = "ReportExport" + revision = 0 + exportEffect: EffectHandle | null = null + artifactKey = "" + appliedEffectId: string | null = null + + start(): void { + this.exportEffect = this.emit("build_report", { + arguments: { revision: ++this.revision }, + onSuccess: "exportFinished", + onFailure: "exportFailed", + onRecovery: "recoverExport", + onStatus: "inspectExport", + recoveryTimeoutMilliseconds: 120_000, + }) + this.schedule({ at: new Date(Date.now() + 30_000), key: "export-watchdog" }).watchdog() + } + + watchdog(): void { + if (this.exportEffect) this.requestEffectRecovery(this.exportEffect) + } + + recoverExport(payload: EffectRetiredPayload): void { + if (payload.effectId !== this.exportEffect?.id || payload.arguments.revision !== this.revision) + return + this.start() + } + + exportFinished(payload: EffectSuccessPayload): void { + this.applyExportResult(payload) + } + + exportFailed(_payload: JsonValue): void {} + + inspectExport(payload: EffectRecoveryPayload): void { + if (payload.effectId !== this.exportEffect?.id) return + if (payload.outcome === EffectRecoveryOutcome.Completed) this.applyExportResult(payload) + if ( + payload.outcome === EffectRecoveryOutcome.Deferred || + payload.outcome === EffectRecoveryOutcome.Pending + ) { + this.schedule({ at: new Date(Date.now() + 30_000), key: "export-watchdog" }).watchdog() + } + } + + private applyExportResult(payload: EffectSuccessPayload): void { + if (payload.effectId !== this.exportEffect?.id || payload.arguments.revision !== this.revision) + return + if (this.appliedEffectId === payload.effectId) return + this.artifactKey = payload.result.artifactKey + this.appliedEffectId = payload.effectId + } +} +``` + +Routing metadata remains until the originating instance is destroyed or pruned; +it survives effect/message pruning but does not pin the instance. Checks after +that boundary fail. Callback delivery and idempotency follow durable mailbox +retention. Retirement survives a crash before callback delivery. + +The Durable Objects backend returns ordinary emit handles using its outbox ID, +but rejects recovery callbacks, timeouts, and recovery intents before committing +the actor turn: it has no shared SQL process-heartbeat registry. See +[effect recovery coordination](effect-recovery.md) for transaction and lock order. + +**External actions still require idempotency.** Retirement fences library state; +it does not cancel the previous JavaScript handler or prove its remote request +stopped. It does not provide exactly-once external execution. + ### Typing your onFailure handler An effect callback is an ordinary actor operation. Its payload always includes diff --git a/docs/effect-recovery.md b/docs/effect-recovery.md new file mode 100644 index 0000000..d885ab9 --- /dev/null +++ b/docs/effect-recovery.md @@ -0,0 +1,54 @@ +# Effect recovery coordination + +Install the additive schema migration and upgrade all effect workers and process +cleanup roles before emitting recovery-enabled effects. Older runtime versions +do not honor the persisted recovery bindings or the new lock protocol. + +The single `emit` API allocates a stable effect ID at staging and returns its +JSON handle. The fenced actor commit persists the effect, actor state, and +optional recovery/status binding together. `requestEffectRecovery` stages a +check on that same transaction connection. It never opens another transaction +while an application commit action holds locks. + +Automatic polling checks processing effects with `onRecovery` in one independent +transaction per candidate. Candidate reads are hints only. Lock order is origin +instance, effects ordered by ID, recovery bindings ordered by effect ID, then +current owner processes ordered by ID. Explicit batches acquire all effect and +binding locks before any process locks. Completion/failure lock the instance +before the effect. Pending claims lock the effect and never subsequently lock +the instance. Mailbox insertion reuses the origin instance lock. + +After waiting for these locks, the decision uses current ownership, a locked +heartbeat, database wall time, and the maximum of the current runtime threshold +and the persisted per-effect override. Missing owners are stale; query errors +are errors. Process shutdown preserves heartbeat evidence and opted-in claims. +Process pruning excludes effect owners; later polling revisits stopped owners +until each effect's individual grace expires. Ordinary effects and pending +retries retain their scheduler behavior. + +Retirement records `retired_at_ms` in `effect_recoveries`, clears the claim, and +uses the existing terminal `completed` effect storage state. The durable binding +distinguishes retirement from successful completion and is checked first by all +recovery observations. No success callback is generated. This avoids rewriting +existing status constraints across PostgreSQL, MySQL, and SQLite. Internal +effect-table status alone is not the recovery outcome. Late completion/failure +must still match a processing claim, which retirement removes. + +The terminal transition and `effect::recovery` mailbox insertion are atomic. +A winning explicit check additionally enqueues its separate status response, +after recovery, keyed by `effect::check:`. Failure of +either insertion rolls the transaction back. Multiple checks share one durable +retirement, with one response per request. A crash after commit cannot lose the +recovery callback. Wake-up signals are hints; mailbox polling provides delivery. + +Bindings belong to the exact originating instance, survive effect/message +pruning, and cascade when the instance is deleted. They neither pin instances nor +authorize cross-actor access. Within that retention lifetime a pruned effect can +report missing or already retired. Outside it, checks fail. Message idempotency +has normal mailbox retention; applications cannot supply internal request IDs. + +See [the watchdog example](api.md#recovering-abandoned-effects). Success and +completed-status repair use one application guard; status never owns replacement. +External systems still require idempotency across retries and replacement +generations. Stale heartbeat evidence grants library recovery permission; it +does not prove the previous handler or remote operation stopped. diff --git a/docs/parity.md b/docs/parity.md index 42fa817..7f14be9 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -217,3 +217,18 @@ Rails generators, Active Record models/controllers, Turbo rendering, and Action Cable are not copied into this package. The Rack dashboard is represented by the framework-neutral Fetch and Node adapter, renderer callbacks, and the same authorization and CSRF boundaries. + +## Effect recovery + +| Capability | Status | Contract | +| ----------------------------------------------- | -------------- | ------------------------------------------------------------------------------------------------------------------------------------ | +| Abandoned SQL effect recovery | Native | Stable emit handles, automatic retirement, staged status checks, durable callbacks, and extending heartbeat grace in both languages. | +| Shared process-heartbeat recovery on Cloudflare | Not applicable | Durable Objects has no shared SQL process registry; recovery options and intents fail before commit. | + +SQL effect recovery uses the same contract in Ruby and JavaScript: one `emit` +returns a stable handle; `onRecovery`/`on_recovery` opts into atomic retirement +and a durable callback; optional `onStatus`/`on_status` answers explicit staged +checks. Per-effect recovery timeouts extend the runtime heartbeat threshold +(milliseconds in JS, seconds in Ruby). Cloudflare returns emit handles but rejects +process-heartbeat recovery options and intents before commit. See +[the transaction protocol](effect-recovery.md). diff --git a/package.json b/package.json index 49b80fb..be4fcf4 100644 --- a/package.json +++ b/package.json @@ -115,8 +115,8 @@ "test": "vitest run", "test:browser": "pnpm run build && playwright test", "test:coverage": "vitest run --coverage", - "test:postgresql": "vitest run test/postgresql.test.ts", - "test:mysql": "vitest run test/mysql.test.ts", + "test:postgresql": "vitest run test/postgresql.test.ts test/effect-recovery.test.ts", + "test:mysql": "vitest run test/mysql.test.ts test/effect-recovery.test.ts", "test:package": "node scripts/release-artifact-smoke.mjs", "test:recovery": "pnpm run build && node examples/failure-recovery/demo.ts", "test:at-least-once": "pnpm run build && node examples/at-least-once/demo.ts", diff --git a/src/actor.ts b/src/actor.ts index f415c39..769eb5c 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -3,6 +3,7 @@ import { getDefaultRuntime } from "./default-runtime.js" import type { StateMigration } from "./definition.js" import { InvalidRejectionCode, Rejected, UnknownOperation } from "./errors.js" import { TRANSMIT_EFFECT } from "./transmit-effect.js" +import { randomUUID } from "./platform/uuid.js" import { createStagedOperationMap, createStagedOperations, @@ -12,7 +13,13 @@ import { type StagedOperations, } from "./reference.js" import { jsonObject, normalizeJson } from "./serialization.js" -import type { ActorIdentifier, JsonObject, JsonValue, MessageContext } from "./types.js" +import type { + ActorIdentifier, + EffectHandle, + JsonObject, + JsonValue, + MessageContext, +} from "./types.js" const observableBroadcastMode = Symbol("solid-objects.observable-broadcast-mode") @@ -47,16 +54,33 @@ export type PayloadBroadcasts = R > export interface EffectIntent { + id?: string name: string arguments: JsonObject successOperation?: string failureOperation?: string + recoveryOperation?: string + statusOperation?: string + recoveryTimeoutMilliseconds?: number } -export interface EffectOptions { +export interface EffectOptions< + Success extends string = string, + Failure extends string = Success, + Recovery extends string = string, + Status extends string = string, +> { arguments?: Record onSuccess?: Exclude onFailure?: Exclude + onRecovery?: Exclude + onStatus?: Exclude + recoveryTimeoutMilliseconds?: number +} + +export interface EffectRecoveryIntent { + effectId: string + requestId: string } type CallbackActor = string extends Callback @@ -95,6 +119,7 @@ export interface OutboundMessageIntent { export interface ActorIntents { effects: EffectIntent[] + effectRecoveries?: EffectRecoveryIntent[] commitActions: CommitActionIntent[] reminders: ReminderIntent[] outboundMessages: OutboundMessageIntent[] @@ -219,22 +244,63 @@ export abstract class Actor { }) } - emit( - this: CallbackActor> & CallbackActor>, + emit< + const Success extends string = never, + const Failure extends string = never, + const Recovery extends string = never, + const Status extends string = never, + >( + this: CallbackActor> & + CallbackActor> & + CallbackActor> & + CallbackActor>, name: string, - options: EffectOptions = {}, - ): void { - for (const callback of [options.onSuccess, options.onFailure]) { + options: EffectOptions = {}, + ): EffectHandle { + for (const callback of [ + options.onSuccess, + options.onFailure, + options.onRecovery, + options.onStatus, + ]) { if (callback !== undefined && !this.#operations.has(String(callback))) { throw new UnknownOperation(`unknown effect callback operation ${JSON.stringify(callback)}`) } } + const timeout = options.recoveryTimeoutMilliseconds + if (timeout !== undefined) { + if (!Number.isSafeInteger(timeout) || timeout <= 0) + throw new TypeError("recoveryTimeoutMilliseconds must be a positive safe integer") + if (options.onRecovery === undefined) + throw new TypeError("recoveryTimeoutMilliseconds requires onRecovery") + } + const id = randomUUID() this.#intents.effects.push({ + id, name, arguments: jsonObject(options.arguments ?? {}), ...(options.onSuccess === undefined ? {} : { successOperation: String(options.onSuccess) }), ...(options.onFailure === undefined ? {} : { failureOperation: String(options.onFailure) }), + ...(options.onRecovery === undefined + ? {} + : { recoveryOperation: String(options.onRecovery) }), + ...(options.onStatus === undefined ? {} : { statusOperation: String(options.onStatus) }), + ...(timeout === undefined ? {} : { recoveryTimeoutMilliseconds: timeout }), }) + return { id } + } + + requestEffectRecovery(handle: EffectHandle): void { + if ( + typeof handle !== "object" || + handle === null || + typeof handle.id !== "string" || + handle.id.length === 0 + ) { + throw new TypeError("effect recovery requires an effect handle") + } + this.#intents.effectRecoveries ??= [] + this.#intents.effectRecoveries.push({ effectId: handle.id, requestId: randomUUID() }) } transmit( @@ -338,6 +404,7 @@ export abstract class Actor { drainIntents(): ActorIntents { return { effects: this.#intents.effects.splice(0), + effectRecoveries: this.#intents.effectRecoveries?.splice(0) ?? [], commitActions: this.#intents.commitActions.splice(0), reminders: this.#intents.reminders.splice(0), outboundMessages: this.#intents.outboundMessages.splice(0), diff --git a/src/cloudflare/engine.ts b/src/cloudflare/engine.ts index 15d3e27..3cc562a 100644 --- a/src/cloudflare/engine.ts +++ b/src/cloudflare/engine.ts @@ -542,6 +542,19 @@ export class ActorEngine { maxResultBytes: this.settings.maxResultBytes, }) const intents = actor.drainIntents() + if ( + intents.effectRecoveries?.length || + intents.effects.some( + (effect) => + effect.recoveryOperation !== undefined || + effect.statusOperation !== undefined || + effect.recoveryTimeoutMilliseconds !== undefined, + ) + ) { + throw new UnsupportedCapability( + "the Durable Objects backend does not support process-heartbeat effect recovery", + ) + } if (intents.commitActions.length > 0) throw new UnsupportedCapability("the Durable Objects backend does not support commitAction") await this.store.atomic(() => { @@ -639,7 +652,7 @@ export class ActorEngine { }): void { const { instance, message, intents, broadcast } = options for (const effect of intents.effects) { - const id = crypto.randomUUID() + const id = effect.id ?? crypto.randomUUID() this.addOutbox({ id, instance, diff --git a/src/core.ts b/src/core.ts index 28ecd7b..ef78a41 100644 --- a/src/core.ts +++ b/src/core.ts @@ -1,5 +1,6 @@ export * from "./actor.js" export * from "./errors.js" +export * from "./effect-recovery.js" export type * from "./types.js" export type * from "./reference.js" export type { ActorRuntime } from "./actor-runtime.js" diff --git a/src/doctor.ts b/src/doctor.ts index 5c4de30..ad77b93 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -77,6 +77,14 @@ const EXPECTED_COLUMNS: Readonly> = { ], reminders: ["id", "instance_id", "operation", "message_operation", "run_at_ms", "status"], effects: ["id", "message_id", "instance_id", "name", "status", "available_at_ms"], + effect_recoveries: [ + "effect_id", + "instance_id", + "recovery_operation", + "status_operation", + "recovery_timeout_ms", + "retired_at_ms", + ], broadcasts: [ "id", "message_id", @@ -202,11 +210,11 @@ export class Doctor { message: `incompatible schema identity ${wrongIdentity.schema_identity}`, }) } - if (versions.join(",") !== "1,2,3,4,5,6,7,8") { + if (versions.join(",") !== "1,2,3,4,5,6,7,8,9") { return check({ name: "schema", status: "fail", - message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8; found ${versions.join(", ")}`, + message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8, 9; found ${versions.join(", ")}`, }) } return check({ diff --git a/src/effect-recovery-coordinator.ts b/src/effect-recovery-coordinator.ts new file mode 100644 index 0000000..a2ef95b --- /dev/null +++ b/src/effect-recovery-coordinator.ts @@ -0,0 +1,236 @@ +import type { EffectRecoveryIntent } from "./actor.js" +import type { RuntimeSettings } from "./configuration.js" +import type { DatabaseConnection } from "./database/types.js" +import { NonRetryableError } from "./errors.js" +import { + EffectRecoveryOutcome, + type EffectRecoveryPayload, + type EffectRetiredPayload, +} from "./effect-recovery.js" +import type { EffectRow, EnqueueInput, MessageRow, ProcessRow } from "./records.js" +import { jsonObject, normalizeJson } from "./serialization.js" + +interface RecoveryBinding { + effect_id: string + instance_id: string + recovery_operation: string | null + status_operation: string | null + recovery_timeout_ms: number | bigint | null + retired_at_ms: number | bigint | null +} + +interface Origin { + id: string + actor_type: string + actor_id: string +} + +export class EffectRecoveryCoordinator { + constructor( + private readonly options: { + settings: RuntimeSettings + enqueue: (connection: DatabaseConnection, input: EnqueueInput) => Promise + }, + ) {} + + async recoverAvailable(): Promise { + const candidates = await this.options.settings.database.connection((connection) => + connection.all( + `SELECT recoveries.* FROM ${this.table("effect_recoveries")} recoveries + JOIN ${this.table("effects")} effects ON effects.id = recoveries.effect_id + WHERE recoveries.retired_at_ms IS NULL AND recoveries.recovery_operation IS NOT NULL + AND effects.status = 'processing' ORDER BY recoveries.instance_id, recoveries.effect_id`, + ), + ) + for (const candidate of candidates) { + await this.options.settings.database.transaction(async (connection) => { + const origin = await connection.get( + `SELECT id, actor_type, actor_id FROM ${this.table("instances")} WHERE id = ?${this.lockClause()}`, + [candidate.instance_id], + ) + if (!origin) return + const effect = await connection.get( + `SELECT * FROM ${this.table("effects")} WHERE id = ? AND instance_id = ?${this.lockClause()}`, + [candidate.effect_id, origin.id], + ) + const binding = await connection.get( + `SELECT * FROM ${this.table("effect_recoveries")} WHERE effect_id = ? AND instance_id = ?${this.lockClause()}`, + [candidate.effect_id, origin.id], + ) + if ( + !effect || + !binding || + !binding.recovery_operation || + binding.retired_at_ms !== null || + effect.status !== "processing" + ) + return + const owner = + effect.claimed_by === null + ? undefined + : await connection.get( + `SELECT * FROM ${this.table("processes")} WHERE id = ?${this.lockClause()}`, + [effect.claimed_by], + ) + const now = await connection.nowMilliseconds() + if (this.ownerFresh({ binding, owner, now })) return + await this.retire({ connection, origin, binding, effect, now }) + }) + } + } + + async check(options: { + connection: DatabaseConnection + origin: Origin + intents: readonly EffectRecoveryIntent[] + }): Promise { + const { connection, origin, intents } = options + const effectIds = [...new Set(intents.map((intent) => intent.effectId))].sort() + if (effectIds.length === 0) return + const placeholders = effectIds.map(() => "?").join(", ") + const effects = new Map( + ( + await connection.all( + `SELECT * FROM ${this.table("effects")} WHERE instance_id = ? AND id IN (${placeholders}) ORDER BY id${this.lockClause()}`, + [origin.id, ...effectIds], + ) + ).map((effect) => [effect.id, effect]), + ) + const bindings = new Map( + ( + await connection.all( + `SELECT * FROM ${this.table("effect_recoveries")} WHERE instance_id = ? AND effect_id IN (${placeholders}) ORDER BY effect_id${this.lockClause()}`, + [origin.id, ...effectIds], + ) + ).map((binding) => [binding.effect_id, binding]), + ) + for (const id of effectIds) { + const binding = bindings.get(id) + if (!binding?.recovery_operation || !binding.status_operation) { + throw new NonRetryableError( + "effect recovery requires an owned handle with onRecovery and onStatus", + ) + } + } + const ownerIds = [ + ...new Set( + [...effects.values()].flatMap((effect) => + effect.claimed_by === null ? [] : [effect.claimed_by], + ), + ), + ].sort() + const owners = new Map() + for (const id of ownerIds) { + const owner = await connection.get( + `SELECT * FROM ${this.table("processes")} WHERE id = ?${this.lockClause()}`, + [id], + ) + if (owner) owners.set(id, owner) + } + const now = await connection.nowMilliseconds() + for (const intent of intents) { + const binding = bindings.get(intent.effectId)! + const effect = effects.get(intent.effectId) + const idempotencyKey = `effect:${intent.effectId}:check:${intent.requestId}` + const existing = await connection.get<{ id: string }>( + `SELECT id FROM ${this.table("messages")} WHERE instance_id = ? AND idempotency_key = ?`, + [origin.id, idempotencyKey], + ) + if (existing) continue + const payload = this.observe({ binding, effect, owners, now }) + if (payload.outcome === EffectRecoveryOutcome.Retired && effect) { + await this.retire({ connection, origin, binding, effect, now }) + } + await this.options.enqueue(connection, { + actorType: origin.actor_type, + actorId: origin.actor_id, + operation: binding.status_operation!, + deliveryMode: "internal", + arguments: jsonObject(payload), + idempotencyKey, + }) + } + } + + private observe(options: { + binding: RecoveryBinding + effect: EffectRow | undefined + owners: ReadonlyMap + now: number + }): EffectRecoveryPayload { + const { binding, effect, owners, now } = options + const effectId = binding.effect_id + if (binding.retired_at_ms !== null) + return { effectId, outcome: EffectRecoveryOutcome.AlreadyRetired } + if (!effect) return { effectId, outcome: EffectRecoveryOutcome.Missing } + const argumentsValue = jsonObject(JSON.parse(effect.arguments)) + if (effect.status === "completed") + return { + effectId, + arguments: argumentsValue, + outcome: EffectRecoveryOutcome.Completed, + result: normalizeJson(JSON.parse(effect.result ?? "null")), + } + if (effect.status === "dead") + return { effectId, arguments: argumentsValue, outcome: EffectRecoveryOutcome.Dead } + if (effect.status === "pending") + return { effectId, arguments: argumentsValue, outcome: EffectRecoveryOutcome.Pending } + const owner = effect.claimed_by === null ? undefined : owners.get(effect.claimed_by) + if (this.ownerFresh({ binding, owner, now })) + return { effectId, arguments: argumentsValue, outcome: EffectRecoveryOutcome.Deferred } + return { effectId, arguments: argumentsValue, outcome: EffectRecoveryOutcome.Retired } + } + + private ownerFresh(options: { + binding: RecoveryBinding + owner: ProcessRow | undefined + now: number + }): boolean { + const timeout = Math.max( + this.options.settings.processAliveThresholdMilliseconds, + Number(options.binding.recovery_timeout_ms ?? 0), + ) + return ( + options.owner !== undefined && Number(options.owner.heartbeat_at_ms) > options.now - timeout + ) + } + + private async retire(options: { + connection: DatabaseConnection + origin: Origin + binding: RecoveryBinding + effect: EffectRow + now: number + }): Promise { + const { connection, origin, binding, effect, now } = options + await connection.run( + `UPDATE ${this.table("effects")} SET status = 'completed', claimed_by = NULL WHERE id = ?`, + [effect.id], + ) + await connection.run( + `UPDATE ${this.table("effect_recoveries")} SET retired_at_ms = ? WHERE effect_id = ?`, + [now, effect.id], + ) + const payload: EffectRetiredPayload = { + effectId: effect.id, + arguments: jsonObject(JSON.parse(effect.arguments)), + outcome: EffectRecoveryOutcome.Retired, + } + await this.options.enqueue(connection, { + actorType: origin.actor_type, + actorId: origin.actor_id, + operation: binding.recovery_operation!, + deliveryMode: "internal", + arguments: jsonObject(payload), + idempotencyKey: `effect:${effect.id}:recovery`, + }) + binding.retired_at_ms = now + } + + private table(name: string): string { + return `${this.options.settings.tableNamePrefix}${name}` + } + private lockClause(): string { + return this.options.settings.database.family === "sqlite" ? "" : " FOR UPDATE" + } +} diff --git a/src/effect-recovery.ts b/src/effect-recovery.ts new file mode 100644 index 0000000..68a2887 --- /dev/null +++ b/src/effect-recovery.ts @@ -0,0 +1,42 @@ +import type { JsonObject, JsonValue } from "./types.js" + +export const EffectRecoveryOutcome = Object.freeze({ + Retired: "retired", + Deferred: "deferred", + Pending: "pending", + Completed: "completed", + Dead: "dead", + AlreadyRetired: "alreadyRetired", + Missing: "missing", +} as const) + +export type EffectRecoveryOutcome = + (typeof EffectRecoveryOutcome)[keyof typeof EffectRecoveryOutcome] + +export type EffectRetiredPayload = { + effectId: string + arguments: Arguments + outcome: typeof EffectRecoveryOutcome.Retired +} + +export type EffectRecoveryPayload< + Arguments extends JsonObject = JsonObject, + Result extends JsonValue = JsonValue, +> = + | EffectRetiredPayload + | { + effectId: string + arguments: Arguments + outcome: typeof EffectRecoveryOutcome.Completed + result: Result + } + | { + effectId: string + arguments?: Arguments + outcome: + | typeof EffectRecoveryOutcome.Deferred + | typeof EffectRecoveryOutcome.Pending + | typeof EffectRecoveryOutcome.Dead + | typeof EffectRecoveryOutcome.AlreadyRetired + | typeof EffectRecoveryOutcome.Missing + } diff --git a/src/index.ts b/src/index.ts index add7d09..792dc8b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -126,6 +126,7 @@ export type { DestroyOptions, EffectContext, EffectFailurePayload, + EffectHandle, EffectSuccessPayload, InvocationOptions, JsonObject, @@ -180,3 +181,4 @@ export { UnknownReminder, UnsupportedDatabase, } from "./errors.js" +export * from "./effect-recovery.js" diff --git a/src/platform/uuid.ts b/src/platform/uuid.ts index e2051e4..9c13cbd 100644 --- a/src/platform/uuid.ts +++ b/src/platform/uuid.ts @@ -1,3 +1,4 @@ export function randomUUID(): string { - return globalThis.crypto.randomUUID() + const platform = globalThis as typeof globalThis & { crypto: { randomUUID(): string } } + return platform.crypto.randomUUID() } diff --git a/src/repository.ts b/src/repository.ts index 1fafd48..93b9e24 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -35,6 +35,7 @@ import type { SerializedError, } from "./types.js" import { VERSION } from "./version.js" +import { EffectRecoveryCoordinator } from "./effect-recovery-coordinator.js" export interface SyncDiagnosticsRecord { message: MessageRow @@ -162,6 +163,7 @@ export class Repository { } async cleanupStaleProcesses(): Promise { + await this.effectRecoveryCoordinator().recoverAvailable() return this.settings.database.transaction(async (connection) => { const now = await connection.nowMilliseconds() const staleAt = now - this.settings.processAliveThresholdMilliseconds @@ -251,7 +253,10 @@ export class Repository { ) await connection.run( `UPDATE ${this.table("effects")} SET status = 'pending', claimed_by = NULL - WHERE status = 'processing' AND claimed_by = ?`, + WHERE status = 'processing' AND claimed_by = ? AND NOT EXISTS ( + SELECT 1 FROM ${this.table("effect_recoveries")} recoveries + WHERE recoveries.effect_id = ${this.table("effects")}.id AND recoveries.recovery_operation IS NOT NULL + )`, [processId], ) await connection.run( @@ -266,9 +271,9 @@ export class Repository { ) await connection.run( `UPDATE ${this.table("processes")} - SET shutdown_state = 'stopped', stopped_at_ms = ?, heartbeat_at_ms = ? + SET shutdown_state = 'stopped', stopped_at_ms = ? WHERE id = ? AND shutdown_state <> 'stopped'`, - [now, now, processId], + [now, processId], ) } @@ -719,13 +724,14 @@ export class Repository { ]) for (const effect of input.intents.effects) { + const effectId = effect.id ?? randomUUID() await connection.run( `INSERT INTO ${this.table("effects")} (id, message_id, instance_id, name, arguments, success_operation, failure_operation, status, max_attempts, available_at_ms) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)`, [ - randomUUID(), + effectId, turn.message.id, turn.instance.id, effect.name, @@ -736,8 +742,28 @@ export class Repository { now, ], ) + if (effect.recoveryOperation !== undefined || effect.statusOperation !== undefined) { + await connection.run( + `INSERT INTO ${this.table("effect_recoveries")} + (effect_id, instance_id, recovery_operation, status_operation, recovery_timeout_ms) + VALUES (?, ?, ?, ?, ?)`, + [ + effectId, + turn.instance.id, + effect.recoveryOperation ?? null, + effect.statusOperation ?? null, + effect.recoveryTimeoutMilliseconds ?? null, + ], + ) + } } + await this.effectRecoveryCoordinator().check({ + connection, + origin: turn.instance, + intents: input.intents.effectRecoveries ?? [], + }) + for (const reminder of input.intents.reminders) { const existing = await connection.get<{ id: string; run_at_ms: number | bigint }>( `SELECT id, run_at_ms FROM ${this.table("reminders")} @@ -1373,6 +1399,9 @@ export class Repository { const table = this.table("processes") return { sql: `${table}.shutdown_state = 'stopped' + AND NOT EXISTS ( + SELECT 1 FROM ${this.table("effects")} effects WHERE effects.claimed_by = ${table}.id + ) AND ${table}.stopped_at_ms IS NOT NULL AND ${table}.stopped_at_ms < ? AND NOT EXISTS ( @@ -1413,6 +1442,7 @@ export class Repository { async resetForTesting(): Promise { const tables = [ + "effect_recoveries", "dead_letters", "claimed_messages", "ready_messages", @@ -1429,13 +1459,17 @@ export class Repository { } async claimEffect(processId: string): Promise { + await this.effectRecoveryCoordinator().recoverAvailable() return this.settings.database.transaction( async (connection) => { const now = await connection.nowMilliseconds() const staleAt = now - this.settings.processAliveThresholdMilliseconds await connection.run( `UPDATE ${this.table("effects")} SET status = 'pending', claimed_by = NULL - WHERE status = 'processing' AND ( + WHERE status = 'processing' AND NOT EXISTS ( + SELECT 1 FROM ${this.table("effect_recoveries")} recoveries + WHERE recoveries.effect_id = ${this.table("effects")}.id AND recoveries.recovery_operation IS NOT NULL + ) AND ( claimed_by IS NULL OR NOT EXISTS ( SELECT 1 FROM ${this.table("processes")} processes WHERE processes.id = ${this.table("effects")}.claimed_by @@ -1475,6 +1509,8 @@ export class Repository { async completeEffect(effect: EffectRow, result: JsonValue): Promise { await this.settings.database.transaction(async (connection) => { + if (!(await this.lockActivationFence(connection, effect.instance_id))) + throw new LostActivation("effect origin no longer exists") const now = await connection.nowMilliseconds() const updated = await connection.run( `UPDATE ${this.table("effects")} @@ -1507,6 +1543,8 @@ export class Repository { }): Promise { const { effect, error, retryable } = options await this.settings.database.transaction(async (connection) => { + if (!(await this.lockActivationFence(connection, effect.instance_id))) + throw new LostActivation("effect origin no longer exists") const now = await connection.nowMilliseconds() const errorRecord = safeError(error) const exhausted = !retryable || Number(effect.attempt_count) >= Number(effect.max_attempts) @@ -1981,6 +2019,13 @@ export class Repository { return " FOR UPDATE SKIP LOCKED" } + private effectRecoveryCoordinator(): EffectRecoveryCoordinator { + return new EffectRecoveryCoordinator({ + settings: this.settings, + enqueue: (connection, input) => this.enqueueInTransaction(connection, input), + }) + } + private rowLockClause(): string { if (this.settings.database.family === "sqlite") return "" return " FOR UPDATE" diff --git a/src/schema.ts b/src/schema.ts index de88974..5514d5c 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -9,7 +9,8 @@ const PROCESS_DRAINING_VERSION = 5 const OBSERVABLE_INVALIDATIONS_VERSION = 6 const KEYED_REMINDERS_VERSION = 7 const POLLING_INDEXES_VERSION = 8 -const LATEST_VERSION = POLLING_INDEXES_VERSION +const EFFECT_RECOVERY_VERSION = 9 +const LATEST_VERSION = EFFECT_RECOVERY_VERSION export async function installSchema(options: { connection: DatabaseConnection @@ -307,6 +308,31 @@ export async function installSchema(options: { }) } + await createTable(`CREATE TABLE IF NOT EXISTS ${table("effect_recoveries")} ( + effect_id TEXT PRIMARY KEY, + instance_id TEXT NOT NULL, + recovery_operation TEXT, + status_operation TEXT, + recovery_timeout_ms INTEGER CHECK (recovery_timeout_ms IS NULL OR recovery_timeout_ms > 0), + retired_at_ms INTEGER, + FOREIGN KEY (instance_id) REFERENCES ${table("instances")}(id) ON DELETE CASCADE + ) STRICT`) + await createIndex({ + connection, + family, + table: table("effect_recoveries"), + name: `${prefix}effect_recoveries_instance`, + columns: "instance_id", + }) + if (!installedVersions.has(EFFECT_RECOVERY_VERSION)) { + await recordMigration({ + connection, + table: table("schema_migrations"), + version: EFFECT_RECOVERY_VERSION, + schemaIdentity, + }) + } + if (installedVersions.has(POLLING_INDEXES_VERSION)) return const pollingIndexes = [ ["effects", `${prefix}effects_poll`, "status, available_at_ms, id"], diff --git a/src/types.ts b/src/types.ts index a70880c..b74d217 100644 --- a/src/types.ts +++ b/src/types.ts @@ -2,6 +2,8 @@ export type JsonPrimitive = null | boolean | number | string export type JsonValue = JsonPrimitive | JsonValue[] | { [key: string]: JsonValue } export type JsonObject = { [key: string]: JsonValue } +export type EffectHandle = { readonly id: string } + export type SerializedError = { name: string message: string diff --git a/test/actor-operations.types.ts b/test/actor-operations.types.ts index 768b517..dbfe073 100644 --- a/test/actor-operations.types.ts +++ b/test/actor-operations.types.ts @@ -119,7 +119,7 @@ export function genericActor(actor: ActorType, callback export class ExistingOverride extends Actor { override emit(name: string, options: { onFailure?: string } = {}) { - super.emit(name, options) + return super.emit(name, options) } // @ts-expect-error broad legacy override cannot promise concrete operation keys override schedule(options: { at: Date }): ScheduledOperations { diff --git a/test/cloudflare/effect-payloads.test.ts b/test/cloudflare/effect-payloads.test.ts index 1ccd733..7d169d8 100644 --- a/test/cloudflare/effect-payloads.test.ts +++ b/test/cloudflare/effect-payloads.test.ts @@ -7,6 +7,21 @@ const authorizationContext = "allowed" const runtime = () => createRuntime({ backend: durableObjects({ namespace: env.ACTORS }) }) describe("Cloudflare effect payloads", () => { + it.each(["startRecoverable", "startStatusOnly", "checkRecovery"] as const)( + "rejects unsupported %s before committing state or effects", + async (operation) => { + const reference = runtime().ref(EffectCallbacks, `unsupported-${operation}`) + await expect(reference.with({ authorizationContext })[operation]()).rejects.toMatchObject({ + name: "MessageFailed", + details: { + name: "UnsupportedCapability", + message: "the Durable Objects backend does not support process-heartbeat effect recovery", + }, + }) + expect((await reference.snapshot({ authorizationContext })).effectHandle).toBeNull() + expect((await reference.snapshot({ authorizationContext })).received).toEqual([]) + }, + ) it.each([null, false, 42, "reply", ["reply"], { reply: "done" }])( "delivers the complete success envelope for %j", async (result) => { @@ -20,6 +35,9 @@ describe("Cloudflare effect payloads", () => { .toEqual([{ effectId: expect.any(String), arguments: argumentsValue, result }]) const [payload] = (await reference.snapshot({ authorizationContext })).received expect(deliveries.get(payload!.effectId)).toBe(1) + expect((await reference.snapshot({ authorizationContext })).effectHandle).toEqual({ + id: payload!.effectId, + }) }, ) diff --git a/test/cloudflare/worker.ts b/test/cloudflare/worker.ts index a280e3a..c6c82f9 100644 --- a/test/cloudflare/worker.ts +++ b/test/cloudflare/worker.ts @@ -4,6 +4,7 @@ import { NonRetryableError, type EffectFailurePayload, type EffectSuccessPayload, + type EffectHandle, type JsonObject, } from "../../src/core.js" import { PortableCounter } from "../support/portable-actor.js" @@ -152,9 +153,10 @@ export class VersionedCounter extends Actor { export class EffectCallbacks extends Actor { static override readonly actorType = "EffectCallbacks" received: (EffectSuccessPayload | EffectFailurePayload)[] = [] + effectHandle: EffectHandle | null = null start(argumentsValue: JsonObject): void { - this.emit("callbackValue", { + this.effectHandle = this.emit("callbackValue", { arguments: argumentsValue, onSuccess: "succeeded", onFailure: "failed", @@ -162,9 +164,23 @@ export class EffectCallbacks extends Actor { } startEmpty(): void { - this.emit("callbackEmpty", { onSuccess: "succeeded" }) + this.effectHandle = this.emit("callbackEmpty", { onSuccess: "succeeded" }) } + startRecoverable(): void { + this.effectHandle = this.emit("callbackEmpty", { onRecovery: "recover" }) + } + + startStatusOnly(): void { + this.effectHandle = this.emit("callbackEmpty", { onStatus: "inspect" }) + } + + checkRecovery(): void { + this.requestEffectRecovery({ id: "unsupported" }) + } + recover(): void {} + inspect(): void {} + succeeded(payload: EffectSuccessPayload): void { this.received.push(payload) } diff --git a/test/dead-letters.test.ts b/test/dead-letters.test.ts index 5ca173f..89f4228 100644 --- a/test/dead-letters.test.ts +++ b/test/dead-letters.test.ts @@ -151,7 +151,7 @@ describe("schema migrations", () => { const broadcastColumns = await runtime.settings.database.connection((connection) => connection.all<{ name: string }>("PRAGMA table_info(solid_objects_broadcasts)"), ) - expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8]) + expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9]) expect(deadLetterColumns.map(({ name }) => name)).toContain("retried_message_id") expect(broadcastColumns.map(({ name }) => name)).toContain("invalidations") expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) @@ -184,7 +184,7 @@ describe("schema migrations", () => { "SELECT version FROM solid_objects_schema_migrations ORDER BY version", ), ) - expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8]) + expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9]) expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) }) diff --git a/test/doctor.test.ts b/test/doctor.test.ts index 9e41f62..8c21e1b 100644 --- a/test/doctor.test.ts +++ b/test/doctor.test.ts @@ -27,7 +27,7 @@ describe("runtime doctor", () => { expect(check(report, "configuration").status).toBe("pass") expect(check(report, "schema")).toMatchObject({ status: "pass", - details: { versions: [1, 2, 3, 4, 5, 6, 7, 8] }, + details: { versions: [1, 2, 3, 4, 5, 6, 7, 8, 9] }, }) expect(check(report, "authorization").status).toBe("pass") expect(check(report, "database").status).toBe("pass") diff --git a/test/effect-recovery.test.ts b/test/effect-recovery.test.ts new file mode 100644 index 0000000..71a1a36 --- /dev/null +++ b/test/effect-recovery.test.ts @@ -0,0 +1,404 @@ +import { afterEach, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { sqlite } from "../src/database/sqlite.js" +import { postgresql } from "../src/database/postgresql.js" +import { mysql } from "../src/database/mysql.js" +import type { + EffectHandle, + EffectSuccessPayload, + EffectFailurePayload, + JsonObject, +} from "../src/types.js" +import type { EffectRecoveryPayload, EffectRetiredPayload } from "../src/effect-recovery.js" +import { LostActivation } from "../src/errors.js" +import { EffectRecoveryCoordinator } from "../src/effect-recovery-coordinator.js" +import type { DatabaseConnection } from "../src/database/types.js" +import type { EffectRow } from "../src/records.js" + +class ReportExport extends Actor { + static override readonly actorType = "ReportExport" + exportEffect: EffectHandle | null = null + notifications: JsonObject[] = [] + + start(): void { + this.exportEffect = this.emit("build_report", { arguments: { revision: 1 } }) ?? null + } + + startRecoverable({ timeoutMilliseconds }: { timeoutMilliseconds?: number } = {}): void { + this.exportEffect = this.emit("build_report", { + arguments: { revision: 1 }, + onRecovery: "recover", + onStatus: "inspect", + onSuccess: "finished", + onFailure: "failed", + ...(timeoutMilliseconds === undefined + ? {} + : { recoveryTimeoutMilliseconds: timeoutMilliseconds }), + }) + } + + check(): void { + if (this.exportEffect) this.requestEffectRecovery(this.exportEffect) + } + recover(payload: EffectRetiredPayload): void { + this.notifications.push({ kind: "recovery", ...payload }) + } + inspect(payload: EffectRecoveryPayload): void { + this.notifications.push({ kind: "status", ...payload }) + } + finished(payload: EffectSuccessPayload): void { + this.notifications.push({ kind: "success", ...payload }) + } + failed(payload: EffectFailurePayload): void { + this.notifications.push({ kind: "failure", ...payload }) + } +} + +let runtime: SolidObjectsRuntime | undefined +afterEach(async () => { + await runtime?.repository.resetForTesting() + await runtime?.close() +}) + +it("returns the same effect identity that the scheduler claims", async () => { + runtime = await createTestRuntime() + await runtime.ref(ReportExport, "export").start() + const effect = await runtime.settings.database.connection((connection) => + connection.get<{ id: string }>(`SELECT id FROM ${runtime!.repository.table("effects")}`), + ) + expect(await runtime.ref(ReportExport, "export").exportEffect).toEqual({ id: effect!.id }) +}) + +it("retires abandoned processing effects before the scheduler can reclaim them", async () => { + runtime = await createTestRuntime() + await runtime.ref(ReportExport, "export").startRecoverable() + await runtime.repository.registerProcess("owner", "effect") + const effect = await runtime.repository.claimEffect("owner") + await runtime.settings.database.connection((connection) => + connection.run( + `UPDATE ${runtime!.repository.table("processes")} SET heartbeat_at_ms = 0 WHERE id = ?`, + ["owner"], + ), + ) + await runtime.repository.registerProcess("replacement", "effect") + expect(await runtime.repository.claimEffect("replacement")).toBeUndefined() + const notifications = await runtime.settings.database.connection((connection) => + connection.all<{ arguments: string }>( + `SELECT arguments FROM ${runtime!.repository.table("messages")} WHERE operation = 'recover'`, + ), + ) + expect(notifications.map((row) => JSON.parse(row.arguments))).toEqual([ + { effectId: effect!.id, arguments: { revision: 1 }, outcome: "retired" }, + ]) +}) + +it("defers indefinitely for a fresh heartbeat without consuming an attempt", async () => { + const effect = await processingEffect() + await runtime!.settings.database.connection((connection) => + connection.run( + `UPDATE ${runtime!.repository.table("effects")} SET available_at_ms = 0 WHERE id = ?`, + [effect.id], + ), + ) + await runtime!.ref(ReportExport, "export").check() + await runtime!.worker().runUntilIdle() + expect(await runtime!.ref(ReportExport, "export").notifications).toEqual([ + { kind: "status", effectId: effect.id, arguments: { revision: 1 }, outcome: "deferred" }, + ]) + expect(await storedEffect(effect.id)).toMatchObject({ + status: "processing", + claimed_by: "owner", + attempt_count: 1n, + }) +}) + +it("preserves a longer grace after ordinary cleanup stops the owner", async () => { + const effect = await processingEffect({ timeoutMilliseconds: 120_000 }) + await ageOwner(75_000) + await runtime!.repository.cleanupStaleProcesses() + expect(await storedEffect(effect.id)).toMatchObject({ status: "processing", claimed_by: "owner" }) + expect(await runtime!.repository.claimEffect("owner")).toBeUndefined() + await ageOwner(125_000) + expect(await runtime!.repository.claimEffect("owner")).toBeUndefined() + expect(await messages("recover")).toHaveLength(1) +}) + +it("keeps pending retries with the existing scheduler", async () => { + const effect = await processingEffect() + await runtime!.repository.failEffect({ effect, error: new Error("retry"), retryable: true }) + await runtime!.ref(ReportExport, "export").check() + expect((await messages("inspect"))[0]).toMatchObject({ effectId: effect.id, outcome: "pending" }) + expect(await storedEffect(effect.id)).toMatchObject({ status: "pending", attempt_count: 1n }) + expect(await messages("recover")).toHaveLength(0) +}) + +it("returns recorded null results and preserves the normal success callback", async () => { + const effect = await processingEffect() + await runtime!.repository.completeEffect(effect, null) + await runtime!.ref(ReportExport, "export").check() + expect(await messages("inspect")).toEqual([ + { effectId: effect.id, arguments: { revision: 1 }, outcome: "completed", result: null }, + ]) + expect(await messages("finished")).toEqual([ + { effectId: effect.id, arguments: { revision: 1 }, result: null }, + ]) + expect(await messages("recover")).toHaveLength(0) +}) + +it("preserves dead effects and their failure callbacks", async () => { + const effect = await processingEffect() + await runtime!.repository.failEffect({ effect, error: new Error("terminal"), retryable: false }) + await runtime!.ref(ReportExport, "export").check() + expect((await messages("inspect"))[0]).toMatchObject({ outcome: "dead" }) + expect(await messages("failed")).toHaveLength(1) + expect(await messages("recover")).toHaveLength(0) +}) + +it("deduplicates retirement across repeated checks and fences late completion and failure", async () => { + const effect = await processingEffect() + await ageOwner(70_000) + await runtime!.ref(ReportExport, "export").check() + await runtime!.ref(ReportExport, "export").check() + expect(await messages("recover")).toHaveLength(1) + expect((await messages("inspect")).map((payload) => payload.outcome)).toEqual([ + "retired", + "alreadyRetired", + ]) + await expect(runtime!.repository.completeEffect(effect, "late")).rejects.toBeInstanceOf( + LostActivation, + ) + await expect( + runtime!.repository.failEffect({ effect, error: new Error("late"), retryable: true }), + ).rejects.toBeInstanceOf(LostActivation) + await expect( + runtime!.repository.failEffect({ effect, error: new Error("late"), retryable: false }), + ).rejects.toBeInstanceOf(LostActivation) + expect(await messages("finished")).toHaveLength(0) + expect(await messages("failed")).toHaveLength(0) +}) + +it("rolls back retirement if the second mailbox insert fails", async () => { + const effect = await processingEffect() + await ageOwner(70_000) + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime!.settings, + enqueue: (connection, input) => { + if (input.operation === "inspect") throw new Error("injected second insert failure") + return runtime!.repository.enqueueInTransaction(connection, input) + }, + }) + await expect( + runtime!.settings.database.transaction(async (connection) => { + const origin = await lockOrigin({ connection, effect }) + await coordinator.check({ + connection, + origin, + intents: [{ effectId: effect.id, requestId: "rollback" }], + }) + }), + ).rejects.toThrow("injected second insert failure") + expect(await storedEffect(effect.id)).toMatchObject({ status: "processing", claimed_by: "owner" }) + expect(await messages("recover")).toHaveLength(0) + await runtime!.ref(ReportExport, "export").check() + expect((await messages("inspect"))[0]).toMatchObject({ outcome: "retired" }) +}) + +it("reports missing from the owned binding without exposing another actor", async () => { + const effect = await processingEffect() + await runtime!.settings.database.connection((connection) => + connection.run(`DELETE FROM ${runtime!.repository.table("effects")} WHERE id = ?`, [effect.id]), + ) + await runtime!.ref(ReportExport, "export").check() + expect(await messages("inspect")).toEqual([{ effectId: effect.id, outcome: "missing" }]) + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime!.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + await expect( + runtime!.settings.database.transaction((connection) => + coordinator.check({ + connection, + origin: { id: "foreign", actor_type: "ReportExport", actor_id: "foreign" }, + intents: [{ effectId: effect.id, requestId: "foreign" }], + }), + ), + ).rejects.toThrow("owned handle") +}) + +async function createTestRuntime(): Promise { + const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL + const database = connectionString?.startsWith("postgresql:") + ? postgresql({ connectionString, maximumConnections: 8 }) + : connectionString?.startsWith("mysql:") + ? mysql({ connectionString }) + : sqlite({ path: ":memory:" }) + const created = createRuntime({ + database, + tableNamePrefix: "recovery_test_", + authorizeMessage: () => true, + authorizeQuery: () => true, + maxAttempts: 2, + retryDelayMilliseconds: () => 60_000, + }) + created.register(ReportExport) + await created.install() + await created.repository.resetForTesting() + return created +} + +it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( + "locks the origin before the effect when completion races retirement", + async () => { + const effect = await processingEffect() + await ageOwner(70_000) + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime!.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + let completion: Promise = Promise.resolve() + await runtime!.settings.database.transaction(async (connection) => { + const origin = await lockOrigin({ connection, effect }) + const process = await connection.get<{ id: number }>("SELECT pg_backend_pid() AS id") + completion = runtime!.repository + .completeEffect(effect, "late") + .catch((error: unknown) => error) + await waitForBlockedTransaction({ connection, processId: process!.id }) + await coordinator.check({ + connection, + origin, + intents: [{ effectId: effect.id, requestId: "lock-order" }], + }) + }) + expect(await completion).toBeInstanceOf(LostActivation) + expect(await messages("recover")).toHaveLength(1) + expect(await messages("finished")).toHaveLength(0) + }, +) + +it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( + "rechecks a refreshed heartbeat after waiting for the owner lock", + async () => { + const effect = await processingEffect() + await ageOwner(70_000) + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime!.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + let recovery: Promise = Promise.resolve() + await runtime!.settings.database.transaction(async (connection) => { + await connection.get( + `SELECT id FROM ${runtime!.repository.table("processes")} WHERE id = 'owner' FOR UPDATE`, + ) + const process = await connection.get<{ id: number }>("SELECT pg_backend_pid() AS id") + recovery = coordinator.recoverAvailable() + await waitForBlockedTransaction({ connection, processId: process!.id }) + await connection.run( + `UPDATE ${runtime!.repository.table("processes")} SET heartbeat_at_ms = ? WHERE id = 'owner'`, + [await connection.nowMilliseconds()], + ) + }) + await recovery + expect(await storedEffect(effect.id)).toMatchObject({ + status: "processing", + claimed_by: "owner", + }) + expect(await messages("recover")).toHaveLength(0) + }, +) + +it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( + "two blocked automatic recovery passes enqueue one durable callback", + async () => { + const effect = await processingEffect() + await ageOwner(70_000) + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime!.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + let recoveries: Promise = Promise.resolve([]) + await runtime!.settings.database.transaction(async (connection) => { + await lockOrigin({ connection, effect }) + const process = await connection.get<{ id: number }>("SELECT pg_backend_pid() AS id") + recoveries = Promise.all([coordinator.recoverAvailable(), coordinator.recoverAvailable()]) + await waitForBlockedTransaction({ connection, processId: process!.id, count: 2 }) + }) + await recoveries + expect(await messages("recover")).toHaveLength(1) + await runtime!.worker().runUntilIdle() + expect( + (await runtime!.ref(ReportExport, "export").notifications).filter( + (payload) => payload.kind === "recovery", + ), + ).toHaveLength(1) + }, +) + +async function waitForBlockedTransaction(options: { + connection: DatabaseConnection + processId: number + count?: number +}): Promise { + const deadline = Date.now() + 4_000 + while (Date.now() < deadline) { + await options.connection.get("SELECT pg_stat_clear_snapshot()") + const row = await options.connection.get<{ count: string }>( + "WITH RECURSIVE blocked(pid) AS (SELECT pid FROM pg_stat_activity WHERE ? = ANY(pg_blocking_pids(pid)) UNION SELECT activity.pid FROM pg_stat_activity activity JOIN blocked ON blocked.pid = ANY(pg_blocking_pids(activity.pid))) SELECT count(*)::text AS count FROM blocked", + [options.processId], + ) + if (Number(row!.count) >= (options.count ?? 1)) return + } + throw new Error("transaction did not reach the lock barrier") +} + +async function processingEffect( + options: { timeoutMilliseconds?: number } = {}, +): Promise { + runtime = await createTestRuntime() + await runtime.ref(ReportExport, "export").startRecoverable(options) + await runtime.repository.registerProcess("owner", "effect") + const effect = await runtime.repository.claimEffect("owner") + if (!effect) throw new Error("expected a processing effect") + return effect +} + +async function ageOwner(milliseconds: number): Promise { + await runtime!.settings.database.connection(async (connection) => { + const now = await connection.nowMilliseconds() + await connection.run( + `UPDATE ${runtime!.repository.table("processes")} SET heartbeat_at_ms = ? WHERE id = ?`, + [now - milliseconds, "owner"], + ) + }) +} + +function storedEffect(id: string): Promise { + return runtime!.settings.database.connection((connection) => + connection.get( + `SELECT * FROM ${runtime!.repository.table("effects")} WHERE id = ?`, + [id], + ), + ) +} + +async function messages(operation: string): Promise { + const rows = await runtime!.settings.database.connection((connection) => + connection.all<{ arguments: string }>( + `SELECT arguments FROM ${runtime!.repository.table("messages")} WHERE operation = ? ORDER BY sequence`, + [operation], + ), + ) + return rows.map((row) => JSON.parse(row.arguments)) +} + +async function lockOrigin(options: { + connection: DatabaseConnection + effect: EffectRow +}): Promise<{ id: string; actor_type: string; actor_id: string }> { + const origin = await options.connection.get<{ id: string; actor_type: string; actor_id: string }>( + `SELECT id, actor_type, actor_id FROM ${runtime!.repository.table("instances")} WHERE id = ?${runtime!.settings.database.family === "sqlite" ? "" : " FOR UPDATE"}`, + [options.effect.instance_id], + ) + if (!origin) throw new Error("origin missing") + return origin +} diff --git a/test/effect-recovery.types.ts b/test/effect-recovery.types.ts new file mode 100644 index 0000000..6210e94 --- /dev/null +++ b/test/effect-recovery.types.ts @@ -0,0 +1,70 @@ +import { + Actor, + EffectRecoveryOutcome, + type EffectHandle, + type EffectRecoveryPayload, + type EffectRetiredPayload, +} from "../src/index.js" +import { EffectRecoveryOutcome as CoreOutcome } from "../src/core.js" +import { expectTypeOf } from "vitest" + +class ReportExport extends Actor { + start(): EffectHandle { + const handle = this.emit("build_report", { + onRecovery: "recover", + onStatus: "inspect", + recoveryTimeoutMilliseconds: 120_000, + }) + this.requestEffectRecovery(handle) + expectTypeOf(this.emit("plain")).toEqualTypeOf() + // @ts-expect-error Recovery callbacks must name actor operations. + this.emit("build_report", { onRecovery: "recvoer" }) + // @ts-expect-error Status callbacks must name actor operations. + this.emit("build_report", { onStatus: "inspec" }) + // @ts-expect-error A handle requires its effect ID. + this.requestEffectRecovery({}) + return handle + } + + recover(payload: EffectRetiredPayload<{ revision: number }>): void { + expectTypeOf(payload.outcome).toEqualTypeOf<"retired">() + expectTypeOf(payload.arguments.revision).toEqualTypeOf() + // @ts-expect-error Retirement does not provide a success result. + payload.result + } + + inspect(payload: EffectRecoveryPayload<{ revision: number }, string | null>): void { + switch (payload.outcome) { + case EffectRecoveryOutcome.Completed: + expectTypeOf(payload.result).toEqualTypeOf() + expectTypeOf(payload.arguments.revision).toEqualTypeOf() + return + case EffectRecoveryOutcome.Retired: + this.recover(payload) + return + case EffectRecoveryOutcome.Deferred: + case EffectRecoveryOutcome.Pending: + case EffectRecoveryOutcome.Dead: + case EffectRecoveryOutcome.AlreadyRetired: + case EffectRecoveryOutcome.Missing: + // @ts-expect-error Noncompleted observations cannot provide a success result. + payload.result + return + default: + expectTypeOf(payload).toEqualTypeOf() + } + } +} + +expectTypeOf(CoreOutcome.Retired).toEqualTypeOf<"retired">() +// @ts-expect-error Retired payloads require original arguments. +export const invalidRetired: EffectRetiredPayload = { effectId: "id", outcome: "retired" } +// @ts-expect-error Completed observations require a result, including null. +export const invalidCompleted: EffectRecoveryPayload = { + effectId: "id", + outcome: "completed", + arguments: {}, +} +// @ts-expect-error The wire value is retired, never recovered. +export const invalidOutcome: EffectRecoveryPayload = { effectId: "id", outcome: "recovered" } +export { ReportExport } diff --git a/test/fixtures/effect-payload-consumer.mts b/test/fixtures/effect-payload-consumer.mts index 8847bb6..60cf7c7 100644 --- a/test/fixtures/effect-payload-consumer.mts +++ b/test/fixtures/effect-payload-consumer.mts @@ -1,5 +1,34 @@ import type { EffectFailurePayload, EffectSuccessPayload, SerializedError } from "solid-objects" import type { EffectFailurePayload as CoreFailurePayload } from "solid-objects/core" +import { + Actor, + EffectRecoveryOutcome, + type EffectRecoveryPayload, + type EffectRetiredPayload, + type EffectHandle, +} from "solid-objects" + +export function recoveryResult( + payload: EffectRecoveryPayload<{ revision: number }, string>, +): string | null { + if (payload.outcome === EffectRecoveryOutcome.Completed) return payload.result + return null +} + +export class ReportExport extends Actor { + start(): EffectHandle { + return this.emit("build_report", { + onRecovery: "recover", + recoveryTimeoutMilliseconds: 120_000, + }) + } + recover(payload: EffectRetiredPayload<{ revision: number }>): number { + return payload.arguments.revision + } +} + +// @ts-expect-error Published retirement contracts require original arguments. +export const invalidRetirement: EffectRetiredPayload = { effectId: "id", outcome: "retired" } type RunArguments = { generation: number } diff --git a/test/runtime.test.ts b/test/runtime.test.ts index a46d3e2..d24ef84 100644 --- a/test/runtime.test.ts +++ b/test/runtime.test.ts @@ -310,9 +310,11 @@ describe("actor reminders", () => { expect(() => actor.emit("effect", { onFailure: unknownCallback })).toThrow(UnknownOperation) expect(() => actor.emit("effect", { onSuccess: unknownCallback })).toThrow(UnknownOperation) expect(actor.hasIntents()).toBe(false) - expect(actor.emit("effect", { onSuccess: "increment", onFailure: "increment" })).toBeUndefined() + const handle = actor.emit("effect", { onSuccess: "increment", onFailure: "increment" }) + expect(handle.id).toEqual(expect.any(String)) expect(actor.drainIntents().effects).toEqual([ { + id: handle.id, name: "effect", arguments: {}, successOperation: "increment", From 8326f27cb7ddf97bda7887950ec54f76b1ef46f6 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 12:04:14 -0700 Subject: [PATCH 2/7] fix: Bound recovery scans and wake callbacks --- docs/effect-recovery.md | 11 +- examples/failure-recovery/actor.ts | 20 +- examples/failure-recovery/demo.ts | 54 ++++- examples/failure-recovery/worker.ts | 10 +- src/actor.ts | 10 +- src/effect-recovery-coordinator.ts | 27 ++- src/runtime.ts | 20 +- src/wake-up-notification.ts | 19 ++ test/effect-recovery.test.ts | 310 +++++++++++++++++++++++++++- 9 files changed, 434 insertions(+), 47 deletions(-) create mode 100644 src/wake-up-notification.ts diff --git a/docs/effect-recovery.md b/docs/effect-recovery.md index d885ab9..e29d1aa 100644 --- a/docs/effect-recovery.md +++ b/docs/effect-recovery.md @@ -10,8 +10,11 @@ optional recovery/status binding together. `requestEffectRecovery` stages a check on that same transaction connection. It never opens another transaction while an application commit action holds locks. -Automatic polling checks processing effects with `onRecovery` in one independent -transaction per candidate. Candidate reads are hints only. Lock order is origin +Automatic polling checks at most `claimScanLimit` stale candidates per pass, +prefiltering with database time, owner heartbeat, and the effective timeout. It +does not lock fresh owners or their actors, even when the global liveness floor +has elapsed but an effect's extended grace has not. Each candidate gets one +independent transaction; unlocked candidate reads remain hints only. Lock order is origin instance, effects ordered by ID, recovery bindings ordered by effect ID, then current owner processes ordered by ID. Explicit batches acquire all effect and binding locks before any process locks. Completion/failure lock the instance @@ -39,7 +42,9 @@ A winning explicit check additionally enqueues its separate status response, after recovery, keyed by `effect::check:`. Failure of either insertion rolls the transaction back. Multiple checks share one durable retirement, with one response per request. A crash after commit cannot lose the -recovery callback. Wake-up signals are hints; mailbox polling provides delivery. +recovery callback. Successful retirement wakes actor workers after commit. +Wake-up failures are logged without changing the committed decision; mailbox +polling provides delivery. Bindings belong to the exact originating instance, survive effect/message pruning, and cascade when the instance is deleted. They neither pin instances nor diff --git a/examples/failure-recovery/actor.ts b/examples/failure-recovery/actor.ts index c372732..0718f3b 100644 --- a/examples/failure-recovery/actor.ts +++ b/examples/failure-recovery/actor.ts @@ -1,6 +1,24 @@ import { appendFile, access, writeFile } from "node:fs/promises" import { join } from "node:path" -import { Actor } from "solid-objects" +import { Actor, type EffectHandle, type EffectRetiredPayload } from "solid-objects" + +export class RecoverableReport extends Actor { + static override readonly actorType = "RecoverableReport" + exportEffect: EffectHandle | null = null + recoveryCount = 0 + + start(): void { + this.exportEffect = this.emit("build_report", { + arguments: { revision: 1 }, + onRecovery: "recoverExport", + }) + } + + recoverExport(payload: EffectRetiredPayload<{ revision: number }>): void { + if (payload.effectId !== this.exportEffect?.id || payload.arguments.revision !== 1) return + this.recoveryCount += 1 + } +} export class RecoveryCounter extends Actor { static override readonly actorType = "RecoveryCounter" diff --git a/examples/failure-recovery/demo.ts b/examples/failure-recovery/demo.ts index ba0f3a7..491dff8 100644 --- a/examples/failure-recovery/demo.ts +++ b/examples/failure-recovery/demo.ts @@ -7,7 +7,7 @@ import { fileURLToPath } from "node:url" import { fork, type ChildProcess } from "node:child_process" import { createRuntime, type ActorReference, type MessageReference } from "solid-objects" import { sqlite } from "solid-objects/database/sqlite" -import { RecoveryCounter } from "./actor.ts" +import { RecoverableReport, RecoveryCounter } from "./actor.ts" import { assertSerializedExecution, parseSerializationEvent, @@ -46,11 +46,15 @@ const runtime = createRuntime({ try { runtime.register(RecoveryCounter) + runtime.register(RecoverableReport) await runtime.install() const serialization = await proveSerialization() const crash = await proveCrashRecovery() const fencing = await proveFencing() - process.stdout.write(`${JSON.stringify({ serialization, crash, fencing }, null, 2)}\n`) + const effectRecovery = await proveEffectRecovery() + process.stdout.write( + `${JSON.stringify({ serialization, crash, fencing, effectRecovery }, null, 2)}\n`, + ) } finally { await runtime.close() await rm(directory, { recursive: true }) @@ -140,15 +144,51 @@ async function recoveryResult(options: { return { attempts, finalState: snapshot.count, repeatedEffects: effects.length } } -function spawnWorker(): { +async function proveEffectRecovery(): Promise<{ recoveryCallbacks: number }> { + const reference = runtime.ref(RecoverableReport, "report") + await reference.start() + await runtime.repository.registerProcess("abandoned-effect-owner", "effect") + const effect = await runtime.repository.claimEffect("abandoned-effect-owner") + assert.ok(effect) + await runtime.settings.database.connection((connection) => + connection.run( + `UPDATE ${runtime.repository.table("processes")} SET heartbeat_at_ms = 0 WHERE id = ?`, + ["abandoned-effect-owner"], + ), + ) + const retiringWorker = spawnWorker({ mode: "retire-effects" }) + const stopped = retiringWorker.finished.catch(() => undefined) + try { + await Promise.race([ + retiringWorker.waitFor((message) => message.event === "effects.retired"), + retiringWorker.finished.then(() => { + throw new Error("worker exited before retirement") + }), + ]) + } finally { + retiringWorker.child.kill("SIGKILL") + await stopped + } + assert.equal((await reference.snapshot()).recoveryCount, 0) + await spawnWorker().finished + assert.equal((await reference.snapshot()).recoveryCount, 1) + assert.equal(await runtime.repository.claimEffect("abandoned-effect-owner"), undefined) + return { recoveryCallbacks: 1 } +} + +function spawnWorker(options: { mode?: "retire-effects" } = {}): { child: ChildProcess finished: Promise waitFor(predicate: (message: WorkerMessage) => boolean): Promise } { - const child = fork(fileURLToPath(new URL("./worker.ts", import.meta.url)), [databasePath], { - cwd: fileURLToPath(new URL("../..", import.meta.url)), - stdio: ["ignore", "pipe", "pipe", "ipc"], - }) + const child = fork( + fileURLToPath(new URL("./worker.ts", import.meta.url)), + [databasePath, ...(options.mode ? [options.mode] : [])], + { + cwd: fileURLToPath(new URL("../..", import.meta.url)), + stdio: ["ignore", "pipe", "pipe", "ipc"], + }, + ) const messages: WorkerMessage[] = [] const listeners = new Set<(message: WorkerMessage) => void>() let stderr = "" diff --git a/examples/failure-recovery/worker.ts b/examples/failure-recovery/worker.ts index 44d8914..12c48ea 100644 --- a/examples/failure-recovery/worker.ts +++ b/examples/failure-recovery/worker.ts @@ -1,6 +1,6 @@ import { createRuntime } from "solid-objects" import { sqlite } from "solid-objects/database/sqlite" -import { RecoveryCounter } from "./actor.ts" +import { RecoverableReport, RecoveryCounter } from "./actor.ts" const databasePath = requiredArgument(2) const runtime = createRuntime({ @@ -24,7 +24,15 @@ const runtime = createRuntime({ }) runtime.register(RecoveryCounter) +runtime.register(RecoverableReport) await runtime.install() +if (process.argv[3] === "retire-effects") { + await runtime.repository.cleanupStaleProcesses() + process.send?.({ event: "effects.retired" }) + await new Promise(() => { + process.on("message", () => {}) + }) +} const worker = runtime.worker() try { diff --git a/src/actor.ts b/src/actor.ts index 769eb5c..336c66d 100644 --- a/src/actor.ts +++ b/src/actor.ts @@ -268,12 +268,10 @@ export abstract class Actor { } } const timeout = options.recoveryTimeoutMilliseconds - if (timeout !== undefined) { - if (!Number.isSafeInteger(timeout) || timeout <= 0) - throw new TypeError("recoveryTimeoutMilliseconds must be a positive safe integer") - if (options.onRecovery === undefined) - throw new TypeError("recoveryTimeoutMilliseconds requires onRecovery") - } + if (timeout !== undefined && (!Number.isSafeInteger(timeout) || timeout <= 0)) + throw new TypeError("recoveryTimeoutMilliseconds must be a positive safe integer") + if (timeout !== undefined && options.onRecovery === undefined) + throw new TypeError("recoveryTimeoutMilliseconds requires onRecovery") const id = randomUUID() this.#intents.effects.push({ id, diff --git a/src/effect-recovery-coordinator.ts b/src/effect-recovery-coordinator.ts index a2ef95b..b6a4574 100644 --- a/src/effect-recovery-coordinator.ts +++ b/src/effect-recovery-coordinator.ts @@ -9,6 +9,7 @@ import { } from "./effect-recovery.js" import type { EffectRow, EnqueueInput, MessageRow, ProcessRow } from "./records.js" import { jsonObject, normalizeJson } from "./serialization.js" +import { notifyWakeUp } from "./wake-up-notification.js" interface RecoveryBinding { effect_id: string @@ -34,16 +35,23 @@ export class EffectRecoveryCoordinator { ) {} async recoverAvailable(): Promise { - const candidates = await this.options.settings.database.connection((connection) => - connection.all( + const candidates = await this.options.settings.database.connection(async (connection) => { + const now = await connection.nowMilliseconds() + const threshold = this.options.settings.processAliveThresholdMilliseconds + return connection.all( `SELECT recoveries.* FROM ${this.table("effect_recoveries")} recoveries JOIN ${this.table("effects")} effects ON effects.id = recoveries.effect_id + LEFT JOIN ${this.table("processes")} owners ON owners.id = effects.claimed_by WHERE recoveries.retired_at_ms IS NULL AND recoveries.recovery_operation IS NOT NULL - AND effects.status = 'processing' ORDER BY recoveries.instance_id, recoveries.effect_id`, - ), - ) + AND effects.status = 'processing' + AND (owners.id IS NULL OR owners.heartbeat_at_ms <= ? - + CASE WHEN recoveries.recovery_timeout_ms > ? THEN recoveries.recovery_timeout_ms ELSE ? END) + ORDER BY recoveries.instance_id, recoveries.effect_id LIMIT ?`, + [now, threshold, threshold, this.options.settings.claimScanLimit], + ) + }) for (const candidate of candidates) { - await this.options.settings.database.transaction(async (connection) => { + const retired = await this.options.settings.database.transaction(async (connection) => { const origin = await connection.get( `SELECT id, actor_type, actor_id FROM ${this.table("instances")} WHERE id = ?${this.lockClause()}`, [candidate.instance_id], @@ -75,7 +83,14 @@ export class EffectRecoveryCoordinator { const now = await connection.nowMilliseconds() if (this.ownerFresh({ binding, owner, now })) return await this.retire({ connection, origin, binding, effect, now }) + return true }) + if (retired) + notifyWakeUp({ + adapter: this.options.settings.wakeUp, + logger: this.options.settings.logger, + role: "actors", + }) } } diff --git a/src/runtime.ts b/src/runtime.ts index 6c557bb..26ed144 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -122,6 +122,7 @@ import { SolidObjectsTestHelper } from "./test-helper.js" import { waitFor, Worker } from "./worker.js" import { EffectWorker } from "./effect-worker.js" import type { WakeUpRole } from "./wake-up.js" +import { notifyWakeUp } from "./wake-up-notification.js" import { withDatabaseDeadline } from "./database/deadline.js" import type { DatabaseConnection } from "./database/types.js" import { evaluateActorTurn, readActorObservables } from "./turn.js" @@ -1210,7 +1211,8 @@ export class SolidObjectsRuntime { } }, }) - if (intents.outboundMessages.length > 0) this.wakeUp("actors") + if (intents.outboundMessages.length > 0 || (intents.effectRecoveries?.length ?? 0) > 0) + this.wakeUp("actors") if (intents.effects.length > 0) this.wakeUp("effects") if (intents.reminders.length > 0) this.wakeUp("reminders") if (broadcastProjectionValue !== undefined) this.wakeUp("broadcasts") @@ -1933,21 +1935,7 @@ export class SolidObjectsRuntime { } private wakeUp(role: WakeUpRole): void { - try { - Promise.resolve(this.settings.wakeUp.notify(role)).catch((error: unknown) => { - this.logWakeUpFailure(role, error) - }) - } catch (error) { - this.logWakeUpFailure(role, error) - } - } - - private logWakeUpFailure(role: WakeUpRole, error: unknown): void { - this.settings.logger.error({ - event: "solid_objects.wake_up.failed", - role, - errorName: error instanceof Error ? error.name : "Error", - }) + notifyWakeUp({ adapter: this.settings.wakeUp, logger: this.settings.logger, role }) } private async authorize(options: { diff --git a/src/wake-up-notification.ts b/src/wake-up-notification.ts new file mode 100644 index 0000000..604ef18 --- /dev/null +++ b/src/wake-up-notification.ts @@ -0,0 +1,19 @@ +import type { Logger } from "./types.js" +import type { WakeUpAdapter, WakeUpRole } from "./wake-up.js" + +export function notifyWakeUp(options: { + adapter: WakeUpAdapter + logger: Logger + role: WakeUpRole +}): void { + const logFailure = (errorName: string): void => { + options.logger.error({ event: "solid_objects.wake_up.failed", role: options.role, errorName }) + } + try { + Promise.resolve(options.adapter.notify(options.role)).catch((error) => + logFailure(error instanceof Error ? error.name : "Error"), + ) + } catch (error) { + logFailure(error instanceof Error ? error.name : "Error") + } +} diff --git a/test/effect-recovery.test.ts b/test/effect-recovery.test.ts index 71a1a36..acce136 100644 --- a/test/effect-recovery.test.ts +++ b/test/effect-recovery.test.ts @@ -11,10 +11,13 @@ import type { JsonObject, } from "../src/types.js" import type { EffectRecoveryPayload, EffectRetiredPayload } from "../src/effect-recovery.js" -import { LostActivation } from "../src/errors.js" +import { LostActivation, MailboxFull } from "../src/errors.js" import { EffectRecoveryCoordinator } from "../src/effect-recovery-coordinator.js" import type { DatabaseConnection } from "../src/database/types.js" import type { EffectRow } from "../src/records.js" +import { Repository } from "../src/repository.js" +import { PausingClaimDatabase } from "./support/pausing-claim-database.js" +import { deferred } from "./support/fenced-commit.js" class ReportExport extends Actor { static override readonly actorType = "ReportExport" @@ -41,6 +44,13 @@ class ReportExport extends Actor { check(): void { if (this.exportEffect) this.requestEffectRecovery(this.exportEffect) } + + startStatusOnly(): void { + this.exportEffect = this.emit("build_report", { + arguments: { revision: 1 }, + onStatus: "inspect", + }) + } recover(payload: EffectRetiredPayload): void { this.notifications.push({ kind: "recovery", ...payload }) } @@ -59,6 +69,7 @@ let runtime: SolidObjectsRuntime | undefined afterEach(async () => { await runtime?.repository.resetForTesting() await runtime?.close() + runtime = undefined }) it("returns the same effect identity that the scheduler claims", async () => { @@ -82,7 +93,9 @@ it("retires abandoned processing effects before the scheduler can reclaim them", ), ) await runtime.repository.registerProcess("replacement", "effect") + const wakeUp = await runtime.settings.wakeUp.watch("actors") expect(await runtime.repository.claimEffect("replacement")).toBeUndefined() + expect(await wakeUp.wait({ timeoutMilliseconds: 0 })).toBe(true) const notifications = await runtime.settings.database.connection((connection) => connection.all<{ arguments: string }>( `SELECT arguments FROM ${runtime!.repository.table("messages")} WHERE operation = 'recover'`, @@ -109,8 +122,8 @@ it("defers indefinitely for a fresh heartbeat without consuming an attempt", asy expect(await storedEffect(effect.id)).toMatchObject({ status: "processing", claimed_by: "owner", - attempt_count: 1n, }) + expect(Number((await storedEffect(effect.id))!.attempt_count)).toBe(1) }) it("preserves a longer grace after ordinary cleanup stops the owner", async () => { @@ -129,7 +142,8 @@ it("keeps pending retries with the existing scheduler", async () => { await runtime!.repository.failEffect({ effect, error: new Error("retry"), retryable: true }) await runtime!.ref(ReportExport, "export").check() expect((await messages("inspect"))[0]).toMatchObject({ effectId: effect.id, outcome: "pending" }) - expect(await storedEffect(effect.id)).toMatchObject({ status: "pending", attempt_count: 1n }) + expect(await storedEffect(effect.id)).toMatchObject({ status: "pending" }) + expect(Number((await storedEffect(effect.id))!.attempt_count)).toBe(1) expect(await messages("recover")).toHaveLength(0) }) @@ -226,6 +240,100 @@ it("reports missing from the owned binding without exposing another actor", asyn ).rejects.toThrow("owned handle") }) +it("rolls back both callbacks when only one mailbox slot remains", async () => { + const effect = await processingEffect() + await ageOwner(70_000) + runtime!.settings.maxMailboxLength = 1 + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime!.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + await expect( + runtime!.settings.database.transaction(async (connection) => { + const origin = await lockOrigin({ connection, effect }) + await coordinator.check({ + connection, + origin, + intents: [{ effectId: effect.id, requestId: "full" }], + }) + }), + ).rejects.toBeInstanceOf(MailboxFull) + expect(await storedEffect(effect.id)).toMatchObject({ status: "processing", claimed_by: "owner" }) + expect(await messages("recover")).toHaveLength(0) + expect(await messages("inspect")).toHaveLength(0) +}) + +it("surfaces an owner query failure without deciding abandonment", async () => { + const effect = await processingEffect() + await ageOwner(70_000) + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime!.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + await expect( + runtime!.settings.database.transaction(async (connection) => { + const origin = await lockOrigin({ connection, effect }) + const failingConnection: DatabaseConnection = { + run: connection.run.bind(connection), + all: connection.all.bind(connection), + nowMilliseconds: connection.nowMilliseconds.bind(connection), + get: (sql: string, parameters?: readonly unknown[]) => + connection.get( + sql.includes(runtime!.repository.table("processes")) + ? `SELECT absent_recovery_column FROM ${runtime!.repository.table("processes")}` + : sql, + parameters, + ), + } + await coordinator.check({ + connection: failingConnection, + origin, + intents: [{ effectId: effect.id, requestId: "lookup-error" }], + }) + }), + ).rejects.toThrow() + expect(await storedEffect(effect.id)).toMatchObject({ status: "processing", claimed_by: "owner" }) + expect(await messages("recover")).toHaveLength(0) + expect(await messages("inspect")).toHaveLength(0) +}) + +it.each([0, -1, Infinity, NaN, 1.5, Number.MAX_SAFE_INTEGER + 1])( + "rejects invalid recovery timeout %s before staging", + (timeout) => { + const actor = new ReportExport("invalid") + actor.prepare(new Set(["recover", "inspect", "finished", "failed"])) + expect(() => actor.startRecoverable({ timeoutMilliseconds: timeout })).toThrow(TypeError) + expect(actor.hasIntents()).toBe(false) + }, +) + +it("requires recovery opt-in for a timeout and both bindings for an explicit check", async () => { + const actor = new ReportExport("invalid") + expect(() => actor.emit("build_report", { recoveryTimeoutMilliseconds: 120_000 })).toThrow( + "requires onRecovery", + ) + runtime = await createTestRuntime() + await runtime.ref(ReportExport, "export").start() + const effect = await runtime.settings.database.connection((connection) => + connection.get(`SELECT * FROM ${runtime!.repository.table("effects")}`), + ) + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + await expect( + runtime.settings.database.transaction(async (connection) => { + const origin = await lockOrigin({ connection, effect: effect! }) + await coordinator.check({ + connection, + origin, + intents: [{ effectId: effect!.id, requestId: "unbound" }], + }) + }), + ).rejects.toThrow("onRecovery and onStatus") + expect(await storedEffect(effect!.id)).toMatchObject({ status: "pending" }) +}) + async function createTestRuntime(): Promise { const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL const database = connectionString?.startsWith("postgresql:") @@ -247,6 +355,48 @@ async function createTestRuntime(): Promise { return created } +it("does not opt a status-only effect into retirement", async () => { + runtime = await createTestRuntime() + await runtime.ref(ReportExport, "export").startStatusOnly() + await runtime.repository.registerProcess("owner", "effect") + const effect = await runtime.repository.claimEffect("owner") + await ageOwner(70_000) + await runtime.repository.registerProcess("new-owner", "effect") + const reclaimed = await runtime.repository.claimEffect("new-owner") + expect(reclaimed?.id).toBe(effect?.id) + expect(reclaimed?.claimed_by).toBe("new-owner") + expect(await messages("recover")).toHaveLength(0) + expect(await messages("inspect")).toHaveLength(0) +}) + +it("applies the current runtime floor and handles a missing processing owner", async () => { + const effect = await processingEffect({ timeoutMilliseconds: 1 }) + await ageOwner(70_000) + runtime!.settings.processAliveThresholdMilliseconds = 120_000 + expect(await runtime!.repository.claimEffect("owner")).toBeUndefined() + expect(await storedEffect(effect.id)).toMatchObject({ status: "processing", claimed_by: "owner" }) + await runtime!.settings.database.connection((connection) => + connection.run( + `UPDATE ${runtime!.repository.table("effects")} SET claimed_by = NULL WHERE id = ?`, + [effect.id], + ), + ) + expect(await runtime!.repository.claimEffect("owner")).toBeUndefined() + expect(await messages("recover")).toHaveLength(1) +}) + +it("bounds each automatic pass and revisits remaining abandoned effects", async () => { + await processingEffect() + await runtime!.ref(ReportExport, "other").startRecoverable() + expect(await runtime!.repository.claimEffect("owner")).toBeDefined() + await ageOwner(70_000) + runtime!.settings.claimScanLimit = 1 + expect(await runtime!.repository.claimEffect("owner")).toBeUndefined() + expect(await messages("recover")).toHaveLength(1) + expect(await runtime!.repository.claimEffect("owner")).toBeUndefined() + expect(await messages("recover")).toHaveLength(2) +}) + it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( "locks the origin before the effect when completion races retirement", async () => { @@ -256,13 +406,14 @@ it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( settings: runtime!.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) - let completion: Promise = Promise.resolve() + let completion: Promise = Promise.resolve() await runtime!.settings.database.transaction(async (connection) => { const origin = await lockOrigin({ connection, effect }) const process = await connection.get<{ id: number }>("SELECT pg_backend_pid() AS id") - completion = runtime!.repository - .completeEffect(effect, "late") - .catch((error: unknown) => error) + completion = runtime!.repository.completeEffect(effect, "late").catch((error) => { + if (!(error instanceof Error)) throw new TypeError("completion rejected without an Error") + return error + }) await waitForBlockedTransaction({ connection, processId: process!.id }) await coordinator.check({ connection, @@ -362,6 +513,44 @@ async function processingEffect( return effect } +it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( + "claiming pending work does not wait on a fresh recovery candidate", + async () => { + const effect = await processingEffect({ timeoutMilliseconds: 120_000 }) + await ageOwner(75_000) + await runtime!.ref(ReportExport, "other").start() + await runtime!.repository.registerProcess("other-owner", "effect") + let claim: Promise = Promise.resolve(undefined) + try { + await runtime!.settings.database.transaction(async (connection) => { + await lockOrigin({ connection, effect }) + claim = runtime!.repository.claimEffect("other-owner") + const selected = await withDeadline(claim) + expect(selected?.actor_id).toBe("other") + }) + } finally { + await claim + } + }, +) + +async function withDeadline(promise: Promise): Promise { + let timer: ReturnType | undefined + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error("independent work blocked on a fresh recovery candidate")), + 2_000, + ) + }), + ]) + } finally { + clearTimeout(timer) + } +} + async function ageOwner(milliseconds: number): Promise { await runtime!.settings.database.connection(async (connection) => { const now = await connection.nowMilliseconds() @@ -372,6 +561,113 @@ async function ageOwner(milliseconds: number): Promise { }) } +it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( + "rechecks ownership when a pending claim wins before the recovery check", + async () => { + runtime = await createTestRuntime() + await runtime.ref(ReportExport, "export").startRecoverable() + const handle = await runtime.ref(ReportExport, "export").exportEffect + const effect = await storedEffect(handle!.id) + await runtime.repository.registerProcess("owner", "effect") + const pausedDatabase = new PausingClaimDatabase({ + database: runtime.settings.database, + table: "effects", + }) + const claimant = new Repository({ ...runtime.settings, database: pausedDatabase }) + const claim = claimant.claimEffect("owner") + const originLocked = deferred() + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + try { + await pausedDatabase.waitUntilClaimLocked() + const recovery = runtime.settings.database.transaction(async (connection) => { + const origin = await lockOrigin({ connection, effect: effect! }) + originLocked.resolve() + await coordinator.check({ + connection, + origin, + intents: [{ effectId: effect!.id, requestId: "claim-wins" }], + }) + }) + await originLocked.promise + pausedDatabase.resume() + await Promise.all([claim, recovery]) + expect((await messages("inspect"))[0]).toMatchObject({ outcome: "deferred" }) + expect(await messages("recover")).toHaveLength(0) + } finally { + pausedDatabase.resume() + await claim + } + }, +) + +it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( + "leaves a pending effect claimable after the recovery check wins", + async () => { + runtime = await createTestRuntime() + await runtime.ref(ReportExport, "export").startRecoverable() + const handle = await runtime.ref(ReportExport, "export").exportEffect + const effect = await storedEffect(handle!.id) + await runtime.repository.registerProcess("owner", "effect") + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + await runtime.settings.database.transaction(async (connection) => { + const origin = await lockOrigin({ connection, effect: effect! }) + await coordinator.check({ + connection, + origin, + intents: [{ effectId: effect!.id, requestId: "check-wins" }], + }) + expect(await runtime!.repository.claimEffect("owner")).toBeUndefined() + }) + expect((await runtime.repository.claimEffect("owner"))?.id).toBe(effect!.id) + expect((await messages("inspect"))[0]).toMatchObject({ outcome: "pending" }) + expect(await messages("recover")).toHaveLength(0) + }, +) + +it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( + "concurrent explicit checks and a replay share one retirement", + async () => { + const effect = await processingEffect() + await ageOwner(70_000) + const coordinator = new EffectRecoveryCoordinator({ + settings: runtime!.settings, + enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), + }) + const check = (requestId: string) => + runtime!.settings.database.transaction(async (connection) => { + const origin = await lockOrigin({ connection, effect }) + await coordinator.check({ + connection, + origin, + intents: [{ effectId: effect.id, requestId }], + }) + }) + let checks: Promise = Promise.resolve([]) + try { + await runtime!.settings.database.transaction(async (connection) => { + await lockOrigin({ connection, effect }) + const process = await connection.get<{ id: number }>("SELECT pg_backend_pid() AS id") + checks = Promise.all([check("one"), check("two")]) + await waitForBlockedTransaction({ connection, processId: process!.id, count: 2 }) + }) + } finally { + await checks + } + await check("one") + expect(await messages("recover")).toHaveLength(1) + expect((await messages("inspect")).map((payload) => payload.outcome)).toEqual([ + "retired", + "alreadyRetired", + ]) + }, +) + function storedEffect(id: string): Promise { return runtime!.settings.database.connection((connection) => connection.get( From 38f021657eb46981a155ede5c2ca00ac1cb1d7c5 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 12:10:46 -0700 Subject: [PATCH 3/7] test: Reuse database parameter types --- test/effect-recovery.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/effect-recovery.test.ts b/test/effect-recovery.test.ts index acce136..2040c70 100644 --- a/test/effect-recovery.test.ts +++ b/test/effect-recovery.test.ts @@ -277,7 +277,10 @@ it("surfaces an owner query failure without deciding abandonment", async () => { run: connection.run.bind(connection), all: connection.all.bind(connection), nowMilliseconds: connection.nowMilliseconds.bind(connection), - get: (sql: string, parameters?: readonly unknown[]) => + get: ( + sql: string, + parameters?: Parameters[1], + ) => connection.get( sql.includes(runtime!.repository.table("processes")) ? `SELECT absent_recovery_column FROM ${runtime!.repository.table("processes")}` From 65e21c0cb9a4e034844ec7e908602edff11494c8 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 12:15:45 -0700 Subject: [PATCH 4/7] chore: Prepare version 0.15.0 --- CHANGELOG.md | 2 +- package.json | 2 +- src/version.ts | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ce29789..b963a3b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## Unreleased +## 0.15.0 - 2026-09-15 - Return a stable `EffectHandle` from every `emit`. Overrides/wrappers must return the handle; code expecting `void`/`undefined` needs migration. diff --git a/package.json b/package.json index be4fcf4..99436e2 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "solid-objects", - "version": "0.14.9", + "version": "0.15.0", "description": "Race-free realtime state per application identity, backed by your SQL database", "type": "module", "license": "MIT", diff --git a/src/version.ts b/src/version.ts index f5bce34..92d00d6 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const VERSION = "0.14.9" +export const VERSION = "0.15.0" From c677b4ba091ee423924d7e0c55fe40621df971cc Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 12:20:19 -0700 Subject: [PATCH 5/7] test: Flatten recovery adapter selection --- test/effect-recovery.test.ts | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/test/effect-recovery.test.ts b/test/effect-recovery.test.ts index 2040c70..e9fb76d 100644 --- a/test/effect-recovery.test.ts +++ b/test/effect-recovery.test.ts @@ -13,7 +13,7 @@ import type { import type { EffectRecoveryPayload, EffectRetiredPayload } from "../src/effect-recovery.js" import { LostActivation, MailboxFull } from "../src/errors.js" import { EffectRecoveryCoordinator } from "../src/effect-recovery-coordinator.js" -import type { DatabaseConnection } from "../src/database/types.js" +import type { Database, DatabaseConnection } from "../src/database/types.js" import type { EffectRow } from "../src/records.js" import { Repository } from "../src/repository.js" import { PausingClaimDatabase } from "./support/pausing-claim-database.js" @@ -337,13 +337,16 @@ it("requires recovery opt-in for a timeout and both bindings for an explicit che expect(await storedEffect(effect!.id)).toMatchObject({ status: "pending" }) }) -async function createTestRuntime(): Promise { +function createTestDatabase(): Database { const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL - const database = connectionString?.startsWith("postgresql:") - ? postgresql({ connectionString, maximumConnections: 8 }) - : connectionString?.startsWith("mysql:") - ? mysql({ connectionString }) - : sqlite({ path: ":memory:" }) + if (connectionString?.startsWith("postgresql:")) + return postgresql({ connectionString, maximumConnections: 8 }) + if (connectionString?.startsWith("mysql:")) return mysql({ connectionString }) + return sqlite({ path: ":memory:" }) +} + +async function createTestRuntime(): Promise { + const database = createTestDatabase() const created = createRuntime({ database, tableNamePrefix: "recovery_test_", From 81693fdee7e6ca40c6c6a04f6062a55afc6d3cda Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 13:08:08 -0700 Subject: [PATCH 6/7] fix: Resume heartbeats after database errors --- CHANGELOG.md | 2 ++ docs/api.md | 5 ++++ docs/parity.md | 3 +++ src/worker.ts | 11 +++++++- test/effect-recovery.test.ts | 51 ++++++++++++++++++++++++++++++++---- 5 files changed, 66 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b963a3b..a73b166 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## 0.15.0 - 2026-09-15 +- Report transient process-heartbeat errors and keep retrying while effects run, + so one failed update cannot permanently disable owner freshness maintenance. - Return a stable `EffectHandle` from every `emit`. Overrides/wrappers must return the handle; code expecting `void`/`undefined` needs migration. - Add SQL effect recovery coordination with `onRecovery`, optional `onStatus`, diff --git a/docs/api.md b/docs/api.md index da4b66d..7de0fe7 100644 --- a/docs/api.md +++ b/docs/api.md @@ -185,6 +185,11 @@ override and the runtime's current `processAliveThresholdMilliseconds` (default owner's database heartbeat, not effect duration or progress. An owner that keeps heartbeating protects its effect indefinitely. +Heartbeat update errors emit `solid_objects.process.heartbeat_failed` and retry +at the configured interval without consuming effect attempts. An outage lasting +beyond the freshness window can still permit recovery; this does not cancel +external work or extend the timeout. + `EffectRetiredPayload` is the `onRecovery` envelope: `effectId`, original `arguments`, and `outcome: EffectRecoveryOutcome.Retired`. No outcome guard is needed in that callback. `EffectRecoveryPayload` is the diff --git a/docs/parity.md b/docs/parity.md index 7f14be9..69379e7 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -220,6 +220,9 @@ same authorization and CSRF boundaries. ## Effect recovery +Both runtimes maintain heartbeats during effect execution and retry failed +updates at the configured interval, reporting `process.heartbeat_failed`. + | Capability | Status | Contract | | ----------------------------------------------- | -------------- | ------------------------------------------------------------------------------------------------------------------------------------ | | Abandoned SQL effect recovery | Native | Stable emit handles, automatic retirement, staged status checks, durable callbacks, and extending heartbeat grace in both languages. | diff --git a/src/worker.ts b/src/worker.ts index 8bb05f0..c1e5e0a 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -332,6 +332,15 @@ async function heartbeatUntilStopped(options: { while (!signal.aborted) { await waitFor(runtime.settings.processHeartbeatIntervalMilliseconds, signal) if (signal.aborted) return - await runtime.repository.heartbeatProcess(processId) + try { + await runtime.repository.heartbeatProcess(processId) + } catch (error) { + const attributes = { processId, errorName: error instanceof Error ? error.name : "Error" } + runtime.settings.logger.warn({ + event: "solid_objects.process.heartbeat_failed", + ...attributes, + }) + runtime.emitInstrumentation("process.heartbeat_failed", attributes) + } } } diff --git a/test/effect-recovery.test.ts b/test/effect-recovery.test.ts index e9fb76d..ea094b4 100644 --- a/test/effect-recovery.test.ts +++ b/test/effect-recovery.test.ts @@ -18,6 +18,7 @@ import type { EffectRow } from "../src/records.js" import { Repository } from "../src/repository.js" import { PausingClaimDatabase } from "./support/pausing-claim-database.js" import { deferred } from "./support/fenced-commit.js" +import { withProcessHeartbeat } from "../src/worker.js" class ReportExport extends Actor { static override readonly actorType = "ReportExport" @@ -72,6 +73,46 @@ afterEach(async () => { runtime = undefined }) +it("resumes process heartbeats after a database error", async () => { + runtime = await createTestRuntime() + runtime.settings.processHeartbeatIntervalMilliseconds = 10 + const resumed = deferred() + const release = deferred() + const events: string[] = [] + runtime.settings.instrumentation = ({ name }) => events.push(name) + await runtime.repository.registerProcess("transient-owner", "effect") + const heartbeatProcess = runtime.repository.heartbeatProcess.bind(runtime.repository) + let attempts = 0 + runtime.repository.heartbeatProcess = async (processId) => { + attempts += 1 + if (attempts === 1) + await runtime!.settings.database.connection((connection) => + connection.run( + `SELECT absent_heartbeat_column FROM ${runtime!.repository.table("processes")}`, + ), + ) + await heartbeatProcess(processId) + resumed.resolve() + } + const running = withProcessHeartbeat({ + runtime, + processId: "transient-owner", + operation: () => release.promise, + }).catch((error) => { + if (error instanceof Error) return error + throw error + }) + try { + await withDeadline(resumed.promise, "heartbeat did not resume after the database error") + } finally { + release.resolve() + await running + } + expect(await running).toBeUndefined() + expect(attempts).toBeGreaterThanOrEqual(2) + expect(events.filter((name) => name === "solid_objects.process.heartbeat_failed")).toHaveLength(1) +}) + it("returns the same effect identity that the scheduler claims", async () => { runtime = await createTestRuntime() await runtime.ref(ReportExport, "export").start() @@ -540,16 +581,16 @@ it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( }, ) -async function withDeadline(promise: Promise): Promise { +async function withDeadline( + promise: Promise, + message = "independent work blocked on a fresh recovery candidate", +): Promise { let timer: ReturnType | undefined try { return await Promise.race([ promise, new Promise((_, reject) => { - timer = setTimeout( - () => reject(new Error("independent work blocked on a fresh recovery candidate")), - 2_000, - ) + timer = setTimeout(() => reject(new Error(message)), 2_000) }), ]) } finally { From 9de7c2ccdc1e8b21c79571ad057300876778211b Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Tue, 15 Sep 2026 13:14:10 -0700 Subject: [PATCH 7/7] test: Allow time for cold MySQL schema setup --- test/effect-recovery.test.ts | 2 +- test/mysql.test.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/test/effect-recovery.test.ts b/test/effect-recovery.test.ts index ea094b4..23202c0 100644 --- a/test/effect-recovery.test.ts +++ b/test/effect-recovery.test.ts @@ -111,7 +111,7 @@ it("resumes process heartbeats after a database error", async () => { expect(await running).toBeUndefined() expect(attempts).toBeGreaterThanOrEqual(2) expect(events.filter((name) => name === "solid_objects.process.heartbeat_failed")).toHaveLength(1) -}) +}, 30_000) it("returns the same effect identity that the scheduler claims", async () => { runtime = await createTestRuntime() diff --git a/test/mysql.test.ts b/test/mysql.test.ts index 933131c..01f06f5 100644 --- a/test/mysql.test.ts +++ b/test/mysql.test.ts @@ -462,7 +462,7 @@ describeMySQL("MySQL adapter", () => { await localDatabase.close() await serverDatabase.close() } - }) + }, 30_000) it("enforces deadlines without leaking session settings", async () => { if (!connectionString) throw new Error("MySQL connection string is required")