diff --git a/CHANGELOG.md b/CHANGELOG.md index 8365c93..a73b166 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 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`, + 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..7de0fe7 100644 --- a/docs/api.md +++ b/docs/api.md @@ -162,6 +162,145 @@ 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. + +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 +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..e29d1aa --- /dev/null +++ b/docs/effect-recovery.md @@ -0,0 +1,59 @@ +# 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 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 +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. 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 +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..69379e7 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -217,3 +217,21 @@ 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 + +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. | +| 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/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/package.json b/package.json index 49b80fb..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", @@ -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..336c66d 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,61 @@ 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 && (!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, 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 +402,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..b6a4574 --- /dev/null +++ b/src/effect-recovery-coordinator.ts @@ -0,0 +1,251 @@ +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" +import { notifyWakeUp } from "./wake-up-notification.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(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' + 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) { + 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], + ) + 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 }) + return true + }) + if (retired) + notifyWakeUp({ + adapter: this.options.settings.wakeUp, + logger: this.options.settings.logger, + role: "actors", + }) + } + } + + 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/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/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/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" 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/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/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..23202c0 --- /dev/null +++ b/test/effect-recovery.test.ts @@ -0,0 +1,747 @@ +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, MailboxFull } from "../src/errors.js" +import { EffectRecoveryCoordinator } from "../src/effect-recovery-coordinator.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" +import { deferred } from "./support/fenced-commit.js" +import { withProcessHeartbeat } from "../src/worker.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) + } + + startStatusOnly(): void { + this.exportEffect = this.emit("build_report", { + arguments: { revision: 1 }, + onStatus: "inspect", + }) + } + 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() + 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) +}, 30_000) + +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") + 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'`, + ), + ) + 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", + }) + expect(Number((await storedEffect(effect.id))!.attempt_count)).toBe(1) +}) + +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" }) + expect(Number((await storedEffect(effect.id))!.attempt_count)).toBe(1) + 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") +}) + +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?: Parameters[1], + ) => + 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" }) +}) + +function createTestDatabase(): Database { + const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL + 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_", + authorizeMessage: () => true, + authorizeQuery: () => true, + maxAttempts: 2, + retryDelayMilliseconds: () => 60_000, + }) + created.register(ReportExport) + await created.install() + await created.repository.resetForTesting() + 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 () => { + 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) => { + if (!(error instanceof Error)) throw new TypeError("completion rejected without an Error") + return 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 +} + +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, + 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(message)), 2_000) + }), + ]) + } finally { + clearTimeout(timer) + } +} + +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"], + ) + }) +} + +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( + `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/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") 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",