From b81c879bdf2fb91e3e6f1b7c31f294636ad78d9c Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Thu, 3 Sep 2026 07:56:21 -0700 Subject: [PATCH 1/3] fix: optimize polling claims Install canonical ordered polling indexes and use adapter-aware row locks for outbox claims. Split mixed recovery paths and scope SQL claim transactions to read committed so concurrent MySQL workers do not serialize behind gap locks. --- CHANGELOG.md | 18 ++ docs/api.md | 8 +- docs/parity.md | 18 +- package.json | 2 +- src/application-database.ts | 5 +- src/database/mysql.ts | 11 +- src/database/postgresql.ts | 14 +- src/database/types.ts | 5 + src/doctor.ts | 4 +- src/index.ts | 8 +- src/repository.ts | 346 +++++++++++++++++-------- src/schema.ts | 45 +++- src/version.ts | 2 +- test/dead-letters.test.ts | 58 ++++- test/doctor.test.ts | 2 +- test/mysql.test.ts | 123 +++++++++ test/outboxes.test.ts | 44 ++++ test/polling-queries.test.ts | 111 ++++++++ test/postgresql.test.ts | 43 +++ test/support/pausing-claim-database.ts | 100 +++++++ 20 files changed, 833 insertions(+), 134 deletions(-) create mode 100644 test/polling-queries.test.ts create mode 100644 test/support/pausing-claim-database.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 069a3e5..9972af4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,23 @@ # Changelog +## 0.14.6 - 2026-09-03 + +- Poll effects, reminders, and broadcasts through ordered indexes installed by + schema migration 8. PostgreSQL and MySQL claim one row with + `FOR UPDATE SKIP LOCKED`; SQLite keeps its serialized transaction path. The + broadcast revision guard has its own `(instance_id, state_revision, status)` + index instead of rescanning the outbox for every candidate. Claim transactions + use read committed isolation on PostgreSQL and MySQL so preceding recovery + work cannot turn row skipping into InnoDB gap-lock contention. +- Replace bulk broadcast and reminder recovery updates with separate available + and stale probes, then claim the oldest locked candidate across each pair. + This preserves global delivery order, per-instance broadcast revision order, + stale-process recovery, and at-least-once delivery. +- On 50,000 production-shaped rows, SQLite, PostgreSQL 18, and MySQL 8.4 all + move from scans and explicit sorts to ordered index probes. The worst SQLite + broadcast probe fell from 944 ms to 0.018 ms; PostgreSQL's probes finish in + 0.02-0.09 ms and MySQL's in 0.10-1.2 ms. + ## 0.14.5 - 2026-08-29 - Record the Ruby state warning in the parity ledger. The row said the Ruby gem diff --git a/docs/api.md b/docs/api.md index d472749..ae42d42 100644 --- a/docs/api.md +++ b/docs/api.md @@ -206,8 +206,12 @@ rows for `preview()` and rows actually deleted for `prune()`. `InstrumentationEvent` type the host integration contract. `JsonPrimitive`, `JsonValue`, `JsonObject`, `DeepReadonly`, `ActorIdentifier`, `MessageContext`, `MessageStatus`, and `Logger` are shared types. `Database`, -`DatabaseConnection`, `DatabaseFamily`, and `RunResult` support custom database -and commit-action integration. +`DatabaseConnection`, `DatabaseFamily`, `DatabaseTransactionOptions`, and +`RunResult` support custom database and commit-action integration. A custom +PostgreSQL or MySQL adapter must honor +`transaction(callback, { isolationLevel: "read_committed" })`; outbox claiming +uses that isolation with row locks to avoid InnoDB gap-lock contention. SQLite +adapters may treat the option as their ordinary serialized transaction. `BroadcastEvent.observables` contains changed value-broadcast projections. `BroadcastEvent.invalidations` contains changed invalidation-only names. The diff --git a/docs/parity.md b/docs/parity.md index 1a6b9a7..1e6df05 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -80,15 +80,15 @@ such boundary between a gem and its dependents. ## Databases and wake-up -| Capability | Status | TypeScript shape or remaining work | -| ------------------------ | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| SQLite | Native | Uses built-in `node:sqlite`, serialized process-local access, bounded transient writer retries, foreign keys, strict tables, database time, and deadline-bounded access and lock waits. | -| PostgreSQL | Native | Optional `pg` 8.23 peer, bounded pooling, 64-bit schema, row-locked sequences, server checks, and deadline-bounded pool, statement, and lock waits. | -| MySQL | Native | Optional `mysql2` 3.23 peer, bounded pooling, InnoDB schema, row-locked sequences, scoped deadlock retry, and deadline-bounded pool, query, and lock waits. Ruby also tests a second client, `trilogy`; Node has no comparable second MySQL client, so only `mysql2` is tracked here. | -| Durable polling fallback | Native | Every role progresses without a notification service. | -| In-process wake-up | Native | A generation-based default adapter prevents claim-to-wait signal loss; commits wake role-specific waiters and polling remains the fallback. | -| 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. | +| Capability | Status | TypeScript shape or remaining work | +| ------------------------ | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| SQLite | Native | Uses built-in `node:sqlite`, serialized process-local access, bounded transient writer retries, foreign keys, strict tables, database time, and deadline-bounded access and lock waits. | +| PostgreSQL | Native | Optional `pg` 8.23 peer, bounded pooling, 64-bit schema, row-locked sequences, server checks, and deadline-bounded pool, statement, and lock waits. | +| MySQL | Native | Optional `mysql2` 3.23 peer, bounded pooling, InnoDB schema, row-locked sequences, scoped deadlock retry, and deadline-bounded pool, query, and lock waits. Ruby also tests a second client, `trilogy`; Node has no comparable second MySQL client, so only `mysql2` is tracked here. | +| Durable polling fallback | Native | Every role progresses without a notification service. Effects, reminders, and broadcasts use canonical ordered polling indexes; PostgreSQL and MySQL claim with `FOR UPDATE SKIP LOCKED`. Broadcasts and reminders use separate available and stale-recovery probes and preserve the oldest-first choice across each pair. | +| In-process wake-up | Native | A generation-based default adapter prevents claim-to-wait signal loss; commits wake role-specific waiters and polling remains the fallback. | +| 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 diff --git a/package.json b/package.json index bb08605..ae6d664 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "solid-objects", - "version": "0.14.5", + "version": "0.14.6", "description": "Race-free realtime state per application identity, backed by your SQL database", "type": "module", "license": "MIT", diff --git a/src/application-database.ts b/src/application-database.ts index c793423..f4d1bb4 100644 --- a/src/application-database.ts +++ b/src/application-database.ts @@ -1,5 +1,5 @@ import { applicationWritesForbidden } from "./context.js" -import type { Database, DatabaseConnection } from "./database/types.js" +import type { Database, DatabaseConnection, DatabaseTransactionOptions } from "./database/types.js" import { ApplicationWriteForbidden } from "./errors.js" export function guardApplicationDatabase(database: Database): Database { @@ -27,8 +27,9 @@ class GuardedApplicationDatabase implements Database { transaction( callback: (connection: DatabaseConnection) => Promise, + options: DatabaseTransactionOptions = {}, ): Promise { - return this.database.transaction((connection) => callback(guardConnection(connection))) + return this.database.transaction((connection) => callback(guardConnection(connection)), options) } close(): Promise { diff --git a/src/database/mysql.ts b/src/database/mysql.ts index 6210162..e8e0fd5 100644 --- a/src/database/mysql.ts +++ b/src/database/mysql.ts @@ -6,7 +6,12 @@ import mysqlDriver, { type RowDataPacket, } from "mysql2/promise" import type { ExecuteValues } from "mysql2" -import type { Database, DatabaseConnection, RunResult } from "./types.js" +import type { + Database, + DatabaseConnection, + DatabaseTransactionOptions, + RunResult, +} from "./types.js" import { acquireBeforeDatabaseDeadline, databaseDeadlineRemainingMilliseconds, @@ -152,6 +157,7 @@ export class MySQLDatabase implements Database { async transaction( callback: (connection: DatabaseConnection) => Promise, + options: DatabaseTransactionOptions = {}, ): Promise { return withDatabaseTransaction(this, async () => { const deadlineActive = databaseDeadlineRemainingMilliseconds() !== undefined @@ -160,6 +166,9 @@ export class MySQLDatabase implements Database { (lateConnection) => lateConnection.release(), ) try { + if (options.isolationLevel === "read_committed") { + await connection.query("SET TRANSACTION ISOLATION LEVEL READ COMMITTED") + } await connection.beginTransaction() const remaining = requireDatabaseDeadlineRemaining() if (remaining !== undefined) { diff --git a/src/database/postgresql.ts b/src/database/postgresql.ts index 1e57a41..847ebc5 100644 --- a/src/database/postgresql.ts +++ b/src/database/postgresql.ts @@ -1,7 +1,12 @@ import "../platform/node.js" import { Pool, TypeOverrides, types, type PoolClient, type PoolConfig } from "pg" import { postgresqlSql } from "./postgresql-sql.js" -import type { Database, DatabaseConnection, RunResult } from "./types.js" +import type { + Database, + DatabaseConnection, + DatabaseTransactionOptions, + RunResult, +} from "./types.js" import { PostgreSQLWakeUpAdapter, type PostgreSQLWakeUpFailure } from "../wake-up/postgresql.js" import { acquireBeforeDatabaseDeadline, @@ -154,6 +159,7 @@ export class PostgreSQLDatabase implements Database { async transaction( callback: (connection: DatabaseConnection) => Promise, + options: DatabaseTransactionOptions = {}, ): Promise { return withDatabaseTransaction(this, async () => { const deadlineActive = databaseDeadlineRemainingMilliseconds() !== undefined @@ -162,7 +168,11 @@ export class PostgreSQLDatabase implements Database { (lateClient) => lateClient.release(), ) try { - await client.query("BEGIN") + await client.query( + options.isolationLevel === "read_committed" + ? "BEGIN ISOLATION LEVEL READ COMMITTED" + : "BEGIN", + ) const remaining = requireDatabaseDeadlineRemaining() if (remaining !== undefined) { await applyPostgreSQLDeadline({ client, milliseconds: remaining, scope: "transaction" }) diff --git a/src/database/types.ts b/src/database/types.ts index d675922..e8cb383 100644 --- a/src/database/types.ts +++ b/src/database/types.ts @@ -5,6 +5,10 @@ export interface RunResult { lastInsertId?: string } +export interface DatabaseTransactionOptions { + isolationLevel?: "read_committed" +} + export interface DatabaseConnection { run(sql: string, parameters?: readonly unknown[]): Promise get(sql: string, parameters?: readonly unknown[]): Promise @@ -19,6 +23,7 @@ export interface Database { connection(callback: (connection: DatabaseConnection) => Promise): Promise transaction( callback: (connection: DatabaseConnection) => Promise, + options?: DatabaseTransactionOptions, ): Promise close(): Promise } diff --git a/src/doctor.ts b/src/doctor.ts index c000be8..5c4de30 100644 --- a/src/doctor.ts +++ b/src/doctor.ts @@ -202,11 +202,11 @@ export class Doctor { message: `incompatible schema identity ${wrongIdentity.schema_identity}`, }) } - if (versions.join(",") !== "1,2,3,4,5,6,7") { + if (versions.join(",") !== "1,2,3,4,5,6,7,8") { return check({ name: "schema", status: "fail", - message: `expected schema migrations 1, 2, 3, 4, 5, 6; found ${versions.join(", ")}`, + message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8; found ${versions.join(", ")}`, }) } return check({ diff --git a/src/index.ts b/src/index.ts index fe2b8b8..3a28d97 100644 --- a/src/index.ts +++ b/src/index.ts @@ -133,7 +133,13 @@ export type { MessageStatus, SnapshotOptions, } from "./types.js" -export type { Database, DatabaseConnection, DatabaseFamily, RunResult } from "./database/types.js" +export type { + Database, + DatabaseConnection, + DatabaseFamily, + DatabaseTransactionOptions, + RunResult, +} from "./database/types.js" export { ApplicationWriteForbidden, ActorCallCycle, diff --git a/src/repository.ts b/src/repository.ts index 33671b6..a652be1 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1348,42 +1348,45 @@ export class Repository { } async claimEffect(processId: string): Promise { - return this.settings.database.transaction(async (connection) => { - const now = await connection.nowMilliseconds() - const staleAt = now - this.settings.processAliveThresholdMilliseconds - await connection.run( - `UPDATE ${this.table("effects")} SET status = 'pending', claimed_by = NULL - WHERE status = 'processing' AND ( - claimed_by IS NULL OR NOT EXISTS ( - SELECT 1 FROM ${this.table("processes")} processes - WHERE processes.id = ${this.table("effects")}.claimed_by - AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? - ) - )`, - [staleAt], - ) - const effect = await connection.get( - `SELECT effects.*, instances.actor_type, instances.actor_id - FROM ${this.table("effects")} effects - JOIN ${this.table("instances")} instances ON instances.id = effects.instance_id - WHERE effects.status = 'pending' AND effects.available_at_ms <= ? - ORDER BY effects.available_at_ms, effects.id - LIMIT 1`, - [now], - ) - if (!effect) return undefined - const claimed = await connection.run( - `UPDATE ${this.table("effects")} - SET status = 'processing', claimed_by = ?, attempt_count = attempt_count + 1 - WHERE id = ? AND status = 'pending'`, - [processId, effect.id], - ) - if (claimed.changes !== 1) return undefined - effect.status = "processing" - effect.claimed_by = processId - effect.attempt_count = Number(effect.attempt_count) + 1 - return effect - }) + return this.settings.database.transaction( + async (connection) => { + const now = await connection.nowMilliseconds() + const staleAt = now - this.settings.processAliveThresholdMilliseconds + await connection.run( + `UPDATE ${this.table("effects")} SET status = 'pending', claimed_by = NULL + WHERE status = 'processing' AND ( + claimed_by IS NULL OR NOT EXISTS ( + SELECT 1 FROM ${this.table("processes")} processes + WHERE processes.id = ${this.table("effects")}.claimed_by + AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? + ) + )`, + [staleAt], + ) + const effect = await connection.get( + `SELECT effects.*, instances.actor_type, instances.actor_id + FROM ${this.table("effects")} effects + JOIN ${this.table("instances")} instances ON instances.id = effects.instance_id + WHERE effects.status = 'pending' AND effects.available_at_ms <= ? + ORDER BY effects.available_at_ms, effects.id + LIMIT 1${this.claimLockClause()}`, + [now], + ) + if (!effect) return undefined + const claimed = await connection.run( + `UPDATE ${this.table("effects")} + SET status = 'processing', claimed_by = ?, attempt_count = attempt_count + 1 + WHERE id = ? AND status = 'pending'`, + [processId, effect.id], + ) + if (claimed.changes !== 1) return undefined + effect.status = "processing" + effect.claimed_by = processId + effect.attempt_count = Number(effect.attempt_count) + 1 + return effect + }, + { isolationLevel: "read_committed" }, + ) } async completeEffect(effect: EffectRow, result: JsonValue): Promise { @@ -1462,42 +1465,97 @@ export class Repository { processId: string, options: { nowMilliseconds?: number } = {}, ): Promise { - return this.settings.database.transaction(async (connection) => { - const databaseNow = await connection.nowMilliseconds() - const dueAt = options.nowMilliseconds ?? databaseNow - const staleAt = databaseNow - this.settings.processAliveThresholdMilliseconds - await connection.run( - `UPDATE ${this.table("reminders")} SET claimed_by = NULL, claimed_at_ms = NULL - WHERE claimed_by IS NOT NULL AND ( - claimed_at_ms <= ? OR NOT EXISTS ( + return this.settings.database.transaction( + async (connection) => { + const databaseNow = await connection.nowMilliseconds() + const dueAt = options.nowMilliseconds ?? databaseNow + const staleAt = databaseNow - this.settings.processAliveThresholdMilliseconds + const available = await this.findAvailableReminder(connection, dueAt) + const stale = await this.findStaleReminder({ connection, dueAt, staleAt }) + const reminder = earliestReminder(available, stale) + if (!reminder) return undefined + const claimed = await this.claimReminderCandidate({ + connection, + reminder, + processId, + claimedAt: databaseNow, + staleAt, + }) + if (claimed.changes !== 1) return undefined + reminder.claimed_by = processId + reminder.claimed_at_ms = databaseNow + return reminder + }, + { isolationLevel: "read_committed" }, + ) + } + + private findAvailableReminder( + connection: DatabaseConnection, + dueAt: number, + ): Promise { + return connection.get( + `SELECT reminders.*, instances.actor_type, instances.actor_id + FROM ${this.table("reminders")} reminders + JOIN ${this.table("instances")} instances ON instances.id = reminders.instance_id + WHERE reminders.status = 'scheduled' AND reminders.run_at_ms <= ? + AND reminders.claimed_by IS NULL + ORDER BY reminders.run_at_ms, reminders.id + LIMIT 1${this.claimLockClause()}`, + [dueAt], + ) + } + + private findStaleReminder(options: { + connection: DatabaseConnection + dueAt: number + staleAt: number + }): Promise { + const { connection, dueAt, staleAt } = options + return connection.get( + `SELECT reminders.*, instances.actor_type, instances.actor_id + FROM ${this.table("reminders")} reminders + JOIN ${this.table("instances")} instances ON instances.id = reminders.instance_id + WHERE reminders.status = 'scheduled' AND reminders.run_at_ms <= ? + AND reminders.claimed_by IS NOT NULL AND ( + reminders.claimed_at_ms <= ? OR NOT EXISTS ( SELECT 1 FROM ${this.table("processes")} processes - WHERE processes.id = ${this.table("reminders")}.claimed_by + WHERE processes.id = reminders.claimed_by AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? ) - )`, - [staleAt, staleAt], - ) - const reminder = await connection.get( - `SELECT reminders.*, instances.actor_type, instances.actor_id - FROM ${this.table("reminders")} reminders - JOIN ${this.table("instances")} instances ON instances.id = reminders.instance_id - WHERE reminders.status = 'scheduled' AND reminders.run_at_ms <= ? - AND reminders.claimed_by IS NULL - ORDER BY reminders.run_at_ms, reminders.id - LIMIT 1`, - [dueAt], - ) - if (!reminder) return undefined - const claimed = await connection.run( + ) + ORDER BY reminders.run_at_ms, reminders.id + LIMIT 1${this.claimLockClause()}`, + [dueAt, staleAt, staleAt], + ) + } + + private claimReminderCandidate(options: { + connection: DatabaseConnection + reminder: ReminderRow + processId: string + claimedAt: number + staleAt: number + }): Promise<{ changes: number }> { + const { connection, reminder, processId, claimedAt, staleAt } = options + if (reminder.claimed_by === null) { + return connection.run( `UPDATE ${this.table("reminders")} SET claimed_by = ?, claimed_at_ms = ? WHERE id = ? AND status = 'scheduled' AND claimed_by IS NULL`, - [processId, databaseNow, reminder.id], + [processId, claimedAt, reminder.id], ) - if (claimed.changes !== 1) return undefined - reminder.claimed_by = processId - reminder.claimed_at_ms = databaseNow - return reminder - }) + } + return connection.run( + `UPDATE ${this.table("reminders")} SET claimed_by = ?, claimed_at_ms = ? + WHERE id = ? AND status = 'scheduled' AND claimed_by = ? AND ( + claimed_at_ms <= ? OR NOT EXISTS ( + SELECT 1 FROM ${this.table("processes")} processes + WHERE processes.id = ${this.table("reminders")}.claimed_by + AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? + ) + )`, + [processId, claimedAt, reminder.id, reminder.claimed_by, staleAt, staleAt], + ) } async enqueueReminder( @@ -1570,46 +1628,101 @@ export class Repository { } async claimBroadcast(processId: string): Promise { - return this.settings.database.transaction(async (connection) => { - const now = await connection.nowMilliseconds() - const staleAt = now - this.settings.processAliveThresholdMilliseconds - await connection.run( - `UPDATE ${this.table("broadcasts")} SET status = 'pending', claimed_by = NULL - WHERE status = 'processing' AND ( - claimed_by IS NULL OR NOT EXISTS ( - SELECT 1 FROM ${this.table("processes")} processes - WHERE processes.id = ${this.table("broadcasts")}.claimed_by - AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? - ) - )`, - [staleAt], - ) - const broadcast = await connection.get( - `SELECT broadcasts.* FROM ${this.table("broadcasts")} broadcasts - WHERE broadcasts.status = 'pending' AND broadcasts.available_at_ms <= ? - AND NOT EXISTS ( - SELECT 1 FROM ${this.table("broadcasts")} earlier - WHERE earlier.instance_id = broadcasts.instance_id - AND earlier.state_revision < broadcasts.state_revision - AND earlier.status IN ('pending', 'processing') - ) - ORDER BY broadcasts.available_at_ms, broadcasts.id - LIMIT 1`, - [now], - ) - if (!broadcast) return undefined - const claimed = await connection.run( + return this.settings.database.transaction( + async (connection) => { + const now = await connection.nowMilliseconds() + const staleAt = now - this.settings.processAliveThresholdMilliseconds + const pending = await this.findPendingBroadcast(connection, now) + const stale = await this.findStaleBroadcast(connection, staleAt) + const broadcast = earliestBroadcast(pending, stale) + if (!broadcast) return undefined + const claimed = await this.claimBroadcastCandidate({ + connection, + broadcast, + processId, + staleAt, + }) + if (claimed.changes !== 1) return undefined + broadcast.status = "processing" + broadcast.claimed_by = processId + broadcast.attempt_count = Number(broadcast.attempt_count) + 1 + return broadcast + }, + { isolationLevel: "read_committed" }, + ) + } + + private findPendingBroadcast( + connection: DatabaseConnection, + now: number, + ): Promise { + return connection.get( + `SELECT broadcasts.* FROM ${this.table("broadcasts")} broadcasts + WHERE broadcasts.status = 'pending' AND broadcasts.available_at_ms <= ? + AND NOT EXISTS ( + SELECT 1 FROM ${this.table("broadcasts")} earlier + WHERE earlier.instance_id = broadcasts.instance_id + AND earlier.state_revision < broadcasts.state_revision + AND earlier.status IN ('pending', 'processing') + ) + ORDER BY broadcasts.available_at_ms, broadcasts.id + LIMIT 1${this.claimLockClause()}`, + [now], + ) + } + + private findStaleBroadcast( + connection: DatabaseConnection, + staleAt: number, + ): Promise { + return connection.get( + `SELECT broadcasts.* FROM ${this.table("broadcasts")} broadcasts + WHERE broadcasts.status = 'processing' AND ( + broadcasts.claimed_by IS NULL OR NOT EXISTS ( + SELECT 1 FROM ${this.table("processes")} processes + WHERE processes.id = broadcasts.claimed_by + AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? + ) + ) + AND NOT EXISTS ( + SELECT 1 FROM ${this.table("broadcasts")} earlier + WHERE earlier.instance_id = broadcasts.instance_id + AND earlier.state_revision < broadcasts.state_revision + AND earlier.status IN ('pending', 'processing') + ) + ORDER BY broadcasts.available_at_ms, broadcasts.id + LIMIT 1${this.claimLockClause()}`, + [staleAt], + ) + } + + private claimBroadcastCandidate(options: { + connection: DatabaseConnection + broadcast: ClaimableBroadcast + processId: string + staleAt: number + }): Promise<{ changes: number }> { + const { connection, broadcast, processId, staleAt } = options + if (broadcast.status === "pending") { + return connection.run( `UPDATE ${this.table("broadcasts")} SET status = 'processing', claimed_by = ?, attempt_count = attempt_count + 1 WHERE id = ? AND status = 'pending'`, [processId, broadcast.id], ) - if (claimed.changes !== 1) return undefined - broadcast.status = "processing" - broadcast.claimed_by = processId - broadcast.attempt_count = Number(broadcast.attempt_count) + 1 - return broadcast - }) + } + return connection.run( + `UPDATE ${this.table("broadcasts")} + SET claimed_by = ?, attempt_count = attempt_count + 1 + WHERE id = ? AND status = 'processing' AND ( + claimed_by IS NULL OR NOT EXISTS ( + SELECT 1 FROM ${this.table("processes")} processes + WHERE processes.id = ${this.table("broadcasts")}.claimed_by + AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? + ) + )`, + [processId, broadcast.id, staleAt], + ) } async completeBroadcast(broadcast: BroadcastRow): Promise { @@ -1642,6 +1755,11 @@ export class Repository { }) } + private claimLockClause(): string { + if (this.settings.database.family === "sqlite") return "" + return " FOR UPDATE SKIP LOCKED" + } + private findInstance(options: { connection: DatabaseConnection actorType: string @@ -1723,6 +1841,32 @@ export class Repository { } } +type ClaimableBroadcast = BroadcastRow & { status: "pending" | "processing" } + +function earliestBroadcast( + first: ClaimableBroadcast | undefined, + second: ClaimableBroadcast | undefined, +): ClaimableBroadcast | undefined { + if (!first) return second + if (!second) return first + const availableAtDifference = Number(first.available_at_ms) - Number(second.available_at_ms) + if (availableAtDifference < 0) return first + if (availableAtDifference > 0) return second + return first.id < second.id ? first : second +} + +function earliestReminder( + first: ReminderRow | undefined, + second: ReminderRow | undefined, +): ReminderRow | undefined { + if (!first) return second + if (!second) return first + const runAtDifference = Number(first.run_at_ms) - Number(second.run_at_ms) + if (runAtDifference < 0) return first + if (runAtDifference > 0) return second + return first.id < second.id ? first : second +} + function nextReminderRun(options: { previousRun: number interval: number diff --git a/src/schema.ts b/src/schema.ts index 3df10a6..de88974 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -8,7 +8,8 @@ const PROCESS_IDENTITY_VERSION = 4 const PROCESS_DRAINING_VERSION = 5 const OBSERVABLE_INVALIDATIONS_VERSION = 6 const KEYED_REMINDERS_VERSION = 7 -const LATEST_VERSION = KEYED_REMINDERS_VERSION +const POLLING_INDEXES_VERSION = 8 +const LATEST_VERSION = POLLING_INDEXES_VERSION export async function installSchema(options: { connection: DatabaseConnection @@ -290,18 +291,42 @@ export async function installSchema(options: { }) } - if (installedVersions.has(OBSERVABLE_INVALIDATIONS_VERSION)) return - await connection.run( - `ALTER TABLE ${table("broadcasts")} ADD COLUMN ${family === "postgresql" ? "IF NOT EXISTS " : ""}invalidations ${family === "mysql" ? "LONGTEXT" : "TEXT"}`, - ) - await connection.run( - `UPDATE ${table("broadcasts")} SET invalidations = ? WHERE invalidations IS NULL`, - ["[]"], - ) + if (!installedVersions.has(OBSERVABLE_INVALIDATIONS_VERSION)) { + await connection.run( + `ALTER TABLE ${table("broadcasts")} ADD COLUMN ${family === "postgresql" ? "IF NOT EXISTS " : ""}invalidations ${family === "mysql" ? "LONGTEXT" : "TEXT"}`, + ) + await connection.run( + `UPDATE ${table("broadcasts")} SET invalidations = ? WHERE invalidations IS NULL`, + ["[]"], + ) + await recordMigration({ + connection, + table: table("schema_migrations"), + version: OBSERVABLE_INVALIDATIONS_VERSION, + schemaIdentity, + }) + } + + if (installedVersions.has(POLLING_INDEXES_VERSION)) return + const pollingIndexes = [ + ["effects", `${prefix}effects_poll`, "status, available_at_ms, id"], + ["reminders", `${prefix}reminders_due`, "status, run_at_ms, id"], + ["broadcasts", `${prefix}broadcasts_poll`, "status, available_at_ms, id"], + ["broadcasts", `${prefix}broadcasts_instance_revision`, "instance_id, state_revision, status"], + ] as const + for (const [tableName, name, columns] of pollingIndexes) { + await createIndex({ + connection, + family, + table: table(tableName), + name, + columns, + }) + } await recordMigration({ connection, table: table("schema_migrations"), - version: OBSERVABLE_INVALIDATIONS_VERSION, + version: POLLING_INDEXES_VERSION, schemaIdentity, }) } diff --git a/src/version.ts b/src/version.ts index d7f0246..70013c3 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const VERSION = "0.14.5" +export const VERSION = "0.14.6" diff --git a/test/dead-letters.test.ts b/test/dead-letters.test.ts index cd065c7..5ca173f 100644 --- a/test/dead-letters.test.ts +++ b/test/dead-letters.test.ts @@ -151,9 +151,41 @@ describe("schema migrations", () => { const broadcastColumns = await runtime.settings.database.connection((connection) => connection.all<{ name: string }>("PRAGMA table_info(solid_objects_broadcasts)"), ) - expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7]) + expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8]) expect(deadLetterColumns.map(({ name }) => name)).toContain("retried_message_id") expect(broadcastColumns.map(({ name }) => name)).toContain("invalidations") + expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) + }) + + it("adds polling indexes to an existing version-seven database", async () => { + const directory = mkdtempSync(join(tmpdir(), "solid-objects-schema-")) + temporaryDirectories.push(directory) + const path = join(directory, "database.sqlite3") + runtime = configuredRuntime({ database: sqlite({ path }) }) + await runtime.install() + await runtime.close() + runtime = undefined + const database = new DatabaseSync(path) + database.exec(` + DELETE FROM solid_objects_schema_migrations WHERE version = 8; + DROP INDEX solid_objects_effects_poll; + DROP INDEX solid_objects_reminders_due; + DROP INDEX solid_objects_broadcasts_poll; + DROP INDEX solid_objects_broadcasts_instance_revision; + `) + database.close() + runtime = configuredRuntime({ database: sqlite({ path }) }) + + await runtime.install() + await runtime.install() + + const versions = await runtime.settings.database.connection((connection) => + connection.all<{ version: number | bigint }>( + "SELECT version FROM solid_objects_schema_migrations ORDER BY version", + ), + ) + expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8]) + expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS) }) it("preserves version-two request IDs as legacy idempotency keys", async () => { @@ -194,6 +226,30 @@ describe("schema migrations", () => { }) }) +const POLLING_INDEX_COLUMNS = { + solid_objects_effects_poll: ["status", "available_at_ms", "id"], + solid_objects_reminders_due: ["status", "run_at_ms", "id"], + solid_objects_broadcasts_poll: ["status", "available_at_ms", "id"], + solid_objects_broadcasts_instance_revision: ["instance_id", "state_revision", "status"], +} + +async function installedPollingIndexes( + currentRuntime: SolidObjectsRuntime, +): Promise> { + return currentRuntime.settings.database.connection(async (connection) => + Object.fromEntries( + await Promise.all( + Object.keys(POLLING_INDEX_COLUMNS).map(async (name) => [ + name, + (await connection.all<{ name: string }>(`PRAGMA index_info(${name})`)).map( + ({ name: columnName }) => columnName, + ), + ]), + ), + ), + ) +} + async function createDeadLetter(currentRuntime: SolidObjectsRuntime) { const message = await PoisonActor.ref("one").send.run({ source: "test" }) expect(await currentRuntime.worker().runUntilIdle()).toBe(1) diff --git a/test/doctor.test.ts b/test/doctor.test.ts index 7dda70f..9e41f62 100644 --- a/test/doctor.test.ts +++ b/test/doctor.test.ts @@ -27,7 +27,7 @@ describe("runtime doctor", () => { expect(check(report, "configuration").status).toBe("pass") expect(check(report, "schema")).toMatchObject({ status: "pass", - details: { versions: [1, 2, 3, 4, 5, 6, 7] }, + details: { versions: [1, 2, 3, 4, 5, 6, 7, 8] }, }) expect(check(report, "authorization").status).toBe("pass") expect(check(report, "database").status).toBe("pass") diff --git a/test/mysql.test.ts b/test/mysql.test.ts index 7aee4fb..0cf4b87 100644 --- a/test/mysql.test.ts +++ b/test/mysql.test.ts @@ -8,6 +8,7 @@ import { DatabaseDeadlineExceeded } from "../src/errors.js" import { withDatabaseDeadline } from "../src/database/deadline.js" import type { JsonObject } from "../src/types.js" import { createDashboard } from "../src/web/index.js" +import { PausingClaimDatabase } from "./support/pausing-claim-database.js" const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL class TransmitProofCounter extends Actor { @@ -132,6 +133,128 @@ describe("MySQL SQL compatibility", () => { }) describeMySQL("MySQL adapter", () => { + it("lets concurrent effect claimants skip locked work", async () => { + if (!connectionString) throw new Error("MySQL connection string is required") + database = mysql({ connectionString, maximumConnections: 5 }) + const pausingDatabase = new PausingClaimDatabase({ database, table: "effects" }) + runtime = configure({ + database: pausingDatabase, + tableNamePrefix: "mysql_test_", + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + logger: quietLogger, + }) + runtime.register(MySQLWorkflow) + await runtime.install() + await MySQLWorkflow.ref(`first-${crypto.randomUUID()}`).start() + await MySQLWorkflow.ref(`second-${crypto.randomUUID()}`).start() + await runtime.repository.registerProcess("first-effect-worker", "effect_worker") + await runtime.repository.registerProcess("second-effect-worker", "effect_worker") + + const firstClaim = runtime.repository.claimEffect("first-effect-worker") + await pausingDatabase.waitUntilClaimLocked() + const secondAttempt = await withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimEffect("second-effect-worker"), + ).then( + (value) => ({ value }), + (error: unknown) => ({ error }), + ) + pausingDatabase.resume() + const first = await firstClaim + if ("error" in secondAttempt) throw secondAttempt.error + const second = secondAttempt.value + + expect(first).toBeDefined() + expect(second).toBeDefined() + expect(second?.id).not.toBe(first?.id) + }) + + it("lets concurrent reminder claimants skip locked work", async () => { + if (!connectionString) throw new Error("MySQL connection string is required") + database = mysql({ connectionString, maximumConnections: 5 }) + const pausingDatabase = new PausingClaimDatabase({ database, table: "reminders" }) + runtime = configure({ + database: pausingDatabase, + tableNamePrefix: "mysql_test_", + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + logger: quietLogger, + }) + runtime.register(MySQLWorkflow) + await runtime.install() + await MySQLWorkflow.ref(`first-${crypto.randomUUID()}`).start() + await MySQLWorkflow.ref(`second-${crypto.randomUUID()}`).start() + await runtime.repository.registerProcess("first-reminder-worker", "reminder_worker") + await runtime.repository.registerProcess("second-reminder-worker", "reminder_worker") + const reminders = await runtime.settings.database.connection((connection) => + connection.all<{ id: string }>( + `SELECT id FROM ${runtime?.repository.table("reminders")} WHERE status = 'scheduled'`, + ), + ) + expect(reminders).toHaveLength(2) + + const firstClaim = runtime.repository.claimReminder("first-reminder-worker") + await pausingDatabase.waitUntilClaimLocked() + const secondAttempt = await withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimReminder("second-reminder-worker"), + ).then( + (value) => ({ value }), + (error: unknown) => ({ error }), + ) + pausingDatabase.resume() + const first = await firstClaim + if ("error" in secondAttempt) throw secondAttempt.error + const second = secondAttempt.value + + expect(first).toBeDefined() + expect(second).toBeDefined() + expect(second?.id).not.toBe(first?.id) + }) + + it("lets concurrent broadcast claimants skip locked work", async () => { + if (!connectionString) throw new Error("MySQL connection string is required") + database = mysql({ connectionString, maximumConnections: 5 }) + const pausingDatabase = new PausingClaimDatabase({ + database, + table: "broadcasts", + pollingQueriesBeforePause: 2, + }) + runtime = configure({ + database: pausingDatabase, + tableNamePrefix: "mysql_test_", + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + broadcast: async () => {}, + logger: quietLogger, + }) + runtime.register(MySQLWorkflow) + await runtime.install() + await MySQLWorkflow.ref(`first-${crypto.randomUUID()}`).start() + await MySQLWorkflow.ref(`second-${crypto.randomUUID()}`).start() + await runtime.repository.registerProcess("first-broadcast-worker", "broadcast_worker") + await runtime.repository.registerProcess("second-broadcast-worker", "broadcast_worker") + + const firstClaim = runtime.repository.claimBroadcast("first-broadcast-worker") + await pausingDatabase.waitUntilClaimLocked() + const secondAttempt = await withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimBroadcast("second-broadcast-worker"), + ).then( + (value) => ({ value }), + (error: unknown) => ({ error }), + ) + pausingDatabase.resume() + const first = await firstClaim + if ("error" in secondAttempt) throw secondAttempt.error + const second = secondAttempt.value + + expect(first).toBeDefined() + expect(second).toBeDefined() + expect(second?.id).not.toBe(first?.id) + }) + it("stages, drains, and ingests transmit envelopes on MySQL", async () => { if (!connectionString) throw new Error("MySQL connection string is required") const localDatabase = mysql({ connectionString, maximumConnections: 5 }) diff --git a/test/outboxes.test.ts b/test/outboxes.test.ts index b5c6a2b..bc9c724 100644 --- a/test/outboxes.test.ts +++ b/test/outboxes.test.ts @@ -279,6 +279,28 @@ describe("reminders", () => { expect(await alarm.fired).toBe(1) }) + it("recovers the oldest reminder across available and stale work", async () => { + runtime = configuredRuntime() + await runtime.install() + await Alarm.ref("stale-oldest").arm() + await abandonProcess("abandoned-reminder", "reminder_scheduler") + const stale = await runtime.repository.claimReminder("abandoned-reminder") + if (!stale) throw new Error("reminder was not claimed") + await staleProcess("abandoned-reminder") + await Alarm.ref("available-newer").arm() + await runtime.settings.database.connection((connection) => + connection.run( + `UPDATE ${runtime?.repository.table("reminders")} + SET run_at_ms = CASE id WHEN ? THEN 0 ELSE 1 END`, + [stale.id], + ), + ) + + const recovered = await runtime.repository.claimReminder("replacement-reminder-worker") + + expect(recovered?.id).toBe(stale.id) + }) + it("pauses a reminder whose message no longer exists", async () => { runtime = configuredRuntime({ logger: { @@ -363,6 +385,28 @@ describe("observable broadcasts", () => { expect(events).toHaveLength(1) }) + + it("recovers the oldest broadcast across pending and stale work", async () => { + runtime = configuredRuntime({ broadcast: async () => {} }) + await runtime.install() + await ObservableCounter.ref("stale-oldest").increment() + await abandonProcess("abandoned-broadcast", "broadcast_worker") + const stale = await runtime.repository.claimBroadcast("abandoned-broadcast") + if (!stale) throw new Error("broadcast was not claimed") + await staleProcess("abandoned-broadcast") + await ObservableCounter.ref("pending-newer").increment() + await runtime.settings.database.connection((connection) => + connection.run( + `UPDATE ${runtime?.repository.table("broadcasts")} + SET available_at_ms = CASE actor_id WHEN 'stale-oldest' THEN 0 ELSE 1 END`, + ), + ) + + const recovered = await runtime.repository.claimBroadcast("replacement-broadcast-worker") + + expect(recovered?.id).toBe(stale.id) + expect(Number(recovered?.attempt_count)).toBe(2) + }) }) function configuredRuntime( diff --git a/test/polling-queries.test.ts b/test/polling-queries.test.ts new file mode 100644 index 0000000..08c4be1 --- /dev/null +++ b/test/polling-queries.test.ts @@ -0,0 +1,111 @@ +import { afterEach, describe, expect, it } from "vitest" +import type { Database, DatabaseConnection } from "../src/database/types.js" +import { sqlite } from "../src/database/sqlite.js" +import { configure, type SolidObjectsRuntime } from "../src/runtime.js" + +let runtime: SolidObjectsRuntime | undefined + +afterEach(async () => { + await runtime?.close() + runtime = undefined +}) + +describe("polling queries", () => { + it("locks one candidate per indexed query without combining recovery paths", async () => { + const database = sqlite({ path: ":memory:" }) + const installer = configuredRuntime(database) + await installer.install() + const statements: string[] = [] + runtime = configuredRuntime(new RecordingPostgreSQLDatabase(database, statements)) + + await runtime.repository.claimEffect("effect-worker") + await runtime.repository.claimReminder("reminder-worker") + await runtime.repository.claimBroadcast("broadcast-worker") + + const pollingStatements = statements.filter((statement) => + /FROM solid_objects_(effects|reminders|broadcasts) /.test(statement), + ) + expect(pollingStatements).toHaveLength(5) + expect(pollingStatements).toEqual( + expect.arrayContaining([ + expect.stringMatching( + /WHERE effects\.status = 'pending'.*ORDER BY effects\.available_at_ms, effects\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + ), + expect.stringMatching( + /WHERE reminders\.status = 'scheduled'.*reminders\.claimed_by IS NULL.*ORDER BY reminders\.run_at_ms, reminders\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + ), + expect.stringMatching( + /WHERE reminders\.status = 'scheduled'.*reminders\.claimed_by IS NOT NULL.*ORDER BY reminders\.run_at_ms, reminders\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + ), + expect.stringMatching( + /WHERE broadcasts\.status = 'pending'.*ORDER BY broadcasts\.available_at_ms, broadcasts\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + ), + expect.stringMatching( + /WHERE broadcasts\.status = 'processing'.*ORDER BY broadcasts\.available_at_ms, broadcasts\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + ), + ]), + ) + const pendingBroadcast = pollingStatements.find((statement) => + statement.includes("broadcasts.status = 'pending'"), + ) + expect(pendingBroadcast).not.toMatch(/broadcasts\.status = 'processing'/) + const availableReminder = pollingStatements.find((statement) => + statement.includes("reminders.claimed_by IS NULL"), + ) + expect(availableReminder).not.toMatch(/reminders\.claimed_by IS NOT NULL/) + }) +}) + +class RecordingPostgreSQLDatabase implements Database { + readonly family = "postgresql" as const + readonly schemaIdentity: string + + constructor( + private readonly database: Database, + private readonly statements: string[], + ) { + this.schemaIdentity = database.schemaIdentity + } + + connection( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return this.database.connection((connection) => callback(this.recordingConnection(connection))) + } + + transaction( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return this.database.transaction((connection) => callback(this.recordingConnection(connection))) + } + + transactionActive(): boolean { + return this.database.transactionActive?.() ?? false + } + + close(): Promise { + return this.database.close() + } + + private recordingConnection(connection: DatabaseConnection): DatabaseConnection { + return { + run: (sql, parameters) => connection.run(sql, parameters), + get: (sql: string, parameters?: readonly unknown[]) => { + this.statements.push(sql.replace(/\s+/g, " ").trim()) + return connection.get(sql.replace(/\s+FOR UPDATE SKIP LOCKED\s*$/i, ""), parameters) + }, + all: (sql: string, parameters?: readonly unknown[]) => + connection.all(sql, parameters), + nowMilliseconds: () => connection.nowMilliseconds(), + } + } +} + +function configuredRuntime(database: Database): SolidObjectsRuntime { + return configure({ + database, + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + }) +} diff --git a/test/postgresql.test.ts b/test/postgresql.test.ts index a1aa8c0..cdd4bd7 100644 --- a/test/postgresql.test.ts +++ b/test/postgresql.test.ts @@ -15,6 +15,7 @@ import { DatabaseDeadlineExceeded } from "../src/errors.js" import { withDatabaseDeadline } from "../src/database/deadline.js" import type { JsonObject } from "../src/types.js" import { createDashboard } from "../src/web/index.js" +import { PausingClaimDatabase } from "./support/pausing-claim-database.js" const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL const describePostgreSQL = connectionString?.startsWith("postgresql:") ? describe : describe.skip @@ -166,6 +167,48 @@ describe("PostgreSQL SQL parameters", () => { }) describePostgreSQL("PostgreSQL adapter", () => { + it("lets concurrent broadcast claimants skip locked work", async () => { + if (!connectionString) throw new Error("PostgreSQL connection string is required") + database = postgresql({ connectionString, maximumConnections: 5 }) + const pausingDatabase = new PausingClaimDatabase({ + database, + table: "broadcasts", + pollingQueriesBeforePause: 2, + }) + runtime = configure({ + database: pausingDatabase, + tableNamePrefix: "postgresql_test_", + authorizeMessage: () => true, + authorizeQuery: () => true, + authorizeDestroy: () => true, + broadcast: async () => {}, + logger: quietLogger, + }) + runtime.register(PostgreSQLWorkflow) + await runtime.install() + await PostgreSQLWorkflow.ref(`first-${crypto.randomUUID()}`).start() + await PostgreSQLWorkflow.ref(`second-${crypto.randomUUID()}`).start() + await runtime.repository.registerProcess("first-broadcast-worker", "broadcast_worker") + await runtime.repository.registerProcess("second-broadcast-worker", "broadcast_worker") + + const firstClaim = runtime.repository.claimBroadcast("first-broadcast-worker") + await pausingDatabase.waitUntilClaimLocked() + const secondAttempt = await withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimBroadcast("second-broadcast-worker"), + ).then( + (value) => ({ value }), + (error: unknown) => ({ error }), + ) + pausingDatabase.resume() + const first = await firstClaim + if ("error" in secondAttempt) throw secondAttempt.error + const second = secondAttempt.value + + expect(first).toBeDefined() + expect(second).toBeDefined() + expect(second?.id).not.toBe(first?.id) + }) + it("stages, drains, and ingests transmit envelopes on PostgreSQL", async () => { if (!connectionString) throw new Error("PostgreSQL connection string is required") const localDatabase = postgresql({ connectionString, maximumConnections: 5 }) diff --git a/test/support/pausing-claim-database.ts b/test/support/pausing-claim-database.ts new file mode 100644 index 0000000..8b8b66d --- /dev/null +++ b/test/support/pausing-claim-database.ts @@ -0,0 +1,100 @@ +import type { + Database, + DatabaseConnection, + DatabaseFamily, + DatabaseTransactionOptions, +} from "../../src/database/types.js" + +type ClaimTable = "broadcasts" | "effects" | "reminders" + +export class PausingClaimDatabase implements Database { + readonly family: DatabaseFamily + readonly schemaIdentity: string + private readonly claimLocked: Promise + private readonly resumeClaim: Promise + private resolveClaimLocked: () => void = () => {} + private resolveResumeClaim: () => void = () => {} + private pollingQueryCount = 0 + private paused = false + private readonly table: ClaimTable + private readonly pollingQueriesBeforePause: number + + constructor(options: { + database: Database + table: ClaimTable + pollingQueriesBeforePause?: number + }) { + this.database = options.database + this.table = options.table + this.pollingQueriesBeforePause = options.pollingQueriesBeforePause ?? 1 + const { database } = options + this.family = database.family + this.schemaIdentity = database.schemaIdentity + this.claimLocked = new Promise((resolve) => { + this.resolveClaimLocked = resolve + }) + this.resumeClaim = new Promise((resolve) => { + this.resolveResumeClaim = resolve + }) + } + + private readonly database: Database + + waitUntilClaimLocked(): Promise { + return this.claimLocked + } + + resume(): void { + this.resolveResumeClaim() + } + + connection( + callback: (connection: DatabaseConnection) => Promise, + ): Promise { + return this.database.connection((connection) => callback(this.pausingConnection(connection))) + } + + transaction( + callback: (connection: DatabaseConnection) => Promise, + options: DatabaseTransactionOptions = {}, + ): Promise { + return this.database.transaction( + (connection) => callback(this.pausingConnection(connection)), + options, + ) + } + + transactionActive(): boolean { + return this.database.transactionActive?.() ?? false + } + + close(): Promise { + return this.database.close() + } + + private pausingConnection(connection: DatabaseConnection): DatabaseConnection { + return { + run: (sql, parameters) => connection.run(sql, parameters), + get: async (sql: string, parameters?: readonly unknown[]) => { + const row = await connection.get(sql, parameters) + if (!isPollingQuery(sql, this.table)) return row + this.pollingQueryCount += 1 + if (this.paused || this.pollingQueryCount < this.pollingQueriesBeforePause) return row + this.paused = true + this.resolveClaimLocked() + await this.resumeClaim + return row + }, + all: (sql: string, parameters?: readonly unknown[]) => + connection.all(sql, parameters), + nowMilliseconds: () => connection.nowMilliseconds(), + } + } +} + +function isPollingQuery(sql: string, table: ClaimTable): boolean { + return new RegExp( + `FROM\\s+\\S*${table}\\s+${table}[\\s\\S]*FOR UPDATE SKIP LOCKED\\s*$`, + "i", + ).test(sql) +} From 50b5f21bf84591a530f4f9b5a530533072603d5a Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Thu, 3 Sep 2026 08:22:11 -0700 Subject: [PATCH 2/3] fix: keep claim probes indexable Load actor identity after claiming the outbox row. This prevents MySQL from driving the locking query through the instances join, sorting and locking every due reminder before LIMIT can select one. --- CHANGELOG.md | 12 ++++--- src/repository.ts | 64 +++++++++++++++++++++++------------- test/polling-queries.test.ts | 1 + 3 files changed, 50 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9972af4..094d74a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,11 +4,13 @@ - Poll effects, reminders, and broadcasts through ordered indexes installed by schema migration 8. PostgreSQL and MySQL claim one row with - `FOR UPDATE SKIP LOCKED`; SQLite keeps its serialized transaction path. The - broadcast revision guard has its own `(instance_id, state_revision, status)` - index instead of rescanning the outbox for every candidate. Claim transactions - use read committed isolation on PostgreSQL and MySQL so preceding recovery - work cannot turn row skipping into InnoDB gap-lock contention. + `FOR UPDATE SKIP LOCKED`; candidate probes avoid joins and load actor identity + by primary key after the claim. SQLite keeps its serialized transaction path. + The broadcast revision guard has its own + `(instance_id, state_revision, status)` index instead of rescanning the outbox + for every candidate. Claim transactions use read committed isolation on + PostgreSQL and MySQL so preceding recovery work cannot turn row skipping into + InnoDB gap-lock contention. - Replace bulk broadcast and reminder recovery updates with separate available and stale probes, then claim the oldest locked candidate across each pair. This preserves global delivery order, per-instance broadcast revision order, diff --git a/src/repository.ts b/src/repository.ts index a652be1..be2e0b4 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1363,10 +1363,9 @@ export class Repository { )`, [staleAt], ) - const effect = await connection.get( - `SELECT effects.*, instances.actor_type, instances.actor_id + const effect = await connection.get( + `SELECT effects.* FROM ${this.table("effects")} effects - JOIN ${this.table("instances")} instances ON instances.id = effects.instance_id WHERE effects.status = 'pending' AND effects.available_at_ms <= ? ORDER BY effects.available_at_ms, effects.id LIMIT 1${this.claimLockClause()}`, @@ -1380,10 +1379,14 @@ export class Repository { [processId, effect.id], ) if (claimed.changes !== 1) return undefined - effect.status = "processing" - effect.claimed_by = processId - effect.attempt_count = Number(effect.attempt_count) + 1 - return effect + const identity = await this.loadActorIdentity(connection, effect.instance_id) + return { + ...effect, + ...identity, + status: "processing", + claimed_by: processId, + attempt_count: Number(effect.attempt_count) + 1, + } }, { isolationLevel: "read_committed" }, ) @@ -1482,9 +1485,13 @@ export class Repository { staleAt, }) if (claimed.changes !== 1) return undefined - reminder.claimed_by = processId - reminder.claimed_at_ms = databaseNow - return reminder + const identity = await this.loadActorIdentity(connection, reminder.instance_id) + return { + ...reminder, + ...identity, + claimed_by: processId, + claimed_at_ms: databaseNow, + } }, { isolationLevel: "read_committed" }, ) @@ -1493,11 +1500,10 @@ export class Repository { private findAvailableReminder( connection: DatabaseConnection, dueAt: number, - ): Promise { - return connection.get( - `SELECT reminders.*, instances.actor_type, instances.actor_id + ): Promise { + return connection.get( + `SELECT reminders.* FROM ${this.table("reminders")} reminders - JOIN ${this.table("instances")} instances ON instances.id = reminders.instance_id WHERE reminders.status = 'scheduled' AND reminders.run_at_ms <= ? AND reminders.claimed_by IS NULL ORDER BY reminders.run_at_ms, reminders.id @@ -1510,12 +1516,11 @@ export class Repository { connection: DatabaseConnection dueAt: number staleAt: number - }): Promise { + }): Promise { const { connection, dueAt, staleAt } = options - return connection.get( - `SELECT reminders.*, instances.actor_type, instances.actor_id + return connection.get( + `SELECT reminders.* FROM ${this.table("reminders")} reminders - JOIN ${this.table("instances")} instances ON instances.id = reminders.instance_id WHERE reminders.status = 'scheduled' AND reminders.run_at_ms <= ? AND reminders.claimed_by IS NOT NULL AND ( reminders.claimed_at_ms <= ? OR NOT EXISTS ( @@ -1532,7 +1537,7 @@ export class Repository { private claimReminderCandidate(options: { connection: DatabaseConnection - reminder: ReminderRow + reminder: ReminderCandidate processId: string claimedAt: number staleAt: number @@ -1760,6 +1765,18 @@ export class Repository { return " FOR UPDATE SKIP LOCKED" } + private async loadActorIdentity( + connection: DatabaseConnection, + instanceId: string, + ): Promise { + const identity = await connection.get( + `SELECT actor_type, actor_id FROM ${this.table("instances")} WHERE id = ?`, + [instanceId], + ) + if (!identity) throw new Error("claimed outbox instance does not exist") + return identity + } + private findInstance(options: { connection: DatabaseConnection actorType: string @@ -1842,6 +1859,9 @@ export class Repository { } type ClaimableBroadcast = BroadcastRow & { status: "pending" | "processing" } +type ActorIdentity = Pick +type EffectCandidate = Omit +type ReminderCandidate = Omit function earliestBroadcast( first: ClaimableBroadcast | undefined, @@ -1856,9 +1876,9 @@ function earliestBroadcast( } function earliestReminder( - first: ReminderRow | undefined, - second: ReminderRow | undefined, -): ReminderRow | undefined { + first: ReminderCandidate | undefined, + second: ReminderCandidate | undefined, +): ReminderCandidate | undefined { if (!first) return second if (!second) return first const runAtDifference = Number(first.run_at_ms) - Number(second.run_at_ms) diff --git a/test/polling-queries.test.ts b/test/polling-queries.test.ts index 08c4be1..42f2e66 100644 --- a/test/polling-queries.test.ts +++ b/test/polling-queries.test.ts @@ -26,6 +26,7 @@ describe("polling queries", () => { /FROM solid_objects_(effects|reminders|broadcasts) /.test(statement), ) expect(pollingStatements).toHaveLength(5) + for (const statement of pollingStatements) expect(statement).not.toMatch(/\bJOIN\b/) expect(pollingStatements).toEqual( expect.arrayContaining([ expect.stringMatching( From 1d461288a41d4155bd2ead85b7cf7058f4735a2f Mon Sep 17 00:00:00 2001 From: Lucas Carlson Date: Thu, 3 Sep 2026 08:32:27 -0700 Subject: [PATCH 3/3] fix: avoid locking unused candidates Split polling needs both category heads to preserve global order, but locking both can hide one from another claimant. Peek without locks, lock only the selected row, and retry past candidates locked elsewhere. --- CHANGELOG.md | 8 +- docs/parity.md | 18 +- src/repository.ts | 251 ++++++++++++++++++++----- test/mysql.test.ts | 86 ++++++--- test/polling-queries.test.ts | 14 +- test/postgresql.test.ts | 38 +++- test/support/pausing-claim-database.ts | 27 ++- 7 files changed, 336 insertions(+), 106 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 094d74a..fb90795 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,9 +12,11 @@ PostgreSQL and MySQL so preceding recovery work cannot turn row skipping into InnoDB gap-lock contention. - Replace bulk broadcast and reminder recovery updates with separate available - and stale probes, then claim the oldest locked candidate across each pair. - This preserves global delivery order, per-instance broadcast revision order, - stale-process recovery, and at-least-once delivery. + and stale probes. Compare category heads without locking them, lock only the + oldest candidate by primary key, and retry past work already locked by another + claimant. This preserves global delivery order, concurrent progress, + per-instance broadcast revision order, stale-process recovery, and + at-least-once delivery. - On 50,000 production-shaped rows, SQLite, PostgreSQL 18, and MySQL 8.4 all move from scans and explicit sorts to ordered index probes. The worst SQLite broadcast probe fell from 944 ms to 0.018 ms; PostgreSQL's probes finish in diff --git a/docs/parity.md b/docs/parity.md index 1e6df05..2d88638 100644 --- a/docs/parity.md +++ b/docs/parity.md @@ -80,15 +80,15 @@ such boundary between a gem and its dependents. ## Databases and wake-up -| Capability | Status | TypeScript shape or remaining work | -| ------------------------ | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| SQLite | Native | Uses built-in `node:sqlite`, serialized process-local access, bounded transient writer retries, foreign keys, strict tables, database time, and deadline-bounded access and lock waits. | -| PostgreSQL | Native | Optional `pg` 8.23 peer, bounded pooling, 64-bit schema, row-locked sequences, server checks, and deadline-bounded pool, statement, and lock waits. | -| MySQL | Native | Optional `mysql2` 3.23 peer, bounded pooling, InnoDB schema, row-locked sequences, scoped deadlock retry, and deadline-bounded pool, query, and lock waits. Ruby also tests a second client, `trilogy`; Node has no comparable second MySQL client, so only `mysql2` is tracked here. | -| Durable polling fallback | Native | Every role progresses without a notification service. Effects, reminders, and broadcasts use canonical ordered polling indexes; PostgreSQL and MySQL claim with `FOR UPDATE SKIP LOCKED`. Broadcasts and reminders use separate available and stale-recovery probes and preserve the oldest-first choice across each pair. | -| In-process wake-up | Native | A generation-based default adapter prevents claim-to-wait signal loss; commits wake role-specific waiters and polling remains the fallback. | -| 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. | +| Capability | Status | TypeScript shape or remaining work | +| ------------------------ | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| SQLite | Native | Uses built-in `node:sqlite`, serialized process-local access, bounded transient writer retries, foreign keys, strict tables, database time, and deadline-bounded access and lock waits. | +| PostgreSQL | Native | Optional `pg` 8.23 peer, bounded pooling, 64-bit schema, row-locked sequences, server checks, and deadline-bounded pool, statement, and lock waits. | +| MySQL | Native | Optional `mysql2` 3.23 peer, bounded pooling, InnoDB schema, row-locked sequences, scoped deadlock retry, and deadline-bounded pool, query, and lock waits. Ruby also tests a second client, `trilogy`; Node has no comparable second MySQL client, so only `mysql2` is tracked here. | +| Durable polling fallback | Native | Every role progresses without a notification service. Effects, reminders, and broadcasts use canonical ordered polling indexes; PostgreSQL and MySQL lock only the selected row with `FOR UPDATE SKIP LOCKED`. Broadcasts and reminders compare separate available and stale-recovery probes, preserve the oldest-first choice, and retry past candidates locked by another claimant. | +| In-process wake-up | Native | A generation-based default adapter prevents claim-to-wait signal loss; commits wake role-specific waiters and polling remains the fallback. | +| 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 diff --git a/src/repository.ts b/src/repository.ts index be2e0b4..10e1012 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1473,42 +1473,71 @@ export class Repository { const databaseNow = await connection.nowMilliseconds() const dueAt = options.nowMilliseconds ?? databaseNow const staleAt = databaseNow - this.settings.processAliveThresholdMilliseconds - const available = await this.findAvailableReminder(connection, dueAt) - const stale = await this.findStaleReminder({ connection, dueAt, staleAt }) - const reminder = earliestReminder(available, stale) - if (!reminder) return undefined - const claimed = await this.claimReminderCandidate({ - connection, - reminder, - processId, - claimedAt: databaseNow, - staleAt, - }) - if (claimed.changes !== 1) return undefined - const identity = await this.loadActorIdentity(connection, reminder.instance_id) - return { - ...reminder, - ...identity, - claimed_by: processId, - claimed_at_ms: databaseNow, + const skippedCandidateIds: string[] = [] + while (true) { + const available = await this.findAvailableReminder({ + connection, + dueAt, + skippedCandidateIds, + }) + const stale = await this.findStaleReminder({ + connection, + dueAt, + staleAt, + skippedCandidateIds, + }) + const candidate = earliestReminder(available, stale) + if (!candidate) return undefined + const reminder = await this.lockReminderCandidate({ + connection, + candidate, + dueAt, + staleAt, + }) + if (!reminder) { + skippedCandidateIds.push(candidate.id) + continue + } + const claimed = await this.claimReminderCandidate({ + connection, + reminder, + processId, + claimedAt: databaseNow, + staleAt, + }) + if (claimed.changes !== 1) { + skippedCandidateIds.push(reminder.id) + continue + } + const identity = await this.loadActorIdentity(connection, reminder.instance_id) + return { + ...reminder, + ...identity, + claimed_by: processId, + claimed_at_ms: databaseNow, + } } }, { isolationLevel: "read_committed" }, ) } - private findAvailableReminder( - connection: DatabaseConnection, - dueAt: number, - ): Promise { + private findAvailableReminder(options: { + connection: DatabaseConnection + dueAt: number + skippedCandidateIds: string[] + }): Promise { + const { connection, dueAt, skippedCandidateIds } = options + const exclusion = candidateExclusion("reminders", skippedCandidateIds) return connection.get( `SELECT reminders.* FROM ${this.table("reminders")} reminders WHERE reminders.status = 'scheduled' AND reminders.run_at_ms <= ? AND reminders.claimed_by IS NULL + ${exclusion.sql} ORDER BY reminders.run_at_ms, reminders.id - LIMIT 1${this.claimLockClause()}`, - [dueAt], + LIMIT 1`, + [dueAt, ...exclusion.parameters], ) } @@ -1516,8 +1545,10 @@ export class Repository { connection: DatabaseConnection dueAt: number staleAt: number + skippedCandidateIds: string[] }): Promise { - const { connection, dueAt, staleAt } = options + const { connection, dueAt, staleAt, skippedCandidateIds } = options + const exclusion = candidateExclusion("reminders", skippedCandidateIds) return connection.get( `SELECT reminders.* FROM ${this.table("reminders")} reminders @@ -1529,9 +1560,41 @@ export class Repository { AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? ) ) + ${exclusion.sql} ORDER BY reminders.run_at_ms, reminders.id + LIMIT 1`, + [dueAt, staleAt, staleAt, ...exclusion.parameters], + ) + } + + private lockReminderCandidate(options: { + connection: DatabaseConnection + candidate: ReminderCandidate + dueAt: number + staleAt: number + }): Promise { + const { connection, candidate, dueAt, staleAt } = options + if (candidate.claimed_by === null) { + return connection.get( + `SELECT reminders.* FROM ${this.table("reminders")} reminders + WHERE reminders.id = ? AND reminders.status = 'scheduled' + AND reminders.run_at_ms <= ? AND reminders.claimed_by IS NULL + LIMIT 1${this.claimLockClause()}`, + [candidate.id, dueAt], + ) + } + return connection.get( + `SELECT reminders.* FROM ${this.table("reminders")} reminders + WHERE reminders.id = ? AND reminders.status = 'scheduled' + AND reminders.run_at_ms <= ? AND reminders.claimed_by IS NOT NULL AND ( + reminders.claimed_at_ms <= ? OR NOT EXISTS ( + SELECT 1 FROM ${this.table("processes")} processes + WHERE processes.id = reminders.claimed_by + AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? + ) + ) LIMIT 1${this.claimLockClause()}`, - [dueAt, staleAt, staleAt], + [candidate.id, dueAt, staleAt, staleAt], ) } @@ -1637,30 +1700,57 @@ export class Repository { async (connection) => { const now = await connection.nowMilliseconds() const staleAt = now - this.settings.processAliveThresholdMilliseconds - const pending = await this.findPendingBroadcast(connection, now) - const stale = await this.findStaleBroadcast(connection, staleAt) - const broadcast = earliestBroadcast(pending, stale) - if (!broadcast) return undefined - const claimed = await this.claimBroadcastCandidate({ - connection, - broadcast, - processId, - staleAt, - }) - if (claimed.changes !== 1) return undefined - broadcast.status = "processing" - broadcast.claimed_by = processId - broadcast.attempt_count = Number(broadcast.attempt_count) + 1 - return broadcast + const skippedCandidateIds: string[] = [] + while (true) { + const pending = await this.findPendingBroadcast({ + connection, + now, + skippedCandidateIds, + }) + const stale = await this.findStaleBroadcast({ + connection, + staleAt, + skippedCandidateIds, + }) + const candidate = earliestBroadcast(pending, stale) + if (!candidate) return undefined + const broadcast = await this.lockBroadcastCandidate({ + connection, + candidate, + now, + staleAt, + }) + if (!broadcast) { + skippedCandidateIds.push(candidate.id) + continue + } + const claimed = await this.claimBroadcastCandidate({ + connection, + broadcast, + processId, + staleAt, + }) + if (claimed.changes !== 1) { + skippedCandidateIds.push(broadcast.id) + continue + } + broadcast.status = "processing" + broadcast.claimed_by = processId + broadcast.attempt_count = Number(broadcast.attempt_count) + 1 + return broadcast + } }, { isolationLevel: "read_committed" }, ) } - private findPendingBroadcast( - connection: DatabaseConnection, - now: number, - ): Promise { + private findPendingBroadcast(options: { + connection: DatabaseConnection + now: number + skippedCandidateIds: string[] + }): Promise { + const { connection, now, skippedCandidateIds } = options + const exclusion = candidateExclusion("broadcasts", skippedCandidateIds) return connection.get( `SELECT broadcasts.* FROM ${this.table("broadcasts")} broadcasts WHERE broadcasts.status = 'pending' AND broadcasts.available_at_ms <= ? @@ -1670,16 +1760,20 @@ export class Repository { AND earlier.state_revision < broadcasts.state_revision AND earlier.status IN ('pending', 'processing') ) + ${exclusion.sql} ORDER BY broadcasts.available_at_ms, broadcasts.id - LIMIT 1${this.claimLockClause()}`, - [now], + LIMIT 1`, + [now, ...exclusion.parameters], ) } - private findStaleBroadcast( - connection: DatabaseConnection, - staleAt: number, - ): Promise { + private findStaleBroadcast(options: { + connection: DatabaseConnection + staleAt: number + skippedCandidateIds: string[] + }): Promise { + const { connection, staleAt, skippedCandidateIds } = options + const exclusion = candidateExclusion("broadcasts", skippedCandidateIds) return connection.get( `SELECT broadcasts.* FROM ${this.table("broadcasts")} broadcasts WHERE broadcasts.status = 'processing' AND ( @@ -1695,9 +1789,50 @@ export class Repository { AND earlier.state_revision < broadcasts.state_revision AND earlier.status IN ('pending', 'processing') ) + ${exclusion.sql} ORDER BY broadcasts.available_at_ms, broadcasts.id + LIMIT 1`, + [staleAt, ...exclusion.parameters], + ) + } + + private lockBroadcastCandidate(options: { + connection: DatabaseConnection + candidate: ClaimableBroadcast + now: number + staleAt: number + }): Promise { + const { connection, candidate, now, staleAt } = options + if (candidate.status === "pending") { + return connection.get( + `SELECT broadcasts.* FROM ${this.table("broadcasts")} broadcasts + WHERE broadcasts.id = ? AND broadcasts.status = 'pending' + AND broadcasts.available_at_ms <= ? AND NOT EXISTS ( + SELECT 1 FROM ${this.table("broadcasts")} earlier + WHERE earlier.instance_id = broadcasts.instance_id + AND earlier.state_revision < broadcasts.state_revision + AND earlier.status IN ('pending', 'processing') + ) + LIMIT 1${this.claimLockClause()}`, + [candidate.id, now], + ) + } + return connection.get( + `SELECT broadcasts.* FROM ${this.table("broadcasts")} broadcasts + WHERE broadcasts.id = ? AND broadcasts.status = 'processing' AND ( + broadcasts.claimed_by IS NULL OR NOT EXISTS ( + SELECT 1 FROM ${this.table("processes")} processes + WHERE processes.id = broadcasts.claimed_by + AND processes.shutdown_state = 'running' AND processes.heartbeat_at_ms > ? + ) + ) AND NOT EXISTS ( + SELECT 1 FROM ${this.table("broadcasts")} earlier + WHERE earlier.instance_id = broadcasts.instance_id + AND earlier.state_revision < broadcasts.state_revision + AND earlier.status IN ('pending', 'processing') + ) LIMIT 1${this.claimLockClause()}`, - [staleAt], + [candidate.id, staleAt], ) } @@ -1863,6 +1998,18 @@ type ActorIdentity = Pick type EffectCandidate = Omit type ReminderCandidate = Omit +function candidateExclusion( + tableAlias: "broadcasts" | "reminders", + skippedCandidateIds: string[], +): { sql: string; parameters: string[] } { + if (skippedCandidateIds.length === 0) return { sql: "", parameters: [] } + const placeholders = skippedCandidateIds.map(() => "?").join(", ") + return { + sql: `AND ${tableAlias}.id NOT IN (${placeholders})`, + parameters: skippedCandidateIds, + } +} + function earliestBroadcast( first: ClaimableBroadcast | undefined, second: ClaimableBroadcast | undefined, diff --git a/test/mysql.test.ts b/test/mysql.test.ts index 0cf4b87..86f40e2 100644 --- a/test/mysql.test.ts +++ b/test/mysql.test.ts @@ -8,7 +8,7 @@ import { DatabaseDeadlineExceeded } from "../src/errors.js" import { withDatabaseDeadline } from "../src/database/deadline.js" import type { JsonObject } from "../src/types.js" import { createDashboard } from "../src/web/index.js" -import { PausingClaimDatabase } from "./support/pausing-claim-database.js" +import { captureAttempt, PausingClaimDatabase } from "./support/pausing-claim-database.js" const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL class TransmitProofCounter extends Actor { @@ -154,11 +154,10 @@ describeMySQL("MySQL adapter", () => { const firstClaim = runtime.repository.claimEffect("first-effect-worker") await pausingDatabase.waitUntilClaimLocked() - const secondAttempt = await withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => - runtime!.repository.claimEffect("second-effect-worker"), - ).then( - (value) => ({ value }), - (error: unknown) => ({ error }), + const secondAttempt = await captureAttempt( + withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimEffect("second-effect-worker"), + ), ) pausingDatabase.resume() const first = await firstClaim @@ -170,10 +169,14 @@ describeMySQL("MySQL adapter", () => { expect(second?.id).not.toBe(first?.id) }) - it("lets concurrent reminder claimants skip locked work", async () => { + it("does not hide reminder work across concurrent recovery probes", async () => { if (!connectionString) throw new Error("MySQL connection string is required") database = mysql({ connectionString, maximumConnections: 5 }) - const pausingDatabase = new PausingClaimDatabase({ database, table: "reminders" }) + const pausingDatabase = new PausingClaimDatabase({ + database, + table: "reminders", + pollingQueriesBeforePause: 2, + }) runtime = configure({ database: pausingDatabase, tableNamePrefix: "mysql_test_", @@ -190,30 +193,43 @@ describeMySQL("MySQL adapter", () => { await runtime.repository.registerProcess("second-reminder-worker", "reminder_worker") const reminders = await runtime.settings.database.connection((connection) => connection.all<{ id: string }>( - `SELECT id FROM ${runtime?.repository.table("reminders")} WHERE status = 'scheduled'`, + `SELECT id FROM ${runtime?.repository.table("reminders")} + WHERE status = 'scheduled' ORDER BY id`, ), ) expect(reminders).toHaveLength(2) + const [staleReminder, availableReminder] = reminders + if (!staleReminder || !availableReminder) throw new Error("expected two reminders") + await runtime.settings.database.connection(async (connection) => { + await connection.run( + `UPDATE ${runtime?.repository.table("reminders")} + SET claimed_by = 'missing-reminder-worker', claimed_at_ms = 0, run_at_ms = 0 + WHERE id = ?`, + [staleReminder.id], + ) + await connection.run( + `UPDATE ${runtime?.repository.table("reminders")} SET run_at_ms = 1 WHERE id = ?`, + [availableReminder.id], + ) + }) const firstClaim = runtime.repository.claimReminder("first-reminder-worker") await pausingDatabase.waitUntilClaimLocked() - const secondAttempt = await withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => - runtime!.repository.claimReminder("second-reminder-worker"), - ).then( - (value) => ({ value }), - (error: unknown) => ({ error }), + const secondAttempt = await captureAttempt( + withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimReminder("second-reminder-worker"), + ), ) pausingDatabase.resume() const first = await firstClaim if ("error" in secondAttempt) throw secondAttempt.error const second = secondAttempt.value - expect(first).toBeDefined() - expect(second).toBeDefined() - expect(second?.id).not.toBe(first?.id) + expect(first?.id).toBe(staleReminder.id) + expect(second?.id).toBe(availableReminder.id) }) - it("lets concurrent broadcast claimants skip locked work", async () => { + it("does not hide broadcast work across concurrent recovery probes", async () => { if (!connectionString) throw new Error("MySQL connection string is required") database = mysql({ connectionString, maximumConnections: 5 }) const pausingDatabase = new PausingClaimDatabase({ @@ -236,23 +252,41 @@ describeMySQL("MySQL adapter", () => { await MySQLWorkflow.ref(`second-${crypto.randomUUID()}`).start() await runtime.repository.registerProcess("first-broadcast-worker", "broadcast_worker") await runtime.repository.registerProcess("second-broadcast-worker", "broadcast_worker") + const broadcasts = await runtime.settings.database.connection((connection) => + connection.all<{ id: string }>( + `SELECT id FROM ${runtime?.repository.table("broadcasts")} ORDER BY id`, + ), + ) + const [staleBroadcast, pendingBroadcast] = broadcasts + if (!staleBroadcast || !pendingBroadcast) throw new Error("expected two broadcasts") + await runtime.settings.database.connection(async (connection) => { + await connection.run( + `UPDATE ${runtime?.repository.table("broadcasts")} + SET status = 'processing', claimed_by = 'missing-broadcast-worker', + attempt_count = 1, available_at_ms = 0 + WHERE id = ?`, + [staleBroadcast.id], + ) + await connection.run( + `UPDATE ${runtime?.repository.table("broadcasts")} SET available_at_ms = 1 WHERE id = ?`, + [pendingBroadcast.id], + ) + }) const firstClaim = runtime.repository.claimBroadcast("first-broadcast-worker") await pausingDatabase.waitUntilClaimLocked() - const secondAttempt = await withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => - runtime!.repository.claimBroadcast("second-broadcast-worker"), - ).then( - (value) => ({ value }), - (error: unknown) => ({ error }), + const secondAttempt = await captureAttempt( + withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimBroadcast("second-broadcast-worker"), + ), ) pausingDatabase.resume() const first = await firstClaim if ("error" in secondAttempt) throw secondAttempt.error const second = secondAttempt.value - expect(first).toBeDefined() - expect(second).toBeDefined() - expect(second?.id).not.toBe(first?.id) + expect(first?.id).toBe(staleBroadcast.id) + expect(second?.id).toBe(pendingBroadcast.id) }) it("stages, drains, and ingests transmit envelopes on MySQL", async () => { diff --git a/test/polling-queries.test.ts b/test/polling-queries.test.ts index 42f2e66..0a860cc 100644 --- a/test/polling-queries.test.ts +++ b/test/polling-queries.test.ts @@ -11,7 +11,7 @@ afterEach(async () => { }) describe("polling queries", () => { - it("locks one candidate per indexed query without combining recovery paths", async () => { + it("keeps ordered probes indexable without combining recovery paths", async () => { const database = sqlite({ path: ":memory:" }) const installer = configuredRuntime(database) await installer.install() @@ -33,19 +33,23 @@ describe("polling queries", () => { /WHERE effects\.status = 'pending'.*ORDER BY effects\.available_at_ms, effects\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, ), expect.stringMatching( - /WHERE reminders\.status = 'scheduled'.*reminders\.claimed_by IS NULL.*ORDER BY reminders\.run_at_ms, reminders\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + /WHERE reminders\.status = 'scheduled'.*reminders\.claimed_by IS NULL.*ORDER BY reminders\.run_at_ms, reminders\.id LIMIT 1$/, ), expect.stringMatching( - /WHERE reminders\.status = 'scheduled'.*reminders\.claimed_by IS NOT NULL.*ORDER BY reminders\.run_at_ms, reminders\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + /WHERE reminders\.status = 'scheduled'.*reminders\.claimed_by IS NOT NULL.*ORDER BY reminders\.run_at_ms, reminders\.id LIMIT 1$/, ), expect.stringMatching( - /WHERE broadcasts\.status = 'pending'.*ORDER BY broadcasts\.available_at_ms, broadcasts\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + /WHERE broadcasts\.status = 'pending'.*ORDER BY broadcasts\.available_at_ms, broadcasts\.id LIMIT 1$/, ), expect.stringMatching( - /WHERE broadcasts\.status = 'processing'.*ORDER BY broadcasts\.available_at_ms, broadcasts\.id LIMIT 1 FOR UPDATE SKIP LOCKED$/, + /WHERE broadcasts\.status = 'processing'.*ORDER BY broadcasts\.available_at_ms, broadcasts\.id LIMIT 1$/, ), ]), ) + const recoveryProbes = pollingStatements.filter( + (statement) => !statement.includes("FROM solid_objects_effects effects"), + ) + for (const statement of recoveryProbes) expect(statement).not.toMatch(/FOR UPDATE/) const pendingBroadcast = pollingStatements.find((statement) => statement.includes("broadcasts.status = 'pending'"), ) diff --git a/test/postgresql.test.ts b/test/postgresql.test.ts index cdd4bd7..b850d03 100644 --- a/test/postgresql.test.ts +++ b/test/postgresql.test.ts @@ -15,7 +15,7 @@ import { DatabaseDeadlineExceeded } from "../src/errors.js" import { withDatabaseDeadline } from "../src/database/deadline.js" import type { JsonObject } from "../src/types.js" import { createDashboard } from "../src/web/index.js" -import { PausingClaimDatabase } from "./support/pausing-claim-database.js" +import { captureAttempt, PausingClaimDatabase } from "./support/pausing-claim-database.js" const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL const describePostgreSQL = connectionString?.startsWith("postgresql:") ? describe : describe.skip @@ -167,7 +167,7 @@ describe("PostgreSQL SQL parameters", () => { }) describePostgreSQL("PostgreSQL adapter", () => { - it("lets concurrent broadcast claimants skip locked work", async () => { + it("does not hide broadcast work across concurrent recovery probes", async () => { if (!connectionString) throw new Error("PostgreSQL connection string is required") database = postgresql({ connectionString, maximumConnections: 5 }) const pausingDatabase = new PausingClaimDatabase({ @@ -190,23 +190,41 @@ describePostgreSQL("PostgreSQL adapter", () => { await PostgreSQLWorkflow.ref(`second-${crypto.randomUUID()}`).start() await runtime.repository.registerProcess("first-broadcast-worker", "broadcast_worker") await runtime.repository.registerProcess("second-broadcast-worker", "broadcast_worker") + const broadcasts = await runtime.settings.database.connection((connection) => + connection.all<{ id: string }>( + `SELECT id FROM ${runtime?.repository.table("broadcasts")} ORDER BY id`, + ), + ) + const [staleBroadcast, pendingBroadcast] = broadcasts + if (!staleBroadcast || !pendingBroadcast) throw new Error("expected two broadcasts") + await runtime.settings.database.connection(async (connection) => { + await connection.run( + `UPDATE ${runtime?.repository.table("broadcasts")} + SET status = 'processing', claimed_by = 'missing-broadcast-worker', + attempt_count = 1, available_at_ms = 0 + WHERE id = ?`, + [staleBroadcast.id], + ) + await connection.run( + `UPDATE ${runtime?.repository.table("broadcasts")} SET available_at_ms = 1 WHERE id = ?`, + [pendingBroadcast.id], + ) + }) const firstClaim = runtime.repository.claimBroadcast("first-broadcast-worker") await pausingDatabase.waitUntilClaimLocked() - const secondAttempt = await withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => - runtime!.repository.claimBroadcast("second-broadcast-worker"), - ).then( - (value) => ({ value }), - (error: unknown) => ({ error }), + const secondAttempt = await captureAttempt( + withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimBroadcast("second-broadcast-worker"), + ), ) pausingDatabase.resume() const first = await firstClaim if ("error" in secondAttempt) throw secondAttempt.error const second = secondAttempt.value - expect(first).toBeDefined() - expect(second).toBeDefined() - expect(second?.id).not.toBe(first?.id) + expect(first?.id).toBe(staleBroadcast.id) + expect(second?.id).toBe(pendingBroadcast.id) }) it("stages, drains, and ingests transmit envelopes on PostgreSQL", async () => { diff --git a/test/support/pausing-claim-database.ts b/test/support/pausing-claim-database.ts index 8b8b66d..63228d9 100644 --- a/test/support/pausing-claim-database.ts +++ b/test/support/pausing-claim-database.ts @@ -79,7 +79,13 @@ export class PausingClaimDatabase implements Database { const row = await connection.get(sql, parameters) if (!isPollingQuery(sql, this.table)) return row this.pollingQueryCount += 1 - if (this.paused || this.pollingQueryCount < this.pollingQueriesBeforePause) return row + if ( + this.paused || + (this.pollingQueryCount < this.pollingQueriesBeforePause && + !isCandidateLockQuery(sql, this.table)) + ) { + return row + } this.paused = true this.resolveClaimLocked() await this.resumeClaim @@ -92,9 +98,28 @@ export class PausingClaimDatabase implements Database { } } +export async function captureAttempt( + promise: Promise, +): Promise<{ value: Result } | { error: Error }> { + try { + return { value: await promise } + } catch (error) { + return { error: normalizeError(error) } + } +} + function isPollingQuery(sql: string, table: ClaimTable): boolean { return new RegExp( `FROM\\s+\\S*${table}\\s+${table}[\\s\\S]*FOR UPDATE SKIP LOCKED\\s*$`, "i", ).test(sql) } + +function isCandidateLockQuery(sql: string, table: ClaimTable): boolean { + return new RegExp(`WHERE\\s+${table}\\.id\\s*=\\s*\\?`, "i").test(sql) +} + +function normalizeError(error: unknown): Error { + if (error instanceof Error) return error + return new Error("claim attempt rejected with a non-Error value", { cause: error }) +}