From 2ae893072992d0de93dbc6d68a717558d68f738c Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 09:49:59 -0700 Subject: [PATCH 1/9] fix: isolate caller worker lifecycles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Give overlapping synchronous callers separate reusable workers so one pass cannot discard another turn’s staged intents. Refresh each runtime role’s process registration before claiming work after stale cleanup or retention. --- CHANGELOG.md | 12 ++ docs/parity.md | 4 +- package.json | 4 +- src/broadcast-worker.ts | 2 - src/effect-worker.ts | 2 - src/reminder-scheduler.ts | 2 - src/runtime.ts | 27 +++- src/worker.ts | 2 - test/caller-lifecycle.test.ts | 270 ++++++++++++++++++++++++++++++++++ 9 files changed, 307 insertions(+), 18 deletions(-) create mode 100644 test/caller-lifecycle.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 423c355..66833e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## Unreleased + +- Give overlapping synchronous caller passes separate, reusable workers so + cleanup cannot discard another turn's staged effects or other intents. + Calls to different actors still execute concurrently; workers are reused + once their pass and activation cleanup finish. +- Re-establish actor, effect, reminder, and broadcast process registrations + before each pass, including after stale-process cleanup and retention. + Explicitly stopped workers stay stopped. No schema or setting changes. +- Cover caller concurrency and process retention on SQLite, PostgreSQL, and + MySQL, with `maxAttempts: 1` and both default and single-message pass limits. + ## 0.17.7 - 2026-10-12 - Renew retained activations between short turns and across worker passes, diff --git a/docs/parity.md b/docs/parity.md index 5105402..05be29e 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -59,7 +59,7 @@ is needed for the JavaScript stale-result race fix. | Bounded activation passes and hot-actor fairness | Native | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting. | | Bounded claim candidate scan | Native | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race. | | Backpressure and payload caps | Partial | Serialization enforces a shared maximum JSON nesting depth, raising `InvalidPayload`, and an optional caller-supplied `maxBytes` limit, raising `PayloadTooLarge`; keyed reminder names are bounded to 255 characters in JS and 191 in Ruby. These existing database limits remain runtime-specific; use at most 191 ASCII characters for a name shared across runtimes. Distributed per-actor rate limits, global admission control, and cache-capacity eviction are not planned in either runtime. They are hot, request-path, and loss-tolerant, so one durable ordered message per check is the wrong shape. Solid Objects Pro answers them with grouped and ephemeral operations, which [fit](fit.md) describes. | -| Idle activation cache | Native | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown. | +| Idle activation cache | Native | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown. Overlapping synchronous caller passes own separate workers, preserving staged intents and concurrent execution across actors; Ruby already isolates each invocation's activation. | | Transactional effects and outcome operations | Native | At-least-once handlers receive immutable stable effect, attempt, source-message, and actor identity; success and failure operations also receive the originally staged arguments for correlation. Typed callback envelopes are exported. | | Actor-to-actor delivery | Native | `sendTo(reference).operation()` stages delivery in the source actor commit. | | One-shot and recurring reminders | Native | Scheduling, cancellation by name/key/handle, read-your-writes inspection, catch-up policy, stale-claim recovery, pausing, and authorized resume. Cancellation cannot recall an already-enqueued occurrence. | @@ -84,7 +84,7 @@ Ruby publishes RBS contracts with native Ruby field names; this does not change | Additional supervised components | Native | `registerComponent()` builds, validates, runs, and stops application components with the runtime. | | Dead-letter inspection and retry | Native | `runtime.deadLetters` provides deny-by-default immutable inspection and idempotent durable retry linkage. | | Reconciliation reads | Native | Authorized cursor pages cover active, quiet, and orphaned instances; bounded state batches are migrated and deeply frozen. | -| Message, process, and opt-in instance retention | Native | Supervised scheduling bounds message and process growth; authorized manual APIs add preview and keep destructive instance expiration explicit. SQL instance pruning uses an index on actor type and update time. | +| Message, process, and opt-in instance retention | Native | Supervised scheduling bounds message and process growth; authorized manual APIs add preview and keep destructive instance expiration explicit. Reused JS workers refresh their registration before each pass after cleanup or pruning; Ruby callers already reload and replace stale registrations. SQL instance pruning uses an index on actor type and update time. | | Doctor and schema verification | Native | Structured checks cover configuration, schema/version shape, adapter server versions, neutral-context policy probes, live roles, and a targeted round trip. | | CLI | Native | The packaged executable loads an application runtime and exposes start, diagnostics, processes, dead letters, reminders, and explicit retention pruning as JSON. | | Operator dashboard | Native | Fetch and Node/Connect mounting, authorized views and actions, CSRF, filtering, paging, charts, and extensions match the Rack dashboard. Both dashboards expose individual retries and pause/resume; bulk redrive is available through the administration APIs. Pause does not interrupt an in-flight turn. | diff --git a/package.json b/package.json index baff34b..4684f26 100644 --- a/package.json +++ b/package.json @@ -118,8 +118,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/activation-lease.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", - "test:mysql": "vitest run test/mysql.test.ts test/activation-lease.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", + "test:postgresql": "vitest run test/postgresql.test.ts test/activation-lease.test.ts test/caller-lifecycle.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", + "test:mysql": "vitest run test/mysql.test.ts test/activation-lease.test.ts test/caller-lifecycle.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.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/broadcast-worker.ts b/src/broadcast-worker.ts index f31892e..f324365 100644 --- a/src/broadcast-worker.ts +++ b/src/broadcast-worker.ts @@ -28,7 +28,6 @@ export class BroadcastWorker { async runOnce(): Promise { if (this.stopping) return 0 await this.ensureRegistered() - await this.runtime.repository.heartbeatProcess(this.processId) const broadcast = await this.runtime.repository.claimBroadcast(this.processId) if (!broadcast) return 0 await withProcessHeartbeat({ @@ -88,7 +87,6 @@ export class BroadcastWorker { } private async ensureRegistered(): Promise { - if (this.registered) return await this.runtime.repository.registerProcess(this.processId, "broadcast_worker") this.registered = true } diff --git a/src/effect-worker.ts b/src/effect-worker.ts index 3446d4d..adb4b31 100644 --- a/src/effect-worker.ts +++ b/src/effect-worker.ts @@ -28,7 +28,6 @@ export class EffectWorker { async runOnce(): Promise { if (this.stopping) return 0 await this.ensureRegistered() - await this.runtime.repository.heartbeatProcess(this.processId) const effect = await this.runtime.repository.claimEffect(this.processId) if (!effect) return 0 await withProcessHeartbeat({ @@ -88,7 +87,6 @@ export class EffectWorker { } private async ensureRegistered(): Promise { - if (this.registered) return await this.runtime.repository.registerProcess(this.processId, "effect_worker") this.registered = true } diff --git a/src/reminder-scheduler.ts b/src/reminder-scheduler.ts index cfdcaa2..2c21706 100644 --- a/src/reminder-scheduler.ts +++ b/src/reminder-scheduler.ts @@ -32,7 +32,6 @@ export class ReminderScheduler { throw new TypeError("reminder test time must be a valid date") } await this.ensureRegistered() - await this.runtime.repository.heartbeatProcess(this.processId) const reminder = await this.runtime.repository.claimReminder(this.processId, { ...(nowMilliseconds === undefined ? {} : { nowMilliseconds }), }) @@ -109,7 +108,6 @@ export class ReminderScheduler { } private async ensureRegistered(): Promise { - if (this.registered) return await this.runtime.repository.registerProcess(this.processId, "reminder_scheduler") this.registered = true } diff --git a/src/runtime.ts b/src/runtime.ts index b2107c0..a2771eb 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -216,7 +216,8 @@ export class SolidObjectsRuntime { private readonly effects = new Map() private readonly commitActions = new Map() private readonly additionalComponents: ComponentRegistration[] = [] - private callerWorker: Worker | undefined + private readonly callerWorkers = new Set() + private readonly idleCallerWorkers: Worker[] = [] private running = false private pollingOnlyWakeUpWarningEmitted = false private pollingOnlyWakeUpWarningCheck: Promise | undefined @@ -496,7 +497,6 @@ export class SolidObjectsRuntime { messageReference: MessageReference, options: { timeout: number; deadline: number }, ): Promise> { - this.callerWorker ??= new Worker(this, { setupFailure: "throw" }) while (performance.now() < options.deadline) { let snapshot: { message: MessageRow | undefined; status: MessageStatus } try { @@ -513,7 +513,7 @@ export class SolidObjectsRuntime { if (status === "completed") return readonlyCopy(parseResult(message.result)) as DeepReadonly if (status === "dead") throw messageFailure(message) - const processed = await this.callerWorker.runOnce({ activationRetention: "release" }) + const processed = await this.assistCaller() const remaining = options.deadline - performance.now() if (processed === 0 && remaining > 0) await new Promise((resolve) => @@ -549,6 +549,22 @@ export class SolidObjectsRuntime { throw await this.syncTimeout(messageReference, options.timeout) } + private async assistCaller(): Promise { + const worker = this.idleCallerWorkers.pop() ?? new Worker(this, { setupFailure: "throw" }) + this.callerWorkers.add(worker) + try { + return await worker.runOnce({ activationRetention: "release" }) + } finally { + this.idleCallerWorkers.push(worker) + } + } + + private async stopCallerWorkers(): Promise { + await Promise.all([...this.callerWorkers].map((worker) => worker.stop())) + this.callerWorkers.clear() + this.idleCallerWorkers.length = 0 + } + private messageSnapshotBeforeDeadline( messageReference: MessageReference, deadline: number, @@ -1493,7 +1509,7 @@ export class SolidObjectsRuntime { async close(): Promise { if (this.running) throw new Error("abort runtime.run() and wait for it before closing the runtime") - await this.callerWorker?.stop() + await this.stopCallerWorkers() this.realtime.close() await this.closeWakeUp() this.observers.clear() @@ -1545,8 +1561,7 @@ export class SolidObjectsRuntime { async resetForTesting(): Promise { if (this.running) throw new Error("abort runtime.run() before resetting test state") - await this.callerWorker?.stop() - this.callerWorker = undefined + await this.stopCallerWorkers() this.realtime.close() await this.discardWakeUp() await this.repository.resetForTesting() diff --git a/src/worker.ts b/src/worker.ts index 762ef10..d3cf2b1 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -53,7 +53,6 @@ export class Worker { try { if (this.stopping) return 0 await this.ensureRegistered() - await this.runtime.repository.heartbeatProcess(this.processId) await this.maintainCachedActivations() let cached: CachedActivation | undefined = await this.claimCachedActivation() let turn = cached?.turn ?? (await this.runtime.repository.claim(this.processId)) @@ -190,7 +189,6 @@ export class Worker { } private async ensureRegistered(): Promise { - if (this.registered) return await this.runtime.repository.registerProcess(this.processId, "worker") this.registered = true } diff --git a/test/caller-lifecycle.test.ts b/test/caller-lifecycle.test.ts new file mode 100644 index 0000000..527d235 --- /dev/null +++ b/test/caller-lifecycle.test.ts @@ -0,0 +1,270 @@ +import { afterEach, expect, it, vi } from "vitest" +import { Actor } from "../src/actor.js" +import type { InstrumentationEvent } from "../src/configuration.js" +import { mysql } from "../src/database/mysql.js" +import { postgresql } from "../src/database/postgresql.js" +import { sqlite } from "../src/database/sqlite.js" +import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { Worker } from "../src/worker.js" + +class CallerCounter extends Actor { + static override readonly actorType = "CallerCounter" + static beforeCommit: (label: string) => Promise = async () => {} + static deactivated: () => void = () => {} + count = 0 + + async increment(options: { label: string }): Promise { + this.count += 1 + this.emit("caller-effect", { arguments: { label: options.label } }) + await CallerCounter.beforeCommit(options.label) + return this.count + } + + override onDeactivate(): void { + CallerCounter.deactivated() + } +} + +let runtime: SolidObjectsRuntime +let controller: AbortController | undefined +let running: Promise | undefined +const events: InstrumentationEvent[] = [] + +afterEach(async () => { + controller?.abort() + await running + await runtime?.resetForTesting() + await runtime?.close() + vi.restoreAllMocks() + events.length = 0 + controller = undefined + running = undefined + CallerCounter.beforeCommit = async () => {} + CallerCounter.deactivated = () => {} +}) + +async function setup(options: { maxMessagesPerActivationPass?: number } = {}): Promise { + const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL + const database = connectionString?.startsWith("postgres") + ? postgresql({ connectionString }) + : connectionString?.startsWith("mysql") + ? mysql({ connectionString }) + : sqlite({ path: ":memory:" }) + runtime = configure({ + database, + tableNamePrefix: "caller_test_", + authorizeMessage: () => true, + authorizeAdministration: () => true, + broadcast: async () => {}, + maxAttempts: 1, + instrumentation: (event) => events.push(event), + ...options, + }) + runtime.register(CallerCounter) + runtime.registerEffect("caller-effect", async () => {}) + await runtime.install() + await runtime.repository.resetForTesting() +} + +async function effectLabels(): Promise { + const rows = await runtime.settings.database.connection((connection) => + connection.all<{ arguments: string }>( + `SELECT arguments FROM ${runtime.repository.table("effects")}`, + ), + ) + return rows.map((row) => (JSON.parse(row.arguments) as { label: string }).label).sort() +} + +it.each([50, 1])( + "preserves effects while callers overlap with pass limit %i", + async (passLimit) => { + await setup({ maxMessagesPerActivationPass: passLimit }) + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + const idlePass = Promise.withResolvers() + let executing = false + let deactivatedDuringTurn = false + CallerCounter.beforeCommit = async (label) => { + if (label !== "second") return + executing = true + entered.resolve() + await release.promise + executing = false + } + CallerCounter.deactivated = () => { + if (executing) deactivatedDuringTurn = true + } + const runOnce = Worker.prototype.runOnce + vi.spyOn(Worker.prototype, "runOnce").mockImplementation(async function ( + this: Worker, + options, + ) { + const count = await runOnce.call(this, options) + if (count === 0) idlePass.resolve() + return count + }) + const reference = CallerCounter.ref("shared") + const first = await reference.send.increment({ label: "first" }) + const second = reference.increment({ label: "second" }) + let third: Promise | undefined + try { + await beforeDeadline(entered.promise) + third = reference.increment({ label: "third" }) + await beforeDeadline(idlePass.promise) + } finally { + release.resolve() + await Promise.all([second, third]) + } + expect(await first.result()).toBe(1) + expect(await second).toBe(2) + expect(await third).toBe(3) + expect(await effectLabels()).toEqual(["first", "second", "third"]) + expect(deactivatedDuringTurn).toBe(false) + expect(events.filter((event) => event.name === "solid_objects.activation.lost")).toEqual([]) + expect(events.filter((event) => event.name === "solid_objects.message.completed")).toHaveLength( + 3, + ) + expect( + events + .filter((event) => event.name === "solid_objects.message.started") + .map((event) => event.attempt), + ).toEqual([1, 1, 1]) + }, +) + +it.each(["worker", "effectWorker", "reminderScheduler", "broadcastWorker"] as const)( + "restores the process record before an idle %s resumes", + async (role) => { + await setup() + const worker = runtime[role]() + try { + expect(await worker.runOnce()).toBe(0) + 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 - runtime.settings.processAliveThresholdMilliseconds - 1, worker.processId], + ) + }) + await runtime.cleanupProcesses() + await runtime.settings.database.connection(async (connection) => { + const now = await connection.nowMilliseconds() + await connection.run( + `UPDATE ${runtime.repository.table("processes")} SET stopped_at_ms = ? WHERE id = ?`, + [now - 8 * 86_400_000, worker.processId], + ) + }) + expect((await runtime.retention.prune({ target: "processes" })).count).toBe(1) + expect(await worker.runOnce()).toBe(0) + expect((await runtime.repository.listProcesses()).rows).toEqual([ + expect.objectContaining({ id: worker.processId, shutdown_state: "running" }), + ]) + } finally { + await worker.stop() + } + expect(await worker.runOnce()).toBe(0) + expect((await runtime.repository.listProcesses()).rows[0]?.shutdown_state).toBe("stopped") + }, +) + +it("allows synchronous calls to different actors to execute together", async () => { + await setup() + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + const labels: string[] = [] + CallerCounter.beforeCommit = async (label) => { + labels.push(label) + if (labels.length === 2) entered.resolve() + await release.promise + } + const first = CallerCounter.ref("first").increment({ label: "first" }) + const second = CallerCounter.ref("second").increment({ label: "second" }) + try { + await beforeDeadline(entered.promise) + } finally { + release.resolve() + await Promise.all([first, second]) + } + expect(await first).toBe(1) + expect(await second).toBe(1) + expect(await effectLabels()).toEqual(["first", "second"]) + const processIds = (await runtime.repository.listProcesses()).rows.map((row) => row.id).sort() + expect(processIds).toHaveLength(2) + CallerCounter.beforeCommit = async () => {} + expect( + await Promise.all([ + CallerCounter.ref("first").increment({ label: "first-again" }), + CallerCounter.ref("second").increment({ label: "second-again" }), + ]), + ).toEqual([2, 2]) + expect((await runtime.repository.listProcesses()).rows.map((row) => row.id).sort()).toEqual( + processIds, + ) + await runtime.resetForTesting() + expect(await CallerCounter.ref("reset").increment({ label: "reset" })).toBe(1) + expect((await runtime.repository.listProcesses()).rows).toHaveLength(1) +}) + +async function beforeDeadline(operation: Promise): Promise { + let timeout: ReturnType | undefined + try { + await Promise.race([ + operation, + new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error("caller did not reach the barrier")), 3000) + }), + ]) + } finally { + clearTimeout(timeout) + } +} + +it.each(["stopped", "pruned"])( + "recovers a %s caller registration after runtime shutdown", + async (state) => { + await setup() + const reference = CallerCounter.ref("retained-process") + expect(await reference.increment({ label: "before" })).toBe(1) + const caller = (await runtime.repository.listProcesses()).rows[0]! + controller = new AbortController() + running = runtime.run(controller.signal) + await vi.waitFor(async () => + expect((await runtime.repository.listProcesses()).rows.length).toBeGreaterThan(1), + ) + 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 - runtime.settings.processAliveThresholdMilliseconds - 1, caller.id], + ) + }) + await runtime.cleanupProcesses() + expect( + (await runtime.repository.listProcesses()).rows.find((row) => row.id === caller.id) + ?.shutdown_state, + ).toBe("stopped") + if (state === "pruned") { + await runtime.settings.database.connection(async (connection) => { + const now = await connection.nowMilliseconds() + await connection.run( + `UPDATE ${runtime.repository.table("processes")} SET stopped_at_ms = ? WHERE id = ?`, + [now - 8 * 86_400_000, caller.id], + ) + }) + expect((await runtime.retention.prune({ target: "processes" })).count).toBe(1) + } + controller.abort() + await running + running = undefined + expect(await reference.increment({ label: "after" })).toBe(2) + expect( + (await runtime.repository.listProcesses()).rows.find((row) => row.id === caller.id) + ?.shutdown_state, + ).toBe("running") + expect( + events + .filter((event) => event.name === "solid_objects.message.started") + .map((event) => event.attempt), + ).toEqual([1, 1]) + }, +) From 1d4d254680248a4df2827810ed7e8fa620f0c6da Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 10:35:48 -0700 Subject: [PATCH 2/9] fix: discard stopped caller workers after reset --- CHANGELOG.md | 2 ++ src/runtime.ts | 2 +- test/caller-lifecycle.test.ts | 26 ++++++++++++++++++++++++++ 3 files changed, 29 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 66833e4..6a25faf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ cleanup cannot discard another turn's staged effects or other intents. Calls to different actors still execute concurrently; workers are reused once their pass and activation cleanup finish. + A caller interrupted by test reset cannot return its stopped worker to + the pool and stall subsequent calls. - Re-establish actor, effect, reminder, and broadcast process registrations before each pass, including after stale-process cleanup and retention. Explicitly stopped workers stay stopped. No schema or setting changes. diff --git a/src/runtime.ts b/src/runtime.ts index a2771eb..c92b296 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -555,7 +555,7 @@ export class SolidObjectsRuntime { try { return await worker.runOnce({ activationRetention: "release" }) } finally { - this.idleCallerWorkers.push(worker) + if (!worker.stopped()) this.idleCallerWorkers.push(worker) } } diff --git a/test/caller-lifecycle.test.ts b/test/caller-lifecycle.test.ts index 527d235..a7112c5 100644 --- a/test/caller-lifecycle.test.ts +++ b/test/caller-lifecycle.test.ts @@ -4,6 +4,7 @@ import type { InstrumentationEvent } from "../src/configuration.js" import { mysql } from "../src/database/mysql.js" import { postgresql } from "../src/database/postgresql.js" import { sqlite } from "../src/database/sqlite.js" +import { ActorDestroyed } from "../src/errors.js" import { configure, type SolidObjectsRuntime } from "../src/runtime.js" import { Worker } from "../src/worker.js" @@ -219,6 +220,31 @@ async function beforeDeadline(operation: Promise): Promise { } } +it("discards a stopped caller that finishes after test reset", async () => { + await setup() + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + CallerCounter.beforeCommit = async (label) => { + if (label !== "interrupted") return + entered.resolve() + await release.promise + } + const reference = CallerCounter.ref("reset-during-call").with({ timeoutMilliseconds: 1000 }) + const interrupted = reference.increment({ label: "interrupted" }).catch((error: unknown) => error) + try { + await beforeDeadline(entered.promise) + await runtime.resetForTesting() + } finally { + release.resolve() + await interrupted + } + + expect(await interrupted).toBeInstanceOf(ActorDestroyed) + expect(await reference.increment({ label: "after-reset" })).toBe(1) + expect(await reference.increment({ label: "reused" })).toBe(2) + expect(await effectLabels()).toEqual(["after-reset", "reused"]) +}) + it.each(["stopped", "pruned"])( "recovers a %s caller registration after runtime shutdown", async (state) => { From 59dd0d3325223ad187ac08ba02093e0bab7b1898 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 10:42:52 -0700 Subject: [PATCH 3/9] test: narrow interrupted caller errors --- test/caller-lifecycle.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/caller-lifecycle.test.ts b/test/caller-lifecycle.test.ts index a7112c5..76f466f 100644 --- a/test/caller-lifecycle.test.ts +++ b/test/caller-lifecycle.test.ts @@ -230,7 +230,10 @@ it("discards a stopped caller that finishes after test reset", async () => { await release.promise } const reference = CallerCounter.ref("reset-during-call").with({ timeoutMilliseconds: 1000 }) - const interrupted = reference.increment({ label: "interrupted" }).catch((error: unknown) => error) + const interrupted = reference.increment({ label: "interrupted" }).catch((error: unknown) => { + if (error instanceof Error) return error + throw new TypeError("Expected an Error rejection") + }) try { await beforeDeadline(entered.promise) await runtime.resetForTesting() From 5979a9ec363815a8f30ec3bdd631e08d8a28536f Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 10:54:54 -0700 Subject: [PATCH 4/9] test: drain retries in the recovery demo A lost activation counts as an attempted turn, so the demo worker could exit before its reclaimed message committed on a delayed CI runner. Drain available turns and assert completion before checking state. Additional attempts can repeat external effects while the durable state still commits exactly once. Reproduce the failure with a real SQLite lease expiry before the change. The new regression and a Node 24.4.0 subprocess stall now pass, along with 638 tests and the type, style, and documentation checks. --- CHANGELOG.md | 3 + examples/failure-recovery/demo.ts | 11 ++-- examples/failure-recovery/process-messages.ts | 10 ++++ examples/failure-recovery/worker.ts | 7 +-- test/failure-recovery-worker.test.ts | 56 +++++++++++++++++++ 5 files changed, 77 insertions(+), 10 deletions(-) create mode 100644 examples/failure-recovery/process-messages.ts create mode 100644 test/failure-recovery-worker.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a25faf..bc01aab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,9 @@ Explicitly stopped workers stay stopped. No schema or setting changes. - Cover caller concurrency and process retention on SQLite, PostgreSQL, and MySQL, with `maxAttempts: 1` and both default and single-message pass limits. +- Drain recovered turns in the failure-recovery demo before checking committed + state. Verify completion and allow additional fenced retries when a worker + loses its lease on a delayed runner. ## 0.17.7 - 2026-10-12 diff --git a/examples/failure-recovery/demo.ts b/examples/failure-recovery/demo.ts index 491dff8..552006b 100644 --- a/examples/failure-recovery/demo.ts +++ b/examples/failure-recovery/demo.ts @@ -72,7 +72,7 @@ async function proveSerialization(): Promise finished)) - await Promise.all(messages.map((message) => message.result())) + for (const message of messages) assert.equal(await message.status(), "completed") const events = await readJsonLines( join(controlDirectory, "serialization.jsonl"), parseSerializationEvent, @@ -99,7 +99,7 @@ async function proveCrashRecovery(): Promise<{ await wait(350) const recoveryWorker = spawnWorker() await recoveryWorker.finished - await message.result() + assert.equal(await message.status(), "completed") return recoveryResult({ reference, message, controlDirectory }) } @@ -118,7 +118,7 @@ async function proveFencing(): Promise<{ await wait(350) const recoveryWorker = spawnWorker() await recoveryWorker.finished - await message.result() + assert.equal(await message.status(), "completed") staleWorker.child.kill("SIGCONT") await writeFile(join(controlDirectory, "release-first-attempt"), "") await staleWorker.waitFor((entry) => entry.event === "solid_objects.activation.lost") @@ -138,9 +138,10 @@ async function recoveryResult(options: { join(options.controlDirectory, "external-effects.jsonl"), parseExternalEffectEvent, ) - assert.equal(attempts, 2) + assert.ok(attempts >= 2) assert.equal(snapshot.count, 1) - assert.equal(effects.length, 2) + assert.ok(effects.length >= 2) + assert.ok(effects.length <= attempts) return { attempts, finalState: snapshot.count, repeatedEffects: effects.length } } diff --git a/examples/failure-recovery/process-messages.ts b/examples/failure-recovery/process-messages.ts new file mode 100644 index 0000000..69765ba --- /dev/null +++ b/examples/failure-recovery/process-messages.ts @@ -0,0 +1,10 @@ +export async function processMessages(worker: { + runUntilIdle(): Promise +}): Promise { + let processed = 0 + for (let attempt = 0; attempt < 200 && processed === 0; attempt += 1) { + processed = await worker.runUntilIdle() + if (processed === 0) await new Promise((resolve) => setTimeout(resolve, 10)) + } + return processed +} diff --git a/examples/failure-recovery/worker.ts b/examples/failure-recovery/worker.ts index 12c48ea..9f372ed 100644 --- a/examples/failure-recovery/worker.ts +++ b/examples/failure-recovery/worker.ts @@ -1,6 +1,7 @@ import { createRuntime } from "solid-objects" import { sqlite } from "solid-objects/database/sqlite" import { RecoverableReport, RecoveryCounter } from "./actor.ts" +import { processMessages } from "./process-messages.ts" const databasePath = requiredArgument(2) const runtime = createRuntime({ @@ -36,11 +37,7 @@ if (process.argv[3] === "retire-effects") { const worker = runtime.worker() try { - let processed = 0 - for (let attempt = 0; attempt < 200 && processed === 0; attempt += 1) { - processed = await worker.runOnce({ activationRetention: "release" }) - if (processed === 0) await new Promise((resolve) => setTimeout(resolve, 10)) - } + const processed = await processMessages(worker) process.send?.({ event: "worker.finished", processed }) } finally { await worker.stop() diff --git a/test/failure-recovery-worker.test.ts b/test/failure-recovery-worker.test.ts new file mode 100644 index 0000000..e64a2a8 --- /dev/null +++ b/test/failure-recovery-worker.test.ts @@ -0,0 +1,56 @@ +import { expect, it } from "vitest" +import { processMessages } from "../examples/failure-recovery/process-messages.js" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { createRuntime } from "../src/runtime.js" + +it("drains recovered messages after a demo worker loses its lease", async () => { + class RecoveryCounter extends Actor { + static override readonly actorType = "RecoveryWorkerCounter" + count = 0 + + increment(options: { block?: "first-attempt" }): number { + if (options.block && this.currentMessage?.attempt === 1) { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 300) + } + this.count += 1 + return this.count + } + } + + const lostActivations: string[] = [] + const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + leaseDurationMilliseconds: 200, + leaseRenewalIntervalMilliseconds: 50, + maxMessagesPerActivationPass: 1, + maxAttempts: 1, + authorizeMessage: () => true, + authorizeQuery: () => true, + instrumentation: ({ name }) => { + if (name === "solid_objects.activation.lost") lostActivations.push(name) + }, + }) + runtime.register(RecoveryCounter) + const worker = runtime.worker() + + try { + await runtime.install() + const reference = runtime.ref(RecoveryCounter, "counter") + const first = await reference.send.increment({ block: "first-attempt" }) + const second = await reference.send.increment({}) + + await processMessages(worker) + + expect(await first.status()).toBe("completed") + expect(await second.status()).toBe("completed") + expect(await first.result()).toBe(1) + expect(await second.result()).toBe(2) + expect((await reference.snapshot()).count).toBe(2) + expect(lostActivations).toHaveLength(1) + expect(Number((await runtime.repository.findMessage(first.id))?.attempt_count)).toBe(2) + } finally { + await worker.stop() + await runtime.close() + } +}) From 78338eb4d42e78c41ccfc1f75d5c9b4d500e18b3 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 11:01:18 -0700 Subject: [PATCH 5/9] test: control time in commit fencing checks The PostgreSQL and MySQL commit-fencing tests could lose their 100 ms lease before reaching the commit barrier on a delayed runner. A 200 ms handler stall reproduces the 15-second timeout in both adapters. Keep that stall and advance a controlled database clock only after the commit has acquired its locks. The competing claim must still wait and must not recover the committed message. No runtime defaults change. Validate 69 PostgreSQL checks, 56 MySQL checks with 7 conditional skips, 638 default-suite checks with 32 conditional skips, types, and format. --- CHANGELOG.md | 3 +++ test/mysql.test.ts | 8 +++--- test/postgresql.test.ts | 8 +++--- test/support/fenced-commit.ts | 50 +++++++++++++++++++---------------- 4 files changed, 40 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bc01aab..4b24349 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,9 @@ - Drain recovered turns in the failure-recovery demo before checking committed state. Verify completion and allow additional fenced retries when a worker loses its lease on a delayed runner. +- Advance the database test clock after acquiring commit locks in PostgreSQL + and MySQL fencing tests, so runner delays cannot expire the lease before + the commit barrier is reached. ## 0.17.7 - 2026-10-12 diff --git a/test/mysql.test.ts b/test/mysql.test.ts index c0c8e2c..899fc57 100644 --- a/test/mysql.test.ts +++ b/test/mysql.test.ts @@ -9,7 +9,7 @@ import { withDatabaseDeadline } from "../src/database/deadline.js" import type { JsonObject } from "../src/types.js" import { createDashboard } from "../src/web/index.js" import { captureAttempt, PausingClaimDatabase } from "./support/pausing-claim-database.js" -import { deferred, delay, waitForActivationExpiration } from "./support/fenced-commit.js" +import { controlledDatabaseClock, deferred, delay } from "./support/fenced-commit.js" const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL class TransmitProofCounter extends Actor { @@ -101,6 +101,7 @@ class MySQLFencedCommitActor extends Actor { count = 0 increment(): number { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 200) this.count += 1 this.commitAction("pause_fenced_commit") return this.count @@ -192,8 +193,9 @@ describeMySQL("MySQL adapter", () => { it("keeps a fenced commit exclusive after its lease expires", async () => { if (!connectionString) throw new Error("MySQL connection string is required") database = mysql({ connectionString, maximumConnections: 5 }) + const clock = controlledDatabaseClock(database) runtime = configure({ - database, + database: clock.database, tableNamePrefix: "mysql_test_", authorizeMessage: () => true, authorizeQuery: () => true, @@ -216,7 +218,7 @@ describeMySQL("MySQL adapter", () => { const firstRun = firstWorker.runOnce() await commitStarted.promise try { - await waitForActivationExpiration(runtime, MySQLFencedCommitActor.actorType, actorId) + clock.advanceMilliseconds(101) await runtime.repository.registerProcess("replacement-worker", "worker") const replacementClaim = runtime.repository.claim("replacement-worker") const replacementBeforeRelease = await Promise.race([ diff --git a/test/postgresql.test.ts b/test/postgresql.test.ts index 19b6f90..c21706a 100644 --- a/test/postgresql.test.ts +++ b/test/postgresql.test.ts @@ -16,7 +16,7 @@ import { withDatabaseDeadline } from "../src/database/deadline.js" import type { JsonObject } from "../src/types.js" import { createDashboard } from "../src/web/index.js" import { captureAttempt, PausingClaimDatabase } from "./support/pausing-claim-database.js" -import { deferred, delay, waitForActivationExpiration } from "./support/fenced-commit.js" +import { controlledDatabaseClock, deferred, delay } from "./support/fenced-commit.js" const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL const describePostgreSQL = connectionString?.startsWith("postgresql:") ? describe : describe.skip @@ -104,6 +104,7 @@ class PostgreSQLFencedCommitActor extends Actor { count = 0 increment(): number { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 200) this.count += 1 this.commitAction("pause_fenced_commit") return this.count @@ -226,8 +227,9 @@ describePostgreSQL("PostgreSQL adapter", () => { it("keeps a fenced commit exclusive after its lease expires", async () => { if (!connectionString) throw new Error("PostgreSQL connection string is required") database = postgresql({ connectionString, maximumConnections: 5 }) + const clock = controlledDatabaseClock(database) runtime = configure({ - database, + database: clock.database, tableNamePrefix: "postgresql_test_", authorizeMessage: () => true, authorizeQuery: () => true, @@ -250,7 +252,7 @@ describePostgreSQL("PostgreSQL adapter", () => { const firstRun = firstWorker.runOnce() await commitStarted.promise try { - await waitForActivationExpiration(runtime, PostgreSQLFencedCommitActor.actorType, actorId) + clock.advanceMilliseconds(101) await runtime.repository.registerProcess("replacement-worker", "worker") const replacementClaim = runtime.repository.claim("replacement-worker") const replacementBeforeRelease = await Promise.race([ diff --git a/test/support/fenced-commit.ts b/test/support/fenced-commit.ts index 142eedc..a5694ac 100644 --- a/test/support/fenced-commit.ts +++ b/test/support/fenced-commit.ts @@ -1,4 +1,4 @@ -import type { SolidObjectsRuntime } from "../../src/runtime.js" +import type { Database, DatabaseConnection } from "../../src/database/types.js" export function deferred(): { promise: Promise; resolve: () => void } { let resolve = () => {} @@ -12,27 +12,31 @@ export function delay(milliseconds: number): Promise { return new Promise((resolve) => setTimeout(resolve, milliseconds)) } -export async function waitForActivationExpiration( - runtime: SolidObjectsRuntime, - actorType: string, - actorId: string, -): Promise { - const expiration = await runtime.settings.database.connection((connection) => - connection.get<{ activation_expires_at_ms: number | bigint }>( - `SELECT activation_expires_at_ms FROM ${runtime.repository.table("instances")} - WHERE actor_type = ? AND actor_id = ?`, - [actorType, actorId], - ), - ) - if (!expiration) throw new Error("active actor instance was not found") - const expiresAt = Number(expiration.activation_expires_at_ms) - const deadline = Date.now() + 2_000 - while (Date.now() < deadline) { - const now = await runtime.settings.database.connection((connection) => - connection.nowMilliseconds(), - ) - if (now > expiresAt) return - await delay(Math.min(Math.max(expiresAt - now + 1, 1), 10)) +export function controlledDatabaseClock(database: Database): { + database: Database + advanceMilliseconds(milliseconds: number): void +} { + let now = Date.now() + const clockedConnection = (connection: DatabaseConnection): DatabaseConnection => ({ + run: connection.run.bind(connection), + get: connection.get.bind(connection), + all: connection.all.bind(connection), + nowMilliseconds: async () => now, + }) + return { + advanceMilliseconds: (milliseconds) => { + now += milliseconds + }, + database: { + family: database.family, + schemaIdentity: database.schemaIdentity, + transactionActive: () => database.transactionActive?.() ?? false, + ...(database.wakeUp ? { wakeUp: database.wakeUp.bind(database) } : {}), + connection: (callback) => + database.connection((connection) => callback(clockedConnection(connection))), + transaction: (callback, options) => + database.transaction((connection) => callback(clockedConnection(connection)), options), + close: () => database.close(), + }, } - throw new Error("activation lease did not expire") } From a330aaf187cd9c6782ec83447408bee954563d28 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 11:08:47 -0700 Subject: [PATCH 6/9] chore: prepare JavaScript 0.17.8 Align the package version, exported runtime version, and dated changelog so the caller lifecycle fixes can be tagged after merging. --- CHANGELOG.md | 2 +- package.json | 2 +- src/version.ts | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b24349..715248e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## Unreleased +## 0.17.8 - 2026-10-12 - Give overlapping synchronous caller passes separate, reusable workers so cleanup cannot discard another turn's staged effects or other intents. diff --git a/package.json b/package.json index 4684f26..7467fd0 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "solid-objects", - "version": "0.17.7", + "version": "0.17.8", "description": "SQL-backed virtual actor library for TypeScript and Node.js, with durable state, ordered operations, and automatic activation on SQLite, PostgreSQL, or MySQL", "type": "module", "license": "MIT", diff --git a/src/version.ts b/src/version.ts index 7baf0ef..3fad9f0 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const VERSION = "0.17.7" +export const VERSION = "0.17.8" From 89d4f96c1282e2158bee2124b1b43f544994f0a4 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 11:54:22 -0700 Subject: [PATCH 7/9] fix: back off interrupted message recovery Independently reported lease loss could retry indefinitely while resolved SQLite promises prevented timers and input from running. Yield between busy worker and drain passes, preserving immediate instrumentation aborts. Recover attempt one immediately, then persist the existing retry delay for total attempts minus one. Preserve fences, mailbox order, and handler failure limits. Add child-process starvation regressions, durable recovery coverage, and the matching Ruby telemetry contract and documentation. Baseline: 20 new regression failures and two passing controls. All pass with the fix, alongside full SQLite, PostgreSQL, and MySQL suites. --- CHANGELOG.md | 13 ++ compatibility/telemetry-events.json | 15 ++ docs/agents.md | 3 + docs/configuration.md | 26 ++- docs/observability.md | 31 +-- docs/operations.md | 3 +- docs/parity.md | 46 ++--- docs/virtual-actors.md | 4 + package.json | 4 +- src/broadcast-worker.ts | 4 + src/effect-worker.ts | 4 + src/platform/node.ts | 2 + src/reminder-scheduler.ts | 4 + src/repository.ts | 30 ++- src/test-helper.ts | 8 +- src/worker.ts | 4 + src/yield-to-event-loop.ts | 11 ++ test/activation-recovery-backoff.test.ts | 236 +++++++++++++++++++++++ test/fixtures/busy-worker.ts | 145 ++++++++++++++ test/worker-fairness.test.ts | 110 +++++++++++ 20 files changed, 657 insertions(+), 46 deletions(-) create mode 100644 src/yield-to-event-loop.ts create mode 100644 test/activation-recovery-backoff.test.ts create mode 100644 test/fixtures/busy-worker.ts create mode 100644 test/worker-fairness.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 715248e..978ded5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## 0.17.8 - 2026-10-12 +- Back off repeated interrupted-message recovery using `retryDelayMilliseconds`. + The first claimed attempt recovers immediately; after attempts 2, 3, 4, + and onward, the delay callback receives 1, 2, 3, and onward (defaults: + 1, 2, 4 seconds, capped at 60 seconds). Total persisted claim attempts, + including earlier handler failures, determine the delay. The ready timestamp + survives worker replacement and preserves mailbox order while other actors + can proceed. Lease loss alone still does not dead-letter a message; handler + failure attempt limits, setting names, defaults, and schema stay unchanged. +- Yield to timers and input between busy passes in actor, effect, reminder, + and broadcast workers and drain helpers, including repeated lease loss. + Timer-driven aborts can stop the runtime between attempts; an instrumentation + abort stops before another pass. Synchronous handlers remain non-preemptible. + - Give overlapping synchronous caller passes separate, reusable workers so cleanup cannot discard another turn's staged effects or other intents. Calls to different actors still execute concurrently; workers are reused diff --git a/compatibility/telemetry-events.json b/compatibility/telemetry-events.json index 19669db..fb5ccea 100644 --- a/compatibility/telemetry-events.json +++ b/compatibility/telemetry-events.json @@ -84,6 +84,21 @@ "sequence" ] }, + { + "name": "activation.lost", + "attributes": [ + "actorId", + "actorType", + "attempt", + "deliveryMode", + "durationMilliseconds", + "instanceId", + "messageId", + "operation", + "requestId", + "sequence" + ] + }, { "name": "activation.started", "attributes": ["actorId", "actorType", "generation", "instanceId", "ownerId"], diff --git a/docs/agents.md b/docs/agents.md index 1f31c66..1dc4684 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -206,6 +206,9 @@ Obey these rules in actor code: - Do not call an external API in an operation. Use `this.emit()` and an effect handler. - Write each operation so that it can run again. Delivery is at least once. +- Repeated lease losses back off but do not become dead letters. Observe + `activation.lost` and set an application stopping policy if a hard bound is + required; handler-failure attempt limits do not bound lease loss. - A reminder changes state only when it runs, and it runs only while a process calls `runtime.run(signal)`. Do not compute expiry from the clock in a query; read the state that the reminder committed. diff --git a/docs/configuration.md b/docs/configuration.md index d74b0af..7a387a6 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -21,11 +21,11 @@ through `runtime.ref(ActorClass, actorId)`. Both validate options immediately. | `maxStateBytes` | `5_242_880` | Positive persisted actor-state byte limit. | | `warnStateBytes` | `131_072` | Positive persisted actor-state byte threshold that reports one warning. | | `maxResultBytes` | `1_048_576` | Positive operation-result byte limit. | -| `maxAttempts` | `5` | Positive maximum operation, effect, and broadcast attempts. | +| `maxAttempts` | `5` | Positive attempt limit evaluated on handler failure. | | `maxMessagesPerActivationPass` | `50` | Positive integer turn budget before fairness yield. | | `maxActivationDurationMilliseconds` | `5_000` | Positive elapsed-time budget before fairness yield. | | `claimScanLimit` | `100` | Positive integer ready-actor candidates considered per global claim. | -| `retryDelayMilliseconds` | exponential 1-60 seconds | Receives the next attempt number and returns its delay. | +| `retryDelayMilliseconds` | exponential 1-60 seconds | Receives a retry index; returns milliseconds (see below). | The limits measure normalized JSON encoded as UTF-8. `maxResultBytes` also limits each computed snapshot getter and effect-handler result. Explicit @@ -36,6 +36,22 @@ names enter the broadcast outbox. `retryDelayMilliseconds` should return a non-negative finite number; an invalid application callback will fail the affected failure path rather than schedule an invalid timestamp. +Interrupted actor messages use the existing `retryDelayMilliseconds` callback +on recovery. After the first claimed attempt they are immediately eligible; +after persisted attempts 2, 3, 4, ... the callback receives 1, 2, 3, ... . Its +default delays are 1, 2, 4 seconds up to 60 seconds. This is the total claim count, +including earlier handler failures, not a separate lease-loss counter. The +availability timestamp is stored transactionally in ready membership, survives +worker replacement and stale-process cleanup, and keeps the message ahead of +later messages for that actor. Other actors remain eligible. Custom delays, +including zero, remain authoritative. + +Lease loss alone never creates a dead letter, including with `maxAttempts: 1`. +A handler failure still uses the full attempt count with `maxAttempts` and +passes that count to `retryDelayMilliseconds`, as before. Applications needing +a hard retry or elapsed-time bound must observe `activation.lost` and stop or +intervene. No setting or schema change is required. + `maxStateBytes` is a hard limit that fails the turn. `warnStateBytes` is a soft threshold that keeps the turn. A commit above the threshold reports one `solid_objects.state.large` instrumentation event with the actor type, the actor @@ -76,6 +92,12 @@ and `false` for a timeout. An older adapter that returns `void` remains compatible and keeps the fast cadence. Wake-ups reduce latency, while database polling remains the correctness path. +Busy SQL workers and testing drains give the event loop a turn between passes. +In Node this lets timers and input run even with synchronous SQLite calls. +An instrumentation callback can abort `runtime.run()` before its next pass. +This does not interrupt a synchronous handler: responsiveness is bounded by +one pass, or by a single overlong handler that must return first. + ## Retention and cleanup | Option | Default | Contract | diff --git a/docs/observability.md b/docs/observability.md index a8c93da..deb6bad 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -44,21 +44,22 @@ result. When an exporter or observer raises, both runtimes log Delivery is best effort and synchronous callbacks should be short; JavaScript does not await exporters. Telemetry is not a durable audit trail. -| Event | Meaning | -| ------------------------------------------- | ---------------------------------------------------------------------------- | -| `activation.started/completed/failed` | Local actor activation hook lifecycle | -| `message.started/completed/failed/rejected` | One attempt's execution outcome | -| `message.retry` | Failed attempt durably queued for another attempt | -| `dead_letter.created` | Message exhausted retries or failed permanently | -| `commit_action.started/completed/failed` | One registered commit action inside the commit transaction | -| `mailbox.depth` | On-demand diagnostic sample; `depth` is null when the sample is truncated | -| `reminder.enqueued` | Due reminder dispatch; lateness is measured from due time | -| `outbox.age` | Delivery observation; age is time since the item's current availability time | -| `recovery.reclaimed` | A previously claimed, interrupted message begins another attempt | -| `recovery.completed/failed` | Durable effect recovery callback commits or enters the dead-letter queue | -| `snapshot.read` | Authorized snapshot constructed without exposing its contents | -| `realtime.connected/disconnected` | Actor subscription added or removed | -| `payload_broadcast.failed` | One personalized payload failed; `payload` names it | +| Event | Meaning | +| ------------------------------------------- | ------------------------------------------------------------------------------------- | +| `activation.started/completed/failed` | Local actor activation hook lifecycle | +| `message.started/completed/failed/rejected` | One attempt's execution outcome | +| `activation.lost` | A message turn lost its activation fence; includes message ID and total attempt count | +| `message.retry` | Failed attempt durably queued for another attempt | +| `dead_letter.created` | Message exhausted retries or failed permanently | +| `commit_action.started/completed/failed` | One registered commit action inside the commit transaction | +| `mailbox.depth` | On-demand diagnostic sample; `depth` is null when the sample is truncated | +| `reminder.enqueued` | Due reminder dispatch; lateness is measured from due time | +| `outbox.age` | Delivery observation; age is time since the item's current availability time | +| `recovery.reclaimed` | A previously claimed, interrupted message begins another attempt | +| `recovery.completed/failed` | Durable effect recovery callback commits or enters the dead-letter queue | +| `snapshot.read` | Authorized snapshot constructed without exposing its contents | +| `realtime.connected/disconnected` | Actor subscription added or removed | +| `payload_broadcast.failed` | One personalized payload failed; `payload` names it | Events describe local observations. Concurrent deletion, crashes, and failed exporters can omit events. Never infer exactly-once delivery from event counts. diff --git a/docs/operations.md b/docs/operations.md index 375d061..08dc5dd 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -230,7 +230,8 @@ for the same time emits nothing. ## Dead letters, retry, and redrive -A message that exhausts its attempts becomes a dead letter. An effect or a +A handler failure that exhausts its attempts becomes a dead letter. +Lease loss alone remains pending with [recovery backoff](configuration.md). An effect or a broadcast that exhausts its attempts stays in its own table with `status = 'dead'`. All three are read and retried through one receiver, which carries the kind: diff --git a/docs/parity.md b/docs/parity.md index 05be29e..77cfd2e 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -4,7 +4,7 @@ This ledger tracks capability parity with the Ruby `solid_objects` gem. Parity preserves a capability and its correctness or security boundary. It does not copy a Rails API into Node. -Reference: Ruby `solid_objects` 0.17.4. The JavaScript package began at the Ruby +Reference: Ruby `solid_objects` 0.17.5. The JavaScript package began at the Ruby design's `0.12` capability generation; that version number did not imply earlier JavaScript releases. @@ -48,28 +48,28 @@ SQL waiters read results and status from one statement. Ruby already checks completion and returns the result from the same loaded message; no Ruby change is needed for the JavaScript stale-result race fix. -| Capability | Status | TypeScript shape or remaining work | -| ------------------------------------------------------------------------------------------------------------- | ------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Actor registry, durable identity, JSON state, and adjacent state migrations | Native | Ordinary classes, static actor types, inferred state, explicit migrations, and isolated runtime context across every actor-instance callback. | -| Query and projection purity | Native | Both runtimes reject state mutation and staged effects, recovery checks, commit actions, reminders, or outbound messages in queries, observable projections, and personalized payloads. Projection guards compare the complete staged work, including same-count replacements. Query and observable violations fail terminally with `QueryMutatedState`; payload violations fail only that payload. Snapshot projection reads enforce the same rule. | -| JSON property preservation | Native | Reserved property names such as `__proto__`, `constructor`, and `prototype` remain ordinary data, including through size checks and immutable copies. Both suites consume `compatibility/json-values.json`. | -| Fluent committed calls and background delivery | Native | `await reference.operation()` and `reference.send.operation()`. | -| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native | Both runtimes renew between turns when due and retain the fence while a claim remains, allowing expired-turn recovery in mailbox order even with one allowed attempt. Reclaiming increments the attempt count; lease loss alone does not dead-letter a message. Relational ready/claimed membership tables, distinct generated request IDs and caller idempotency keys, durable history, adapter-appropriate sequence locking, and PostgreSQL/MySQL row locks held from fence validation through commit. | -| Domain rejection and strict poison ordering | Native | Rejections accept JavaScript identifier-style codes and roll back without retry; invalid codes fail terminally, while retryable failures block later operations until completion or dead-lettering. | -| Bounded activation passes and hot-actor fairness | Native | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting. | -| Bounded claim candidate scan | Native | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race. | -| Backpressure and payload caps | Partial | Serialization enforces a shared maximum JSON nesting depth, raising `InvalidPayload`, and an optional caller-supplied `maxBytes` limit, raising `PayloadTooLarge`; keyed reminder names are bounded to 255 characters in JS and 191 in Ruby. These existing database limits remain runtime-specific; use at most 191 ASCII characters for a name shared across runtimes. Distributed per-actor rate limits, global admission control, and cache-capacity eviction are not planned in either runtime. They are hot, request-path, and loss-tolerant, so one durable ordered message per check is the wrong shape. Solid Objects Pro answers them with grouped and ephemeral operations, which [fit](fit.md) describes. | -| Idle activation cache | Native | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown. Overlapping synchronous caller passes own separate workers, preserving staged intents and concurrent execution across actors; Ruby already isolates each invocation's activation. | -| Transactional effects and outcome operations | Native | At-least-once handlers receive immutable stable effect, attempt, source-message, and actor identity; success and failure operations also receive the originally staged arguments for correlation. Typed callback envelopes are exported. | -| Actor-to-actor delivery | Native | `sendTo(reference).operation()` stages delivery in the source actor commit. | -| One-shot and recurring reminders | Native | Scheduling, cancellation by name/key/handle, read-your-writes inspection, catch-up policy, stale-claim recovery, pausing, and authorized resume. Cancellation cannot recall an already-enqueued occurrence. | -| Same-database commit actions | Native | Registered actions receive source-message identity, mailbox sequence, activation generation, and the fenced transaction connection. | -| Ambient transaction rejection | Native | Committed calls and message waits fail before blocking when the current async context already owns a transaction on the Solid Objects adapter. | -| Direct application-write isolation during actor code | Partial | `guardApplicationDatabase()` fails closed for operations, projections, migrations, and commit actions; only the supplied fenced commit-action connection may write. Unwrapped clients cannot be intercepted. | -| Committed snapshots | Native | `snapshot()` returns authorized persisted fields and inferred getters from one read-only committed state image; realtime replay reads explicit observables without mailbox history. | -| Actor destruction and incarnation fencing | Native | Authorized cascading deletion creates a fresh instance ID on recreation; an authorized waiter receives `ActorDestroyed` when that incarnation disappears. | -| Result recovery and sync timeout diagnostics | Native | Status, result, outcome, and wait reauthorize the original stored operation and arguments. All delivery modes retain immutable JSON results under the result-size cap. Result reads raise terminal rejection/failure errors; outcome returns them as data. Whole-call deadlines distinguish enqueue, wait, database, activation, and mailbox blockers. | -| Result lookup by request ID and idempotency key | Native | `runtime.findBy({ requestId })` and `reference.findBy({ idempotencyKey })` rebuild a `MessageReference`, authorized with the hook the original call ran. An actor remembers the keys of its own finished turns, so a key lookup separates a pruned message from one that never existed. | +| Capability | Status | TypeScript shape or remaining work | +| ------------------------------------------------------------------------------------------------------------- | ------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Actor registry, durable identity, JSON state, and adjacent state migrations | Native | Ordinary classes, static actor types, inferred state, explicit migrations, and isolated runtime context across every actor-instance callback. | +| Query and projection purity | Native | Both runtimes reject state mutation and staged effects, recovery checks, commit actions, reminders, or outbound messages in queries, observable projections, and personalized payloads. Projection guards compare the complete staged work, including same-count replacements. Query and observable violations fail terminally with `QueryMutatedState`; payload violations fail only that payload. Snapshot projection reads enforce the same rule. | +| JSON property preservation | Native | Reserved property names such as `__proto__`, `constructor`, and `prototype` remain ordinary data, including through size checks and immutable copies. Both suites consume `compatibility/json-values.json`. | +| Fluent committed calls and background delivery | Native | `await reference.operation()` and `reference.send.operation()`. | +| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native | Both runtimes renew between turns when due and retain the fence while a claim remains, allowing expired-turn recovery in mailbox order even with one allowed attempt. Reclaiming increments the attempt count; lease loss alone does not dead-letter a message. Recovery after attempt one is immediate; subsequent interruptions use the existing retry delay with total attempts minus one, stored in ready membership. Mailbox order and other-actor progress are preserved. Both runtimes emit `activation.lost` with the message ID and attempt. JS busy workers and drains yield to timers and input between passes; Ruby uses threads and waits after lost turns. Relational ready/claimed membership tables, distinct generated request IDs and caller idempotency keys, durable history, adapter-appropriate sequence locking, and PostgreSQL/MySQL row locks held from fence validation through commit. | +| Domain rejection and strict poison ordering | Native | Rejections accept JavaScript identifier-style codes and roll back without retry; invalid codes fail terminally, while retryable failures block later operations until completion or dead-lettering. | +| Bounded activation passes and hot-actor fairness | Native | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting. | +| Bounded claim candidate scan | Native | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race. | +| Backpressure and payload caps | Partial | Serialization enforces a shared maximum JSON nesting depth, raising `InvalidPayload`, and an optional caller-supplied `maxBytes` limit, raising `PayloadTooLarge`; keyed reminder names are bounded to 255 characters in JS and 191 in Ruby. These existing database limits remain runtime-specific; use at most 191 ASCII characters for a name shared across runtimes. Distributed per-actor rate limits, global admission control, and cache-capacity eviction are not planned in either runtime. They are hot, request-path, and loss-tolerant, so one durable ordered message per check is the wrong shape. Solid Objects Pro answers them with grouped and ephemeral operations, which [fit](fit.md) describes. | +| Idle activation cache | Native | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown. Overlapping synchronous caller passes own separate workers, preserving staged intents and concurrent execution across actors; Ruby already isolates each invocation's activation. | +| Transactional effects and outcome operations | Native | At-least-once handlers receive immutable stable effect, attempt, source-message, and actor identity; success and failure operations also receive the originally staged arguments for correlation. Typed callback envelopes are exported. | +| Actor-to-actor delivery | Native | `sendTo(reference).operation()` stages delivery in the source actor commit. | +| One-shot and recurring reminders | Native | Scheduling, cancellation by name/key/handle, read-your-writes inspection, catch-up policy, stale-claim recovery, pausing, and authorized resume. Cancellation cannot recall an already-enqueued occurrence. | +| Same-database commit actions | Native | Registered actions receive source-message identity, mailbox sequence, activation generation, and the fenced transaction connection. | +| Ambient transaction rejection | Native | Committed calls and message waits fail before blocking when the current async context already owns a transaction on the Solid Objects adapter. | +| Direct application-write isolation during actor code | Partial | `guardApplicationDatabase()` fails closed for operations, projections, migrations, and commit actions; only the supplied fenced commit-action connection may write. Unwrapped clients cannot be intercepted. | +| Committed snapshots | Native | `snapshot()` returns authorized persisted fields and inferred getters from one read-only committed state image; realtime replay reads explicit observables without mailbox history. | +| Actor destruction and incarnation fencing | Native | Authorized cascading deletion creates a fresh instance ID on recreation; an authorized waiter receives `ActorDestroyed` when that incarnation disappears. | +| Result recovery and sync timeout diagnostics | Native | Status, result, outcome, and wait reauthorize the original stored operation and arguments. All delivery modes retain immutable JSON results under the result-size cap. Result reads raise terminal rejection/failure errors; outcome returns them as data. Whole-call deadlines distinguish enqueue, wait, database, activation, and mailbox blockers. | +| Result lookup by request ID and idempotency key | Native | `runtime.findBy({ requestId })` and `reference.findBy({ idempotencyKey })` rebuild a `MessageReference`, authorized with the hook the original call ran. An actor remembers the keys of its own finished turns, so a key lookup separates a pruned message from one that never existed. | Effect callback envelopes are typed with `EffectFailurePayload`, `EffectSuccessPayload`, and `SerializedError` in both SQL and Cloudflare. diff --git a/docs/virtual-actors.md b/docs/virtual-actors.md index 091e138..1153e18 100644 --- a/docs/virtual-actors.md +++ b/docs/virtual-actors.md @@ -213,6 +213,10 @@ run again without harm. after a crash or a lost lease. - Ordered turns do not cancel stale JavaScript. Fencing stops a stale activation from a commit, but that code can continue to run. +- An interrupted first attempt recovers immediately. Repeated interruptions + use growing, durable retry delays while keeping mailbox order. Lease loss + alone has no attempt limit; use instrumentation and an application stopping + policy when a hard bound is required. - External effects can run more than once. Use the stable effect ID, or another durable key, as the idempotency key at the provider. - There are no transactions across actor identities, and there is no replay of diff --git a/package.json b/package.json index 7467fd0..f1df76f 100644 --- a/package.json +++ b/package.json @@ -118,8 +118,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/activation-lease.test.ts test/caller-lifecycle.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", - "test:mysql": "vitest run test/mysql.test.ts test/activation-lease.test.ts test/caller-lifecycle.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", + "test:postgresql": "vitest run test/postgresql.test.ts test/activation-lease.test.ts test/activation-recovery-backoff.test.ts test/caller-lifecycle.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.test.ts", + "test:mysql": "vitest run test/mysql.test.ts test/activation-lease.test.ts test/activation-recovery-backoff.test.ts test/caller-lifecycle.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts test/transmit-ordering.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/broadcast-worker.ts b/src/broadcast-worker.ts index f324365..417fe85 100644 --- a/src/broadcast-worker.ts +++ b/src/broadcast-worker.ts @@ -2,6 +2,7 @@ import { randomUUID } from "./platform/uuid.js" import type { SolidObjectsRuntime } from "./runtime.js" import { withProcessHeartbeat } from "./worker.js" import { PollingBackoff } from "./polling-backoff.js" +import { yieldToEventLoop } from "./yield-to-event-loop.js" export class BroadcastWorker { readonly processId = randomUUID() @@ -45,6 +46,7 @@ export class BroadcastWorker { const count = await this.runOnce() if (count === 0) break processed += count + await yieldToEventLoop() } return processed } @@ -57,6 +59,8 @@ export class BroadcastWorker { const processed = await this.runOnce() if (processed > 0) { this.pollingBackoff.reset("work") + if (signal.aborted || this.stopping) break + await yieldToEventLoop() continue } const notified = await wakeUp.wait({ diff --git a/src/effect-worker.ts b/src/effect-worker.ts index adb4b31..943f3c1 100644 --- a/src/effect-worker.ts +++ b/src/effect-worker.ts @@ -2,6 +2,7 @@ import { randomUUID } from "./platform/uuid.js" import type { SolidObjectsRuntime } from "./runtime.js" import { withProcessHeartbeat } from "./worker.js" import { PollingBackoff } from "./polling-backoff.js" +import { yieldToEventLoop } from "./yield-to-event-loop.js" export class EffectWorker { readonly processId = randomUUID() @@ -45,6 +46,7 @@ export class EffectWorker { const count = await this.runOnce() if (count === 0) break processed += count + await yieldToEventLoop() } return processed } @@ -57,6 +59,8 @@ export class EffectWorker { const processed = await this.runOnce() if (processed > 0) { this.pollingBackoff.reset("work") + if (signal.aborted || this.stopping) break + await yieldToEventLoop() continue } const notified = await wakeUp.wait({ diff --git a/src/platform/node.ts b/src/platform/node.ts index a880561..d46c3ff 100644 --- a/src/platform/node.ts +++ b/src/platform/node.ts @@ -2,8 +2,10 @@ import { AsyncLocalStorage } from "node:async_hooks" import { hostname } from "node:os" import { registerContextStoreFactory } from "./context-store.js" import { registerHostIdentity } from "./host-identity.js" +import { registerEventLoopYield } from "../yield-to-event-loop.js" registerContextStoreFactory(() => new AsyncLocalStorage()) +registerEventLoopYield(() => new Promise((resolve) => setTimeout(() => setImmediate(resolve), 0))) registerHostIdentity({ hostname: hostname(), hostProcessId: process.pid, diff --git a/src/reminder-scheduler.ts b/src/reminder-scheduler.ts index 2c21706..6e52317 100644 --- a/src/reminder-scheduler.ts +++ b/src/reminder-scheduler.ts @@ -2,6 +2,7 @@ import { randomUUID } from "./platform/uuid.js" import { UnknownOperation } from "./errors.js" import type { SolidObjectsRuntime } from "./runtime.js" import { PollingBackoff } from "./polling-backoff.js" +import { yieldToEventLoop } from "./yield-to-event-loop.js" export class ReminderScheduler { readonly processId = randomUUID() @@ -66,6 +67,7 @@ export class ReminderScheduler { const count = await this.runOnce() if (count === 0) break processed += count + await yieldToEventLoop() } return processed } @@ -78,6 +80,8 @@ export class ReminderScheduler { const processed = await this.runOnce() if (processed > 0) { this.pollingBackoff.reset("work") + if (signal.aborted || this.stopping) break + await yieldToEventLoop() continue } const notified = await wakeUp.wait({ diff --git a/src/repository.ts b/src/repository.ts index f3da4cf..c1db1b7 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -227,7 +227,12 @@ export class Repository { `INSERT INTO ${this.table("ready_messages")} (message_id, instance_id, sequence, available_at_ms) VALUES (?, ?, ?, ?) ON CONFLICT(message_id) DO NOTHING`, - [claim.message_id, claim.instance_id, currentClaim.sequence, now], + [ + claim.message_id, + claim.instance_id, + currentClaim.sequence, + await this.recoveryAvailableAt({ connection, messageId: claim.message_id, now }), + ], ) const ownsActivation = instance.activation_owner_id === processId && @@ -2211,6 +2216,22 @@ export class Repository { ) } + private async recoveryAvailableAt(options: { + connection: DatabaseConnection + messageId: string + now: number + }): Promise { + const { connection, messageId, now } = options + const message = await connection.get<{ attempt_count: number | bigint }>( + `SELECT attempt_count FROM ${this.table("messages")} WHERE id = ?`, + [messageId], + ) + if (!message) throw new Error("claimed message does not exist") + const attempt = Number(message.attempt_count) + if (attempt <= 1) return now + return now + this.settings.retryDelayMilliseconds(attempt - 1) + } + private async recoverExpiredClaims(connection: DatabaseConnection, now: number): Promise { const expired = await connection.all<{ message_id: string @@ -2258,7 +2279,12 @@ export class Repository { await connection.run( `INSERT INTO ${this.table("ready_messages")}(message_id, instance_id, sequence, available_at_ms) VALUES (?, ?, ?, ?) ON CONFLICT(message_id) DO NOTHING`, - [claim.message_id, claim.instance_id, currentClaim.sequence, now], + [ + claim.message_id, + claim.instance_id, + currentClaim.sequence, + await this.recoveryAvailableAt({ connection, messageId: claim.message_id, now }), + ], ) const released = await connection.run( `UPDATE ${this.table("instances")} diff --git a/src/test-helper.ts b/src/test-helper.ts index 30f844f..7da8ad7 100644 --- a/src/test-helper.ts +++ b/src/test-helper.ts @@ -1,6 +1,7 @@ import type { SolidObjectsRuntime } from "./runtime.js" import type { LongRunningComponent } from "./types.js" import { broadcastsEnabled } from "./configuration.js" +import { yieldToEventLoop } from "./yield-to-event-loop.js" export type TestHelperRole = "actors" | "effects" | "reminders" | "broadcasts" @@ -40,7 +41,11 @@ export class SolidObjectsTestHelper { try { for (let pass = 0; pass < maxPasses; pass += 1) { let processed = 0 - for (const runner of sequence) processed += await runner.runOnce() + for (const runner of sequence) { + const count = await runner.runOnce() + processed += count + if (count > 0) await yieldToEventLoop() + } if (processed === 0) return total total += processed } @@ -70,6 +75,7 @@ export class SolidObjectsTestHelper { const count = await scheduler.runOnce({ now: options.now }) if (count === 0) return processed processed += count + await yieldToEventLoop() } return processed } finally { diff --git a/src/worker.ts b/src/worker.ts index d3cf2b1..0f2ac07 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -9,6 +9,7 @@ import { import type { ActivationLease, ClaimedTurn } from "./records.js" import type { SolidObjectsRuntime } from "./runtime.js" import { PollingBackoff } from "./polling-backoff.js" +import { yieldToEventLoop } from "./yield-to-event-loop.js" interface CachedActivation { actor: Actor @@ -132,6 +133,7 @@ export class Worker { const count = await this.runOnce() if (count === 0) break processed += count + await yieldToEventLoop() } return processed } finally { @@ -147,6 +149,8 @@ export class Worker { const processed = await this.runOnce() if (processed > 0) { this.pollingBackoff.reset("work") + if (signal.aborted || this.stopping) break + await yieldToEventLoop() continue } const notified = await wakeUp.wait({ diff --git a/src/yield-to-event-loop.ts b/src/yield-to-event-loop.ts new file mode 100644 index 0000000..2665cde --- /dev/null +++ b/src/yield-to-event-loop.ts @@ -0,0 +1,11 @@ +let yieldToHost = (): Promise => { + return new Promise((resolve) => setTimeout(resolve, 0)) +} + +export function registerEventLoopYield(callback: () => Promise): void { + yieldToHost = callback +} + +export function yieldToEventLoop(): Promise { + return yieldToHost() +} diff --git a/test/activation-recovery-backoff.test.ts b/test/activation-recovery-backoff.test.ts new file mode 100644 index 0000000..5b48848 --- /dev/null +++ b/test/activation-recovery-backoff.test.ts @@ -0,0 +1,236 @@ +import { afterEach, expect, it } from "vitest" +import type { InstrumentationEvent } from "../src/configuration.js" +import { expectPortableAttributes } from "./support/portable-telemetry.js" +import { Actor } from "../src/actor.js" +import { mysql } from "../src/database/mysql.js" +import { postgresql } from "../src/database/postgresql.js" +import { sqlite } from "../src/database/sqlite.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { controlledDatabaseClock } from "./support/fenced-commit.js" + +class RecoveryCounter extends Actor { + static override readonly actorType = "RecoveryBackoffCounter" + static beforeCommit: (label: string) => void = () => {} + count = 0 + + increment(options: { label: string }): number { + this.count += 1 + this.emit("applied", { arguments: options }) + RecoveryCounter.beforeCommit(options.label) + return this.count + } +} + +let events: InstrumentationEvent[] +let runtime: SolidObjectsRuntime +let clock: ReturnType + +afterEach(async () => { + await runtime?.repository.resetForTesting() + await runtime?.close() + RecoveryCounter.beforeCommit = () => {} +}) + +async function setup( + options: { retryDelayMilliseconds?: (attempt: number) => number; maxAttempts?: number } = {}, +): Promise { + const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL + const database = connectionString?.startsWith("postgres") + ? postgresql({ connectionString }) + : connectionString?.startsWith("mysql") + ? mysql({ connectionString }) + : sqlite({ path: ":memory:" }) + clock = controlledDatabaseClock(database) + events = [] + runtime = createRuntime({ + instrumentation: (event) => events.push(event), + database: clock.database, + tableNamePrefix: "backoff_test_", + leaseDurationMilliseconds: 200, + leaseRenewalIntervalMilliseconds: 50, + maxAttempts: 1, + authorizeMessage: () => true, + authorizeQuery: () => true, + ...options, + }) + runtime.register(RecoveryCounter) + runtime.registerEffect("applied", async () => {}) + await runtime.install() + await runtime.repository.resetForTesting() + RecoveryCounter.beforeCommit = (label) => { + if (label === "slow") clock.advanceMilliseconds(201) + } +} + +async function remainingDelay(messageId: string): Promise { + return runtime.settings.database.connection(async (connection) => { + const ready = await connection.get<{ available_at_ms: number | bigint }>( + `SELECT available_at_ms FROM ${runtime.repository.table("ready_messages")} WHERE message_id = ?`, + [messageId], + ) + if (!ready) throw new Error("message has no ready membership") + return Number(ready.available_at_ms) - (await connection.nowMilliseconds()) + }) +} + +it("recovers once immediately then backs off without blocking other actors or reordering", async () => { + await setup() + const worker = runtime.worker() + const reference = runtime.ref(RecoveryCounter, "slow") + const first = await reference.send.increment({ label: "slow" }) + const later = await reference.send.increment({ label: "later" }) + try { + expect(await worker.runOnce()).toBe(1) + expect(await worker.runOnce()).toBe(1) + const other = await runtime.ref(RecoveryCounter, "other").send.increment({ label: "other" }) + expect(await worker.runOnce()).toBe(1) + expect(await other.result()).toBe(1) + expect(await first.status()).toBe("ready") + expect(await later.status()).toBe("ready") + expect(await remainingDelay(first.id)).toBe(1_000) + expect((await reference.snapshot()).count).toBe(0) + + clock.advanceMilliseconds(999) + expect(await worker.runOnce()).toBe(0) + clock.advanceMilliseconds(1) + expect(await worker.runOnce()).toBe(1) + expect(await worker.runOnce()).toBe(0) + expect(await remainingDelay(first.id)).toBe(2_000) + expect(Number((await runtime.repository.findMessage(first.id))?.attempt_count)).toBe(3) + + RecoveryCounter.beforeCommit = () => {} + clock.advanceMilliseconds(2_000) + expect(await worker.runOnce()).toBe(2) + expect(await first.result()).toBe(1) + expect(await later.result()).toBe(2) + const effects = await runtime.settings.database.connection((connection) => + connection.all<{ arguments: string }>( + `SELECT arguments FROM ${runtime.repository.table("effects")}`, + ), + ) + expect( + effects.map((effect) => (JSON.parse(effect.arguments) as { label: string }).label).sort(), + ).toEqual(["later", "other", "slow"]) + expect(await first.status()).toBe("completed") + const losses = events.filter((event) => event.name === "solid_objects.activation.lost") + expect(losses.map((event) => event.attempt)).toEqual([1, 2, 3]) + expect(losses.every((event) => event.messageId === first.id)).toBe(true) + losses.forEach(expectPortableAttributes) + } finally { + await worker.stop() + } +}) + +it.each(["stop", "stale cleanup", "replacement worker"])( + "preserves custom recovery delays through %s", + async (recovery) => { + const delayArguments: number[] = [] + await setup({ + retryDelayMilliseconds: (attempt) => { + delayArguments.push(attempt) + return attempt * 25 + }, + }) + const firstWorker = runtime.worker() + const replacement = runtime.worker() + const message = await runtime.ref(RecoveryCounter, "one").send.increment({ label: "slow" }) + try { + await firstWorker.runOnce() + await firstWorker.runOnce() + if (recovery === "stop") await firstWorker.stop() + if (recovery === "stale cleanup") { + clock.advanceMilliseconds(runtime.settings.processAliveThresholdMilliseconds + 1) + await runtime.repository.cleanupStaleProcesses() + } + expect(await replacement.runOnce()).toBe(0) + expect(delayArguments).toEqual([1]) + expect(await remainingDelay(message.id)).toBe(25) + expect(await message.status()).toBe("ready") + await firstWorker.stop() + expect(await remainingDelay(message.id)).toBe(25) + + clock.advanceMilliseconds(25) + expect(await replacement.runOnce()).toBe(1) + expect(await replacement.runOnce()).toBe(0) + expect(delayArguments).toEqual([1, 2]) + expect(await remainingDelay(message.id)).toBe(50) + expect(Number((await runtime.repository.findMessage(message.id))?.attempt_count)).toBe(3) + } finally { + await firstWorker.stop() + await replacement.stop() + } + }, +) + +it("keeps handler failures terminal with one allowed attempt", async () => { + await setup() + RecoveryCounter.beforeCommit = () => { + throw new Error("handler failed") + } + const worker = runtime.worker() + try { + const message = await runtime.ref(RecoveryCounter, "failure").send.increment({ label: "fail" }) + await worker.runOnce() + expect(await message.status()).toBe("dead") + expect(Number((await runtime.repository.findMessage(message.id))?.attempt_count)).toBe(1) + } finally { + await worker.stop() + } +}) + +it("caps default recovery waits at sixty seconds without dead-lettering lease loss", async () => { + await setup() + const worker = runtime.worker() + try { + const message = await runtime + .ref(RecoveryCounter, "persistent") + .send.increment({ label: "slow" }) + await worker.runOnce() + await worker.runOnce() + for (const delay of [1_000, 2_000, 4_000, 8_000, 16_000, 32_000, 60_000, 60_000]) { + expect(await worker.runOnce()).toBe(0) + expect(await remainingDelay(message.id)).toBe(delay) + expect(await message.status()).toBe("ready") + clock.advanceMilliseconds(delay) + expect(await worker.runOnce()).toBe(1) + } + } finally { + await worker.stop() + } +}) + +it("uses total claims for recovery after an earlier handler failure", async () => { + const delayArguments: number[] = [] + await setup({ + maxAttempts: 3, + retryDelayMilliseconds: (attempt) => { + delayArguments.push(attempt) + return attempt * 25 + }, + }) + const worker = runtime.worker() + try { + const message = await runtime.ref(RecoveryCounter, "mixed").send.increment({ label: "slow" }) + RecoveryCounter.beforeCommit = () => { + throw new Error("handler failed") + } + await worker.runOnce() + expect(delayArguments).toEqual([1]) + clock.advanceMilliseconds(25) + RecoveryCounter.beforeCommit = () => clock.advanceMilliseconds(201) + await worker.runOnce() + expect(await worker.runOnce()).toBe(0) + expect(delayArguments).toEqual([1, 1]) + expect(await remainingDelay(message.id)).toBe(25) + expect(Number((await runtime.repository.findMessage(message.id))?.attempt_count)).toBe(2) + clock.advanceMilliseconds(25) + RecoveryCounter.beforeCommit = () => { + throw new Error("handler failed again") + } + await worker.runOnce() + expect(await message.status()).toBe("dead") + expect(Number((await runtime.repository.findMessage(message.id))?.attempt_count)).toBe(3) + } finally { + await worker.stop() + } +}) diff --git a/test/fixtures/busy-worker.ts b/test/fixtures/busy-worker.ts new file mode 100644 index 0000000..b254f97 --- /dev/null +++ b/test/fixtures/busy-worker.ts @@ -0,0 +1,145 @@ +import { writeSync } from "node:fs" +import { Actor, createRuntime } from "../../src/index.js" +import { sqlite } from "../../src/database/sqlite.js" + +const scenario = process.argv[2] ?? "lease-timer" +const role = process.argv[3] ?? "actors" +const execution = process.argv[4] ?? "run" +const leaseScenario = scenario.startsWith("lease-") +const messageCount = Number(process.env.BUSY_MESSAGE_COUNT ?? 100) +const turnMilliseconds = Number(process.env.BUSY_TURN_MILLISECONDS ?? 2) +const controller = new AbortController() +let completed = 0 +let ticks = 0 +let completedAtFirstTick = -1 +let inputReceived = false +let lastTick = 0 +let maximumTimerGapMilliseconds = 0 +const attempts: number[] = [] +const ticksAtLoss: number[] = [] + +function block(milliseconds: number): void { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, milliseconds) +} + +function finishWork(): void { + block(turnMilliseconds) + completed += 1 + if (completed === messageCount) controller.abort() +} + +class BusyActor extends Actor { + static override readonly actorType = "BusyActor" + count = 0 + + increment(): void { + if (leaseScenario) block(400) + this.count += 1 + } + + stage(): void { + for (let index = 0; index < messageCount; index += 1) { + if (role === "effects") this.emit("busy-effect") + if (role === "reminders") this.schedule({ at: new Date(0), key: String(index) }).increment() + } + } + + override observables(): Record { + return { count: this.count } + } +} + +const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + maxAttempts: 1, + leaseDurationMilliseconds: leaseScenario ? 100 : 30_000, + leaseRenewalIntervalMilliseconds: leaseScenario ? 30 : 5_000, + workerCount: role === "actors" ? 1 : 0, + effectWorkerCount: role === "effects" ? 1 : 0, + reminderSchedulerCount: role === "reminders" ? 1 : 0, + broadcastWorkerCount: role === "broadcasts" ? 1 : 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + authorizeMessage: () => true, + authorizeQuery: () => true, + wakeUp: "in_process", + logger: { debug() {}, info() {}, warn() {}, error() {} }, + ...(role === "broadcasts" ? { broadcast: async () => finishWork() } : {}), + instrumentation: (event) => { + if (event.name === "solid_objects.activation.lost") { + attempts.push(event.attempt) + ticksAtLoss.push(ticks) + writeSync(1, `${JSON.stringify({ event: "lost", attempts, ticks })}\n`) + if (scenario === "lease-callback" && attempts.length === 3) controller.abort() + } + if (!lastTick || leaseScenario) return + if (role === "actors" && event.name === "solid_objects.message.completed") finishWork() + if (role === "reminders" && event.name === "solid_objects.reminder.enqueued") finishWork() + }, +}) +runtime.register(BusyActor) +runtime.registerEffect("busy-effect", async () => finishWork()) +await runtime.install() +const reference = runtime.ref(BusyActor, "one") +if (role === "actors" || role === "broadcasts") { + for (let index = 0; index < (leaseScenario ? 1 : messageCount); index += 1) { + await reference.send.increment() + } +} else { + await reference.send.stage() +} +if (role !== "actors") { + const setupWorker = runtime.worker() + await setupWorker.runUntilIdle() + await setupWorker.stop() +} + +const startedAt = performance.now() +lastTick = startedAt +const timer = setInterval(() => { + if (ticks === 0) completedAtFirstTick = completed + const now = performance.now() + maximumTimerGapMilliseconds = Math.max(maximumTimerGapMilliseconds, now - lastTick) + lastTick = now + ticks += 1 +}, 20) +const abortTimer = + scenario === "lease-timer" ? setTimeout(() => controller.abort(), 900) : undefined +process.stdin.once("data", () => { + inputReceived = true +}) +writeSync(1, `${JSON.stringify({ event: "ready" })}\n`) +try { + if (execution === "run") await runtime.run(controller.signal) + else if (execution === "drain") + await runtime.testing.drain({ + roles: [role as "actors" | "effects" | "reminders" | "broadcasts"], + }) + else if (execution === "due-reminders") await runtime.testing.runDueReminders({ now: new Date() }) + else { + const worker = + role === "actors" + ? runtime.worker() + : role === "effects" + ? runtime.effectWorker() + : role === "reminders" + ? runtime.reminderScheduler() + : runtime.broadcastWorker() + try { + await worker.runUntilIdle() + } finally { + await worker.stop() + } + } + const elapsedMilliseconds = performance.now() - startedAt + maximumTimerGapMilliseconds = Math.max(maximumTimerGapMilliseconds, performance.now() - lastTick) + writeSync( + 1, + `${JSON.stringify({ event: "finished", elapsedMilliseconds, maximumTimerGapMilliseconds, ticks, completedAtFirstTick, inputReceived, attempts, ticksAtLoss, completed, count: (await reference.snapshot()).count })}\n`, + ) +} finally { + clearInterval(timer) + clearTimeout(abortTimer) + process.stdin.destroy() + await runtime.close() +} diff --git a/test/worker-fairness.test.ts b/test/worker-fairness.test.ts new file mode 100644 index 0000000..64f7414 --- /dev/null +++ b/test/worker-fairness.test.ts @@ -0,0 +1,110 @@ +import { spawn } from "node:child_process" +import { fileURLToPath } from "node:url" +import { expect, it } from "vitest" + +interface WorkerResult { + event: string + elapsedMilliseconds: number + maximumTimerGapMilliseconds: number + ticks: number + completedAtFirstTick: number + inputReceived: boolean + attempts: number[] + ticksAtLoss: number[] + completed: number + count: number +} + +async function runWorker(options: { + scenario: string + role?: string + execution?: string +}): Promise { + const child = spawn( + process.execPath, + [ + "--experimental-transform-types", + "--disable-warning=ExperimentalWarning", + "--import", + fileURLToPath(new URL("./support/register-source-hooks.mjs", import.meta.url)), + fileURLToPath(new URL("./fixtures/busy-worker.ts", import.meta.url)), + options.scenario, + options.role ?? "actors", + options.execution ?? "run", + ], + { stdio: ["pipe", "pipe", "pipe"] }, + ) + let stdout = "" + let stderr = "" + let timedOut = false + child.stdout.on("data", (chunk: Buffer) => { + stdout += chunk.toString() + if (stdout.includes('"event":"ready"') && !child.stdin.writableEnded) child.stdin.end("input\n") + }) + child.stderr.on("data", (chunk: Buffer) => { + stderr += chunk.toString() + }) + child.stdin.on("error", () => {}) + const timeout = setTimeout(() => { + timedOut = true + child.kill("SIGKILL") + }, 4_000) + try { + const code = await new Promise((resolve, reject) => { + child.once("error", reject) + child.once("close", resolve) + }) + expect(timedOut, `child exceeded external time limit:\n${stdout}\n${stderr}`).toBe(false) + expect(code, stderr).toBe(0) + const result = stdout + .trim() + .split("\n") + .map((line) => JSON.parse(line) as WorkerResult) + .find((entry) => entry.event === "finished") + expect(result).toBeDefined() + return result! + } finally { + clearTimeout(timeout) + if (child.exitCode === null) child.kill("SIGKILL") + } +} + +it("lets timers and input run between lost attempts and honors a timer abort", async () => { + const result = await runWorker({ scenario: "lease-timer" }) + expect(result.ticks).toBeGreaterThan(0) + expect(result.inputReceived).toBe(true) + expect(result.ticksAtLoss[1]).toBeGreaterThan(result.ticksAtLoss[0]!) + expect(result.elapsedMilliseconds).toBeLessThan(2_500) + expect(result.count).toBe(0) +}, 6_000) + +it("stops at the third lost attempt when instrumentation aborts the run", async () => { + const result = await runWorker({ scenario: "lease-callback" }) + expect(result.attempts).toEqual([1, 2, 3]) + expect(result.elapsedMilliseconds).toBeLessThan(3_000) + expect(result.count).toBe(0) +}, 6_000) + +for (const role of ["actors", "effects", "reminders", "broadcasts"]) { + for (const execution of ["run", "until-idle", "drain"]) { + it(`lets timers and input run during continuous ${role} work with ${execution}`, async () => { + const result = await runWorker({ scenario: "short", role, execution }) + expect(result.completed).toBe(100) + expect(result.completedAtFirstTick).toBeLessThan(100) + expect(result.ticks).toBeGreaterThan(0) + expect(result.inputReceived).toBe(true) + }, 6_000) + } +} + +it("lets timers and input run while draining due reminders", async () => { + const result = await runWorker({ + scenario: "short", + role: "reminders", + execution: "due-reminders", + }) + expect(result.completed).toBe(100) + expect(result.completedAtFirstTick).toBeLessThan(100) + expect(result.ticks).toBeGreaterThan(0) + expect(result.inputReceived).toBe(true) +}, 6_000) From 96f21379812ffb369a7057e7e9a90a94a9b359c6 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 12:04:20 -0700 Subject: [PATCH 8/9] test: flatten adapter and worker selection Address review feedback by keeping regression fixture choices explicit. Preserve the adapter and worker mappings while removing nested conditionals. Validation: pnpm run format:check, pnpm run check, and pnpm test (660 passed, 32 conditional skips). --- CHANGELOG.md | 1 + test/activation-recovery-backoff.test.ts | 14 +++++++++----- test/caller-lifecycle.test.ts | 14 +++++++++----- test/fixtures/busy-worker.ts | 16 ++++++++-------- 4 files changed, 27 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 978ded5..a87d3a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ Explicitly stopped workers stay stopped. No schema or setting changes. - Cover caller concurrency and process retention on SQLite, PostgreSQL, and MySQL, with `maxAttempts: 1` and both default and single-message pass limits. + Keep adapter and worker-role selection explicit in the regression fixtures. - Drain recovered turns in the failure-recovery demo before checking committed state. Verify completion and allow additional fenced retries when a worker loses its lease on a delayed runner. diff --git a/test/activation-recovery-backoff.test.ts b/test/activation-recovery-backoff.test.ts index 5b48848..bf94690 100644 --- a/test/activation-recovery-backoff.test.ts +++ b/test/activation-recovery-backoff.test.ts @@ -5,6 +5,7 @@ import { Actor } from "../src/actor.js" import { mysql } from "../src/database/mysql.js" import { postgresql } from "../src/database/postgresql.js" import { sqlite } from "../src/database/sqlite.js" +import type { Database } from "../src/database/types.js" import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" import { controlledDatabaseClock } from "./support/fenced-commit.js" @@ -35,11 +36,14 @@ async function setup( options: { retryDelayMilliseconds?: (attempt: number) => number; maxAttempts?: number } = {}, ): Promise { const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL - const database = connectionString?.startsWith("postgres") - ? postgresql({ connectionString }) - : connectionString?.startsWith("mysql") - ? mysql({ connectionString }) - : sqlite({ path: ":memory:" }) + let database: Database + if (connectionString?.startsWith("postgres")) { + database = postgresql({ connectionString }) + } else if (connectionString?.startsWith("mysql")) { + database = mysql({ connectionString }) + } else { + database = sqlite({ path: ":memory:" }) + } clock = controlledDatabaseClock(database) events = [] runtime = createRuntime({ diff --git a/test/caller-lifecycle.test.ts b/test/caller-lifecycle.test.ts index 76f466f..f0fb993 100644 --- a/test/caller-lifecycle.test.ts +++ b/test/caller-lifecycle.test.ts @@ -4,6 +4,7 @@ import type { InstrumentationEvent } from "../src/configuration.js" import { mysql } from "../src/database/mysql.js" import { postgresql } from "../src/database/postgresql.js" import { sqlite } from "../src/database/sqlite.js" +import type { Database } from "../src/database/types.js" import { ActorDestroyed } from "../src/errors.js" import { configure, type SolidObjectsRuntime } from "../src/runtime.js" import { Worker } from "../src/worker.js" @@ -46,11 +47,14 @@ afterEach(async () => { async function setup(options: { maxMessagesPerActivationPass?: number } = {}): Promise { const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL - const database = connectionString?.startsWith("postgres") - ? postgresql({ connectionString }) - : connectionString?.startsWith("mysql") - ? mysql({ connectionString }) - : sqlite({ path: ":memory:" }) + let database: Database + if (connectionString?.startsWith("postgres")) { + database = postgresql({ connectionString }) + } else if (connectionString?.startsWith("mysql")) { + database = mysql({ connectionString }) + } else { + database = sqlite({ path: ":memory:" }) + } runtime = configure({ database, tableNamePrefix: "caller_test_", diff --git a/test/fixtures/busy-worker.ts b/test/fixtures/busy-worker.ts index b254f97..9ddfb3f 100644 --- a/test/fixtures/busy-worker.ts +++ b/test/fixtures/busy-worker.ts @@ -94,6 +94,13 @@ if (role !== "actors") { await setupWorker.stop() } +function workerForRole() { + if (role === "actors") return runtime.worker() + if (role === "effects") return runtime.effectWorker() + if (role === "reminders") return runtime.reminderScheduler() + return runtime.broadcastWorker() +} + const startedAt = performance.now() lastTick = startedAt const timer = setInterval(() => { @@ -117,14 +124,7 @@ try { }) else if (execution === "due-reminders") await runtime.testing.runDueReminders({ now: new Date() }) else { - const worker = - role === "actors" - ? runtime.worker() - : role === "effects" - ? runtime.effectWorker() - : role === "reminders" - ? runtime.reminderScheduler() - : runtime.broadcastWorker() + const worker = workerForRole() try { await worker.runUntilIdle() } finally { From ec3ddc3b3ad5364c3019ead4d6c4399f0b43f898 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Mon, 12 Oct 2026 12:12:22 -0700 Subject: [PATCH 9/9] fix: report lease loss in all turn outcomes Handle fenced failures and rejections like lost completions, preserving recovery and activation.lost callbacks. Flatten shutdown checks in all worker loops as requested in review. Before the fix, two new recovery tests raised LostActivation and two callback shutdown cases exceeded their external time limits. All pass now. Type, formatting and full SQLite/PostgreSQL/MySQL suites pass. --- CHANGELOG.md | 3 ++ docs/parity.md | 44 ++++++++-------- src/broadcast-worker.ts | 2 +- src/effect-worker.ts | 2 +- src/reminder-scheduler.ts | 2 +- src/runtime.ts | 67 +++++++++++++----------- src/worker.ts | 2 +- test/activation-recovery-backoff.test.ts | 43 +++++++++++++++ test/fixtures/busy-worker.ts | 7 ++- test/worker-fairness.test.ts | 16 +++--- 10 files changed, 122 insertions(+), 66 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a87d3a1..af9ae0b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,9 @@ and broadcast workers and drain helpers, including repeated lease loss. Timer-driven aborts can stop the runtime between attempts; an instrumentation abort stops before another pass. Synchronous handlers remain non-preemptible. +- Report `activation.lost` when a handler failure or rejection cannot commit + because its lease expired. These attempts recover with the same fencing and + backoff as lost completions, and instrumentation can stop the worker. - Give overlapping synchronous caller passes separate, reusable workers so cleanup cannot discard another turn's staged effects or other intents. diff --git a/docs/parity.md b/docs/parity.md index 77cfd2e..b897e22 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -48,28 +48,28 @@ SQL waiters read results and status from one statement. Ruby already checks completion and returns the result from the same loaded message; no Ruby change is needed for the JavaScript stale-result race fix. -| Capability | Status | TypeScript shape or remaining work | -| ------------------------------------------------------------------------------------------------------------- | ------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Actor registry, durable identity, JSON state, and adjacent state migrations | Native | Ordinary classes, static actor types, inferred state, explicit migrations, and isolated runtime context across every actor-instance callback. | -| Query and projection purity | Native | Both runtimes reject state mutation and staged effects, recovery checks, commit actions, reminders, or outbound messages in queries, observable projections, and personalized payloads. Projection guards compare the complete staged work, including same-count replacements. Query and observable violations fail terminally with `QueryMutatedState`; payload violations fail only that payload. Snapshot projection reads enforce the same rule. | -| JSON property preservation | Native | Reserved property names such as `__proto__`, `constructor`, and `prototype` remain ordinary data, including through size checks and immutable copies. Both suites consume `compatibility/json-values.json`. | -| Fluent committed calls and background delivery | Native | `await reference.operation()` and `reference.send.operation()`. | -| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native | Both runtimes renew between turns when due and retain the fence while a claim remains, allowing expired-turn recovery in mailbox order even with one allowed attempt. Reclaiming increments the attempt count; lease loss alone does not dead-letter a message. Recovery after attempt one is immediate; subsequent interruptions use the existing retry delay with total attempts minus one, stored in ready membership. Mailbox order and other-actor progress are preserved. Both runtimes emit `activation.lost` with the message ID and attempt. JS busy workers and drains yield to timers and input between passes; Ruby uses threads and waits after lost turns. Relational ready/claimed membership tables, distinct generated request IDs and caller idempotency keys, durable history, adapter-appropriate sequence locking, and PostgreSQL/MySQL row locks held from fence validation through commit. | -| Domain rejection and strict poison ordering | Native | Rejections accept JavaScript identifier-style codes and roll back without retry; invalid codes fail terminally, while retryable failures block later operations until completion or dead-lettering. | -| Bounded activation passes and hot-actor fairness | Native | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting. | -| Bounded claim candidate scan | Native | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race. | -| Backpressure and payload caps | Partial | Serialization enforces a shared maximum JSON nesting depth, raising `InvalidPayload`, and an optional caller-supplied `maxBytes` limit, raising `PayloadTooLarge`; keyed reminder names are bounded to 255 characters in JS and 191 in Ruby. These existing database limits remain runtime-specific; use at most 191 ASCII characters for a name shared across runtimes. Distributed per-actor rate limits, global admission control, and cache-capacity eviction are not planned in either runtime. They are hot, request-path, and loss-tolerant, so one durable ordered message per check is the wrong shape. Solid Objects Pro answers them with grouped and ephemeral operations, which [fit](fit.md) describes. | -| Idle activation cache | Native | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown. Overlapping synchronous caller passes own separate workers, preserving staged intents and concurrent execution across actors; Ruby already isolates each invocation's activation. | -| Transactional effects and outcome operations | Native | At-least-once handlers receive immutable stable effect, attempt, source-message, and actor identity; success and failure operations also receive the originally staged arguments for correlation. Typed callback envelopes are exported. | -| Actor-to-actor delivery | Native | `sendTo(reference).operation()` stages delivery in the source actor commit. | -| One-shot and recurring reminders | Native | Scheduling, cancellation by name/key/handle, read-your-writes inspection, catch-up policy, stale-claim recovery, pausing, and authorized resume. Cancellation cannot recall an already-enqueued occurrence. | -| Same-database commit actions | Native | Registered actions receive source-message identity, mailbox sequence, activation generation, and the fenced transaction connection. | -| Ambient transaction rejection | Native | Committed calls and message waits fail before blocking when the current async context already owns a transaction on the Solid Objects adapter. | -| Direct application-write isolation during actor code | Partial | `guardApplicationDatabase()` fails closed for operations, projections, migrations, and commit actions; only the supplied fenced commit-action connection may write. Unwrapped clients cannot be intercepted. | -| Committed snapshots | Native | `snapshot()` returns authorized persisted fields and inferred getters from one read-only committed state image; realtime replay reads explicit observables without mailbox history. | -| Actor destruction and incarnation fencing | Native | Authorized cascading deletion creates a fresh instance ID on recreation; an authorized waiter receives `ActorDestroyed` when that incarnation disappears. | -| Result recovery and sync timeout diagnostics | Native | Status, result, outcome, and wait reauthorize the original stored operation and arguments. All delivery modes retain immutable JSON results under the result-size cap. Result reads raise terminal rejection/failure errors; outcome returns them as data. Whole-call deadlines distinguish enqueue, wait, database, activation, and mailbox blockers. | -| Result lookup by request ID and idempotency key | Native | `runtime.findBy({ requestId })` and `reference.findBy({ idempotencyKey })` rebuild a `MessageReference`, authorized with the hook the original call ran. An actor remembers the keys of its own finished turns, so a key lookup separates a pruned message from one that never existed. | +| Capability | Status | TypeScript shape or remaining work | +| ------------------------------------------------------------------------------------------------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Actor registry, durable identity, JSON state, and adjacent state migrations | Native | Ordinary classes, static actor types, inferred state, explicit migrations, and isolated runtime context across every actor-instance callback. | +| Query and projection purity | Native | Both runtimes reject state mutation and staged effects, recovery checks, commit actions, reminders, or outbound messages in queries, observable projections, and personalized payloads. Projection guards compare the complete staged work, including same-count replacements. Query and observable violations fail terminally with `QueryMutatedState`; payload violations fail only that payload. Snapshot projection reads enforce the same rule. | +| JSON property preservation | Native | Reserved property names such as `__proto__`, `constructor`, and `prototype` remain ordinary data, including through size checks and immutable copies. Both suites consume `compatibility/json-values.json`. | +| Fluent committed calls and background delivery | Native | `await reference.operation()` and `reference.send.operation()`. | +| Ordered mailbox, sequence allocation, idempotency, retries, dead letters, leases, renewal, and fenced commits | Native | Both runtimes renew between turns when due and retain the fence while a claim remains, allowing expired-turn recovery in mailbox order even with one allowed attempt. Reclaiming increments the attempt count; lease loss alone does not dead-letter a message. Recovery after attempt one is immediate; subsequent interruptions use the existing retry delay with total attempts minus one, stored in ready membership. Mailbox order and other-actor progress are preserved. Both runtimes emit `activation.lost` with the message ID and attempt, including losses while saving a handler failure or rejection. JS busy workers and drains yield to timers and input between passes; Ruby uses threads and waits after lost turns. Relational ready/claimed membership tables, distinct generated request IDs and caller idempotency keys, durable history, adapter-appropriate sequence locking, and PostgreSQL/MySQL row locks held from fence validation through commit. | +| Domain rejection and strict poison ordering | Native | Rejections accept JavaScript identifier-style codes and roll back without retry; invalid codes fail terminally, while retryable failures block later operations until completion or dead-lettering. | +| Bounded activation passes and hot-actor fairness | Native | Configurable turn-count and elapsed-time budgets bound each pass, then move only that actor's already-due memberships behind actors already waiting. | +| Bounded claim candidate scan | Native | A configurable ordered scan continues to another ready actor when a worker loses the first candidate's lease race. | +| Backpressure and payload caps | Partial | Serialization enforces a shared maximum JSON nesting depth, raising `InvalidPayload`, and an optional caller-supplied `maxBytes` limit, raising `PayloadTooLarge`; keyed reminder names are bounded to 255 characters in JS and 191 in Ruby. These existing database limits remain runtime-specific; use at most 191 ASCII characters for a name shared across runtimes. Distributed per-actor rate limits, global admission control, and cache-capacity eviction are not planned in either runtime. They are hot, request-path, and loss-tolerant, so one durable ordered message per check is the wrong shape. Solid Objects Pro answers them with grouped and ephemeral operations, which [fit](fit.md) describes. | +| Idle activation cache | Native | Long-running workers retain hydrated actors under renewable fenced leases, restore public state after failed turns, and release on timeout, fairness yield, lease loss, or shutdown. Overlapping synchronous caller passes own separate workers, preserving staged intents and concurrent execution across actors; Ruby already isolates each invocation's activation. | +| Transactional effects and outcome operations | Native | At-least-once handlers receive immutable stable effect, attempt, source-message, and actor identity; success and failure operations also receive the originally staged arguments for correlation. Typed callback envelopes are exported. | +| Actor-to-actor delivery | Native | `sendTo(reference).operation()` stages delivery in the source actor commit. | +| One-shot and recurring reminders | Native | Scheduling, cancellation by name/key/handle, read-your-writes inspection, catch-up policy, stale-claim recovery, pausing, and authorized resume. Cancellation cannot recall an already-enqueued occurrence. | +| Same-database commit actions | Native | Registered actions receive source-message identity, mailbox sequence, activation generation, and the fenced transaction connection. | +| Ambient transaction rejection | Native | Committed calls and message waits fail before blocking when the current async context already owns a transaction on the Solid Objects adapter. | +| Direct application-write isolation during actor code | Partial | `guardApplicationDatabase()` fails closed for operations, projections, migrations, and commit actions; only the supplied fenced commit-action connection may write. Unwrapped clients cannot be intercepted. | +| Committed snapshots | Native | `snapshot()` returns authorized persisted fields and inferred getters from one read-only committed state image; realtime replay reads explicit observables without mailbox history. | +| Actor destruction and incarnation fencing | Native | Authorized cascading deletion creates a fresh instance ID on recreation; an authorized waiter receives `ActorDestroyed` when that incarnation disappears. | +| Result recovery and sync timeout diagnostics | Native | Status, result, outcome, and wait reauthorize the original stored operation and arguments. All delivery modes retain immutable JSON results under the result-size cap. Result reads raise terminal rejection/failure errors; outcome returns them as data. Whole-call deadlines distinguish enqueue, wait, database, activation, and mailbox blockers. | +| Result lookup by request ID and idempotency key | Native | `runtime.findBy({ requestId })` and `reference.findBy({ idempotencyKey })` rebuild a `MessageReference`, authorized with the hook the original call ran. An actor remembers the keys of its own finished turns, so a key lookup separates a pruned message from one that never existed. | Effect callback envelopes are typed with `EffectFailurePayload`, `EffectSuccessPayload`, and `SerializedError` in both SQL and Cloudflare. diff --git a/src/broadcast-worker.ts b/src/broadcast-worker.ts index 417fe85..14b999a 100644 --- a/src/broadcast-worker.ts +++ b/src/broadcast-worker.ts @@ -57,9 +57,9 @@ export class BroadcastWorker { while (!signal.aborted && !this.stopping) { const wakeUp = await this.runtime.watchWakeUp("broadcasts") const processed = await this.runOnce() + if (signal.aborted || this.stopping) break if (processed > 0) { this.pollingBackoff.reset("work") - if (signal.aborted || this.stopping) break await yieldToEventLoop() continue } diff --git a/src/effect-worker.ts b/src/effect-worker.ts index 943f3c1..6523b47 100644 --- a/src/effect-worker.ts +++ b/src/effect-worker.ts @@ -57,9 +57,9 @@ export class EffectWorker { while (!signal.aborted && !this.stopping) { const wakeUp = await this.runtime.watchWakeUp("effects") const processed = await this.runOnce() + if (signal.aborted || this.stopping) break if (processed > 0) { this.pollingBackoff.reset("work") - if (signal.aborted || this.stopping) break await yieldToEventLoop() continue } diff --git a/src/reminder-scheduler.ts b/src/reminder-scheduler.ts index 6e52317..6cf429e 100644 --- a/src/reminder-scheduler.ts +++ b/src/reminder-scheduler.ts @@ -78,9 +78,9 @@ export class ReminderScheduler { while (!signal.aborted && !this.stopping) { const wakeUp = await this.runtime.watchWakeUp("reminders") const processed = await this.runOnce() + if (signal.aborted || this.stopping) break if (processed > 0) { this.pollingBackoff.reset("work") - if (signal.aborted || this.stopping) break await yieldToEventLoop() continue } diff --git a/src/runtime.ts b/src/runtime.ts index c92b296..30cef90 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -1429,45 +1429,48 @@ export class SolidObjectsRuntime { await renewal actor.discardIntents() if (stateBefore) restoreActorState({ actor, definition, state: stateBefore }) - if (error instanceof LostActivation) { - this.emitInstrumentation("activation.lost", { + try { + if (error instanceof LostActivation) throw error + if (error instanceof Rejected) { + await this.repository.reject(turn, { + code: error.code, + message: error.message, + details: error.details, + }) + this.emitInstrumentation("message.rejected", { + ...messageInstrumentation(turn.message), + code: error.code, + durationMilliseconds: performance.now() - startedAt, + }) + return { actor, retainActivation: activated, activated } + } + const retryable = !(error instanceof NonRetryableError) + const outcome = await this.repository.fail(turn, { + error, + retryable, + }) + this.emitInstrumentation("message.failed", { ...messageInstrumentation(turn.message), + retryable, + errorName: error instanceof Error ? error.name : "Error", durationMilliseconds: performance.now() - startedAt, + outcome, }) - return { actor, retainActivation: false, activated } - } - if (error instanceof Rejected) { - await this.repository.reject(turn, { - code: error.code, - message: error.message, - details: error.details, - }) - this.emitInstrumentation("message.rejected", { + if (outcome === "retrying") + this.emitInstrumentation("message.retry", messageInstrumentation(turn.message)) + if (outcome === "dead" && recoveryMessage(turn.message)) + this.emitInstrumentation("recovery.failed", messageInstrumentation(turn.message)) + if (outcome === "dead") + this.emitInstrumentation("dead_letter.created", messageInstrumentation(turn.message)) + return { actor, retainActivation: activated, activated } + } catch (outcomeError) { + if (!(outcomeError instanceof LostActivation)) throw outcomeError + this.emitInstrumentation("activation.lost", { ...messageInstrumentation(turn.message), - code: error.code, durationMilliseconds: performance.now() - startedAt, }) - return { actor, retainActivation: activated, activated } + return { actor, retainActivation: false, activated } } - const retryable = !(error instanceof NonRetryableError) - const outcome = await this.repository.fail(turn, { - error, - retryable, - }) - this.emitInstrumentation("message.failed", { - ...messageInstrumentation(turn.message), - retryable, - errorName: error instanceof Error ? error.name : "Error", - durationMilliseconds: performance.now() - startedAt, - outcome, - }) - if (outcome === "retrying") - this.emitInstrumentation("message.retry", messageInstrumentation(turn.message)) - if (outcome === "dead" && recoveryMessage(turn.message)) - this.emitInstrumentation("recovery.failed", messageInstrumentation(turn.message)) - if (outcome === "dead") - this.emitInstrumentation("dead_letter.created", messageInstrumentation(turn.message)) - return { actor, retainActivation: activated, activated } } } diff --git a/src/worker.ts b/src/worker.ts index 0f2ac07..ab8f865 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -147,9 +147,9 @@ export class Worker { while (!signal.aborted && !this.stopping) { const wakeUp = await this.runtime.watchWakeUp("actors") const processed = await this.runOnce() + if (signal.aborted || this.stopping) break if (processed > 0) { this.pollingBackoff.reset("work") - if (signal.aborted || this.stopping) break await yieldToEventLoop() continue } diff --git a/test/activation-recovery-backoff.test.ts b/test/activation-recovery-backoff.test.ts index bf94690..c409c05 100644 --- a/test/activation-recovery-backoff.test.ts +++ b/test/activation-recovery-backoff.test.ts @@ -6,6 +6,7 @@ import { mysql } from "../src/database/mysql.js" import { postgresql } from "../src/database/postgresql.js" import { sqlite } from "../src/database/sqlite.js" import type { Database } from "../src/database/types.js" +import { Rejected } from "../src/errors.js" import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" import { controlledDatabaseClock } from "./support/fenced-commit.js" @@ -166,6 +167,48 @@ it.each(["stop", "stale cleanup", "replacement worker"])( }, ) +it.each(["failure", "rejection"])( + "reports lease loss while saving a handler %s and recovers", + async (outcome) => { + await setup() + const worker = runtime.worker() + const reference = runtime.ref(RecoveryCounter, "one") + const message = await reference.send.increment({ label: "slow" }) + RecoveryCounter.beforeCommit = () => { + clock.advanceMilliseconds(201) + if (outcome === "rejection") throw new Rejected({ code: "declined", message: "declined" }) + throw new Error("handler failed") + } + try { + await worker.runOnce() + const losses = events.filter((event) => event.name === "solid_objects.activation.lost") + expect(losses.map((event) => event.attempt)).toEqual([1]) + expect(losses.map((event) => event.messageId)).toEqual([message.id]) + losses.forEach(expectPortableAttributes) + expect((await reference.snapshot()).count).toBe(0) + expect((await runtime.repository.findMessage(message.id))?.error).toBeNull() + expect((await runtime.repository.findMessage(message.id))?.rejection).toBeNull() + expect(await message.status()).toBe("claimed") + expect( + events.some((event) => + ["solid_objects.message.failed", "solid_objects.message.rejected"].includes(event.name), + ), + ).toBe(false) + + RecoveryCounter.beforeCommit = () => {} + expect(await worker.runOnce()).toBe(1) + expect(await message.result()).toBe(1) + expect(Number((await runtime.repository.findMessage(message.id))?.attempt_count)).toBe(2) + const effects = await runtime.settings.database.connection((connection) => + connection.all(`SELECT id FROM ${runtime.repository.table("effects")}`), + ) + expect(effects).toHaveLength(1) + } finally { + await worker.stop() + } + }, +) + it("keeps handler failures terminal with one allowed attempt", async () => { await setup() RecoveryCounter.beforeCommit = () => { diff --git a/test/fixtures/busy-worker.ts b/test/fixtures/busy-worker.ts index 9ddfb3f..06656b6 100644 --- a/test/fixtures/busy-worker.ts +++ b/test/fixtures/busy-worker.ts @@ -1,5 +1,5 @@ import { writeSync } from "node:fs" -import { Actor, createRuntime } from "../../src/index.js" +import { Actor, Rejected, createRuntime } from "../../src/index.js" import { sqlite } from "../../src/database/sqlite.js" const scenario = process.argv[2] ?? "lease-timer" @@ -34,6 +34,9 @@ class BusyActor extends Actor { increment(): void { if (leaseScenario) block(400) + if (scenario === "lease-callback-failure") throw new Error("handler failed") + if (scenario === "lease-callback-rejection") + throw new Rejected({ code: "declined", message: "declined" }) this.count += 1 } @@ -70,7 +73,7 @@ const runtime = createRuntime({ attempts.push(event.attempt) ticksAtLoss.push(ticks) writeSync(1, `${JSON.stringify({ event: "lost", attempts, ticks })}\n`) - if (scenario === "lease-callback" && attempts.length === 3) controller.abort() + if (scenario.startsWith("lease-callback") && attempts.length === 3) controller.abort() } if (!lastTick || leaseScenario) return if (role === "actors" && event.name === "solid_objects.message.completed") finishWork() diff --git a/test/worker-fairness.test.ts b/test/worker-fairness.test.ts index 64f7414..ba2dd17 100644 --- a/test/worker-fairness.test.ts +++ b/test/worker-fairness.test.ts @@ -78,12 +78,16 @@ it("lets timers and input run between lost attempts and honors a timer abort", a expect(result.count).toBe(0) }, 6_000) -it("stops at the third lost attempt when instrumentation aborts the run", async () => { - const result = await runWorker({ scenario: "lease-callback" }) - expect(result.attempts).toEqual([1, 2, 3]) - expect(result.elapsedMilliseconds).toBeLessThan(3_000) - expect(result.count).toBe(0) -}, 6_000) +it.each(["lease-callback", "lease-callback-failure", "lease-callback-rejection"])( + "stops at the third lost attempt when instrumentation aborts %s", + async (scenario) => { + const result = await runWorker({ scenario }) + expect(result.attempts).toEqual([1, 2, 3]) + expect(result.elapsedMilliseconds).toBeLessThan(3_000) + expect(result.count).toBe(0) + }, + 6_000, +) for (const role of ["actors", "effects", "reminders", "broadcasts"]) { for (const execution of ["run", "until-idle", "drain"]) {