fix(core): an idle DispatchLoop wakes at the next due time, and a sole writer skips the cold-start ramp

.changeset/29919-dispatch-loop-next-due-wake.md

@@ -0,0 +1,17 @@
+---
+'@objectstack/core': patch
+'@objectstack/service-messaging': patch
+'@objectstack/service-queue': patch
+---
+
+fix(core,service-messaging,service-queue): an idle dispatch loop wakes at the next due time, and a sole writer's loop no longer re-runs the idle ramp after every boot
+
+Clause-②: yes (widening)
+Metadata-protocol: none
+
+- **`@objectstack/core`: `DispatchLoop` wakes at the next due time.** A new optional `DispatchLoopOptions.nextDueAt()` answers the earliest instant work becomes claimable (a deferred row, a claim's expiry). The loop asks it once per idle streak, after the first empty tick, and never sleeps past the answer, so a deferred row goes out within one tick of coming due whatever `maxIdleIntervalMs` is. The answer is remembered rather than re-read on every ceiling tick, so an idle loop's steady-state cost is unchanged; the new `DispatchLoop.wakeAt(dueAtMs)` lets a producer in the same process announce a deferred row at no read. An answer at or before the tick that found nothing is not chased, and a rejected tick or read still counts as idle, so neither can make the loop spin.
+- **`@objectstack/core`: `DispatchLoopOptions.soleWriter`** (default `false`). On a host where no other process writes the store, an empty tick sleeps straight to the next due time or the ceiling instead of re-running the ramp from `intervalMs`, which only ever bought latency on rows another process wrote. A fresh sole-writer loop over an empty store now ticks once in its first minute; before, every boot re-ran the ramp (six ticks in the first minute at the dispatchers' 500 ms base, five at the queue's 1 s).
+- **`@objectstack/service-messaging`:** `INotificationOutbox.nextDueAt` / `IHttpOutbox.nextDueAt` (optional; both built-in stores implement it: two reads, the earliest future `next_attempt_at` / `next_retry_at` and the oldest `in_flight` claim plus the claim TTL), wired into `NotificationDispatcher` and `HttpDispatcher`; `soleWriter` on both dispatchers' options, and `dispatchSoleWriter` on `MessagingServicePlugin`. Retries, quiet hours, digest windows and a crashed node's claims are taken at their due time instead of up to `dispatchMaxIdleIntervalMs` late.
+- **`@objectstack/service-queue`:** `DbQueueAdapter` reads the earliest future `scheduled_for` of its subscribed queues (one read per idle streak), a deferred `publish()` announces its due time to the loop through `wakeAt` instead of waiting out the backoff, and `DbQueueAdapterOptions.soleWriter` (reachable as `QueueServicePlugin` `db.soleWriter`) skips the ramp.
+
+Nothing changes for a host that sets neither `soleWriter` flag except the due-time wake: the ramp keeps its shape, cut short only when a due row is earlier, and an idle stretch costs one extra read at its start.

content/docs/automation/webhooks.mdx

@@ -452,16 +452,39 @@ 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.
+`in_flight` rows claimed longer than `claimTtlMs` ago return to `pending` — then
+asks once, across every partition, whether any `pending` row is due. Only when
+one is does it claim partition by partition. A tick over an empty outbox
+therefore costs **two** statements, both reads, whatever `partitionCount` is:
+the reap's "anything expired?" read (its UPDATE runs only when that finds a
+row) and the "anything due?" read.
+
+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.
+
+The loop also **wakes at the next due time**. After the first empty tick of an
+idle stretch it reads, once, the earliest instant a row becomes claimable
+without anyone writing: the soonest future `next_retry_at`, or the moment the
+oldest `in_flight` claim passes `claimTtlMs`. It never sleeps past that
+instant, so a retry is attempted when it comes due and a crashed node's claim is
+recovered when its TTL passes, whatever the ceiling. That costs two more reads
+at the start of each idle stretch, not one per tick. The backoff is left with
+one job: a row **another process** wrote, which is noticed within one
+backed-off interval, never more than `maxIdleIntervalMs` late.
+
+On a host where no other process writes these tables, that job is empty. Set
+`dispatchSoleWriter: true` on `MessagingServicePlugin` there, for example on a
+hosted tenant kernel with one kernel per tenant database. An empty tick then
+sleeps straight to the next due time or the ceiling, instead of re-running the
+ramp from `intervalMs` after every boot and every burst, and a fresh kernel's
+dispatchers tick once in their first minute. The job queue takes the same flag
+as `DbQueueAdapter`'s `soleWriter`, set through `QueueServicePlugin`'s
+`db.soleWriter`. Leave both off when a second node or a worker process writes
+the same database. Default: off.
 
 Within a held partition, the lock-holder claims a batch with an **atomic
 conditional UPDATE** rather than `SELECT … FOR UPDATE SKIP LOCKED`:
@@ -697,7 +720,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`; 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). |
+| 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 wakes when the oldest claim it read at the start of its idle stretch passes its TTL, and ticks at least every `maxIdleIntervalMs` (default 30s) in any case, so recovery takes about `claimTtlMs` for a claim it saw and at most `claimTtlMs + maxIdleIntervalMs` (~35s at defaults) for one made after that read. |
 | A send outlasts the claim TTL and another dispatcher re-claims the row | The re-claiming dispatcher posts the delivery again (at-least-once — the receiver can see it twice; see the last row). The first dispatcher's late outcome is **not** written over the live attempt: its ack carries the claim credential (`claimed_by`, `claimed_at`) its claim stamped, the conditional update matches nothing, and it logs `http-dispatcher: ack refused, claim no longer held`. The row records the outcome of the attempt that still holds it. |
 | 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. |

packages/core/src/dispatch-loop.test.ts

@@ -0,0 +1,294 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * #29919 — `DispatchLoop` wakes at the next due time, and a sole writer's loop
+ * does not re-run the idle ramp after a cold start.
+ *
+ * Before #29919 every `start()` re-ran the backoff's ramp from `intervalMs`
+ * against stores that were empty: with the dispatchers' 500 ms base, six ticks
+ * in a fresh loop's first minute (0, 1, 3, 7, 15, 31 s), every one idle, while the only thing the ramp ever bought — early
+ * notice of a row another process wrote — cannot happen on a sole writer. And
+ * a deferred row was noticed only at the next backed-off tick, up to the
+ * ceiling late.
+ *
+ * Every leg runs on vitest's fake timers, so "when did a tick start" is an
+ * exact reading rather than a sleep-and-hope.
+ */
+
+import { describe, it, expect, afterEach, vi } from 'vitest';
+import { DispatchLoop, type DispatchLoopOptions } from './dispatch-loop.js';
+
+const BASE = 500;
+const MINUTE = 60_000;
+const FIVE_MINUTES = 5 * MINUTE;
+
+afterEach(() => { vi.useRealTimers(); });
+
+/** Fake timers with the clock at 0, so a tick's start reads as ms since `start()`. */
+function useClockAtZero(): void {
+    vi.useFakeTimers();
+    vi.setSystemTime(0);
+}
+
+/**
+ * A store of deferred rows: `runTick` claims every row due by the fake clock,
+ * `nextDueAt` answers the earliest row still in the future. Both are recorded.
+ */
+function harness(options: Partial<DispatchLoopOptions> & { dueRows?: number[] } = {}) {
+    const dueRows = [...(options.dueRows ?? [])];
+    const tickStarts: number[] = [];
+    const claimedAt: number[] = [];
+    let dueReads = 0;
+    const errors: unknown[] = [];
+    const loop = new DispatchLoop({
+        intervalMs: BASE,
+        runTick: async () => {
+            const now = Date.now();
+            tickStarts.push(now);
+            let claimed = 0;
+            for (let i = dueRows.length - 1; i >= 0; i--) {
+                if (dueRows[i]! <= now) {
+                    dueRows.splice(i, 1);
+                    claimedAt.push(now);
+                    claimed++;
+                }
+            }
+            return claimed;
+        },
+        nextDueAt: async () => {
+            dueReads++;
+            const future = dueRows.filter((t) => t > Date.now());
+            return future.length > 0 ? Math.min(...future) : null;
+        },
+        onTickError: (err) => { errors.push(err); },
+        ...options,
+    });
+    return {
+        loop,
+        tickStarts,
+        claimedAt,
+        errors,
+        dueRows,
+        get dueReads() { return dueReads; },
+    };
+}
+
+describe('#29919 DispatchLoop — a sole writer does not ramp from a cold start', () => {
+    it('a fresh loop on an empty store ticks ONCE in its first minute', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+
+        expect(h.tickStarts).toEqual([0]);
+        // The one tick's one due read — nothing deferred, so the ceiling.
+        expect(h.dueReads).toBe(1);
+    });
+
+    it('CONTROL: the same loop without soleWriter still runs the ramp it always had', async () => {
+        useClockAtZero();
+        const h = harness({ maxIdleIntervalMs: FIVE_MINUTES });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+
+        // 1, 2, 4, 8, 16 s apart — six ticks. The harness can see a ramp, so the
+        // single tick above is a reading about soleWriter, not a dead clock.
+        expect(h.tickStarts).toEqual([0, 1_000, 3_000, 7_000, 15_000, 31_000]);
+    });
+
+    it('without a nextDueAt reader, soleWriter alone sleeps straight to the ceiling', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: 30_000, nextDueAt: undefined });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(2 * MINUTE);
+        await h.loop.stop();
+
+        expect(h.tickStarts).toEqual([0, 30_000, 60_000, 90_000, 120_000]);
+    });
+
+    it('steady state costs no extra read: an idle hour at the ceiling reads the due time once', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: 30_000 });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(60 * MINUTE);
+        await h.loop.stop();
+
+        // 3600 s / 30 s, plus the tick start() runs.
+        expect(h.tickStarts).toHaveLength(121);
+        expect(h.dueReads).toBe(1);
+    });
+});
+
+describe('#29919 DispatchLoop — wakes at the next due time', () => {
+    it('a row due at T+20 s is claimed at T+20 s, under a five-minute ceiling', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES, dueRows: [20_000] });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+
+        expect(h.claimedAt).toEqual([20_000]);
+        // The cold-start tick, the due tick, and the tick after a claim
+        // (`intervalMs` later) — then nothing is deferred and the ceiling wins.
+        expect(h.tickStarts).toEqual([0, 20_000, 20_500]);
+    });
+
+    it('the ramp keeps its shape, cut short only by the due time (no soleWriter)', async () => {
+        useClockAtZero();
+        const h = harness({ maxIdleIntervalMs: FIVE_MINUTES, dueRows: [2_000] });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(4_000);
+        await h.loop.stop();
+
+        expect(h.claimedAt).toEqual([2_000]);
+        expect(h.tickStarts).toEqual([0, 1_000, 2_000, 2_500, 3_500]);
+    });
+
+    it('consecutive deferred rows are each claimed on time — the read is re-asked after each due time', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES, dueRows: [10_000, 40_000] });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+
+        expect(h.claimedAt).toEqual([10_000, 40_000]);
+    });
+
+    it('wakeAt(): a deferred row written later in the streak is claimed on time with no read', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(10_000);
+        const readsBefore = h.dueReads;
+        const ticksBefore = h.tickStarts.length;
+
+        h.dueRows.push(35_000);
+        h.loop.wakeAt(35_000);
+        // No tick now: the hint only moves the timer.
+        await vi.advanceTimersByTimeAsync(1);
+        expect(h.tickStarts).toHaveLength(ticksBefore);
+
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+        expect(h.claimedAt).toEqual([35_000]);
+        // The hint cost no read; the read after the claim is the next streak's.
+        expect(h.dueReads).toBe(readsBefore + 1);
+    });
+
+    it('wakeAt() keeps a timer that is already due sooner', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES, dueRows: [20_000] });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(1);
+        h.dueRows.push(50_000);
+        h.loop.wakeAt(50_000);
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+
+        expect(h.claimedAt).toEqual([20_000, 50_000]);
+    });
+});
+
+describe('#29919 DispatchLoop — wake() still delivers at once', () => {
+    it('a wake on a sole writer sleeping towards its ceiling ticks immediately', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+
+        h.dueRows.push(MINUTE);
+        h.loop.wake();
+        await vi.advanceTimersByTimeAsync(0);
+        await h.loop.stop();
+
+        expect(h.claimedAt).toEqual([MINUTE]);
+    });
+});
+
+describe('#29919 DispatchLoop — nothing it reads can make it spin', () => {
+    it('a rejected tick counts as idle: a failing store is ticked once per ceiling, not hammered', async () => {
+        useClockAtZero();
+        const h = harness({
+            soleWriter: true,
+            maxIdleIntervalMs: 30_000,
+            runTick: async () => { h.tickStarts.push(Date.now()); throw new Error('store down'); },
+        });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+
+        expect(h.tickStarts).toEqual([0, 30_000, 60_000]);
+        expect(h.errors).toHaveLength(3);
+        // A rejected tick asks for no due time.
+        expect(h.dueReads).toBe(0);
+    });
+
+    it('a rejected tick after a due time was learned still sleeps until that due time, not 0 ms', async () => {
+        useClockAtZero();
+        const tickStarts: number[] = [];
+        const loop = new DispatchLoop({
+            intervalMs: BASE,
+            maxIdleIntervalMs: FIVE_MINUTES,
+            soleWriter: true,
+            // The cold-start tick finds nothing; every later tick rejects.
+            runTick: async () => {
+                tickStarts.push(Date.now());
+                if (tickStarts.length > 1) throw new Error('store down');
+                return 0;
+            },
+            nextDueAt: async () => 20_000,
+            onTickError: () => {},
+        });
+        loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await loop.stop();
+
+        // The due tick at 20 s rejects; the next is a ceiling away.
+        expect(tickStarts).toEqual([0, 20_000]);
+    });
+
+    it('a due time at or before the tick that found nothing is not chased (visible, but not this loop\'s to claim)', async () => {
+        useClockAtZero();
+        const h = harness({
+            soleWriter: true,
+            maxIdleIntervalMs: 30_000,
+            // Another node holds the row's partition: it is due, and never claimable here.
+            runTick: async () => { h.tickStarts.push(Date.now()); return 0; },
+            nextDueAt: async () => 0,
+        });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+
+        expect(h.tickStarts).toEqual([0, 30_000, 60_000]);
+    });
+
+    it('a rejected due read is reported, and the loop sleeps by its backoff alone', async () => {
+        useClockAtZero();
+        const h = harness({
+            soleWriter: true,
+            maxIdleIntervalMs: 30_000,
+            nextDueAt: async () => { throw new Error('read failed'); },
+        });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await h.loop.stop();
+
+        expect(h.tickStarts).toEqual([0, 30_000, 60_000]);
+        expect(h.errors).toHaveLength(3);
+    });
+
+    it('stop() is final: a wakeAt() after it is a no-op', async () => {
+        useClockAtZero();
+        const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        h.loop.start();
+        await vi.advanceTimersByTimeAsync(1);
+        await h.loop.stop();
+        h.loop.wakeAt(10_000);
+        await vi.advanceTimersByTimeAsync(MINUTE);
+
+        expect(h.tickStarts).toEqual([0]);
+    });
+});