diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cc0e89..b17df28 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,21 @@ package release is useful for users. tick and its sleep is never lost. Same-process only; a cross-process wake (Postgres `LISTEN`/`NOTIFY`) is a planned follow-up. Opt-in and fully backward compatible — omit the `waker` and the loop is an unchanged pure poller. +- **Added `WakeSocketServer` / `WakeSocketClient` — the cross-process wake for + processes on the same machine** (the classic app + `start:worker` split sharing + one database, where an in-memory `notify()` can't cross the boundary). The + worker listens on a unix domain socket (Windows: a `\\.\pipe\…` name) and feeds + incoming connections into its `OutboxWaker`; producers hold a + `WakeSocketClient` on the same path and `notify()` after the enqueueing + transaction commits — fire-and-forget, never throwing into the request path. + Built on `node:net` alone (zero new dependencies, dialect-agnostic; for the + SQLite store this covers every supported deployment, since processes sharing a + SQLite file are on one machine by definition). The server recovers a stale + socket path left by a crashed predecessor and refuses a path a live server + owns; polling remains the backstop, so a failed wake only costs one poll + interval. The shared `WakeSignal` interface lets producers swap + `OutboxWaker` ↔ `WakeSocketClient` without touching domain code. A + cross-MACHINE wake (Postgres `LISTEN`/`NOTIFY`) remains the planned follow-up. - Internal simplifications surfaced by the full-package mutation pass (no behavior change): `headerToString` drops a redundant `undefined` guard (the fall-through already returns `undefined`), and the Kafka inbox diff --git a/packages/messaging/README.md b/packages/messaging/README.md index 2219ba8..47f3c64 100644 --- a/packages/messaging/README.md +++ b/packages/messaging/README.md @@ -87,17 +87,39 @@ await this.txHost.withTransaction(async () => { waker.notify(); ``` -**Same process only.** `notify()` can't cross process boundaries — a worker in a -*separate* process from the producer still relies on polling. A cross-process wake -(e.g. Postgres `LISTEN`/`NOTIFY`) is a planned, dialect-specific follow-up; -`OutboxWaker` is the dialect-agnostic half that works today and gives that bridge -something to drive. +**Across processes on the same machine** — the classic split where the HTTP app +and the worker (`npm run start:worker`) are separate processes sharing one +database — an in-memory `notify()` can't cross the boundary. The +`WakeSocketServer`/`WakeSocketClient` pair bridges it over a unix domain socket +(a `\\.\pipe\…` name on Windows), built on `node:net` alone: + +```ts +// worker process — feed incoming wakes into the loop's waker +const waker = new OutboxWaker(); +const wakeServer = new WakeSocketServer({ path: env.outboxWakeSocket, waker }); +await wakeServer.listen(); // recovers a stale path left by a crashed worker +runWorkerLoop(claimer, { pollIntervalMs: 2_000, waker, signal }); + +// app process — same path, fire-and-forget after the commit +const wake = new WakeSocketClient({ path: env.outboxWakeSocket }); +wake.notify(); // never throws; a failed wake only costs one poll interval +``` + +Producers can depend on the shared `WakeSignal` shape (`{ notify(): void }`) so +switching topology — single process (`OutboxWaker`) ↔ app + worker processes +(`WakeSocketClient`) — never touches domain code. Note that for the SQLite store +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. ## 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 for same-process workers (see above). -- **Roadmap:** a cross-process wake (Postgres `LISTEN`/`NOTIFY`) to extend `OutboxWaker` past process boundaries; 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`; 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. 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/index.ts b/packages/messaging/index.ts index a770273..ea29138 100644 --- a/packages/messaging/index.ts +++ b/packages/messaging/index.ts @@ -10,4 +10,5 @@ export * from './outbox-claimer.service'; export * from './inbox.service'; export * from './outbox-waker'; export * from './outbox-worker'; +export * from './wake-socket'; export * from './messaging.module'; diff --git a/packages/messaging/outbox-waker.ts b/packages/messaging/outbox-waker.ts index 33eb3f4..ef299c6 100644 --- a/packages/messaging/outbox-waker.ts +++ b/packages/messaging/outbox-waker.ts @@ -1,3 +1,13 @@ +/** + * Anything a producer can nudge after committing an event: the in-process + * {@link OutboxWaker}, or a {@link WakeSocketClient} when the worker runs in a + * separate process. Producers depend on this shape so swapping deployment + * topology (single process ↔ app + worker processes) doesn't touch domain code. + */ +export interface WakeSignal { + notify(): void; +} + /** * An in-process wake signal for {@link runWorkerLoop}. * @@ -20,7 +30,7 @@ * lost-wakeup race where an event commits in the sliver between a tick and its * sleep. */ -export class OutboxWaker { +export class OutboxWaker implements WakeSignal { /** Resolver of the wait currently parked, or `null` when none is waiting. */ #wake: (() => void) | null = null; /** A `notify()` that arrived with no waiter parked; consumed by the next wait. */ diff --git a/packages/messaging/test/wake-socket.spec.ts b/packages/messaging/test/wake-socket.spec.ts new file mode 100644 index 0000000..ea76f52 --- /dev/null +++ b/packages/messaging/test/wake-socket.spec.ts @@ -0,0 +1,155 @@ +import { strict as assert } from 'node:assert'; +import { unlinkSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { after, describe, test } from 'node:test'; +import { OutboxWaker } from '../outbox-waker'; +import { WakeSocketClient, WakeSocketServer } from '../wake-socket'; + +// Unix-socket paths have a low length cap (~104 bytes on macOS) — keep names short. +let seq = 0; +function socketPath(): string { + seq += 1; + return join(tmpdir(), `wk-${process.pid}-${seq}.sock`); +} + +const openServers: WakeSocketServer[] = []; +async function startServer( + options: ConstructorParameters[0], +): Promise { + const server = new WakeSocketServer(options); + await server.listen(); + openServers.push(server); + return server; +} + +after(async () => { + for (const server of openServers) { + await server.close(); + } +}); + +describe('WakeSocketServer + WakeSocketClient', () => { + test('a client notify() crosses the socket and wakes a parked waker', async () => { + const path = socketPath(); + const waker = new OutboxWaker(); + await startServer({ path, waker }); + + const start = Date.now(); + const parked = waker.wait(8_000); // backstop far above the runtime bound + new WakeSocketClient({ path }).notify(); + await parked; + assert.ok( + Date.now() - start < 1_500, + 'the socket wake should cut the 8s wait short', + ); + }); + + test('a notify() BEFORE the worker parks is latched by the waker (no lost wakeup)', async () => { + const path = socketPath(); + const waker = new OutboxWaker(); + await startServer({ path, waker }); + + new WakeSocketClient({ path }).notify(); + // Give the connection time to land while nobody is parked… + await new Promise((resolve) => setTimeout(resolve, 50)); + // …then the next wait must resolve from the latch, not the 8s backstop. + const start = Date.now(); + await waker.wait(8_000); + assert.ok(Date.now() - start < 1_500, 'latched socket wake was lost'); + }); + + test('recovers a stale socket path left by a crashed predecessor', async () => { + const path = socketPath(); + writeFileSync(path, ''); // a dead filesystem entry occupying the path + const waker = new OutboxWaker(); + await startServer({ path, waker }); // must unlink the stale entry and bind + + const start = Date.now(); + const parked = waker.wait(8_000); + new WakeSocketClient({ path }).notify(); + await parked; + assert.ok(Date.now() - start < 1_500, 'server on a recovered path must work'); + }); + + test('refuses the path when a LIVE server already owns it', async () => { + const path = socketPath(); + await startServer({ path, waker: new OutboxWaker() }); + + const second = new WakeSocketServer({ path, waker: new OutboxWaker() }); + await assert.rejects( + () => second.listen(), + /already listening/, + 'a live server is a conflict, not something to unlink', + ); + }); + + test('a non-EADDRINUSE bind error is rethrown as-is', async () => { + const server = new WakeSocketServer({ + path: join(tmpdir(), 'no-such-dir-wake', 'w.sock'), // parent dir missing + waker: new OutboxWaker(), + }); + await assert.rejects( + () => server.listen(), + (error: NodeJS.ErrnoException) => error.code !== 'EADDRINUSE', + ); + }); + + test('close() is idempotent and safe before listen and after manual unlink', async () => { + const unlistened = new WakeSocketServer({ + path: socketPath(), + waker: new OutboxWaker(), + }); + await unlistened.close(); // never listened — early return + + const path = socketPath(); + const server = new WakeSocketServer({ path, waker: new OutboxWaker() }); + await server.listen(); + unlinkSync(path); // yank the file out from under close() + await server.close(); // tryUnlink's catch path — must not throw + await server.close(); // double close — early return again + }); + + test('an accepted-socket error is swallowed into onError, not thrown', () => { + // The listener guards a teardown race that is not deterministically + // triggerable over a real socket, so it is unit-tested directly: it must + // route to onError (and tolerate onError being absent) without throwing. + const errors: unknown[] = []; + const withHandler = new WakeSocketServer({ + path: socketPath(), + waker: new OutboxWaker(), + onError: (error) => errors.push(error), + }); + type HasListener = { onSocketError: (error: unknown) => void }; + const boom = new Error('teardown race'); + (withHandler as unknown as HasListener).onSocketError(boom); + assert.deepEqual(errors, [boom]); + + const withoutHandler = new WakeSocketServer({ + path: socketPath(), + waker: new OutboxWaker(), + }); + (withoutHandler as unknown as HasListener).onSocketError(boom); // no throw + }); + + test('client notify() against a missing path reports to onError and never throws', async () => { + const errors: unknown[] = []; + const client = new WakeSocketClient({ + path: socketPath(), // nothing listening here + onError: (error) => errors.push(error), + }); + client.notify(); + const deadline = Date.now() + 5_000; + while (errors.length === 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + assert.equal(errors.length, 1); + assert.equal((errors[0] as NodeJS.ErrnoException).code, 'ENOENT'); + }); + + test('client notify() without onError still swallows the failure', async () => { + const client = new WakeSocketClient({ path: socketPath() }); + client.notify(); // must not throw now… + await new Promise((resolve) => setTimeout(resolve, 100)); // …or async-crash later + }); +}); diff --git a/packages/messaging/wake-socket.ts b/packages/messaging/wake-socket.ts new file mode 100644 index 0000000..d83fea4 --- /dev/null +++ b/packages/messaging/wake-socket.ts @@ -0,0 +1,154 @@ +import { unlinkSync } from 'node:fs'; +import { connect, createServer, type Server, type Socket } from 'node:net'; +import type { WakeSignal } from './outbox-waker'; + +/** + * The cross-process half of the wake story, for processes on the SAME machine — + * the classic deployment where the HTTP app and the outbox worker + * (`runWorkerLoop`) are separate processes sharing one database. An in-memory + * `OutboxWaker.notify()` cannot cross that boundary; this bridge carries it over + * a unix domain socket (a `\\.\pipe\…` name on Windows), using only `node:net` — + * no new dependencies, no dialect coupling. + * + * Topology: the worker process runs a {@link WakeSocketServer} pointed at its + * `OutboxWaker`; each producer process holds a {@link WakeSocketClient} on the + * same path and calls `notify()` after its enqueueing transaction commits. The + * connection itself is the entire signal — there is no payload protocol to + * version — and polling remains the backstop, so a failed or missed wake only + * widens latency back to one poll interval, never drops an event. + * + * Cross-MACHINE wake (e.g. Postgres `LISTEN`/`NOTIFY`) is a separate, dialect- + * specific concern this bridge deliberately does not attempt. + */ +export interface WakeSocketOptions { + /** + * Unix-domain-socket path (Windows: a `\\.\pipe\` name). Both processes must + * use the same value; one server owns a path at a time. + */ + path: string; + /** + * Socket-level errors are reported here instead of thrown — a wake is an + * optimization, so its failures must never reach the request path. + */ + onError?: (error: unknown) => void; +} + +export interface WakeSocketServerOptions extends WakeSocketOptions { + /** The wake target — the same `OutboxWaker` passed to `runWorkerLoop`. */ + waker: WakeSignal; +} + +/** + * Worker-process side: listens on the socket path and forwards every incoming + * connection to `waker.notify()`. Handles the crashed-predecessor case: a stale + * socket file (bind refused, probe connect refused) is unlinked and rebound; a + * LIVE server on the path (probe connect succeeds) is a real conflict and throws. + */ +export class WakeSocketServer { + #server: Server | null = null; + + constructor(private readonly options: WakeSocketServerOptions) {} + + async listen(): Promise { + try { + this.#server = await this.bind(); + } catch (error) { + if (!isAddrInUse(error)) { + throw error; + } + // EADDRINUSE: a live worker owns the path, or a crashed one left a stale + // socket file behind. Probe to tell the two apart. + if (await isAlive(this.options.path)) { + throw new Error( + `another wake-socket server is already listening on ${this.options.path}`, + ); + } + tryUnlink(this.options.path); + this.#server = await this.bind(); + } + } + + /** Stop accepting wakes and remove the socket file. Idempotent. */ + async close(): Promise { + const server = this.#server; + if (!server) { + return; + } + this.#server = null; + await new Promise((resolve) => { + server.close(() => resolve()); + }); + tryUnlink(this.options.path); + } + + private bind(): Promise { + return new Promise((resolve, reject) => { + const server = createServer((socket) => this.onConnection(socket)); + server.once('error', reject); + server.listen(this.options.path, () => { + server.removeListener('error', reject); + resolve(server); + }); + }); + } + + private onConnection(socket: Socket): void { + // The connection IS the signal: guard, hang up, wake. + socket.once('error', this.onSocketError); + socket.destroy(); + this.options.waker.notify(); + } + + // A property (not a method) so the exact listener identity is unit-testable; + // a teardown-race error on an accepted socket must be swallowed, not crash the + // worker — the wake is best-effort by contract. + private readonly onSocketError = (error: unknown): void => { + this.options.onError?.(error); + }; +} + +/** + * Producer-process side: `notify()` fires one short-lived connection at the + * server — fire-and-forget, never throws, never blocks the request path. A + * failure (worker down, path missing) goes to `onError` and costs nothing but + * latency: the worker's poll interval remains the delivery backstop. + */ +export class WakeSocketClient implements WakeSignal { + constructor(private readonly options: WakeSocketOptions) {} + + notify(): void { + const socket = connect(this.options.path); + socket.once('connect', () => socket.destroy()); + socket.once('error', (error) => { + socket.destroy(); + this.options.onError?.(error); + }); + } +} + +function isAddrInUse(error: unknown): boolean { + return (error as NodeJS.ErrnoException).code === 'EADDRINUSE'; +} + +/** Probe whether something is actually accepting on the path. */ +function isAlive(path: string): Promise { + return new Promise((resolve) => { + const probe = connect(path); + probe.once('connect', () => { + probe.destroy(); + resolve(true); + }); + probe.once('error', () => { + probe.destroy(); + resolve(false); + }); + }); +} + +function tryUnlink(path: string): void { + try { + unlinkSync(path); + } catch { + // Already gone, or a Windows pipe name with no filesystem entry. + } +}