diff --git a/CHANGELOG.md b/CHANGELOG.md index 840922b..1696310 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,33 @@ This project follows semantic versioning for the published package. Sample, documentation, and CI-only changes may remain in `Unreleased` until the next package release is useful for users. +## Unreleased + +- **Added the cross-machine wake for the Postgres dialect — `LISTEN`/`NOTIFY`.** + Completes the wake tiers from 0.4.0: `new PostgresOutboxStore({ wakeChannel })` + piggybacks `pg_notify` on the enqueue transaction (Postgres delivers it **on + commit** and drops it on rollback, so the wake is atomic with the event + becoming visible — no post-commit discipline needed), and the new + `PostgresWakeListener` (at `@nest-native/messaging/postgres`) holds a + dedicated `LISTEN` connection that feeds the worker's `OutboxWaker`, + reconnecting with a fixed delay when the connection drops. Best-effort by the + same contract as the other tiers: notifications missed during a reconnect gap + are not recovered — polling remains the delivery backstop, so a lost wake only + costs one poll interval, never an event. Channels are allow-listed to an + identifier-safe charset (`LISTEN` cannot be parameterized), capped at + Postgres's 63-byte identifier limit (beyond it `LISTEN` silently truncates + while `pg_notify` RAISES — which would abort the caller's business + transaction), and double-quoted, keeping `LISTEN` case-sensitivity aligned + with `pg_notify`'s exact-string channel. The listener survives the pg + end-during-connect edge (a client ended mid-connect never settles its + `connect()` promise — the session races it against the connection's own + `end`/`error` events so `stop()` cannot hang), guards a throwing custom + `WakeSignal` from crashing the worker, validates `reconnectDelayMs`, and the + documented factory sets `keepAlive: true` so a half-open LISTEN socket is + detected by the OS. Postgres-only by nature; SQLite/MySQL use the + `WakeSocket` tier. Verified against real Postgres: delivered on commit, + dropped on rollback. + ## 0.4.0 - 2026-07-19 - **Added `OutboxWaker` — an in-process wake for the worker loop.** The worker diff --git a/packages/messaging/README.md b/packages/messaging/README.md index 47f3c64..3c4edf4 100644 --- a/packages/messaging/README.md +++ b/packages/messaging/README.md @@ -111,15 +111,38 @@ switching topology — single process (`OutboxWaker`) ↔ app + worker processes this covers every supported deployment: processes sharing a SQLite file are by definition on one machine. -**Across machines** the wake would have to travel through shared infrastructure -(e.g. Postgres `LISTEN`/`NOTIFY`) — a planned, dialect-specific follow-up. Those -workers rely on polling today. +**Across machines** the wake travels through the one thing every worker already +shares — the database. On the **Postgres** dialect, `LISTEN`/`NOTIFY` carries it: + +```ts +// producer side — one option on the store; pg_notify rides the enqueue +// transaction, so Postgres delivers the wake ON COMMIT and drops it on +// rollback (the signal is atomic with the event becoming visible) +new PostgresOutboxStore({ wakeChannel: 'outbox_wake' }) + +// worker side — a DEDICATED (non-pooled) LISTEN connection feeds the waker +import { PostgresWakeListener } from '@nest-native/messaging/postgres'; + +const listener = new PostgresWakeListener({ + connect: () => new pg.Client({ connectionString, keepAlive: true }), // fresh client per attempt + channel: 'outbox_wake', + waker, // the same OutboxWaker passed to runWorkerLoop +}); +listener.start(); +// on shutdown: await listener.stop(); +``` + +The listener reconnects (default every 5s) when its connection drops; +notifications missed during the gap are not recovered — polling remains the +backstop, exactly as with the other tiers. `LISTEN`/`NOTIFY` is Postgres-only: +SQLite and MySQL deployments use the socket tier above (for SQLite that is the +whole story anyway — its processes share one machine by definition). ## Status & scope - **Drivers:** SQLite (better-sqlite3, sync), Postgres (`pg`, async), and MySQL (`mysql2`, async) via per-dialect stores. - **Transports:** in-process (default, `@nest-native/messaging/in-process` — no broker, at-least-once via the claimer) and Kafka (`@nest-native/kafka`). -- **Latency:** the worker drains a backlog immediately and only idles at `pollIntervalMs`; an `OutboxWaker` cuts that idle wait in-process, and the `WakeSocket` pair carries the wake across processes on the same machine (see above). -- **Roadmap:** a cross-machine wake (Postgres `LISTEN`/`NOTIFY`); additional transports. CDC (Debezium) is an intentional non-goal — this is the app-level outbox. +- **Latency:** the worker drains a backlog immediately and only idles at `pollIntervalMs`; the wake tiers cut that idle wait — `OutboxWaker` in-process, the `WakeSocket` pair across processes on one machine, and Postgres `LISTEN`/`NOTIFY` across machines (see above). +- **Roadmap:** additional transports. CDC (Debezium) is an intentional non-goal — this is the app-level outbox. Part of the [nest-native](https://github.com/nest-native) family. Not affiliated with the NestJS core team. MIT licensed. diff --git a/packages/messaging/dialects/postgres/index.ts b/packages/messaging/dialects/postgres/index.ts index 1cd6a3b..c7554e4 100644 --- a/packages/messaging/dialects/postgres/index.ts +++ b/packages/messaging/dialects/postgres/index.ts @@ -3,3 +3,4 @@ export * from './schema'; export * from './outbox-store'; export * from './inbox-store'; +export * from './wake'; diff --git a/packages/messaging/dialects/postgres/outbox-store.ts b/packages/messaging/dialects/postgres/outbox-store.ts index 933a1e9..042cfd0 100644 --- a/packages/messaging/dialects/postgres/outbox-store.ts +++ b/packages/messaging/dialects/postgres/outbox-store.ts @@ -8,9 +8,20 @@ import type { ResolvedClaimerConfig, } from '../../interfaces'; import { outboxEvents } from './schema'; +import { assertValidWakeChannel } from './wake'; type Db = NodePgDatabase>; +export interface PostgresOutboxStoreOptions { + /** + * When set, `enqueue` also runs `pg_notify(wakeChannel, '')` on the same + * handle — inside the caller's transaction, so Postgres delivers the wake + * **on commit** and drops it on rollback: the signal is atomic with the event + * becoming visible. Pair it with a `PostgresWakeListener` on the workers. + */ + wakeChannel?: string; +} + /** * Postgres (node-postgres) outbox store. Every method is **asynchronous** — * `enqueue` awaits the insert (call it with `await` inside an async @@ -18,6 +29,12 @@ type Db = NodePgDatabase>; * transaction. */ export class PostgresOutboxStore implements OutboxStore { + constructor(private readonly options: PostgresOutboxStoreOptions = {}) { + if (this.options.wakeChannel !== undefined) { + assertValidWakeChannel(this.options.wakeChannel); + } + } + async enqueue(db: unknown, input: EnqueueInput): Promise { const now = new Date().toISOString(); const [row] = await (db as Db) @@ -34,6 +51,13 @@ export class PostgresOutboxStore implements OutboxStore { createdAt: now, }) .returning(); + if (this.options.wakeChannel !== undefined) { + // pg_notify takes the channel as a plain string parameter (safely + // parameterized, unlike LISTEN's identifier) — delivered on commit. + await (db as Db).execute( + sql`select pg_notify(${this.options.wakeChannel}, '')`, + ); + } return row; } diff --git a/packages/messaging/dialects/postgres/wake.ts b/packages/messaging/dialects/postgres/wake.ts new file mode 100644 index 0000000..61adda4 --- /dev/null +++ b/packages/messaging/dialects/postgres/wake.ts @@ -0,0 +1,205 @@ +import type { WakeSignal } from '../../outbox-waker'; + +/** + * The cross-MACHINE wake, for the Postgres dialect: workers on other hosts + * can't be reached by an in-memory `OutboxWaker` or a same-machine + * `WakeSocket…` pair, but they all share the database — so the database carries + * the signal. Producers `NOTIFY` (see `PostgresOutboxStore`'s `wakeChannel` + * option, which piggybacks `pg_notify` on the enqueue transaction — Postgres + * delivers it **on commit** and drops it on rollback, so the wake is atomic + * with the event becoming visible); each worker holds a `LISTEN` connection + * that feeds its `OutboxWaker`. + * + * Best-effort by the same contract as every other wake tier: the listener + * reconnects with a fixed delay when its connection drops, and any + * notification missed during the gap is NOT recovered — polling remains the + * delivery backstop, so a lost wake only costs one poll interval, never an + * event. Correctness never depends on the wake arriving. + */ + +/** + * The slice of `pg.Client` the listener uses — structural, so tests fake it + * hermetically and any LISTEN-capable client satisfies it. Must be a + * DEDICATED, non-pooled connection: notifications arrive only on the exact + * connection that issued `LISTEN`, and a pool may recycle it. + */ +export interface WakeListenConnection { + // `Promise`, not `Promise`: pg.Client's promise overloads + // resolve with values (`connect()` → the client), and TS's void-return + // leniency does not pierce the Promise generic. + connect(): Promise; + query(text: string): Promise; + end(): Promise; + on(event: 'notification', listener: () => void): unknown; + on(event: 'error', listener: (error: unknown) => void): unknown; + on(event: 'end', listener: () => void): unknown; +} + +export interface PostgresWakeListenerOptions { + /** + * Factory for a fresh dedicated connection per attempt — e.g. + * `() => new pg.Client({ connectionString, keepAlive: true })`. A factory + * (not an instance) because every reconnect needs a new client. Set + * `keepAlive: true` (pg leaves it off by default): a LISTEN connection is + * pure-receive, so without TCP keepalives a half-open death (NAT/firewall + * idle eviction, a host vanishing without FIN/RST) is never detected — the + * listener would sit parked forever believing it is healthy while every + * wake silently falls back to the polling interval. + */ + connect: () => WakeListenConnection; + /** The wake target — the same `OutboxWaker` passed to `runWorkerLoop`. */ + waker: WakeSignal; + /** NOTIFY channel to listen on; must match the producer side. Default `outbox_wake`. */ + channel?: string; + /** Delay before re-connecting after a drop or failed attempt. Default 5000ms. */ + reconnectDelayMs?: number; + /** Connection/listen failures are reported here instead of thrown. */ + onError?: (error: unknown) => void; +} + +// LISTEN takes an identifier (it cannot be parameterized like pg_notify's +// string argument), so the channel is allow-listed to a safe charset and then +// double-quoted — quoting also keeps LISTEN case-sensitive, matching the exact +// string the producer passes to pg_notify. +const CHANNEL_PATTERN = /^[A-Za-z_][A-Za-z0-9_]*$/; + +export function assertValidWakeChannel(channel: string): void { + // 63 = NAMEDATALEN - 1, the Postgres identifier limit. Beyond it, LISTEN + // silently truncates while pg_notify RAISES — which would abort the caller's + // business transaction on every enqueue. The charset is ASCII-only, so + // characters equal bytes. + if (!CHANNEL_PATTERN.test(channel) || channel.length > 63) { + throw new Error( + `invalid wake channel ${JSON.stringify(channel)}: must match ${String(CHANNEL_PATTERN)} and be at most 63 characters`, + ); + } +} + +/** + * Holds a dedicated `LISTEN` connection and forwards every notification on the + * channel to `waker.notify()`. `start()` launches a supervision loop that + * reconnects (with `reconnectDelayMs` between attempts) until `stop()`; + * failures go to `onError`, never to the caller — the worker must boot and keep + * polling even when Postgres is briefly unreachable. + */ +export class PostgresWakeListener { + readonly #channel: string; + readonly #reconnectDelayMs: number; + #loop: Promise | null = null; + #current: WakeListenConnection | null = null; + #stopped = false; + #abortSleep: (() => void) | null = null; + + constructor(private readonly options: PostgresWakeListenerOptions) { + this.#channel = options.channel ?? 'outbox_wake'; + assertValidWakeChannel(this.#channel); + this.#reconnectDelayMs = options.reconnectDelayMs ?? 5_000; + // Fail fast on a config that would turn a Postgres outage into a hot + // reconnect loop (0/negative/NaN all collapse to ~immediate retries). + if (!Number.isFinite(this.#reconnectDelayMs) || this.#reconnectDelayMs < 1) { + throw new Error( + `invalid reconnectDelayMs ${this.#reconnectDelayMs}: must be a finite number >= 1`, + ); + } + } + + /** Launch the supervision loop (idempotent — a second call is a no-op). */ + start(): void { + this.#loop ??= this.#supervise(); + } + + /** End the current connection and stop reconnecting. Awaitable; idempotent. */ + async stop(): Promise { + this.#stopped = true; + this.#abortSleep?.(); + const current = this.#current; + if (current) { + await this.#endQuietly(current); // emits 'end' → the parked session resolves + } + await this.#loop; + } + + async #supervise(): Promise { + while (!this.#stopped) { + try { + await this.#session(); + } catch (error) { + this.options.onError?.(error); + } + if (!this.#stopped) { + await this.#sleep(this.#reconnectDelayMs); + } + } + } + + /** One connection lifetime: connect, LISTEN, park until drop/stop. */ + async #session(): Promise { + const client = this.options.connect(); + this.#current = client; + try { + // Park BEFORE connect(), and race connect against it: a client that + // stop() ends mid-connect emits 'end' but its connect() promise NEVER + // settles (pg skips the connect callback once `_ending` is set) — an + // un-raced `await client.connect()` would deadlock stop() forever. The + // early 'error' listener also keeps an async connection error from + // becoming an uncaught event that crashes the worker. + let dead = false; + const parked = new Promise((resolve, reject) => { + client.on('error', (error) => { + dead = true; + reject(error); + }); + client.on('end', () => { + dead = true; + resolve(); + }); + }); + parked.catch(() => undefined); // consumed via the awaits below; never unhandled + client.on('notification', () => this.#forwardWake()); + await Promise.race([client.connect(), parked]); + if (dead) { + return; // ended (stop, or a server drop) before LISTEN could be issued + } + await client.query(`LISTEN "${this.#channel}"`); + await parked; + } finally { + this.#current = null; + await this.#endQuietly(client); + } + } + + // pg emits 'notification' synchronously from its socket-data handler; a + // throwing custom WakeSignal must be routed to onError, not thrown into that + // path (an uncaught exception there kills the worker process). The stock + // OutboxWaker cannot throw, but WakeSignal is a public structural seam. + #forwardWake(): void { + try { + this.options.waker.notify(); + } catch (error) { + this.options.onError?.(error); + } + } + + async #endQuietly(client: WakeListenConnection): Promise { + try { + await client.end(); + } catch (error) { + this.options.onError?.(error); + } + } + + /** Reconnect backoff that `stop()` can cut short. */ + #sleep(ms: number): Promise { + return new Promise((resolve) => { + const timer = setTimeout(() => { + this.#abortSleep = null; + resolve(); + }, ms); + this.#abortSleep = () => { + clearTimeout(timer); + this.#abortSleep = null; + resolve(); + }; + }); + } +} diff --git a/packages/messaging/test/integration/round-trip.spec.ts b/packages/messaging/test/integration/round-trip.spec.ts index 74655b9..e74eb8c 100644 --- a/packages/messaging/test/integration/round-trip.spec.ts +++ b/packages/messaging/test/integration/round-trip.spec.ts @@ -14,6 +14,7 @@ import { outboxEvents as pgOutboxEvents, PostgresInboxStore, PostgresOutboxStore, + PostgresWakeListener, } from '../../dialects/postgres'; // Gated end-to-end tests against a REAL database. They skip unless the matching @@ -200,4 +201,46 @@ describe('Postgres round-trip (real service)', { skip: !POSTGRES_URL }, () => { assert.equal((seen.rows as { c: number }[])[0].c, 1); void pgInboxEvents; }); + + test('LISTEN/NOTIFY wake: delivered on commit, dropped on rollback', async () => { + const pg = await import('pg'); + let wakes = 0; + const listener = new PostgresWakeListener({ + // A dedicated client, NOT the pool: notifications arrive only on the + // exact connection that issued LISTEN. + connect: () => new pg.Client({ connectionString: POSTGRES_URL, keepAlive: true }), + channel: 'outbox_wake_it', + waker: { notify: () => (wakes += 1) }, + }); + listener.start(); + try { + const wakeStore = new PostgresOutboxStore({ wakeChannel: 'outbox_wake_it' }); + + // Give the listener a beat to establish LISTEN before the first NOTIFY. + await new Promise((resolve) => setTimeout(resolve, 300)); + + // Committed enqueue → exactly one wake arrives. + await db.transaction(async (tx) => { + await wakeStore.enqueue(tx, { topic: 'wake.commit', payload: { n: 1 } }); + }); + const deadline = Date.now() + 5_000; + while (wakes === 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.equal(wakes, 1, 'a committed enqueue must wake the listener'); + + // Rolled-back enqueue → Postgres drops the notification with the tx. + await assert.rejects( + db.transaction(async (tx) => { + await wakeStore.enqueue(tx, { topic: 'wake.rollback', payload: { n: 2 } }); + throw new Error('force rollback'); + }), + /force rollback/, + ); + await new Promise((resolve) => setTimeout(resolve, 500)); + assert.equal(wakes, 1, 'a rolled-back enqueue must NOT wake the listener'); + } finally { + await listener.stop(); + } + }); }); diff --git a/packages/messaging/test/postgres-store.spec.ts b/packages/messaging/test/postgres-store.spec.ts index efcbcdd..9d5e493 100644 --- a/packages/messaging/test/postgres-store.spec.ts +++ b/packages/messaging/test/postgres-store.spec.ts @@ -167,6 +167,41 @@ describe('PostgresInboxStore', () => { }); }); +describe('PostgresOutboxStore wakeChannel', () => { + test('enqueue fires pg_notify on the configured channel (observed via LISTEN)', async () => { + // A dedicated PGlite so we can reach its listen() API directly — the shared + // beforeEach db hides the raw client behind drizzle. + const raw = new PGlite(); + const wakeDb = drizzle(raw); + for (const stmt of DDL.split(';')) { + const trimmed = stmt.trim(); + if (trimmed) await wakeDb.execute(trimmed); + } + let wakes = 0; + await raw.listen('outbox_wake_test', () => { + wakes += 1; + }); + + const store = new PostgresOutboxStore({ wakeChannel: 'outbox_wake_test' }); + const row = await store.enqueue(wakeDb, { topic: 't', payload: { a: 1 } }); + assert.equal(row.status, 'pending'); // the insert itself is unchanged + + const deadline = Date.now() + 5_000; + while (wakes === 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + assert.equal(wakes, 1, 'pg_notify must reach the listener'); + await raw.close(); + }); + + test('rejects an unsafe wakeChannel at construction', () => { + assert.throws( + () => new PostgresOutboxStore({ wakeChannel: 'x"; DROP TABLE y' }), + /invalid wake channel/, + ); + }); +}); + describe('isPgUniqueViolation', () => { test('matches 23505 (direct or wrapped in cause), rejects others', () => { assert.equal(isPgUniqueViolation({ code: '23505' }), true); diff --git a/packages/messaging/test/wake-postgres.spec.ts b/packages/messaging/test/wake-postgres.spec.ts new file mode 100644 index 0000000..120e93c --- /dev/null +++ b/packages/messaging/test/wake-postgres.spec.ts @@ -0,0 +1,385 @@ +import { strict as assert } from 'node:assert'; +import { describe, test } from 'node:test'; +import { OutboxWaker } from '../outbox-waker'; +import { + assertValidWakeChannel, + PostgresWakeListener, + type WakeListenConnection, +} from '../dialects/postgres'; + +type Listener = (...args: unknown[]) => void; + +/** A scriptable stand-in for the LISTEN slice of `pg.Client`. */ +class FakeConnection implements WakeListenConnection { + queries: string[] = []; + connectCalls = 0; + endCalls = 0; + failConnect = false; + hangConnect = false; + failQuery = false; + failEnd = false; + #listeners = new Map(); + + async connect(): Promise { + this.connectCalls += 1; + if (this.failConnect) throw new Error('connect refused'); + if (this.hangConnect) { + // Models pg's end()-during-connect behavior: the connect promise never + // settles once the client is ending (only 'end' fires). + return new Promise(() => {}); + } + } + + async query(text: string): Promise { + this.queries.push(text); + if (this.failQuery) throw new Error('query failed'); + return undefined; + } + + async end(): Promise { + this.endCalls += 1; + if (this.failEnd && this.endCalls === 1) throw new Error('end failed'); + this.emit('end'); // pg.Client emits 'end' once the connection closes + } + + on(event: string, listener: Listener): unknown { + const list = this.#listeners.get(event) ?? []; + list.push(listener); + this.#listeners.set(event, list); + return this; + } + + emit(event: string, ...args: unknown[]): void { + for (const listener of this.#listeners.get(event) ?? []) { + listener(...args); + } + } +} + +/** Factory that hands out pre-built fakes in order and records demand. */ +function connectionQueue(...connections: FakeConnection[]): { + factory: () => WakeListenConnection; + handedOut: () => number; +} { + let index = 0; + return { + factory: () => { + const connection = connections[index]; + assert.ok(connection, `factory exhausted after ${index} connections`); + index += 1; + return connection; + }, + handedOut: () => index, + }; +} + +async function until( + predicate: () => boolean, + label: string, + timeoutMs = 5_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + assert.ok(Date.now() < deadline, `timed out waiting for ${label}`); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +describe('assertValidWakeChannel', () => { + test('accepts identifier-safe channels and rejects everything else', () => { + assertValidWakeChannel('outbox_wake'); + assertValidWakeChannel('Wake_1'); + assert.throws(() => assertValidWakeChannel('bad-channel'), /invalid wake channel/); + assert.throws(() => assertValidWakeChannel('x"; DROP TABLE y'), /invalid wake channel/); + assert.throws(() => assertValidWakeChannel(''), /invalid wake channel/); + }); + + test('enforces the 63-byte Postgres identifier limit', () => { + // Beyond 63, LISTEN silently truncates while pg_notify RAISES — aborting + // the caller's business transaction on every enqueue. + assertValidWakeChannel('a'.repeat(63)); + assert.throws(() => assertValidWakeChannel('a'.repeat(64)), /invalid wake channel/); + }); +}); + +describe('PostgresWakeListener', () => { + test('connects, LISTENs on the quoted default channel, and forwards notifications', async () => { + const connection = new FakeConnection(); + const waker = new OutboxWaker(); + const listener = new PostgresWakeListener({ + connect: connectionQueue(connection).factory, + waker, + }); + listener.start(); + await until(() => connection.queries.length === 1, 'LISTEN to be issued'); + assert.deepEqual(connection.queries, ['LISTEN "outbox_wake"']); + + const start = Date.now(); + const parked = waker.wait(8_000); + connection.emit('notification'); + await parked; + assert.ok(Date.now() - start < 1_500, 'notification must cut the wait short'); + await listener.stop(); + }); + + test('quotes a custom mixed-case channel (LISTEN must stay case-sensitive)', async () => { + const connection = new FakeConnection(); + const listener = new PostgresWakeListener({ + connect: connectionQueue(connection).factory, + waker: new OutboxWaker(), + channel: 'Wake_1', + }); + listener.start(); + await until(() => connection.queries.length === 1, 'LISTEN to be issued'); + assert.deepEqual(connection.queries, ['LISTEN "Wake_1"']); + await listener.stop(); + }); + + test('a connection error is reported and a fresh connection re-LISTENs', async () => { + const first = new FakeConnection(); + const second = new FakeConnection(); + const errors: unknown[] = []; + const waker = new OutboxWaker(); + const listener = new PostgresWakeListener({ + connect: connectionQueue(first, second).factory, + waker, + reconnectDelayMs: 5, + onError: (error) => errors.push(error), + }); + listener.start(); + await until(() => first.queries.length === 1, 'first LISTEN'); + first.emit('error', new Error('connection reset')); + await until(() => second.queries.length === 1, 'second LISTEN after reconnect'); + assert.match((errors[0] as Error).message, /connection reset/); + + // The NEW connection must still deliver wakes. + const parked = waker.wait(8_000); + const start = Date.now(); + second.emit('notification'); + await parked; + assert.ok(Date.now() - start < 1_500); + await listener.stop(); + }); + + test('a refused connect is reported and retried', async () => { + const refusing = new FakeConnection(); + refusing.failConnect = true; + const healthy = new FakeConnection(); + const errors: unknown[] = []; + const listener = new PostgresWakeListener({ + connect: connectionQueue(refusing, healthy).factory, + waker: new OutboxWaker(), + reconnectDelayMs: 5, + onError: (error) => errors.push(error), + }); + listener.start(); + await until(() => healthy.queries.length === 1, 'retry after refused connect'); + assert.match((errors[0] as Error).message, /connect refused/); + await listener.stop(); + }); + + test('a failed LISTEN query is reported and retried', async () => { + const broken = new FakeConnection(); + broken.failQuery = true; + const healthy = new FakeConnection(); + const errors: unknown[] = []; + const listener = new PostgresWakeListener({ + connect: connectionQueue(broken, healthy).factory, + waker: new OutboxWaker(), + reconnectDelayMs: 5, + onError: (error) => errors.push(error), + }); + listener.start(); + await until(() => healthy.queries.length === 1, 'retry after failed LISTEN'); + assert.match((errors[0] as Error).message, /query failed/); + await listener.stop(); + }); + + test('a server-side clean end (no error) also reconnects', async () => { + const first = new FakeConnection(); + const second = new FakeConnection(); + const listener = new PostgresWakeListener({ + connect: connectionQueue(first, second).factory, + waker: new OutboxWaker(), + reconnectDelayMs: 5, + }); + listener.start(); + await until(() => first.queries.length === 1, 'first LISTEN'); + first.emit('end'); // server hung up without an error event + await until(() => second.queries.length === 1, 'reconnect after clean end'); + await listener.stop(); + }); + + test('stop() ends the live connection, halts the loop, and is idempotent', async () => { + const connection = new FakeConnection(); + const queue = connectionQueue(connection); + const listener = new PostgresWakeListener({ + connect: queue.factory, + waker: new OutboxWaker(), + reconnectDelayMs: 5, + }); + listener.start(); + await until(() => connection.queries.length === 1, 'LISTEN'); + await listener.stop(); + assert.ok(connection.endCalls >= 1, 'stop must end the connection'); + assert.equal(queue.handedOut(), 1, 'no reconnect after stop'); + await listener.stop(); // second stop is a no-op + }); + + test('stop() during the reconnect backoff exits without a new connection', async () => { + const failing = new FakeConnection(); + failing.failConnect = true; + const spare = new FakeConnection(); + const queue = connectionQueue(failing, spare); + const errors: unknown[] = []; + const listener = new PostgresWakeListener({ + connect: queue.factory, + waker: new OutboxWaker(), + reconnectDelayMs: 60_000, // long backoff: only an aborted sleep exits fast + onError: (error) => errors.push(error), + }); + listener.start(); + await until(() => errors.length === 1, 'the failed connect to be reported'); + const start = Date.now(); + await listener.stop(); + assert.ok(Date.now() - start < 4_000, 'stop must cut the 60s backoff short'); + assert.equal(queue.handedOut(), 1, 'the spare connection must never be used'); + }); + + test('stop() before start() resolves immediately', async () => { + const listener = new PostgresWakeListener({ + connect: connectionQueue().factory, + waker: new OutboxWaker(), + }); + await listener.stop(); + }); + + test('start() twice runs a single supervision loop', async () => { + const connection = new FakeConnection(); + const queue = connectionQueue(connection); + const listener = new PostgresWakeListener({ + connect: queue.factory, + waker: new OutboxWaker(), + }); + listener.start(); + listener.start(); + await until(() => connection.queries.length === 1, 'LISTEN'); + assert.equal(queue.handedOut(), 1); + await listener.stop(); + }); + + test('an error while closing the old connection is reported, and the loop survives', async () => { + const first = new FakeConnection(); + first.failEnd = true; // cleanup after the drop will itself fail once + const second = new FakeConnection(); + const errors: unknown[] = []; + const listener = new PostgresWakeListener({ + connect: connectionQueue(first, second).factory, + waker: new OutboxWaker(), + reconnectDelayMs: 5, + onError: (error) => errors.push(error), + }); + listener.start(); + await until(() => first.queries.length === 1, 'first LISTEN'); + first.emit('error', new Error('connection reset')); + await until(() => second.queries.length === 1, 'reconnect despite failed cleanup'); + const messages = errors.map((e) => (e as Error).message); + assert.ok(messages.includes('connection reset')); + assert.ok(messages.includes('end failed')); + await listener.stop(); + }); + + test('failures without an onError handler are still swallowed', async () => { + const failing = new FakeConnection(); + failing.failConnect = true; + const healthy = new FakeConnection(); + const listener = new PostgresWakeListener({ + connect: connectionQueue(failing, healthy).factory, + waker: new OutboxWaker(), + reconnectDelayMs: 5, + }); + listener.start(); // must not crash the process + await until(() => healthy.queries.length === 1, 'silent retry'); + await listener.stop(); + }); + + test('rejects an unsafe channel at construction', () => { + assert.throws( + () => + new PostgresWakeListener({ + connect: connectionQueue().factory, + waker: new OutboxWaker(), + channel: 'wake; DROP TABLE outbox_events', + }), + /invalid wake channel/, + ); + }); + + test('rejects a reconnectDelayMs that would hot-loop', () => { + for (const bad of [0, -5, Number.NaN]) { + assert.throws( + () => + new PostgresWakeListener({ + connect: connectionQueue().factory, + waker: new OutboxWaker(), + reconnectDelayMs: bad, + }), + /invalid reconnectDelayMs/, + `reconnectDelayMs=${bad} must be rejected`, + ); + } + }); + + test('stop() during a hung connect() resolves promptly (pg end-during-connect deadlock)', async () => { + // pg never settles connect() once end() was called mid-connect; only 'end' + // fires. An un-raced `await connect()` would park stop() forever. + const connection = new FakeConnection(); + connection.hangConnect = true; + const listener = new PostgresWakeListener({ + connect: connectionQueue(connection).factory, + waker: new OutboxWaker(), + }); + listener.start(); + await until(() => connection.connectCalls === 1, 'connect to be in flight'); + const start = Date.now(); + await listener.stop(); // end() → 'end' → the raced session returns + assert.ok(Date.now() - start < 4_000, 'stop must not await the dead connect'); + assert.equal(connection.queries.length, 0, 'LISTEN must never be issued'); + }); + + test('a throwing custom WakeSignal is routed to onError, never thrown into pg', async () => { + const connection = new FakeConnection(); + const errors: unknown[] = []; + const listener = new PostgresWakeListener({ + connect: connectionQueue(connection).factory, + waker: { + notify: () => { + throw new Error('user waker exploded'); + }, + }, + onError: (error) => errors.push(error), + }); + listener.start(); + await until(() => connection.queries.length === 1, 'LISTEN'); + connection.emit('notification'); // must not become an uncaught exception + assert.equal(errors.length, 1); + assert.match((errors[0] as Error).message, /user waker exploded/); + await listener.stop(); + }); + + test('a throwing custom WakeSignal without onError is still swallowed', async () => { + const connection = new FakeConnection(); + const listener = new PostgresWakeListener({ + connect: connectionQueue(connection).factory, + waker: { + notify: () => { + throw new Error('boom'); + }, + }, + }); + listener.start(); + await until(() => connection.queries.length === 1, 'LISTEN'); + connection.emit('notification'); // no onError, still no crash + await listener.stop(); + }); +}); diff --git a/website/docs/api-reference.md b/website/docs/api-reference.md index a446d28..e64d781 100644 --- a/website/docs/api-reference.md +++ b/website/docs/api-reference.md @@ -80,6 +80,7 @@ interface WorkerLoopOptions { signal?: AbortSignal; // abort to stop the loop onTick?: (report: TickReport) => void; onError?: (error: unknown) => void; + waker?: OutboxWaker; // optional event-driven wake (see below) } ``` @@ -87,6 +88,38 @@ Loops `claimer.tick()`: when a tick claims a batch it loops immediately to drain the backlog; when it claims nothing it waits `pollIntervalMs`. A throwing tick is reported via `onError` and the loop continues. +### Wake tiers (cutting the idle latency) + +The idle `pollIntervalMs` wait is the worst-case latency for a lone event +landing in an idle outbox. Three opt-in tiers cut it — polling always remains +the delivery backstop, so a missed wake only costs one poll interval: + +```ts +// same process: notify() after the enqueueing transaction commits +const waker = new OutboxWaker(); +runWorkerLoop(claimer, { waker, signal }); +waker.notify(); + +// separate processes, same machine: a unix-domain-socket bridge +const server = new WakeSocketServer({ path, waker }); // worker side +await server.listen(); +new WakeSocketClient({ path }).notify(); // producer side + +// separate machines (Postgres only): LISTEN/NOTIFY through the database +new PostgresOutboxStore({ wakeChannel: 'outbox_wake' }) // pg_notify rides the tx +const listener = new PostgresWakeListener({ // worker side + connect: () => new pg.Client({ connectionString, keepAlive: true }), // dedicated, non-pooled + channel: 'outbox_wake', + waker, +}); +listener.start(); // reconnects on drops; await listener.stop() on shutdown +``` + +Producers can depend on the shared `WakeSignal` shape (`{ notify(): void }`), so +switching deployment topology never touches domain code. With `wakeChannel`, +Postgres delivers the notify **on commit** and drops it on rollback — the wake +is atomic with the event becoming visible. + ### `InboxService` Injectable (only registered when an `inboxStore` is supplied). The idempotent