feat: add WakeSocketServer/WakeSocketClient — cross-process wake (same machine)

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

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.

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';

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. */

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<typeof WakeSocketServer>[0],
+): Promise<WakeSocketServer> {
+  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
+  });
+});