diff --git a/CHANGELOG.md b/CHANGELOG.md index 069a3e5..fb90795 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,27 @@ # 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`; 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. 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 + 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..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. | -| 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/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..10e1012 100644 --- a/src/repository.ts +++ b/src/repository.ts @@ -1348,42 +1348,48 @@ 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.* + FROM ${this.table("effects")} effects + 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 + 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" }, + ) } async completeEffect(effect: EffectRow, result: JsonValue): Promise { @@ -1462,42 +1468,162 @@ 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 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(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`, + [dueAt, ...exclusion.parameters], + ) + } + + private findStaleReminder(options: { + connection: DatabaseConnection + dueAt: number + staleAt: number + skippedCandidateIds: string[] + }): Promise { + const { connection, dueAt, staleAt, 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 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], + ) + ${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], ) - if (!reminder) return undefined - const claimed = await connection.run( + } + 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()}`, + [candidate.id, dueAt, staleAt, staleAt], + ) + } + + private claimReminderCandidate(options: { + connection: DatabaseConnection + reminder: ReminderCandidate + 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 +1696,173 @@ 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( + return this.settings.database.transaction( + async (connection) => { + const now = await connection.nowMilliseconds() + const staleAt = now - this.settings.processAliveThresholdMilliseconds + 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(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 <= ? + 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') + ) + ${exclusion.sql} + ORDER BY broadcasts.available_at_ms, broadcasts.id + LIMIT 1`, + [now, ...exclusion.parameters], + ) + } + + 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 ( + 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') + ) + ${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.status = 'pending' AND broadcasts.available_at_ms <= ? - AND NOT EXISTS ( + 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') ) - ORDER BY broadcasts.available_at_ms, broadcasts.id - LIMIT 1`, - [now], + LIMIT 1${this.claimLockClause()}`, + [candidate.id, now], ) - if (!broadcast) return undefined - const claimed = await connection.run( + } + 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()}`, + [candidate.id, 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 +1895,23 @@ export class Repository { }) } + private claimLockClause(): string { + if (this.settings.database.family === "sqlite") return "" + 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 @@ -1723,6 +1993,47 @@ export class Repository { } } +type ClaimableBroadcast = BroadcastRow & { status: "pending" | "processing" } +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, +): 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: 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) + 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..86f40e2 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 { captureAttempt, PausingClaimDatabase } from "./support/pausing-claim-database.js" const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL class TransmitProofCounter extends Actor { @@ -132,6 +133,162 @@ 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 captureAttempt( + withDatabaseDeadline({ timeoutMilliseconds: 1_000 }, () => + runtime!.repository.claimEffect("second-effect-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) + }) + + 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", + pollingQueriesBeforePause: 2, + }) + 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' 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 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?.id).toBe(staleReminder.id) + expect(second?.id).toBe(availableReminder.id) + }) + + 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({ + 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 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 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?.id).toBe(staleBroadcast.id) + expect(second?.id).toBe(pendingBroadcast.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..0a860cc --- /dev/null +++ b/test/polling-queries.test.ts @@ -0,0 +1,116 @@ +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("keeps ordered probes indexable 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) + for (const statement of pollingStatements) expect(statement).not.toMatch(/\bJOIN\b/) + 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$/, + ), + expect.stringMatching( + /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$/, + ), + expect.stringMatching( + /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'"), + ) + 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..b850d03 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 { captureAttempt, 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,66 @@ describe("PostgreSQL SQL parameters", () => { }) describePostgreSQL("PostgreSQL adapter", () => { + 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({ + 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 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 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?.id).toBe(staleBroadcast.id) + expect(second?.id).toBe(pendingBroadcast.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..63228d9 --- /dev/null +++ b/test/support/pausing-claim-database.ts @@ -0,0 +1,125 @@ +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 && + !isCandidateLockQuery(sql, this.table)) + ) { + 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(), + } + } +} + +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 }) +}