Merge pull request #51 from cardmagic/feat/effect-recovery

Coordinate abandoned effect recovery

Author: cardmagic

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

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<Arguments>` is the `onRecovery` envelope: `effectId`, original
+`arguments`, and `outcome: EffectRecoveryOutcome.Retired`. No outcome guard is
+needed in that callback. `EffectRecoveryPayload<Arguments, Result>` 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<ReportArguments>): void {
+    if (payload.effectId !== this.exportEffect?.id || payload.arguments.revision !== this.revision)
+      return
+    this.start()
+  }
+
+  exportFinished(payload: EffectSuccessPayload<ReportArguments, ReportResult>): void {
+    this.applyExportResult(payload)
+  }
+
+  exportFailed(_payload: JsonValue): void {}
+
+  inspectExport(payload: EffectRecoveryPayload<ReportArguments, ReportResult>): 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<ReportArguments, ReportResult>): 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

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:<id>:recovery` mailbox insertion are atomic.
+A winning explicit check additionally enqueues its separate status response,
+after recovery, keyed by `effect:<id>:check:<internal-request-id>`. 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.

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).

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"

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<void>
   waitFor(predicate: (message: WorkerMessage) => boolean): Promise<WorkerMessage>
 } {
-  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 = ""

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<void>(() => {
+    process.on("message", () => {})
+  })
+}
 const worker = runtime.worker()
 
 try {

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",