feat: add OutboxWaker — in-process wake to cut worker idle latency

CHANGELOG.md

@@ -8,6 +8,17 @@ package release is useful for users.
 
 ## Unreleased
 
+- **Added `OutboxWaker` — an in-process wake for the worker loop.** The worker
+  only waits `pollIntervalMs` when a tick claims nothing, so that interval is the
+  worst-case latency for a lone event landing in an idle outbox. Pass an
+  `OutboxWaker` to `runWorkerLoop({ waker })` and call `waker.notify()` after the
+  enqueueing transaction commits to cut that idle wait short — the worker relays
+  immediately instead of on the next poll. Polling stays the backstop, so a missed
+  or absent `notify()` never stalls delivery (it only widens latency back to one
+  interval), and wakes are latched so an event committed in the sliver between a
+  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.
 - 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

@@ -54,10 +54,50 @@ npm install @nest-native/kafka                                     # only for th
 
 See the [00-showcase sample](https://github.com/nest-native/messaging/tree/main/sample/00-showcase) for a runnable end-to-end example on SQLite.
 
+### Cutting the idle latency (`OutboxWaker`)
+
+`runWorkerLoop` is self-clocking: after a tick that claims a full batch it loops
+again immediately to drain the backlog, and it only waits `pollIntervalMs`
+(default 2s) when a tick claims **nothing**. So that interval is the worst-case
+latency for a *lone* event landing in an otherwise-idle outbox — not a per-event
+tax. Under load, throughput is never gated by it.
+
+When even that idle latency matters — a user-facing "we've got it" sitting behind
+an outbox event — turning the poll interval down works but has a floor and a
+DB-load cost. The better lever is an **in-process wake**: pass an `OutboxWaker`
+to the loop and `notify()` it right after the enqueueing transaction commits, so
+the worker relays now instead of on the next poll. Polling stays the backstop, so
+a missed or absent `notify()` never stalls delivery — it only widens latency back
+to one interval.
+
+```ts
+import { OutboxWaker, runWorkerLoop } from '@nest-native/messaging';
+
+const waker = new OutboxWaker();
+
+// worker: the idle wait is now woken early by notify()
+runWorkerLoop(claimer, { pollIntervalMs: 2_000, waker, signal });
+
+// request path: notify AFTER the transaction commits (before commit the row
+// isn't visible to the claimer's own transaction yet)
+await this.txHost.withTransaction(async () => {
+  await this.outbox.enqueue({ topic: 'order.paid', payload });
+  // ...business writes...
+});
+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.
+
 ## 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`).
-- **Roadmap:** 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 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.
 
 Part of the [nest-native](https://github.com/nest-native) family. Not affiliated with the NestJS core team. MIT licensed.

packages/messaging/index.ts

@@ -8,5 +8,6 @@ export * from './tokens';
 export * from './outbox-producer.service';
 export * from './outbox-claimer.service';
 export * from './inbox.service';
+export * from './outbox-waker';
 export * from './outbox-worker';
 export * from './messaging.module';

packages/messaging/outbox-waker.ts

@@ -0,0 +1,74 @@
+/**
+ * An in-process wake signal for {@link runWorkerLoop}.
+ *
+ * The worker only pays the idle `pollIntervalMs` latency when a tick claims
+ * nothing. A producer that has just committed a new event can call
+ * {@link OutboxWaker.notify} to cut that idle wait short so the next tick runs
+ * immediately — turning the poll interval from a per-event latency tax into a
+ * mere backstop. A missed or absent `notify()` never stalls delivery; it only
+ * widens latency back to one poll interval, so correctness never depends on it.
+ *
+ * **Same-process only.** `notify()` cannot cross process boundaries — a worker
+ * running in a separate process from the producer still relies on polling. A
+ * cross-process wake (e.g. Postgres `LISTEN`/`NOTIFY`) is a future, dialect-
+ * specific addition; this primitive is the dialect-agnostic half that works today
+ * and gives that future bridge something to drive.
+ *
+ * **Wakes are latched.** A `notify()` that arrives while the worker is mid-tick
+ * (after an idle tick decided to wait, but before — or between — waits) is
+ * remembered, so the very next idle wait returns at once. This closes the classic
+ * lost-wakeup race where an event commits in the sliver between a tick and its
+ * sleep.
+ */
+export class OutboxWaker {
+  /** 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. */
+  #pending = false;
+
+  /**
+   * Wake the idle worker so its next tick runs now instead of after the poll
+   * interval. Safe to call at any time and from anywhere in the same process —
+   * ideally right after the transaction that enqueued the event commits (before
+   * commit, the row is not yet visible to the claimer's own transaction).
+   */
+  notify(): void {
+    const wake = this.#wake;
+    if (wake) {
+      // A waiter is parked: resolve it now. Clear the slot first so a re-entrant
+      // notify (or the settle path) latches rather than double-firing.
+      this.#wake = null;
+      wake();
+    } else {
+      // Nobody waiting yet — latch it for the next wait so the wake isn't lost.
+      this.#pending = true;
+    }
+  }
+
+  /**
+   * Internal — used by {@link runWorkerLoop} in place of a plain sleep. Resolves
+   * on the first of: a {@link notify}, the `ms` timer (the poll backstop), or
+   * `signal` aborting. A latched pending wake (or an already-aborted signal)
+   * resolves synchronously.
+   */
+  wait(ms: number, signal?: AbortSignal): Promise<void> {
+    if (this.#pending || signal?.aborted) {
+      this.#pending = false;
+      return Promise.resolve();
+    }
+    return new Promise<void>((resolve) => {
+      // One idempotent settle for all three trigger sources: whichever fires
+      // first disarms the others (clears the timer, drops the abort listener,
+      // frees the wake slot), so settle runs at most once per wait.
+      const settle = (): void => {
+        clearTimeout(timer);
+        signal?.removeEventListener('abort', settle);
+        this.#wake = null;
+        resolve();
+      };
+      const timer = setTimeout(settle, ms);
+      this.#wake = settle;
+      signal?.addEventListener('abort', settle, { once: true });
+    });
+  }
+}

packages/messaging/outbox-worker.ts

@@ -1,5 +1,6 @@
 import type { ClaimerConfig } from './interfaces';
 import type { OutboxClaimer, TickReport } from './outbox-claimer.service';
+import type { OutboxWaker } from './outbox-waker';
 
 export interface WorkerLoopOptions {
   /** Delay between ticks when the last tick claimed nothing (default 2000ms). */
@@ -12,30 +13,44 @@ export interface WorkerLoopOptions {
   onTick?: (report: TickReport) => void;
   /** Called when a tick throws — the loop reports and continues. */
   onError?: (error: unknown) => void;
+  /**
+   * Optional in-process wake. When provided, a producer that calls
+   * {@link OutboxWaker.notify} after committing an event cuts the idle wait short
+   * so the next tick runs immediately; `pollIntervalMs` becomes a backstop rather
+   * than a per-event latency tax. Omit it to keep pure polling.
+   */
+  waker?: OutboxWaker;
 }
 
 /**
  * Runs `claimer.tick()` in a loop until `signal` aborts. 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 after the same wait.
+ *
+ * Pass a {@link OutboxWaker} to have that idle wait woken early by an in-process
+ * `notify()`; without one, the loop is a pure poller.
  */
 export async function runWorkerLoop(
   claimer: OutboxClaimer,
   options: WorkerLoopOptions = {},
 ): Promise<void> {
   const pollIntervalMs = options.pollIntervalMs ?? 2_000;
-  const { signal } = options;
+  const { signal, waker } = options;
+  // The idle/error wait is either woken early by the waker or a plain sleep.
+  const wait = waker
+    ? (ms: number): Promise<void> => waker.wait(ms, signal)
+    : (ms: number): Promise<void> => sleep(ms, signal);
   while (!signal?.aborted) {
     try {
       const report = await claimer.tick(options.claimer);
       options.onTick?.(report);
       if (report.claimed === 0) {
-        await sleep(pollIntervalMs, signal);
+        await wait(pollIntervalMs);
       }
     } catch (error) {
       options.onError?.(error);
-      await sleep(pollIntervalMs, signal);
+      await wait(pollIntervalMs);
     }
   }
 }

packages/messaging/test/waker.spec.ts

@@ -0,0 +1,82 @@
+import { strict as assert } from 'node:assert';
+import { getEventListeners } from 'node:events';
+import { describe, test } from 'node:test';
+import { OutboxWaker } from '../outbox-waker';
+
+describe('OutboxWaker', () => {
+  test('wait() resolves on the backstop timer when nothing wakes it', async () => {
+    const waker = new OutboxWaker();
+    const start = Date.now();
+    await waker.wait(40);
+    const gap = Date.now() - start;
+    assert.ok(gap >= 25, `expected the timer to elapse, got ${gap}ms`);
+    assert.ok(gap < 1_000, `timer overshot: ${gap}ms`);
+  });
+
+  test('notify() wakes a parked wait immediately', async () => {
+    const waker = new OutboxWaker();
+    const start = Date.now();
+    const parked = waker.wait(8_000); // long backstop we should never reach
+    waker.notify();
+    await parked;
+    assert.ok(Date.now() - start < 1_000, 'notify should cut the long wait short');
+  });
+
+  test('a notify() with no waiter is latched and consumed by the next wait', async () => {
+    const waker = new OutboxWaker();
+    waker.notify(); // nobody parked yet — must latch
+    waker.notify(); // a second one collapses into the same single latch
+    const start = Date.now();
+    await waker.wait(8_000); // returns at once from the latch
+    assert.ok(Date.now() - start < 1_000, 'latched wake should resolve the next wait');
+
+    // The latch is one-shot: the following wait falls through to the backstop.
+    const secondStart = Date.now();
+    await waker.wait(40);
+    assert.ok(
+      Date.now() - secondStart >= 25,
+      'the latch must be consumed, not sticky',
+    );
+  });
+
+  test('wait() returns immediately when the signal is already aborted', async () => {
+    const waker = new OutboxWaker();
+    const controller = new AbortController();
+    controller.abort();
+    const start = Date.now();
+    await waker.wait(8_000, controller.signal);
+    assert.ok(Date.now() - start < 1_000);
+  });
+
+  test('an abort during the wait wakes it and removes the listener', async () => {
+    const waker = new OutboxWaker();
+    const controller = new AbortController();
+    const start = Date.now();
+    const parked = waker.wait(8_000, controller.signal);
+    setTimeout(() => controller.abort(), 20);
+    await parked;
+    assert.ok(Date.now() - start < 1_000);
+    // settle must drop the abort listener — no leak across waits.
+    assert.equal(getEventListeners(controller.signal, 'abort').length, 0);
+  });
+
+  test('notify() with a signal present still cleans up the abort listener', async () => {
+    const waker = new OutboxWaker();
+    const controller = new AbortController();
+    const parked = waker.wait(8_000, controller.signal);
+    waker.notify();
+    await parked;
+    assert.equal(getEventListeners(controller.signal, 'abort').length, 0);
+  });
+
+  test('works without a signal: both the backstop and notify paths', async () => {
+    const waker = new OutboxWaker();
+    // notify path with no signal (removeEventListener/addEventListener optional
+    // chains take their absent branch).
+    const start = Date.now();
+    const parked = waker.wait(8_000);
+    waker.notify();
+    await parked;
+    assert.ok(Date.now() - start < 1_000);
+  });
+});

packages/messaging/test/worker.spec.ts

@@ -3,6 +3,7 @@ import { strict as assert } from 'node:assert';
 import { getEventListeners } from 'node:events';
 import { describe, test } from 'node:test';
 import type { OutboxClaimer, TickReport } from '../outbox-claimer.service';
+import { OutboxWaker } from '../outbox-waker';
 import { runWorkerLoop } from '../outbox-worker';
 
 const report = (claimed: number): TickReport => ({
@@ -193,6 +194,45 @@ describe('runWorkerLoop', () => {
     assert.equal(calls, 2);
   });
 
+  test('a waker cuts the idle wait short instead of sleeping the full interval', async () => {
+    // The idle wait is set to 8s — far above the runtime bound — so the test can
+    // only pass if the waker's notify() (not the timer) ends the wait.
+    const controller = new AbortController();
+    const waker = new OutboxWaker();
+    const timestamps: number[] = [];
+    let calls = 0;
+    await runWorkerLoop(
+      fakeClaimer(async () => {
+        calls += 1;
+        timestamps.push(Date.now());
+        if (calls === 1) setTimeout(() => waker.notify(), 20);
+        if (calls === 2) controller.abort();
+        return report(0); // always idle, so every gap is a waker-woken wait
+      }),
+      { pollIntervalMs: 8_000, signal: controller.signal, waker },
+    );
+    assert.equal(calls, 2);
+    const gap = timestamps[1]! - timestamps[0]!;
+    assert.ok(gap < 1_500, `waker should cut the 8s idle wait, got ${gap}ms`);
+  });
+
+  test('a waker-driven loop still stops promptly on abort (waker path honours the signal)', async () => {
+    const controller = new AbortController();
+    const waker = new OutboxWaker();
+    let calls = 0;
+    const start = Date.now();
+    await runWorkerLoop(
+      fakeClaimer(async () => {
+        calls += 1;
+        setTimeout(() => controller.abort(), 25); // abort mid idle-wait
+        return report(0);
+      }),
+      { pollIntervalMs: 8_000, signal: controller.signal, waker },
+    );
+    assert.equal(calls, 1);
+    assert.ok(Date.now() - start < 4_000, 'abort must short-circuit the waker wait');
+  });
+
   test('runs without a signal: the loop keeps going (and never settles)', async () => {
     let calls = 0;
     let settled = false;