From edc37a0038f02a9b36b2d81b787f1f1fffa5af31 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 01:07:54 +0000
Subject: [PATCH 1/5] wip(core): DispatchLoop wakes at the next due time and
 skips the ramp for a sole writer (#29919)

---
 packages/core/src/dispatch-loop.ts            | 155 +++++++++++++++++-
 .../service-messaging/src/dispatcher.ts       |  35 +++-
 .../service-messaging/src/http-dispatcher.ts  |  38 +++--
 .../service-messaging/src/http-outbox.ts      |  23 +++
 .../services/service-messaging/src/index.ts   |   2 +
 .../src/memory-http-outbox.ts                 |  15 ++
 .../service-messaging/src/memory-outbox.ts    |  15 ++
 .../src/messaging-service-plugin.ts           |  23 ++-
 .../service-messaging/src/next-due.ts         |  26 +++
 .../services/service-messaging/src/outbox.ts  |  21 +++
 .../service-messaging/src/sql-http-outbox.ts  |  26 +++
 .../service-messaging/src/sql-outbox.ts       |  26 +++
 .../service-queue/src/db-queue-adapter.ts     |  69 ++++++--
 13 files changed, 437 insertions(+), 37 deletions(-)
 create mode 100644 packages/services/service-messaging/src/next-due.ts

diff --git a/packages/core/src/dispatch-loop.ts b/packages/core/src/dispatch-loop.ts
index 5d063e15e07..3fcbc6a5eba 100644
--- a/packages/core/src/dispatch-loop.ts
+++ b/packages/core/src/dispatch-loop.ts
@@ -37,6 +37,27 @@
  * 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.
+ *
+ * ## Waking at the next due time (#29919)
+ *
+ * Two of those three are not unknowable: a deferred row and a claim carry the
+ * instant they become claimable. A loop given {@link DispatchLoopOptions.nextDueAt}
+ * asks for the earliest such instant once, after the first empty tick of an
+ * idle streak (after `start()`, after a tick that claimed work, after a wake,
+ * and after the due time it was sleeping towards arrives), and never sleeps
+ * past it: a deferred row goes out within one tick of coming due, whatever the
+ * ceiling. The answer is remembered, not re-read on every ceiling tick, so an
+ * idle loop's steady-state cost is unchanged; a deferred row written in this
+ * process later in the streak reaches the loop through
+ * {@link DispatchLoop.wakeAt} instead, at no read at all.
+ *
+ * Only the third case — a row another process wrote — is left to the backoff.
+ * A host on which no other process writes the store (a hosted tenant kernel:
+ * one kernel per tenant database) says so with
+ * {@link DispatchLoopOptions.soleWriter}, and an empty tick then sleeps
+ * straight to the next due time or the ceiling instead of re-running the
+ * `intervalMs` ramp. Before #29919 every boot re-ran that ramp against empty
+ * stores: five ticks in the first minute per loop, every one of them idle.
  */
 
 /** Default ceiling of the idle backoff, in ms — see {@link DispatchLoopOptions.maxIdleIntervalMs}. */
@@ -50,6 +71,30 @@ export interface DispatchLoopOptions {
      * A value at or below `intervalMs` disables the backoff.
      */
     maxIdleIntervalMs?: number;
+    /**
+     * [#29919] The earliest instant (epoch ms, on `Date.now()`'s clock) at
+     * which work this loop's tick would claim becomes claimable — a deferred
+     * row's due time, a claim's expiry — or `null` when nothing is deferred.
+     * Read once per idle streak, after its first empty tick; the loop then never
+     * sleeps past the answer. An answer at or before the start of the tick that
+     * just found nothing is not chased (that work was visible and not claimable
+     * by this loop — another node's partition, say), so it cannot make the loop
+     * spin. A rejection is reported through `onTickError` and the loop sleeps as
+     * if it had no answer.
+     *
+     * Absent, the loop sleeps by the backoff alone, as it always has.
+     */
+    nextDueAt?: () => Promise<number | null | undefined>;
+    /**
+     * [#29919] No other process writes the store this loop drains: every row it
+     * can claim is written here, by a producer that calls
+     * {@link DispatchLoop.wake} (or {@link DispatchLoop.wakeAt} for a deferred
+     * row). The idle backoff's ramp only ever bought latency on rows another
+     * process wrote, so with this set an empty tick sleeps straight to the
+     * ceiling, or to the next due time when {@link nextDueAt} is given. Default
+     * `false`: the ramp from `intervalMs` is kept.
+     */
+    soleWriter?: boolean;
     /** 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. */
@@ -61,7 +106,17 @@ export class DispatchLoop {
     private readonly maxIdleIntervalMs: number;
     private readonly runTick: () => Promise<number>;
     private readonly onTickError: (err: unknown) => void;
+    private readonly nextDueAt: (() => Promise<number | null | undefined>) | undefined;
+    private readonly soleWriter: boolean;
     private timer: ReturnType<typeof setTimeout> | undefined;
+    /** When the pending timer fires (epoch ms); meaningful only while `timer` is set. */
+    private timerFiresAt = 0;
+    /** [#29919] Earliest known instant work becomes claimable, from the last read and every hint since. */
+    private knownDueAt: number | null = null;
+    /** [#29919] The next empty tick reads {@link DispatchLoopOptions.nextDueAt}. */
+    private dueReadOwed = false;
+    /** [#29919] Hints ({@link DispatchLoop.wakeAt}) since the running tick began — a read in flight cannot see them. */
+    private hintedDueAt: number | null = null;
     private running = false;
     private inflightTick: Promise<void> | undefined;
     /** Consecutive loop ticks that claimed nothing — the idle backoff's exponent. */
@@ -78,6 +133,8 @@ export class DispatchLoop {
         );
         this.runTick = options.runTick;
         this.onTickError = options.onTickError;
+        this.nextDueAt = options.nextDueAt;
+        this.soleWriter = options.soleWriter ?? false;
     }
 
     /** Begin the loop; the first tick runs immediately. Idempotent. */
@@ -85,6 +142,8 @@ export class DispatchLoop {
         if (this.running) return;
         this.running = true;
         this.idleTicks = 0;
+        this.knownDueAt = null;
+        this.dueReadOwed = true;
         this.loopTick();
     }
 
@@ -106,9 +165,33 @@ export class DispatchLoop {
     wake(): void {
         if (!this.running) return;
         this.idleTicks = 0;
+        this.dueReadOwed = true;
         this.loopTick();
     }
 
+    /**
+     * [#29919] Work becomes claimable at `dueAtMs` (epoch ms, on `Date.now()`'s
+     * clock) — a deferred row this process just wrote: make sure a tick runs
+     * then. Costs no read and no tick now, and leaves the idle backoff where it
+     * is; a timer already due sooner is kept. An instant already past ticks at
+     * once. No-op while stopped, or for a non-finite instant.
+     */
+    wakeAt(dueAtMs: number): void {
+        if (!this.running || !Number.isFinite(dueAtMs)) return;
+        this.knownDueAt = earliest(this.knownDueAt, dueAtMs);
+        this.hintedDueAt = earliest(this.hintedDueAt, dueAtMs);
+        // The running tick schedules its successor from `knownDueAt` when it settles.
+        if (this.inflightTick) return;
+        if (this.timer !== undefined && this.timerFiresAt <= dueAtMs) return;
+        const delayMs = dueAtMs - Date.now();
+        if (delayMs <= 0) {
+            this.dueReadOwed = true;
+            this.loopTick();
+            return;
+        }
+        this.schedule(delayMs, true);
+    }
+
     /**
      * 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.
@@ -121,13 +204,22 @@ export class DispatchLoop {
             return;
         }
         const startedAt = Date.now();
+        this.hintedDueAt = null;
         this.inflightTick = this.runTick()
-            .then((claimed) => {
-                this.idleTicks = claimed > 0 ? 0 : this.idleTicks + 1;
+            .then(async (claimed) => {
+                if (claimed > 0) {
+                    this.idleTicks = 0;
+                    // The work just done may have deferred more (a retry, a backoff).
+                    this.dueReadOwed = true;
+                    return;
+                }
+                this.idleTicks += 1;
+                if (this.nextDueAt && this.dueReadOwed) await this.readNextDueAt();
             })
             .catch((err) => {
                 // A failing store is not work: back off rather than hammer it.
                 this.idleTicks += 1;
+                this.dueReadOwed = true;
                 this.onTickError(err);
             })
             .finally(() => {
@@ -139,25 +231,67 @@ export class DispatchLoop {
                     this.loopTick();
                     return;
                 }
-                this.schedule(Math.max(0, this.nextIntervalMs() - (Date.now() - startedAt)));
+                const { delayMs, forDue } = this.nextDelay(startedAt);
+                this.schedule(Math.max(0, delayMs - (Date.now() - startedAt)), forDue);
             });
     }
 
+    /** [#29919] One read of the next due time, folded with the hints that arrived while it ran. */
+    private async readNextDueAt(): Promise<void> {
+        this.dueReadOwed = false;
+        try {
+            const due = await this.nextDueAt!();
+            this.knownDueAt = earliest(
+                typeof due === 'number' && Number.isFinite(due) ? due : null,
+                this.hintedDueAt,
+            );
+        } catch (err) {
+            // No answer: sleep by the backoff alone, and ask again next idle tick.
+            this.dueReadOwed = true;
+            this.onTickError(err);
+        }
+    }
+
     /**
-     * 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`.
+     * Delay before the next loop tick, measured from the START of the last one
+     * (`startedAt`), and whether it is a due time's: the backoff's delay
+     * ({@link nextIntervalMs}), cut short by the earliest known due time.
+     */
+    private nextDelay(startedAt: number): { delayMs: number; forDue: boolean } {
+        const backoffMs = this.nextIntervalMs();
+        if (this.idleTicks === 0 || this.knownDueAt === null) return { delayMs: backoffMs, forDue: false };
+        if (this.knownDueAt <= startedAt) {
+            // Claimable before this tick began, and the tick claimed nothing: the
+            // work is not this loop's to take right now (another node holds its
+            // partition, say). Chasing it would spin; the backoff covers it.
+            this.knownDueAt = null;
+            return { delayMs: backoffMs, forDue: false };
+        }
+        const untilDueMs = this.knownDueAt - startedAt;
+        return untilDueMs < backoffMs ? { delayMs: untilDueMs, forDue: true } : { delayMs: backoffMs, forDue: false };
+    }
+
+    /**
+     * The backoff's 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` — or the
+     * cap at once after an empty tick when the loop is the store's sole writer.
      */
     private nextIntervalMs(): number {
         if (this.idleTicks === 0) return this.intervalMs;
+        if (this.soleWriter) return this.maxIdleIntervalMs;
         // 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 {
+    private schedule(delayMs: number, forDue: boolean): void {
         this.clearTimer();
+        this.timerFiresAt = Date.now() + delayMs;
         this.timer = setTimeout(() => {
             this.timer = undefined;
+            // The due time arrived: the tick takes that work, and the next idle
+            // tick asks for the one after it.
+            if (forDue) this.dueReadOwed = true;
             this.loopTick();
         }, delayMs);
         // Don't keep the event loop alive solely for the dispatcher.
@@ -170,3 +304,10 @@ export class DispatchLoop {
         this.timer = undefined;
     }
 }
+
+/** The earlier of two optional instants. */
+function earliest(a: number | null, b: number | null): number | null {
+    if (a === null) return b;
+    if (b === null) return a;
+    return Math.min(a, b);
+}
diff --git a/packages/services/service-messaging/src/dispatcher.ts b/packages/services/service-messaging/src/dispatcher.ts
index 08af9d3f509..31a7719b6c3 100644
--- a/packages/services/service-messaging/src/dispatcher.ts
+++ b/packages/services/service-messaging/src/dispatcher.ts
@@ -71,13 +71,24 @@ export interface NotificationDispatcherOptions {
      * {@link NotificationDispatcher.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 deferred row coming due (retry schedule, quiet hours, a digest window),
-     * a row enqueued by a process this dispatcher does not serve, and a crashed
-     * node's `in_flight` rows — reaped at most `claimTtlMs` + this after their
-     * claim, where the fixed interval gave `claimTtlMs` + `intervalMs`.
+     * While idle this bounds how late the loop notices a row enqueued by a
+     * process this dispatcher does not serve. A deferred row coming due (retry
+     * schedule, quiet hours, a digest window) and a crashed node's `in_flight`
+     * rows passing `claimTtlMs` are not left to it: with an outbox that answers
+     * {@link INotificationOutbox.nextDueAt} the loop sleeps no later than the
+     * earliest of them (#29919), so each is taken within one tick of coming
+     * due. An outbox without it leaves both to this bound too.
      */
     maxIdleIntervalMs?: number;
+    /**
+     * [#29919] No other process writes this dispatcher's outbox — every row is
+     * enqueued in this process, which wakes the dispatcher (a hosted tenant
+     * kernel: one kernel per tenant database). An idle tick then sleeps straight
+     * to the next due time or the ceiling instead of re-running the backoff's
+     * ramp from `intervalMs`, which only ever bought latency on rows another
+     * process wrote. Default `false`.
+     */
+    soleWriter?: boolean;
     lockTtlMs?: number;
     claimTtlMs?: number;
     rng?: () => number;
@@ -127,7 +138,7 @@ export interface NotificationDispatcherOptions {
  */
 export class NotificationDispatcher {
     private readonly opts: Required<
-        Omit<NotificationDispatcherOptions, 'rng' | 'logger' | 'onAttempt' | 'cluster' | 'now' | 'maxIdleIntervalMs' | 'textRenderer'>
+        Omit<NotificationDispatcherOptions, 'rng' | 'logger' | 'onAttempt' | 'cluster' | 'now' | 'maxIdleIntervalMs' | 'soleWriter' | 'textRenderer'>
     > &
         Pick<NotificationDispatcherOptions, 'rng' | 'logger' | 'onAttempt' | 'now' | 'textRenderer'> & { cluster: DispatchCluster };
     /** [#17610, #17623] The timer loop — idle backoff, wake, stop — shared with `HttpDispatcher`. */
@@ -153,9 +164,21 @@ export class NotificationDispatcher {
             onAttempt: options.onAttempt,
             textRenderer: options.textRenderer,
         };
+        const { outbox } = options;
         this.loop = new DispatchLoop({
             intervalMs,
             maxIdleIntervalMs: options.maxIdleIntervalMs,
+            soleWriter: options.soleWriter,
+            // [#29919] Read once per idle streak; an outbox without the read
+            // leaves the loop on its backoff alone. An injected clock's answer
+            // is re-based onto the loop's (`Date.now()`).
+            nextDueAt: outbox.nextDueAt
+                ? async () => {
+                    const now = this.opts.now?.();
+                    const due = await outbox.nextDueAt!({ claimTtlMs: this.opts.claimTtlMs, now });
+                    return due === null || now === undefined ? due : Date.now() + (due - now);
+                }
+                : undefined,
             runTick: () => this.runTick(),
             onTickError: (err) => {
                 this.opts.logger?.warn?.('notification-dispatcher: tick failed', {
diff --git a/packages/services/service-messaging/src/http-dispatcher.ts b/packages/services/service-messaging/src/http-dispatcher.ts
index 05580d9afda..7292e10b269 100644
--- a/packages/services/service-messaging/src/http-dispatcher.ts
+++ b/packages/services/service-messaging/src/http-dispatcher.ts
@@ -78,17 +78,23 @@ export interface HttpDispatcherOptions {
      * 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`.
+     * While idle this bounds how late the loop notices a row enqueued by a
+     * process this dispatcher does not serve. A retry coming due and a crashed
+     * node's `in_flight` rows passing `claimTtlMs` are not left to it: with an
+     * outbox that answers {@link IHttpOutbox.nextDueAt} the loop sleeps no
+     * later than the earliest of them (#29919), so each is taken within one
+     * tick of coming due. An outbox without it leaves both to this bound too.
      */
     maxIdleIntervalMs?: number;
+    /**
+     * [#29919] No other process writes this dispatcher's outbox — every row is
+     * enqueued in this process, which wakes the dispatcher (a hosted tenant
+     * kernel: one kernel per tenant database). An idle tick then sleeps straight
+     * to the next due time or the ceiling instead of re-running the backoff's
+     * ramp from `intervalMs`, which only ever bought latency on rows another
+     * process wrote. Default `false`.
+     */
+    soleWriter?: boolean;
     /** Per-partition lock TTL. Default = 5 × intervalMs. */
     lockTtlMs?: number;
     /** Visibility timeout for claimed rows. Default = 2 × lockTtlMs. */
@@ -107,7 +113,7 @@ export interface HttpDispatcherOptions {
 
 export class HttpDispatcher {
     private readonly opts: Required<
-        Omit<HttpDispatcherOptions, 'fetchImpl' | 'rng' | 'logger' | 'onAttempt' | 'cluster' | 'now' | 'maxIdleIntervalMs'>
+        Omit<HttpDispatcherOptions, 'fetchImpl' | 'rng' | 'logger' | 'onAttempt' | 'cluster' | 'now' | 'maxIdleIntervalMs' | 'soleWriter'>
     > &
         Pick<HttpDispatcherOptions, 'fetchImpl' | 'rng' | 'logger' | 'onAttempt' | 'now'> & {
             cluster: DispatchCluster;
@@ -133,9 +139,21 @@ export class HttpDispatcher {
             logger: options.logger,
             onAttempt: options.onAttempt,
         };
+        const { outbox } = options;
         this.loop = new DispatchLoop({
             intervalMs,
             maxIdleIntervalMs: options.maxIdleIntervalMs,
+            soleWriter: options.soleWriter,
+            // [#29919] Read once per idle streak; an outbox without the read
+            // leaves the loop on its backoff alone. An injected clock's answer
+            // is re-based onto the loop's (`Date.now()`).
+            nextDueAt: outbox.nextDueAt
+                ? async () => {
+                    const now = this.opts.now?.();
+                    const due = await outbox.nextDueAt!({ claimTtlMs: this.opts.claimTtlMs, now });
+                    return due === null || now === undefined ? due : Date.now() + (due - now);
+                }
+                : undefined,
             runTick: () => this.runTick(),
             onTickError: (err) => {
                 this.opts.logger?.warn?.('http-dispatcher: tick failed', {
diff --git a/packages/services/service-messaging/src/http-outbox.ts b/packages/services/service-messaging/src/http-outbox.ts
index fa47699b44c..1d7944ff152 100644
--- a/packages/services/service-messaging/src/http-outbox.ts
+++ b/packages/services/service-messaging/src/http-outbox.ts
@@ -348,6 +348,14 @@ export interface HttpDueWorkOptions {
     now?: number;
 }
 
+/** [#29919] Options for {@link IHttpOutbox.nextDueAt}. */
+export interface HttpNextDueOptions {
+    /** Visibility timeout — an `in_flight` row becomes reapable once claimed longer ago than this. */
+    claimTtlMs: number;
+    /** "Now" reference, ms since epoch. Defaults to Date.now(). */
+    now?: number;
+}
+
 export interface HttpAckSuccess {
     success: true;
     httpStatus: number;
@@ -656,6 +664,21 @@ export interface IHttpOutbox {
      */
     hasDueWork?(opts: HttpDueWorkOptions): Promise<boolean>;
 
+    /**
+     * [#29919] The earliest instant (ms since epoch) at which a row becomes
+     * claimable without anyone writing to the outbox — a `pending` row's
+     * `nextRetryAt` still after `now`, or the instant an `in_flight` row's claim
+     * passes `claimTtlMs` (already past if no reap has run since) — or `null`
+     * when there is none. The dispatcher's loop sleeps no later than
+     * this after an idle tick.
+     *
+     * A read and nothing more, stale the moment it returns like
+     * {@link hasDueWork}: a row written afterwards reaches the loop by the
+     * writer's `wake()`. Optional: without it the dispatcher sleeps by its idle
+     * backoff alone. Both built-in stores implement it.
+     */
+    nextDueAt?(opts: HttpNextDueOptions): Promise<number | null>;
+
     /**
      * 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 33f2f886001..206c94c2164 100644
--- a/packages/services/service-messaging/src/index.ts
+++ b/packages/services/service-messaging/src/index.ts
@@ -140,6 +140,7 @@ export type {
     ReapOptions,
     // [#28383] The dispatcher's one-read "is anything due?" probe.
     DueWorkOptions,
+    NextDueOptions,
     AckResult,
 } from './outbox.js';
 // [commit 1a47a5368] `ack()`'s status precondition refuses with this, so a caller that
@@ -172,6 +173,7 @@ export type {
     HttpReapOptions,
     // [#28383] The dispatcher's one-read "is anything due?" probe.
     HttpDueWorkOptions,
+    HttpNextDueOptions,
     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 11d5cc8e451..98894823288 100644
--- a/packages/services/service-messaging/src/memory-http-outbox.ts
+++ b/packages/services/service-messaging/src/memory-http-outbox.ts
@@ -19,6 +19,7 @@ import {
     type HttpDeliveryStatus,
     type HttpReapOptions,
     type HttpDueWorkOptions,
+    type HttpNextDueOptions,
     type IHttpOutbox,
     type RedeliverOptions,
     type UndeliverableHttpInput,
@@ -110,6 +111,20 @@ export class MemoryHttpOutbox implements IHttpOutbox {
         return false;
     }
 
+    /** [#29919] Earliest future `nextRetryAt` of a `pending` row, or claim expiry of an `in_flight` one. */
+    async nextDueAt(opts: HttpNextDueOptions): Promise<number | null> {
+        const now = opts.now ?? Date.now();
+        let due: number | null = null;
+        for (const row of this.rows.values()) {
+            let at: number | undefined;
+            if (row.status === 'pending' && row.nextRetryAt !== undefined && row.nextRetryAt > now) at = row.nextRetryAt;
+            // reapExpired takes a claim strictly older than the TTL.
+            if (row.status === 'in_flight' && row.claimedAt !== undefined) at = row.claimedAt + opts.claimTtlMs + 1;
+            if (at !== undefined && (due === null || at < due)) due = at;
+        }
+        return due;
+    }
+
     async claim(opts: HttpClaimOptions): Promise<HttpDelivery[]> {
         const now = opts.now ?? Date.now();
         const claimed: HttpDelivery[] = [];
diff --git a/packages/services/service-messaging/src/memory-outbox.ts b/packages/services/service-messaging/src/memory-outbox.ts
index 9c5ce9bad31..4c8c76688f3 100644
--- a/packages/services/service-messaging/src/memory-outbox.ts
+++ b/packages/services/service-messaging/src/memory-outbox.ts
@@ -11,6 +11,7 @@ import type {
     NotificationDeliveryRecord,
     ReapOptions,
     DueWorkOptions,
+    NextDueOptions,
 } from './outbox.js';
 import {
     NotificationAckError,
@@ -82,6 +83,20 @@ export class MemoryNotificationOutbox implements INotificationOutbox {
         return false;
     }
 
+    /** [#29919] Earliest future `nextAttemptAt` of a `pending` row, or claim expiry of an `in_flight` one. */
+    async nextDueAt(opts: NextDueOptions): Promise<number | null> {
+        const now = opts.now ?? this.clock();
+        let due: number | null = null;
+        for (const r of this.rows.values()) {
+            let at: number | undefined;
+            if (r.status === 'pending' && r.nextAttemptAt != null && r.nextAttemptAt > now) at = r.nextAttemptAt;
+            // reapExpired takes a claim strictly older than the TTL.
+            if (r.status === 'in_flight' && r.claimedAt != null) at = r.claimedAt + opts.claimTtlMs + 1;
+            if (at !== undefined && (due === null || at < due)) due = at;
+        }
+        return due;
+    }
+
     async claim(opts: ClaimOptions): Promise<ClaimedDeliveryRecord[]> {
         const now = opts.now ?? this.clock();
         // Reap stale in_flight — unless the caller already reaped this pass.
diff --git a/packages/services/service-messaging/src/messaging-service-plugin.ts b/packages/services/service-messaging/src/messaging-service-plugin.ts
index 5c71ea0674f..57d0ba0beb2 100644
--- a/packages/services/service-messaging/src/messaging-service-plugin.ts
+++ b/packages/services/service-messaging/src/messaging-service-plugin.ts
@@ -123,12 +123,24 @@ export interface MessagingServicePluginOptions {
      * 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.
+     * While idle it bounds how late a dispatcher notices a row enqueued by
+     * another process; a deferred delivery coming due (retry, quiet hours,
+     * digest window) and a crashed node's claim passing its timeout wake the
+     * dispatcher at their due time instead (#29919). A value at or below
+     * `dispatchIntervalMs` disables the backoff.
      */
     dispatchMaxIdleIntervalMs?: number;
+    /**
+     * [#29919] This kernel is the only process that writes its database's
+     * delivery outboxes (a hosted tenant kernel: one kernel per tenant
+     * database), so every row is enqueued here and wakes its dispatcher. Both
+     * dispatchers then sleep straight from an idle tick to the next due row or
+     * `dispatchMaxIdleIntervalMs`, instead of re-running the backoff's ramp
+     * from `dispatchIntervalMs` after every boot and every burst (default
+     * `false`). Leave it off when another process — a second node, a worker —
+     * writes the same tables: the ramp is what notices its rows sooner.
+     */
+    dispatchSoleWriter?: boolean;
     /**
      * Topics that bypass the per-user preference matrix (ADR-0030 P2) — e.g.
      * security/system alerts users must not be able to mute. Exact match, or a
@@ -218,6 +230,7 @@ export class MessagingServicePlugin implements Plugin {
             partitionCount: 8,
             dispatchIntervalMs: 500,
             dispatchMaxIdleIntervalMs: DEFAULT_MAX_IDLE_INTERVAL_MS,
+            dispatchSoleWriter: false,
             mandatoryTopics: [],
             ...options,
         };
@@ -561,6 +574,7 @@ export class MessagingServicePlugin implements Plugin {
                     partitionCount: this.options.partitionCount,
                     intervalMs: this.options.dispatchIntervalMs,
                     maxIdleIntervalMs: this.options.dispatchMaxIdleIntervalMs,
+                    soleWriter: this.options.dispatchSoleWriter,
                     logger: ctx.logger,
                     textRenderer,
                 });
@@ -583,6 +597,7 @@ export class MessagingServicePlugin implements Plugin {
                     partitionCount: this.options.partitionCount,
                     intervalMs: this.options.dispatchIntervalMs,
                     maxIdleIntervalMs: this.options.dispatchMaxIdleIntervalMs,
+                    soleWriter: this.options.dispatchSoleWriter,
                     logger: ctx.logger,
                 });
                 this.httpDispatcher.start();
diff --git a/packages/services/service-messaging/src/next-due.ts b/packages/services/service-messaging/src/next-due.ts
new file mode 100644
index 00000000000..9c642f9e5d2
--- /dev/null
+++ b/packages/services/service-messaging/src/next-due.ts
@@ -0,0 +1,26 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * [#29919] The one fold both SQL outboxes' `nextDueAt` end in: the earlier of
+ * a deferred row's due time and the instant the oldest claim becomes reapable.
+ *
+ * Both cells are `Field.number` epoch-ms columns, read back raw from whatever
+ * driver holds them, so each is coerced and a cell that denotes no instant
+ * (`null`, a non-numeric value) contributes nothing rather than a `NaN` that
+ * would poison the comparison. A claim is reaped once it is STRICTLY older than
+ * the TTL (`claimed_at < now - claimTtlMs`), hence the `+ 1`.
+ */
+export function earliestDueAt(deferredAt: unknown, claimedAt: unknown, claimTtlMs: number): number | null {
+    const candidates: number[] = [];
+    const deferred = instantMs(deferredAt);
+    if (deferred !== null) candidates.push(deferred);
+    const claimed = instantMs(claimedAt);
+    if (claimed !== null) candidates.push(claimed + claimTtlMs + 1);
+    return candidates.length > 0 ? Math.min(...candidates) : null;
+}
+
+function instantMs(cell: unknown): number | null {
+    if (cell === null || cell === undefined || cell === '') return null;
+    const n = Number(cell);
+    return Number.isFinite(n) ? n : null;
+}
diff --git a/packages/services/service-messaging/src/outbox.ts b/packages/services/service-messaging/src/outbox.ts
index 26ad59b2fe6..3e1dbadb036 100644
--- a/packages/services/service-messaging/src/outbox.ts
+++ b/packages/services/service-messaging/src/outbox.ts
@@ -157,6 +157,14 @@ export interface DueWorkOptions {
     now?: number;
 }
 
+/** [#29919] Options for {@link INotificationOutbox.nextDueAt}. */
+export interface NextDueOptions {
+    /** Visibility timeout — an `in_flight` row becomes reapable once claimed longer ago than this. */
+    claimTtlMs: number;
+    /** "Now" reference, ms. Defaults to the store's clock. */
+    now?: number;
+}
+
 export interface AckSuccess {
     success: true;
     durationMs?: number;
@@ -296,6 +304,19 @@ export interface INotificationOutbox {
      * implement it.
      */
     hasDueWork?(opts: DueWorkOptions): Promise<boolean>;
+    /**
+     * [#29919] The earliest instant (ms) at which a row becomes claimable
+     * without anyone writing to the outbox — a `pending` row's `nextAttemptAt`
+     * still after `now` (a retry, quiet hours, a digest window closing), or the
+     * instant an `in_flight` row's claim passes `claimTtlMs` (already past if
+     * no reap has run since) — or `null` when there is none. The dispatcher's loop sleeps no later than this after an idle tick.
+     *
+     * A read and nothing more, stale the moment it returns like
+     * {@link hasDueWork}: a row written afterwards reaches the loop by the
+     * writer's `wake()`. Optional: without it the dispatcher sleeps by its idle
+     * backoff alone. Both built-in stores implement it.
+     */
+    nextDueAt?(opts: NextDueOptions): Promise<number | null>;
     claim(opts: ClaimOptions): Promise<ClaimedDeliveryRecord[]>;
     /**
      * Record the outcome of ONE dispatch attempt on a row this caller claimed,
diff --git a/packages/services/service-messaging/src/sql-http-outbox.ts b/packages/services/service-messaging/src/sql-http-outbox.ts
index e7ba88f3dc1..97dca0b69b6 100644
--- a/packages/services/service-messaging/src/sql-http-outbox.ts
+++ b/packages/services/service-messaging/src/sql-http-outbox.ts
@@ -30,10 +30,12 @@ import {
     type HttpDeliveryStatus,
     type HttpReapOptions,
     type HttpDueWorkOptions,
+    type HttpNextDueOptions,
     type IHttpOutbox,
     type RedeliverOptions,
     type UndeliverableHttpInput,
 } from './http-outbox.js';
+import { earliestDueAt } from './next-due.js';
 import { SYS_HTTP_DELIVERY } from './objects/http-delivery.object.js';
 
 export interface SqlHttpOutboxOptions {
@@ -244,6 +246,30 @@ export class SqlHttpOutbox implements IHttpOutbox {
         return due.length > 0;
     }
 
+    /**
+     * [#29919] Two reads, asked only after a tick that found nothing: the
+     * earliest future `next_retry_at` of a `pending` row in any partition, and
+     * the oldest `in_flight` claim, on the reaper's `status, claimed_at` index.
+     * The first sorts the deferred `pending` rows; there are none on an idle
+     * outbox.
+     */
+    async nextDueAt(opts: HttpNextDueOptions): Promise<number | null> {
+        const now = opts.now ?? Date.now();
+        const deferred = await this.engine.find(this.objectName, {
+            where: { status: 'pending', next_retry_at: { $gt: now } },
+            fields: ['next_retry_at'],
+            orderBy: [{ field: 'next_retry_at', order: 'asc' }],
+            limit: 1,
+        }, { context: DISPATCHER_SYSTEM_CONTEXT });
+        const claimed = await this.engine.find(this.objectName, {
+            where: { status: 'in_flight' },
+            fields: ['claimed_at'],
+            orderBy: [{ field: 'claimed_at', order: 'asc' }],
+            limit: 1,
+        }, { context: DISPATCHER_SYSTEM_CONTEXT });
+        return earliestDueAt(deferred[0]?.next_retry_at, claimed[0]?.claimed_at, opts.claimTtlMs);
+    }
+
     async claim(opts: HttpClaimOptions): Promise<HttpDelivery[]> {
         const now = opts.now ?? Date.now();
 
diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts
index edcf8ef1d8d..c8ca45c3052 100644
--- a/packages/services/service-messaging/src/sql-outbox.ts
+++ b/packages/services/service-messaging/src/sql-outbox.ts
@@ -12,9 +12,11 @@ import type {
     NotificationDeliveryRecord,
     ReapOptions,
     DueWorkOptions,
+    NextDueOptions,
 } from './outbox.js';
 import { hashPartition } from './backoff.js';
 import { toEpochMs } from './audit-timestamp.js';
+import { earliestDueAt } from './next-due.js';
 import {
     DISPATCHER_SYSTEM_CONTEXT,
     OUTBOX_SYSTEM_CONTEXT,
@@ -172,6 +174,30 @@ export class SqlNotificationOutbox implements INotificationOutbox {
         return due.length > 0;
     }
 
+    /**
+     * [#29919] Two reads, asked only after a tick that found nothing: the
+     * earliest future `next_attempt_at` of a `pending` row — every deferred
+     * row, batched (digest window) or not, in any partition — and the oldest
+     * `in_flight` claim, on the reaper's `status, claimed_at` index. The first
+     * sorts the deferred `pending` rows; there are none on an idle outbox.
+     */
+    async nextDueAt(opts: NextDueOptions): Promise<number | null> {
+        const now = opts.now ?? Date.now();
+        const deferred = await this.engine.find(this.objectName, {
+            where: { status: 'pending', next_attempt_at: { $gt: now } },
+            fields: ['next_attempt_at'],
+            orderBy: [{ field: 'next_attempt_at', order: 'asc' }],
+            limit: 1,
+        }, { context: DISPATCHER_SYSTEM_CONTEXT });
+        const claimed = await this.engine.find(this.objectName, {
+            where: { status: 'in_flight' },
+            fields: ['claimed_at'],
+            orderBy: [{ field: 'claimed_at', order: 'asc' }],
+            limit: 1,
+        }, { context: DISPATCHER_SYSTEM_CONTEXT });
+        return earliestDueAt(deferred[0]?.next_attempt_at, claimed[0]?.claimed_at, opts.claimTtlMs);
+    }
+
     async claim(opts: ClaimOptions): Promise<ClaimedDeliveryRecord[]> {
         const now = opts.now ?? Date.now();
 
diff --git a/packages/services/service-queue/src/db-queue-adapter.ts b/packages/services/service-queue/src/db-queue-adapter.ts
index 67be673ab50..efdc89a59e4 100644
--- a/packages/services/service-queue/src/db-queue-adapter.ts
+++ b/packages/services/service-queue/src/db-queue-adapter.ts
@@ -146,13 +146,22 @@ export interface DbQueueAdapterOptions {
    * straight back to `pollIntervalMs`.
    *
    * A value at or below `pollIntervalMs` disables the backoff and restores the
-   * flat poll. What the backoff costs is latency on work this process was not
-   * told about — a row another node wrote, a deferred row coming due, a
-   * crashed worker's lease expiring: never noticed more than this long after it
-   * became claimable. Work published through THIS adapter wakes the loop, so it
-   * is unaffected.
+   * flat poll. What the backoff costs is latency on a row another node wrote:
+   * never noticed more than this long after it became claimable. Work
+   * published through THIS adapter wakes the loop, and a deferred row coming
+   * due wakes it at its `scheduled_for` (#29919), so neither waits on this.
    */
   maxIdleIntervalMs?: number;
+  /**
+   * [#29919] No other process writes this database's `sys_job_queue` (a hosted
+   * tenant kernel: one kernel per tenant database), so every row is published
+   * through this adapter and wakes its loop. An idle tick then sleeps straight
+   * to the next deferred row or {@link maxIdleIntervalMs} instead of
+   * re-running the backoff's ramp from `pollIntervalMs` after every boot and
+   * every burst — the ramp only ever bought latency on rows another process
+   * wrote. Default `false`.
+   */
+  soleWriter?: boolean;
   /** Max messages claimed per poll tick (default 10) */
   batchSize?: number;
   /** Lease duration before another worker may reclaim (ms, default 30000) */
@@ -225,6 +234,7 @@ export class DbQueueAdapter implements IQueueService {
     this.opts = {
       pollIntervalMs: o.pollIntervalMs ?? 1000,
       maxIdleIntervalMs: o.maxIdleIntervalMs ?? DEFAULT_MAX_IDLE_INTERVAL_MS,
+      soleWriter: o.soleWriter ?? false,
       batchSize: o.batchSize ?? 10,
       leaseMs: o.leaseMs ?? 30_000,
       idempotencyWindowMs: o.idempotencyWindowMs ?? 24 * 60 * 60 * 1000,
@@ -372,11 +382,14 @@ export class DbQueueAdapter implements IQueueService {
     }, { context: SYSTEM_CTX });
 
     // [#17612] Tick now rather than wait out the idle backoff — but only for a
-    // row that is claimable THIS instant. A deferred row is exactly the case
-    // the loop's contract already covers (noticed within one backed-off
-    // interval of coming due), and waking for it would reset the backoff to
-    // `pollIntervalMs` for a tick guaranteed to claim nothing.
-    if (Date.parse(scheduledFor) <= now.getTime()) this.wake();
+    // row that is claimable THIS instant: waking for a deferred row would reset
+    // the backoff to `pollIntervalMs` for a tick guaranteed to claim nothing.
+    // [#29919] A deferred row instead tells the loop when it comes due — no
+    // read, no tick now — so it goes out on time even with a minutes-long
+    // ceiling, rather than up to a backed-off interval late.
+    const untilDueMs = Date.parse(scheduledFor) - now.getTime();
+    if (untilDueMs <= 0) this.wake();
+    else this.loop?.wakeAt(Date.now() + untilDueMs);
 
     return id;
   }
@@ -479,6 +492,8 @@ export class DbQueueAdapter implements IQueueService {
     this.loop = new DispatchLoop({
       intervalMs: this.opts.pollIntervalMs,
       maxIdleIntervalMs: this.opts.maxIdleIntervalMs,
+      soleWriter: this.opts.soleWriter,
+      nextDueAt: () => this.nextDueAt(),
       runTick: () => this.pollOnce(),
       onTickError: (err) => { this.logger?.warn?.('DbQueueAdapter: poll tick failed', err); },
     });
@@ -518,6 +533,40 @@ export class DbQueueAdapter implements IQueueService {
     return processed;
   }
 
+  /**
+   * [#29919] When the earliest deferred `pending` row of a subscribed queue
+   * comes due, on the loop's clock (`Date.now()`), or `null` when none is
+   * deferred. One read for every subscribed queue, asked by the loop once per
+   * idle streak; it sorts the deferred rows, of which an idle queue has none.
+   *
+   * `scheduled_for` is the only due time a queue row carries: every write that
+   * returns a row to `pending` clears its lease (`locked_until`), so no lease
+   * expiry gates a `pending` row.
+   */
+  private async nextDueAt(): Promise<number | null> {
+    const queues = [...this.handlers.keys()];
+    if (queues.length === 0) return null;
+    const now = this.now();
+    const rows = await this.engine.find(QUEUE_TABLE, {
+      where: {
+        queue: { $in: queues },
+        status: 'pending',
+        scheduled_for: { $gt: now.toISOString() },
+      },
+      fields: ['scheduled_for'],
+      orderBy: [{ field: 'scheduled_for', order: 'asc' }],
+      limit: 1,
+      context: SYSTEM_CTX,
+    });
+    const cell = rows?.[0]?.scheduled_for;
+    // Instants, not strings — the dialects materialize `scheduled_for`
+    // differently (ISO text on SQLite/turso, a `Date` on Postgres/MySQL).
+    const dueMs = cell ? new Date(cell).getTime() : NaN;
+    if (!Number.isFinite(dueMs)) return null;
+    // Re-based onto the loop's clock, so an injected `clock` keeps its meaning.
+    return Date.now() + (dueMs - now.getTime());
+  }
+
   // ── Internals ────────────────────────────────────────────────────
 
   private async claimBatch(queue: string, max: number): Promise<any[]> {

From 141fc4916c109d83e6251d6f50cc2710fb57b199 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 01:16:13 +0000
Subject: [PATCH 2/5] test: pin the next-due wake and the sole-writer cold
 start (#29919)

---
 packages/core/src/dispatch-loop.test.ts       | 294 ++++++++++++++++++
 .../src/dispatcher-idle-backoff.test.ts       |  43 ++-
 .../dispatcher-idle-cost.integration.test.ts  |  40 +++
 .../src/http-dispatcher-idle-backoff.test.ts  |  39 ++-
 ...p-dispatcher-idle-cost.integration.test.ts |  24 ++
 .../src/db-queue-idle-backoff.test.ts         | 101 +++++-
 6 files changed, 535 insertions(+), 6 deletions(-)
 create mode 100644 packages/core/src/dispatch-loop.test.ts

diff --git a/packages/core/src/dispatch-loop.test.ts b/packages/core/src/dispatch-loop.test.ts
new file mode 100644
index 00000000000..ce1f76220d6
--- /dev/null
+++ b/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]);
+    });
+});
diff --git a/packages/services/service-messaging/src/dispatcher-idle-backoff.test.ts b/packages/services/service-messaging/src/dispatcher-idle-backoff.test.ts
index 8ec2e0217aa..928c45ecf7e 100644
--- a/packages/services/service-messaging/src/dispatcher-idle-backoff.test.ts
+++ b/packages/services/service-messaging/src/dispatcher-idle-backoff.test.ts
@@ -73,7 +73,7 @@ function sendProbe(hold?: { recipient: string; until: Promise<void> }): SendProb
     };
 }
 
-function setup(options: { maxIdleIntervalMs?: number; probe?: SendProbe } = {}) {
+function setup(options: { maxIdleIntervalMs?: number; probe?: SendProbe; soleWriter?: boolean } = {}) {
     const outbox = new TickRecordingOutbox(1);
     const probe = options.probe ?? sendProbe();
     const dispatcher = new NotificationDispatcher({
@@ -84,6 +84,7 @@ function setup(options: { maxIdleIntervalMs?: number; probe?: SendProbe } = {})
         partitionCount: 1,
         intervalMs: BASE,
         maxIdleIntervalMs: options.maxIdleIntervalMs,
+        soleWriter: options.soleWriter,
     });
     return { outbox, probe, dispatcher };
 }
@@ -254,3 +255,43 @@ describe('#17610 MessagingService.emit() wakes the dispatcher its outbox is wire
         await dispatcher.stop();
     });
 });
+
+describe('#29919 NotificationDispatcher — a sole writer wakes at the next due time', () => {
+    const FIVE_MINUTES = 5 * MINUTE;
+
+    it('a fresh sole-writer dispatcher on an empty outbox ticks ONCE in its first minute', async () => {
+        const { outbox, dispatcher } = setup({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        dispatcher.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await dispatcher.stop();
+        expect(outbox.tickStarts).toHaveLength(1);
+    });
+
+    it('a deferred delivery written before boot is sent the moment it comes due, not a ceiling later', async () => {
+        const { outbox, probe, dispatcher } = setup({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        const t0 = Date.now();
+        await outbox.enqueue({ ...row('u1'), notBefore: t0 + 20_000 });
+        dispatcher.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await dispatcher.stop();
+
+        expect(probe.sent).toEqual(['u1']);
+        expect(outbox.tickStarts.slice(0, 2)).toEqual([t0, t0 + 20_000]);
+    });
+
+    it('a crashed node\'s claim is reaped and re-sent once its claim TTL passes, not a ceiling later', async () => {
+        const { outbox, probe, dispatcher } = setup({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        const t0 = Date.now();
+        await outbox.enqueue(row('u1'));
+        // Claimed by a node that then died: in flight, never acked.
+        const [claimed] = await outbox.claim({ nodeId: 'dead-node', limit: 1, claimTtlMs: 5_000 });
+        expect(claimed?.recipientId).toBe('u1');
+        dispatcher.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await dispatcher.stop();
+
+        expect(probe.sent).toEqual(['u1']);
+        // claimTtlMs defaults to 2 × lockTtlMs = 10 × intervalMs; reapable just after.
+        expect(outbox.tickStarts[1]! - t0).toBe(10 * BASE + 1);
+    });
+});
diff --git a/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts b/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts
index b48253c9f27..0dad580ea02 100644
--- a/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts
+++ b/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts
@@ -290,3 +290,43 @@ describe('#17610 #28383 NotificationDispatcher — idle tick cost', () => {
         expect(sent).toEqual(['u1']);
     });
 });
+
+describe('#29919 SqlNotificationOutbox.nextDueAt — the one read a sole-writer loop wakes by', () => {
+    it('an empty outbox answers null in TWO statements, both reads', async () => {
+        statements = [];
+        expect(await outbox.nextDueAt({ claimTtlMs: TTL })).toBeNull();
+        expect(statements).toHaveLength(2);
+        expect(statements.every((sql) => /^select /i.test(sql))).toBe(true);
+    });
+
+    it('answers the earliest deferred row — batched or not — and ignores a row already due', async () => {
+        const now = Date.now();
+        await outbox.enqueue({ notificationId: 'n_now', recipientId: 'u1', channel: 'inbox', payload: { title: 't' } });
+        await outbox.enqueue({
+            notificationId: 'n_later', recipientId: 'u2', channel: 'inbox', payload: { title: 't' },
+            notBefore: now + 60_000,
+        });
+        expect(await outbox.nextDueAt({ claimTtlMs: TTL, now })).toBe(now + 60_000);
+
+        // A digest window closing is a deferred row too.
+        await outbox.enqueue({
+            notificationId: 'n_digest', recipientId: 'u3', channel: 'inbox', payload: { title: 't' },
+            digestKey: 'u3|inbox|w1', notBefore: now + 30_000,
+        });
+        expect(await outbox.nextDueAt({ claimTtlMs: TTL, now })).toBe(now + 30_000);
+    });
+
+    it('answers when the oldest in-flight claim becomes reapable, when that is sooner', async () => {
+        const now = Date.now();
+        await outbox.enqueue({
+            notificationId: 'n_later', recipientId: 'u2', channel: 'inbox', payload: { title: 't' },
+            notBefore: now + 10 * TTL,
+        });
+        await outbox.enqueue({ notificationId: 'n_claimed', recipientId: 'u1', channel: 'inbox', payload: { title: 't' } });
+        const [claimed] = await outbox.claim({ nodeId: 'node-crashed', limit: 10, claimTtlMs: TTL, now: now - 1_000 });
+        expect(claimed?.recipientId).toBe('u1');
+
+        // `reapExpired` takes a claim strictly older than the TTL.
+        expect(await outbox.nextDueAt({ claimTtlMs: TTL, now })).toBe(now - 1_000 + TTL + 1);
+    });
+});
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
index e48aa9c8fa1..b8b1f3afa8d 100644
--- a/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts
+++ b/packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts
@@ -88,7 +88,7 @@ function fetchProbe(hold?: { url: string; until: Promise<void> }): FetchProbe {
     };
 }
 
-function setup(options: { maxIdleIntervalMs?: number; probe?: FetchProbe } = {}) {
+function setup(options: { maxIdleIntervalMs?: number; probe?: FetchProbe; soleWriter?: boolean } = {}) {
     const outbox = new TickRecordingOutbox();
     const probe = options.probe ?? fetchProbe();
     const dispatcher = new HttpDispatcher({
@@ -98,6 +98,7 @@ function setup(options: { maxIdleIntervalMs?: number; probe?: FetchProbe } = {})
         partitionCount: 1,
         intervalMs: BASE,
         maxIdleIntervalMs: options.maxIdleIntervalMs,
+        soleWriter: options.soleWriter,
     });
     return { outbox, probe, dispatcher };
 }
@@ -314,3 +315,39 @@ describe('#17623 MessagingService wakes the HTTP dispatcher its outbox is wired
         await dispatcher.stop();
     });
 });
+
+describe('#29919 HttpDispatcher — a sole writer wakes at the next due time', () => {
+    const FIVE_MINUTES = 5 * MINUTE;
+
+    it('a fresh sole-writer dispatcher on an empty outbox ticks ONCE in its first minute', async () => {
+        const { outbox, dispatcher } = setup({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        dispatcher.start();
+        await vi.advanceTimersByTimeAsync(MINUTE);
+        await dispatcher.stop();
+        expect(outbox.tickStarts).toHaveLength(1);
+    });
+
+    it('every retry is attempted the moment it comes due, under a five-minute ceiling', 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: FIVE_MINUTES, soleWriter: true,
+            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);
+        for (let i = 0; i < 3; i++) expect(attemptedAt[i + 1]! - outbox.retriesDueAt[i]!).toBe(0);
+        expect((await outbox.list())[0]).toMatchObject({ status: 'success', attempts: 4 });
+    });
+});
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
index 4986abb915d..9cef1b66a2a 100644
--- 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
@@ -264,3 +264,27 @@ describe('#17623 #28383 HttpDispatcher — idle tick cost', () => {
         expect(posted).toEqual(['https://receiver.example/r_legacy']);
     });
 });
+
+describe('#29919 SqlHttpOutbox.nextDueAt — the one read a sole-writer loop wakes by', () => {
+    it('an empty outbox answers null in TWO statements, both reads', async () => {
+        statements = [];
+        expect(await outbox.nextDueAt({ claimTtlMs: TTL })).toBeNull();
+        expect(statements).toHaveLength(2);
+        expect(statements.every((sql) => /^select /i.test(sql))).toBe(true);
+    });
+
+    it('answers a retry\'s due time, ignores a row already due, and answers a claim\'s expiry when sooner', async () => {
+        const now = Date.now();
+        await outbox.enqueue(delivery('r_now'));
+        await outbox.enqueue(delivery('r_retry'));
+        const claimed = await outbox.claim({ nodeId: 'node-a', limit: 10, claimTtlMs: TTL, now });
+        const retry = claimed.find((d) => d.refId === 'r_retry')!;
+        await outbox.ack(retry.id, { success: false, httpStatus: 500, durationMs: 1, nextRetryAt: now + 10 * TTL }, retry);
+        // r_now is still in flight under node-a, claimed at `now`.
+        expect(await outbox.nextDueAt({ claimTtlMs: TTL, now })).toBe(now + TTL + 1);
+
+        const live = claimed.find((d) => d.refId === 'r_now')!;
+        await outbox.ack(live.id, { success: true, httpStatus: 200, durationMs: 1 }, live);
+        expect(await outbox.nextDueAt({ claimTtlMs: TTL, now })).toBe(now + 10 * TTL);
+    });
+});
diff --git a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
index cf95f5f818a..95c5c970985 100644
--- a/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
+++ b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts
@@ -55,6 +55,7 @@ function makeCountingEngine() {
                 switch (op) {
                     case '$lte': if (cell == null || !(String(cell) <= String(target))) return false; break;
                     case '$lt': if (cell == null || !(String(cell) < String(target))) return false; break;
+                    case '$gt': if (cell == null || !(String(cell) > String(target))) return false; break;
                     case '$ne': if (cell === target) return false; break;
                     case '$in': if (!(target as unknown[]).includes(cell)) return false; break;
                     default: throw new Error(`fake driver: unsupported operator ${op}`);
@@ -162,8 +163,9 @@ describe('#17612 DbQueueAdapter — idle poll cost', () => {
         await vi.advanceTimersByTimeAsync(HOUR);
         await adapter.stop();
 
-        // 3600 s / 1 s, plus the tick `start()` runs immediately.
-        expect(engine.calls.find).toBe(3601);
+        // 3600 s / 1 s, plus the tick `start()` runs immediately, plus the ONE
+        // due-time read of the idle streak (#29919) — once, not once a tick.
+        expect(engine.calls.find).toBe(3602);
     });
 
     it('backs off while idle: the same hour, same queue, is 124 reads instead of 3600', async () => {
@@ -176,8 +178,9 @@ describe('#17612 DbQueueAdapter — idle poll cost', () => {
         await adapter.stop();
 
         // 1s,2s,4s,8s,16s then the 30 s ceiling — pinned exactly, because a
-        // bound alone is met by a loop that stopped.
-        expect(engine.calls.find).toBe(124);
+        // bound alone is met by a loop that stopped — plus the streak's one
+        // due-time read (#29919).
+        expect(engine.calls.find).toBe(125);
         // Nothing was claimed, so nothing was written.
         expect(engine.calls.update).toBe(0);
     });
@@ -288,3 +291,93 @@ describe('#17612 DbQueueAdapter — idle poll cost', () => {
         expect(engine.calls.find).toBe(findsAtStop);
     });
 });
+
+describe('#29919 DbQueueAdapter — wake at the next due time, no ramp for a sole writer', () => {
+    const FIVE_MINUTES = 5 * 60_000;
+
+    it('a sole writer on an idle queue ticks ONCE in a fresh kernel\'s first minute: two reads, not five', async () => {
+        vi.useFakeTimers();
+        const engine = makeCountingEngine();
+        const adapter = adapterOn(engine, { soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        await adapter.subscribe('q1', async () => {});
+        adapter.start();
+        await vi.advanceTimersByTimeAsync(60_000);
+        await adapter.stop();
+
+        // The cold-start tick's claim read and its one due-time read.
+        expect(engine.calls.find).toBe(2);
+    });
+
+    it('CONTROL: without soleWriter the same minute still ramps — 1 + 5 claim reads and the due read', async () => {
+        vi.useFakeTimers();
+        const engine = makeCountingEngine();
+        const adapter = adapterOn(engine, { maxIdleIntervalMs: FIVE_MINUTES });
+        await adapter.subscribe('q1', async () => {});
+        adapter.start();
+        await vi.advanceTimersByTimeAsync(60_000);
+        await adapter.stop();
+
+        // Ticks at 0, 2, 6, 14, 30 s (1 s base, doubling), then 62 s.
+        expect(engine.calls.find).toBe(6);
+    });
+
+    it('a row scheduled before boot is claimed within one tick of coming due, under a five-minute ceiling', async () => {
+        vi.useFakeTimers();
+        const engine = makeCountingEngine();
+        const writer = adapterOn(engine);
+        const t0 = Date.now();
+        await writer.publish('q1', { n: 1 }, { delay: 20_000 });
+
+        const adapter = adapterOn(engine, { soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        const at: number[] = [];
+        await adapter.subscribe('q1', async () => { at.push(Date.now()); });
+        adapter.start();
+        await vi.advanceTimersByTimeAsync(60_000);
+        await adapter.stop();
+
+        expect(at).toHaveLength(1);
+        expect(at[0]! - t0).toBe(20_000);
+    });
+
+    it('a deferred publish in this process goes out on time with no extra read — it hints, it does not wake', async () => {
+        vi.useFakeTimers();
+        const engine = makeCountingEngine();
+        const adapter = adapterOn(engine, { soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        const at: number[] = [];
+        await adapter.subscribe('q1', async () => { at.push(Date.now()); });
+        adapter.start();
+        await vi.advanceTimersByTimeAsync(10_000);
+        const findsWhileIdle = engine.calls.find;
+
+        const publishedAt = Date.now();
+        await adapter.publish('q1', { n: 1 }, { delay: 20_000 });
+        await vi.advanceTimersByTimeAsync(1);
+        // No tick and no read now — the publish's own dedup reads nothing (no key).
+        expect(engine.calls.find).toBe(findsWhileIdle);
+
+        await vi.advanceTimersByTimeAsync(60_000);
+        await adapter.stop();
+        expect(at).toHaveLength(1);
+        expect(at[0]! - publishedAt).toBe(20_000);
+    });
+
+    it('a failing retry is re-attempted at its backoff, not at the ceiling', async () => {
+        vi.useFakeTimers();
+        const engine = makeCountingEngine();
+        const adapter = adapterOn(engine, { soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
+        const at: number[] = [];
+        await adapter.subscribe('q1', async () => {
+            at.push(Date.now());
+            if (at.length === 1) throw new Error('first attempt fails');
+        });
+        adapter.start();
+        await vi.advanceTimersByTimeAsync(10_000);
+        await adapter.publish('q1', { n: 1 }, { backoff: { type: 'fixed', delayMs: 15_000 } });
+        await vi.advanceTimersByTimeAsync(60_000);
+        await adapter.stop();
+
+        expect(at).toHaveLength(2);
+        expect(at[1]! - at[0]!).toBe(15_000);
+    });
+});
+

From 886c3568f7645484274eb92e5a0c9efd13316e9e Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 01:18:08 +0000
Subject: [PATCH 3/5] chore(changeset): DispatchLoop next-due wake and
 sole-writer cold start (#29919)

Claude-Session: https://claude.ai/code/session_01Utp6CY2ugcVZ5KKbTbWpXw
Co-authored-by: Claude <noreply@anthropic.com>
---
 .changeset/29919-dispatch-loop-next-due-wake.md | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)
 create mode 100644 .changeset/29919-dispatch-loop-next-due-wake.md

diff --git a/.changeset/29919-dispatch-loop-next-due-wake.md b/.changeset/29919-dispatch-loop-next-due-wake.md
new file mode 100644
index 00000000000..e93288f72b4
--- /dev/null
+++ b/.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.

From 2d31b75648040288192da3f50961eea5919f852e Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 01:21:59 +0000
Subject: [PATCH 4/5] test(service-messaging): type the claim credential in the
 next-due integration pin (#29919)

Claude-Session: https://claude.ai/code/session_01Utp6CY2ugcVZ5KKbTbWpXw
Co-authored-by: Claude <noreply@anthropic.com>
---
 .../src/http-dispatcher-idle-cost.integration.test.ts     | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

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
index 9cef1b66a2a..f9772adf184 100644
--- 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
@@ -61,7 +61,7 @@ import { HttpDelivery, SYS_HTTP_DELIVERY } from './objects/http-delivery.object.
 import { HttpDispatcher } from './http-dispatcher.js';
 import { hashPartition } from './backoff.js';
 import type { FetchImpl } from './http-sender.js';
-import type { EnqueueHttpInput, HttpClaimCredential, IHttpOutbox } from './http-outbox.js';
+import type { EnqueueHttpInput, HttpClaimCredential, HttpDelivery as HttpDeliveryRecord, IHttpOutbox } from './http-outbox.js';
 import { installDefaultOrganization } from './default-organization.testkit.js';
 
 /** The production default. */
@@ -266,6 +266,8 @@ describe('#17623 #28383 HttpDispatcher — idle tick cost', () => {
 });
 
 describe('#29919 SqlHttpOutbox.nextDueAt — the one read a sole-writer loop wakes by', () => {
+    const credential = (d: HttpDeliveryRecord): HttpClaimCredential => ({ claimedBy: d.claimedBy!, claimedAt: d.claimedAt! });
+
     it('an empty outbox answers null in TWO statements, both reads', async () => {
         statements = [];
         expect(await outbox.nextDueAt({ claimTtlMs: TTL })).toBeNull();
@@ -279,12 +281,12 @@ describe('#29919 SqlHttpOutbox.nextDueAt — the one read a sole-writer loop wak
         await outbox.enqueue(delivery('r_retry'));
         const claimed = await outbox.claim({ nodeId: 'node-a', limit: 10, claimTtlMs: TTL, now });
         const retry = claimed.find((d) => d.refId === 'r_retry')!;
-        await outbox.ack(retry.id, { success: false, httpStatus: 500, durationMs: 1, nextRetryAt: now + 10 * TTL }, retry);
+        await outbox.ack(retry.id, { success: false, httpStatus: 500, durationMs: 1, nextRetryAt: now + 10 * TTL }, credential(retry));
         // r_now is still in flight under node-a, claimed at `now`.
         expect(await outbox.nextDueAt({ claimTtlMs: TTL, now })).toBe(now + TTL + 1);
 
         const live = claimed.find((d) => d.refId === 'r_now')!;
-        await outbox.ack(live.id, { success: true, httpStatus: 200, durationMs: 1 }, live);
+        await outbox.ack(live.id, { success: true, httpStatus: 200, durationMs: 1 }, credential(live));
         expect(await outbox.nextDueAt({ claimTtlMs: TTL, now })).toBe(now + 10 * TTL);
     });
 });

From a1a9fbd308c37468d652a5f60daa7cde68e4c095 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 5 Nov 2026 02:49:36 +0000
Subject: [PATCH 5/5] docs(webhooks): the HTTP dispatcher's idle tick is two
 reads, it wakes at the next due time, and dispatchSoleWriter / db.soleWriter
 skip the ramp (#29919)

Claude-Session: https://claude.ai/code/session_01Utp6CY2ugcVZ5KKbTbWpXw
Co-authored-by: Claude <noreply@anthropic.com>
---
 content/docs/automation/webhooks.mdx | 45 +++++++++++++++++++++-------
 1 file changed, 34 insertions(+), 11 deletions(-)

diff --git a/content/docs/automation/webhooks.mdx b/content/docs/automation/webhooks.mdx
index b04fc6c5bbe..81f4ad53494 100644
--- a/content/docs/automation/webhooks.mdx
+++ b/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. |