From b60f53ce25a30b60759df4eea7550fa93c107dfd Mon Sep 17 00:00:00 2001 From: rodrigobnogueira Date: Sun, 19 Jul 2026 01:23:17 -0300 Subject: [PATCH] =?UTF-8?q?feat:=20add=20OutboxWaker=20=E2=80=94=20in-proc?= =?UTF-8?q?ess=20wake=20to=20cut=20worker=20idle=20latency?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `runWorkerLoop` 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 (under load the loop drains immediately and is never gated by it). When that idle latency matters — a user-facing signal behind an outbox event — turning the poll down has a floor and a DB-load cost. `OutboxWaker` is the better lever: pass it to `runWorkerLoop({ waker })` and call `waker.notify()` after the enqueueing transaction commits, so the worker relays now instead of on the next poll. Polling stays the backstop — a missed or absent notify never stalls delivery, 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. Dialect-agnostic and same-process only; a cross-process wake (Postgres LISTEN/NOTIFY) is a planned follow-up. Fully backward compatible: omit the waker and the loop is an unchanged pure poller. - outbox-waker.ts: the primitive (notify + internal latched wait), 100% covered - outbox-worker.ts: optional `waker` option; idle/error wait uses it when present - tests: waker unit specs + worker-loop waker path; latch mutant hand-verified - README + CHANGELOG: the low-latency-wake recipe and the same-process caveat --- CHANGELOG.md | 11 ++++ packages/messaging/README.md | 42 ++++++++++++- packages/messaging/index.ts | 1 + packages/messaging/outbox-waker.ts | 74 +++++++++++++++++++++++ packages/messaging/outbox-worker.ts | 21 ++++++- packages/messaging/test/waker.spec.ts | 82 ++++++++++++++++++++++++++ packages/messaging/test/worker.spec.ts | 40 +++++++++++++ 7 files changed, 267 insertions(+), 4 deletions(-) create mode 100644 packages/messaging/outbox-waker.ts create mode 100644 packages/messaging/test/waker.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 597de47..7cc0e89 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/packages/messaging/README.md b/packages/messaging/README.md index 8d18691..2219ba8 100644 --- a/packages/messaging/README.md +++ b/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. diff --git a/packages/messaging/index.ts b/packages/messaging/index.ts index bc456f9..a770273 100644 --- a/packages/messaging/index.ts +++ b/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'; diff --git a/packages/messaging/outbox-waker.ts b/packages/messaging/outbox-waker.ts new file mode 100644 index 0000000..33eb3f4 --- /dev/null +++ b/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 { + if (this.#pending || signal?.aborted) { + this.#pending = false; + return Promise.resolve(); + } + return new Promise((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 }); + }); + } +} diff --git a/packages/messaging/outbox-worker.ts b/packages/messaging/outbox-worker.ts index 048bbd1..8620deb 100644 --- a/packages/messaging/outbox-worker.ts +++ b/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,6 +13,13 @@ 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; } /** @@ -19,23 +27,30 @@ export interface WorkerLoopOptions { * 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 { 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 => waker.wait(ms, signal) + : (ms: number): Promise => 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); } } } diff --git a/packages/messaging/test/waker.spec.ts b/packages/messaging/test/waker.spec.ts new file mode 100644 index 0000000..7c605aa --- /dev/null +++ b/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); + }); +}); diff --git a/packages/messaging/test/worker.spec.ts b/packages/messaging/test/worker.spec.ts index 7e47c3b..2084c90 100644 --- a/packages/messaging/test/worker.spec.ts +++ b/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;