From ca9338436d51c901bf1d608d76cf891b6bd6c919 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:24:08 -0800 Subject: [PATCH 01/15] feat: add the singleOwner configuration option A process that is the only writer of its SQLite file does not need cross-process signals or polling to stay correct. This option lets it say so. It refuses a database other than SQLite and a wake-up that is not in-process, because only a reliable in-process signal can replace polling. With the option, the runtime selects in-process signalling and the doctor accepts it. The default stays false. --- docs/configuration.md | 27 +++++------ src/configuration.ts | 23 ++++++++++ src/doctor.ts | 2 +- src/runtime.ts | 6 ++- src/wake-up-selection.ts | 12 +++++ test/single-owner-configuration.test.ts | 59 +++++++++++++++++++++++++ test/single-owner-wake-up.test.ts | 52 ++++++++++++++++++++++ 7 files changed, 166 insertions(+), 15 deletions(-) create mode 100644 test/single-owner-configuration.test.ts create mode 100644 test/single-owner-wake-up.test.ts diff --git a/docs/configuration.md b/docs/configuration.md index f2d3fbd..ae6bea3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -65,19 +65,20 @@ as an instruction to divide the actor. See ## Runtime roles and supervision -| Option | Default | Contract | -| ------------------------------------------- | --------------: | ----------------------------------------------------------------- | -| `workerCount` | `1` | Non-negative actor workers. | -| `effectWorkerCount` | `1` | Non-negative effect workers. | -| `broadcastWorkerCount` | `1` | Non-negative broadcast workers when realtime is enabled. | -| `reminderSchedulerCount` | `1` | Non-negative reminder schedulers. | -| `processHeartbeatIntervalMilliseconds` | `15_000` | Positive persisted heartbeat cadence. | -| `processAliveThresholdMilliseconds` | `60_000` | Positive age after which an owner is stale. | -| `shutdownTimeoutMilliseconds` | `15_000` | Positive shared graceful-shutdown budget. | -| `supervisorRestartDelayMilliseconds` | `100` | Positive initial failed-role replacement delay. | -| `supervisorMaximumRestartDelayMilliseconds` | `10_000` | Positive cap no smaller than the initial delay. | -| `wakeUp` | `"automatic"` | A name or an adapter implementing `watch`, `notify`, and `close`. | -| `logger` | console methods | Structured `debug`, `info`, `warn`, and `error` sink. | +| Option | Default | Contract | +| ------------------------------------------- | --------------: | ------------------------------------------------------------------- | +| `workerCount` | `1` | Non-negative actor workers. | +| `effectWorkerCount` | `1` | Non-negative effect workers. | +| `broadcastWorkerCount` | `1` | Non-negative broadcast workers when realtime is enabled. | +| `reminderSchedulerCount` | `1` | Non-negative reminder schedulers. | +| `processHeartbeatIntervalMilliseconds` | `15_000` | Positive persisted heartbeat cadence. | +| `processAliveThresholdMilliseconds` | `60_000` | Positive age after which an owner is stale. | +| `shutdownTimeoutMilliseconds` | `15_000` | Positive shared graceful-shutdown budget. | +| `supervisorRestartDelayMilliseconds` | `100` | Positive initial failed-role replacement delay. | +| `supervisorMaximumRestartDelayMilliseconds` | `10_000` | Positive cap no smaller than the initial delay. | +| `wakeUp` | `"automatic"` | A name or an adapter implementing `watch`, `notify`, and `close`. | +| `singleOwner` | `false` | `true` when this process is the only writer of its SQLite database. | +| `logger` | console methods | Structured `debug`, `info`, `warn`, and `error` sink. | Counts may be zero, but the complete configuration must leave at least one runtime role enabled. Broadcast workers are started only when `broadcast` or diff --git a/src/configuration.ts b/src/configuration.ts index a369d2e..43da659 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -101,6 +101,7 @@ export interface SolidObjectsConfiguration { instrumentation?: (event: InstrumentationEvent) => void broadcast?: (event: BroadcastEvent) => Promise wakeUp?: WakeUpSetting + singleOwner?: boolean redriveBatchSize?: number redriveBatchPauseMilliseconds?: number administrationIdentity?: ( @@ -192,6 +193,7 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime : String(authorizationContext)), logger: configuration.logger ?? consoleLogger, wakeUp: configuration.wakeUp ?? "automatic", + singleOwner: configuration.singleOwner ?? false, authorizeMessage: configuration.authorizeMessage ?? (() => false), authorizeQuery: configuration.authorizeQuery ?? (() => false), authorizeDestroy: configuration.authorizeDestroy ?? (() => false), @@ -244,8 +246,29 @@ function validateWakeUpAdapter(adapter: WakeUpAdapter): void { } } +function validateSingleOwner(settings: RuntimeSettings): void { + if (typeof settings.singleOwner !== "boolean") { + throw new TypeError("singleOwner must be a boolean") + } + if (!settings.singleOwner) return + if (settings.database.family !== "sqlite") { + throw new TypeError( + `singleOwner needs a SQLite database, and this database is ${settings.database.family}`, + ) + } + if (typeof settings.wakeUp !== "string") { + throw new TypeError("singleOwner needs in-process wake-ups, and wakeUp is an adapter") + } + if (settings.wakeUp === "automatic" || settings.wakeUp === "in_process") return + + throw new TypeError( + `singleOwner needs in-process wake-ups, and wakeUp is ${JSON.stringify(settings.wakeUp)}`, + ) +} + function validateSettings(settings: RuntimeSettings): void { validateWakeUp(settings.wakeUp) + validateSingleOwner(settings) if (!/^[a-z][a-z0-9_]*$/.test(settings.tableNamePrefix)) { throw new TypeError("tableNamePrefix must contain lowercase letters, digits, and underscores") } diff --git a/src/doctor.ts b/src/doctor.ts index e1159c3..fc5868a 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -167,7 +167,7 @@ export class Doctor { const summary = `${capability.adapter}: ${capability.reason}${ floor === undefined ? "" : `, floor ${floor} ms` }` - if (capability.crossesProcesses) { + if (capability.crossesProcesses || this.runtime.settings.singleOwner) { return check({ name: "wakeUp", status: "pass", message: summary }) } return check({ diff --git a/src/runtime.ts b/src/runtime.ts index 7882369..e0e3cbb 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -132,7 +132,7 @@ import { waitFor, Worker } from "./worker.js" import { EffectWorker } from "./effect-worker.js" import type { WakeUpAdapter, WakeUpCapability, WakeUpRole, WakeUpWatch } from "./wake-up.js" import { notifyWakeUp } from "./wake-up-notification.js" -import { selectWakeUp, type SelectedWakeUp } from "./wake-up-selection.js" +import { selectWakeUp, singleOwnerWakeUp, type SelectedWakeUp } from "./wake-up-selection.js" import { withDatabaseDeadline } from "./database/deadline.js" import type { DatabaseConnection } from "./database/types.js" import { evaluateActorTurn, readActorObservables } from "./turn.js" @@ -2310,6 +2310,10 @@ export class SolidObjectsRuntime { } private resolveWakeUp(): Promise { + if (this.settings.singleOwner) { + this.wakeUpSelection ??= Promise.resolve(singleOwnerWakeUp()) + return this.wakeUpSelection + } this.wakeUpSelection ??= selectWakeUp({ setting: this.settings.wakeUp, database: this.settings.database, diff --git a/src/wake-up-selection.ts b/src/wake-up-selection.ts index 404d2fc..7c04e90 100644 --- a/src/wake-up-selection.ts +++ b/src/wake-up-selection.ts @@ -42,6 +42,18 @@ export async function selectWakeUp(options: WakeUpSelectionOptions): Promise) { + return buildSettings({ database: sqlite({ path: ":memory:" }), ...configuration }) +} + +function databaseOfFamily(family: "postgresql" | "mysql"): Database { + return { + family, + schemaIdentity: "test", + connection: () => Promise.reject(new Error("unused")), + transaction: () => Promise.reject(new Error("unused")), + close: () => Promise.resolve(), + } +} + +describe("singleOwner configuration", () => { + it("defaults to false", () => { + expect(settingsFor({}).singleOwner).toBe(false) + }) + + it("accepts true with SQLite and in-process wake-ups", () => { + expect(settingsFor({ singleOwner: true }).singleOwner).toBe(true) + expect(settingsFor({ singleOwner: true, wakeUp: "in_process" }).singleOwner).toBe(true) + }) + + it("refuses a value that is not a boolean", () => { + expect(() => settingsFor({ singleOwner: "yes" as unknown as boolean })).toThrow( + "singleOwner must be a boolean", + ) + }) + + it.each(["postgresql", "mysql"] as const)("refuses a %s database", (family) => { + expect(() => buildSettings({ database: databaseOfFamily(family), singleOwner: true })).toThrow( + `singleOwner needs a SQLite database, and this database is ${family}`, + ) + }) + + it.each(["postgresql", "redis"] as const)("refuses the %s wake-up", (wakeUp) => { + expect(() => settingsFor({ singleOwner: true, wakeUp })).toThrow( + `singleOwner needs in-process wake-ups, and wakeUp is ${JSON.stringify(wakeUp)}`, + ) + }) + + it("refuses a wake-up adapter", () => { + expect(() => settingsFor({ singleOwner: true, wakeUp: new InProcessWakeUpAdapter() })).toThrow( + "singleOwner needs in-process wake-ups, and wakeUp is an adapter", + ) + }) + + it("keeps the other wake-up settings valid without singleOwner", () => { + expect(settingsFor({ wakeUp: "redis" }).wakeUp).toBe("redis") + expect(() => buildSettings({ database: databaseOfFamily("postgresql") })).not.toThrow() + }) +}) diff --git a/test/single-owner-wake-up.test.ts b/test/single-owner-wake-up.test.ts new file mode 100644 index 0000000..3141d17 --- /dev/null +++ b/test/single-owner-wake-up.test.ts @@ -0,0 +1,52 @@ +import { afterEach, describe, expect, it, vi } from "vitest" +import { sqlite } from "../src/database/sqlite.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { InProcessWakeUpAdapter } from "../src/wake-up.js" + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + vi.unstubAllEnvs() + await runtime?.close() + runtime = undefined +}) + +function singleOwnerRuntime(wakeUp: "automatic" | "in_process"): SolidObjectsRuntime { + return createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: true, + wakeUp, + authorizeAdministration: () => true, + }) +} + +describe("singleOwner wake-ups", () => { + it.each(["automatic", "in_process"] as const)( + "selects in-process signalling for %s even when a Redis URL is set", + async (wakeUp) => { + vi.stubEnv("SOLID_OBJECTS_REDIS_URL", "redis://127.0.0.1:1") + runtime = singleOwnerRuntime(wakeUp) + + const capability = await runtime.wakeUpCapability() + + expect(capability).toEqual({ + adapter: "in_process", + crossesProcesses: false, + reason: + "singleOwner is set, so one process owns the database and in-process signalling reaches every role", + }) + expect(await runtime.wakeUpAdapter()).toBeInstanceOf(InProcessWakeUpAdapter) + }, + ) + + it("passes the doctor wake-up check, because no other process can commit", async () => { + runtime = singleOwnerRuntime("automatic") + await runtime.install() + + const report = await runtime.doctor.run({ roundTrip: "skip" }) + + expect(report.checks.find((check) => check.name === "wakeUp")).toMatchObject({ + status: "pass", + }) + }) +}) From 887fa46e3f7bddf96ddfb15e1ccb0dc7d144fab9 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:29:37 -0800 Subject: [PATCH 02/15] feat: read when each role has work next A role that does not poll must know when to look again. These queries read the earliest time at which each role can claim something: a ready message at the head of its mailbox, an expiring lease, a pending effect, broadcast or reminder, a claim whose owner becomes stale, a longer recovery timeout, or a running redrive. They mirror the claim conditions, so a due time never points at work that a claim skips. --- src/due-work.ts | 133 +++++++++++++++ test/due-work.test.ts | 367 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 500 insertions(+) create mode 100644 src/due-work.ts create mode 100644 test/due-work.test.ts diff --git a/src/due-work.ts b/src/due-work.ts new file mode 100644 index 0000000..0179a1c --- /dev/null +++ b/src/due-work.ts @@ -0,0 +1,133 @@ +import type { RuntimeSettings } from "./configuration.js" + +interface DueRow { + due_at_ms: number | bigint | null +} + +export class DueWork { + constructor(private readonly settings: RuntimeSettings) {} + + untilActorWork(): Promise { + return this.untilDue(() => ({ + sql: `SELECT MIN(due_at_ms) AS due_at_ms FROM ( + SELECT CASE + WHEN instances.activation_owner_id IS NOT NULL + AND instances.activation_expires_at_ms > ready.available_at_ms + THEN instances.activation_expires_at_ms + ELSE ready.available_at_ms + END AS due_at_ms + FROM ${this.table("ready_messages")} ready + JOIN ${this.table("instances")} instances ON instances.id = ready.instance_id + WHERE instances.paused = 0 + AND NOT EXISTS ( + SELECT 1 FROM ${this.table("ready_messages")} earlier + WHERE earlier.instance_id = ready.instance_id AND earlier.sequence < ready.sequence + ) + AND NOT EXISTS ( + SELECT 1 FROM ${this.table("claimed_messages")} claimed + WHERE claimed.instance_id = ready.instance_id AND claimed.sequence < ready.sequence + ) + UNION ALL + SELECT instances.activation_expires_at_ms AS due_at_ms + FROM ${this.table("claimed_messages")} claimed + JOIN ${this.table("instances")} instances ON instances.id = claimed.instance_id + WHERE instances.activation_expires_at_ms IS NOT NULL + ) due`, + parameters: [], + })) + } + + untilEffectWork(): Promise { + const threshold = this.settings.processAliveThresholdMilliseconds + return this.untilDue((now) => ({ + sql: `SELECT MIN(due_at_ms) AS due_at_ms FROM ( + SELECT effects.available_at_ms AS due_at_ms + FROM ${this.table("effects")} effects + WHERE effects.status = 'pending' + UNION ALL + SELECT CASE + WHEN owners.id IS NULL THEN ? + WHEN recoveries.effect_id IS NULL THEN owners.heartbeat_at_ms + ? + WHEN recoveries.recovery_timeout_ms > ? + THEN owners.heartbeat_at_ms + recoveries.recovery_timeout_ms + ELSE owners.heartbeat_at_ms + ? + END AS due_at_ms + FROM ${this.table("effects")} effects + LEFT JOIN ${this.table("effect_recoveries")} recoveries + ON recoveries.effect_id = effects.id AND recoveries.recovery_operation IS NOT NULL + LEFT JOIN ${this.table("processes")} owners + ON owners.id = effects.claimed_by + AND (recoveries.effect_id IS NOT NULL OR owners.shutdown_state = 'running') + WHERE effects.status = 'processing' + ) due`, + parameters: [now, threshold, threshold, threshold], + })) + } + + untilReminderWork(): Promise { + const threshold = this.settings.processAliveThresholdMilliseconds + return this.untilDue(() => ({ + sql: `SELECT MIN(CASE WHEN stale_at_ms > run_at_ms THEN stale_at_ms ELSE run_at_ms END) + AS due_at_ms FROM ( + SELECT reminders.run_at_ms AS run_at_ms, + CASE + WHEN reminders.claimed_by IS NULL OR owners.id IS NULL THEN reminders.run_at_ms + WHEN reminders.claimed_at_ms < owners.heartbeat_at_ms + THEN reminders.claimed_at_ms + ? + ELSE owners.heartbeat_at_ms + ? + END AS stale_at_ms + FROM ${this.table("reminders")} reminders + LEFT JOIN ${this.table("processes")} owners + ON owners.id = reminders.claimed_by AND owners.shutdown_state = 'running' + WHERE reminders.status = 'scheduled' + ) due`, + parameters: [threshold, threshold], + })) + } + + untilBroadcastWork(): Promise { + const threshold = this.settings.processAliveThresholdMilliseconds + return this.untilDue((now) => ({ + sql: `SELECT MIN(due_at_ms) AS due_at_ms FROM ( + SELECT broadcasts.available_at_ms AS due_at_ms + FROM ${this.table("broadcasts")} broadcasts + WHERE broadcasts.status = 'pending' AND NOT EXISTS ( + SELECT 1 FROM ${this.table("broadcasts")} earlier + WHERE earlier.instance_id = broadcasts.instance_id + AND earlier.state_revision < broadcasts.state_revision + AND earlier.status IN ('pending', 'processing') + ) + UNION ALL + SELECT CASE WHEN owners.id IS NULL THEN ? ELSE owners.heartbeat_at_ms + ? END AS due_at_ms + FROM ${this.table("broadcasts")} broadcasts + LEFT JOIN ${this.table("processes")} owners + ON owners.id = broadcasts.claimed_by AND owners.shutdown_state = 'running' + WHERE broadcasts.status = 'processing' + ) due`, + parameters: [now, threshold], + })) + } + + untilRedriveWork(): Promise { + return this.untilDue((now) => ({ + sql: `SELECT MIN(?) AS due_at_ms FROM ${this.table("redrives")} WHERE status = 'running'`, + parameters: [now], + })) + } + + private async untilDue( + query: (now: number) => { sql: string; parameters: readonly unknown[] }, + ): Promise { + return this.settings.database.connection(async (connection) => { + const now = await connection.nowMilliseconds() + const { sql, parameters } = query(now) + const row = await connection.get(sql, parameters) + if (row?.due_at_ms === null || row?.due_at_ms === undefined) return undefined + return Number(row.due_at_ms) - now + }) + } + + private table(name: string): string { + return `${this.settings.tableNamePrefix}${name}` + } +} diff --git a/test/due-work.test.ts b/test/due-work.test.ts new file mode 100644 index 0000000..4b3489f --- /dev/null +++ b/test/due-work.test.ts @@ -0,0 +1,367 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest" +import { sqlite } from "../src/database/sqlite.js" +import { DueWork } from "../src/due-work.js" +import type { MessageRow } from "../src/records.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" + +const aliveThreshold = 60_000 +const leaseDuration = 30_000 +const tolerance = 1_000 + +let runtime: SolidObjectsRuntime +let dueWork: DueWork + +beforeEach(async () => { + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: true, + processAliveThresholdMilliseconds: aliveThreshold, + leaseDurationMilliseconds: leaseDuration, + }) + await runtime.install() + dueWork = new DueWork(runtime.settings) +}) + +afterEach(async () => { + await runtime.close() +}) + +function databaseNow(): Promise { + return runtime.settings.database.connection((connection) => connection.nowMilliseconds()) +} + +async function write(sql: string, parameters: readonly unknown[] = []): Promise { + await runtime.settings.database.transaction((connection) => connection.run(sql, parameters)) +} + +function enqueue(options: { actorId: string; availableAtMilliseconds?: number }) { + return runtime.repository.enqueue({ + actorType: "Probe", + actorId: options.actorId, + operation: "work", + deliveryMode: "async", + arguments: {}, + ...(options.availableAtMilliseconds === undefined + ? {} + : { availableAtMilliseconds: options.availableAtMilliseconds }), + }) +} + +async function registerOwner(options: { + id: string + heartbeatAtMilliseconds: number + shutdownState?: "running" | "draining" | "stopped" +}): Promise { + await write( + `INSERT INTO solid_objects_processes + (id, kind, hostname, host_process_id, metadata, started_at_ms, heartbeat_at_ms, shutdown_state) + VALUES (?, 'worker', 'other-host', 1, '{}', ?, ?, ?)`, + [ + options.id, + options.heartbeatAtMilliseconds, + options.heartbeatAtMilliseconds, + options.shutdownState ?? "running", + ], + ) +} + +async function insertEffect(options: { + message: MessageRow + id: string + status: "pending" | "processing" + availableAtMilliseconds: number + claimedBy?: string +}): Promise { + await write( + `INSERT INTO solid_objects_effects + (id, message_id, instance_id, name, arguments, status, max_attempts, available_at_ms, claimed_by) + VALUES (?, ?, ?, 'work', '{}', ?, 3, ?, ?)`, + [ + options.id, + options.message.id, + options.message.instance_id, + options.status, + options.availableAtMilliseconds, + options.claimedBy ?? null, + ], + ) +} + +async function insertReminder(options: { + message: MessageRow + id: string + runAtMilliseconds: number + status?: "scheduled" | "paused" + claimedBy?: string + claimedAtMilliseconds?: number +}): Promise { + await write( + `INSERT INTO solid_objects_reminders + (id, instance_id, operation, run_at_ms, arguments, missed_policy, status, claimed_by, claimed_at_ms) + VALUES (?, ?, ?, ?, '{}', 'latest', ?, ?, ?)`, + [ + options.id, + options.message.instance_id, + options.id, + options.runAtMilliseconds, + options.status ?? "scheduled", + options.claimedBy ?? null, + options.claimedAtMilliseconds ?? null, + ], + ) +} + +async function insertBroadcast(options: { + message: MessageRow + id: string + revision: number + status: "pending" | "processing" + availableAtMilliseconds: number + claimedBy?: string +}): Promise { + await write( + `INSERT INTO solid_objects_broadcasts + (id, message_id, instance_id, actor_type, actor_id, state_revision, observables, status, + available_at_ms, claimed_by) + VALUES (?, ?, ?, 'Probe', ?, ?, '{}', ?, ?, ?)`, + [ + options.id, + options.message.id, + options.message.instance_id, + options.message.actor_id, + options.revision, + options.status, + options.availableAtMilliseconds, + options.claimedBy ?? null, + ], + ) +} + +function expectAbout(actual: number | undefined, expected: number): void { + expect(actual).toBeDefined() + expect(actual!).toBeGreaterThan(expected - tolerance) + expect(actual!).toBeLessThanOrEqual(expected) +} + +describe("DueWork", () => { + it("reports no due time when no work exists", async () => { + expect(await dueWork.untilActorWork()).toBeUndefined() + expect(await dueWork.untilEffectWork()).toBeUndefined() + expect(await dueWork.untilReminderWork()).toBeUndefined() + expect(await dueWork.untilBroadcastWork()).toBeUndefined() + expect(await dueWork.untilRedriveWork()).toBeUndefined() + }) + + describe("actor work", () => { + it("reports the earliest ready message", async () => { + const now = await databaseNow() + await enqueue({ actorId: "late", availableAtMilliseconds: now + 9_000 }) + await enqueue({ actorId: "early", availableAtMilliseconds: now + 5_000 }) + + expectAbout(await dueWork.untilActorWork(), 5_000) + }) + + it("waits for the head of a mailbox, not for a later message behind it", async () => { + const now = await databaseNow() + await enqueue({ actorId: "blocked", availableAtMilliseconds: now + 8_000 }) + await enqueue({ actorId: "blocked" }) + + expectAbout(await dueWork.untilActorWork(), 8_000) + }) + + it("skips a paused actor", async () => { + const message = await enqueue({ actorId: "paused" }) + await write("UPDATE solid_objects_instances SET paused = 1 WHERE id = ?", [ + message.instance_id, + ]) + + expect(await dueWork.untilActorWork()).toBeUndefined() + }) + + it("waits for an activation that another owner holds", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "held" }) + await registerOwner({ id: "holder", heartbeatAtMilliseconds: now }) + await write( + `UPDATE solid_objects_instances + SET activation_owner_id = 'holder', activation_token = 'token', activation_expires_at_ms = ? + WHERE id = ?`, + [now + 7_000, message.instance_id], + ) + + expectAbout(await dueWork.untilActorWork(), 7_000) + }) + + it("waits for the lease of a claimed message to expire", async () => { + const now = await databaseNow() + await enqueue({ actorId: "claimed" }) + await registerOwner({ id: "claimer", heartbeatAtMilliseconds: now }) + expect(await runtime.repository.claim("claimer")).toBeDefined() + + expectAbout(await dueWork.untilActorWork(), leaseDuration) + }) + }) + + describe("effect work", () => { + it("reports a pending effect at its available time", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "effects" }) + await insertEffect({ + message, + id: "retry", + status: "pending", + availableAtMilliseconds: now + 4_000, + }) + + expectAbout(await dueWork.untilEffectWork(), 4_000) + }) + + it("reports a processing effect when its owner becomes stale", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "effects" }) + await registerOwner({ id: "owner", heartbeatAtMilliseconds: now }) + await insertEffect({ + message, + id: "running", + status: "processing", + availableAtMilliseconds: now, + claimedBy: "owner", + }) + + expectAbout(await dueWork.untilEffectWork(), aliveThreshold) + }) + + it.each(["missing", "draining"] as const)( + "reports a processing effect at once when its owner is %s", + async (owner) => { + const now = await databaseNow() + const message = await enqueue({ actorId: "effects" }) + if (owner === "draining") { + await registerOwner({ id: owner, heartbeatAtMilliseconds: now, shutdownState: owner }) + } + await insertEffect({ + message, + id: "orphan", + status: "processing", + availableAtMilliseconds: now, + claimedBy: owner, + }) + + expect(await dueWork.untilEffectWork()).toBeLessThanOrEqual(0) + }, + ) + + it("uses the longer recovery timeout of an effect with a recovery operation", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "effects" }) + await registerOwner({ id: "owner", heartbeatAtMilliseconds: now, shutdownState: "stopped" }) + await insertEffect({ + message, + id: "recoverable", + status: "processing", + availableAtMilliseconds: now, + claimedBy: "owner", + }) + await write( + `INSERT INTO solid_objects_effect_recoveries + (effect_id, instance_id, recovery_operation, status_operation, recovery_timeout_ms) + VALUES ('recoverable', ?, 'recover', 'inspect', ?)`, + [message.instance_id, 120_000], + ) + + expectAbout(await dueWork.untilEffectWork(), 120_000) + }) + }) + + describe("reminder work", () => { + it("reports a scheduled reminder at its run time and skips a paused one", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "reminders" }) + await insertReminder({ message, id: "paused", runAtMilliseconds: now, status: "paused" }) + await insertReminder({ message, id: "later", runAtMilliseconds: now + 3_000 }) + + expectAbout(await dueWork.untilReminderWork(), 3_000) + }) + + it("reports a claimed reminder when its claim becomes stale", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "reminders" }) + await registerOwner({ id: "scheduler", heartbeatAtMilliseconds: now + 10_000 }) + await insertReminder({ + message, + id: "claimed", + runAtMilliseconds: now - 1_000, + claimedBy: "scheduler", + claimedAtMilliseconds: now, + }) + + expectAbout(await dueWork.untilReminderWork(), aliveThreshold) + }) + + it("reports a reminder at its run time when its claim owner is gone", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "reminders" }) + await insertReminder({ + message, + id: "orphan", + runAtMilliseconds: now + 2_000, + claimedBy: "missing", + claimedAtMilliseconds: now, + }) + + expectAbout(await dueWork.untilReminderWork(), 2_000) + }) + }) + + describe("broadcast work", () => { + it("waits for the earliest revision of an actor, not for a later one behind it", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "broadcasts" }) + await insertBroadcast({ + message, + id: "first", + revision: 1, + status: "pending", + availableAtMilliseconds: now + 6_000, + }) + await insertBroadcast({ + message, + id: "second", + revision: 2, + status: "pending", + availableAtMilliseconds: now, + }) + + expectAbout(await dueWork.untilBroadcastWork(), 6_000) + }) + + it("reports a processing broadcast when its owner becomes stale", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "broadcasts" }) + await registerOwner({ id: "owner", heartbeatAtMilliseconds: now }) + await insertBroadcast({ + message, + id: "running", + revision: 1, + status: "processing", + availableAtMilliseconds: now, + claimedBy: "owner", + }) + + expectAbout(await dueWork.untilBroadcastWork(), aliveThreshold) + }) + }) + + describe("redrive work", () => { + it("reports a running redrive as due now", async () => { + const now = await databaseNow() + await write( + `INSERT INTO solid_objects_redrives (id, kind, filters, status, started_at_ms) + VALUES ('redrive', 'effect', '{}', 'running', ?)`, + [now], + ) + + expect(await dueWork.untilRedriveWork()).toBe(0) + }) + }) +}) From 3326761ebfd53456513b5d88f0c439df19768421 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:31:47 -0800 Subject: [PATCH 03/15] feat: wait for due work instead of polling An idle runtime with singleOwner sent thousands of statements a minute, because each role waited a fixed interval and rewrote its process row on every pass. With the option, each role now waits for an in-process wake-up or for the next due time that the database reports, bounded at 24 hours. The actor worker also wakes to renew or release a cached activation, and stops when it holds none. Each role writes its process row once at start. The periodic dead-process cleanup does not run. A redrive start wakes the redrive scheduler through a local signal. When a due time has already passed and a claim still finds nothing, the role falls back to its polling interval, so stuck work cannot spin. --- src/broadcast-worker.ts | 8 ++- src/configuration.ts | 7 +- src/effect-worker.ts | 8 ++- src/local-signal.ts | 44 ++++++++++++ src/redrive.ts | 37 +++++++++- src/reminder-scheduler.ts | 8 ++- src/runtime.ts | 10 ++- src/single-owner.ts | 24 +++++++ src/worker.ts | 24 ++++++- test/single-owner-idle.test.ts | 115 ++++++++++++++++++++++++++++++ test/support/counting-database.ts | 75 +++++++++++++++++++ 11 files changed, 352 insertions(+), 8 deletions(-) create mode 100644 src/local-signal.ts create mode 100644 src/single-owner.ts create mode 100644 test/single-owner-idle.test.ts create mode 100644 test/support/counting-database.ts diff --git a/src/broadcast-worker.ts b/src/broadcast-worker.ts index 14b999a..4b926ff 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 { idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class BroadcastWorker { @@ -64,7 +65,11 @@ export class BroadcastWorker { continue } const notified = await wakeUp.wait({ - timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + timeoutMilliseconds: await idleWaitMilliseconds({ + settings: this.runtime.settings, + pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + untilDue: () => this.runtime.dueWork.untilBroadcastWork(), + }), signal, }) if (notified === false) this.pollingBackoff.recordIdle() @@ -91,6 +96,7 @@ export class BroadcastWorker { } private async ensureRegistered(): Promise { + if (this.registered && this.runtime.settings.singleOwner) return await this.runtime.repository.registerProcess(this.processId, "broadcast_worker") this.registered = true } diff --git a/src/configuration.ts b/src/configuration.ts index 43da659..7077dfe 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -373,10 +373,15 @@ function validateSettings(settings: RuntimeSettings): void { settings.reminderSchedulerCount + (broadcastsEnabled(settings) ? settings.broadcastWorkerCount : 0) + (settings.retentionIntervalMilliseconds > 0 ? 1 : 0) + - (settings.deadProcessCleanupIntervalMilliseconds > 0 ? 1 : 0) + (deadProcessCleanupScheduled(settings) ? 1 : 0) if (roleCount === 0) throw new TypeError("at least one runtime role must be configured") } +export function deadProcessCleanupScheduled(settings: RuntimeSettings): boolean { + if (settings.singleOwner) return false + return settings.deadProcessCleanupIntervalMilliseconds > 0 +} + export function broadcastsEnabled(settings: RuntimeSettings): boolean { return ( settings.broadcast !== undefined || diff --git a/src/effect-worker.ts b/src/effect-worker.ts index 6523b47..4d994f9 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 { idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class EffectWorker { @@ -64,7 +65,11 @@ export class EffectWorker { continue } const notified = await wakeUp.wait({ - timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + timeoutMilliseconds: await idleWaitMilliseconds({ + settings: this.runtime.settings, + pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + untilDue: () => this.runtime.dueWork.untilEffectWork(), + }), signal, }) if (notified === false) this.pollingBackoff.recordIdle() @@ -91,6 +96,7 @@ export class EffectWorker { } private async ensureRegistered(): Promise { + if (this.registered && this.runtime.settings.singleOwner) return await this.runtime.repository.registerProcess(this.processId, "effect_worker") this.registered = true } diff --git a/src/local-signal.ts b/src/local-signal.ts new file mode 100644 index 0000000..0df49b4 --- /dev/null +++ b/src/local-signal.ts @@ -0,0 +1,44 @@ +export interface LocalSignalWaitOptions { + timeoutMilliseconds: number + signal?: AbortSignal +} + +export interface LocalSignalWatch { + wait(options: LocalSignalWaitOptions): Promise +} + +export class LocalSignal { + private generation = 0 + private readonly wakers = new Set<() => void>() + + watch(): LocalSignalWatch { + const generation = this.generation + return { wait: (options) => this.wait({ generation, ...options }) } + } + + notify(): void { + this.generation += 1 + const wakers = [...this.wakers] + this.wakers.clear() + for (const wake of wakers) wake() + } + + private wait(options: LocalSignalWaitOptions & { generation: number }): Promise { + if (options.signal?.aborted) return Promise.resolve(false) + if (this.generation !== options.generation) return Promise.resolve(true) + + return new Promise((resolve) => { + const finish = (notified: boolean) => { + clearTimeout(timeout) + this.wakers.delete(wake) + options.signal?.removeEventListener("abort", abort) + resolve(notified) + } + const wake = () => finish(true) + const abort = () => finish(false) + const timeout = setTimeout(() => finish(false), options.timeoutMilliseconds) + this.wakers.add(wake) + options.signal?.addEventListener("abort", abort, { once: true }) + }) + } +} diff --git a/src/redrive.ts b/src/redrive.ts index 8ed1f80..292f672 100644 --- a/src/redrive.ts +++ b/src/redrive.ts @@ -4,6 +4,8 @@ import { randomUUID, sha256Hex } from "./platform/uuid.js" import type { SolidObjectsRuntime } from "./runtime.js" import type { AdministrationOptions } from "./types.js" import { waitFor } from "./worker.js" +import type { LocalSignalWatch } from "./local-signal.js" +import { MAXIMUM_WAIT_MILLISECONDS } from "./single-owner.js" export type RedriveStatus = "running" | "completed" | "cancelled" @@ -183,6 +185,7 @@ export class RedriveManager { if (running) return running throw new RedriveNotStarted(`could not start a ${input.kind} redrive`) } + this.runtime.redriveSignal.notify() return await this.require(id) } @@ -282,8 +285,9 @@ export class RedriveScheduler { async run(signal: AbortSignal): Promise { while (!signal.aborted && !this.stopping) { + const watch = this.runtime.redriveSignal.watch() const moved = await this.advance() - await waitFor(this.pauseMilliseconds(moved), signal) + await this.pause({ moved, watch, signal }) } } @@ -310,6 +314,37 @@ export class RedriveScheduler { } } + private async pause(options: { + moved: boolean + watch: LocalSignalWatch + signal: AbortSignal + }): Promise { + if (options.moved || !this.runtime.settings.singleOwner) { + await waitFor(this.pauseMilliseconds(options.moved), options.signal) + return + } + await options.watch.wait({ + timeoutMilliseconds: await this.idleWaitMilliseconds(), + signal: options.signal, + }) + } + + private async idleWaitMilliseconds(): Promise { + if ((await this.untilRedriveWork()) === undefined) return MAXIMUM_WAIT_MILLISECONDS + return this.runtime.settings.idlePollingIntervalMilliseconds + } + + private async untilRedriveWork(): Promise { + try { + return await this.runtime.dueWork.untilRedriveWork() + } catch (error) { + this.runtime.emitInstrumentation("supervisor.redrive_failed", { + errorName: error instanceof Error ? error.name : "Error", + }) + return 0 + } + } + private pauseMilliseconds(moved: boolean): number { if (moved) return this.runtime.settings.redriveBatchPauseMilliseconds return this.runtime.settings.idlePollingIntervalMilliseconds diff --git a/src/reminder-scheduler.ts b/src/reminder-scheduler.ts index 6cf429e..cae3ea8 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 { idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class ReminderScheduler { @@ -85,7 +86,11 @@ export class ReminderScheduler { continue } const notified = await wakeUp.wait({ - timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + timeoutMilliseconds: await idleWaitMilliseconds({ + settings: this.runtime.settings, + pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + untilDue: () => this.runtime.dueWork.untilReminderWork(), + }), signal, }) if (notified === false) this.pollingBackoff.recordIdle() @@ -112,6 +117,7 @@ export class ReminderScheduler { } private async ensureRegistered(): Promise { + if (this.registered && this.runtime.settings.singleOwner) return await this.runtime.repository.registerProcess(this.processId, "reminder_scheduler") this.registered = true } diff --git a/src/runtime.ts b/src/runtime.ts index e0e3cbb..6112405 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -6,6 +6,7 @@ import { BroadcastWorker } from "./broadcast-worker.js" import { buildSettings, broadcastsEnabled, + deadProcessCleanupScheduled, validateComponent, type BroadcastEvent, type ComponentRegistration, @@ -83,6 +84,8 @@ import { import { RealtimeManager } from "./realtime.js" import type { PayloadEnvelope } from "./browser/index.js" import { MaintenanceScheduler } from "./maintenance-scheduler.js" +import { DueWork } from "./due-work.js" +import { LocalSignal } from "./local-signal.js" import type { BroadcastRow, ClaimedTurn, @@ -197,6 +200,8 @@ function administrationEventId(occurredAtMilliseconds: number): string { export class SolidObjectsRuntime { readonly settings readonly repository + readonly dueWork + readonly redriveSignal = new LocalSignal() readonly deadLetters readonly redrives readonly reconciliation @@ -225,6 +230,7 @@ export class SolidObjectsRuntime { this.repository = new Repository(this.settings, { wakeUpAdapter: () => this.wakeUpAdapter(), }) + this.dueWork = new DueWork(this.settings) this.deadLetters = new DeadLetterManager(this) this.redrives = new RedriveManager(this) this.reconciliation = new ReconciliationManager(this) @@ -327,7 +333,7 @@ export class SolidObjectsRuntime { ? this.settings.broadcastWorkerCount : 0, retentionEnabled: this.settings.retentionIntervalMilliseconds > 0, - deadProcessCleanupEnabled: this.settings.deadProcessCleanupIntervalMilliseconds > 0, + deadProcessCleanupEnabled: deadProcessCleanupScheduled(this.settings), }) const controller = new AbortController() let shutdownDeadline: number | undefined @@ -1972,7 +1978,7 @@ export class SolidObjectsRuntime { }), ] : []), - ...(this.settings.deadProcessCleanupIntervalMilliseconds > 0 + ...(deadProcessCleanupScheduled(this.settings) ? [ () => new MaintenanceScheduler({ diff --git a/src/single-owner.ts b/src/single-owner.ts new file mode 100644 index 0000000..87f619e --- /dev/null +++ b/src/single-owner.ts @@ -0,0 +1,24 @@ +import type { RuntimeSettings } from "./configuration.js" + +export const MAXIMUM_WAIT_MILLISECONDS = 86_400_000 + +export function waitUntilDue(options: { + untilDueMilliseconds: number | undefined + pollingIntervalMilliseconds: number +}): number { + if (options.untilDueMilliseconds === undefined) return MAXIMUM_WAIT_MILLISECONDS + if (options.untilDueMilliseconds < 0) return options.pollingIntervalMilliseconds + return Math.min(options.untilDueMilliseconds + 1, MAXIMUM_WAIT_MILLISECONDS) +} + +export async function idleWaitMilliseconds(options: { + settings: RuntimeSettings + pollingIntervalMilliseconds: number + untilDue: () => Promise +}): Promise { + if (!options.settings.singleOwner) return options.pollingIntervalMilliseconds + return waitUntilDue({ + untilDueMilliseconds: await options.untilDue(), + pollingIntervalMilliseconds: options.pollingIntervalMilliseconds, + }) +} diff --git a/src/worker.ts b/src/worker.ts index ab8f865..c13db48 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 { idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" interface CachedActivation { @@ -154,7 +155,7 @@ export class Worker { continue } const notified = await wakeUp.wait({ - timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + timeoutMilliseconds: await this.idleWaitMilliseconds(), signal, }) if (notified === false) this.pollingBackoff.recordIdle() @@ -192,7 +193,28 @@ export class Worker { } } + private async idleWaitMilliseconds(): Promise { + const databaseWait = await idleWaitMilliseconds({ + settings: this.runtime.settings, + pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + untilDue: () => this.runtime.dueWork.untilActorWork(), + }) + if (!this.runtime.settings.singleOwner) return databaseWait + return Math.min(databaseWait, this.untilCachedActivationDue()) + } + + private untilCachedActivationDue(): number { + let earliest = Number.POSITIVE_INFINITY + for (const activation of this.activations.values()) { + const deactivateAt = + activation.lastUsedAt + this.runtime.settings.idleDeactivationTimeoutMilliseconds + earliest = Math.min(earliest, activation.renewAt, deactivateAt) + } + return Math.max(Math.ceil(earliest - performance.now()), 0) + } + private async ensureRegistered(): Promise { + if (this.registered && this.runtime.settings.singleOwner) return await this.runtime.repository.registerProcess(this.processId, "worker") this.registered = true } diff --git a/test/single-owner-idle.test.ts b/test/single-owner-idle.test.ts new file mode 100644 index 0000000..45eb1c9 --- /dev/null +++ b/test/single-owner-idle.test.ts @@ -0,0 +1,115 @@ +import { afterEach, describe, expect, it, vi } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { CountingDatabase } from "./support/counting-database.js" + +class Counter extends Actor { + static override readonly actorType = "SingleOwnerIdleCounter" + + count = 0 + + increment(): number { + this.count += 1 + return this.count + } +} + +let runtime: SolidObjectsRuntime | undefined +let stopRuntime: (() => Promise) | undefined + +afterEach(async () => { + await stopRuntime?.() + stopRuntime = undefined + vi.useRealTimers() + await runtime?.close() + runtime = undefined +}) + +function useControlledClock(): void { + vi.useFakeTimers({ + toFake: ["setTimeout", "clearTimeout", "setImmediate", "clearImmediate", "performance"], + }) +} + +async function startRuntime(options: { singleOwner: boolean }): Promise { + const database = new CountingDatabase(sqlite({ path: ":memory:" })) + runtime = createRuntime({ + database, + singleOwner: options.singleOwner, + authorizeMessage: () => true, + broadcast: async () => {}, + }) + runtime.register(Counter) + await runtime.install() + const controller = new AbortController() + const running = runtime.run(controller.signal) + stopRuntime = async () => { + controller.abort() + await running + } + return database +} + +async function settle(database: CountingDatabase): Promise { + let observed = -1 + while (observed !== database.statements.length) { + observed = database.statements.length + await vi.advanceTimersByTimeAsync(100) + } +} + +describe("an idle singleOwner runtime", () => { + it("polls without the option, so the measurement can see idle work", async () => { + useControlledClock() + const database = await startRuntime({ singleOwner: false }) + await vi.advanceTimersByTimeAsync(5_000) + const position = database.statements.length + + await vi.advanceTimersByTimeAsync(60_000) + + expect(database.statementsSince(position).length).toBeGreaterThan(100) + }) + + it("sends no statement in 60 seconds", async () => { + useControlledClock() + const database = await startRuntime({ singleOwner: true }) + await settle(database) + const position = database.statements.length + + await vi.advanceTimersByTimeAsync(60_000) + + expect(database.statementsSince(position)).toEqual([]) + }) + + it("renews a lease only while it holds the activation, then sends nothing", async () => { + useControlledClock() + const database = await startRuntime({ singleOwner: true }) + await settle(database) + await runtime!.ref(Counter, "one").send.increment() + await settle(database) + const activationHeld = database.statements.length + + await vi.advanceTimersByTimeAsync(31_000) + + const whileHeld = database.statementsSince(activationHeld) + expect( + whileHeld.filter((statement) => + statement.startsWith("UPDATE solid_objects_instances SET activation_expires_at_ms"), + ).length, + ).toBeGreaterThanOrEqual(2) + expect( + whileHeld.some((statement) => + statement.startsWith( + "UPDATE solid_objects_instances SET activation_owner_id = NULL, activation_token = NULL", + ), + ), + ).toBe(true) + await settle(database) + const released = database.statements.length + + await vi.advanceTimersByTimeAsync(60_000) + + expect(database.statementsSince(released)).toEqual([]) + }) +}) diff --git a/test/support/counting-database.ts b/test/support/counting-database.ts new file mode 100644 index 0000000..b53e4ed --- /dev/null +++ b/test/support/counting-database.ts @@ -0,0 +1,75 @@ +import type { + Database, + DatabaseConnection, + DatabaseFamily, + DatabaseTransactionOptions, +} from "../../src/database/types.js" + +type Parameters = readonly unknown[] | undefined + +export class CountingDatabase implements Database { + readonly family: DatabaseFamily + readonly schemaIdentity: string + readonly statements: string[] = [] + + constructor(private readonly database: Database) { + this.family = database.family + this.schemaIdentity = database.schemaIdentity + } + + connection( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return this.database.connection((connection) => callback(this.counting(connection))) + } + + transaction( + callback: (connection: DatabaseConnection) => Promise, + options?: DatabaseTransactionOptions, + ): Promise { + return this.database.transaction(async (connection) => { + this.statements.push("BEGIN") + try { + const result = await callback(this.counting(connection)) + this.statements.push("COMMIT") + return result + } catch (error) { + this.statements.push("ROLLBACK") + throw error + } + }, options) + } + + transactionActive(): boolean { + return this.database.transactionActive?.() ?? false + } + + close(): Promise { + return this.database.close() + } + + statementsSince(position: number): string[] { + return this.statements.slice(position).map((statement) => statement.replace(/\s+/g, " ")) + } + + private counting(connection: DatabaseConnection): DatabaseConnection { + return { + run: (sql: string, parameters?: Parameters) => { + this.statements.push(sql) + return connection.run(sql, parameters) + }, + get: (sql: string, parameters?: Parameters) => { + this.statements.push(sql) + return connection.get(sql, parameters) + }, + all: (sql: string, parameters?: Parameters) => { + this.statements.push(sql) + return connection.all(sql, parameters) + }, + nowMilliseconds: () => { + this.statements.push("SELECT now") + return connection.nowMilliseconds() + }, + } + } +} From 1797c60869637c3418bf3235d6b7f26716b5f4c1 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:32:13 -0800 Subject: [PATCH 04/15] feat: refuse singleOwner beside a live process The option is safe only while no other process writes the database, so run() now refuses with SingleOwnerConflict when another process row is alive by the usual heartbeat rule. The error names each row and the time at which it becomes stale. Rows of the runtime's own synchronous callers do not count. After the check, the runtime releases stale process rows once, which replaces the periodic cleanup that the option turns off. --- docs/api.md | 8 +- src/errors.ts | 27 +++++ src/index.ts | 2 + src/repository.ts | 13 +++ src/runtime.ts | 23 ++++ test/single-owner-start.test.ts | 179 ++++++++++++++++++++++++++++++++ 6 files changed, 249 insertions(+), 3 deletions(-) create mode 100644 test/single-owner-start.test.ts diff --git a/docs/api.md b/docs/api.md index 3321a01..2909e29 100644 --- a/docs/api.md +++ b/docs/api.md @@ -793,11 +793,13 @@ The root exports `SolidObjectsError` and its supported subclasses: `ScheduleWithoutOperation`, `StateMigrationError`, `ApplicationWriteForbidden`, `UnknownEffect`, and `UnknownCommitAction`; - operational failures: `LostActivation`, `DatabaseDeadlineExceeded`, - `UnknownDeadLetter`, `UnknownReminder`, `ReminderNotPaused`, and - `UnsupportedDatabase`. + `UnknownDeadLetter`, `UnknownReminder`, `ReminderNotPaused`, + `UnsupportedDatabase`, and `SingleOwnerConflict`. `NonRetryableError` is the application subclassing point. `SyncTimeoutDetails` -and `SyncTimeoutWaitingOn` type timeout diagnostics. See +and `SyncTimeoutWaitingOn` type timeout diagnostics. `SingleOwnerConflict` +carries one `SingleOwnerConflictProcess` for each live process row that stopped +a `singleOwner` runtime from starting. See [Errors and recovery](errors-and-recovery.md) before deciding what to catch. ## `solid-objects/database/sqlite` diff --git a/src/errors.ts b/src/errors.ts index a3605db..43b055b 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -17,6 +17,33 @@ export class EnqueueOutcomeUnknown extends SolidObjectsError { } } export class UnsupportedDatabase extends SolidObjectsError {} + +export interface SingleOwnerConflictProcess { + readonly id: string + readonly kind: string + readonly hostname: string + readonly hostProcessId: number + readonly heartbeatAt: Date + readonly staleAt: Date +} + +export class SingleOwnerConflict extends SolidObjectsError { + readonly processes: readonly SingleOwnerConflictProcess[] + + constructor(options: { processes: readonly SingleOwnerConflictProcess[] }) { + const descriptions = options.processes.map( + (process) => + `${process.kind} ${process.id} on ${process.hostname} with process ID ` + + `${process.hostProcessId}, alive until ${process.staleAt.toISOString()}`, + ) + super( + `singleOwner refuses to start, because another process row is alive: ` + + `${descriptions.join("; ")}. Stop that process, or wait until its row is stale, ` + + `and start again`, + ) + this.processes = Object.freeze(options.processes.map((process) => Object.freeze(process))) + } +} export class DatabaseDeadlineExceeded extends SolidObjectsError {} export class SyncEnqueueTimeout extends SolidObjectsError { constructor( diff --git a/src/index.ts b/src/index.ts index 25c658f..af50848 100644 --- a/src/index.ts +++ b/src/index.ts @@ -201,6 +201,8 @@ export { Rejected, ReminderNotPaused, ScheduleWithoutOperation, + SingleOwnerConflict, + type SingleOwnerConflictProcess, SolidObjectsError, StateMigrationError, SyncTimeout, diff --git a/src/repository.ts b/src/repository.ts index c1db1b7..33333f6 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -149,6 +149,19 @@ export class Repository { }) } + async liveProcesses(options: { excludedProcessIds: readonly string[] }): Promise { + const rows = await this.settings.database.connection(async (connection) => { + const now = await connection.nowMilliseconds() + return connection.all( + `SELECT * FROM ${this.table("processes")} + WHERE shutdown_state <> 'stopped' AND heartbeat_at_ms > ? + ORDER BY started_at_ms, id`, + [now - this.settings.processAliveThresholdMilliseconds], + ) + }) + return rows.filter((row) => !options.excludedProcessIds.includes(row.id)) + } + async hasLiveProcessOutsideCurrentHostProcess(): Promise { return this.settings.database.connection(async (connection) => { const now = await connection.nowMilliseconds() diff --git a/src/runtime.ts b/src/runtime.ts index 6112405..f7b550d 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -46,6 +46,7 @@ import { NonRetryableError, QueryMutatedState, Rejected, + SingleOwnerConflict, SyncTimeout, SyncEnqueueTimeout, SyncInsideTransaction, @@ -345,6 +346,7 @@ export class SolidObjectsRuntime { let slots: ComponentSlot[] = [] try { + if (this.settings.singleOwner) await this.claimSingleOwnership() slots = await this.buildComponentSlots() await Promise.all( slots.map((slot) => @@ -1122,6 +1124,27 @@ export class SolidObjectsRuntime { return Object.freeze({ cleaned }) } + private async claimSingleOwnership(): Promise { + const live = await this.repository.liveProcesses({ + excludedProcessIds: [...this.callerWorkers].map((worker) => worker.processId), + }) + if (live.length > 0) { + throw new SingleOwnerConflict({ + processes: live.map((row) => ({ + id: row.id, + kind: row.kind, + hostname: row.hostname, + hostProcessId: Number(row.host_process_id), + heartbeatAt: new Date(Number(row.heartbeat_at_ms)), + staleAt: new Date( + Number(row.heartbeat_at_ms) + this.settings.processAliveThresholdMilliseconds, + ), + })), + }) + } + await this.cleanupStaleProcesses() + } + private async cleanupStaleProcesses(): Promise { const cleaned = await this.repository.cleanupStaleProcesses() if (cleaned > 0) { diff --git a/test/single-owner-start.test.ts b/test/single-owner-start.test.ts new file mode 100644 index 0000000..734fdfc --- /dev/null +++ b/test/single-owner-start.test.ts @@ -0,0 +1,179 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { SingleOwnerConflict } from "../src/errors.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" + +class Counter extends Actor { + static override readonly actorType = "SingleOwnerStartCounter" + + count = 0 + + increment(): number { + this.count += 1 + return this.count + } +} + +const aliveThreshold = 60_000 + +let runtime: SolidObjectsRuntime | undefined +let stopRuntime: (() => Promise) | undefined + +afterEach(async () => { + await stopRuntime?.() + stopRuntime = undefined + await runtime?.close() + runtime = undefined +}) + +async function installedRuntime(): Promise { + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: true, + processAliveThresholdMilliseconds: aliveThreshold, + authorizeMessage: () => true, + authorizeQuery: () => true, + }) + runtime.register(Counter) + await runtime.install() + return runtime +} + +function start(target: SolidObjectsRuntime): Promise { + const controller = new AbortController() + const running = target.run(controller.signal) + stopRuntime = async () => { + controller.abort() + await running + } + return running +} + +async function insertProcess(options: { + id: string + heartbeatAgeMilliseconds: number + shutdownState: "running" | "draining" | "stopped" +}): Promise { + await runtime!.settings.database.transaction(async (connection) => { + const now = await connection.nowMilliseconds() + await connection.run( + `INSERT INTO solid_objects_processes + (id, kind, hostname, host_process_id, metadata, started_at_ms, heartbeat_at_ms, shutdown_state) + VALUES (?, 'worker', 'other-host', 4242, '{}', ?, ?, ?)`, + [ + options.id, + now - options.heartbeatAgeMilliseconds, + now - options.heartbeatAgeMilliseconds, + options.shutdownState, + ], + ) + }) +} + +async function shutdownStateOf(id: string): Promise { + const row = await runtime!.settings.database.connection((connection) => + connection.get<{ shutdown_state: string }>( + "SELECT shutdown_state FROM solid_objects_processes WHERE id = ?", + [id], + ), + ) + return row?.shutdown_state +} + +async function waitForStatus(options: { + status: () => Promise + expected: string + timeoutMilliseconds: number +}): Promise { + const deadline = performance.now() + options.timeoutMilliseconds + let status = await options.status() + while (status !== options.expected && performance.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 10)) + status = await options.status() + } + return status +} + +describe("singleOwner start", () => { + it.each(["running", "draining"] as const)( + "refuses to start while another %s process row is alive", + async (shutdownState) => { + const target = await installedRuntime() + await insertProcess({ id: "other", heartbeatAgeMilliseconds: 1_000, shutdownState }) + + const failure = await target.run(new AbortController().signal).catch((error) => error) + + expect(failure).toBeInstanceOf(SingleOwnerConflict) + expect(failure.message).toContain( + "singleOwner refuses to start, because another process row is alive", + ) + expect(failure.message).toContain("worker other on other-host with process ID 4242") + expect(failure.processes).toEqual([ + expect.objectContaining({ id: "other", kind: "worker", hostProcessId: 4242 }), + ]) + expect(failure.processes[0].staleAt.getTime()).toBe( + failure.processes[0].heartbeatAt.getTime() + aliveThreshold, + ) + expect(await shutdownStateOf("other")).toBe(shutdownState) + }, + ) + + it("starts again once the other process has stopped", async () => { + const target = await installedRuntime() + await insertProcess({ id: "other", heartbeatAgeMilliseconds: 1_000, shutdownState: "running" }) + await expect(target.run(new AbortController().signal)).rejects.toThrow(SingleOwnerConflict) + await target.settings.database.transaction((connection) => + connection.run("UPDATE solid_objects_processes SET shutdown_state = 'stopped'"), + ) + + const message = await target.ref(Counter, "after").send.increment() + void start(target) + + expect( + await waitForStatus({ + status: () => message.status(), + expected: "completed", + timeoutMilliseconds: 2_000, + }), + ).toBe("completed") + }) + + it("ignores the process rows of its own synchronous callers", async () => { + const target = await installedRuntime() + expect(await target.ref(Counter, "caller").increment()).toBe(1) + + const message = await target.ref(Counter, "caller").send.increment() + void start(target) + + expect( + await waitForStatus({ + status: () => message.status(), + expected: "completed", + timeoutMilliseconds: 2_000, + }), + ).toBe("completed") + }) + + it("releases a stale process once at start, so its claimed message runs at once", async () => { + const target = await installedRuntime() + const message = await target.ref(Counter, "stale").send.increment() + await insertProcess({ + id: "crashed", + heartbeatAgeMilliseconds: aliveThreshold + 1_000, + shutdownState: "running", + }) + expect(await target.repository.claim("crashed")).toBeDefined() + + void start(target) + + expect( + await waitForStatus({ + status: () => message.status(), + expected: "completed", + timeoutMilliseconds: 2_000, + }), + ).toBe("completed") + expect(await shutdownStateOf("crashed")).toBe("stopped") + }) +}) From 0df411447fe56167e8da9dd452f92310db907560 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:35:29 -0800 Subject: [PATCH 05/15] fix: keep the row of a single owner fresh on claim A singleOwner worker writes its process row only at start, so after an idle minute the row looks stale by the alive rule. A worker that then claimed work had a stale owner row until its first heartbeat. In that window a second worker could reset and run the same effect, take the same broadcast or reminder, the effect recovery could retire a running effect, and the process cleanup could release a fresh claim. With the option, each claim now writes the claiming row in the same transaction, as registration does. An empty pass still writes nothing. --- src/repository.ts | 77 ++++++++++---- test/single-owner-claims.test.ts | 175 +++++++++++++++++++++++++++++++ test/single-owner-start.test.ts | 6 ++ 3 files changed, 236 insertions(+), 22 deletions(-) create mode 100644 test/single-owner-claims.test.ts diff --git a/src/repository.ts b/src/repository.ts index 33333f6..98e0a68 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -81,31 +81,50 @@ export class Repository { async registerProcess(processId: string, kind: string): Promise { await this.settings.database.transaction(async (connection) => { const now = await connection.nowMilliseconds() - await connection.run( - `INSERT INTO ${this.table("processes")} - (id, kind, hostname, host_process_id, metadata, started_at_ms, heartbeat_at_ms, - shutdown_state) - VALUES (?, ?, ?, ?, ?, ?, ?, 'running') - ON CONFLICT(id) DO UPDATE SET heartbeat_at_ms = excluded.heartbeat_at_ms, - hostname = excluded.hostname, host_process_id = excluded.host_process_id, - metadata = excluded.metadata, shutdown_requested_at_ms = NULL, - stopped_at_ms = NULL, shutdown_state = 'running'`, - [ - processId, - kind, - hostIdentity().hostname, - hostIdentity().hostProcessId, - JSON.stringify({ - solidObjectsVersion: VERSION, - nodeVersion: hostIdentity().runtimeVersion, - }), - now, - now, - ], - ) + await this.writeProcessRow({ connection, processId, kind, now }) }) } + private async writeProcessRow(options: { + connection: DatabaseConnection + processId: string + kind: string + now: number + }): Promise { + await options.connection.run( + `INSERT INTO ${this.table("processes")} + (id, kind, hostname, host_process_id, metadata, started_at_ms, heartbeat_at_ms, + shutdown_state) + VALUES (?, ?, ?, ?, ?, ?, ?, 'running') + ON CONFLICT(id) DO UPDATE SET heartbeat_at_ms = excluded.heartbeat_at_ms, + hostname = excluded.hostname, host_process_id = excluded.host_process_id, + metadata = excluded.metadata, shutdown_requested_at_ms = NULL, + stopped_at_ms = NULL, shutdown_state = 'running'`, + [ + options.processId, + options.kind, + hostIdentity().hostname, + hostIdentity().hostProcessId, + JSON.stringify({ + solidObjectsVersion: VERSION, + nodeVersion: hostIdentity().runtimeVersion, + }), + options.now, + options.now, + ], + ) + } + + private async refreshClaimingProcess(options: { + connection: DatabaseConnection + processId: string + kind: string + now: number + }): Promise { + if (!this.settings.singleOwner) return + await this.writeProcessRow(options) + } + async startDrainingProcess(processId: string): Promise { await this.settings.database.transaction(async (connection) => { const now = await connection.nowMilliseconds() @@ -547,6 +566,7 @@ export class Repository { `UPDATE ${this.table("messages")} SET attempt_count = attempt_count + 1, updated_at_ms = ? WHERE id = ?`, [now, message.id], ) + await this.refreshClaimingProcess({ connection, processId, kind: "worker", now }) message.attempt_count = Number(message.attempt_count) + 1 return { instance, @@ -1607,6 +1627,7 @@ export class Repository { [processId, effect.id], ) if (claimed.changes !== 1) return undefined + await this.refreshClaimingProcess({ connection, processId, kind: "effect_worker", now }) const identity = await this.loadActorIdentity(connection, effect.instance_id) return { ...effect, @@ -1746,6 +1767,12 @@ export class Repository { skippedCandidateIds.push(reminder.id) continue } + await this.refreshClaimingProcess({ + connection, + processId, + kind: "reminder_scheduler", + now: databaseNow, + }) const identity = await this.loadActorIdentity(connection, reminder.instance_id) return { ...reminder, @@ -1979,6 +2006,12 @@ export class Repository { skippedCandidateIds.push(broadcast.id) continue } + await this.refreshClaimingProcess({ + connection, + processId, + kind: "broadcast_worker", + now, + }) broadcast.status = "processing" broadcast.claimed_by = processId broadcast.attempt_count = Number(broadcast.attempt_count) + 1 diff --git a/test/single-owner-claims.test.ts b/test/single-owner-claims.test.ts new file mode 100644 index 0000000..38d573a --- /dev/null +++ b/test/single-owner-claims.test.ts @@ -0,0 +1,175 @@ +import { afterEach, describe, expect, it } from "vitest" +import { sqlite } from "../src/database/sqlite.js" +import type { MessageRow } from "../src/records.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" + +const aliveThreshold = 60_000 +const idleAge = 2 * aliveThreshold + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + await runtime?.close() + runtime = undefined +}) + +async function installedRuntime(options: { singleOwner: boolean }): Promise { + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: options.singleOwner, + processAliveThresholdMilliseconds: aliveThreshold, + authorizeAdministration: () => true, + }) + await runtime.install() + return runtime +} + +async function write(sql: string, parameters: readonly unknown[] = []): Promise { + await runtime!.settings.database.transaction((connection) => connection.run(sql, parameters)) +} + +async function registerIdleWorker(options: { id: string; kind: string }): Promise { + await runtime!.repository.registerProcess(options.id, options.kind) + await write( + "UPDATE solid_objects_processes SET heartbeat_at_ms = heartbeat_at_ms - ? WHERE id = ?", + [idleAge, options.id], + ) +} + +async function processRow(id: string) { + return runtime!.settings.database.connection(async (connection) => ({ + now: await connection.nowMilliseconds(), + row: await connection.get<{ heartbeat_at_ms: number | bigint; shutdown_state: string }>( + "SELECT heartbeat_at_ms, shutdown_state FROM solid_objects_processes WHERE id = ?", + [id], + ), + })) +} + +function enqueue(actorId: string): Promise { + return runtime!.repository.enqueue({ + actorType: "Probe", + actorId, + operation: "work", + deliveryMode: "async", + arguments: {}, + }) +} + +async function seedEffect(options: { recoverable: boolean }): Promise { + const message = await enqueue("effects") + await write( + `INSERT INTO solid_objects_effects + (id, message_id, instance_id, name, arguments, status, max_attempts, available_at_ms) + VALUES ('effect', ?, ?, 'work', '{}', 'pending', 3, 0)`, + [message.id, message.instance_id], + ) + if (!options.recoverable) return + await write( + `INSERT INTO solid_objects_effect_recoveries + (effect_id, instance_id, recovery_operation, status_operation) + VALUES ('effect', ?, 'recover', 'inspect')`, + [message.instance_id], + ) +} + +async function effectState() { + return runtime!.settings.database.connection((connection) => + connection.get<{ status: string; claimed_by: string | null }>( + "SELECT status, claimed_by FROM solid_objects_effects WHERE id = 'effect'", + ), + ) +} + +describe("a singleOwner claim", () => { + it("refreshes the process row of a worker that was idle, in the claim transaction", async () => { + await installedRuntime({ singleOwner: true }) + await registerIdleWorker({ id: "idle", kind: "effect_worker" }) + await write("UPDATE solid_objects_processes SET shutdown_state = 'stopped' WHERE id = 'idle'") + await seedEffect({ recoverable: false }) + + expect(await runtime!.repository.claimEffect("idle")).toBeDefined() + + const { now, row } = await processRow("idle") + expect(row?.shutdown_state).toBe("running") + expect(Number(row?.heartbeat_at_ms)).toBeGreaterThan(now - 1_000) + }) + + it("keeps an effect from a second worker after the first worker was idle", async () => { + await installedRuntime({ singleOwner: true }) + await registerIdleWorker({ id: "first", kind: "effect_worker" }) + await runtime!.repository.registerProcess("second", "effect_worker") + await seedEffect({ recoverable: false }) + + expect(await runtime!.repository.claimEffect("first")).toBeDefined() + expect(await runtime!.repository.claimEffect("second")).toBeUndefined() + + expect(await effectState()).toEqual({ status: "processing", claimed_by: "first" }) + }) + + it("keeps a running effect with a recovery operation from being retired", async () => { + await installedRuntime({ singleOwner: true }) + await registerIdleWorker({ id: "first", kind: "effect_worker" }) + await seedEffect({ recoverable: true }) + + expect(await runtime!.repository.claimEffect("first")).toBeDefined() + await runtime!.processes.cleanup() + + expect(await effectState()).toEqual({ status: "processing", claimed_by: "first" }) + }) + + it("keeps a claimed message from the process cleanup", async () => { + await installedRuntime({ singleOwner: true }) + await registerIdleWorker({ id: "first", kind: "worker" }) + await enqueue("actor") + + expect(await runtime!.repository.claim("first")).toBeDefined() + + expect(await runtime!.processes.cleanup()).toEqual({ cleaned: 0 }) + }) + + it("keeps a reminder from a second scheduler after the first was idle", async () => { + await installedRuntime({ singleOwner: true }) + await registerIdleWorker({ id: "first", kind: "reminder_scheduler" }) + await runtime!.repository.registerProcess("second", "reminder_scheduler") + const message = await enqueue("reminders") + await write( + `INSERT INTO solid_objects_reminders + (id, instance_id, operation, run_at_ms, arguments, missed_policy, status) + VALUES ('reminder', ?, 'work', 0, '{}', 'latest', 'scheduled')`, + [message.instance_id], + ) + + expect(await runtime!.repository.claimReminder("first")).toBeDefined() + expect(await runtime!.repository.claimReminder("second")).toBeUndefined() + }) + + it("keeps a broadcast from a second worker after the first was idle", async () => { + await installedRuntime({ singleOwner: true }) + await registerIdleWorker({ id: "first", kind: "broadcast_worker" }) + await runtime!.repository.registerProcess("second", "broadcast_worker") + const message = await enqueue("broadcasts") + await write( + `INSERT INTO solid_objects_broadcasts + (id, message_id, instance_id, actor_type, actor_id, state_revision, observables, status, + available_at_ms) + VALUES ('broadcast', ?, ?, 'Probe', 'broadcasts', 1, '{}', 'pending', 0)`, + [message.id, message.instance_id], + ) + + expect(await runtime!.repository.claimBroadcast("first")).toBeDefined() + expect(await runtime!.repository.claimBroadcast("second")).toBeUndefined() + }) + + it("leaves the process row alone without the option", async () => { + await installedRuntime({ singleOwner: false }) + await registerIdleWorker({ id: "idle", kind: "effect_worker" }) + const before = await processRow("idle") + await seedEffect({ recoverable: false }) + + expect(await runtime!.repository.claimEffect("idle")).toBeDefined() + + const after = await processRow("idle") + expect(after.row?.heartbeat_at_ms).toBe(before.row?.heartbeat_at_ms) + }) +}) diff --git a/test/single-owner-start.test.ts b/test/single-owner-start.test.ts index 734fdfc..8848629 100644 --- a/test/single-owner-start.test.ts +++ b/test/single-owner-start.test.ts @@ -164,6 +164,12 @@ describe("singleOwner start", () => { shutdownState: "running", }) expect(await target.repository.claim("crashed")).toBeDefined() + await target.settings.database.transaction((connection) => + connection.run( + "UPDATE solid_objects_processes SET heartbeat_at_ms = heartbeat_at_ms - ? WHERE id = 'crashed'", + [aliveThreshold + 1_000], + ), + ) void start(target) From 5ff2dd3252f49c5a82db8f1a950cf86e7733d13d Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:39:18 -0800 Subject: [PATCH 06/15] fix: report a held retry to its own worker A message that failed and waits for its retry keeps the cached activation of its worker. The due time read the lease expiry for every held activation, so the holder slept until it renewed its lease, and the retry ran seconds late. The holder is the one worker that can claim that message, so its due time is now the time the message is available. Other workers still wait for the lease to expire. --- src/due-work.ts | 5 +++-- src/worker.ts | 2 +- test/due-work.test.ts | 27 +++++++++++++++++++++------ 3 files changed, 25 insertions(+), 9 deletions(-) diff --git a/src/due-work.ts b/src/due-work.ts index 0179a1c..6da8024 100644 --- a/src/due-work.ts +++ b/src/due-work.ts @@ -7,11 +7,12 @@ interface DueRow { export class DueWork { constructor(private readonly settings: RuntimeSettings) {} - untilActorWork(): Promise { + untilActorWork(options: { processId: string }): Promise { return this.untilDue(() => ({ sql: `SELECT MIN(due_at_ms) AS due_at_ms FROM ( SELECT CASE WHEN instances.activation_owner_id IS NOT NULL + AND instances.activation_owner_id <> ? AND instances.activation_expires_at_ms > ready.available_at_ms THEN instances.activation_expires_at_ms ELSE ready.available_at_ms @@ -33,7 +34,7 @@ export class DueWork { JOIN ${this.table("instances")} instances ON instances.id = claimed.instance_id WHERE instances.activation_expires_at_ms IS NOT NULL ) due`, - parameters: [], + parameters: [options.processId], })) } diff --git a/src/worker.ts b/src/worker.ts index c13db48..d85ce1e 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -197,7 +197,7 @@ export class Worker { const databaseWait = await idleWaitMilliseconds({ settings: this.runtime.settings, pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, - untilDue: () => this.runtime.dueWork.untilActorWork(), + untilDue: () => this.runtime.dueWork.untilActorWork({ processId: this.processId }), }) if (!this.runtime.settings.singleOwner) return databaseWait return Math.min(databaseWait, this.untilCachedActivationDue()) diff --git a/test/due-work.test.ts b/test/due-work.test.ts index 4b3489f..e5176dd 100644 --- a/test/due-work.test.ts +++ b/test/due-work.test.ts @@ -145,7 +145,7 @@ function expectAbout(actual: number | undefined, expected: number): void { describe("DueWork", () => { it("reports no due time when no work exists", async () => { - expect(await dueWork.untilActorWork()).toBeUndefined() + expect(await dueWork.untilActorWork({ processId: "worker" })).toBeUndefined() expect(await dueWork.untilEffectWork()).toBeUndefined() expect(await dueWork.untilReminderWork()).toBeUndefined() expect(await dueWork.untilBroadcastWork()).toBeUndefined() @@ -158,7 +158,7 @@ describe("DueWork", () => { await enqueue({ actorId: "late", availableAtMilliseconds: now + 9_000 }) await enqueue({ actorId: "early", availableAtMilliseconds: now + 5_000 }) - expectAbout(await dueWork.untilActorWork(), 5_000) + expectAbout(await dueWork.untilActorWork({ processId: "worker" }), 5_000) }) it("waits for the head of a mailbox, not for a later message behind it", async () => { @@ -166,7 +166,7 @@ describe("DueWork", () => { await enqueue({ actorId: "blocked", availableAtMilliseconds: now + 8_000 }) await enqueue({ actorId: "blocked" }) - expectAbout(await dueWork.untilActorWork(), 8_000) + expectAbout(await dueWork.untilActorWork({ processId: "worker" }), 8_000) }) it("skips a paused actor", async () => { @@ -175,7 +175,7 @@ describe("DueWork", () => { message.instance_id, ]) - expect(await dueWork.untilActorWork()).toBeUndefined() + expect(await dueWork.untilActorWork({ processId: "worker" })).toBeUndefined() }) it("waits for an activation that another owner holds", async () => { @@ -189,7 +189,22 @@ describe("DueWork", () => { [now + 7_000, message.instance_id], ) - expectAbout(await dueWork.untilActorWork(), 7_000) + expectAbout(await dueWork.untilActorWork({ processId: "worker" }), 7_000) + }) + + it("reports a message at its available time to the worker that holds the activation", async () => { + const now = await databaseNow() + const message = await enqueue({ actorId: "held", availableAtMilliseconds: now + 5_000 }) + await registerOwner({ id: "holder", heartbeatAtMilliseconds: now }) + await write( + `UPDATE solid_objects_instances + SET activation_owner_id = 'holder', activation_token = 'token', activation_expires_at_ms = ? + WHERE id = ?`, + [now + leaseDuration, message.instance_id], + ) + + expectAbout(await dueWork.untilActorWork({ processId: "holder" }), 5_000) + expectAbout(await dueWork.untilActorWork({ processId: "other" }), leaseDuration) }) it("waits for the lease of a claimed message to expire", async () => { @@ -198,7 +213,7 @@ describe("DueWork", () => { await registerOwner({ id: "claimer", heartbeatAtMilliseconds: now }) expect(await runtime.repository.claim("claimer")).toBeDefined() - expectAbout(await dueWork.untilActorWork(), leaseDuration) + expectAbout(await dueWork.untilActorWork({ processId: "worker" }), leaseDuration) }) }) From 5f24ca003b072230a2e1813f674b230bf12e3fa0 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:39:18 -0800 Subject: [PATCH 07/15] feat: wake actor workers after a singleOwner retry A synchronous caller can record a retry and then stop waiting. With polling off, a background worker that slept before that write learned of the retry only when the caller's lease expired. With singleOwner, the runtime now notifies the actor workers after it records a retry. The new tests show the option at work with hour-long polling intervals: a sent message, an effect, a broadcast, a due reminder, a message retry and an effect retry run promptly, and a reminder or message that a later write moves earlier wakes its role for the earlier time. --- src/runtime.ts | 1 + test/single-owner-work.test.ts | 290 +++++++++++++++++++++++++++++++++ 2 files changed, 291 insertions(+) create mode 100644 test/single-owner-work.test.ts diff --git a/src/runtime.ts b/src/runtime.ts index f7b550d..180a7a7 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -1483,6 +1483,7 @@ export class SolidObjectsRuntime { }) if (outcome === "retrying") this.emitInstrumentation("message.retry", messageInstrumentation(turn.message)) + if (outcome === "retrying" && this.settings.singleOwner) this.wakeUp("actors") if (outcome === "dead" && recoveryMessage(turn.message)) this.emitInstrumentation("recovery.failed", messageInstrumentation(turn.message)) if (outcome === "dead") diff --git a/test/single-owner-work.test.ts b/test/single-owner-work.test.ts new file mode 100644 index 0000000..f2cf796 --- /dev/null +++ b/test/single-owner-work.test.ts @@ -0,0 +1,290 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import type { BroadcastEvent } from "../src/configuration.js" +import { sqlite } from "../src/database/sqlite.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" + +const anHour = 3_600_000 +const delay = 300 +const promptly = 700 + +class Workbench extends Actor { + static override readonly actorType = "SingleOwnerWorkbench" + static flakyAttempts = 0 + static remindedAt: number[] = [] + + pings = 0 + + ping(): number { + this.pings += 1 + return this.pings + } + + emitWork(): void { + this.emit("work") + } + + emitFlakyWork(): void { + this.emit("flaky-work") + } + + remindAt(input: { at: number }): void { + this.schedule({ at: new Date(input.at) }).remind() + } + + remind(): void { + Workbench.remindedAt.push(performance.now()) + } + + flaky(): number { + Workbench.flakyAttempts += 1 + if (Workbench.flakyAttempts === 1) throw new Error("the first attempt fails") + return Workbench.flakyAttempts + } + + override observables(): Record { + return { pings: this.pings } + } +} + +let runtime: SolidObjectsRuntime | undefined +let stopRuntime: (() => Promise) | undefined +let broadcasts: BroadcastEvent[] = [] +let effectRuns: number[] = [] +let flakyEffectRuns: number[] = [] + +afterEach(async () => { + await stopRuntime?.() + stopRuntime = undefined + await runtime?.close() + runtime = undefined + broadcasts = [] + effectRuns = [] + flakyEffectRuns = [] + Workbench.flakyAttempts = 0 + Workbench.remindedAt = [] +}) + +async function startedRuntime(options: { singleOwner: boolean }): Promise { + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: options.singleOwner, + pollingIntervalMilliseconds: anHour, + idlePollingIntervalMilliseconds: anHour, + retryDelayMilliseconds: () => delay, + authorizeMessage: () => true, + authorizeQuery: () => true, + broadcast: async (event) => { + broadcasts.push(event) + }, + }) + runtime.register(Workbench) + runtime.registerEffect("work", () => { + effectRuns.push(performance.now()) + }) + runtime.registerEffect("flaky-work", () => { + flakyEffectRuns.push(performance.now()) + if (flakyEffectRuns.length === 1) throw new Error("the first effect attempt fails") + }) + await runtime.install() + const controller = new AbortController() + const running = runtime.run(controller.signal) + stopRuntime = async () => { + controller.abort() + await running + } + await settled() + return runtime +} + +function settled(): Promise { + return new Promise((resolve) => setTimeout(resolve, 50)) +} + +async function eventually(options: { + read: () => Value | Promise + until: (value: Value) => boolean + timeoutMilliseconds: number +}): Promise { + const deadline = performance.now() + options.timeoutMilliseconds + let value = await options.read() + while (!options.until(value) && performance.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 5)) + value = await options.read() + } + return value +} + +async function databaseNow(): Promise { + return runtime!.settings.database.connection((connection) => connection.nowMilliseconds()) +} + +describe("a singleOwner runtime with work", () => { + it("processes a message that is sent to an actor", async () => { + const target = await startedRuntime({ singleOwner: true }) + + const message = await target.ref(Workbench, "message").send.ping() + + expect( + await eventually({ + read: () => message.status(), + until: (status) => status === "completed", + timeoutMilliseconds: promptly, + }), + ).toBe("completed") + }) + + it("runs an effect", async () => { + const target = await startedRuntime({ singleOwner: true }) + + await target.ref(Workbench, "effect").send.emitWork() + + expect( + await eventually({ + read: () => effectRuns.length, + until: (count) => count === 1, + timeoutMilliseconds: promptly, + }), + ).toBe(1) + }) + + it("delivers a broadcast", async () => { + const target = await startedRuntime({ singleOwner: true }) + + await target.ref(Workbench, "broadcast").send.ping() + + const delivered = await eventually({ + read: () => broadcasts, + until: (events) => events.length === 1, + timeoutMilliseconds: promptly, + }) + expect(delivered).toEqual([ + expect.objectContaining({ actorId: "broadcast", invalidations: ["pings"] }), + ]) + }) + + it("runs a reminder when it comes due", async () => { + const target = await startedRuntime({ singleOwner: true }) + const scheduledAt = performance.now() + + await target.ref(Workbench, "reminder").send.remindAt({ at: Date.now() + delay }) + + const reminded = await eventually({ + read: () => Workbench.remindedAt, + until: (times) => times.length === 1, + timeoutMilliseconds: delay + promptly, + }) + expect(reminded).toHaveLength(1) + expect(reminded[0]! - scheduledAt).toBeGreaterThanOrEqual(delay - 50) + }) + + it("retries a failed message when its retry comes due", async () => { + const target = await startedRuntime({ singleOwner: true }) + + const message = await target.ref(Workbench, "retry").send.flaky() + + expect( + await eventually({ + read: () => message.status(), + until: (status) => status === "completed", + timeoutMilliseconds: delay + promptly, + }), + ).toBe("completed") + expect(Workbench.flakyAttempts).toBe(2) + }) + + it("retries a failed effect when its retry comes due", async () => { + const target = await startedRuntime({ singleOwner: true }) + + await target.ref(Workbench, "effect-retry").send.emitFlakyWork() + + const runs = await eventually({ + read: () => flakyEffectRuns, + until: (times) => times.length === 2, + timeoutMilliseconds: delay + promptly, + }) + expect(runs).toHaveLength(2) + expect(runs[1]! - runs[0]!).toBeGreaterThanOrEqual(delay - 50) + }) + + it("waits an hour for a retry without the option, which shows that the option does the work", async () => { + const target = await startedRuntime({ singleOwner: false }) + + const message = await target.ref(Workbench, "polling").send.flaky() + + expect( + await eventually({ + read: () => message.status(), + until: (status) => status === "completed", + timeoutMilliseconds: delay + promptly, + }), + ).toBe("ready") + }) +}) + +describe("a due time that a later write moves earlier", () => { + it("wakes the reminder scheduler for the earlier time", async () => { + const target = await startedRuntime({ singleOwner: true }) + const reference = target.ref(Workbench, "moved-reminder") + await reference.send.remindAt({ at: Date.now() + anHour }) + await settled() + const movedAt = performance.now() + + await reference.send.remindAt({ at: Date.now() + delay }) + + const reminded = await eventually({ + read: () => Workbench.remindedAt, + until: (times) => times.length === 1, + timeoutMilliseconds: delay + promptly, + }) + expect(reminded).toHaveLength(1) + expect(reminded[0]! - movedAt).toBeGreaterThanOrEqual(delay - 50) + }) + + it("wakes the actor worker for an earlier available message", async () => { + const target = await startedRuntime({ singleOwner: true }) + const now = await databaseNow() + const late = await target + .ref(Workbench, "late") + .send.with({ availableAt: new Date(now + anHour) }) + .ping() + await settled() + + const early = await target + .ref(Workbench, "early") + .send.with({ availableAt: new Date(now + delay) }) + .ping() + + expect( + await eventually({ + read: () => early.status(), + until: (status) => status === "completed", + timeoutMilliseconds: delay + promptly, + }), + ).toBe("completed") + expect(await late.status()).toBe("ready") + }) +}) + +describe("a retry that a synchronous caller records", () => { + it.each([ + { singleOwner: true, notified: true }, + { singleOwner: false, notified: false }, + ])("notifies the actor workers when singleOwner is $singleOwner", async (options) => { + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: options.singleOwner, + retryDelayMilliseconds: () => anHour, + authorizeMessage: () => true, + }) + runtime.register(Workbench) + await runtime.install() + await runtime.ref(Workbench, "caller").send.flaky() + const watch = await runtime.watchWakeUp("actors") + + expect(await runtime.worker().runOnce({ activationRetention: "release" })).toBe(1) + await settled() + + expect(await watch.wait({ timeoutMilliseconds: 0 })).toBe(options.notified) + }) +}) From 9c1ee6ded959f90b1bd11c954e45b357f210b0e1 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:41:32 -0800 Subject: [PATCH 08/15] test: cover singleOwner redrive and recovery Two paths have no message to wake them. A redrive starts through an administration call, and an effect of a crashed owner becomes recoverable only when its recovery timeout ends. The tests show that a singleOwner runtime with hour-long polling runs both on time. --- test/single-owner-recovery.test.ts | 154 +++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 test/single-owner-recovery.test.ts diff --git a/test/single-owner-recovery.test.ts b/test/single-owner-recovery.test.ts new file mode 100644 index 0000000..cbc6778 --- /dev/null +++ b/test/single-owner-recovery.test.ts @@ -0,0 +1,154 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { NonRetryableError } from "../src/errors.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" + +const anHour = 3_600_000 +const aliveThreshold = 1_000 +const promptly = 700 + +class Exporter extends Actor { + static override readonly actorType = "SingleOwnerExporter" + static recoveredAt: number[] = [] + + ping(): void {} + + emitExport(): void { + this.emit("export") + } + + recovered(): void { + Exporter.recoveredAt.push(performance.now()) + } + + inspected(): void {} +} + +let runtime: SolidObjectsRuntime | undefined +let stopRuntime: (() => Promise) | undefined +let exportRuns = 0 + +afterEach(async () => { + await stopRuntime?.() + stopRuntime = undefined + await runtime?.close() + runtime = undefined + exportRuns = 0 + Exporter.recoveredAt = [] +}) + +async function installedRuntime(options: { singleOwner: boolean }): Promise { + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: options.singleOwner, + pollingIntervalMilliseconds: anHour, + idlePollingIntervalMilliseconds: anHour, + processAliveThresholdMilliseconds: aliveThreshold, + authorizeMessage: () => true, + authorizeAdministration: () => true, + }) + runtime.register(Exporter) + runtime.registerEffect("export", () => { + exportRuns += 1 + if (exportRuns === 1) throw new NonRetryableError("the export service was down") + }) + await runtime.install() + return runtime +} + +function start(target: SolidObjectsRuntime): void { + const controller = new AbortController() + const running = target.run(controller.signal) + stopRuntime = async () => { + controller.abort() + await running + } +} + +async function eventually(options: { + read: () => Value | Promise + until: (value: Value) => boolean + timeoutMilliseconds: number +}): Promise { + const deadline = performance.now() + options.timeoutMilliseconds + let value = await options.read() + while (!options.until(value) && performance.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 5)) + value = await options.read() + } + return value +} + +describe("singleOwner recovery work", () => { + it.each([ + { singleOwner: true, runs: 2 }, + { singleOwner: false, runs: 1 }, + ])("runs a redriven effect promptly only when singleOwner is $singleOwner", async (options) => { + const target = await installedRuntime({ singleOwner: options.singleOwner }) + start(target) + await target.ref(Exporter, "redrive").send.emitExport() + expect( + await eventually({ + read: async () => (await target.deadLetters.effects.all()).length, + until: (count) => count === 1, + timeoutMilliseconds: promptly, + }), + ).toBe(1) + + await target.deadLetters.effects.redrive() + + expect( + await eventually({ + read: () => exportRuns, + until: (runs) => runs === 2, + timeoutMilliseconds: promptly, + }), + ).toBe(options.runs) + }) + + it("retires the effect of a crashed owner when its recovery timeout ends", async () => { + const target = await installedRuntime({ singleOwner: true }) + const message = await target.repository.enqueue({ + actorType: Exporter.actorType, + actorId: "recovery", + operation: "ping", + deliveryMode: "async", + arguments: {}, + }) + const recoveryTimeout = aliveThreshold + 600 + await target.settings.database.transaction(async (connection) => { + const now = await connection.nowMilliseconds() + const heartbeat = now - aliveThreshold - 100 + await connection.run( + `INSERT INTO solid_objects_processes + (id, kind, hostname, host_process_id, metadata, started_at_ms, heartbeat_at_ms, shutdown_state) + VALUES ('crashed', 'effect_worker', 'other-host', 1, '{}', ?, ?, 'running')`, + [heartbeat, heartbeat], + ) + await connection.run( + `INSERT INTO solid_objects_effects + (id, message_id, instance_id, name, arguments, status, max_attempts, available_at_ms, claimed_by) + VALUES ('stuck', ?, ?, 'export', '{}', 'processing', 3, ?, 'crashed')`, + [message.id, message.instance_id, heartbeat], + ) + await connection.run( + `INSERT INTO solid_objects_effect_recoveries + (effect_id, instance_id, recovery_operation, status_operation, recovery_timeout_ms) + VALUES ('stuck', ?, 'recovered', 'inspected', ?)`, + [message.instance_id, recoveryTimeout], + ) + }) + const startedAt = performance.now() + + start(target) + + const recovered = await eventually({ + read: () => Exporter.recoveredAt, + until: (times) => times.length === 1, + timeoutMilliseconds: 500 + promptly, + }) + expect(recovered).toHaveLength(1) + expect(recovered[0]! - startedAt).toBeGreaterThanOrEqual(400) + }) +}) From 632e09ba6fc6ae8929dc52ae2d359bbbedb2164b Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:43:15 -0800 Subject: [PATCH 09/15] feat: wake singleOwner roles after a clock jump A role that waits for a due time uses a Node timer, and Node timers stop while the computer sleeps. Without polling, a reminder due during a laptop sleep of eight hours would run hours late. With singleOwner, a watch now compares the wall clock with the process clock every ten seconds, and wakes every role when they differ by a second or more. The roles then read their due times again. The watch sends no query. --- src/runtime.ts | 12 ++++++ src/wall-clock-watch.ts | 37 +++++++++++++++++ test/single-owner-idle.test.ts | 2 +- test/wall-clock-watch.test.ts | 75 ++++++++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+), 1 deletion(-) create mode 100644 src/wall-clock-watch.ts create mode 100644 test/wall-clock-watch.test.ts diff --git a/src/runtime.ts b/src/runtime.ts index 180a7a7..11e6d5f 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -87,6 +87,7 @@ import type { PayloadEnvelope } from "./browser/index.js" import { MaintenanceScheduler } from "./maintenance-scheduler.js" import { DueWork } from "./due-work.js" import { LocalSignal } from "./local-signal.js" +import { WallClockWatch } from "./wall-clock-watch.js" import type { BroadcastRow, ClaimedTurn, @@ -1991,6 +1992,9 @@ export class SolidObjectsRuntime { ) : []), () => new RedriveScheduler(this), + ...(this.settings.singleOwner + ? [() => new WallClockWatch({ onJump: () => this.wakeUpEveryRole() })] + : []), ...(this.settings.retentionIntervalMilliseconds > 0 ? [ () => @@ -2370,6 +2374,14 @@ export class SolidObjectsRuntime { void this.notifyWhenSelected(role) } + private wakeUpEveryRole(): void { + this.wakeUp("actors") + this.wakeUp("effects") + this.wakeUp("reminders") + this.wakeUp("broadcasts") + this.redriveSignal.notify() + } + private async notifyWhenSelected(role: WakeUpRole): Promise { try { const selected = await this.resolveWakeUp() diff --git a/src/wall-clock-watch.ts b/src/wall-clock-watch.ts new file mode 100644 index 0000000..7b31502 --- /dev/null +++ b/src/wall-clock-watch.ts @@ -0,0 +1,37 @@ +import { waitFor } from "./worker.js" + +export const WALL_CLOCK_WATCH_INTERVAL_MILLISECONDS = 10_000 +export const WALL_CLOCK_JUMP_MILLISECONDS = 1_000 + +export class WallClockWatch { + private stopping = false + + constructor(private readonly options: { onJump: () => void }) {} + + async run(signal: AbortSignal): Promise { + let offset = wallClockOffset() + while (!signal.aborted && !this.stopping) { + await waitFor(WALL_CLOCK_WATCH_INTERVAL_MILLISECONDS, signal) + if (signal.aborted || this.stopping) return + const current = wallClockOffset() + if (Math.abs(current - offset) >= WALL_CLOCK_JUMP_MILLISECONDS) this.options.onJump() + offset = current + } + } + + requestShutdown(): void { + this.stopping = true + } + + stopped(): boolean { + return this.stopping + } + + stop(): void { + this.stopping = true + } +} + +function wallClockOffset(): number { + return Date.now() - performance.now() +} diff --git a/test/single-owner-idle.test.ts b/test/single-owner-idle.test.ts index 45eb1c9..918d28f 100644 --- a/test/single-owner-idle.test.ts +++ b/test/single-owner-idle.test.ts @@ -28,7 +28,7 @@ afterEach(async () => { function useControlledClock(): void { vi.useFakeTimers({ - toFake: ["setTimeout", "clearTimeout", "setImmediate", "clearImmediate", "performance"], + toFake: ["setTimeout", "clearTimeout", "setImmediate", "clearImmediate", "performance", "Date"], }) } diff --git a/test/wall-clock-watch.test.ts b/test/wall-clock-watch.test.ts new file mode 100644 index 0000000..33ec242 --- /dev/null +++ b/test/wall-clock-watch.test.ts @@ -0,0 +1,75 @@ +import { afterEach, describe, expect, it, vi } from "vitest" +import { sqlite } from "../src/database/sqlite.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { WallClockWatch } from "../src/wall-clock-watch.js" +import type { WakeUpRole } from "../src/wake-up.js" + +const eightHours = 8 * 3_600_000 + +let runtime: SolidObjectsRuntime | undefined +let stopRuntime: (() => Promise) | undefined + +afterEach(async () => { + await stopRuntime?.() + stopRuntime = undefined + vi.useRealTimers() + await runtime?.close() + runtime = undefined +}) + +function useControlledClocks(): void { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date", "performance"] }) +} + +describe("WallClockWatch", () => { + it("reports a jump of the wall clock against the process clock, as after a sleep", async () => { + useControlledClocks() + let jumps = 0 + const watch = new WallClockWatch({ onJump: () => (jumps += 1) }) + const controller = new AbortController() + const running = watch.run(controller.signal) + + await vi.advanceTimersByTimeAsync(60_000) + expect(jumps).toBe(0) + + vi.setSystemTime(Date.now() + eightHours) + await vi.advanceTimersByTimeAsync(10_000) + expect(jumps).toBe(1) + + await vi.advanceTimersByTimeAsync(60_000) + expect(jumps).toBe(1) + + vi.setSystemTime(Date.now() - eightHours) + await vi.advanceTimersByTimeAsync(10_000) + expect(jumps).toBe(2) + + controller.abort() + await running + }) + + it("wakes every role of a singleOwner runtime after a jump", async () => { + useControlledClocks() + runtime = createRuntime({ database: sqlite({ path: ":memory:" }), singleOwner: true }) + await runtime.install() + const controller = new AbortController() + const running = runtime.run(controller.signal) + stopRuntime = async () => { + controller.abort() + await running + } + await vi.advanceTimersByTimeAsync(1_000) + const roles: WakeUpRole[] = ["actors", "effects", "reminders", "broadcasts"] + const watches = await Promise.all(roles.map((role) => runtime!.watchWakeUp(role))) + const redrive = runtime.redriveSignal.watch() + + vi.setSystemTime(Date.now() + eightHours) + await vi.advanceTimersByTimeAsync(10_000) + + const notified = Promise.all([ + ...watches.map((watch) => watch.wait({ timeoutMilliseconds: 0 })), + redrive.wait({ timeoutMilliseconds: 0 }), + ]) + await vi.advanceTimersByTimeAsync(1) + expect(await notified).toEqual([true, true, true, true, true]) + }) +}) From 93464bd2bad88b4707347557bea9d105f52eab8a Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:53:33 -0800 Subject: [PATCH 10/15] docs: describe the singleOwner option Explain what the option changes, when the runtime refuses it, and why PostgreSQL, MySQL, and cross-process wake-ups are refused. State the guarantee and its limit in the correctness page: it holds only while one process writes the database, and the start check sees another writer only while that writer's row is fresh. Describe the stale rows of an idle single owner and the exclusive SQLite lock in the operations page. Document SingleOwnerConflict and add a changelog entry with the measured idle statement counts. --- CHANGELOG.md | 14 +++++++ docs/configuration.md | 78 ++++++++++++++++++++++++++++++++++++- docs/correctness.md | 56 ++++++++++++++++++++++++-- docs/errors-and-recovery.md | 4 ++ docs/operations.md | 50 +++++++++++++++++++++--- 5 files changed, 192 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2585a78..fc79730 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ ## Unreleased +- Add the `singleOwner` option for a process that is the only writer of its + SQLite database. With the option, an idle runtime sends no query. Each role + waits for an in-process wake-up or for the next due time that the database + reports, instead of a fixed polling interval. Each role writes its process + row at start, at shutdown, and when it claims work. The periodic dead-process + cleanup is replaced by one cleanup at start, and the actor worker stops + waking for leases when it holds no activation. A watch of the wall clock + wakes every role after the computer sleeps. `run()` refuses to start with the + new `SingleOwnerConflict` error when another process row is alive. The option + refuses a database other than SQLite and a wake-up that crosses processes. + The default is `false`, and the default behavior does not change. In an idle + minute, a SQLite runtime with every role sent 2,466 statements by default and + none with the option. + ## 1.0.6 - 2026-11-08 - Refresh the benchmark evidence against the released v1.0.5 source, with diff --git a/docs/configuration.md b/docs/configuration.md index ae6bea3..ced0047 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -91,7 +91,9 @@ Consecutive empty passes double it up to fixed cadence. A custom wake-up adapter must return `true` for a notification 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. +polling remains the correctness path. The one exception is +[`singleOwner`](#single-owner), which replaces polling with wake-ups and due +times. 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. @@ -99,6 +101,77 @@ 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. +## Single owner + +Set `singleOwner: true` when one process is the only writer of its SQLite +database, for example a daemon that embeds the runtime and holds its database +file. With the option, an idle runtime sends no query. The default is `false`, +and the default behavior does not change. + +```typescript +const runtime = createRuntime({ + database: sqlite({ path: "app.sqlite" }), + singleOwner: true, +}) +``` + +With `singleOwner: true`, the runtime changes these behaviors: + +- Each role waits for an in-process wake-up or for the next due time that the + database reports. A due time is, for example, the time of the next ready + message, effect, broadcast, reminder, or retry, the end of a lease, or the + time when the owner of a claim becomes stale. A role with nothing due waits + up to 24 hours and then reads its due time again. +- A write in this process that adds work or moves it earlier wakes the roles + that must react. A role that waits for a later due time then reads the new + due time. +- The actor worker renews a lease only while it holds a cached activation. + When idle deactivation releases the last activation, the worker stops waking + for leases. +- Each role writes its process row once at start and once at shutdown. A claim + also writes the row of the worker that claims, in the claim transaction, so + that running work always has a fresh owner. An empty pass writes nothing. +- No periodic dead-process cleanup runs, and + `deadProcessCleanupIntervalMilliseconds` has no effect. `run()` releases + stale process rows once, at start. +- A watch compares the wall clock with the process clock every 10 seconds. + Node timers stop while the computer sleeps, so after a sleep, or after a jump + of one second or more, every role reads its due time again. The watch sends + no query. +- When a due time has passed and a claim still finds nothing, the role waits + its polling interval before it tries again, so stuck work cannot spin. + `pollingIntervalMilliseconds` and `idlePollingIntervalMilliseconds` set only + this retry. + +Work that is active runs as before. A turn renews its lease, and an effect or a +broadcast sends its process heartbeat while it runs. Automatic retention still +runs at start and every `retentionIntervalMilliseconds`. Set that interval to +`0` to stop it, and prune with `runtime.retention.prune()` instead. + +`run()` refuses to start with `SingleOwnerConflict` when another process row is +alive by the usual rule: the row is not stopped, and its heartbeat is newer +than `processAliveThresholdMilliseconds`. The error names each row and the time +at which the row becomes stale. The rows of the runtime's own synchronous +callers do not count. + +`createRuntime()` refuses the option in these cases: + +- The database is not SQLite. PostgreSQL and MySQL are servers that other + processes can reach. The runtime cannot know that it is the only writer, and + without polling a write from another process can wait up to 24 hours. +- `wakeUp` is `"postgresql"`, `"redis"`, or an adapter. A notification over a + connection can be lost when the connection fails, and the runtime cannot + prove that a custom adapter delivers every signal. Without polling, a lost + signal can delay work for up to 24 hours. With `"automatic"` or + `"in_process"`, the runtime uses in-process signalling and ignores + `SOLID_OBJECTS_REDIS_URL`. + +Do not use the option with `sharedSqliteWasm`, because each tab runs its own +runtime against the shared database. See +[Correctness and delivery semantics](correctness.md#single-owner) for the +guarantee and its limits, and [Operations](operations.md#single-owner) for +monitoring. + ## Retention and cleanup | Option | Default | Contract | @@ -207,7 +280,8 @@ the files only when no process has the database open. An in-memory database `wakeUp` takes a name or an adapter, as `config.cache_store` does in Rails. `"automatic"` is the default. It prefers `SOLID_OBJECTS_REDIS_URL`, then -PostgreSQL notifications, then polling. `"in_process"` opts out, `"postgresql"` +PostgreSQL notifications, then polling. With `singleOwner`, the runtime runs no +selection and uses in-process signalling. `"in_process"` opts out, `"postgresql"` and `"redis"` force one, and an unknown name throws rather than polls quietly. A requested adapter that the environment cannot provide polls instead and says diff --git a/docs/correctness.md b/docs/correctness.md index 7fa9bf1..ac61d4c 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -36,9 +36,10 @@ - Observable outbox rows are claimed in actor revision order. Subscription sessions and browser clients reject duplicate or stale revisions within an incarnation, while a recreated actor starts a new revision sequence. -- Wake-up notifications happen only after commit and never replace polling. - Workers watch before claiming, so an in-process notification between an empty - claim and the following wait is retained. +- Wake-up notifications happen only after commit and never replace polling, + except with [`singleOwner`](#single-owner). Workers watch before claiming, so + an in-process notification between an empty claim and the following wait is + retained. - Activation passes are bounded. Yielding changes ready-membership polling order only; it neither changes durable message sequence nor makes future work due early. @@ -65,6 +66,55 @@ delivery. A state change on an actor with payloads creates a revision broadcast, even when that actor declares no scalar observables. +## Single owner + +With `singleOwner: true`, an idle runtime sends no query, and it still runs all +work when the work arrives or comes due. This guarantee holds only while one +process writes the database. + +The guarantee rests on these rules: + +- Every write that adds work or moves it earlier happens in this process. After + the commit, the write notifies an in-process signal for each role that must + react. A host that enqueues with + `runtime.enqueueInternalMessageInTransaction()` must call + `runtime.announceInternalMessage()` after its commit, as it must without the + option. Without that call, the message waits for the next wake-up. +- A role watches the signal before it claims, so it cannot lose a notification + that arrives during a pass. +- After an empty pass, a role reads the earliest time at which a claim can take + something. The read uses the same conditions as the claims: the head of each + mailbox, lease ends, pending effects, broadcasts, and reminders, claims whose + owners become stale, and recovery timeouts. The role waits until that time, + or for at most 24 hours. +- A claim writes the row of the claiming worker in the claim transaction. A + worker that was idle for longer than `processAliveThresholdMilliseconds` + therefore never holds work under a stale row, and the alive rule protects + running work as it does without the option. + +Fencing and at-least-once delivery do not change. Claims, activation leases, +the commit fence, retries, dead letters, and effect recovery use the same rows +and the same conditions. + +The guarantee has these limits: + +- One process only. A write from another process, for example a second + runtime, a script, or a command, does not wake this runtime. That work waits + for the next local wake-up or due time, for at most 24 hours. +- The start check reads the process rows once. It sees another writer only + while that writer's row is fresh, and an idle single owner does not refresh + its row. Two processes that start at the same moment can both pass the + check. Only an operating system lock, such as SQLite's exclusive locking + mode, keeps a second process out. +- A due time runs on a Node timer. A long synchronous handler delays the timer, + as it delays polling. Node timers stop while the computer sleeps. A watch + sees the jump of the wall clock within 10 seconds and makes every role read + its due time again, so work that came due during a sleep runs at most about + 10 seconds after the computer wakes. +- When a due time has passed and a claim still finds nothing, for example after + an actor setup failure, the role retries at its polling interval, as it does + without the option. + ## Limitations and non-goals - At-least-once execution means actor code may begin more than once. State and diff --git a/docs/errors-and-recovery.md b/docs/errors-and-recovery.md index 859b954..1524cb4 100644 --- a/docs/errors-and-recovery.md +++ b/docs/errors-and-recovery.md @@ -83,6 +83,10 @@ stable public code and does not classify the operation as failed. lookup found no record. - `ReminderNotPaused` means only a completed reminder remains; schedule it again through its actor instead of resuming it. +- `SingleOwnerConflict` means a `singleOwner` runtime found another live + process row at start. Its `processes` list names each row and the time at + which the row becomes stale. Stop the other process, or wait until that time, + and start again. Realtime protocol validation uses `TypeError` for malformed subscription or browser envelopes. Server delivery failures are isolated to the session and diff --git a/docs/operations.md b/docs/operations.md index 08dc5dd..9db6332 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -7,7 +7,8 @@ and work enqueued with `send` stays ready until some process runs the roles. A direct call or an explicit `sync` needs no running role, because the caller's own path executes it. -Runtime roles use durable polling as the correctness fallback. Consecutive +Runtime roles use durable polling as the correctness fallback, unless the +runtime runs as a [single owner](#single-owner). Consecutive empty passes double each role's wait from `pollingIntervalMilliseconds` to `idlePollingIntervalMilliseconds`, which defaults to one second. Processed work and wake-up notifications reset the role to the fast interval. Actor @@ -149,10 +150,10 @@ state from reconciliation code. The runtime automatically prunes expired message and stopped-process history once at startup and every `retentionIntervalMilliseconds`, which defaults to one hour. Stale process ownership is recovered independently every -`deadProcessCleanupIntervalMilliseconds`, which defaults to one minute. Set an -interval to zero to disable its scheduler. Failed passes emit metadata-only -events and retry with bounded exponential backoff without stopping other -runtime roles. +`deadProcessCleanupIntervalMilliseconds`, which defaults to one minute. A +single owner recovers it once, at start. Set an interval to zero to disable its +scheduler. Failed passes emit metadata-only events and retry with bounded +exponential backoff without stopping other runtime roles. Operators can also use `runtime.retention.preview()` and `runtime.retention.prune()`. Both calls require administration authorization; @@ -228,6 +229,45 @@ Scheduling an existing actor operation for a different time emits `solid_objects.reminder.replaced` after the fenced actor commit. Scheduling it for the same time emits nothing. +## Single owner + +With [`singleOwner: true`](configuration.md#single-owner), an idle runtime sends +no query. Its roles wake for an in-process notification, for a due time that +the database reports, or after the wall clock jumps. Expect these differences +from the default: + +- `runtime.administration.processes()` reports the rows of an idle single owner + as `stale`, because a role writes its row only at start, at shutdown, and + when it claims work. The `runtime` doctor check can report no live roles for + the same reason. Both are normal while the runtime is idle. +- The `wakeUp` doctor check passes with in-process signalling, because no other + process writes the database. +- `cleanup()` can mark the rows of an idle single owner `stopped`. The next + claim of each role writes its row again. An idle role holds no claim, so the + cleanup releases nothing. +- `currentPollingIntervalMilliseconds` and the + `solid_objects.polling.interval_changed` events describe only the retry of + due work that a claim cannot take. + +`run()` refuses to start with `SingleOwnerConflict` when another process row is +alive. A process that stopped without cleanup stays alive until its heartbeat +is older than `processAliveThresholdMilliseconds`. The error gives that time +for each row. Stop the other process, or wait until that time, and start again. + +The start check reads the process rows once. It sees another writer only while +that writer's row is fresh, and an idle single owner does not refresh its row. +For a guarantee that the operating system enforces, take an exclusive lock on +the file before the runtime writes to it: + +```typescript +const database = sqlite({ path: "app.sqlite" }) +await database.connection((connection) => connection.run("PRAGMA locking_mode = EXCLUSIVE")) +``` + +SQLite then holds the lock from the first write until the database closes. A +second process, including the `solid-objects` command for the same file, gets +`database is locked`. + ## Dead letters, retry, and redrive A handler failure that exhausts its attempts becomes a dead letter. From bd2b100c3fe1849912073bcc1686b6f2c296d819 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 20:56:40 -0800 Subject: [PATCH 11/15] fix: back off when due work is due now An actor setup failure puts its message back as ready in the same millisecond, so the next due time was exactly now. The worker then waited one millisecond and claimed the same message again, about fifty times a second. A due time of zero now counts as passed, like a negative one, and the role waits its polling interval, as it does without singleOwner. A future due time still wakes the role one millisecond after it. --- src/single-owner.ts | 2 +- test/single-owner-wait.test.ts | 86 ++++++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 1 deletion(-) create mode 100644 test/single-owner-wait.test.ts diff --git a/src/single-owner.ts b/src/single-owner.ts index 87f619e..9121432 100644 --- a/src/single-owner.ts +++ b/src/single-owner.ts @@ -7,7 +7,7 @@ export function waitUntilDue(options: { pollingIntervalMilliseconds: number }): number { if (options.untilDueMilliseconds === undefined) return MAXIMUM_WAIT_MILLISECONDS - if (options.untilDueMilliseconds < 0) return options.pollingIntervalMilliseconds + if (options.untilDueMilliseconds <= 0) return options.pollingIntervalMilliseconds return Math.min(options.untilDueMilliseconds + 1, MAXIMUM_WAIT_MILLISECONDS) } diff --git a/test/single-owner-wait.test.ts b/test/single-owner-wait.test.ts new file mode 100644 index 0000000..a6b96b3 --- /dev/null +++ b/test/single-owner-wait.test.ts @@ -0,0 +1,86 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { ApplicationWriteForbidden } from "../src/errors.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { MAXIMUM_WAIT_MILLISECONDS, waitUntilDue } from "../src/single-owner.js" + +const pollingInterval = 200 + +class BrokenSetup extends Actor { + static override readonly actorType = "SingleOwnerBrokenSetup" + static activations = 0 + + work(): void {} + + protected override onActivate(): void { + BrokenSetup.activations += 1 + throw new ApplicationWriteForbidden("setup cannot write") + } +} + +let runtime: SolidObjectsRuntime | undefined +let stopRuntime: (() => Promise) | undefined + +afterEach(async () => { + await stopRuntime?.() + stopRuntime = undefined + await runtime?.close() + runtime = undefined + BrokenSetup.activations = 0 +}) + +describe("waitUntilDue", () => { + it("waits the maximum when nothing is due", () => { + expect( + waitUntilDue({ untilDueMilliseconds: undefined, pollingIntervalMilliseconds: 100 }), + ).toBe(MAXIMUM_WAIT_MILLISECONDS) + expect(MAXIMUM_WAIT_MILLISECONDS).toBe(86_400_000) + }) + + it("waits one millisecond past a future due time", () => { + expect(waitUntilDue({ untilDueMilliseconds: 1, pollingIntervalMilliseconds: 100 })).toBe(2) + expect(waitUntilDue({ untilDueMilliseconds: 500, pollingIntervalMilliseconds: 100 })).toBe(501) + }) + + it("bounds a far due time at the maximum", () => { + expect( + waitUntilDue({ + untilDueMilliseconds: 10 * MAXIMUM_WAIT_MILLISECONDS, + pollingIntervalMilliseconds: 100, + }), + ).toBe(MAXIMUM_WAIT_MILLISECONDS) + }) + + it("falls back to the polling interval when a due time is now or has passed", () => { + expect(waitUntilDue({ untilDueMilliseconds: 0, pollingIntervalMilliseconds: 100 })).toBe(100) + expect(waitUntilDue({ untilDueMilliseconds: -5, pollingIntervalMilliseconds: 100 })).toBe(100) + }) +}) + +describe("a singleOwner worker with work that it cannot take", () => { + it("retries a failing actor setup at its polling interval instead of spinning", async () => { + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: true, + pollingIntervalMilliseconds: pollingInterval, + idlePollingIntervalMilliseconds: pollingInterval, + authorizeMessage: () => true, + logger: { debug: () => {}, info: () => {}, warn: () => {}, error: () => {} }, + }) + runtime.register(BrokenSetup) + await runtime.install() + await runtime.ref(BrokenSetup, "broken").send.work() + const controller = new AbortController() + const running = runtime.run(controller.signal) + stopRuntime = async () => { + controller.abort() + await running + } + + await new Promise((resolve) => setTimeout(resolve, 5 * pollingInterval)) + + expect(BrokenSetup.activations).toBeGreaterThanOrEqual(2) + expect(BrokenSetup.activations).toBeLessThanOrEqual(10) + }) +}) From 212a4404699380e4efc76fca538bf32b96534a4e Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 21:26:38 -0800 Subject: [PATCH 12/15] fix: remove the abort listener when waitFor ends waitFor added an abort listener to its signal and removed it only when the signal aborted. A loop that waits with the same signal therefore added a listener on every pass. The redrive scheduler added one each idle second without singleOwner, the cleanup scheduler one each minute, and the singleOwner clock watch one each ten seconds. The listeners grow without bound and trigger listener-limit warnings. waitFor now removes its listener when its timer ends. --- src/worker.ts | 16 +++++----- test/wait-for.test.ts | 69 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 9 deletions(-) create mode 100644 test/wait-for.test.ts diff --git a/src/worker.ts b/src/worker.ts index d85ce1e..20a77fb 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -313,15 +313,13 @@ function activationLease(turn: ClaimedTurn): ActivationLease { export function waitFor(milliseconds: number, signal?: AbortSignal): Promise { if (signal?.aborted) return Promise.resolve() return new Promise((resolve) => { - const timeout = setTimeout(resolve, milliseconds) - signal?.addEventListener( - "abort", - () => { - clearTimeout(timeout) - resolve() - }, - { once: true }, - ) + const finish = () => { + clearTimeout(timeout) + signal?.removeEventListener("abort", finish) + resolve() + } + const timeout = setTimeout(finish, milliseconds) + signal?.addEventListener("abort", finish, { once: true }) }) } diff --git a/test/wait-for.test.ts b/test/wait-for.test.ts new file mode 100644 index 0000000..82feb1e --- /dev/null +++ b/test/wait-for.test.ts @@ -0,0 +1,69 @@ +import { getEventListeners } from "node:events" +import { afterEach, describe, expect, it, vi } from "vitest" +import { sqlite } from "../src/database/sqlite.js" +import { RedriveScheduler } from "../src/redrive.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { WallClockWatch } from "../src/wall-clock-watch.js" +import { waitFor } from "../src/worker.js" + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + vi.useRealTimers() + await runtime?.close() + runtime = undefined +}) + +function abortListeners(signal: AbortSignal): number { + return getEventListeners(signal, "abort").length +} + +describe("waitFor", () => { + it("removes its abort listener when its timer ends", async () => { + const controller = new AbortController() + + await waitFor(1, controller.signal) + + expect(abortListeners(controller.signal)).toBe(0) + }) + + it("still ends early when the signal aborts", async () => { + const controller = new AbortController() + const startedAt = performance.now() + const waiting = waitFor(60_000, controller.signal) + + controller.abort() + await waiting + + expect(performance.now() - startedAt).toBeLessThan(1_000) + expect(abortListeners(controller.signal)).toBe(0) + }) +}) + +describe("a long-running role that waits with waitFor", () => { + it("keeps at most one abort listener in a wall clock watch", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date", "performance"] }) + const controller = new AbortController() + const running = new WallClockWatch({ onJump: () => {} }).run(controller.signal) + + await vi.advanceTimersByTimeAsync(600_000) + + expect(abortListeners(controller.signal)).toBeLessThanOrEqual(1) + controller.abort() + await running + }) + + it("keeps at most one abort listener in an idle redrive scheduler without singleOwner", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date", "performance"] }) + runtime = createRuntime({ database: sqlite({ path: ":memory:" }) }) + await runtime.install() + const controller = new AbortController() + const running = new RedriveScheduler(runtime).run(controller.signal) + + await vi.advanceTimersByTimeAsync(60_000) + + expect(abortListeners(controller.signal)).toBeLessThanOrEqual(1) + controller.abort() + await running + }) +}) From 1ae96269948cca7637a2d2880535c6a859435725 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 21:31:32 -0800 Subject: [PATCH 13/15] fix: retry due work at once before backing off A role that wakes for a due time can claim a few milliseconds early, because a Node timer counts from the cached loop time and the database clock rounds to whole milliseconds. The claim then finds nothing, and the next due time reads as exactly now. Since the previous fix, that case waited a full polling interval, an hour with long intervals, so a reminder test failed in two of seventeen suite runs. A due time of now or earlier now retries after 1 ms and doubles the wait up to the polling interval. A future due time starts the back-off again. Stuck work, such as an actor setup failure, still settles at the polling interval. --- docs/configuration.md | 10 +++++--- docs/correctness.md | 7 ++--- src/broadcast-worker.ts | 4 ++- src/effect-worker.ts | 4 ++- src/reminder-scheduler.ts | 4 ++- src/single-owner.ts | 33 ++++++++++++++++++------ src/worker.ts | 4 ++- test/single-owner-wait.test.ts | 47 ++++++++++++++++++++++++---------- 8 files changed, 80 insertions(+), 33 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index ced0047..be3c30b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -138,10 +138,12 @@ With `singleOwner: true`, the runtime changes these behaviors: Node timers stop while the computer sleeps, so after a sleep, or after a jump of one second or more, every role reads its due time again. The watch sends no query. -- When a due time has passed and a claim still finds nothing, the role waits - its polling interval before it tries again, so stuck work cannot spin. - `pollingIntervalMilliseconds` and `idlePollingIntervalMilliseconds` set only - this retry. +- When a due time is now or has passed and a claim still finds nothing, the + role tries again after 1 ms. A timer can end a few milliseconds before the + database clock reaches the due time, and this retry takes the work at once. + Each further retry doubles the wait, up to the polling interval, so stuck + work cannot spin. `pollingIntervalMilliseconds` and + `idlePollingIntervalMilliseconds` set only this limit. Work that is active runs as before. A turn renews its lease, and an effect or a broadcast sends its process heartbeat while it runs. Automatic retention still diff --git a/docs/correctness.md b/docs/correctness.md index ac61d4c..0db722a 100644 --- a/docs/correctness.md +++ b/docs/correctness.md @@ -111,9 +111,10 @@ The guarantee has these limits: sees the jump of the wall clock within 10 seconds and makes every role read its due time again, so work that came due during a sleep runs at most about 10 seconds after the computer wakes. -- When a due time has passed and a claim still finds nothing, for example after - an actor setup failure, the role retries at its polling interval, as it does - without the option. +- When a due time is now or has passed and a claim still finds nothing, for + example after an actor setup failure, the role retries after 1 ms and then + doubles the wait up to its polling interval. Stuck work is then retried at + the polling interval, as it is without the option. ## Limitations and non-goals diff --git a/src/broadcast-worker.ts b/src/broadcast-worker.ts index 4b926ff..54748a0 100644 --- a/src/broadcast-worker.ts +++ b/src/broadcast-worker.ts @@ -2,7 +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 { idleWaitMilliseconds } from "./single-owner.js" +import { DueWait, idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class BroadcastWorker { @@ -10,6 +10,7 @@ export class BroadcastWorker { private registered = false private stopping = false private readonly pollingBackoff: PollingBackoff + private readonly dueWait = new DueWait() constructor(private readonly runtime: SolidObjectsRuntime) { this.pollingBackoff = new PollingBackoff({ @@ -68,6 +69,7 @@ export class BroadcastWorker { timeoutMilliseconds: await idleWaitMilliseconds({ settings: this.runtime.settings, pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + dueWait: this.dueWait, untilDue: () => this.runtime.dueWork.untilBroadcastWork(), }), signal, diff --git a/src/effect-worker.ts b/src/effect-worker.ts index 4d994f9..7cc5186 100644 --- a/src/effect-worker.ts +++ b/src/effect-worker.ts @@ -2,7 +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 { idleWaitMilliseconds } from "./single-owner.js" +import { DueWait, idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class EffectWorker { @@ -10,6 +10,7 @@ export class EffectWorker { private registered = false private stopping = false private readonly pollingBackoff: PollingBackoff + private readonly dueWait = new DueWait() constructor(private readonly runtime: SolidObjectsRuntime) { this.pollingBackoff = new PollingBackoff({ @@ -68,6 +69,7 @@ export class EffectWorker { timeoutMilliseconds: await idleWaitMilliseconds({ settings: this.runtime.settings, pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + dueWait: this.dueWait, untilDue: () => this.runtime.dueWork.untilEffectWork(), }), signal, diff --git a/src/reminder-scheduler.ts b/src/reminder-scheduler.ts index cae3ea8..a0ea20f 100644 --- a/src/reminder-scheduler.ts +++ b/src/reminder-scheduler.ts @@ -2,7 +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 { idleWaitMilliseconds } from "./single-owner.js" +import { DueWait, idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class ReminderScheduler { @@ -10,6 +10,7 @@ export class ReminderScheduler { private registered = false private stopping = false private readonly pollingBackoff: PollingBackoff + private readonly dueWait = new DueWait() constructor(private readonly runtime: SolidObjectsRuntime) { this.pollingBackoff = new PollingBackoff({ @@ -89,6 +90,7 @@ export class ReminderScheduler { timeoutMilliseconds: await idleWaitMilliseconds({ settings: this.runtime.settings, pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + dueWait: this.dueWait, untilDue: () => this.runtime.dueWork.untilReminderWork(), }), signal, diff --git a/src/single-owner.ts b/src/single-owner.ts index 9121432..66f6469 100644 --- a/src/single-owner.ts +++ b/src/single-owner.ts @@ -2,22 +2,39 @@ import type { RuntimeSettings } from "./configuration.js" export const MAXIMUM_WAIT_MILLISECONDS = 86_400_000 -export function waitUntilDue(options: { - untilDueMilliseconds: number | undefined - pollingIntervalMilliseconds: number -}): number { - if (options.untilDueMilliseconds === undefined) return MAXIMUM_WAIT_MILLISECONDS - if (options.untilDueMilliseconds <= 0) return options.pollingIntervalMilliseconds - return Math.min(options.untilDueMilliseconds + 1, MAXIMUM_WAIT_MILLISECONDS) +export class DueWait { + private retryMilliseconds = 1 + + milliseconds(options: { + untilDueMilliseconds: number | undefined + pollingIntervalMilliseconds: number + }): number { + if (options.untilDueMilliseconds === undefined) return this.reset(MAXIMUM_WAIT_MILLISECONDS) + if (options.untilDueMilliseconds > 0) { + return this.reset(Math.min(options.untilDueMilliseconds + 1, MAXIMUM_WAIT_MILLISECONDS)) + } + const retry = Math.min(this.retryMilliseconds, options.pollingIntervalMilliseconds) + this.retryMilliseconds = Math.min( + this.retryMilliseconds * 2, + options.pollingIntervalMilliseconds, + ) + return retry + } + + private reset(milliseconds: number): number { + this.retryMilliseconds = 1 + return milliseconds + } } export async function idleWaitMilliseconds(options: { settings: RuntimeSettings pollingIntervalMilliseconds: number + dueWait: DueWait untilDue: () => Promise }): Promise { if (!options.settings.singleOwner) return options.pollingIntervalMilliseconds - return waitUntilDue({ + return options.dueWait.milliseconds({ untilDueMilliseconds: await options.untilDue(), pollingIntervalMilliseconds: options.pollingIntervalMilliseconds, }) diff --git a/src/worker.ts b/src/worker.ts index 20a77fb..0685c9d 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -9,7 +9,7 @@ import { import type { ActivationLease, ClaimedTurn } from "./records.js" import type { SolidObjectsRuntime } from "./runtime.js" import { PollingBackoff } from "./polling-backoff.js" -import { idleWaitMilliseconds } from "./single-owner.js" +import { DueWait, idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" interface CachedActivation { @@ -25,6 +25,7 @@ export class Worker { private stopping = false private readonly activations = new Map() private readonly pollingBackoff: PollingBackoff + private readonly dueWait = new DueWait() constructor( private readonly runtime: SolidObjectsRuntime, @@ -197,6 +198,7 @@ export class Worker { const databaseWait = await idleWaitMilliseconds({ settings: this.runtime.settings, pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + dueWait: this.dueWait, untilDue: () => this.runtime.dueWork.untilActorWork({ processId: this.processId }), }) if (!this.runtime.settings.singleOwner) return databaseWait diff --git a/test/single-owner-wait.test.ts b/test/single-owner-wait.test.ts index a6b96b3..1b96162 100644 --- a/test/single-owner-wait.test.ts +++ b/test/single-owner-wait.test.ts @@ -3,7 +3,7 @@ import { Actor } from "../src/actor.js" import { sqlite } from "../src/database/sqlite.js" import { ApplicationWriteForbidden } from "../src/errors.js" import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" -import { MAXIMUM_WAIT_MILLISECONDS, waitUntilDue } from "../src/single-owner.js" +import { DueWait, MAXIMUM_WAIT_MILLISECONDS } from "../src/single-owner.js" const pollingInterval = 200 @@ -30,31 +30,48 @@ afterEach(async () => { BrokenSetup.activations = 0 }) -describe("waitUntilDue", () => { +describe("DueWait", () => { it("waits the maximum when nothing is due", () => { + const wait = new DueWait() + expect( - waitUntilDue({ untilDueMilliseconds: undefined, pollingIntervalMilliseconds: 100 }), + wait.milliseconds({ untilDueMilliseconds: undefined, pollingIntervalMilliseconds: 100 }), ).toBe(MAXIMUM_WAIT_MILLISECONDS) expect(MAXIMUM_WAIT_MILLISECONDS).toBe(86_400_000) }) - it("waits one millisecond past a future due time", () => { - expect(waitUntilDue({ untilDueMilliseconds: 1, pollingIntervalMilliseconds: 100 })).toBe(2) - expect(waitUntilDue({ untilDueMilliseconds: 500, pollingIntervalMilliseconds: 100 })).toBe(501) - }) + it("waits one millisecond past a future due time, bounded at the maximum", () => { + const wait = new DueWait() - it("bounds a far due time at the maximum", () => { + expect(wait.milliseconds({ untilDueMilliseconds: 500, pollingIntervalMilliseconds: 100 })).toBe( + 501, + ) expect( - waitUntilDue({ + wait.milliseconds({ untilDueMilliseconds: 10 * MAXIMUM_WAIT_MILLISECONDS, pollingIntervalMilliseconds: 100, }), ).toBe(MAXIMUM_WAIT_MILLISECONDS) }) - it("falls back to the polling interval when a due time is now or has passed", () => { - expect(waitUntilDue({ untilDueMilliseconds: 0, pollingIntervalMilliseconds: 100 })).toBe(100) - expect(waitUntilDue({ untilDueMilliseconds: -5, pollingIntervalMilliseconds: 100 })).toBe(100) + it("retries a due time that is now at once, then backs off to the polling interval", () => { + const wait = new DueWait() + const retries = [0, -5, 0, -1, 0, 0, 0, 0, 0].map((untilDueMilliseconds) => + wait.milliseconds({ untilDueMilliseconds, pollingIntervalMilliseconds: 100 }), + ) + + expect(retries).toEqual([1, 2, 4, 8, 16, 32, 64, 100, 100]) + }) + + it("starts the back-off again after a future due time", () => { + const wait = new DueWait() + for (let pass = 0; pass < 5; pass += 1) { + wait.milliseconds({ untilDueMilliseconds: 0, pollingIntervalMilliseconds: 100 }) + } + + wait.milliseconds({ untilDueMilliseconds: 50, pollingIntervalMilliseconds: 100 }) + + expect(wait.milliseconds({ untilDueMilliseconds: 0, pollingIntervalMilliseconds: 100 })).toBe(1) }) }) @@ -78,9 +95,11 @@ describe("a singleOwner worker with work that it cannot take", () => { await running } + await new Promise((resolve) => setTimeout(resolve, 5 * pollingInterval)) + const afterBackOff = BrokenSetup.activations await new Promise((resolve) => setTimeout(resolve, 5 * pollingInterval)) - expect(BrokenSetup.activations).toBeGreaterThanOrEqual(2) - expect(BrokenSetup.activations).toBeLessThanOrEqual(10) + expect(BrokenSetup.activations - afterBackOff).toBeGreaterThanOrEqual(2) + expect(BrokenSetup.activations - afterBackOff).toBeLessThanOrEqual(8) }) }) From febaab27d7f6b4be972f4aa0f6fdc96486eecbb3 Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 21:41:02 -0800 Subject: [PATCH 14/15] fix: wake actor workers after a release unblocks With singleOwner, a synchronous caller can hold an actor while a new message for that actor arrives. The background worker wakes, finds the actor held, and waits until the caller's lease ends. When the caller then releases the activation, nothing woke that worker, so the message waited about 30 seconds with the default lease. With singleOwner, a release that frees the activation now reports whether ready messages wait for that actor, and the runtime then wakes the actor workers. Without the option, the release sends no extra query. --- src/repository.ts | 14 ++- src/runtime.ts | 3 +- test/single-owner-release.test.ts | 143 ++++++++++++++++++++++++++++++ 3 files changed, 156 insertions(+), 4 deletions(-) create mode 100644 test/single-owner-release.test.ts diff --git a/src/repository.ts b/src/repository.ts index 98e0a68..475c9c5 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -681,10 +681,10 @@ export class Repository { }) } - async releaseActivation(activation: ActivationLease): Promise { - await this.settings.database.transaction(async (connection) => { + async releaseActivation(activation: ActivationLease): Promise<{ readyMessagesWaiting: boolean }> { + return this.settings.database.transaction(async (connection) => { const now = await connection.nowMilliseconds() - await connection.run( + const released = await connection.run( `UPDATE ${this.table("instances")} SET activation_owner_id = NULL, activation_token = NULL, activation_expires_at_ms = NULL, updated_at_ms = ? @@ -703,6 +703,14 @@ export class Repository { activation.instanceId, ], ) + if (!this.settings.singleOwner || released.changes !== 1) { + return { readyMessagesWaiting: false } + } + const waiting = await connection.get<{ present: number | bigint }>( + `SELECT 1 AS present FROM ${this.table("ready_messages")} WHERE instance_id = ? LIMIT 1`, + [activation.instanceId], + ) + return { readyMessagesWaiting: waiting !== undefined } }) } diff --git a/src/runtime.ts b/src/runtime.ts index 11e6d5f..f75958f 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -1527,12 +1527,13 @@ export class SolidObjectsRuntime { }) } finally { actor?.discardIntents() - await this.repository.releaseActivation({ + const released = await this.repository.releaseActivation({ instanceId: turn.instance.id, processId: turn.processId, activationToken: turn.activationToken, activationGeneration: turn.activationGeneration, }) + if (released.readyMessagesWaiting) this.wakeUp("actors") } } diff --git a/test/single-owner-release.test.ts b/test/single-owner-release.test.ts new file mode 100644 index 0000000..f66159f --- /dev/null +++ b/test/single-owner-release.test.ts @@ -0,0 +1,143 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import { CountingDatabase } from "./support/counting-database.js" +import { deferred } from "./support/fenced-commit.js" + +const anHour = 3_600_000 +const promptly = 700 + +class SharedDesk extends Actor { + static override readonly actorType = "SingleOwnerSharedDesk" + static entered = deferred() + static release = deferred() + + visits = 0 + + async hold(): Promise { + SharedDesk.entered.resolve() + await SharedDesk.release.promise + this.visits += 1 + return this.visits + } + + visit(): number { + this.visits += 1 + return this.visits + } +} + +let runtime: SolidObjectsRuntime | undefined +let stopRuntime: (() => Promise) | undefined + +afterEach(async () => { + SharedDesk.release.resolve() + await stopRuntime?.() + stopRuntime = undefined + await runtime?.close() + runtime = undefined + SharedDesk.entered = deferred() + SharedDesk.release = deferred() +}) + +function settled(): Promise { + return new Promise((resolve) => setTimeout(resolve, 50)) +} + +async function eventually(options: { + read: () => Value | Promise + until: (value: Value) => boolean + timeoutMilliseconds: number +}): Promise { + const deadline = performance.now() + options.timeoutMilliseconds + let value = await options.read() + while (!options.until(value) && performance.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 5)) + value = await options.read() + } + return value +} + +describe("an activation release with singleOwner", () => { + it("wakes the actor workers for a message that waited behind a synchronous caller", async () => { + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + singleOwner: true, + maxMessagesPerActivationPass: 1, + pollingIntervalMilliseconds: anHour, + idlePollingIntervalMilliseconds: anHour, + authorizeMessage: () => true, + authorizeQuery: () => true, + }) + runtime.register(SharedDesk) + await runtime.install() + const desk = runtime.ref(SharedDesk, "shared") + const holding = desk.hold() + await SharedDesk.entered.promise + const controller = new AbortController() + const running = runtime.run(controller.signal) + stopRuntime = async () => { + controller.abort() + await running + } + await settled() + const waiting = await desk.send.visit() + await settled() + + SharedDesk.release.resolve() + expect(await holding).toBe(1) + + expect( + await eventually({ + read: () => waiting.status(), + until: (status) => status === "completed", + timeoutMilliseconds: promptly, + }), + ).toBe("completed") + }) +}) + +describe("Repository.releaseActivation", () => { + it.each([ + { singleOwner: true, readyMessagesWaiting: true, readyQueries: 1 }, + { singleOwner: false, readyMessagesWaiting: false, readyQueries: 0 }, + ])("reports a waiting message only when singleOwner is $singleOwner", async (options) => { + const database = new CountingDatabase(sqlite({ path: ":memory:" })) + runtime = createRuntime({ database, singleOwner: options.singleOwner }) + await runtime.install() + await runtime.repository.registerProcess("holder", "worker") + for (const operation of ["first", "second"]) { + await runtime.repository.enqueue({ + actorType: "Probe", + actorId: "busy", + operation, + deliveryMode: "async", + arguments: {}, + }) + } + const turn = await runtime.repository.claim("holder") + await runtime.repository.complete(turn!, { + state: {}, + stateVersion: 1, + result: null, + intents: { effects: [], reminders: [], outboundMessages: [], commitActions: [] }, + executeCommitAction: async () => {}, + }) + const position = database.statements.length + + const released = await runtime.repository.releaseActivation({ + instanceId: turn!.instance.id, + processId: turn!.processId, + activationToken: turn!.activationToken, + activationGeneration: turn!.activationGeneration, + }) + + expect(released).toEqual({ readyMessagesWaiting: options.readyMessagesWaiting }) + expect( + database + .statementsSince(position) + .filter((statement) => statement.includes("FROM solid_objects_ready_messages")), + ).toHaveLength(options.readyQueries) + }) +}) From 7e11a9999ec7fcc2ebd5ffa530ba21da59cd758a Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Sun, 8 Nov 2026 21:41:43 -0800 Subject: [PATCH 15/15] docs: note the abort-listener leak fix The leak also affected runtimes without singleOwner, where the redrive scheduler added a listener each idle second, so the fix belongs in the release notes. --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index fc79730..fb15d0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ The default is `false`, and the default behavior does not change. In an idle minute, a SQLite runtime with every role sent 2,466 statements by default and none with the option. +- Fix an abort-listener leak. A role that waited with the shared shutdown + signal added a listener on every pass and removed it only at shutdown. The + redrive scheduler added one each idle second, which grew memory and printed + listener-limit warnings in a long-running process. ## 1.0.6 - 2026-11-08