diff --git a/.changeset/17623-http-dispatcher-idle-cost.md b/.changeset/17623-http-dispatcher-idle-cost.md new file mode 100644 index 0000000000..08d3d5a130 --- /dev/null +++ b/.changeset/17623-http-dispatcher-idle-cost.md @@ -0,0 +1,26 @@ +--- +'@objectstack/service-messaging': minor +--- + +`HttpDispatcher` reaps once per tick instead of once per partition, backs off while `sys_http_delivery` is idle, and `enqueueHttp()` / `redeliverHttp()` wake it (#17623) + +**What an idle dispatcher cost.** Against an EMPTY `sys_http_delivery` outbox every tick walked `partitionCount` partitions (default 8) and ran `claim()` in each — and each claim opened with the environment-wide visibility-timeout reap before its candidate SELECT. Measured on a real `ObjectQL` + `SqlDriver`: **16 SQL statements a tick, 8 of them the identical reap UPDATE**, on a fixed 500 ms `setInterval` that never let up, one loop per warm kernel. It is the shape #17610 removed from `NotificationDispatcher`, still running beside it. On remote Turso every statement is an HTTP round trip. + +**Now:** + +- **The reap runs once per tick**, before any claim — an idle tick is `1 + partitionCount` = 9 statements. Its predicate names no partition, so one run returns every claim that had expired when the tick began; a claim that expires during the tick is returned by the next one. A crashed node's `in_flight` rows are still recovered within one tick of `claimTtlMs` passing, and a claim is still never re-taken before its TTL. +- **The loop backs off while idle.** Every tick that claims nothing doubles the delay to the next, from `intervalMs` up to `maxIdleIntervalMs` (default 30 s, the notification dispatcher's default). A tick that claims work snaps back to `intervalMs`. With the defaults, ten idle minutes are 24 ticks and 216 statements instead of 1,201 ticks and 19,216. +- **`MessagingServicePlugin`'s `dispatchMaxIdleIntervalMs` sets the ceiling for both dispatchers**, the way `dispatchIntervalMs` and `partitionCount` already govern both. +- **Writes in this process wake the dispatcher.** `MessagingService.setHttpOutbox(outbox, { onEnqueued })` fires after an `enqueueHttp()` that enqueues a delivery — not one that parks an undeliverable record, which is `dead` on arrival — and after a `redeliverHttp()`. The plugin points it at the new `HttpDispatcher.wake()`, which ticks immediately, or once more right after a tick already in flight. + +**Latency bound.** A delivery enqueued or redelivered in the process that runs the dispatcher goes out on the tick `wake()` starts. While idle, work nobody announces is noticed within one backed-off interval, at most `maxIdleIntervalMs` (30 s by default): + +- a retry coming due is attempted less than `min(its delay + intervalMs, maxIdleIntervalMs)` late, because the backoff restarts from `intervalMs` at the attempt that scheduled it; +- a row enqueued by a process that does not run this dispatcher; +- a crashed node's expired claim, recovered within `claimTtlMs` + `maxIdleIntervalMs` (about 35 s at defaults, where it was about 5.5 s). + +Set `dispatchMaxIdleIntervalMs` to `dispatchIntervalMs` to keep the fixed interval. + +**Contract additions — all optional, nothing to change on upgrade.** `IHttpOutbox` gains an optional `reap(opts: HttpReapOptions)` — the visibility-timeout recovery `claim()` already opens with, as a method of its own — and `HttpClaimOptions` gains an optional `skipReap`. Both built-in stores (`SqlHttpOutbox`, `MemoryHttpOutbox`) implement them. A custom outbox without `reap()` keeps working as it is: the dispatcher probes for the method and, when it is absent, lets each claim reap as before — correct, at the old per-claim cost. Direct callers of `claim()` are unaffected: without `skipReap` they reap exactly as before. Also new: `HttpDispatcher.wake()`, the dispatcher's `maxIdleIntervalMs` option, the `HttpReapOptions` type, and `MessagingService.setHttpOutbox`'s optional second argument. + +**One loop, not two copies.** The timer loop — idle backoff, collapsing wakes into one follow-up tick, `stop()` — moved out of `NotificationDispatcher` into a module both dispatchers share. `NotificationDispatcher`'s behaviour and public surface are unchanged; its #17610 tests pass as they were. diff --git a/content/docs/automation/webhooks.mdx b/content/docs/automation/webhooks.mdx index 10d0784933..f94b92ab59 100644 --- a/content/docs/automation/webhooks.mdx +++ b/content/docs/automation/webhooks.mdx @@ -385,6 +385,18 @@ for each partition, attempts to acquire a **per-partition cluster lock** same partition — useful for in-order delivery and connection reuse. On a single-node runtime the lock is an always-grant stub. +Each tick opens with **one** visibility-timeout reap for the whole table — +`in_flight` rows claimed longer than `claimTtlMs` ago return to `pending` — and +then claims partition by partition, so a tick over an empty outbox costs +`1 + partitionCount` statements. The loop **backs off while idle**: every tick +that claims nothing doubles the delay to the next, from `intervalMs` (default +500 ms) up to `maxIdleIntervalMs` (default 30 s; `dispatchMaxIdleIntervalMs` on +`MessagingServicePlugin`), and a tick that claims work snaps it back. +`messaging.enqueueHttp()` and `messaging.redeliverHttp()` wake the dispatcher +running in the same process, so a delivery written there goes out at once. Work +nobody announces — a retry coming due, a row another process wrote — is noticed +within one backed-off interval, never more than `maxIdleIntervalMs` late. + Within a held partition, the lock-holder claims a batch with an **atomic conditional UPDATE** rather than `SELECT … FOR UPDATE SKIP LOCKED`: @@ -609,7 +621,7 @@ A precise table of what the runtime promises and what it does not. |------------------------------------------|-------------------------------------------------------------| | Producer node crashes mid-emit | **Not durable today.** The realtime bus (`InMemoryRealtimeAdapter`) is an unpersisted, in-process pub/sub — an event lost before Stage 3's INSERT is gone, not redelivered (see §4.1). | | Subscriber node crashes after persist | Row exists in `sys_http_delivery`, another node picks it up. | -| Dispatcher node crashes mid-HTTP | Row stays `in_flight` with `claimed_by`; it reverts to `pending` after the claim TTL and is re-posted. The TTL derives from the dispatcher tick (`intervalMs`, default 500ms): `lockTtlMs = 5 × intervalMs`, `claimTtlMs = 2 × lockTtlMs` (so ~5s at defaults), all configurable via `HttpDispatcherOptions`. | +| Dispatcher node crashes mid-HTTP | Row stays `in_flight` with `claimed_by`; the first dispatcher tick after the claim TTL reverts it to `pending` and it is re-posted. The TTL derives from the dispatcher tick (`intervalMs`, default 500ms): `lockTtlMs = 5 × intervalMs`, `claimTtlMs = 2 × lockTtlMs` (so ~5s at defaults), all configurable via `HttpDispatcherOptions`. An idle surviving dispatcher ticks at least every `maxIdleIntervalMs` (default 30s), so recovery takes at most `claimTtlMs + maxIdleIntervalMs` (~35s at defaults). | | Receiver returns 5xx | Retry per backoff schedule until the fixed 8-attempt budget is exhausted (§4.5). | | Receiver returns 4xx | Treated as terminal — no retry, status `dead` immediately. Exception: 408 / 429 are retried. | | Receiver returns 2xx | `status = success`, no more attempts. | diff --git a/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts b/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts index 281829c059..8b85d07982 100644 --- a/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts +++ b/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts @@ -175,6 +175,28 @@ describe('sys_http_delivery — the dispatcher claim path is a classified global await controlUnscopedUpdateMany(SYS_HTTP_DELIVERY); }); + + it("reap() recovers a crashed node's claims in every organization, without a finding", async () => { + const stale = Date.now() - 10 * 60_000; + await seedHttpRow('h_a', 'org_a', { status: 'in_flight', claimed_by: 'dead_node', claimed_at: stale }); + await seedHttpRow('h_b', 'org_b', { status: 'in_flight', claimed_by: 'dead_node', claimed_at: stale }); + + // [#17623] The dispatcher's once-per-tick reap, on its own — the same + // predicate write `claim()` opens with, classified by the same warrant. + const outbox = new SqlHttpOutbox(engine as any, { partitionCount: 1 }); + await outbox.reap({ claimTtlMs: 60_000 }); + + expect(auditedUpdateMany(SYS_HTTP_DELIVERY)).toBe(false); + // Both organizations' abandoned rows are back in the queue with the claim + // cleared — a per-organization reap would have stranded one. + const rows = (await engine.find(SYS_HTTP_DELIVERY, { where: {} })) as any[]; + expect(rows.map((r) => `${r.id}:${r.organization_id}:${r.status}:${r.claimed_by ?? '-'}`).sort()).toEqual([ + 'h_a:org_a:pending:-', + 'h_b:org_b:pending:-', + ]); + + await controlUnscopedUpdateMany(SYS_HTTP_DELIVERY); + }); }); // ─────────────────────────────────────────────────────────────────────────── diff --git a/packages/services/service-messaging/src/dispatch-loop.ts b/packages/services/service-messaging/src/dispatch-loop.ts new file mode 100644 index 0000000000..9e7946c1d3 --- /dev/null +++ b/packages/services/service-messaging/src/dispatch-loop.ts @@ -0,0 +1,163 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The timer loop both outbox dispatchers run — `NotificationDispatcher` over + * `sys_notification_delivery` and `HttpDispatcher` over `sys_http_delivery`. + * + * #17610 wrote this loop inside `NotificationDispatcher`. #17623 found + * `HttpDispatcher` still on the fixed 500 ms `setInterval` the notification side + * had just left, and moved the loop here, so the two dispatchers run one + * implementation of these rules instead of two copies that can drift: + * + * - **Never two ticks at once.** A tick asked for while one is running (a + * {@link DispatchLoop.wake}, typically) becomes ONE follow-up tick, run the + * moment the running one settles, and every request in that window collapses + * into it. The follow-up is not optional: the running tick may already be + * past the partition the new work hashed into. + * - **Idle backoff.** Each tick that claims nothing doubles the delay to the + * next, from `intervalMs` up to `maxIdleIntervalMs`; a tick that claims + * anything, or a wake, snaps it back to `intervalMs`. A tick that REJECTS + * counts as idle — a failing store is not work, and hammering it helps no + * one. Delays are measured from the START of the previous tick. + * - **`stop()` is final.** It cancels the pending timer and any follow-up + * request, then waits out the running tick; a `wake()` after it is a no-op. + * The timer is `unref()`ed, so the loop never keeps a process alive by + * itself. + * + * The price of the backoff is paid in latency by work nobody wakes the loop + * for — a deferred row coming due, a row another process wrote, a crashed + * node's claim passing its timeout: it is noticed within one backed-off + * interval, never more than `maxIdleIntervalMs` after it became claimable. + */ + +/** Default ceiling of the idle backoff, in ms — see {@link DispatchLoopOptions.maxIdleIntervalMs}. */ +export const DEFAULT_MAX_IDLE_INTERVAL_MS = 30_000; + +export interface DispatchLoopOptions { + /** Delay between ticks while ticks claim work, in ms. */ + intervalMs: number; + /** + * Idle backoff ceiling in ms (default {@link DEFAULT_MAX_IDLE_INTERVAL_MS}). + * A value at or below `intervalMs` disables the backoff. + */ + maxIdleIntervalMs?: number; + /** One full pass over the outbox. Resolves to the number of rows it claimed. */ + runTick: () => Promise; + /** A tick rejected. The loop has already counted it as idle; report it. */ + onTickError: (err: unknown) => void; +} + +export class DispatchLoop { + private readonly intervalMs: number; + private readonly maxIdleIntervalMs: number; + private readonly runTick: () => Promise; + private readonly onTickError: (err: unknown) => void; + private timer: ReturnType | undefined; + private running = false; + private inflightTick: Promise | undefined; + /** Consecutive loop ticks that claimed nothing — the idle backoff's exponent. */ + private idleTicks = 0; + /** A tick was asked for while one was running: run one more the moment it settles. */ + private tickRequested = false; + + constructor(options: DispatchLoopOptions) { + this.intervalMs = options.intervalMs; + // A ceiling below the base interval just means "no backoff". + this.maxIdleIntervalMs = Math.max( + options.intervalMs, + options.maxIdleIntervalMs ?? DEFAULT_MAX_IDLE_INTERVAL_MS, + ); + this.runTick = options.runTick; + this.onTickError = options.onTickError; + } + + /** Begin the loop; the first tick runs immediately. Idempotent. */ + start(): void { + if (this.running) return; + this.running = true; + this.idleTicks = 0; + this.loopTick(); + } + + /** Stop the loop and drain the in-flight tick. */ + async stop(): Promise { + if (!this.running) return; + this.running = false; + this.tickRequested = false; + this.clearTimer(); + if (this.inflightTick) { + try { await this.inflightTick; } catch { /* already reported */ } + } + } + + /** + * Work arrived: tick now — or once more the moment the running tick settles — + * and reset the idle backoff. No-op while stopped. + */ + wake(): void { + if (!this.running) return; + this.idleTicks = 0; + this.loopTick(); + } + + /** + * One tick of the loop, then the timer for the next. Never two at once: a + * call that finds a tick in flight becomes a follow-up request instead. + */ + private loopTick(): void { + if (!this.running) return; + this.clearTimer(); + if (this.inflightTick) { + this.tickRequested = true; + return; + } + const startedAt = Date.now(); + this.inflightTick = this.runTick() + .then((claimed) => { + this.idleTicks = claimed > 0 ? 0 : this.idleTicks + 1; + }) + .catch((err) => { + // A failing store is not work: back off rather than hammer it. + this.idleTicks += 1; + this.onTickError(err); + }) + .finally(() => { + this.inflightTick = undefined; + if (!this.running) return; + if (this.tickRequested) { + this.tickRequested = false; + this.idleTicks = 0; + this.loopTick(); + return; + } + this.schedule(Math.max(0, this.nextIntervalMs() - (Date.now() - startedAt))); + }); + } + + /** + * Delay before the next loop tick, measured from the START of the last one: + * `intervalMs` while ticks claim work, doubled for every consecutive empty + * tick after that, capped at `maxIdleIntervalMs`. + */ + private nextIntervalMs(): number { + if (this.idleTicks === 0) return this.intervalMs; + // Exponent clamped so the product stays finite long after the cap wins. + return Math.min(this.maxIdleIntervalMs, this.intervalMs * 2 ** Math.min(this.idleTicks, 30)); + } + + private schedule(delayMs: number): void { + this.clearTimer(); + this.timer = setTimeout(() => { + this.timer = undefined; + this.loopTick(); + }, delayMs); + // Don't keep the event loop alive solely for the dispatcher. + (this.timer as { unref?: () => void })?.unref?.(); + } + + private clearTimer(): void { + if (this.timer === undefined) return; + clearTimeout(this.timer); + this.timer = undefined; + } +} diff --git a/packages/services/service-messaging/src/dispatcher.ts b/packages/services/service-messaging/src/dispatcher.ts index bcd2b32267..3e74d15705 100644 --- a/packages/services/service-messaging/src/dispatcher.ts +++ b/packages/services/service-messaging/src/dispatcher.ts @@ -4,6 +4,7 @@ import type { MessagingChannel, MessagingChannelContext, Notification, SendResul import type { AckResult, ClaimedDeliveryRecord, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; import { classifyDeliveryAttempt } from './backoff.js'; import { renderDigest } from './digest-render.js'; +import { DispatchLoop } from './dispatch-loop.js'; /** Minimal channel-registry surface the dispatcher needs (MessagingService satisfies it). */ export interface ChannelRegistry { @@ -45,9 +46,10 @@ export interface NotificationDispatcherLogger { /** * [#17610] Default ceiling of the idle backoff, in ms — see - * {@link NotificationDispatcherOptions.maxIdleIntervalMs}. + * {@link NotificationDispatcherOptions.maxIdleIntervalMs}. It lives with the + * loop both dispatchers run (#17623) and stays exported from here. */ -export const DEFAULT_MAX_IDLE_INTERVAL_MS = 30_000; +export { DEFAULT_MAX_IDLE_INTERVAL_MS } from './dispatch-loop.js'; export interface NotificationDispatcherOptions { nodeId: string; @@ -112,16 +114,11 @@ export interface NotificationDispatcherOptions { */ export class NotificationDispatcher { private readonly opts: Required< - Omit + Omit > & Pick & { cluster: DispatchCluster }; - private timer: ReturnType | undefined; - private running = false; - private inflightTick: Promise | undefined; - /** [#17610] Consecutive loop ticks that claimed nothing — the idle backoff's exponent. */ - private idleTicks = 0; - /** [#17610] A tick was asked for while one was running: run one more the moment it settles. */ - private tickRequested = false; + /** [#17610, #17623] The timer loop — idle backoff, wake, stop — shared with `HttpDispatcher`. */ + private readonly loop: DispatchLoop; constructor(options: NotificationDispatcherOptions) { const intervalMs = options.intervalMs ?? 500; @@ -135,8 +132,6 @@ export class NotificationDispatcher { partitionCount: options.partitionCount ?? 8, batchSize: options.batchSize ?? 32, intervalMs, - // A ceiling below the base interval just means "no backoff". - maxIdleIntervalMs: Math.max(intervalMs, options.maxIdleIntervalMs ?? DEFAULT_MAX_IDLE_INTERVAL_MS), lockTtlMs, claimTtlMs: options.claimTtlMs ?? lockTtlMs * 2, rng: options.rng, @@ -144,25 +139,27 @@ export class NotificationDispatcher { logger: options.logger, onAttempt: options.onAttempt, }; + this.loop = new DispatchLoop({ + intervalMs, + maxIdleIntervalMs: options.maxIdleIntervalMs, + runTick: () => this.runTick(), + onTickError: (err) => { + this.opts.logger?.warn?.('notification-dispatcher: tick failed', { + nodeId: this.opts.nodeId, + error: (err as Error)?.message ?? String(err), + }); + }, + }); } /** Begin the loop; the first tick runs immediately. Idempotent. */ start(): void { - if (this.running) return; - this.running = true; - this.idleTicks = 0; - this.loopTick(); + this.loop.start(); } /** Stop the loop and drain the in-flight tick. */ async stop(): Promise { - if (!this.running) return; - this.running = false; - this.tickRequested = false; - this.clearTimer(); - if (this.inflightTick) { - try { await this.inflightTick; } catch { /* already logged */ } - } + await this.loop.stop(); } /** @@ -176,9 +173,7 @@ export class NotificationDispatcher { * collapses into that one. No-op while stopped. */ wake(): void { - if (!this.running) return; - this.idleTicks = 0; - this.loopTick(); + this.loop.wake(); } /** Run one full tick (the reap, then all partitions). Exposed for deterministic tests. */ @@ -186,71 +181,6 @@ export class NotificationDispatcher { await this.runTick(); } - /** - * One tick of the loop, then the timer for the next. Never two at once: a - * call that finds a tick in flight becomes a follow-up request instead. - */ - private loopTick(): void { - if (!this.running) return; - this.clearTimer(); - if (this.inflightTick) { - this.tickRequested = true; - return; - } - const startedAt = Date.now(); - this.inflightTick = this.runTick() - .then((claimed) => { - this.idleTicks = claimed > 0 ? 0 : this.idleTicks + 1; - }) - .catch((err) => { - // A failing store is not work: back off rather than hammer it. - this.idleTicks += 1; - this.opts.logger?.warn?.('notification-dispatcher: tick failed', { - nodeId: this.opts.nodeId, - error: (err as Error)?.message ?? String(err), - }); - }) - .finally(() => { - this.inflightTick = undefined; - if (!this.running) return; - if (this.tickRequested) { - this.tickRequested = false; - this.idleTicks = 0; - this.loopTick(); - return; - } - this.schedule(Math.max(0, this.nextIntervalMs() - (Date.now() - startedAt))); - }); - } - - /** - * [#17610] Delay before the next loop tick, measured from the START of the - * last one: `intervalMs` while ticks claim work, doubled for every - * consecutive empty tick after that, capped at `maxIdleIntervalMs`. - */ - private nextIntervalMs(): number { - const { intervalMs, maxIdleIntervalMs } = this.opts; - if (this.idleTicks === 0) return intervalMs; - // Exponent clamped so the product stays finite long after the cap wins. - return Math.min(maxIdleIntervalMs, intervalMs * 2 ** Math.min(this.idleTicks, 30)); - } - - private schedule(delayMs: number): void { - this.clearTimer(); - this.timer = setTimeout(() => { - this.timer = undefined; - this.loopTick(); - }, delayMs); - // Don't keep the event loop alive solely for the dispatcher. - (this.timer as { unref?: () => void })?.unref?.(); - } - - private clearTimer(): void { - if (this.timer === undefined) return; - clearTimeout(this.timer); - this.timer = undefined; - } - /** One full pass: the reap, then every partition. Resolves to the rows claimed. */ private async runTick(): Promise { // [#17610] Visibility-timeout recovery ONCE per tick, BEFORE any claim. diff --git a/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts b/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts new file mode 100644 index 0000000000..b798c59cee --- /dev/null +++ b/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts @@ -0,0 +1,314 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #17623 — the `HttpDispatcher` loop backs off while `sys_http_delivery` is + * idle, and new work wakes it. + * + * The loop used to tick every `intervalMs` (500 ms) forever, whatever it found — + * the shape #17610 removed from `NotificationDispatcher`. Both dispatchers now + * run the one `DispatchLoop` (`dispatch-loop.ts`), whose mechanics are pinned in + * full through the notification side in `dispatcher-idle-backoff.test.ts`. This + * file pins the same behaviour through `HttpDispatcher`, plus what only the HTTP + * side has: a retry schedule the backoff must not starve, and its own ingress — + * `MessagingService.enqueueHttp()` and `redeliverHttp()` — waking it. + * + * Every leg runs on vitest's fake timers, so "when did a tick start" is an + * exact reading rather than a sleep-and-hope. With one partition the dispatcher + * issues exactly one `claim()` per tick, so a tick is timestamped there. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { MemoryHttpOutbox } from './memory-http-outbox.js'; +import { HttpDispatcher } from './http-dispatcher.js'; +import { DEFAULT_MAX_IDLE_INTERVAL_MS } from './dispatcher.js'; +import { MessagingService } from './messaging-service.js'; +import type { FetchImpl } from './http-sender.js'; +import type { EnqueueHttpInput, HttpAckResult, HttpClaimOptions, HttpDelivery } from './http-outbox.js'; + +const BASE = 500; +const CAP = 30_000; +const MINUTE = 60_000; + +const silentLogger = { info() {}, warn() {}, error() {} }; + +/** A memory outbox that records the (fake) instant every tick starts, and every retry it is told to schedule. */ +class TickRecordingOutbox extends MemoryHttpOutbox { + readonly tickStarts: number[] = []; + readonly retriesDueAt: number[] = []; + override async claim(opts: HttpClaimOptions): Promise { + this.tickStarts.push(Date.now()); + return super.claim(opts); + } + override async ack(id: string, result: HttpAckResult): Promise { + if (!result.success && result.nextRetryAt !== undefined) this.retriesDueAt.push(result.nextRetryAt); + return super.ack(id, result); + } +} + +function delivery(refId: string, over: Partial = {}): EnqueueHttpInput { + return { source: 'flow', refId, dedupKey: `d_${refId}`, url: urlOf(refId), payload: { refId }, ...over }; +} + +function urlOf(refId: string): string { + return `https://receiver.example/${refId}`; +} + +function gaps(ts: readonly number[]): number[] { + return ts.slice(1).map((t, i) => t - ts[i]); +} + +function last(ts: readonly number[]): number { + return ts[ts.length - 1]; +} + +/** A 200-answering fetch that records every POST; `posts(n)` resolves once n have gone out. */ +interface FetchProbe { + readonly posted: string[]; + posts(n: number): Promise; + impl: FetchImpl; +} + +function fetchProbe(hold?: { url: string; until: Promise }): FetchProbe { + const posted: string[] = []; + let waiters: Array<{ n: number; resolve: () => void }> = []; + return { + posted, + posts(n) { + if (posted.length >= n) return Promise.resolve(); + return new Promise((resolve) => waiters.push({ n, resolve })); + }, + impl: async (url) => { + posted.push(url); + waiters = waiters.filter((w) => (posted.length >= w.n ? (w.resolve(), false) : true)); + if (hold && url === hold.url) await hold.until; + return { ok: true, status: 200, async text() { return 'ok'; } }; + }, + }; +} + +function setup(options: { maxIdleIntervalMs?: number; probe?: FetchProbe } = {}) { + const outbox = new TickRecordingOutbox(); + const probe = options.probe ?? fetchProbe(); + const dispatcher = new HttpDispatcher({ + nodeId: 'node-test', + outbox, + fetchImpl: probe.impl, + partitionCount: 1, + intervalMs: BASE, + maxIdleIntervalMs: options.maxIdleIntervalMs, + }); + return { outbox, probe, dispatcher }; +} + +beforeEach(() => { vi.useFakeTimers(); }); +afterEach(() => { vi.useRealTimers(); }); + +describe('#17623 HttpDispatcher — idle backoff', () => { + it('doubles the interval on every empty tick, up to maxIdleIntervalMs', async () => { + const { outbox, dispatcher } = setup({ maxIdleIntervalMs: CAP }); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(10 * MINUTE); + await dispatcher.stop(); + + const g = gaps(outbox.tickStarts); + expect(g.slice(0, 6)).toEqual([1_000, 2_000, 4_000, 8_000, 16_000, CAP]); + expect(g.slice(5).every((gap) => gap === CAP)).toBe(true); + // Ten idle minutes: 24 ticks, where the fixed 500 ms loop ran 1,201. + expect(outbox.tickStarts).toHaveLength(24); + }); + + it('defaults the ceiling to 30 s — the notification dispatcher’s default', async () => { + const { outbox, dispatcher } = setup(); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5 * MINUTE); + await dispatcher.stop(); + expect(DEFAULT_MAX_IDLE_INTERVAL_MS).toBe(30_000); + expect(Math.max(...gaps(outbox.tickStarts))).toBe(DEFAULT_MAX_IDLE_INTERVAL_MS); + }); + + it('a ceiling at or below intervalMs disables the backoff', async () => { + const { outbox, dispatcher } = setup({ maxIdleIntervalMs: BASE }); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5_000); + await dispatcher.stop(); + expect(outbox.tickStarts).toHaveLength(11); + expect(new Set(gaps(outbox.tickStarts))).toEqual(new Set([BASE])); + }); + + it('a row nobody announced waits at most one backed-off interval, and the tick that claims it resets the backoff', async () => { + const { outbox, probe, dispatcher } = setup({ maxIdleIntervalMs: CAP }); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5 * MINUTE); // fully backed off + + // Written without a wake() — as a row enqueued by another process is. + const enqueuedAt = Date.now(); + await outbox.enqueue(delivery('r1')); + await vi.advanceTimersByTimeAsync(CAP + BASE); + await dispatcher.stop(); + + expect(probe.posted).toEqual([urlOf('r1')]); + const claimedAt = outbox.tickStarts.findIndex((t) => t >= enqueuedAt); + // The latency bound #17623 trades for the idle savings: one ceiling. + expect(outbox.tickStarts[claimedAt] - enqueuedAt).toBeLessThanOrEqual(CAP); + // …and work found means the very next tick is back on the base interval. + expect(outbox.tickStarts[claimedAt + 1] - outbox.tickStarts[claimedAt]).toBe(BASE); + }); + + it('a retry coming due while backed off is attempted less than min(its delay + intervalMs, maxIdleIntervalMs) late', async () => { + const outbox = new TickRecordingOutbox(); + const attemptedAt: number[] = []; + const statuses = [500, 500, 500, 200]; + const fetchImpl: FetchImpl = async () => { + const status = statuses[Math.min(attemptedAt.length, statuses.length - 1)]; + attemptedAt.push(Date.now()); + return { ok: status < 400, status, async text() { return ''; } }; + }; + const dispatcher = new HttpDispatcher({ + nodeId: 'node-test', outbox, fetchImpl, partitionCount: 1, intervalMs: BASE, maxIdleIntervalMs: CAP, + rng: () => 0.5, // no jitter: the retry delays are exactly 1 s, 10 s, 60 s + }); + await outbox.enqueue(delivery('r1')); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(3 * MINUTE); + await dispatcher.stop(); + + expect(attemptedAt).toHaveLength(4); + expect(outbox.retriesDueAt).toHaveLength(3); + for (let i = 0; i < 3; i++) { + const delay = outbox.retriesDueAt[i] - attemptedAt[i]; + const late = attemptedAt[i + 1] - outbox.retriesDueAt[i]; + expect(late).toBeGreaterThanOrEqual(0); // never before it is due + expect(late).toBeLessThan(Math.min(delay + BASE, CAP)); + } + expect((await outbox.list())[0]).toMatchObject({ status: 'success', attempts: 4 }); + }); +}); + +describe('#17623 HttpDispatcher — wake()', () => { + it('ticks immediately, not at the next backed-off slot, and restarts from intervalMs', async () => { + const { outbox, probe, dispatcher } = setup({ maxIdleIntervalMs: CAP }); + dispatcher.start(); + // 5 min 10 s: between two backed-off ticks, the next one 20 s away. + await vi.advanceTimersByTimeAsync(5 * MINUTE + 10_000); + const ticksBefore = outbox.tickStarts.length; + + await outbox.enqueue(delivery('r1')); + dispatcher.wake(); + // No timer time passes here — only the woken tick's own promise chain. + await probe.posts(1); + expect(outbox.tickStarts).toHaveLength(ticksBefore + 1); + expect(last(outbox.tickStarts)).toBe(Date.now()); + + await vi.advanceTimersByTimeAsync(BASE); + await dispatcher.stop(); + expect(outbox.tickStarts).toHaveLength(ticksBefore + 2); + expect(outbox.tickStarts[ticksBefore + 1] - outbox.tickStarts[ticksBefore]).toBe(BASE); + }); + + it('wakes during a running tick collapse into ONE follow-up tick, run the moment it settles', async () => { + let release!: () => void; + const until = new Promise((resolve) => { release = resolve; }); + const probe = fetchProbe({ url: urlOf('r1'), until }); + const { outbox, dispatcher } = setup({ maxIdleIntervalMs: CAP, probe }); + + await outbox.enqueue(delivery('r1')); + dispatcher.start(); + await probe.posts(1); // tick 1 is now parked inside r1's POST + + // r2 lands after tick 1 has already claimed its batch. + await outbox.enqueue(delivery('r2')); + dispatcher.wake(); + dispatcher.wake(); + dispatcher.wake(); + expect(outbox.tickStarts).toHaveLength(1); // never two ticks at once + + release(); + await probe.posts(2); + // Exactly one follow-up, at the same instant tick 1 settled. + expect(outbox.tickStarts).toHaveLength(2); + expect(outbox.tickStarts[1]).toBe(outbox.tickStarts[0]); + + // Three wakes did not queue three ticks: the next one is a base interval out. + await vi.advanceTimersByTimeAsync(BASE - 1); + expect(outbox.tickStarts).toHaveLength(2); + await dispatcher.stop(); + }); + + it('stop() cancels the pending backed-off tick, and a wake() after stop is a no-op', async () => { + const { outbox, dispatcher } = setup({ maxIdleIntervalMs: CAP }); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(MINUTE); + await dispatcher.stop(); + const ticksAtStop = outbox.tickStarts.length; + + dispatcher.wake(); + await vi.advanceTimersByTimeAsync(10 * MINUTE); + expect(outbox.tickStarts).toHaveLength(ticksAtStop); + }); +}); + +describe('#17623 MessagingService wakes the HTTP dispatcher its outbox is wired to', () => { + function wiredStack() { + const outbox = new TickRecordingOutbox(); + const probe = fetchProbe(); + const service = new MessagingService({ logger: silentLogger }); + const dispatcher = new HttpDispatcher({ + nodeId: 'node-test', + outbox, + fetchImpl: probe.impl, + partitionCount: 1, + intervalMs: BASE, + maxIdleIntervalMs: CAP, + }); + // The seam MessagingServicePlugin wires. + service.setHttpOutbox(outbox, { onEnqueued: () => dispatcher.wake() }); + return { outbox, probe, service, dispatcher }; + } + + it('an enqueueHttp() delivery is POSTed at once by a backed-off dispatcher', async () => { + const { outbox, probe, service, dispatcher } = wiredStack(); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5 * MINUTE + 10_000); + const ticksBefore = outbox.tickStarts.length; + + await service.enqueueHttp(delivery('r1')); + + await probe.posts(1); + expect(probe.posted).toEqual([urlOf('r1')]); + expect(outbox.tickStarts).toHaveLength(ticksBefore + 1); + expect(last(outbox.tickStarts)).toBe(Date.now()); + await dispatcher.stop(); + }); + + it('an enqueueHttp() that PARKS an undeliverable row does not wake it — there is nothing to send', async () => { + const { outbox, service, dispatcher } = wiredStack(); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5 * MINUTE + 10_000); + const ticksBefore = outbox.tickStarts.length; + + await service.enqueueHttp(delivery('r1', { undeliverableReason: 'signing secret could not be resolved' })); + + await vi.advanceTimersByTimeAsync(0); + expect(outbox.tickStarts).toHaveLength(ticksBefore); + expect((await outbox.list())[0]).toMatchObject({ status: 'dead', attempts: 0 }); + await dispatcher.stop(); + }); + + it('a redeliverHttp() puts the row back to pending and wakes the dispatcher to re-send it', async () => { + const { outbox, probe, service, dispatcher } = wiredStack(); + dispatcher.start(); + const id = await service.enqueueHttp(delivery('r1')); + await probe.posts(1); + await vi.advanceTimersByTimeAsync(5 * MINUTE + 10_000); + expect((await outbox.list())[0]).toMatchObject({ status: 'success', attempts: 1 }); + const ticksBefore = outbox.tickStarts.length; + + await service.redeliverHttp(id, { tenantId: undefined }); + + await probe.posts(2); + expect(probe.posted).toEqual([urlOf('r1'), urlOf('r1')]); + expect(outbox.tickStarts).toHaveLength(ticksBefore + 1); + expect(last(outbox.tickStarts)).toBe(Date.now()); + await dispatcher.stop(); + }); +}); diff --git a/packages/services/service-messaging/src/http-dispatcher-idle-cost.integration.test.ts b/packages/services/service-messaging/src/http-dispatcher-idle-cost.integration.test.ts new file mode 100644 index 0000000000..af74f3364a --- /dev/null +++ b/packages/services/service-messaging/src/http-dispatcher-idle-cost.integration.test.ts @@ -0,0 +1,199 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #17623 — what an IDLE `HttpDispatcher` tick costs the store, and that cutting + * it keeps every delivery guarantee. + * + * ## The measurement this pins + * + * Before #17623 one tick over an EMPTY `sys_http_delivery` outbox issued 16 + * statements with the default 8 partitions: every partition's `claim()` opened + * with the environment-wide visibility-timeout reap — 8 identical UPDATEs — + * before its candidate SELECT (8). On remote Turso every statement is an HTTP + * round trip, and the loop never stopped. The reap now runs once per tick, so + * an idle tick is `1 + partitionCount` = 9. It is the shape #17610 removed from + * `NotificationDispatcher`, which pays one more probe per partition for its + * digest pass. + * + * ## Why a real engine + * + * The count is taken on the `IDataEngine` the outbox talks to — the boundary + * this package controls — over a real `ObjectQL` + `SqlDriver` (better-sqlite3 + * `:memory:`), so every counted call is a statement the production outbox + * really issues. On this harness one engine call is one SQL statement. + * + * ## The vacuity traps closed here + * + * - **An upper bound alone is satisfied by a dispatcher that does nothing.** + * The reap count is pinned EXACTLY — once per tick, because zero would + * strand a crashed node's rows forever — and the legs below prove the very + * same harness claims, POSTs and recovers. + * - **"Reap once" could be bought by recovering less.** The recovery leg drives + * a real expired claim through ONE tick; its negative leg proves a claim + * that has not expired is left alone, so the TTL is still a floor. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { SqlHttpOutbox } from './sql-http-outbox.js'; +import { HttpDelivery, SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js'; +import { HttpDispatcher } from './http-dispatcher.js'; +import { hashPartition } from './backoff.js'; +import type { FetchImpl } from './http-sender.js'; +import type { EnqueueHttpInput, IHttpOutbox } from './http-outbox.js'; + +/** The production default. */ +const PARTITIONS = 8; +const TICKS = 10; +const TTL = 60_000; + +let engine: ObjectQL; +let outbox: SqlHttpOutbox; +/** Engine calls against the delivery table, by method. */ +let calls: { update: number; find: number; other: number }; +/** URL of every POST, in order. */ +let posted: string[]; + +const recordingFetch: FetchImpl = async (url) => { + posted.push(url); + return { ok: true, status: 200, async text() { return 'ok'; } }; +}; + +function dispatcher(store: IHttpOutbox = outbox): HttpDispatcher { + return new HttpDispatcher({ + nodeId: 'node-live', + outbox: store, + fetchImpl: recordingFetch, + partitionCount: PARTITIONS, + claimTtlMs: TTL, + intervalMs: 10_000, // ticks are driven manually + }); +} + +function delivery(refId: string): EnqueueHttpInput { + return { source: 'flow', refId, dedupKey: `d_${refId}`, url: `https://receiver.example/${refId}`, payload: { refId } }; +} + +beforeEach(async () => { + const driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + engine = new ObjectQL(); + engine.registerDriver(driver, true); + await engine.init(); + engine.registry.registerObject(HttpDelivery as any, '@objectstack/service-messaging'); + await engine.syncSchemas(); + + calls = { update: 0, find: 0, other: 0 }; + posted = []; + // Count on the engine instance the outbox holds, so the tally is of real + // outbox traffic rather than a stand-in's. + type EngineCall = (name: string, ...rest: unknown[]) => unknown; + const e = engine as unknown as Record; + for (const method of ['find', 'findOne', 'update', 'insert', 'delete', 'count', 'resolveInternalField'] as const) { + if (typeof e[method] !== 'function') continue; + const orig = e[method].bind(engine); + e[method] = (name: string, ...rest: unknown[]) => { + if (name === SYS_HTTP_DELIVERY) { + if (method === 'update') calls.update++; + else if (method === 'find') calls.find++; + else calls.other++; + } + return orig(name, ...rest); + }; + } + outbox = new SqlHttpOutbox(engine as any, { partitionCount: PARTITIONS }); +}); + +afterEach(async () => { + try { await engine?.destroy(); } catch { /* noop */ } +}); + +describe('#17623 HttpDispatcher — idle tick cost', () => { + it('an idle tick is 1 + partitionCount statements: ONE reap, then one candidate probe per partition', async () => { + const d = dispatcher(); + for (let i = 0; i < TICKS; i++) await d.tick(); + + // On an empty outbox the only UPDATE a tick issues is the reap. It is + // environment-wide, so once per tick is all it can use — and exactly + // once, never zero: it is the only crash recovery there is. + expect(calls).toEqual({ update: TICKS, find: TICKS * PARTITIONS, other: 0 }); + // Before #17623: PARTITIONS reaps + PARTITIONS probes = 16 a tick. + expect(calls.update + calls.find + calls.other).toBeLessThanOrEqual(TICKS * (1 + PARTITIONS)); + }); + + it('rows enqueued after an idle stretch go out on the very next tick, in every partition', async () => { + const d = dispatcher(); + for (let i = 0; i < TICKS; i++) await d.tick(); + expect(posted).toEqual([]); + + // Enough distinct anchors that they hash across several of the 8 + // partitions: the next tick must drain all of them, wherever they landed. + const refIds = Array.from({ length: 16 }, (_, i) => `r${i}`); + for (const refId of refIds) await outbox.enqueue(delivery(refId)); + expect(new Set(refIds.map((r) => hashPartition(r, PARTITIONS))).size).toBeGreaterThan(1); + + await d.tick(); + + expect([...posted].sort()).toEqual(refIds.map((r) => `https://receiver.example/${r}`).sort()); + const rows = await outbox.list(); + expect(rows).toHaveLength(16); + expect(rows.every((r) => r.status === 'success' && r.attempts === 1)).toBe(true); + }); + + it("recovers a crashed node's expired claim and delivers it within ONE tick", async () => { + await outbox.enqueue(delivery('r_crashed')); + // A node claims the row and dies: its claim was stamped TTL + 1 ms ago. + const [abandoned] = await outbox.claim({ + nodeId: 'node-crashed', limit: 10, claimTtlMs: TTL, now: Date.now() - TTL - 1, + }); + expect(abandoned?.status).toBe('in_flight'); + + await dispatcher().tick(); + + // The tick's single reap ran BEFORE its claims, so the same tick took + // the row back and delivered it. + expect(posted).toEqual(['https://receiver.example/r_crashed']); + const [row] = await outbox.list(); + expect(row).toMatchObject({ status: 'success', attempts: 1 }); + }); + + it('leaves a claim that has NOT expired alone — the TTL is still a floor', async () => { + await outbox.enqueue(delivery('r_live')); + // Another node's claim, 5 s inside its visibility timeout. + await outbox.claim({ nodeId: 'node-busy', limit: 10, claimTtlMs: TTL, now: Date.now() - TTL + 5_000 }); + + await dispatcher().tick(); + + expect(posted).toEqual([]); + const [row] = await outbox.list(); + expect(row).toMatchObject({ status: 'in_flight', claimedBy: 'node-busy', attempts: 0 }); + }); + + it('an outbox without reap() keeps working: its claims keep reaping, and an expired claim is still recovered', async () => { + // The shape of a store written before `reap()` existed: the same SQL + // store underneath, with that one method not exposed. + const skipReapSeen: unknown[] = []; + const legacy: IHttpOutbox = { + enqueue: (input) => outbox.enqueue(input), + recordUndeliverable: (input) => outbox.recordUndeliverable(input), + claim: (opts) => { skipReapSeen.push(opts.skipReap); return outbox.claim(opts); }, + ack: (id, result) => outbox.ack(id, result), + list: (filter) => outbox.list(filter), + redeliver: (id, options) => outbox.redeliver(id, options), + }; + await outbox.enqueue(delivery('r_legacy')); + await outbox.claim({ nodeId: 'node-crashed', limit: 10, claimTtlMs: TTL, now: Date.now() - TTL - 1 }); + + await dispatcher(legacy).tick(); + + // No claim was told to skip the reap it is now the only source of… + expect(skipReapSeen).toHaveLength(PARTITIONS); + expect(skipReapSeen.every((skip) => skip !== true)).toBe(true); + // …so the expired claim is still recovered and delivered in the same tick. + expect(posted).toEqual(['https://receiver.example/r_legacy']); + }); +}); diff --git a/packages/services/service-messaging/src/http-dispatcher.ts b/packages/services/service-messaging/src/http-dispatcher.ts index 689219fc12..fdec977f13 100644 --- a/packages/services/service-messaging/src/http-dispatcher.ts +++ b/packages/services/service-messaging/src/http-dispatcher.ts @@ -1,6 +1,7 @@ // Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. import type { DispatchCluster, DispatchLockHandle } from './dispatcher.js'; +import { DispatchLoop } from './dispatch-loop.js'; import { classifyAttempt, sendOnce, type FetchImpl } from './http-sender.js'; import type { HttpDelivery, IHttpOutbox } from './http-outbox.js'; @@ -9,8 +10,8 @@ import type { HttpDelivery, IHttpOutbox } from './http-outbox.js'; * (`sys_http_delivery`) and POSTs each row, retrying with backoff and * dead-lettering once the budget is exhausted. * - * Structurally identical to `NotificationDispatcher` / `WebhookDispatcher`: an - * interval loop walks `partitionCount` partitions, each guarded by a + * Structurally identical to `NotificationDispatcher` / `WebhookDispatcher`: a + * timer loop walks `partitionCount` partitions, each guarded by a * per-partition cluster lock; within a held partition it claims a batch * (`pending → in_flight`), sends, and acks. Partition affinity is on the * delivery's `refId`, preserving in-order delivery per source anchor. @@ -18,6 +19,20 @@ import type { HttpDelivery, IHttpOutbox } from './http-outbox.js'; * At-least-once: if the POST succeeds but the ack write fails, the row reverts * to pending after the claim TTL and is re-posted. Receivers MUST be idempotent * on the `X-Objectstack-Delivery` (== row id) header. + * + * ## What an idle tick costs (#17623) + * + * Against an empty outbox one tick is `1 + partitionCount` store round trips: + * ONE visibility-timeout reap for the whole environment, then a candidate probe + * per partition. The reap used to open every partition's claim — + * `partitionCount` identical environment-wide UPDATEs a tick, 8 of the 16 + * statements a tick issued with the default 8 partitions — on a fixed 500 ms + * interval that never let up, one loop per warm kernel: the shape #17610 had + * just removed from `NotificationDispatcher`. Both dispatchers now run the same + * {@link DispatchLoop}, which backs off while idle + * ({@link HttpDispatcherOptions.maxIdleIntervalMs}); {@link HttpDispatcher.wake} + * — which the messaging service calls when `enqueueHttp()` or + * `redeliverHttp()` leaves a row pending — runs the next tick at once. */ const SINGLE_NODE_CLUSTER: DispatchCluster = { @@ -44,8 +59,26 @@ export interface HttpDispatcherOptions { partitionCount?: number; /** Max rows to claim from each partition per tick. Default 32. */ batchSize?: number; - /** Tick interval in ms. Default 500. */ + /** Tick interval in ms while ticks find work. Default 500. */ intervalMs?: number; + /** + * [#17623] Idle backoff ceiling in ms (default `DEFAULT_MAX_IDLE_INTERVAL_MS`, + * 30 s — the notification dispatcher's). Each loop tick that claims nothing + * doubles the delay before the next one, from `intervalMs` up to this; a + * tick that claims anything, or a {@link HttpDispatcher.wake} call, snaps it + * back to `intervalMs`. A value at or below `intervalMs` disables the backoff. + * + * While idle this bounds how late the loop notices work nobody woke it for: + * - a retry coming due — attempted less than `min(its delay + intervalMs, + * maxIdleIntervalMs)` after it is due (plus the time ticks themselves + * take), because the backoff restarts from `intervalMs` at the attempt + * that scheduled the retry; + * - a row enqueued by a process this dispatcher does not serve; + * - a crashed node's `in_flight` rows — reaped at most `claimTtlMs` + this + * after their claim, where the fixed interval gave `claimTtlMs` + + * `intervalMs`. + */ + maxIdleIntervalMs?: number; /** Per-partition lock TTL. Default = 5 × intervalMs. */ lockTtlMs?: number; /** Visibility timeout for claimed rows. Default = 2 × lockTtlMs. */ @@ -64,14 +97,13 @@ export interface HttpDispatcherOptions { export class HttpDispatcher { private readonly opts: Required< - Omit + Omit > & Pick & { cluster: DispatchCluster; }; - private timer: ReturnType | undefined; - private running = false; - private inflightTick: Promise | undefined; + /** [#17623] The timer loop — idle backoff, wake, stop — shared with `NotificationDispatcher`. */ + private readonly loop: DispatchLoop; constructor(options: HttpDispatcherOptions) { const intervalMs = options.intervalMs ?? 500; @@ -91,69 +123,98 @@ export class HttpDispatcher { logger: options.logger, onAttempt: options.onAttempt, }; + this.loop = new DispatchLoop({ + intervalMs, + maxIdleIntervalMs: options.maxIdleIntervalMs, + runTick: () => this.runTick(), + onTickError: (err) => { + this.opts.logger?.warn?.('http-dispatcher: tick failed', { + nodeId: this.opts.nodeId, + error: (err as Error)?.message ?? String(err), + }); + }, + }); } - /** Begin the periodic loop. Safe to call once; subsequent calls are no-ops. */ + /** Begin the loop; the first tick runs immediately. Safe to call once; subsequent calls are no-ops. */ start(): void { - if (this.running) return; - this.running = true; - this.scheduleTick(); - this.timer = setInterval(() => this.scheduleTick(), this.opts.intervalMs); - this.timer.unref?.(); + this.loop.start(); } /** Stop the loop and wait for the in-flight tick to drain. */ async stop(): Promise { - if (!this.running) return; - this.running = false; - if (this.timer) { - clearInterval(this.timer); - this.timer = undefined; - } - if (this.inflightTick) { - try { - await this.inflightTick; - } catch { - /* swallow — already logged */ - } - } + await this.loop.stop(); + } + + /** + * [#17623] Work was just written: tick now and reset the idle backoff. + * + * The messaging service calls this after `enqueueHttp()` enqueues a + * delivery and after `redeliverHttp()` resets one, so a callout raised in + * this process never waits out a backed-off interval. A wake that lands + * while a tick is running queues ONE follow-up tick for the moment it + * settles — the running tick may already be past the partition the row + * hashed into — and every wake in that window collapses into that one. + * No-op while stopped. + */ + wake(): void { + this.loop.wake(); } - /** Run one full tick (all partitions). Exposed for deterministic tests. */ + /** Run one full tick (the reap, then all partitions). Exposed for deterministic tests. */ async tick(): Promise { await this.runTick(); } - private scheduleTick(): void { - if (this.inflightTick) return; - this.inflightTick = this.runTick() - .catch((err) => { - this.opts.logger?.warn?.('http-dispatcher: tick failed', { - nodeId: this.opts.nodeId, - error: (err as Error)?.message ?? String(err), - }); - }) - .finally(() => { - this.inflightTick = undefined; - }); - } + /** One full pass: the reap, then every partition. Resolves to the rows claimed. */ + private async runTick(): Promise { + // [#17623] Visibility-timeout recovery ONCE per tick, BEFORE any claim — + // the arrangement #17610 made for notifications. The reap's predicate + // names no partition, so this one run hands every claim below each row + // that had already expired when the tick began: what reaping inside every + // partition's claim achieved, less the rows that expire DURING this tick, + // which the next tick's reap returns. An abandoned claim is still + // recovered within one tick of `claimTtlMs` passing (one backed-off tick + // while idle, see `maxIdleIntervalMs`), and never before its TTL. + // + // No partition lock is needed, and none was ever in force: the per-claim + // reap, run under partition p's lock, was already rewriting expired rows + // in every other partition. The reap moves only rows past their timeout + // and a claim takes only `pending` rows, so running it here opens no + // interleaving the claim TTL did not already allow. + // + // `reap` is optional on the outbox contract, so a store written before it + // keeps working: without it every claim keeps reaping as it always did — + // correct, at the per-claim cost. + const { outbox } = this.opts; + let reapedForTick = false; + if (outbox.reap) { + await outbox.reap({ claimTtlMs: this.opts.claimTtlMs, now: this.opts.now?.() }); + reapedForTick = true; + } - private async runTick(): Promise { const partitionCount = this.opts.partitionCount; const offset = stableNodeOffset(this.opts.nodeId, partitionCount); + let claimed = 0; for (let step = 0; step < partitionCount; step++) { const i = (offset + step) % partitionCount; - await this.runPartition(i); + claimed += await this.runPartition(i, reapedForTick); } + return claimed; } - private async runPartition(index: number): Promise { + /** + * Claim and POST within one partition's lock. Resolves to the number of rows + * claimed — 0 when another node holds the lock. `skipReap` is true when this + * tick already ran the outbox's `reap()`. + */ + private async runPartition(index: number, skipReap: boolean): Promise { const key = `http.dispatcher.partition.${index}`; const handle: DispatchLockHandle | null = await this.opts.cluster.lock.acquire(key, { ttlMs: this.opts.lockTtlMs, waitMs: 0, }); - if (!handle) return; + if (!handle) return 0; try { const claimed = await this.opts.outbox.claim({ @@ -162,13 +223,17 @@ export class HttpDispatcher { partition: { index, count: this.opts.partitionCount }, claimTtlMs: this.opts.claimTtlMs, now: this.opts.now?.(), + // [#17623] Reaped once for the whole tick in runTick(), when the + // outbox has a reap() to run. + skipReap, }); - if (claimed.length === 0) return; + if (claimed.length === 0) return 0; await handle.renew?.(this.opts.lockTtlMs); for (const row of claimed) { if (handle.isHeld && !handle.isHeld()) break; await this.processRow(row); } + return claimed.length; } finally { await handle.release(); } diff --git a/packages/services/service-messaging/src/http-outbox.ts b/packages/services/service-messaging/src/http-outbox.ts index 23ae79fde7..cc98ae84bf 100644 --- a/packages/services/service-messaging/src/http-outbox.ts +++ b/packages/services/service-messaging/src/http-outbox.ts @@ -313,6 +313,27 @@ export interface HttpClaimOptions { claimTtlMs: number; /** "Now" reference, ms since epoch. Defaults to Date.now(). */ now?: number; + /** + * [#17623] Skip the visibility-timeout reap this call otherwise runs before + * claiming. Default `false`: a direct `claim()` stays self-contained — the + * call that wants stale `in_flight` rows back is the one that recovers them. + * + * A caller that already ran {@link IHttpOutbox.reap} for this pass passes + * `true`. `HttpDispatcher` reaps once per tick and then claims every + * partition; the reap is environment-wide — its predicate names no + * partition — so repeating it per claim recovers nothing the first run did + * not, and on an idle outbox with the default 8 partitions it was 8 of a + * tick's 16 statements. + */ + skipReap?: boolean; +} + +/** [#17623] Options for {@link IHttpOutbox.reap}. */ +export interface HttpReapOptions { + /** Visibility timeout — `in_flight` rows claimed longer ago than this revert to `pending`. */ + claimTtlMs: number; + /** "Now" reference, ms since epoch. Defaults to Date.now(). */ + now?: number; } export interface HttpAckSuccess { @@ -492,6 +513,23 @@ export interface IHttpOutbox { */ recordUndeliverable(input: UndeliverableHttpInput): Promise; + /** + * [#17623] The visibility-timeout recovery on its own: every `in_flight` row + * whose claim is older than `claimTtlMs` reverts to `pending` with its claim + * cleared — across the WHOLE store, every partition and every organization. + * It is the step {@link claim} runs first unless told `skipReap`, exposed so + * a dispatcher can run it once per pass instead of once per partition. + * + * Safe to run at any moment and from any number of nodes, with exactly the + * claim-TTL semantics {@link claim} already has: it moves only rows already + * past their timeout, and {@link claim} takes only `pending` rows. + * + * Optional, so an outbox written before it keeps working unchanged: the + * dispatcher probes for it and, when it is absent, lets every claim reap as + * before — correct, at the per-claim cost. Both built-in stores implement it. + */ + reap?(opts: HttpReapOptions): Promise; + /** * Atomically claim up to `limit` rows whose `nextRetryAt <= now` (or null) * and matching the partition predicate. Claimed rows MUST be marked diff --git a/packages/services/service-messaging/src/index.ts b/packages/services/service-messaging/src/index.ts index d1ba1dabb9..40fac79407 100644 --- a/packages/services/service-messaging/src/index.ts +++ b/packages/services/service-messaging/src/index.ts @@ -134,6 +134,8 @@ export type { HttpDeliveryStatus, EnqueueHttpInput, HttpClaimOptions, + // [#17623] The dispatcher's once-per-tick visibility-timeout recovery. + HttpReapOptions, HttpAckResult, HttpAckSuccess, HttpAckFailure, diff --git a/packages/services/service-messaging/src/memory-http-outbox.ts b/packages/services/service-messaging/src/memory-http-outbox.ts index dabf78a5c0..69c7fb6762 100644 --- a/packages/services/service-messaging/src/memory-http-outbox.ts +++ b/packages/services/service-messaging/src/memory-http-outbox.ts @@ -12,6 +12,7 @@ import { type HttpClaimOptions, type HttpDelivery, type HttpDeliveryStatus, + type HttpReapOptions, type IHttpOutbox, type RedeliverOptions, type UndeliverableHttpInput, @@ -90,22 +91,16 @@ export class MemoryHttpOutbox implements IHttpOutbox { return id; } + async reap(opts: HttpReapOptions): Promise { + this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); + } + async claim(opts: HttpClaimOptions): Promise { const now = opts.now ?? Date.now(); const claimed: HttpDelivery[] = []; - for (const row of this.rows.values()) { - if ( - row.status === 'in_flight' && - row.claimedAt !== undefined && - now - row.claimedAt > opts.claimTtlMs - ) { - row.status = 'pending'; - row.claimedBy = undefined; - row.claimedAt = undefined; - row.updatedAt = now; - } - } + // Reap stale in_flight — unless the caller already reaped this pass (#17623). + if (!opts.skipReap) this.reapExpired(now, opts.claimTtlMs); for (const row of this.rows.values()) { if (claimed.length >= opts.limit) break; @@ -124,6 +119,22 @@ export class MemoryHttpOutbox implements IHttpOutbox { return claimed; } + /** Visibility-timeout recovery: every expired `in_flight` claim reverts to `pending`. */ + private reapExpired(now: number, claimTtlMs: number): void { + for (const row of this.rows.values()) { + if ( + row.status === 'in_flight' && + row.claimedAt !== undefined && + now - row.claimedAt > claimTtlMs + ) { + row.status = 'pending'; + row.claimedBy = undefined; + row.claimedAt = undefined; + row.updatedAt = now; + } + } + } + async ack(id: string, result: HttpAckResult): Promise { const row = this.rows.get(id); if (!row) return; diff --git a/packages/services/service-messaging/src/messaging-service-plugin.ts b/packages/services/service-messaging/src/messaging-service-plugin.ts index 9454859202..6e5b72a739 100644 --- a/packages/services/service-messaging/src/messaging-service-plugin.ts +++ b/packages/services/service-messaging/src/messaging-service-plugin.ts @@ -48,14 +48,17 @@ export interface MessagingServicePluginOptions { /** Dispatcher tick interval in ms while there is work (default 500). */ dispatchIntervalMs?: number; /** - * [#17610] Ceiling in ms for the notification dispatcher's idle backoff - * (default 30000). Consecutive ticks that claim nothing double the interval - * from `dispatchIntervalMs` up to this; a tick that claims work snaps it - * back, and an `emit()` that enqueues deliveries wakes the dispatcher at - * once. While idle it bounds how late the dispatcher notices work nobody - * woke it for: a deferred delivery coming due (retry, quiet hours, digest - * window), a row enqueued by another process, a crashed node's claim passing - * its timeout. A value at or below `dispatchIntervalMs` disables the backoff. + * [#17610, #17623] Ceiling in ms for the idle backoff of BOTH dispatchers — + * notification and outbound-HTTP — the way `dispatchIntervalMs` and + * `partitionCount` already govern both (default 30000). Consecutive ticks + * that claim nothing double the interval from `dispatchIntervalMs` up to + * this; a tick that claims work snaps it back, and the write that makes work + * wakes its dispatcher at once — `emit()` enqueueing deliveries, + * `enqueueHttp()` enqueueing a callout, `redeliverHttp()` resetting one. + * While idle it bounds how late a dispatcher notices work nobody woke it + * for: a deferred delivery coming due (retry, quiet hours, digest window), a + * row enqueued by another process, a crashed node's claim passing its + * timeout. A value at or below `dispatchIntervalMs` disables the backoff. */ dispatchMaxIdleIntervalMs?: number; /** @@ -324,18 +327,21 @@ export class MessagingServicePlugin implements Plugin { // the Flow `http` node (and, going forward, webhook fan-out) with // the same retry / dead-letter substrate as notifications. const httpOutbox = new SqlHttpOutbox(engine, { partitionCount: this.options.partitionCount }); - service.setHttpOutbox(httpOutbox); + // [#17623] The same seam as `setOutbox` above, resolved at call + // time for the same reasons. + service.setHttpOutbox(httpOutbox, { onEnqueued: () => this.httpDispatcher?.wake() }); this.httpDispatcher = new HttpDispatcher({ nodeId: `http-${process.pid}-${randomUUID().slice(0, 8)}`, outbox: httpOutbox, cluster, partitionCount: this.options.partitionCount, intervalMs: this.options.dispatchIntervalMs, + maxIdleIntervalMs: this.options.dispatchMaxIdleIntervalMs, logger: ctx.logger, }); this.httpDispatcher.start(); ctx.logger.info( - `[messaging] HTTP delivery on (sys_http_delivery outbox + dispatcher, ${this.options.partitionCount} partitions)`, + `[messaging] HTTP delivery on (sys_http_delivery outbox + dispatcher, ${this.options.partitionCount} partitions, idle backoff up to ${Math.max(this.options.dispatchIntervalMs, this.options.dispatchMaxIdleIntervalMs)}ms)`, ); }); } diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index b3a0ba77f4..36757546eb 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -184,6 +184,8 @@ export class MessagingService { /** [#17610] Fired after an `emit()` enqueues deliveries — see {@link setOutbox}. */ private onDeliveriesEnqueued?: () => void; private httpOutbox?: IHttpOutbox; + /** [#17623] Fired after a write leaves an HTTP delivery `pending` — see {@link setHttpOutbox}. */ + private onHttpDeliveryEnqueued?: () => void; /** [#8069] Producer vetoes over redelivery, keyed by `HttpDelivery.source`. */ private readonly redeliverGuards = new Map(); @@ -224,9 +226,20 @@ export class MessagingService { * the plugin at `kernel:ready` once the data engine is resolvable. Once set, * {@link enqueueHttp} persists durable rows the {@link HttpDispatcher} * drains with retry / dead-letter; the Flow `http` node enqueues through it. + * + * [#17623] `onEnqueued` fires after every write through this service that + * leaves a row `pending` for the dispatcher: an {@link enqueueHttp} that + * enqueues a delivery — not one that parks an undeliverable record, which is + * `dead` on arrival — and a {@link redeliverHttp} that resets one. The plugin + * points it at `HttpDispatcher.wake()`: the dispatcher backs its tick + * interval off while the outbox is idle, and a row this process just wrote + * should go out on the next tick, not the next backed-off one. It belongs to + * the outbox it was attached with — attaching another outbox replaces (or + * clears) it. */ - setHttpOutbox(outbox: IHttpOutbox): void { + setHttpOutbox(outbox: IHttpOutbox, options: { onEnqueued?: () => void } = {}): void { this.httpOutbox = outbox; + this.onHttpDeliveryEnqueued = options.onEnqueued; } /** @@ -256,7 +269,10 @@ export class MessagingService { if (undeliverableReason !== undefined) { return this.httpOutbox.recordUndeliverable({ ...rest, reason: undeliverableReason }); } - return this.httpOutbox.enqueue(input); + const id = await this.httpOutbox.enqueue(input); + // [#17623] Wake the dispatcher — see `setHttpOutbox`. + this.onHttpDeliveryEnqueued?.(); + return id; } /** @@ -306,10 +322,13 @@ export class MessagingService { if (!this.httpOutbox) { throw new Error('messaging: HTTP delivery outbox not configured'); } - return this.httpOutbox.redeliver(id, { + const row = await this.httpOutbox.redeliver(id, { tenantId: options.tenantId, guard: (row) => this.redeliverGuards.get(row.source)?.(row), }); + // [#17623] The row is `pending` again: wake the dispatcher — see `setHttpOutbox`. + this.onHttpDeliveryEnqueued?.(); + return row; } /** List HTTP delivery rows (admin/tests). Empty when no outbox is wired. */ diff --git a/packages/services/service-messaging/src/plugin-enqueue-wakes-dispatcher.test.ts b/packages/services/service-messaging/src/plugin-enqueue-wakes-dispatcher.test.ts index 3c4f12a81b..2cd804df0a 100644 --- a/packages/services/service-messaging/src/plugin-enqueue-wakes-dispatcher.test.ts +++ b/packages/services/service-messaging/src/plugin-enqueue-wakes-dispatcher.test.ts @@ -22,9 +22,14 @@ * straight into the outbox table — no `emit()`, so no wake — is still * `pending` after a real wait. Without it, a dispatcher ticking fast for any * unrelated reason would pass the positive leg vacuously. + * + * #17623 put `HttpDispatcher` on the same loop, woken by `enqueueHttp()`. Its + * leg below has the same shape — the plugin's own wiring, a real engine, a + * negative control on the same boot — with `globalThis.fetch` stubbed, since + * that is what the plugin's HTTP dispatcher sends through. */ -import { describe, it, expect, afterEach } from 'vitest'; +import { describe, it, expect, afterEach, vi } from 'vitest'; import { ObjectKernel } from '@objectstack/core'; import { ObjectQLPlugin } from '@objectstack/objectql'; import { SqlDriver } from '@objectstack/driver-sql'; @@ -33,6 +38,7 @@ import type { IDataEngine } from '@objectstack/spec/contracts'; import { MessagingServicePlugin } from './messaging-service-plugin.js'; import type { MessagingService } from './messaging-service.js'; import { DELIVERY_OBJECT } from './sql-outbox.js'; +import { SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js'; import { hashPartition } from './backoff.js'; /** No timer-driven tick inside the test's lifetime. */ @@ -64,6 +70,7 @@ afterEach(async () => { while (openDrivers.length) { try { await openDrivers.pop()?.disconnect?.(); } catch { /* noop */ } } + vi.unstubAllGlobals(); }); async function bootMessagingKernel() { @@ -98,6 +105,19 @@ async function statusOf(engine: IDataEngine, id: string): Promise { return found?.status; } +async function httpStatusOf(engine: IDataEngine, id: string): Promise { + const found = await engine.findOne(SYS_HTTP_DELIVERY, { where: { id }, fields: ['status'] }); + return found?.status; +} + +async function eventually(predicate: () => Promise, budgetMs: number): Promise { + const deadline = Date.now() + budgetMs; + while (!(await predicate())) { + if (Date.now() > deadline) throw new Error(`condition not met within ${budgetMs} ms`); + await sleep(10); + } +} + describe('#17610 MessagingServicePlugin — emit() wakes the backed-off dispatcher', () => { it('sends an emitted notification at once, while a row nobody announced stays pending', async () => { const { engine, messaging } = await bootMessagingKernel(); @@ -144,3 +164,55 @@ describe('#17610 MessagingServicePlugin — emit() wakes the backed-off dispatch expect(await statusOf(engine, 'dlv_unannounced')).toBe('success'); }, 20_000); }); + +describe('#17623 MessagingServicePlugin — enqueueHttp() wakes the backed-off HTTP dispatcher', () => { + it('POSTs an enqueued delivery at once, while a row nobody announced stays pending', async () => { + const posted: string[] = []; + // The plugin constructs its HttpDispatcher without a fetchImpl, so it + // sends through `globalThis.fetch`. + vi.stubGlobal('fetch', async (url: string) => { + posted.push(url); + return { ok: true, status: 200, async text() { return 'ok'; } }; + }); + const { engine, messaging } = await bootMessagingKernel(); + + // NEGATIVE CONTROL — a ready `pending` row written straight into the + // outbox table, the way a process with no dispatcher of its own would. + const now = new Date(); + await engine.insert(SYS_HTTP_DELIVERY, { + id: 'hdl_unannounced', + source: 'flow', + ref_id: 'ref_quiet', + dedup_key: 'quiet', + url: 'https://receiver.example/quiet', + method: 'POST', + payload_json: '{}', + partition_key: hashPartition('ref_quiet', PARTITIONS), + status: 'pending', + attempts: 0, + created_at: now, + updated_at: now, + }); + await sleep(CONTROL_WAIT_MS); + expect(await httpStatusOf(engine, 'hdl_unannounced')).toBe('pending'); + expect(posted).toEqual([]); + + // The ingress: enqueueHttp() writes and wakes. The woken tick drains the + // whole partition, so the unannounced row goes out with it. + const id = await messaging.enqueueHttp({ + source: 'flow', + refId: 'ref_loud', + dedupKey: 'loud', + url: 'https://receiver.example/loud', + payload: { title: 'loud' }, + }); + + await until(() => posted.length === 2, DELIVERY_BUDGET_MS); + expect([...posted].sort()).toEqual(['https://receiver.example/loud', 'https://receiver.example/quiet']); + await eventually( + async () => (await httpStatusOf(engine, id)) === 'success' + && (await httpStatusOf(engine, 'hdl_unannounced')) === 'success', + DELIVERY_BUDGET_MS, + ); + }, 20_000); +}); diff --git a/packages/services/service-messaging/src/sql-http-outbox.ts b/packages/services/service-messaging/src/sql-http-outbox.ts index d1d60d27f0..0f5f6bec13 100644 --- a/packages/services/service-messaging/src/sql-http-outbox.ts +++ b/packages/services/service-messaging/src/sql-http-outbox.ts @@ -15,6 +15,7 @@ import { type HttpClaimOptions, type HttpDelivery, type HttpDeliveryStatus, + type HttpReapOptions, type IHttpOutbox, type RedeliverOptions, type UndeliverableHttpInput, @@ -95,9 +96,10 @@ interface DeliveryRow { * **No UPDATE here writes `updated_at`** (#4765) — same rule, same reason as * {@link SqlNotificationOutbox}: the platform's `sys_stamp_audit_update` hook * owns that column, a caller-supplied value is stripped as `readonly` (#2948) - * with a WARN per call, and `claim()`'s unconditional reap UPDATE runs on every - * dispatcher tick — so writing it turned an idle dev server into a console - * firehose while changing nothing about the stored row. + * with a WARN per call, and the visibility-timeout reap is an unconditional + * UPDATE that runs on every dispatcher tick (`reap()`, or `claim()` unless told + * `skipReap`) — so writing it turned an idle dev server into a console firehose + * while changing nothing about the stored row. */ export class SqlHttpOutbox implements IHttpOutbox { private readonly objectName: string; @@ -202,20 +204,16 @@ export class SqlHttpOutbox implements IHttpOutbox { } } + async reap(opts: HttpReapOptions): Promise { + await this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); + } + async claim(opts: HttpClaimOptions): Promise { const now = opts.now ?? Date.now(); - // 1. Reap stale in_flight rows — visibility-timeout recovery. - await this.engine.update( - this.objectName, - { status: 'pending', claimed_by: null, claimed_at: null }, - // Environment-wide by design: recovers rows a crashed node abandoned, - // for every organization. Warrant in `outbox-dispatcher-scope.ts`. - dispatcherSweepOptions({ - status: 'in_flight', - claimed_at: { $lt: now - opts.claimTtlMs }, - }), - ); + // 1. Reap stale in_flight rows — visibility-timeout recovery — unless the + // caller already ran `reap()` for this pass (#17623). + if (!opts.skipReap) await this.reapExpired(now, opts.claimTtlMs); // 2. Pick candidate ids. const partitionFilter = opts.partition ? { partition_key: opts.partition.index } : {}; @@ -253,6 +251,24 @@ export class SqlHttpOutbox implements IHttpOutbox { return claimed.map((r) => this.toDelivery(r, headerColumns)); } + /** + * The visibility-timeout reap: ONE predicate UPDATE returning every expired + * `in_flight` claim to `pending`. No partition in the predicate — it spans + * the whole table by construction. + */ + private async reapExpired(now: number, claimTtlMs: number): Promise { + await this.engine.update( + this.objectName, + { status: 'pending', claimed_by: null, claimed_at: null }, + // Environment-wide by design: recovers rows a crashed node abandoned, + // for every organization. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherSweepOptions({ + status: 'in_flight', + claimed_at: { $lt: now - claimTtlMs }, + }), + ); + } + /** * [#8118] Recover `headers_json` for a batch of just-claimed rows. *