diff --git a/CHANGELOG.md b/CHANGELOG.md index 99a86cd..c218d80 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,24 @@ # Changelog -## Unreleased +## 0.13.1 - 2026-08-16 + +- Back idle actor, effect, reminder, and broadcast polling off exponentially + from the configured fast interval to a new one-second idle ceiling. Any + processed work or wake-up resets the role immediately, and actor polling + remains capped by the lease-renewal interval. +- Expose each role's current polling interval and emit + `solid_objects.polling.interval_changed` instrumentation for every idle, + work, and wake-up transition. +- Warn once when live processes share the database without a configured + cross-process wake-up adapter. +- Preserve older custom wake-up adapters that return `void`; return `true` for + notifications and `false` for timeouts from the built-in PostgreSQL, Redis, + and in-process adapters so adaptive polling can distinguish them. +- Add a reproducible four-role SQLite idle benchmark. +- **Behavior change:** `pollingIntervalMilliseconds` is now the fast interval + after activity, not a constant idle cadence. Existing explicit values back + off to `idlePollingIntervalMilliseconds`, which defaults to `1_000`. Set + both options to the same value to preserve a fixed cadence. ## 0.13.0 - 2026-08-16 diff --git a/README.md b/README.md index 8c43bb4..8adebe9 100644 --- a/README.md +++ b/README.md @@ -57,11 +57,11 @@ processes submit them concurrently. ## Run it now with SQLite -Node.js 24.15 or newer is required. The `0.13.0` release includes a +Node.js 24.15 or newer is required. The `0.13.1` release includes a packaged quickstart: ```bash -npm exec --yes --package=solid-objects@0.13.0 -- solid-objects quickstart +npm exec --yes --package=solid-objects@0.13.1 -- solid-objects quickstart ``` The command needs no repository checkout, database server, Redis, container, or @@ -157,6 +157,12 @@ Redis is optional wake-up infrastructure. It can reduce notification latency for a multi-process MySQL deployment, but the relational database remains the durable source of truth and polling remains the recovery path. +Idle roles back off from the configured 100 ms fast polling interval to one +second. Processed work and wake-up notifications reset that interval +immediately. The default wake-up reaches only the current Node process; use the +PostgreSQL or optional Redis adapter when separate processes need low-latency +delivery. The runtime warns once when it sees that topology without an adapter. + ## Good and poor fits | Good fit | Poor fit | diff --git a/benchmarks/idle.ts b/benchmarks/idle.ts new file mode 100644 index 0000000..8c3a034 --- /dev/null +++ b/benchmarks/idle.ts @@ -0,0 +1,175 @@ +import { readFile } from "node:fs/promises" +import { cpus, platform, release } from "node:os" +import { createRuntime } from "solid-objects" +import { sqlite } from "solid-objects/database/sqlite" +import type { WakeUpAdapter, WakeUpRole, WakeUpWatch } from "../src/wake-up.ts" + +const roles = ["actors", "effects", "reminders", "broadcasts"] as const +const intervals = option("intervals", "20,100,500") + .split(",") + .map((value) => positiveNumber(value, "intervals")) +const warmupMilliseconds = positiveNumber(option("warmup", "3000"), "warmup") +const durationMilliseconds = positiveNumber(option("duration", "10000"), "duration") + +async function main(): Promise { + const packageMetadata = JSON.parse( + await readFile(new URL("../package.json", import.meta.url), "utf8"), + ) as { version: string } + const results = [] + const databaseVersion = await readDatabaseVersion() + + for (const pollingIntervalMilliseconds of intervals) { + results.push(await measure(pollingIntervalMilliseconds)) + } + + process.stdout.write( + `${JSON.stringify( + { + measuredAt: new Date().toISOString(), + packageVersion: packageMetadata.version, + runtime: { + node: process.version, + platform: `${platform()} ${release()}`, + cpu: cpus()[0]?.model ?? "unknown", + logicalCpus: cpus().length, + }, + database: { adapter: "sqlite", version: databaseVersion, path: ":memory:" }, + methodology: { + roles, + warmupMilliseconds, + durationMilliseconds, + cpuPercent: "process user plus system CPU time divided by wall time", + }, + results, + }, + null, + 2, + )}\n`, + ) +} + +async function readDatabaseVersion(): Promise { + const database = sqlite({ path: ":memory:" }) + try { + return await database.connection(async (connection) => { + const row = await connection.get<{ version: string }>("SELECT sqlite_version() AS version") + return row?.version ?? "unknown" + }) + } finally { + await database.close() + } +} + +async function measure(pollingIntervalMilliseconds: number) { + const wakeUp = new CountingWakeUpAdapter() + const runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds, + workerCount: 1, + effectWorkerCount: 1, + reminderSchedulerCount: 1, + broadcastWorkerCount: 1, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + authorizeSubscription: () => true, + broadcast: async () => {}, + wakeUp, + }) + await runtime.install() + const controller = new AbortController() + const running = [ + runtime.worker().run(controller.signal), + runtime.effectWorker().run(controller.signal), + runtime.reminderScheduler().run(controller.signal), + runtime.broadcastWorker().run(controller.signal), + ] + + try { + await wait(warmupMilliseconds) + wakeUp.resetCounts() + const cpuStartedAt = process.cpuUsage() + const wallStartedAt = performance.now() + await wait(durationMilliseconds) + const elapsedMilliseconds = performance.now() - wallStartedAt + const cpuUsage = process.cpuUsage(cpuStartedAt) + const polls = wakeUp.pollCounts() + const totalPolls = Object.values(polls).reduce((total, count) => total + count, 0) + + return { + pollingIntervalMilliseconds, + idlePollingIntervalMilliseconds: runtime.settings.idlePollingIntervalMilliseconds, + polls, + pollsPerSecond: round((totalPolls * 1_000) / elapsedMilliseconds), + idleCpuPercent: round( + ((cpuUsage.user + cpuUsage.system) / 1_000 / elapsedMilliseconds) * 100, + ), + } + } finally { + controller.abort() + await Promise.all(running) + await runtime.close() + } +} + +class CountingWakeUpAdapter implements WakeUpAdapter { + private readonly counts = new Map() + + watch(role: WakeUpRole): WakeUpWatch { + return { + wait: async ({ timeoutMilliseconds, signal }) => { + this.counts.set(role, (this.counts.get(role) ?? 0) + 1) + return new Promise((resolve) => { + let settled = false + const finish = () => { + if (settled) return + settled = true + clearTimeout(timeout) + signal?.removeEventListener("abort", finish) + resolve(false) + } + const timeout = setTimeout(finish, timeoutMilliseconds) + signal?.addEventListener("abort", finish, { once: true }) + if (signal?.aborted) finish() + }) + }, + } + } + + notify(_role: WakeUpRole): void {} + + close(): void {} + + resetCounts(): void { + this.counts.clear() + } + + pollCounts(): Record { + return Object.fromEntries(roles.map((role) => [role, this.counts.get(role) ?? 0])) as Record< + WakeUpRole, + number + > + } +} + +function option(name: string, fallback: string): string { + const prefix = `--${name}=` + return ( + process.argv.find((argument) => argument.startsWith(prefix))?.slice(prefix.length) ?? fallback + ) +} + +function positiveNumber(value: string, name: string): number { + const number = Number(value) + if (!Number.isFinite(number) || number <= 0) throw new TypeError(`${name} must be positive`) + return number +} + +function wait(milliseconds: number): Promise { + return new Promise((resolve) => setTimeout(resolve, milliseconds)) +} + +function round(value: number): number { + return Math.round(value * 1_000) / 1_000 +} + +await main() diff --git a/docs/api.md b/docs/api.md index e5bd92f..526317b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -179,14 +179,17 @@ Factories should create fresh mutable state and `stop()` should be idempotent. exported for test runners and hosts that intentionally operate roles outside `runtime.run()`. Runtime factory methods create the same classes. Each provides `runOnce()`, bounded `runUntilIdle()`, `run(signal)`, `requestShutdown()`, -`stopped()`, and `stop()`. Manual roles still register process ownership and -must be stopped. Prefer `runtime.run()` in production and `runtime.testing` in -tests. +`stopped()`, `stop()`, and the inspectable +`currentPollingIntervalMilliseconds`. Manual roles still register process +ownership and must be stopped. Prefer `runtime.run()` in production and +`runtime.testing` in tests. `InProcessWakeUpAdapter`, `WakeUpAdapter`, `WakeUpRole`, `WakeUpWatch`, and `WakeUpWaitOptions` define the notification extension. A watch must be obtained before checking durable state so a notification cannot fall between claim and -wait. +wait. `WakeUpWatch.wait()` returns `true` for a notification and `false` for a +timeout or cancellation. A legacy `void` result remains accepted and preserves +the fast polling cadence. ### Errors diff --git a/docs/architecture.md b/docs/architecture.md index 1352fbe..a9c8af6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -25,7 +25,9 @@ Each role takes a generation watch before checking for work. A post-commit wake-up therefore cannot fall into the gap between an empty claim and the worker's wait. The default adapter broadcasts within one process; polling remains active as the durable fallback and custom adapters can bridge process -boundaries. +boundaries. Empty passes double the role's wait up to the configured idle +ceiling. Work or a notification resets it to the fast interval, and an actor +worker's ceiling never exceeds its lease-renewal interval. Each runtime role occupies a supervised factory slot. An unexpected promise resolution or rejection cleans up that instance, waits with capped exponential @@ -63,7 +65,8 @@ per runtime listens on role-specific channels before the worker checks durable state, which closes the listener-startup race without holding a polling connection per worker. A notification advances a process-local role generation and wakes every matching waiter. Reconnection and notification loss fall back -to the ordinary polling interval. +to adaptive polling, whose current wait can be as long as the configured idle +ceiling. The optional Redis adapter provides the same role generations through Pub/Sub for deployments that already operate Redis. It keeps commands and subscriptions diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 6548631..c0dcfbf 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -3,6 +3,46 @@ The benchmark harness measures committed actor operations. It is intended to show tradeoffs and catch large regressions, not to predict application capacity. +## Idle polling + +The idle harness measures process CPU and empty database passes for the four +runtime roles: + +```bash +pnpm run benchmark:idle +``` + +It warms each interval for three seconds, measures for ten seconds, and reports +process user plus system CPU time divided by wall time. + +Measured on August 16, 2026 on an Apple M5 with Node.js 26.7.0 and in-memory +SQLite. The before run used 0.13.0; the after run used the prepared 0.13.1 tree. +Each run started one actor, effect, reminder, and broadcast role. + +| Fast interval | Before polls/s | Before CPU | After polls/s | After CPU | +| ------------: | -------------: | ---------: | ------------: | --------: | +| 20 ms | 188.78 | 3.254% | 4.000 | 0.129% | +| 100 ms | 39.596 | 0.906% | 3.999 | 0.121% | +| 500 ms | 7.999 | 0.251% | 3.999 | 0.104% | + +The after run reached the one-second ceiling for all four roles. These are +developer-laptop measurements, not a CPU guarantee; timer scheduling, JIT, +database path, and unrelated host activity affect short samples. + +Five SQLite samples measured durable enqueue through committed completion after +2.5 seconds of idleness. The polling-only multi-process harness submits just +after an empty pass, so it measures approximately the full polling wait rather +than average arrival latency. + +| Topology | 0.13.0 p50 | Prepared 0.13.1 p50 | +| ------------------------------- | ---------: | ------------------: | +| One process, in-process wake-up | 2.589 ms | 2.662 ms | +| Two processes, polling only | 107.945 ms | 1,006.232 ms | + +The local wake-up keeps the one-process path prompt after backoff. The +polling-only row is the explicit tradeoff: use PostgreSQL notifications or +optional Redis Pub/Sub when separate processes need low-latency delivery. + ## Scenarios - `warm-hot`: all operations target one previously created identity. diff --git a/docs/configuration.md b/docs/configuration.md index 2264a05..44df7d2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -11,6 +11,7 @@ through `runtime.ref(ActorClass, actorId)`. Both validate options immediately. | `database` | required | A `Database` adapter. | | `tableNamePrefix` | `"solid_objects_"` | Lowercase letters, digits, and underscores; must start with a letter. | | `pollingIntervalMilliseconds` | `100` | Positive durable-work polling interval. | +| `idlePollingIntervalMilliseconds` | `1_000` | Positive ceiling after consecutive empty polling passes. | | `syncPollingIntervalMilliseconds` | `50` | Positive result-wait polling interval. | | `leaseDurationMilliseconds` | `30_000` | Positive activation lease; must exceed renewal interval. | | `leaseRenewalIntervalMilliseconds` | `10_000` | Positive activation renewal cadence. | @@ -52,8 +53,16 @@ affected failure path rather than schedule an invalid timestamp. Counts may be zero, but the complete configuration must leave at least one runtime role enabled. Broadcast workers are started only when `broadcast` or -`authorizeSubscription` is configured. Wake-ups reduce latency; durable polling -remains the correctness path. +`authorizeSubscription` is configured. + +`pollingIntervalMilliseconds` is the fast interval after work or a wake-up. +Consecutive empty passes double it up to +`idlePollingIntervalMilliseconds`. Actor workers never wait longer than +`leaseRenewalIntervalMilliseconds`. Set the fast and idle values equal for a +fixed cadence. A custom wake-up adapter should 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. ## Retention and cleanup diff --git a/docs/operations.md b/docs/operations.md index 94a67b3..110ec9c 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -1,12 +1,30 @@ # Operations -Runtime roles use durable polling as the correctness fallback. The default -generation-based wake-up adapter interrupts waits for new actor messages, -effects, reminders, and broadcasts in the same Node process. Notification -errors are isolated and logged by role and error class without failing the -committed work. Graceful shutdown stops new claims and allows active turns to -finish within `shutdownTimeoutMilliseconds`, which defaults to 15 seconds. A -component still running or stopping at the deadline emits +Runtime roles use durable polling as the correctness fallback. 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 +workers clamp the ceiling to `leaseRenewalIntervalMilliseconds` while they may +hold cached activations. + +The default generation-based wake-up adapter interrupts waits for new actor +messages, effects, reminders, and broadcasts in the same Node process. It does +not cross a process boundary. When live processes share the database without a +configured adapter, the runtime logs +`solid_objects.polling_only_cross_process_wake_up` once. Use PostgreSQL +notifications or optional Redis Pub/Sub when separate processes need prompt +delivery; without one, newly committed work can wait up to the current idle +polling interval. Notification errors are isolated and logged by role and error +class without failing the committed work. + +Each role exposes `currentPollingIntervalMilliseconds`. +`solid_objects.polling.interval_changed` reports the role, reason, previous +interval, and current interval. The polling-only warning is also emitted as +`solid_objects.polling.only_cross_process_wake_up` instrumentation. + +Graceful shutdown stops new claims and allows active turns to finish within +`shutdownTimeoutMilliseconds`, which defaults to 15 seconds. A component still +running or stopping at the deadline emits `solid_objects.supervisor.component_shutdown_timeout`; the runtime then returns without pretending JavaScript code was forcibly terminated. Operators should monitor oldest ready work, claimed work, dead letters, effect failures, diff --git a/docs/parity.md b/docs/parity.md index 875291c..8b1fe64 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -4,11 +4,11 @@ This ledger tracks capability parity with the Ruby `solid_objects` gem. Parity means preserving a capability and its correctness or security boundary, not copying a Rails API into Node. -Reference: Ruby `solid_objects` 0.13.0. The JavaScript package began at the +Reference: Ruby `solid_objects` 0.13.1. The JavaScript package began at the Ruby design's `0.12` capability generation; that version number did not imply earlier JavaScript releases. -The Node `0.13.0` implementation has capability parity with that reference. Its +The Node `0.13.1` implementation has capability parity with that reference. Its relational runtime, correctness boundaries, administration, diagnostics, operator dashboard, realtime projections, browser behavior, and supported adapters have native equivalents. Rails-specific rendering surfaces are diff --git a/docs/releasing.md b/docs/releasing.md index 14486fa..efe9c60 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -38,8 +38,8 @@ npm trust github solid-objects \ 4. Create and push an annotated tag matching the package version: ```shell - git tag -a v0.13.0 -m "Version 0.13.0" - git push origin v0.13.0 + git tag -a v0.13.1 -m "Version 0.13.1" + git push origin v0.13.1 ``` The tag runs the complete CI matrix. The publish job starts only after every diff --git a/package.json b/package.json index db5b7f6..59d5f18 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "solid-objects", - "version": "0.13.0", + "version": "0.13.1", "description": "Race-free realtime state per application identity, backed by your SQL database", "type": "module", "license": "MIT", @@ -85,6 +85,7 @@ "test:redis": "vitest run test/redis-wake-up.test.ts", "test:watch": "vitest", "benchmark": "pnpm run build && node benchmarks/run.ts", + "benchmark:idle": "pnpm run build && node benchmarks/idle.ts", "pack:check": "pnpm pack --dry-run && node scripts/check-package.mjs", "prepack": "pnpm run build" }, diff --git a/scripts/release-artifact-smoke.mjs b/scripts/release-artifact-smoke.mjs index c58912d..b171520 100644 --- a/scripts/release-artifact-smoke.mjs +++ b/scripts/release-artifact-smoke.mjs @@ -5,6 +5,7 @@ import { join, resolve } from "node:path" import { spawn } from "node:child_process" const repositoryRoot = resolve(import.meta.dirname, "..") +const packageDefinition = JSON.parse(await readFile(join(repositoryRoot, "package.json"), "utf8")) const temporaryDirectory = await mkdtemp(join(tmpdir(), "solid-objects-package-")) const artifactDirectory = join(temporaryDirectory, "artifact") const projectDirectory = join(temporaryDirectory, "project") @@ -21,7 +22,7 @@ try { ), )[0] assert.equal(packed.name, "solid-objects") - assert.equal(packed.version, "0.13.0") + assert.equal(packed.version, packageDefinition.version) const packagedPaths = new Set(packed.files.map((file) => file.path)) for (const expectedPath of [ @@ -50,7 +51,7 @@ try { const installedPackage = JSON.parse( await readFile(join(projectDirectory, "node_modules/solid-objects/package.json"), "utf8"), ) - assert.equal(installedPackage.version, "0.13.0") + assert.equal(installedPackage.version, packageDefinition.version) const resolvedModule = ( await run( diff --git a/src/broadcast-worker.ts b/src/broadcast-worker.ts index 3fe137a..57c9653 100644 --- a/src/broadcast-worker.ts +++ b/src/broadcast-worker.ts @@ -1,13 +1,29 @@ import { randomUUID } from "node:crypto" import type { SolidObjectsRuntime } from "./runtime.js" import { withProcessHeartbeat } from "./worker.js" +import { PollingBackoff } from "./polling-backoff.js" export class BroadcastWorker { readonly processId = randomUUID() private registered = false private stopping = false + private readonly pollingBackoff: PollingBackoff - constructor(private readonly runtime: SolidObjectsRuntime) {} + constructor(private readonly runtime: SolidObjectsRuntime) { + this.pollingBackoff = new PollingBackoff({ + minimumIntervalMilliseconds: runtime.settings.pollingIntervalMilliseconds, + maximumIntervalMilliseconds: runtime.settings.idlePollingIntervalMilliseconds, + onChange: (transition) => + runtime.emitInstrumentation("polling.interval_changed", { + role: "broadcasts", + ...transition, + }), + }) + } + + get currentPollingIntervalMilliseconds(): number { + return this.pollingBackoff.currentIntervalMilliseconds + } async runOnce(): Promise { if (this.stopping) return 0 @@ -36,15 +52,20 @@ export class BroadcastWorker { async run(signal: AbortSignal): Promise { await this.ensureRegistered() + await this.runtime.warnIfPollingIsOnlyCrossProcessWakeUp() while (!signal.aborted && !this.stopping) { const wakeUp = await this.runtime.settings.wakeUp.watch("broadcasts") const processed = await this.runOnce() - if (processed === 0) { - await wakeUp.wait({ - timeoutMilliseconds: this.runtime.settings.pollingIntervalMilliseconds, - signal, - }) + if (processed > 0) { + this.pollingBackoff.reset("work") + continue } + const notified = await wakeUp.wait({ + timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + signal, + }) + if (notified === false) this.pollingBackoff.recordIdle() + else this.pollingBackoff.reset("wake_up") } await this.stop() } diff --git a/src/configuration.ts b/src/configuration.ts index 7b9fcab..014ad0f 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -40,6 +40,7 @@ export interface SolidObjectsConfiguration { database: Database tableNamePrefix?: string pollingIntervalMilliseconds?: number + idlePollingIntervalMilliseconds?: number syncPollingIntervalMilliseconds?: number leaseDurationMilliseconds?: number leaseRenewalIntervalMilliseconds?: number @@ -96,6 +97,7 @@ export interface RuntimeSettings extends Required< broadcast?: (event: BroadcastEvent) => Promise instrumentation?: (event: InstrumentationEvent) => void wakeUp: WakeUpAdapter + wakeUpConfigured: boolean authorizationPoliciesConfigured: Readonly> } @@ -111,6 +113,7 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime database: configuration.database, tableNamePrefix: configuration.tableNamePrefix ?? "solid_objects_", pollingIntervalMilliseconds: configuration.pollingIntervalMilliseconds ?? 100, + idlePollingIntervalMilliseconds: configuration.idlePollingIntervalMilliseconds ?? 1_000, syncPollingIntervalMilliseconds: configuration.syncPollingIntervalMilliseconds ?? 50, leaseDurationMilliseconds: configuration.leaseDurationMilliseconds ?? 30_000, leaseRenewalIntervalMilliseconds: configuration.leaseRenewalIntervalMilliseconds ?? 10_000, @@ -152,6 +155,7 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime pruneBatchSize: configuration.pruneBatchSize ?? 1_000, logger: configuration.logger ?? consoleLogger, wakeUp: configuration.wakeUp ?? new InProcessWakeUpAdapter(), + wakeUpConfigured: configuration.wakeUp !== undefined, authorizeMessage: configuration.authorizeMessage ?? (() => false), authorizeQuery: configuration.authorizeQuery ?? (() => false), authorizeDestroy: configuration.authorizeDestroy ?? (() => false), @@ -199,6 +203,7 @@ function validateSettings(settings: RuntimeSettings): void { const positive: Record = { pollingIntervalMilliseconds: settings.pollingIntervalMilliseconds, + idlePollingIntervalMilliseconds: settings.idlePollingIntervalMilliseconds, syncPollingIntervalMilliseconds: settings.syncPollingIntervalMilliseconds, leaseDurationMilliseconds: settings.leaseDurationMilliseconds, leaseRenewalIntervalMilliseconds: settings.leaseRenewalIntervalMilliseconds, diff --git a/src/effect-worker.ts b/src/effect-worker.ts index 6a2994a..afbd669 100644 --- a/src/effect-worker.ts +++ b/src/effect-worker.ts @@ -1,13 +1,29 @@ import { randomUUID } from "node:crypto" import type { SolidObjectsRuntime } from "./runtime.js" import { withProcessHeartbeat } from "./worker.js" +import { PollingBackoff } from "./polling-backoff.js" export class EffectWorker { readonly processId = randomUUID() private registered = false private stopping = false + private readonly pollingBackoff: PollingBackoff - constructor(private readonly runtime: SolidObjectsRuntime) {} + constructor(private readonly runtime: SolidObjectsRuntime) { + this.pollingBackoff = new PollingBackoff({ + minimumIntervalMilliseconds: runtime.settings.pollingIntervalMilliseconds, + maximumIntervalMilliseconds: runtime.settings.idlePollingIntervalMilliseconds, + onChange: (transition) => + runtime.emitInstrumentation("polling.interval_changed", { + role: "effects", + ...transition, + }), + }) + } + + get currentPollingIntervalMilliseconds(): number { + return this.pollingBackoff.currentIntervalMilliseconds + } async runOnce(): Promise { if (this.stopping) return 0 @@ -36,15 +52,20 @@ export class EffectWorker { async run(signal: AbortSignal): Promise { await this.ensureRegistered() + await this.runtime.warnIfPollingIsOnlyCrossProcessWakeUp() while (!signal.aborted && !this.stopping) { const wakeUp = await this.runtime.settings.wakeUp.watch("effects") const processed = await this.runOnce() - if (processed === 0) { - await wakeUp.wait({ - timeoutMilliseconds: this.runtime.settings.pollingIntervalMilliseconds, - signal, - }) + if (processed > 0) { + this.pollingBackoff.reset("work") + continue } + const notified = await wakeUp.wait({ + timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + signal, + }) + if (notified === false) this.pollingBackoff.recordIdle() + else this.pollingBackoff.reset("wake_up") } await this.stop() } diff --git a/src/polling-backoff.ts b/src/polling-backoff.ts new file mode 100644 index 0000000..3f38b17 --- /dev/null +++ b/src/polling-backoff.ts @@ -0,0 +1,53 @@ +export type PollingBackoffReason = "idle" | "wake_up" | "work" + +export interface PollingBackoffTransition { + previousIntervalMilliseconds: number + currentIntervalMilliseconds: number + reason: PollingBackoffReason +} + +export interface PollingBackoffOptions { + minimumIntervalMilliseconds: number + maximumIntervalMilliseconds: number + onChange?: (transition: PollingBackoffTransition) => void +} + +export class PollingBackoff { + private intervalMilliseconds: number + private readonly minimumIntervalMilliseconds: number + private readonly maximumIntervalMilliseconds: number + private readonly onChange: ((transition: PollingBackoffTransition) => void) | undefined + + constructor(options: PollingBackoffOptions) { + this.minimumIntervalMilliseconds = options.minimumIntervalMilliseconds + this.intervalMilliseconds = options.minimumIntervalMilliseconds + this.maximumIntervalMilliseconds = Math.max( + options.minimumIntervalMilliseconds, + options.maximumIntervalMilliseconds, + ) + this.onChange = options.onChange + } + + get currentIntervalMilliseconds(): number { + return this.intervalMilliseconds + } + + recordIdle(): void { + this.change(Math.min(this.intervalMilliseconds * 2, this.maximumIntervalMilliseconds), "idle") + } + + reset(reason: Exclude): void { + this.change(this.minimumIntervalMilliseconds, reason) + } + + private change(intervalMilliseconds: number, reason: PollingBackoffReason): void { + if (intervalMilliseconds === this.intervalMilliseconds) return + const previousIntervalMilliseconds = this.intervalMilliseconds + this.intervalMilliseconds = intervalMilliseconds + this.onChange?.({ + previousIntervalMilliseconds, + currentIntervalMilliseconds: intervalMilliseconds, + reason, + }) + } +} diff --git a/src/reminder-scheduler.ts b/src/reminder-scheduler.ts index 1926c3b..30be5c1 100644 --- a/src/reminder-scheduler.ts +++ b/src/reminder-scheduler.ts @@ -1,13 +1,29 @@ import { randomUUID } from "node:crypto" import { UnknownOperation } from "./errors.js" import type { SolidObjectsRuntime } from "./runtime.js" +import { PollingBackoff } from "./polling-backoff.js" export class ReminderScheduler { readonly processId = randomUUID() private registered = false private stopping = false + private readonly pollingBackoff: PollingBackoff - constructor(private readonly runtime: SolidObjectsRuntime) {} + constructor(private readonly runtime: SolidObjectsRuntime) { + this.pollingBackoff = new PollingBackoff({ + minimumIntervalMilliseconds: runtime.settings.pollingIntervalMilliseconds, + maximumIntervalMilliseconds: runtime.settings.idlePollingIntervalMilliseconds, + onChange: (transition) => + runtime.emitInstrumentation("polling.interval_changed", { + role: "reminders", + ...transition, + }), + }) + } + + get currentPollingIntervalMilliseconds(): number { + return this.pollingBackoff.currentIntervalMilliseconds + } async runOnce(options: { now?: Date } = {}): Promise { if (this.stopping) return 0 @@ -57,15 +73,20 @@ export class ReminderScheduler { async run(signal: AbortSignal): Promise { await this.ensureRegistered() + await this.runtime.warnIfPollingIsOnlyCrossProcessWakeUp() while (!signal.aborted && !this.stopping) { const wakeUp = await this.runtime.settings.wakeUp.watch("reminders") const processed = await this.runOnce() - if (processed === 0) { - await wakeUp.wait({ - timeoutMilliseconds: this.runtime.settings.pollingIntervalMilliseconds, - signal, - }) + if (processed > 0) { + this.pollingBackoff.reset("work") + continue } + const notified = await wakeUp.wait({ + timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + signal, + }) + if (notified === false) this.pollingBackoff.recordIdle() + else this.pollingBackoff.reset("wake_up") } await this.stop() } diff --git a/src/repository.ts b/src/repository.ts index 49f9018..c806b6d 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -118,6 +118,20 @@ export class Repository { }) } + async hasLiveProcessOutsideCurrentHostProcess(): Promise { + return this.settings.database.connection(async (connection) => { + const now = await connection.nowMilliseconds() + const row = await connection.get<{ present: number | bigint }>( + `SELECT 1 AS present FROM ${this.table("processes")} + WHERE shutdown_state <> 'stopped' AND heartbeat_at_ms > ? + AND (hostname <> ? OR host_process_id <> ?) + LIMIT 1`, + [now - this.settings.processAliveThresholdMilliseconds, hostname(), process.pid], + ) + return row !== undefined + }) + } + async cleanupStaleProcesses(): Promise { return this.settings.database.transaction(async (connection) => { const now = await connection.nowMilliseconds() diff --git a/src/runtime.ts b/src/runtime.ts index f67715b..7b09f0e 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -164,6 +164,8 @@ export class SolidObjectsRuntime { private readonly additionalComponents: ComponentRegistration[] = [] private callerWorker: Worker | undefined private running = false + private pollingOnlyWakeUpWarningEmitted = false + private pollingOnlyWakeUpWarningCheck: Promise | undefined constructor(configuration: SolidObjectsConfiguration) { this.settings = buildSettings(configuration) @@ -1303,6 +1305,35 @@ export class SolidObjectsRuntime { clearDefaultRuntime(this) } + async warnIfPollingIsOnlyCrossProcessWakeUp(): Promise { + if (this.settings.wakeUpConfigured || this.pollingOnlyWakeUpWarningEmitted) return + if (this.pollingOnlyWakeUpWarningCheck) return this.pollingOnlyWakeUpWarningCheck + const check = this.checkPollingOnlyCrossProcessWakeUp() + this.pollingOnlyWakeUpWarningCheck = check + try { + await check + } finally { + if (this.pollingOnlyWakeUpWarningCheck === check) { + this.pollingOnlyWakeUpWarningCheck = undefined + } + } + } + + private async checkPollingOnlyCrossProcessWakeUp(): Promise { + if (!(await this.repository.hasLiveProcessOutsideCurrentHostProcess())) return + if (this.pollingOnlyWakeUpWarningEmitted) return + this.pollingOnlyWakeUpWarningEmitted = true + const attributes = { + pollingIntervalMilliseconds: this.settings.pollingIntervalMilliseconds, + idlePollingIntervalMilliseconds: this.settings.idlePollingIntervalMilliseconds, + } + this.settings.logger.warn({ + event: "solid_objects.polling_only_cross_process_wake_up", + ...attributes, + }) + this.emitInstrumentation("polling.only_cross_process_wake_up", attributes) + } + async resetForTesting(): Promise { if (this.running) throw new Error("abort runtime.run() before resetting test state") await this.callerWorker?.stop() diff --git a/src/version.ts b/src/version.ts index 9410c0c..29a922e 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const VERSION = "0.13.0" +export const VERSION = "0.13.1" diff --git a/src/wake-up.ts b/src/wake-up.ts index bac3b94..fc451f5 100644 --- a/src/wake-up.ts +++ b/src/wake-up.ts @@ -6,7 +6,7 @@ export interface WakeUpWaitOptions { } export interface WakeUpWatch { - wait(options: WakeUpWaitOptions): Promise + wait(options: WakeUpWaitOptions): Promise } export interface WakeUpAdapter { @@ -17,7 +17,7 @@ export interface WakeUpAdapter { export class InProcessWakeUpAdapter implements WakeUpAdapter { private readonly generations = new Map() - private readonly waiters = new Map void>>() + private readonly waiters = new Map void>>() private closed = false watch(role: WakeUpRole): WakeUpWatch { @@ -33,14 +33,14 @@ export class InProcessWakeUpAdapter implements WakeUpAdapter { const waiters = this.waiters.get(role) if (!waiters) return this.waiters.delete(role) - for (const wake of waiters) wake() + for (const wake of waiters) wake(true) } close(): void { if (this.closed) return this.closed = true for (const waiters of this.waiters.values()) { - for (const wake of waiters) wake() + for (const wake of waiters) wake(false) } this.waiters.clear() } @@ -50,32 +50,33 @@ export class InProcessWakeUpAdapter implements WakeUpAdapter { generation: number timeoutMilliseconds: number signal?: AbortSignal - }): Promise { - if (this.closed || options.signal?.aborted) return Promise.resolve() - if (this.generation(options.role) !== options.generation) return Promise.resolve() + }): Promise { + if (this.closed || options.signal?.aborted) return Promise.resolve(false) + if (this.generation(options.role) !== options.generation) return Promise.resolve(true) return new Promise((resolve) => { let settled = false const waiters = this.waiters.get(options.role) ?? new Set() - const finish = () => { + const finish = (notified: boolean) => { if (settled) return settled = true clearTimeout(timeout) waiters.delete(finish) if (waiters.size === 0) this.waiters.delete(options.role) - options.signal?.removeEventListener("abort", finish) - resolve() + options.signal?.removeEventListener("abort", abort) + resolve(notified) } - const timeout = setTimeout(finish, options.timeoutMilliseconds) + const abort = () => finish(false) + const timeout = setTimeout(() => finish(false), options.timeoutMilliseconds) waiters.add(finish) this.waiters.set(options.role, waiters) - options.signal?.addEventListener("abort", finish, { once: true }) + options.signal?.addEventListener("abort", abort, { once: true }) if ( this.closed || options.signal?.aborted || this.generation(options.role) !== options.generation ) { - finish() + finish(!this.closed && !options.signal?.aborted) } }) } diff --git a/src/wake-up/postgresql.ts b/src/wake-up/postgresql.ts index 8901270..14f297b 100644 --- a/src/wake-up/postgresql.ts +++ b/src/wake-up/postgresql.ts @@ -25,7 +25,7 @@ export class PostgreSQLWakeUpAdapter implements WakeUpAdapter { private readonly channels = new Map() private readonly rolesByChannel = new Map() private readonly generations = new Map() - private readonly waiters = new Map void>>() + private readonly waiters = new Map void>>() private readonly listenedRoles = new Set() private readonly listening = new Map>() private readonly onListenerError: (failure: PostgreSQLWakeUpFailure) => void @@ -181,45 +181,46 @@ export class PostgreSQLWakeUpAdapter implements WakeUpAdapter { generation: number timeoutMilliseconds: number signal?: AbortSignal - }): Promise { - if (this.closed || options.signal?.aborted) return Promise.resolve() - if (this.generation(options.role) !== options.generation) return Promise.resolve() + }): Promise { + if (this.closed || options.signal?.aborted) return Promise.resolve(false) + if (this.generation(options.role) !== options.generation) return Promise.resolve(true) return new Promise((resolve) => { let settled = false const waiters = this.waiters.get(options.role) ?? new Set() - const finish = () => { + const finish = (notified: boolean) => { if (settled) return settled = true clearTimeout(timeout) waiters.delete(finish) if (waiters.size === 0) this.waiters.delete(options.role) - options.signal?.removeEventListener("abort", finish) - resolve() + options.signal?.removeEventListener("abort", abort) + resolve(notified) } - const timeout = setTimeout(finish, options.timeoutMilliseconds) + const abort = () => finish(false) + const timeout = setTimeout(() => finish(false), options.timeoutMilliseconds) waiters.add(finish) this.waiters.set(options.role, waiters) - options.signal?.addEventListener("abort", finish, { once: true }) + options.signal?.addEventListener("abort", abort, { once: true }) if ( this.closed || options.signal?.aborted || this.generation(options.role) !== options.generation ) { - finish() + finish(!this.closed && !options.signal?.aborted) } }) } - private wake(role: WakeUpRole): void { + private wake(role: WakeUpRole, notified = true): void { this.generations.set(role, this.generation(role) + 1) const waiters = this.waiters.get(role) if (!waiters) return this.waiters.delete(role) - for (const finish of waiters) finish() + for (const finish of waiters) finish(notified) } private wakeEveryRole(): void { - for (const role of ROLES) this.wake(role) + for (const role of ROLES) this.wake(role, false) } private generation(role: WakeUpRole): number { diff --git a/src/worker.ts b/src/worker.ts index 62c9e95..89eef67 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -8,6 +8,7 @@ import { } from "./errors.js" import type { ActivationLease, ClaimedTurn } from "./records.js" import type { SolidObjectsRuntime } from "./runtime.js" +import { PollingBackoff } from "./polling-backoff.js" interface CachedActivation { actor: Actor @@ -21,11 +22,32 @@ export class Worker { private registered = false private stopping = false private readonly activations = new Map() + private readonly pollingBackoff: PollingBackoff constructor( private readonly runtime: SolidObjectsRuntime, private readonly options: { setupFailure?: "report" | "throw" } = {}, - ) {} + ) { + this.pollingBackoff = new PollingBackoff({ + minimumIntervalMilliseconds: Math.min( + runtime.settings.pollingIntervalMilliseconds, + runtime.settings.leaseRenewalIntervalMilliseconds, + ), + maximumIntervalMilliseconds: Math.min( + runtime.settings.idlePollingIntervalMilliseconds, + runtime.settings.leaseRenewalIntervalMilliseconds, + ), + onChange: (transition) => + runtime.emitInstrumentation("polling.interval_changed", { + role: "actors", + ...transition, + }), + }) + } + + get currentPollingIntervalMilliseconds(): number { + return this.pollingBackoff.currentIntervalMilliseconds + } async runOnce(options: { activationRetention?: "retain" | "release" } = {}): Promise { try { @@ -117,18 +139,20 @@ export class Worker { async run(signal: AbortSignal): Promise { await this.ensureRegistered() + await this.runtime.warnIfPollingIsOnlyCrossProcessWakeUp() while (!signal.aborted && !this.stopping) { const wakeUp = await this.runtime.settings.wakeUp.watch("actors") const processed = await this.runOnce() - if (processed === 0) { - await wakeUp.wait({ - timeoutMilliseconds: Math.min( - this.runtime.settings.pollingIntervalMilliseconds, - this.runtime.settings.leaseRenewalIntervalMilliseconds, - ), - signal, - }) + if (processed > 0) { + this.pollingBackoff.reset("work") + continue } + const notified = await wakeUp.wait({ + timeoutMilliseconds: this.pollingBackoff.currentIntervalMilliseconds, + signal, + }) + if (notified === false) this.pollingBackoff.recordIdle() + else this.pollingBackoff.reset("wake_up") } await this.stop() } diff --git a/test/definition.test.ts b/test/definition.test.ts index 1ccf0f3..b650f24 100644 --- a/test/definition.test.ts +++ b/test/definition.test.ts @@ -212,6 +212,9 @@ describe("runtime configuration", () => { expect(() => buildSettings({ database, maxActivationDurationMilliseconds: 0 })).toThrow( "maxActivationDurationMilliseconds must be positive", ) + expect(() => buildSettings({ database, idlePollingIntervalMilliseconds: 0 })).toThrow( + "idlePollingIntervalMilliseconds must be positive", + ) expect(() => buildSettings({ database, shutdownTimeoutMilliseconds: 0 })).toThrow( "shutdownTimeoutMilliseconds must be positive", ) diff --git a/test/polling-backoff.test.ts b/test/polling-backoff.test.ts new file mode 100644 index 0000000..108b7b4 --- /dev/null +++ b/test/polling-backoff.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it } from "vitest" +import { PollingBackoff } from "../src/polling-backoff.js" + +describe("PollingBackoff", () => { + it("doubles empty-poll intervals until the idle ceiling", () => { + const backoff = new PollingBackoff({ + minimumIntervalMilliseconds: 25, + maximumIntervalMilliseconds: 1_000, + }) + + const intervals = [backoff.currentIntervalMilliseconds] + for (let emptyPolls = 0; emptyPolls < 7; emptyPolls += 1) { + backoff.recordIdle() + intervals.push(backoff.currentIntervalMilliseconds) + } + + expect(intervals).toEqual([25, 50, 100, 200, 400, 800, 1_000, 1_000]) + }) + + it("resets to the fast interval after processed work", () => { + const transitions: Array<{ + previousIntervalMilliseconds: number + currentIntervalMilliseconds: number + reason: string + }> = [] + const backoff = new PollingBackoff({ + minimumIntervalMilliseconds: 25, + maximumIntervalMilliseconds: 1_000, + onChange: (transition) => transitions.push(transition), + }) + backoff.recordIdle() + backoff.recordIdle() + + backoff.reset("work") + + expect(backoff.currentIntervalMilliseconds).toBe(25) + expect(transitions.at(-1)).toEqual({ + previousIntervalMilliseconds: 100, + currentIntervalMilliseconds: 25, + reason: "work", + }) + }) +}) diff --git a/test/polling-loop.test.ts b/test/polling-loop.test.ts new file mode 100644 index 0000000..a7fa0dc --- /dev/null +++ b/test/polling-loop.test.ts @@ -0,0 +1,392 @@ +import { afterEach, describe, expect, it, vi } from "vitest" +import { sqlite } from "../src/database/sqlite.js" +import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js" +import type { InstrumentationEvent } from "../src/configuration.js" +import type { WakeUpAdapter, WakeUpRole, WakeUpWatch } from "../src/wake-up.js" + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + await runtime?.close() + runtime = undefined +}) + +describe("idle polling", () => { + it("backs an idle worker off to the configured ceiling and reports each transition", async () => { + const controller = new AbortController() + const intervals: number[] = [] + const events: InstrumentationEvent[] = [] + const wakeUp = new ImmediateTimeoutWakeUpAdapter((interval) => { + intervals.push(interval) + if (intervals.length === 7) controller.abort() + }) + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + workerCount: 1, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + instrumentation: (event) => events.push(event), + wakeUp, + }) + await runtime.install() + const worker = runtime.worker() + + await worker.run(controller.signal) + + expect(intervals).toEqual([25, 50, 100, 200, 400, 800, 1_000]) + expect(worker.currentPollingIntervalMilliseconds).toBe(1_000) + expect( + events + .filter(({ name }) => name === "solid_objects.polling.interval_changed") + .map(({ attributes }) => attributes), + ).toEqual([ + { + role: "actors", + reason: "idle", + previousIntervalMilliseconds: 25, + currentIntervalMilliseconds: 50, + }, + { + role: "actors", + reason: "idle", + previousIntervalMilliseconds: 50, + currentIntervalMilliseconds: 100, + }, + { + role: "actors", + reason: "idle", + previousIntervalMilliseconds: 100, + currentIntervalMilliseconds: 200, + }, + { + role: "actors", + reason: "idle", + previousIntervalMilliseconds: 200, + currentIntervalMilliseconds: 400, + }, + { + role: "actors", + reason: "idle", + previousIntervalMilliseconds: 400, + currentIntervalMilliseconds: 800, + }, + { + role: "actors", + reason: "idle", + previousIntervalMilliseconds: 800, + currentIntervalMilliseconds: 1_000, + }, + ]) + }) + + it("backs an idle effect worker off to the configured ceiling", async () => { + const controller = new AbortController() + const intervals: number[] = [] + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + workerCount: 0, + effectWorkerCount: 1, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + wakeUp: new ImmediateTimeoutWakeUpAdapter((interval) => { + intervals.push(interval) + if (intervals.length === 7) controller.abort() + }), + }) + await runtime.install() + const worker = runtime.effectWorker() + + await worker.run(controller.signal) + + expect(intervals).toEqual([25, 50, 100, 200, 400, 800, 1_000]) + expect(worker.currentPollingIntervalMilliseconds).toBe(1_000) + }) + + it("backs idle reminder and broadcast roles off to the configured ceiling", async () => { + for (const role of ["reminders", "broadcasts"] as const) { + const controller = new AbortController() + const intervals: number[] = [] + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + workerCount: 0, + effectWorkerCount: 0, + reminderSchedulerCount: role === "reminders" ? 1 : 0, + broadcastWorkerCount: role === "broadcasts" ? 1 : 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + ...(role === "broadcasts" ? { authorizeSubscription: () => true } : {}), + wakeUp: new ImmediateTimeoutWakeUpAdapter((interval) => { + intervals.push(interval) + if (intervals.length === 7) controller.abort() + }), + }) + await runtime.install() + const component = + role === "reminders" ? runtime.reminderScheduler() : runtime.broadcastWorker() + + await component.run(controller.signal) + + expect(intervals, role).toEqual([25, 50, 100, 200, 400, 800, 1_000]) + expect(component.currentPollingIntervalMilliseconds, role).toBe(1_000) + await runtime.close() + runtime = undefined + } + }) + + it("never backs an actor worker off beyond its lease renewal interval", async () => { + const controller = new AbortController() + const intervals: number[] = [] + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + leaseDurationMilliseconds: 300, + leaseRenewalIntervalMilliseconds: 100, + workerCount: 1, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + wakeUp: new ImmediateTimeoutWakeUpAdapter((interval) => { + intervals.push(interval) + if (intervals.length === 5) controller.abort() + }), + }) + await runtime.install() + + await runtime.worker().run(controller.signal) + + expect(intervals).toEqual([25, 50, 100, 100, 100]) + }) + + it("resets an actor worker to its fast interval after processed work", async () => { + const controller = new AbortController() + const intervals: number[] = [] + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + workerCount: 1, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + wakeUp: new ImmediateTimeoutWakeUpAdapter((interval) => { + intervals.push(interval) + if (intervals.length === 3) controller.abort() + }), + }) + await runtime.install() + const worker = runtime.worker() + vi.spyOn(worker, "runOnce") + .mockResolvedValueOnce(0) + .mockResolvedValueOnce(0) + .mockResolvedValueOnce(1) + .mockResolvedValue(0) + + await worker.run(controller.signal) + + expect(intervals).toEqual([25, 50, 25]) + }) + + it("resets an actor worker to its fast interval after a wake-up", async () => { + const controller = new AbortController() + const intervals: number[] = [] + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + workerCount: 1, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + wakeUp: new SequencedWakeUpAdapter({ + results: [false, false, true, false], + waiting: (interval) => { + intervals.push(interval) + if (intervals.length === 4) controller.abort() + }, + }), + }) + await runtime.install() + const worker = runtime.worker() + vi.spyOn(worker, "runOnce").mockResolvedValue(0) + + await worker.run(controller.signal) + + expect(intervals).toEqual([25, 50, 100, 25]) + }) + + it("keeps a legacy wake-up adapter at the fast interval", async () => { + const controller = new AbortController() + const intervals: number[] = [] + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + workerCount: 1, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + wakeUp: new LegacyWakeUpAdapter((interval) => { + intervals.push(interval) + if (intervals.length === 3) controller.abort() + }), + }) + await runtime.install() + const worker = runtime.worker() + vi.spyOn(worker, "runOnce").mockResolvedValue(0) + + await worker.run(controller.signal) + + expect(intervals).toEqual([25, 25, 25]) + }) + + it("warns once when another process shares the database without a wake-up adapter", async () => { + const logger = { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + } + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + pollingIntervalMilliseconds: 25, + workerCount: 2, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + logger, + }) + await runtime.install() + await runtime.repository.registerProcess("other-process", "worker") + await runtime.settings.database.connection((connection) => + connection.run( + `UPDATE ${runtime?.repository.table("processes")} SET host_process_id = ? WHERE id = ?`, + [process.pid + 1, "other-process"], + ), + ) + const controller = new AbortController() + const running = runtime.run(controller.signal) + + try { + await vi.waitFor( + () => { + expect(logger.warn).toHaveBeenCalledTimes(1) + }, + { timeout: 500, interval: 10 }, + ) + } finally { + controller.abort() + await running + } + + expect(logger.warn).toHaveBeenCalledWith({ + event: "solid_objects.polling_only_cross_process_wake_up", + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + }) + }) + + it("does not warn when a cross-process wake-up adapter is configured", async () => { + const logger = { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + } + runtime = createRuntime({ + database: sqlite({ path: ":memory:" }), + workerCount: 1, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + logger, + wakeUp: new ImmediateTimeoutWakeUpAdapter(() => {}), + }) + await runtime.install() + await runtime.repository.registerProcess("other-process", "worker") + await runtime.settings.database.connection((connection) => + connection.run( + `UPDATE ${runtime?.repository.table("processes")} SET host_process_id = ? WHERE id = ?`, + [process.pid + 1, "other-process"], + ), + ) + + await runtime.warnIfPollingIsOnlyCrossProcessWakeUp() + + expect(logger.warn).not.toHaveBeenCalled() + }) +}) + +class ImmediateTimeoutWakeUpAdapter implements WakeUpAdapter { + constructor(private readonly waiting: (timeoutMilliseconds: number) => void) {} + + watch(_role: WakeUpRole): WakeUpWatch { + return { + wait: async ({ timeoutMilliseconds }) => { + this.waiting(timeoutMilliseconds) + return false + }, + } + } + + notify(_role: WakeUpRole): void {} + + close(): void {} +} + +class SequencedWakeUpAdapter implements WakeUpAdapter { + private index = 0 + + constructor( + private readonly options: { + results: readonly boolean[] + waiting: (timeoutMilliseconds: number) => void + }, + ) {} + + watch(_role: WakeUpRole): WakeUpWatch { + return { + wait: async ({ timeoutMilliseconds }) => { + this.options.waiting(timeoutMilliseconds) + const result = this.options.results[this.index] + this.index += 1 + return result ?? false + }, + } + } + + notify(_role: WakeUpRole): void {} + + close(): void {} +} + +class LegacyWakeUpAdapter implements WakeUpAdapter { + constructor(private readonly waiting: (timeoutMilliseconds: number) => void) {} + + watch(_role: WakeUpRole): WakeUpWatch { + return { + wait: async ({ timeoutMilliseconds }) => { + this.waiting(timeoutMilliseconds) + }, + } + } + + notify(_role: WakeUpRole): void {} + + close(): void {} +} diff --git a/test/postgresql.test.ts b/test/postgresql.test.ts index 3915519..baf0f3f 100644 --- a/test/postgresql.test.ts +++ b/test/postgresql.test.ts @@ -192,20 +192,43 @@ describePostgreSQL("PostgreSQL adapter", () => { const actorWaits = actorWatches .slice(0, 2) .map((watch) => watch.wait({ timeoutMilliseconds: 10_000 })) - let effectResolved = false - void effectWatch.wait({ timeoutMilliseconds: 10_000 }).then(() => { - effectResolved = true + let effectResult: boolean | void | undefined + void effectWatch.wait({ timeoutMilliseconds: 10_000 }).then((result) => { + effectResult = result }) const startedAt = performance.now() await notifier.notify("actors") - await Promise.all([...actorWaits, actorWatches[2]!.wait({ timeoutMilliseconds: 10_000 })]) + await expect( + Promise.all([...actorWaits, actorWatches[2]!.wait({ timeoutMilliseconds: 10_000 })]), + ).resolves.toEqual([true, true, true]) expect(performance.now() - startedAt).toBeLessThan(1_000) - expect(effectResolved).toBe(false) + expect(effectResult).toBeUndefined() await listener.close() await new Promise((resolve) => setImmediate(resolve)) - expect(effectResolved).toBe(true) + expect(effectResult).toBe(false) + }) + + it("distinguishes a PostgreSQL wake-up from a polling timeout", async () => { + if (!connectionString) throw new Error("PostgreSQL connection string is required") + const listener = postgresqlWakeUp({ + connectionString, + channelPrefix: "postgresql_test_wait_result", + }) + const notifier = postgresqlWakeUp({ + connectionString, + channelPrefix: "postgresql_test_wait_result", + }) + wakeUps.push(listener, notifier) + + const timedOutWatch = await listener.watch("actors") + await expect(timedOutWatch.wait({ timeoutMilliseconds: 1 })).resolves.toBe(false) + const notifiedWatch = await listener.watch("actors") + const notified = notifiedWatch.wait({ timeoutMilliseconds: 10_000 }) + await notifier.notify("actors") + + await expect(notified).resolves.toBe(true) }) it("reconnects a listener after PostgreSQL closes its session", async () => { @@ -236,12 +259,12 @@ describePostgreSQL("PostgreSQL adapter", () => { ), ) expect(terminated?.terminated).toBe(true) - await interruptedWait + await expect(interruptedWait).resolves.toBe(false) const reconnectedWatch = await listener.watch("actors") const reconnectedWait = reconnectedWatch.wait({ timeoutMilliseconds: 10_000 }) await notifier.notify("actors") - await reconnectedWait + await expect(reconnectedWait).resolves.toBe(true) expect(failures).toContain("connection") }) diff --git a/test/process-administration.test.ts b/test/process-administration.test.ts index 9940efb..7483245 100644 --- a/test/process-administration.test.ts +++ b/test/process-administration.test.ts @@ -42,7 +42,7 @@ describe("process administration", () => { hostProcessId: process.pid, metadata: { nodeVersion: process.version, - solidObjectsVersion: "0.13.0", + solidObjectsVersion: "0.13.1", }, shutdownState: "running", shutdownRequestedAt: null, diff --git a/test/redis-wake-up.test.ts b/test/redis-wake-up.test.ts index c809acb..9f88f65 100644 --- a/test/redis-wake-up.test.ts +++ b/test/redis-wake-up.test.ts @@ -32,7 +32,7 @@ describe("Redis wake-up configuration", () => { const startedAt = performance.now() const watch = await adapter.watch("actors") - await watch.wait({ timeoutMilliseconds: 1 }) + await expect(watch.wait({ timeoutMilliseconds: 1 })).resolves.toBe(false) await adapter.notify("actors") expect(performance.now() - startedAt).toBeLessThan(1_000) @@ -55,19 +55,21 @@ describeRedis("Redis wake-up adapter", () => { const actorWaits = actorWatches .slice(0, 2) .map((watch) => watch.wait({ timeoutMilliseconds: 10_000 })) - let effectResolved = false - void effectWatch.wait({ timeoutMilliseconds: 10_000 }).then(() => { - effectResolved = true + let effectResult: boolean | void | undefined + void effectWatch.wait({ timeoutMilliseconds: 10_000 }).then((result) => { + effectResult = result }) const startedAt = performance.now() await notifier.notify("actors") - await Promise.all([...actorWaits, actorWatches[2]!.wait({ timeoutMilliseconds: 10_000 })]) + await expect( + Promise.all([...actorWaits, actorWatches[2]!.wait({ timeoutMilliseconds: 10_000 })]), + ).resolves.toEqual([true, true, true]) expect(performance.now() - startedAt).toBeLessThan(1_000) - expect(effectResolved).toBe(false) + expect(effectResult).toBeUndefined() listener.close() await new Promise((resolve) => setImmediate(resolve)) - expect(effectResolved).toBe(true) + expect(effectResult).toBe(false) }) }) diff --git a/test/wake-up.test.ts b/test/wake-up.test.ts index 83fff32..732e556 100644 --- a/test/wake-up.test.ts +++ b/test/wake-up.test.ts @@ -44,6 +44,13 @@ afterEach(async () => { }) describe("in-process wake-up", () => { + it("distinguishes a polling timeout from a wake-up", async () => { + const wakeUp = new InProcessWakeUpAdapter() + const watch = wakeUp.watch("actors") + + await expect(watch.wait({ timeoutMilliseconds: 1 })).resolves.toBe(false) + }) + it("does not miss a signal sent before waiting", async () => { const wakeUp = new InProcessWakeUpAdapter() const watch = wakeUp.watch("actors") @@ -57,7 +64,7 @@ describe("in-process wake-up", () => { setTimeout(() => reject(new Error("wake-up was missed")), 100), ), ]), - ).resolves.toBeUndefined() + ).resolves.toBe(true) }) it("wakes every waiter for a role without waking other roles", async () => { @@ -123,6 +130,45 @@ describe("in-process wake-up", () => { expect(await message.result()).toBeNull() }) + it("processes a local message promptly after the worker reaches its idle ceiling", async () => { + runtime = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + authorizeQuery: () => true, + pollingIntervalMilliseconds: 25, + idlePollingIntervalMilliseconds: 1_000, + workerCount: 1, + effectWorkerCount: 0, + reminderSchedulerCount: 0, + retentionIntervalMilliseconds: 0, + deadProcessCleanupIntervalMilliseconds: 0, + }) + runtime.register(WakeTarget) + await runtime.install() + const worker = runtime.worker() + const controller = new AbortController() + const running = worker.run(controller.signal) + await vi.waitFor( + () => { + expect(worker.currentPollingIntervalMilliseconds).toBe(1_000) + }, + { timeout: 3_000, interval: 10 }, + ) + const startedAt = performance.now() + + const message = await WakeTarget.ref("backed-off").send.receive() + await vi.waitFor( + async () => { + expect(await message.status()).toBe("completed") + }, + { timeout: 400, interval: 5 }, + ) + + expect(performance.now() - startedAt).toBeLessThan(400) + controller.abort() + await running + }) + it("isolates notification failures from durable work", async () => { const logger = { debug: () => {},