diff --git a/CHANGELOG.md b/CHANGELOG.md index 2585a78..fb15d0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,24 @@ ## 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. +- 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 - Refresh the benchmark evidence against the released v1.0.5 source, with 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/docs/configuration.md b/docs/configuration.md index f2d3fbd..be3c30b 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 @@ -90,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. @@ -98,6 +101,79 @@ 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 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 +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 | @@ -206,7 +282,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..0db722a 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,56 @@ 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 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 - 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. diff --git a/src/broadcast-worker.ts b/src/broadcast-worker.ts index 14b999a..54748a0 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 { DueWait, idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class BroadcastWorker { @@ -9,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({ @@ -64,7 +66,12 @@ export class BroadcastWorker { continue } const notified = await wakeUp.wait({ - timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + timeoutMilliseconds: await idleWaitMilliseconds({ + settings: this.runtime.settings, + pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + dueWait: this.dueWait, + untilDue: () => this.runtime.dueWork.untilBroadcastWork(), + }), signal, }) if (notified === false) this.pollingBackoff.recordIdle() @@ -91,6 +98,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 a369d2e..7077dfe 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") } @@ -350,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/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/due-work.ts b/src/due-work.ts new file mode 100644 index 0000000..6da8024 --- /dev/null +++ b/src/due-work.ts @@ -0,0 +1,134 @@ +import type { RuntimeSettings } from "./configuration.js" + +interface DueRow { + due_at_ms: number | bigint | null +} + +export class DueWork { + constructor(private readonly settings: RuntimeSettings) {} + + 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 + 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: [options.processId], + })) + } + + 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/src/effect-worker.ts b/src/effect-worker.ts index 6523b47..7cc5186 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 { DueWait, idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class EffectWorker { @@ -9,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({ @@ -64,7 +66,12 @@ export class EffectWorker { continue } const notified = await wakeUp.wait({ - timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + timeoutMilliseconds: await idleWaitMilliseconds({ + settings: this.runtime.settings, + pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + dueWait: this.dueWait, + untilDue: () => this.runtime.dueWork.untilEffectWork(), + }), signal, }) if (notified === false) this.pollingBackoff.recordIdle() @@ -91,6 +98,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/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/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..a0ea20f 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 { DueWait, idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" export class ReminderScheduler { @@ -9,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({ @@ -85,7 +87,12 @@ export class ReminderScheduler { continue } const notified = await wakeUp.wait({ - timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + timeoutMilliseconds: await idleWaitMilliseconds({ + settings: this.runtime.settings, + pollingIntervalMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + dueWait: this.dueWait, + untilDue: () => this.runtime.dueWork.untilReminderWork(), + }), signal, }) if (notified === false) this.pollingBackoff.recordIdle() @@ -112,6 +119,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/repository.ts b/src/repository.ts index c1db1b7..475c9c5 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() @@ -149,6 +168,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() @@ -534,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, @@ -648,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 = ? @@ -670,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 } }) } @@ -1594,6 +1635,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, @@ -1733,6 +1775,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, @@ -1966,6 +2014,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/src/runtime.ts b/src/runtime.ts index 7882369..f75958f 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, @@ -45,6 +46,7 @@ import { NonRetryableError, QueryMutatedState, Rejected, + SingleOwnerConflict, SyncTimeout, SyncEnqueueTimeout, SyncInsideTransaction, @@ -83,6 +85,9 @@ 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 { WallClockWatch } from "./wall-clock-watch.js" import type { BroadcastRow, ClaimedTurn, @@ -132,7 +137,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" @@ -197,6 +202,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 +232,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 +335,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 @@ -339,6 +347,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) => @@ -1116,6 +1125,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) { @@ -1454,6 +1484,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") @@ -1496,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") } } @@ -1961,6 +1993,9 @@ export class SolidObjectsRuntime { ) : []), () => new RedriveScheduler(this), + ...(this.settings.singleOwner + ? [() => new WallClockWatch({ onJump: () => this.wakeUpEveryRole() })] + : []), ...(this.settings.retentionIntervalMilliseconds > 0 ? [ () => @@ -1972,7 +2007,7 @@ export class SolidObjectsRuntime { }), ] : []), - ...(this.settings.deadProcessCleanupIntervalMilliseconds > 0 + ...(deadProcessCleanupScheduled(this.settings) ? [ () => new MaintenanceScheduler({ @@ -2310,6 +2345,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, @@ -2336,6 +2375,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/single-owner.ts b/src/single-owner.ts new file mode 100644 index 0000000..66f6469 --- /dev/null +++ b/src/single-owner.ts @@ -0,0 +1,41 @@ +import type { RuntimeSettings } from "./configuration.js" + +export const MAXIMUM_WAIT_MILLISECONDS = 86_400_000 + +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 options.dueWait.milliseconds({ + untilDueMilliseconds: await options.untilDue(), + pollingIntervalMilliseconds: options.pollingIntervalMilliseconds, + }) +} 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 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/src/worker.ts b/src/worker.ts index ab8f865..0685c9d 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 { DueWait, idleWaitMilliseconds } from "./single-owner.js" import { yieldToEventLoop } from "./yield-to-event-loop.js" interface CachedActivation { @@ -24,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, @@ -154,7 +156,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 +194,29 @@ export class Worker { } } + private async idleWaitMilliseconds(): Promise { + 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 + 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 } @@ -291,15 +315,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/due-work.test.ts b/test/due-work.test.ts new file mode 100644 index 0000000..e5176dd --- /dev/null +++ b/test/due-work.test.ts @@ -0,0 +1,382 @@ +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({ processId: "worker" })).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({ processId: "worker" }), 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({ processId: "worker" }), 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({ processId: "worker" })).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({ 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 () => { + 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({ processId: "worker" }), 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) + }) + }) +}) 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-configuration.test.ts b/test/single-owner-configuration.test.ts new file mode 100644 index 0000000..c135743 --- /dev/null +++ b/test/single-owner-configuration.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it } from "vitest" +import { buildSettings, type SolidObjectsConfiguration } from "../src/configuration.js" +import type { Database } from "../src/database/types.js" +import { sqlite } from "../src/database/sqlite.js" +import { InProcessWakeUpAdapter } from "../src/wake-up.js" + +function settingsFor(configuration: Partial) { + 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-idle.test.ts b/test/single-owner-idle.test.ts new file mode 100644 index 0000000..918d28f --- /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", "Date"], + }) +} + +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/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) + }) +}) 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) + }) +}) diff --git a/test/single-owner-start.test.ts b/test/single-owner-start.test.ts new file mode 100644 index 0000000..8848629 --- /dev/null +++ b/test/single-owner-start.test.ts @@ -0,0 +1,185 @@ +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() + 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) + + expect( + await waitForStatus({ + status: () => message.status(), + expected: "completed", + timeoutMilliseconds: 2_000, + }), + ).toBe("completed") + expect(await shutdownStateOf("crashed")).toBe("stopped") + }) +}) diff --git a/test/single-owner-wait.test.ts b/test/single-owner-wait.test.ts new file mode 100644 index 0000000..1b96162 --- /dev/null +++ b/test/single-owner-wait.test.ts @@ -0,0 +1,105 @@ +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 { DueWait, MAXIMUM_WAIT_MILLISECONDS } 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("DueWait", () => { + it("waits the maximum when nothing is due", () => { + const wait = new DueWait() + + expect( + 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, bounded at the maximum", () => { + const wait = new DueWait() + + expect(wait.milliseconds({ untilDueMilliseconds: 500, pollingIntervalMilliseconds: 100 })).toBe( + 501, + ) + expect( + wait.milliseconds({ + untilDueMilliseconds: 10 * MAXIMUM_WAIT_MILLISECONDS, + pollingIntervalMilliseconds: 100, + }), + ).toBe(MAXIMUM_WAIT_MILLISECONDS) + }) + + 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) + }) +}) + +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)) + const afterBackOff = BrokenSetup.activations + await new Promise((resolve) => setTimeout(resolve, 5 * pollingInterval)) + + expect(BrokenSetup.activations - afterBackOff).toBeGreaterThanOrEqual(2) + expect(BrokenSetup.activations - afterBackOff).toBeLessThanOrEqual(8) + }) +}) 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", + }) + }) +}) 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) + }) +}) 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() + }, + } + } +} 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 + }) +}) 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]) + }) +})