fix(service-messaging): HttpDispatcher reaps once per tick, backs off while idle, wakes on enqueueHttp

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

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

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);
+    });
 });
 
 // ───────────────────────────────────────────────────────────────────────────

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<number>;
+    /** 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<number>;
+    private readonly onTickError: (err: unknown) => void;
+    private timer: ReturnType<typeof setTimeout> | undefined;
+    private running = false;
+    private inflightTick: Promise<void> | 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<void> {
+        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;
+    }
+}