diff --git a/CHANGELOG.md b/CHANGELOG.md index 15d6144..8feec93 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,35 @@ ## Unreleased +- Select a wake-up adapter automatically. `wakeUp` now takes a name or an + adapter, as `config.cache_store` does in Rails, and defaults to + `"automatic"`. Selection prefers `SOLID_OBJECTS_REDIS_URL`, then PostgreSQL + notifications, then polling. `"in_process"` opts out, and an unknown name + throws rather than polls quietly. +- PostgreSQL applications that configure nothing now use notifications. They + gain cross-process wake-up, one dedicated listening client, and one `NOTIFY` + per commit. Set `wakeUp: "in_process"` to keep polling. +- Prove the PostgreSQL notification path before selecting it, because `LISTEN` + does not survive a transaction pooler such as PgBouncer. Selection listens on + a probe channel, sends one `NOTIFY` from a second connection, and waits up to + two seconds for it to arrive. A probe that does not deliver falls back to + polling and warns once. +- Report the resolved choice. `runtime.wakeUpCapability()` names the adapter, + whether it crosses processes, its measured floor, and why it was chosen. The + doctor reports it as a `wakeUp` check, and the polling-only warning now fires + on what was installed rather than on whether a setting was set. +- Keep the capability a configured adapter reports about itself. A configured + `InProcessWakeUpAdapter` now reports `in_process` and warns, rather than + claim that it crosses processes. +- Poll rather than pretend when a requested adapter cannot be built. + `wakeUp: "postgresql"` on a database with no notification channel, + `wakeUp: "redis"` without `SOLID_OBJECTS_REDIS_URL`, and a Redis URL without + the `redis` package each log `solid_objects.wake_up.unavailable` once and + report the reason in the capability, so the doctor warns rather than claim a + cross-process wake-up that cannot happen. +- Select once per runtime. `runtime.wakeUpAdapter()` memoises the selection, so + callers that race for the first use share one probe rather than run one each. + - Add reminder reading. `reminder()` returns one armed alarm as a `ScheduledReminder`, and `reminders()` lists every key of one operation. Both apply the intents staged so far in the turn, so a read agrees with what the diff --git a/docs/api.md b/docs/api.md index 204aefe..582e7a6 100644 --- a/docs/api.md +++ b/docs/api.md @@ -664,6 +664,25 @@ 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. +`configuration.wakeUp` takes a name or an adapter. `WakeUpSetting` is that +union, `WakeUpName` is one of `WAKE_UP_NAMES`, and the default is +`"automatic"`. Selection prefers `SOLID_OBJECTS_REDIS_URL`, then PostgreSQL +notifications, then polling. An unknown name throws rather than polls. + +`selectWakeUp(options)` runs that choice and returns a `SelectedWakeUp`, which +pairs the adapter with a `WakeUpCapability`. `WakeUpSelectionOptions` names the +inputs: the setting, the database, the idle polling interval, a logger, and an +optional Redis URL and probe timeout. `WakeUpCapability` reports +`WakeUpAdapterName`, whether the adapter crosses processes, the measured floor +in milliseconds, and the reason. `runtime.wakeUpAdapter()` and +`runtime.wakeUpCapability()` return the choice, which is made once per runtime. + +An adapter may declare its own `defaultCapability`, which is what it reports +about itself. Selection keeps what a configured adapter declares rather than +assume, and `runtime.wakeUpCapability()` reports what selection installed. `NotificationWakeUpAdapter` adds +`channelFor(role)`, which a database that offers a notification channel +provides through `database.wakeUp(options)`. + ### Errors The root exports `SolidObjectsError` and its supported subclasses: diff --git a/docs/configuration.md b/docs/configuration.md index 6c24112..f36a519 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -49,19 +49,19 @@ 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` | `InProcessWakeUpAdapter` | 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`. | +| `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 @@ -158,6 +158,31 @@ short capped backoff only when no synchronous deadline is active. | `applicationName` | `"solid-objects"` | | `onPoolError` | structured `console.error` | +`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"` +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 +so. `"postgresql"` on a database with no notification channel, `"redis"` without +`SOLID_OBJECTS_REDIS_URL`, and a Redis URL without the `redis` package each log +`solid_objects.wake_up.unavailable` once and report the reason in the +capability. Only a name that does not exist is refused, because a typo cannot be +honoured at all. + +Selection runs once per runtime, on first use. `runtime.wakeUpCapability()` +reports what it chose, whether that choice crosses processes, its measured +floor in milliseconds, and why. The doctor reports the same record, and the +polling-only warning fires on what was installed rather than on whether a +setting was set. An adapter that declares its own `defaultCapability` keeps it, so a +configured `InProcessWakeUpAdapter` still warns. + +Automatic selection proves the PostgreSQL path before it chooses it. It listens +on a probe channel, sends one `NOTIFY` from a second connection, and waits up to +two seconds for it to arrive. A probe that does not deliver falls back to +polling and warns once, because `LISTEN` does not survive a transaction pooler +such as PgBouncer. + `database.wakeUp(options)` creates a dedicated notification adapter using the same connection string. Its options are `channelPrefix = "solid_objects"`, `applicationName = "solid-objects-wake-up"`, and `onListenerError`. Use a diff --git a/docs/operations.md b/docs/operations.md index e17def4..bfb00d1 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -14,16 +14,25 @@ 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 +The runtime selects a wake-up adapter on first use. It prefers +`SOLID_OBJECTS_REDIS_URL`, then PostgreSQL notifications, then polling. +The generation-based in-process 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 of them, newly committed work can wait for the current -idle polling interval. The runtime isolates notification errors and logs them by +not cross a process boundary. When live processes share the database and the +installed adapter does not cross processes, the runtime logs +`solid_objects.polling_only_cross_process_wake_up` once. Without a +cross-process adapter, newly committed work can wait for the current idle +polling interval. The runtime isolates notification errors and logs them by role and error class. The committed work does not fail. +`runtime.wakeUpCapability()` and the `wakeUp` doctor check report the adapter +that is installed, whether it crosses processes, its measured floor, and why it +was chosen. On PostgreSQL, selection first proves the path: it listens on a +probe channel, notifies it from a second connection, and waits for the +notification. A probe that does not arrive logs +`solid_objects.wake_up.pooled_session` once and falls back to polling, because +`LISTEN` does not survive a transaction pooler such as PgBouncer. + The warning excludes process rows with the current hostname and host process ID. It can therefore appear during a rolling deployment or restart overlap when an older and newer process briefly share the same database. A process that stopped diff --git a/docs/parity.md b/docs/parity.md index 7410553..ea4d443 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -106,11 +106,28 @@ Ruby field names; this does not change runtime delivery semantics. | PostgreSQL wake-up | Native | `database.wakeUp()` uses one dedicated event-driven client, role-specific `LISTEN/NOTIFY`, generation fencing, reconnectable listeners, and durable polling fallback. | | Redis wake-up | Native | An optional `redis` peer provides role-specific Pub/Sub over separate lazy publisher/subscriber connections, with bounded failures and durable polling fallback. | -Every wake-up adapter above is opt-in. Neither runtime selects one -automatically. An application that configures nothing keeps polling. Each -runtime warns once when live processes share a database without a configured -cross-process adapter. This limit is intentional in both runtimes. It is not a -gap between them. +Both runtimes select a wake-up adapter automatically. `wakeUp` takes a name or +an adapter and defaults to `"automatic"`, which prefers a configured Redis URL, +then PostgreSQL notifications, then polling. Selection proves the PostgreSQL +path with a probe notification, because `LISTEN` does not survive a transaction +pooler. Each runtime reports what it installed, whether that crosses processes, +its measured floor, and why, and warns once when live processes share a +database and the installed adapter cannot reach them. MySQL still polls in both +runtimes, because MySQL has no notification channel. + +A requested adapter that the environment cannot provide polls instead and says +so in both runtimes, rather than claim a cross-process wake-up that cannot +happen. Only a name that does not exist is refused. + +Three details differ, and each follows from the language rather than from the +feature. The pooled-session warning is emitted once per runtime here and once +per process in Ruby, because this runtime supports several runtimes in one +process. A configured adapter must implement `watch`, `notify`, and `close` +here, while Ruby requires `signal` and `wait` and treats `watch` and `stop` as +optional, which is each runtime's own adapter contract. A selection that a later +edit of the settings makes impossible is reported once here and raised in Ruby; +both runtimes refuse an unknown name when the configuration is built, so this +only reaches code that changes the setting afterwards. ## Realtime and browser behavior diff --git a/src/broadcast-worker.ts b/src/broadcast-worker.ts index 670ee47..f31892e 100644 --- a/src/broadcast-worker.ts +++ b/src/broadcast-worker.ts @@ -54,7 +54,7 @@ export class BroadcastWorker { await this.ensureRegistered() await this.runtime.warnIfPollingIsOnlyCrossProcessWakeUp() while (!signal.aborted && !this.stopping) { - const wakeUp = await this.runtime.settings.wakeUp.watch("broadcasts") + const wakeUp = await this.runtime.watchWakeUp("broadcasts") const processed = await this.runOnce() if (processed > 0) { this.pollingBackoff.reset("work") diff --git a/src/configuration.ts b/src/configuration.ts index ed27fe9..b55ab38 100644 --- a/src/configuration.ts +++ b/src/configuration.ts @@ -1,7 +1,9 @@ import { InvalidActor } from "./errors.js" import type { Database } from "./database/types.js" import type { DeepReadonly, JsonObject, JsonValue, Logger, LongRunningComponent } from "./types.js" -import { InProcessWakeUpAdapter, type WakeUpAdapter } from "./wake-up.js" +import { WAKE_UP_NAMES, type WakeUpAdapter, type WakeUpSetting } from "./wake-up.js" + +const wakeUpNames: readonly string[] = WAKE_UP_NAMES export interface AuthorizationInput { actorType: string @@ -79,7 +81,7 @@ export interface SolidObjectsConfiguration { authorizeSubscription?: (input: SubscriptionAuthorizationInput) => boolean | Promise instrumentation?: (event: InstrumentationEvent) => void broadcast?: (event: BroadcastEvent) => Promise - wakeUp?: WakeUpAdapter + wakeUp?: WakeUpSetting } export interface BroadcastEvent { @@ -97,8 +99,7 @@ export interface RuntimeSettings extends Required< logger: Logger broadcast?: (event: BroadcastEvent) => Promise instrumentation?: (event: InstrumentationEvent) => void - wakeUp: WakeUpAdapter - wakeUpConfigured: boolean + wakeUp: WakeUpSetting authorizationPoliciesConfigured: Readonly> } @@ -156,8 +157,7 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime processRetentionMilliseconds: configuration.processRetentionMilliseconds ?? 7 * 86_400_000, pruneBatchSize: configuration.pruneBatchSize ?? 1_000, logger: configuration.logger ?? consoleLogger, - wakeUp: configuration.wakeUp ?? new InProcessWakeUpAdapter(), - wakeUpConfigured: configuration.wakeUp !== undefined, + wakeUp: configuration.wakeUp ?? "automatic", authorizeMessage: configuration.authorizeMessage ?? (() => false), authorizeQuery: configuration.authorizeQuery ?? (() => false), authorizeDestroy: configuration.authorizeDestroy ?? (() => false), @@ -193,12 +193,25 @@ export function validateComponent(component: LongRunningComponent): void { } } -function validateSettings(settings: RuntimeSettings): void { +function validateWakeUp(setting: WakeUpSetting): void { + if (typeof setting !== "string") return validateWakeUpAdapter(setting) + if (wakeUpNames.includes(setting)) return + + throw new TypeError( + `unknown wakeUp ${JSON.stringify(setting)}, expected one of ${WAKE_UP_NAMES.join(", ")} or an adapter`, + ) +} + +function validateWakeUpAdapter(adapter: WakeUpAdapter): void { for (const name of ["watch", "notify", "close"] as const) { - if (typeof settings.wakeUp[name] !== "function") { - throw new TypeError(`wakeUp must implement ${name}`) - } + if (typeof adapter[name] === "function") continue + + throw new TypeError(`wakeUp must implement ${name}`) } +} + +function validateSettings(settings: RuntimeSettings): void { + validateWakeUp(settings.wakeUp) if (!/^[a-z][a-z0-9_]*$/.test(settings.tableNamePrefix)) { throw new TypeError("tableNamePrefix must contain lowercase letters, digits, and underscores") } diff --git a/src/database/types.ts b/src/database/types.ts index e8cb383..f061432 100644 --- a/src/database/types.ts +++ b/src/database/types.ts @@ -1,3 +1,5 @@ +import type { NotificationWakeUpAdapter } from "../wake-up.js" + export type DatabaseFamily = "sqlite" | "postgresql" | "mysql" export interface RunResult { @@ -20,6 +22,7 @@ export interface Database { readonly family: DatabaseFamily readonly schemaIdentity: string transactionActive?(): boolean + wakeUp?(options?: { channelPrefix?: string }): NotificationWakeUpAdapter connection(callback: (connection: DatabaseConnection) => Promise): Promise transaction( callback: (connection: DatabaseConnection) => Promise, diff --git a/src/doctor.ts b/src/doctor.ts index ad65018..19cfc9c 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -117,6 +117,7 @@ export class Doctor { schema, await this.checkAuthorization(), await this.checkDatabase(), + await this.checkWakeUp(), ] checks.push( schema.status === "fail" @@ -145,6 +146,26 @@ export class Doctor { }) } + private async checkWakeUp(): Promise { + try { + const capability = await this.runtime.wakeUpCapability() + const floor = capability.measuredFloorMilliseconds + const summary = `${capability.adapter}: ${capability.reason}${ + floor === undefined ? "" : `, floor ${floor} ms` + }` + if (capability.crossesProcesses) { + return check({ name: "wakeUp", status: "pass", message: summary }) + } + return check({ + name: "wakeUp", + status: "warn", + message: `${summary}; a commit in one process cannot wake another`, + }) + } catch (error) { + return failedCheck("wakeUp", error) + } + } + private checkConfiguration(): DoctorCheck { return check({ name: "configuration", diff --git a/src/effect-recovery-coordinator.ts b/src/effect-recovery-coordinator.ts index b6a4574..8146f23 100644 --- a/src/effect-recovery-coordinator.ts +++ b/src/effect-recovery-coordinator.ts @@ -9,6 +9,7 @@ import { } from "./effect-recovery.js" import type { EffectRow, EnqueueInput, MessageRow, ProcessRow } from "./records.js" import { jsonObject, normalizeJson } from "./serialization.js" +import type { WakeUpAdapter } from "./wake-up.js" import { notifyWakeUp } from "./wake-up-notification.js" interface RecoveryBinding { @@ -30,6 +31,7 @@ export class EffectRecoveryCoordinator { constructor( private readonly options: { settings: RuntimeSettings + wakeUpAdapter: () => Promise enqueue: (connection: DatabaseConnection, input: EnqueueInput) => Promise }, ) {} @@ -87,7 +89,7 @@ export class EffectRecoveryCoordinator { }) if (retired) notifyWakeUp({ - adapter: this.options.settings.wakeUp, + adapter: await this.options.wakeUpAdapter(), logger: this.options.settings.logger, role: "actors", }) diff --git a/src/effect-worker.ts b/src/effect-worker.ts index 68f4653..3446d4d 100644 --- a/src/effect-worker.ts +++ b/src/effect-worker.ts @@ -54,7 +54,7 @@ export class EffectWorker { await this.ensureRegistered() await this.runtime.warnIfPollingIsOnlyCrossProcessWakeUp() while (!signal.aborted && !this.stopping) { - const wakeUp = await this.runtime.settings.wakeUp.watch("effects") + const wakeUp = await this.runtime.watchWakeUp("effects") const processed = await this.runOnce() if (processed > 0) { this.pollingBackoff.reset("work") diff --git a/src/index.ts b/src/index.ts index 769b8b6..2e0ec5a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -51,11 +51,22 @@ export { export { BroadcastWorker } from "./broadcast-worker.js" export { InProcessWakeUpAdapter, + WAKE_UP_NAMES, + type NotificationWakeUpAdapter, type WakeUpAdapter, + type WakeUpAdapterName, + type WakeUpCapability, + type WakeUpName, type WakeUpRole, + type WakeUpSetting, type WakeUpWaitOptions, type WakeUpWatch, } from "./wake-up.js" +export { + selectWakeUp, + type SelectedWakeUp, + type WakeUpSelectionOptions, +} from "./wake-up-selection.js" export { parseSubscriptionRequest, RealtimeManager, diff --git a/src/reminder-scheduler.ts b/src/reminder-scheduler.ts index 96b7fe8..cfdcaa2 100644 --- a/src/reminder-scheduler.ts +++ b/src/reminder-scheduler.ts @@ -75,7 +75,7 @@ export class ReminderScheduler { await this.ensureRegistered() await this.runtime.warnIfPollingIsOnlyCrossProcessWakeUp() while (!signal.aborted && !this.stopping) { - const wakeUp = await this.runtime.settings.wakeUp.watch("reminders") + const wakeUp = await this.runtime.watchWakeUp("reminders") const processed = await this.runOnce() if (processed > 0) { this.pollingBackoff.reset("work") diff --git a/src/repository.ts b/src/repository.ts index 0e7296d..4840d28 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -36,6 +36,7 @@ import type { } from "./types.js" import { VERSION } from "./version.js" import { EffectRecoveryCoordinator } from "./effect-recovery-coordinator.js" +import type { WakeUpAdapter } from "./wake-up.js" export interface SyncDiagnosticsRecord { message: MessageRow @@ -67,7 +68,10 @@ interface MessageClaimFenceRow { } export class Repository { - constructor(private readonly settings: RuntimeSettings) {} + constructor( + private readonly settings: RuntimeSettings, + private readonly options: { wakeUpAdapter: () => Promise }, + ) {} table(name: string): string { return `${this.settings.tableNamePrefix}${name}` @@ -2056,6 +2060,7 @@ export class Repository { private effectRecoveryCoordinator(): EffectRecoveryCoordinator { return new EffectRecoveryCoordinator({ settings: this.settings, + wakeUpAdapter: this.options.wakeUpAdapter, enqueue: (connection, input) => this.enqueueInTransaction(connection, input), }) } diff --git a/src/runtime.ts b/src/runtime.ts index 8af5fee..259869d 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -122,8 +122,9 @@ import type { import { SolidObjectsTestHelper } from "./test-helper.js" import { waitFor, Worker } from "./worker.js" import { EffectWorker } from "./effect-worker.js" -import type { WakeUpRole } from "./wake-up.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 { withDatabaseDeadline } from "./database/deadline.js" import type { DatabaseConnection } from "./database/types.js" import { evaluateActorTurn, readActorObservables } from "./turn.js" @@ -201,10 +202,14 @@ export class SolidObjectsRuntime { private running = false private pollingOnlyWakeUpWarningEmitted = false private pollingOnlyWakeUpWarningCheck: Promise | undefined + private wakeUpSelection: Promise | undefined + private wakeUpSelectionFailureLogged = false constructor(configuration: SolidObjectsConfiguration) { this.settings = buildSettings(configuration) - this.repository = new Repository(this.settings) + this.repository = new Repository(this.settings, { + wakeUpAdapter: () => this.wakeUpAdapter(), + }) this.deadLetters = new DeadLetterManager(this) this.reconciliation = new ReconciliationManager(this) this.retention = new RetentionManager(this) @@ -1344,13 +1349,25 @@ export class SolidObjectsRuntime { throw new Error("abort runtime.run() and wait for it before closing the runtime") await this.callerWorker?.stop() this.realtime.close() - await this.settings.wakeUp.close() + await this.closeWakeUp() await this.settings.database.close() clearDefaultRuntime(this) } + async wakeUpAdapter(): Promise { + return (await this.resolveWakeUp()).adapter + } + + async wakeUpCapability(): Promise { + return (await this.resolveWakeUp()).capability + } + + async watchWakeUp(role: WakeUpRole): Promise { + return await (await this.wakeUpAdapter()).watch(role) + } + async warnIfPollingIsOnlyCrossProcessWakeUp(): Promise { - if (this.settings.wakeUpConfigured || this.pollingOnlyWakeUpWarningEmitted) return + if (this.pollingOnlyWakeUpWarningEmitted) return if (this.pollingOnlyWakeUpWarningCheck) return this.pollingOnlyWakeUpWarningCheck const check = this.checkPollingOnlyCrossProcessWakeUp() this.pollingOnlyWakeUpWarningCheck = check @@ -1364,6 +1381,7 @@ export class SolidObjectsRuntime { } private async checkPollingOnlyCrossProcessWakeUp(): Promise { + if ((await this.wakeUpCapability()).crossesProcesses) return if (!(await this.repository.hasLiveProcessOutsideCurrentHostProcess())) return if (this.pollingOnlyWakeUpWarningEmitted) return this.pollingOnlyWakeUpWarningEmitted = true @@ -1383,6 +1401,7 @@ export class SolidObjectsRuntime { await this.callerWorker?.stop() this.callerWorker = undefined this.realtime.close() + await this.discardWakeUp() await this.repository.resetForTesting() } @@ -1963,8 +1982,46 @@ export class SolidObjectsRuntime { }) } + private resolveWakeUp(): Promise { + this.wakeUpSelection ??= selectWakeUp({ + setting: this.settings.wakeUp, + database: this.settings.database, + idlePollingIntervalMilliseconds: this.settings.idlePollingIntervalMilliseconds, + logger: this.settings.logger, + }) + return this.wakeUpSelection + } + + private async closeWakeUp(): Promise { + const selection = this.wakeUpSelection + if (!selection) return + const selected = await selection.catch(() => undefined) + await selected?.adapter.close() + } + + private async discardWakeUp(): Promise { + await this.closeWakeUp() + this.wakeUpSelection = undefined + this.wakeUpSelectionFailureLogged = false + } + private wakeUp(role: WakeUpRole): void { - notifyWakeUp({ adapter: this.settings.wakeUp, logger: this.settings.logger, role }) + void this.notifyWhenSelected(role) + } + + private async notifyWhenSelected(role: WakeUpRole): Promise { + try { + const selected = await this.resolveWakeUp() + notifyWakeUp({ adapter: selected.adapter, logger: this.settings.logger, role }) + } catch (error) { + this.reportWakeUpSelectionFailure(error instanceof Error ? error.name : "Error") + } + } + + private reportWakeUpSelectionFailure(errorName: string): void { + if (this.wakeUpSelectionFailureLogged) return + this.wakeUpSelectionFailureLogged = true + this.settings.logger.error({ event: "solid_objects.wake_up.selection_failed", errorName }) } private async authorize(options: { diff --git a/src/wake-up-selection.ts b/src/wake-up-selection.ts new file mode 100644 index 0000000..404d2fc --- /dev/null +++ b/src/wake-up-selection.ts @@ -0,0 +1,253 @@ +import type { Database } from "./database/types.js" +import type { Logger } from "./types.js" +import { + InProcessWakeUpAdapter, + WAKE_UP_NAMES, + type NotificationWakeUpAdapter, + type WakeUpAdapter, + type WakeUpCapability, + type WakeUpSetting, +} from "./wake-up.js" + +export const REDIS_URL_VARIABLE = "SOLID_OBJECTS_REDIS_URL" +export const POSTGRESQL_FLOOR_MILLISECONDS = 2.9 +export const REDIS_FLOOR_MILLISECONDS = 5.7 +export const PROBE_TIMEOUT_MILLISECONDS = 2_000 +export const PROBE_CHANNEL_PREFIX = "solid_objects_probe" + +const PROBE_ROLE = "actors" as const + +type NotificationAdapterFactory = (options?: { + channelPrefix?: string +}) => NotificationWakeUpAdapter + +export interface SelectedWakeUp { + readonly adapter: WakeUpAdapter + readonly capability: WakeUpCapability +} + +export interface WakeUpSelectionOptions { + setting: WakeUpSetting + database: Database + idlePollingIntervalMilliseconds: number + logger: Logger + redisUrl?: string + probeTimeoutMilliseconds?: number +} + +export async function selectWakeUp(options: WakeUpSelectionOptions): Promise { + const { setting } = options + if (typeof setting !== "string") return configured(setting) + if (setting === "automatic") return await automatic(options) + return await named({ ...options, name: setting }) +} + +function configured(adapter: WakeUpAdapter): SelectedWakeUp { + if (adapter.defaultCapability) return { adapter, capability: adapter.defaultCapability } + return { + adapter, + capability: { + adapter: "configured", + crossesProcesses: true, + reason: "an adapter was configured, so selection did not run", + }, + } +} + +async function named(options: WakeUpSelectionOptions & { name: string }): Promise { + if (options.name === "in_process") { + return { + adapter: new InProcessWakeUpAdapter(), + capability: { + adapter: "in_process", + crossesProcesses: false, + reason: "in-process signalling was requested", + }, + } + } + if (options.name === "postgresql") return requestedPostgresql(options) + if (options.name === "redis") return await requestedRedis(options) + + throw new TypeError( + `unknown wakeUp ${JSON.stringify(options.name)}, expected one of ${WAKE_UP_NAMES.join(", ")} or an adapter`, + ) +} + +function requestedPostgresql(options: WakeUpSelectionOptions): SelectedWakeUp { + const wakeUp = notificationFactory(options.database) + if (!wakeUp) { + return unavailable({ + options, + reason: + `wakeUp "postgresql" needs a database with a notification channel, ` + + `and ${options.database.family} provides none`, + }) + } + return { + adapter: wakeUp(), + capability: { + adapter: "postgresql_notify", + crossesProcesses: true, + measuredFloorMilliseconds: POSTGRESQL_FLOOR_MILLISECONDS, + reason: "PostgreSQL LISTEN was requested", + }, + } +} + +async function requestedRedis(options: WakeUpSelectionOptions): Promise { + const url = redisUrl(options) + if (url === undefined) { + return unavailable({ + options, + reason: `wakeUp "redis" needs ${REDIS_URL_VARIABLE}, which is not set`, + }) + } + return await redisSelection({ options, url, reason: "Redis was requested" }) +} + +async function automatic(options: WakeUpSelectionOptions): Promise { + const url = redisUrl(options) + if (url !== undefined) { + return await redisSelection({ + options, + url, + reason: `${REDIS_URL_VARIABLE} is set, so Redis carries the signal between processes`, + }) + } + const wakeUp = notificationFactory(options.database) + if (options.database.family !== "postgresql" || !wakeUp) { + return polling({ + idlePollingIntervalMilliseconds: options.idlePollingIntervalMilliseconds, + reason: `${options.database.family} has no notification channel and ${REDIS_URL_VARIABLE} is not set`, + }) + } + return await postgresqlSelection({ options, wakeUp }) +} + +async function postgresqlSelection(input: { + options: WakeUpSelectionOptions + wakeUp: NotificationAdapterFactory +}): Promise { + if (!(await notificationsDeliver(input))) return pooled(input.options) + return { + adapter: input.wakeUp(), + capability: { + adapter: "postgresql_notify", + crossesProcesses: true, + measuredFloorMilliseconds: POSTGRESQL_FLOOR_MILLISECONDS, + reason: + "a probe notification arrived, so PostgreSQL LISTEN carries the signal between processes", + }, + } +} + +async function notificationsDeliver(input: { + options: WakeUpSelectionOptions + wakeUp: NotificationAdapterFactory +}): Promise { + const probe = input.wakeUp({ channelPrefix: PROBE_CHANNEL_PREFIX }) + try { + const watch = await probe.watch(PROBE_ROLE) + await input.options.database.connection((connection) => + connection.run("SELECT pg_notify(?, ?)", [probe.channelFor(PROBE_ROLE), PROBE_ROLE]), + ) + const timeoutMilliseconds = input.options.probeTimeoutMilliseconds ?? PROBE_TIMEOUT_MILLISECONDS + return (await watch.wait({ timeoutMilliseconds })) === true + } catch { + return false + } finally { + await closeQuietly(probe) + } +} + +function pooled(options: WakeUpSelectionOptions): SelectedWakeUp { + options.logger.warn({ + event: "solid_objects.wake_up.pooled_session", + reason: + "PostgreSQL notifications were not selected because a probe notification did not arrive", + }) + return polling({ + idlePollingIntervalMilliseconds: options.idlePollingIntervalMilliseconds, + reason: + "a probe notification did not arrive, so LISTEN cannot carry the signal between processes; " + + "a transaction pooler such as PgBouncer is the usual cause", + }) +} + +function unavailable(input: { options: WakeUpSelectionOptions; reason: string }): SelectedWakeUp { + input.options.logger.warn({ event: "solid_objects.wake_up.unavailable", reason: input.reason }) + return polling({ + idlePollingIntervalMilliseconds: input.options.idlePollingIntervalMilliseconds, + reason: input.reason, + }) +} + +function polling(input: { + idlePollingIntervalMilliseconds: number + reason: string +}): SelectedWakeUp { + return { + adapter: new InProcessWakeUpAdapter(), + capability: { + adapter: "polling", + crossesProcesses: false, + measuredFloorMilliseconds: input.idlePollingIntervalMilliseconds, + reason: input.reason, + }, + } +} + +async function redisSelection(input: { + options: WakeUpSelectionOptions + url: string + reason: string +}): Promise { + const redis = await importRedis() + if (!redis) { + return unavailable({ + options: input.options, + reason: `${REDIS_URL_VARIABLE} is set, and the redis package is not installed`, + }) + } + return { + adapter: new redis.RedisWakeUpAdapter({ url: input.url }), + capability: { + adapter: "redis", + crossesProcesses: true, + measuredFloorMilliseconds: REDIS_FLOOR_MILLISECONDS, + reason: input.reason, + }, + } +} + +async function importRedis(): Promise { + try { + return await import("./wake-up/redis.js") + } catch { + return undefined + } +} + +function notificationFactory(database: Database): NotificationAdapterFactory | undefined { + const wakeUp = database.wakeUp + if (!wakeUp) return undefined + return (options) => wakeUp.call(database, options) +} + +function redisUrl(options: WakeUpSelectionOptions): string | undefined { + const value = options.redisUrl ?? environmentRedisUrl() + if (value === undefined || value.length === 0) return undefined + return value +} + +function environmentRedisUrl(): string | undefined { + return globalThis.process?.env?.[REDIS_URL_VARIABLE] +} + +async function closeQuietly(adapter: WakeUpAdapter): Promise { + try { + await adapter.close() + } catch { + return + } +} diff --git a/src/wake-up.ts b/src/wake-up.ts index fc451f5..8acbecc 100644 --- a/src/wake-up.ts +++ b/src/wake-up.ts @@ -9,13 +9,40 @@ export interface WakeUpWatch { wait(options: WakeUpWaitOptions): Promise } +export type WakeUpAdapterName = + "in_process" | "polling" | "postgresql_notify" | "redis" | "configured" + +export interface WakeUpCapability { + readonly adapter: WakeUpAdapterName + readonly crossesProcesses: boolean + readonly measuredFloorMilliseconds?: number + readonly reason: string +} + +export const WAKE_UP_NAMES = ["automatic", "in_process", "postgresql", "redis"] as const + +export type WakeUpName = (typeof WAKE_UP_NAMES)[number] + export interface WakeUpAdapter { watch(role: WakeUpRole): WakeUpWatch | Promise notify(role: WakeUpRole): void | Promise close(): void | Promise + readonly defaultCapability?: WakeUpCapability } +export interface NotificationWakeUpAdapter extends WakeUpAdapter { + channelFor(role: WakeUpRole): string +} + +export type WakeUpSetting = WakeUpName | WakeUpAdapter + export class InProcessWakeUpAdapter implements WakeUpAdapter { + readonly defaultCapability: WakeUpCapability = { + adapter: "in_process", + crossesProcesses: false, + reason: "in-process signalling, which a commit in another process cannot reach", + } + private readonly generations = new Map() private readonly waiters = new Map void>>() private closed = false diff --git a/src/wake-up/postgresql.ts b/src/wake-up/postgresql.ts index 14f297b..5e6232c 100644 --- a/src/wake-up/postgresql.ts +++ b/src/wake-up/postgresql.ts @@ -1,5 +1,10 @@ import { Client, type ClientConfig, type Notification } from "pg" -import type { WakeUpAdapter, WakeUpRole, WakeUpWatch } from "../wake-up.js" +import type { + NotificationWakeUpAdapter, + WakeUpCapability, + WakeUpRole, + WakeUpWatch, +} from "../wake-up.js" const ROLES = [ "actors", @@ -20,7 +25,14 @@ export interface PostgreSQLWakeUpOptions { onListenerError?: (failure: PostgreSQLWakeUpFailure) => void } -export class PostgreSQLWakeUpAdapter implements WakeUpAdapter { +export class PostgreSQLWakeUpAdapter implements NotificationWakeUpAdapter { + readonly defaultCapability: WakeUpCapability = { + adapter: "postgresql_notify", + crossesProcesses: true, + measuredFloorMilliseconds: 2.9, + reason: "PostgreSQL LISTEN carries the signal between processes", + } + private readonly clientConfiguration: ClientConfig private readonly channels = new Map() private readonly rolesByChannel = new Map() @@ -65,6 +77,10 @@ export class PostgreSQLWakeUpAdapter implements WakeUpAdapter { })) } + channelFor(role: WakeUpRole): string { + return this.channel(role) + } + async watch(role: WakeUpRole): Promise { await this.ensureListening(role) const generation = this.generation(role) diff --git a/src/wake-up/redis.ts b/src/wake-up/redis.ts index 82ddd6d..15da70f 100644 --- a/src/wake-up/redis.ts +++ b/src/wake-up/redis.ts @@ -2,6 +2,7 @@ import { createClient, type RedisClientType } from "redis" import { InProcessWakeUpAdapter, type WakeUpAdapter, + type WakeUpCapability, type WakeUpRole, type WakeUpWatch, } from "../wake-up.js" @@ -26,6 +27,13 @@ export interface RedisWakeUpOptions { } export class RedisWakeUpAdapter implements WakeUpAdapter { + readonly defaultCapability: WakeUpCapability = { + adapter: "redis", + crossesProcesses: true, + measuredFloorMilliseconds: 5.7, + reason: "Redis carries the signal between processes", + } + private readonly local = new InProcessWakeUpAdapter() private readonly publisher: RedisClientType private readonly subscriber: RedisClientType diff --git a/src/worker.ts b/src/worker.ts index c1e5e0a..fbc68e8 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -141,7 +141,7 @@ export class Worker { await this.ensureRegistered() await this.runtime.warnIfPollingIsOnlyCrossProcessWakeUp() while (!signal.aborted && !this.stopping) { - const wakeUp = await this.runtime.settings.wakeUp.watch("actors") + const wakeUp = await this.runtime.watchWakeUp("actors") const processed = await this.runOnce() if (processed > 0) { this.pollingBackoff.reset("work") diff --git a/test/effect-recovery.test.ts b/test/effect-recovery.test.ts index 23202c0..e8a5c55 100644 --- a/test/effect-recovery.test.ts +++ b/test/effect-recovery.test.ts @@ -134,7 +134,7 @@ it("retires abandoned processing effects before the scheduler can reclaim them", ), ) await runtime.repository.registerProcess("replacement", "effect") - const wakeUp = await runtime.settings.wakeUp.watch("actors") + const wakeUp = await runtime.watchWakeUp("actors") expect(await runtime.repository.claimEffect("replacement")).toBeUndefined() expect(await wakeUp.wait({ timeoutMilliseconds: 0 })).toBe(true) const notifications = await runtime.settings.database.connection((connection) => @@ -237,6 +237,7 @@ it("rolls back retirement if the second mailbox insert fails", async () => { const effect = await processingEffect() await ageOwner(70_000) const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime!.settings, enqueue: (connection, input) => { if (input.operation === "inspect") throw new Error("injected second insert failure") @@ -267,6 +268,7 @@ it("reports missing from the owned binding without exposing another actor", asyn await runtime!.ref(ReportExport, "export").check() expect(await messages("inspect")).toEqual([{ effectId: effect.id, outcome: "missing" }]) const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime!.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -286,6 +288,7 @@ it("rolls back both callbacks when only one mailbox slot remains", async () => { await ageOwner(70_000) runtime!.settings.maxMailboxLength = 1 const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime!.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -308,6 +311,7 @@ it("surfaces an owner query failure without deciding abandonment", async () => { const effect = await processingEffect() await ageOwner(70_000) const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime!.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -362,6 +366,7 @@ it("requires recovery opt-in for a timeout and both bindings for an explicit che connection.get(`SELECT * FROM ${runtime!.repository.table("effects")}`), ) const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -450,6 +455,7 @@ it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( const effect = await processingEffect() await ageOwner(70_000) const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime!.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -480,6 +486,7 @@ it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( const effect = await processingEffect() await ageOwner(70_000) const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime!.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -511,6 +518,7 @@ it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( const effect = await processingEffect() await ageOwner(70_000) const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime!.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -620,10 +628,14 @@ it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( database: runtime.settings.database, table: "effects", }) - const claimant = new Repository({ ...runtime.settings, database: pausedDatabase }) + const claimant = new Repository( + { ...runtime.settings, database: pausedDatabase }, + { wakeUpAdapter: () => runtime!.wakeUpAdapter() }, + ) const claim = claimant.claimEffect("owner") const originLocked = deferred() const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -659,6 +671,7 @@ it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( const effect = await storedEffect(handle!.id) await runtime.repository.registerProcess("owner", "effect") const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) @@ -683,6 +696,7 @@ it.skipIf(!process.env.SOLID_OBJECTS_DATABASE_URL?.startsWith("postgresql:"))( const effect = await processingEffect() await ageOwner(70_000) const coordinator = new EffectRecoveryCoordinator({ + wakeUpAdapter: () => runtime!.wakeUpAdapter(), settings: runtime!.settings, enqueue: (connection, input) => runtime!.repository.enqueueInTransaction(connection, input), }) diff --git a/test/polling-loop.test.ts b/test/polling-loop.test.ts index 91f7d86..2b3918d 100644 --- a/test/polling-loop.test.ts +++ b/test/polling-loop.test.ts @@ -2,7 +2,12 @@ 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" +import { + InProcessWakeUpAdapter, + type WakeUpAdapter, + type WakeUpRole, + type WakeUpWatch, +} from "../src/wake-up.js" let runtime: SolidObjectsRuntime | undefined @@ -365,6 +370,39 @@ describe("idle polling", () => { expect(logger.warn).not.toHaveBeenCalled() }) + + it("warns when the configured adapter reports that it stays in one process", 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 InProcessWakeUpAdapter(), + }) + 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).toHaveBeenCalledWith( + expect.objectContaining({ event: "solid_objects.polling_only_cross_process_wake_up" }), + ) + }) }) class ImmediateTimeoutWakeUpAdapter implements WakeUpAdapter { diff --git a/test/postgresql.test.ts b/test/postgresql.test.ts index 4eb1b96..19b6f90 100644 --- a/test/postgresql.test.ts +++ b/test/postgresql.test.ts @@ -729,6 +729,41 @@ describePostgreSQL("PostgreSQL adapter", () => { expect(dashboardResponse.status).toBe(200) expect(await dashboardResponse.text()).toContain("flow") }) + + it("selects PostgreSQL notifications after a probe notification arrives", async () => { + if (!connectionString) throw new Error("PostgreSQL connection string is required") + database = postgresql({ connectionString }) + runtime = configure({ + database, + tableNamePrefix: "postgresql_test_", + authorizeMessage: () => true, + logger: quietLogger, + }) + + const capability = await runtime.wakeUpCapability() + const adapter = await runtime.wakeUpAdapter() + + expect(capability.adapter).toBe("postgresql_notify") + expect(capability.crossesProcesses).toBe(true) + expect(capability.reason).toMatch(/probe notification arrived/i) + expect(adapter).toBeInstanceOf(PostgreSQLWakeUpAdapter) + }) + + it("wakes a listener from a second connection, which a pooled session could not", async () => { + if (!connectionString) throw new Error("PostgreSQL connection string is required") + database = postgresql({ connectionString }) + const listener = database.wakeUp({ channelPrefix: "postgresql_probe_test" }) + try { + const watch = await listener.watch("actors") + await database.connection((connection) => + connection.run("SELECT pg_notify(?, ?)", [listener.channelFor("actors"), "actors"]), + ) + + expect(await watch.wait({ timeoutMilliseconds: 2_000 })).toBe(true) + } finally { + await listener.close() + } + }) }) function dashboardContext() { diff --git a/test/wake-up-selection.test.ts b/test/wake-up-selection.test.ts new file mode 100644 index 0000000..7abee14 --- /dev/null +++ b/test/wake-up-selection.test.ts @@ -0,0 +1,465 @@ +import { afterEach, describe, expect, it, vi } from "vitest" +import { Actor } from "../src/actor.js" +import { sqlite } from "../src/database/sqlite.js" +import type { + Database, + DatabaseConnection, + DatabaseFamily, + DatabaseTransactionOptions, +} from "../src/database/types.js" +import { configure, type SolidObjectsRuntime } from "../src/runtime.js" +import { + InProcessWakeUpAdapter, + type NotificationWakeUpAdapter, + type WakeUpAdapter, + type WakeUpRole, + type WakeUpSetting, + type WakeUpWatch, +} from "../src/wake-up.js" +import { selectWakeUp } from "../src/wake-up-selection.js" + +const silentLogger = { + debug: () => undefined, + info: () => undefined, + warn: () => undefined, + error: () => undefined, +} + +class ProbeWakeUpAdapter implements NotificationWakeUpAdapter { + closed = false + readonly watchedRoles: WakeUpRole[] = [] + + constructor(private readonly options: { delivers: boolean; channelPrefix: string }) {} + + channelFor(role: WakeUpRole): string { + return `${this.options.channelPrefix}_${role}` + } + + watch(role: WakeUpRole): WakeUpWatch { + this.watchedRoles.push(role) + return { wait: () => Promise.resolve(this.options.delivers) } + } + + notify(): void {} + + close(): void { + this.closed = true + } +} + +class NotifyingDatabase implements Database { + readonly family: DatabaseFamily + readonly schemaIdentity: string + readonly adapters: ProbeWakeUpAdapter[] = [] + readonly notifiedChannels: string[] = [] + + constructor( + private readonly options: { + database: Database + delivers: boolean + refuses?: boolean + unreachable?: boolean + family?: DatabaseFamily + }, + ) { + this.family = options.family ?? "postgresql" + this.schemaIdentity = options.database.schemaIdentity + } + + wakeUp(options: { channelPrefix?: string } = {}): NotificationWakeUpAdapter { + if (this.options.refuses) throw new Error("this database refuses a notification adapter") + const adapter = new ProbeWakeUpAdapter({ + delivers: this.options.delivers, + channelPrefix: options.channelPrefix ?? "solid_objects", + }) + this.adapters.push(adapter) + return adapter + } + + connection( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + if (this.options.unreachable) return Promise.reject(new Error("the database is unreachable")) + return this.options.database.connection((connection) => callback(this.intercept(connection))) + } + + transaction( + callback: (connection: DatabaseConnection) => Promise, + options?: DatabaseTransactionOptions, + ): Promise { + return this.options.database.transaction(callback, options) + } + + close(): Promise { + return this.options.database.close() + } + + private intercept(connection: DatabaseConnection): DatabaseConnection { + return { + run: (sql, parameters = []) => { + if (!sql.includes("pg_notify")) return connection.run(sql, parameters) + this.notifiedChannels.push(String(parameters[0])) + return Promise.resolve({ changes: 0 }) + }, + get: (sql, parameters) => connection.get(sql, parameters), + all: (sql, parameters) => connection.all(sql, parameters), + nowMilliseconds: () => connection.nowMilliseconds(), + } + } +} + +class CustomWakeUpAdapter implements WakeUpAdapter { + watch(): WakeUpWatch { + return { wait: () => Promise.resolve(false) } + } + + notify(): void {} + + close(): void {} +} + +class SelectionCounter extends Actor { + static override readonly actorType = "WakeUpSelectionCounter" + + count = 0 + + increment(): void { + this.count += 1 + } +} + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + await runtime?.close() + runtime = undefined + delete process.env["SOLID_OBJECTS_REDIS_URL"] +}) + +function selectionOptions(options: { database: Database; setting?: WakeUpSetting }) { + return { + setting: options.setting ?? "automatic", + database: options.database, + idlePollingIntervalMilliseconds: 1_000, + logger: silentLogger, + } +} + +describe("wake-up selection", () => { + it("keeps an explicitly configured adapter", async () => { + const configured = new InProcessWakeUpAdapter() + const database = sqlite({ path: ":memory:" }) + + const selected = await selectWakeUp(selectionOptions({ database, setting: configured })) + + expect(selected.adapter).toBe(configured) + await database.close() + }) + + it("keeps the capability a configured adapter reports about itself", async () => { + const database = sqlite({ path: ":memory:" }) + + const selected = await selectWakeUp( + selectionOptions({ database, setting: new InProcessWakeUpAdapter() }), + ) + + expect(selected.capability.adapter).toBe("in_process") + expect(selected.capability.crossesProcesses).toBe(false) + await database.close() + }) + + it("records a configured adapter that reports no capability as configured", async () => { + const database = sqlite({ path: ":memory:" }) + + const selected = await selectWakeUp( + selectionOptions({ database, setting: new CustomWakeUpAdapter() }), + ) + + expect(selected.capability.adapter).toBe("configured") + expect(selected.capability.crossesProcesses).toBe(true) + await database.close() + }) + + it("selects where the runtime defines no process global, as a browser does not", async () => { + const database = sqlite({ path: ":memory:" }) + vi.stubGlobal("process", undefined) + try { + const selected = await selectWakeUp(selectionOptions({ database })) + + expect(selected.capability.adapter).toBe("polling") + } finally { + vi.unstubAllGlobals() + await database.close() + } + }) + + it("opts out of selection when in_process is requested", async () => { + const database = sqlite({ path: ":memory:" }) + + const selected = await selectWakeUp(selectionOptions({ database, setting: "in_process" })) + + expect(selected.capability.adapter).toBe("in_process") + expect(selected.capability.crossesProcesses).toBe(false) + expect(selected.capability.reason).toMatch(/requested/i) + await database.close() + }) + + it("selects redis on any database when a url is set", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: true }) + process.env["SOLID_OBJECTS_REDIS_URL"] = "redis://127.0.0.1:6379/15" + + const selected = await selectWakeUp(selectionOptions({ database })) + + expect(selected.capability.adapter).toBe("redis") + expect(selected.capability.crossesProcesses).toBe(true) + expect(database.adapters).toHaveLength(0) + await selected.adapter.close() + await database.close() + }) + + it("polls when the database cannot answer the probe", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: true, unreachable: true }) + + const selected = await selectWakeUp(selectionOptions({ database })) + + expect(selected.capability.adapter).toBe("polling") + expect(selected.capability.crossesProcesses).toBe(false) + await database.close() + }) + + it("polls and warns when a requested adapter has no notification channel", async () => { + const database = sqlite({ path: ":memory:" }) + const warnings: { event?: string }[] = [] + const logger = { ...silentLogger, warn: (entry: { event?: string }) => warnings.push(entry) } + + const selected = await selectWakeUp({ + ...selectionOptions({ database, setting: "postgresql" }), + logger, + }) + + expect(selected.capability.adapter).toBe("polling") + expect(selected.capability.reason).toMatch(/notification channel/i) + expect(warnings.map(({ event }) => event)).toEqual(["solid_objects.wake_up.unavailable"]) + await database.close() + }) + + it("selects a requested name without probing", async () => { + const database = sqlite({ path: ":memory:" }) + process.env["SOLID_OBJECTS_REDIS_URL"] = "redis://127.0.0.1:6379/15" + + const selected = await selectWakeUp(selectionOptions({ database, setting: "redis" })) + + expect(selected.capability.adapter).toBe("redis") + expect(selected.capability.crossesProcesses).toBe(true) + expect(selected.capability.reason).toMatch(/requested/i) + await selected.adapter.close() + await database.close() + }) + + it("polls and warns when redis is requested without a url", async () => { + const database = sqlite({ path: ":memory:" }) + const warnings: { event?: string }[] = [] + const logger = { ...silentLogger, warn: (entry: { event?: string }) => warnings.push(entry) } + + const selected = await selectWakeUp({ + ...selectionOptions({ database, setting: "redis" }), + logger, + }) + + expect(selected.capability.adapter).toBe("polling") + expect(selected.capability.reason).toMatch(/SOLID_OBJECTS_REDIS_URL/) + expect(warnings.map(({ event }) => event)).toEqual(["solid_objects.wake_up.unavailable"]) + await database.close() + }) + + it("refuses a configured adapter that cannot notify", () => { + const adapter: WakeUpAdapter = { + watch: () => ({ wait: () => Promise.resolve(false) }), + notify: () => undefined, + close: () => undefined, + } + Reflect.deleteProperty(adapter, "notify") + + expect(() => configure({ database: sqlite({ path: ":memory:" }), wakeUp: adapter })).toThrow( + /must implement notify/, + ) + }) + + it("refuses an unknown name when the configuration is built", () => { + expect(() => + configure({ + database: sqlite({ path: ":memory:" }), + wakeUp: "carrier_pigeon" as WakeUpSetting, + }), + ).toThrow(/carrier_pigeon/) + }) + + it("refuses an unknown name rather than polling quietly", async () => { + const database = sqlite({ path: ":memory:" }) + + await expect( + selectWakeUp({ + ...selectionOptions({ database }), + setting: "carrier_pigeon" as WakeUpSetting, + }), + ).rejects.toThrow(/carrier_pigeon/) + await database.close() + }) + + it("polls on a database without a notification channel", async () => { + const database = sqlite({ path: ":memory:" }) + + const selected = await selectWakeUp(selectionOptions({ database })) + + expect(selected.capability.adapter).toBe("polling") + expect(selected.capability.crossesProcesses).toBe(false) + expect(selected.capability.measuredFloorMilliseconds).toBe(1_000) + expect(selected.capability.reason).toMatch(/no notification channel/i) + expect(selected.adapter.defaultCapability?.adapter).toBe("in_process") + await database.close() + }) + + it("selects PostgreSQL notifications when a probe notification arrives", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: true }) + + const selected = await selectWakeUp(selectionOptions({ database })) + + expect(selected.capability.adapter).toBe("postgresql_notify") + expect(selected.capability.crossesProcesses).toBe(true) + expect(selected.capability.reason).toMatch(/probe notification arrived/i) + await database.close() + }) + + it("notifies the probe channel from the database rather than from the listener", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: true }) + + await selectWakeUp(selectionOptions({ database })) + + expect(database.notifiedChannels).toHaveLength(1) + expect(database.notifiedChannels[0]).not.toBe("solid_objects_actors") + expect(database.adapters[0]?.closed).toBe(true) + await database.close() + }) + + it("polls when a probe notification does not arrive", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: false }) + + const selected = await selectWakeUp(selectionOptions({ database })) + + expect(selected.capability.adapter).toBe("polling") + expect(selected.capability.crossesProcesses).toBe(false) + expect(selected.capability.reason).toMatch(/pooler/i) + await database.close() + }) + + it("warns once about a pooled PostgreSQL session", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: false }) + const warnings: { event?: string }[] = [] + const logger = { ...silentLogger, warn: (entry: { event?: string }) => warnings.push(entry) } + + await selectWakeUp({ ...selectionOptions({ database }), logger }) + + expect(warnings).toHaveLength(1) + await database.close() + }) +}) + +describe("runtime wake-up selection", () => { + it("selects once when callers race for the adapter", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: true }) + runtime = configure({ database, authorizeMessage: () => true }) + + const selections = await Promise.all(Array.from({ length: 8 }, () => runtime?.wakeUpAdapter())) + + expect(database.adapters).toHaveLength(2) + expect(new Set(selections).size).toBe(1) + }) + + it("does not select again after the runtime closes", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: true }) + const closing = configure({ database, authorizeMessage: () => true }) + const selected = await closing.wakeUpAdapter() + await closing.close() + + expect(await closing.wakeUpAdapter()).toBe(selected) + expect(database.adapters).toHaveLength(2) + }) + + it("reports a selection that cannot run once rather than on every commit", async () => { + const errors: { event?: string }[] = [] + const logger = { ...silentLogger, error: (entry: { event?: string }) => errors.push(entry) } + const inner = sqlite({ path: ":memory:" }) + runtime = configure({ + database: new NotifyingDatabase({ + database: inner, + delivers: true, + refuses: true, + family: "sqlite", + }), + wakeUp: "postgresql", + logger, + authorizeMessage: () => true, + }) + runtime.register(SelectionCounter) + await runtime.install() + + await runtime.ref(SelectionCounter, "one").send.increment() + await runtime.ref(SelectionCounter, "one").send.increment() + await vi.waitFor(() => + expect(errors.some(({ event }) => event === "solid_objects.wake_up.selection_failed")).toBe( + true, + ), + ) + + expect( + errors.filter(({ event }) => event === "solid_objects.wake_up.selection_failed"), + ).toHaveLength(1) + }) + + it("reports the capability of the adapter that is installed", async () => { + runtime = configure({ + database: sqlite({ path: ":memory:" }), + authorizeMessage: () => true, + }) + + const capability = await runtime.wakeUpCapability() + + expect(capability.adapter).toBe("polling") + expect(capability.crossesProcesses).toBe(false) + }) + + it("warns in the doctor about a configured in-process adapter", async () => { + runtime = configure({ + database: sqlite({ path: ":memory:" }), + authorizeAdministration: () => true, + wakeUp: new InProcessWakeUpAdapter(), + }) + await runtime.install() + + const report = await runtime.doctor.run({ roundTrip: "skip" }) + const wakeUp = report.checks.find(({ name }) => name === "wakeUp") + + expect(wakeUp?.status).toBe("warn") + expect(wakeUp?.message).toMatch(/in_process/) + }) + + it("passes the doctor when the installed adapter crosses processes", async () => { + const inner = sqlite({ path: ":memory:" }) + const database = new NotifyingDatabase({ database: inner, delivers: true }) + runtime = configure({ database, authorizeAdministration: () => true }) + + const report = await runtime.doctor.run({ roundTrip: "skip" }) + const wakeUp = report.checks.find(({ name }) => name === "wakeUp") + + expect(wakeUp?.status).toBe("pass") + expect(wakeUp?.message).toMatch(/postgresql_notify/) + }) +})