Skip to content

Commit 771b2db

Browse files
fix(core): an idle DispatchLoop wakes at the next due time, and a sole writer skips the cold-start ramp (#30225)
<!-- claude-projects-attribution: {"github_login":"os-bill"} --> _Requested by **Bill** · [project thread](https://claude.ai/code/project/chan_016iMXWSZDzqWRWvWzdNo8wg?thread=cmsg_016iMXWSZDzqWRWvWzdNo8wgFmm31mRxusPCgeS2bBQHgg)_ domain:server#5 · os-bill · session_01Utp6CYVV6fcAQrNTDQRvPb ## 维护者速读 - 改了什么:投递调度器和任务队列共用的轮询循环,空闲时改为一次读出"下一条到期时间"并睡到那时;宿主声明自己是唯一写入方时(云端租户内核),开机不再从 0.5 秒起一倍倍退避。 - 为什么改:全新内核第一分钟里三个循环对着空表白轮询约 30 条语句;同进程写入本来就会立即唤醒循环。现在单写者内核每个循环第一分钟只跑一次。 - 风险与回滚:默认不开单写者,老宿主只多一条"到期时间"读(每段空闲一次);重试、静默时段、摘要窗口改为准点发出。回滚即撤本 PR,无数据迁移。 - 先看:`packages/core/src/dispatch-loop.ts`(`nextDelay` 与防空转判断)、`packages/services/service-queue/src/db-queue-adapter.ts`(延迟发布改用 `wakeAt`)、`packages/services/service-messaging/src/sql-outbox.ts`(`nextDueAt` 两条读)。 - 两分钟验证:`pnpm --filter @objectstack/core exec vitest run --project local src/dispatch-loop.test.ts`,预期 15 条全绿,其中"fresh loop ticks ONCE in its first minute"对照组为 6 次。 - 需要你做的:无,席位落地。 Fixes #29919 Clause-②: yes (widening) — measured on the built entry closure: `@objectstack/core` `dist/index.d.ts` gains `DispatchLoopOptions.nextDueAt`, `DispatchLoopOptions.soleWriter` and `DispatchLoop.wakeAt`; `@objectstack/service-messaging` gains optional `INotificationOutbox.nextDueAt` / `IHttpOutbox.nextDueAt`, the `NextDueOptions` / `HttpNextDueOptions` types, `soleWriter` on both dispatchers' options and `MessagingServicePluginOptions.dispatchSoleWriter`; `@objectstack/service-queue` gains `DbQueueAdapterOptions.soleWriter`. Every addition is optional; nothing is removed or narrowed. Metadata-protocol: none ## What changes **`DispatchLoop` wakes at the next due time (card item 1).** A loop given `nextDueAt()` asks for the earliest instant work becomes claimable once per idle streak: after the first empty tick following `start()`, a tick that claimed work, a `wake()`, or the arrival of the due time it was sleeping towards. It then never sleeps past that instant. The answer is remembered, not re-read on every ceiling tick, so an idle loop's steady-state cost does not move. A deferred row written in this process later in the streak reaches the loop through the new `wakeAt(dueAtMs)`, which moves the timer and costs no read and no tick. Two guards keep it from spinning. A due time at or before the start of the tick that just found nothing is dropped, because that work was visible and not claimable by this loop (another node holds its partition, for example). A rejected tick or a rejected read counts as idle and sleeps by the backoff. **The single-writer signal (card item 2).** No existing host signal says "I am the only writer", and none is needed in `packages/spec`: the dispatchers' and the queue's options are TypeScript plugin options. `soleWriter` (default `false`) on `DispatchLoopOptions`, on both dispatchers, as `MessagingServicePlugin` `dispatchSoleWriter`, and on `DbQueueAdapterOptions` (reachable as `QueueServicePlugin` `db.soleWriter`) makes an empty tick sleep straight to the next due time or the ceiling instead of re-running the ramp from `intervalMs`. Item 1 alone does not meet "Done when": without the flag the ramp keeps its shape (the CONTROL legs below read 6 ticks in the first minute), and only the due time cuts it short. **Due-time sources**, each read only after an idle tick: | Loop | Read | Sources | |---|---|---| | `NotificationDispatcher` | `SqlNotificationOutbox.nextDueAt`, 2 SELECTs | earliest future `next_attempt_at` of a `pending` row (retry, quiet hours, digest window close), oldest `in_flight` `claimed_at` + `claimTtlMs` + 1 (the reaper takes a claim strictly older than the TTL) | | `HttpDispatcher` | `SqlHttpOutbox.nextDueAt`, 2 SELECTs | earliest future `next_retry_at`, oldest `in_flight` claim + TTL + 1 | | `DbQueueAdapter` | 1 SELECT over the subscribed queues | earliest future `scheduled_for` | The queue has no lease expiry to read: every write that returns a row to `pending` clears `locked_until`, so no lease gates a `pending` row. The in-flight read uses the reaper's `status, claimed_at` index; the deferred reads sort the deferred `pending` rows, and an idle store has none. A deferred `publish()` used to skip the wake on purpose, so as not to reset the backoff. It now calls `wakeAt`, so the row goes out at its `scheduled_for` even under a minutes-long ceiling. **Docs (head `a1a9fbd3`, a docs-only commit the seat asked for at review).** `content/docs/automation/webhooks.mdx`: an idle tick costs two reads whatever `partitionCount` is (the old `1 + partitionCount` figure was stale since #28383); a new paragraph on waking at the next due time (a retry is attempted when due, a crashed node's claim is recovered when its TTL passes; the backoff is left only for rows another process wrote); `dispatchSoleWriter` and the queue's `db.soleWriter`: what they do, who turns them on (one kernel per tenant database), who leaves them off (a second node or worker on the same database), default off; the crash row of the failure table matches. `kernel/cluster.mdx`, `plugins/packages.mdx` and `api/environment-routing.mdx` were read and nothing in them is made false by this PR. ## Statement counts, before and after Taken on this repo's harnesses, at the client (knex `query` events on the real `ObjectQL` + `SqlDriver` better-sqlite3 harness for the outboxes; engine calls on the counting engine for the queue). The card's figures were taken at the Turso client in cloud, which counts more per read, so the absolute numbers differ. The tick counts are the same. | Loop, fresh kernel, empty store, first minute, 5-minute ceiling | before (`main`) | after, `soleWriter` | after, default | |---|---|---|---| | `NotificationDispatcher` | 6 ticks × 2 = **12** | 1 tick × 2 + due read 2 = **4** | 6 ticks × 2 + 2 = 14 | | `HttpDispatcher` | 6 × 2 = **12** | 2 + 2 = **4** | 14 | | `DbQueueAdapter` (one queue, 1 s base) | 5 ticks × 1 = **5** | 1 + 1 = **2** | 6 | | Total | **29** | **10** | 34 | Steady state at the ceiling is unchanged: an idle hour at 30 s reads the due time once (`steady state costs no extra read` pin, 121 ticks, 1 read). Without the flag, an idle stretch costs one extra read at its start. That is the price of "a deferred row goes out within one tick of coming due" on a multi-writer host. ## Tests <details><summary>New pins (fake timers, exact instants)</summary> - `packages/core/src/dispatch-loop.test.ts` (new, 15): a fresh sole-writer loop ticks once in its first minute, with a CONTROL showing the ramp the harness can see (0, 1, 3, 7, 15, 31 s). `soleWriter` alone goes to the ceiling. Steady state reads the due time once an hour. A row due at T+20 s is claimed at T+20 s under a 5-minute ceiling. The ramp is cut short only by a due time. Consecutive deferred rows are each on time. `wakeAt` costs no read and keeps a sooner timer. `wake()` ticks at once. A rejected tick is idle (one tick per ceiling) and keeps a learned due time without spinning. A due time at or before the empty tick is not chased. A rejected read falls back to the backoff. `stop()` is final for `wakeAt`. - `db-queue-idle-backoff.test.ts` (+5, 2 moved): a sole-writer queue reads 2 in the first minute (CONTROL 6), a row scheduled before boot runs at exactly +20 s, a deferred in-process publish runs at exactly +20 s with no extra read, and a failing retry is re-attempted at its 15 s backoff. The two #17612 exact counts move by the one due read of the streak (3601 → 3602, 124 → 125), with a comment saying so. - `dispatcher-idle-backoff.test.ts` (+3) and `http-dispatcher-idle-backoff.test.ts` (+2): one tick in a sole-writer dispatcher's first minute. A deferred delivery written before boot is sent at +20 s. A crashed node's claim is re-sent at claim TTL + 1 ms. Every HTTP retry is attempted at exactly its due time under a 5-minute ceiling. - `dispatcher-idle-cost.integration.test.ts` (+3) and `http-dispatcher-idle-cost.integration.test.ts` (+2), on a real `ObjectQL` + `SqlDriver`: `nextDueAt` on an empty outbox is null in exactly 2 SELECT statements. It answers the earliest deferred row, digest rows included, and ignores a row already due. It answers the oldest claim's expiry when that is sooner. </details> <details><summary>Ablations: each mutation made through scripts/ablation-replace.mjs, landed on disk, red, then restored with blob == HEAD</summary> All three suites resolve `@objectstack/core` to its source through their vitest alias, so no `dist/` rebuild is involved. | Mutation | Suite | Red | |---|---|---| | A1: put the ramp back (delete `if (this.soleWriter) return this.maxIdleIntervalMs;`) | core | 3 (ticks ONCE; soleWriter alone; steady state) | | A1 | queue | 1 (sole writer first minute) | | A1 | messaging | 3 (both dispatchers' first minute, and the deferred / claim legs) | | A2: ignore the due time in `nextDelay` | core 5 · queue 2 · messaging 3 | every due-time leg | | A3: drop the guard that discards a due time at or before the tick's start | core 2 | the not-chased leg and the rejected-after-due leg spin | | A4: drop the queue's `wakeAt` on a deferred publish | queue 1 | deferred in-process publish | </details> ## Acceptance notes - Item 2 lands as `soleWriter` / `dispatchSoleWriter` / `db.soleWriter`. Cloud turns it on for hosted tenant kernels together with the queue's ceiling under cloud#3413; nothing in this repo turns it on. With cloud's 5-minute `dispatchMaxIdleIntervalMs`, each dispatcher ticks once in the first minute. The queue ticks once only after cloud also raises `db.maxIdleIntervalMs` (the card says that is cloud's change); at the framework's 30 s default a sole-writer queue ticks at 0 s and 30 s. - Default hosts: the ramp is unchanged. `content/docs/automation/webhooks.mdx` is updated in this PR (see Docs above); its stale `1 + partitionCount` figure is corrected here. - No `packages/spec` key was added. Seat 4's #29918 adds no host signal, so nothing is shared with it. ## Verification <details><summary>Local gates and suites (head sha in the os-dev-report)</summary> Head `def0361dc` (after merging `origin/main` at `dfafae813`; no incoming commit touches `packages/core`, `service-messaging` or `service-queue`). - `pnpm --filter @objectstack/core --filter @objectstack/service-messaging --filter @objectstack/service-queue run test`: core 134 files / 2989 tests, service-messaging 66 / 812, service-queue 5 / 82, all passed (lock VERDICT command-exit 0). - `run typecheck` on the same three: exit 0 (core includes `check:test-typecheck`). - `pnpm lint`: exit 0. - `node scripts/pm/dispatch-gates.mjs --commands`: 95 derived families. The 65 code families ran at `def0361d` (65 × exit 0; the docs commit touches no code), and the 30 families the docs page adds ran at `a1a9fbd3` (30 × exit 0). `--ran` reconciliation: 95 derived, 95 run, 0 NOT-MEASURED. Readings: project file `server-seat5/gates-29919-ran-r2.txt`. - Before the merge the same list ran on `2d31b756` with three exit 3 (prerequisites: `check-plugin-teardown-shape --self-test` on a shallow clone, `check:dual-build-cjs-loads` and `check:i18n` without the built CLI closure). On this head all three ran after `git fetch` of the fixture commit and the gate's own named build: exit 0. </details> --- _Generated by [Claude Code](https://claude.ai/code/session_01Utp6CY2ugcVZ5KKbTbWpXw)_ <!-- signed: domain:server#5 · os-bill · session_01Utp6CYVV6fcAQrNTDQRvPb --> --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent bb86ff9 commit 771b2db

21 files changed

Lines changed: 1026 additions & 55 deletions
Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
---
2+
'@objectstack/core': patch
3+
'@objectstack/service-messaging': patch
4+
'@objectstack/service-queue': patch
5+
---
6+
7+
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
8+
9+
Clause-②: yes (widening)
10+
Metadata-protocol: none
11+
12+
- **`@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.
13+
- **`@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).
14+
- **`@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.
15+
- **`@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.
16+
17+
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‎

Lines changed: 34 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -452,16 +452,39 @@ same partition — useful for in-order delivery and connection reuse. On a
452452
single-node runtime the lock is an always-grant stub.
453453

454454
Each tick opens with **one** visibility-timeout reap for the whole table —
455-
`in_flight` rows claimed longer than `claimTtlMs` ago return to `pending` — and
456-
then claims partition by partition, so a tick over an empty outbox costs
457-
`1 + partitionCount` statements. The loop **backs off while idle**: every tick
458-
that claims nothing doubles the delay to the next, from `intervalMs` (default
459-
500 ms) up to `maxIdleIntervalMs` (default 30 s; `dispatchMaxIdleIntervalMs` on
460-
`MessagingServicePlugin`), and a tick that claims work snaps it back.
461-
`messaging.enqueueHttp()` and `messaging.redeliverHttp()` wake the dispatcher
462-
running in the same process, so a delivery written there goes out at once. Work
463-
nobody announces — a retry coming due, a row another process wrote — is noticed
464-
within one backed-off interval, never more than `maxIdleIntervalMs` late.
455+
`in_flight` rows claimed longer than `claimTtlMs` ago return to `pending` — then
456+
asks once, across every partition, whether any `pending` row is due. Only when
457+
one is does it claim partition by partition. A tick over an empty outbox
458+
therefore costs **two** statements, both reads, whatever `partitionCount` is:
459+
the reap's "anything expired?" read (its UPDATE runs only when that finds a
460+
row) and the "anything due?" read.
461+
462+
The loop **backs off while idle**: every tick that claims nothing doubles the
463+
delay to the next, from `intervalMs` (default 500 ms) up to `maxIdleIntervalMs`
464+
(default 30 s; `dispatchMaxIdleIntervalMs` on `MessagingServicePlugin`), and a
465+
tick that claims work snaps it back. `messaging.enqueueHttp()` and
466+
`messaging.redeliverHttp()` wake the dispatcher running in the same process, so
467+
a delivery written there goes out at once.
468+
469+
The loop also **wakes at the next due time**. After the first empty tick of an
470+
idle stretch it reads, once, the earliest instant a row becomes claimable
471+
without anyone writing: the soonest future `next_retry_at`, or the moment the
472+
oldest `in_flight` claim passes `claimTtlMs`. It never sleeps past that
473+
instant, so a retry is attempted when it comes due and a crashed node's claim is
474+
recovered when its TTL passes, whatever the ceiling. That costs two more reads
475+
at the start of each idle stretch, not one per tick. The backoff is left with
476+
one job: a row **another process** wrote, which is noticed within one
477+
backed-off interval, never more than `maxIdleIntervalMs` late.
478+
479+
On a host where no other process writes these tables, that job is empty. Set
480+
`dispatchSoleWriter: true` on `MessagingServicePlugin` there, for example on a
481+
hosted tenant kernel with one kernel per tenant database. An empty tick then
482+
sleeps straight to the next due time or the ceiling, instead of re-running the
483+
ramp from `intervalMs` after every boot and every burst, and a fresh kernel's
484+
dispatchers tick once in their first minute. The job queue takes the same flag
485+
as `DbQueueAdapter`'s `soleWriter`, set through `QueueServicePlugin`'s
486+
`db.soleWriter`. Leave both off when a second node or a worker process writes
487+
the same database. Default: off.
465488

466489
Within a held partition, the lock-holder claims a batch with an **atomic
467490
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.
697720
|------------------------------------------|-------------------------------------------------------------|
698721
| 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). |
699722
| Subscriber node crashes after persist | Row exists in `sys_http_delivery`, another node picks it up. |
700-
| 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). |
723+
| 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. |
701724
| 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. |
702725
| Receiver returns 5xx | Retry per backoff schedule until the fixed 8-attempt budget is exhausted (§4.5). |
703726
| Receiver returns 4xx | Treated as terminal — no retry, status `dead` immediately. Exception: 408 / 429 are retried. |
Lines changed: 294 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,294 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
/**
4+
* #29919 — `DispatchLoop` wakes at the next due time, and a sole writer's loop
5+
* does not re-run the idle ramp after a cold start.
6+
*
7+
* Before #29919 every `start()` re-ran the backoff's ramp from `intervalMs`
8+
* against stores that were empty: with the dispatchers' 500 ms base, six ticks
9+
* 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
10+
* notice of a row another process wrote — cannot happen on a sole writer. And
11+
* a deferred row was noticed only at the next backed-off tick, up to the
12+
* ceiling late.
13+
*
14+
* Every leg runs on vitest's fake timers, so "when did a tick start" is an
15+
* exact reading rather than a sleep-and-hope.
16+
*/
17+
18+
import { describe, it, expect, afterEach, vi } from 'vitest';
19+
import { DispatchLoop, type DispatchLoopOptions } from './dispatch-loop.js';
20+
21+
const BASE = 500;
22+
const MINUTE = 60_000;
23+
const FIVE_MINUTES = 5 * MINUTE;
24+
25+
afterEach(() => { vi.useRealTimers(); });
26+
27+
/** Fake timers with the clock at 0, so a tick's start reads as ms since `start()`. */
28+
function useClockAtZero(): void {
29+
vi.useFakeTimers();
30+
vi.setSystemTime(0);
31+
}
32+
33+
/**
34+
* A store of deferred rows: `runTick` claims every row due by the fake clock,
35+
* `nextDueAt` answers the earliest row still in the future. Both are recorded.
36+
*/
37+
function harness(options: Partial<DispatchLoopOptions> & { dueRows?: number[] } = {}) {
38+
const dueRows = [...(options.dueRows ?? [])];
39+
const tickStarts: number[] = [];
40+
const claimedAt: number[] = [];
41+
let dueReads = 0;
42+
const errors: unknown[] = [];
43+
const loop = new DispatchLoop({
44+
intervalMs: BASE,
45+
runTick: async () => {
46+
const now = Date.now();
47+
tickStarts.push(now);
48+
let claimed = 0;
49+
for (let i = dueRows.length - 1; i >= 0; i--) {
50+
if (dueRows[i]! <= now) {
51+
dueRows.splice(i, 1);
52+
claimedAt.push(now);
53+
claimed++;
54+
}
55+
}
56+
return claimed;
57+
},
58+
nextDueAt: async () => {
59+
dueReads++;
60+
const future = dueRows.filter((t) => t > Date.now());
61+
return future.length > 0 ? Math.min(...future) : null;
62+
},
63+
onTickError: (err) => { errors.push(err); },
64+
...options,
65+
});
66+
return {
67+
loop,
68+
tickStarts,
69+
claimedAt,
70+
errors,
71+
dueRows,
72+
get dueReads() { return dueReads; },
73+
};
74+
}
75+
76+
describe('#29919 DispatchLoop — a sole writer does not ramp from a cold start', () => {
77+
it('a fresh loop on an empty store ticks ONCE in its first minute', async () => {
78+
useClockAtZero();
79+
const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
80+
h.loop.start();
81+
await vi.advanceTimersByTimeAsync(MINUTE);
82+
await h.loop.stop();
83+
84+
expect(h.tickStarts).toEqual([0]);
85+
// The one tick's one due read — nothing deferred, so the ceiling.
86+
expect(h.dueReads).toBe(1);
87+
});
88+
89+
it('CONTROL: the same loop without soleWriter still runs the ramp it always had', async () => {
90+
useClockAtZero();
91+
const h = harness({ maxIdleIntervalMs: FIVE_MINUTES });
92+
h.loop.start();
93+
await vi.advanceTimersByTimeAsync(MINUTE);
94+
await h.loop.stop();
95+
96+
// 1, 2, 4, 8, 16 s apart — six ticks. The harness can see a ramp, so the
97+
// single tick above is a reading about soleWriter, not a dead clock.
98+
expect(h.tickStarts).toEqual([0, 1_000, 3_000, 7_000, 15_000, 31_000]);
99+
});
100+
101+
it('without a nextDueAt reader, soleWriter alone sleeps straight to the ceiling', async () => {
102+
useClockAtZero();
103+
const h = harness({ soleWriter: true, maxIdleIntervalMs: 30_000, nextDueAt: undefined });
104+
h.loop.start();
105+
await vi.advanceTimersByTimeAsync(2 * MINUTE);
106+
await h.loop.stop();
107+
108+
expect(h.tickStarts).toEqual([0, 30_000, 60_000, 90_000, 120_000]);
109+
});
110+
111+
it('steady state costs no extra read: an idle hour at the ceiling reads the due time once', async () => {
112+
useClockAtZero();
113+
const h = harness({ soleWriter: true, maxIdleIntervalMs: 30_000 });
114+
h.loop.start();
115+
await vi.advanceTimersByTimeAsync(60 * MINUTE);
116+
await h.loop.stop();
117+
118+
// 3600 s / 30 s, plus the tick start() runs.
119+
expect(h.tickStarts).toHaveLength(121);
120+
expect(h.dueReads).toBe(1);
121+
});
122+
});
123+
124+
describe('#29919 DispatchLoop — wakes at the next due time', () => {
125+
it('a row due at T+20 s is claimed at T+20 s, under a five-minute ceiling', async () => {
126+
useClockAtZero();
127+
const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES, dueRows: [20_000] });
128+
h.loop.start();
129+
await vi.advanceTimersByTimeAsync(MINUTE);
130+
await h.loop.stop();
131+
132+
expect(h.claimedAt).toEqual([20_000]);
133+
// The cold-start tick, the due tick, and the tick after a claim
134+
// (`intervalMs` later) — then nothing is deferred and the ceiling wins.
135+
expect(h.tickStarts).toEqual([0, 20_000, 20_500]);
136+
});
137+
138+
it('the ramp keeps its shape, cut short only by the due time (no soleWriter)', async () => {
139+
useClockAtZero();
140+
const h = harness({ maxIdleIntervalMs: FIVE_MINUTES, dueRows: [2_000] });
141+
h.loop.start();
142+
await vi.advanceTimersByTimeAsync(4_000);
143+
await h.loop.stop();
144+
145+
expect(h.claimedAt).toEqual([2_000]);
146+
expect(h.tickStarts).toEqual([0, 1_000, 2_000, 2_500, 3_500]);
147+
});
148+
149+
it('consecutive deferred rows are each claimed on time — the read is re-asked after each due time', async () => {
150+
useClockAtZero();
151+
const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES, dueRows: [10_000, 40_000] });
152+
h.loop.start();
153+
await vi.advanceTimersByTimeAsync(MINUTE);
154+
await h.loop.stop();
155+
156+
expect(h.claimedAt).toEqual([10_000, 40_000]);
157+
});
158+
159+
it('wakeAt(): a deferred row written later in the streak is claimed on time with no read', async () => {
160+
useClockAtZero();
161+
const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
162+
h.loop.start();
163+
await vi.advanceTimersByTimeAsync(10_000);
164+
const readsBefore = h.dueReads;
165+
const ticksBefore = h.tickStarts.length;
166+
167+
h.dueRows.push(35_000);
168+
h.loop.wakeAt(35_000);
169+
// No tick now: the hint only moves the timer.
170+
await vi.advanceTimersByTimeAsync(1);
171+
expect(h.tickStarts).toHaveLength(ticksBefore);
172+
173+
await vi.advanceTimersByTimeAsync(MINUTE);
174+
await h.loop.stop();
175+
expect(h.claimedAt).toEqual([35_000]);
176+
// The hint cost no read; the read after the claim is the next streak's.
177+
expect(h.dueReads).toBe(readsBefore + 1);
178+
});
179+
180+
it('wakeAt() keeps a timer that is already due sooner', async () => {
181+
useClockAtZero();
182+
const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES, dueRows: [20_000] });
183+
h.loop.start();
184+
await vi.advanceTimersByTimeAsync(1);
185+
h.dueRows.push(50_000);
186+
h.loop.wakeAt(50_000);
187+
await vi.advanceTimersByTimeAsync(MINUTE);
188+
await h.loop.stop();
189+
190+
expect(h.claimedAt).toEqual([20_000, 50_000]);
191+
});
192+
});
193+
194+
describe('#29919 DispatchLoop — wake() still delivers at once', () => {
195+
it('a wake on a sole writer sleeping towards its ceiling ticks immediately', async () => {
196+
useClockAtZero();
197+
const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
198+
h.loop.start();
199+
await vi.advanceTimersByTimeAsync(MINUTE);
200+
201+
h.dueRows.push(MINUTE);
202+
h.loop.wake();
203+
await vi.advanceTimersByTimeAsync(0);
204+
await h.loop.stop();
205+
206+
expect(h.claimedAt).toEqual([MINUTE]);
207+
});
208+
});
209+
210+
describe('#29919 DispatchLoop — nothing it reads can make it spin', () => {
211+
it('a rejected tick counts as idle: a failing store is ticked once per ceiling, not hammered', async () => {
212+
useClockAtZero();
213+
const h = harness({
214+
soleWriter: true,
215+
maxIdleIntervalMs: 30_000,
216+
runTick: async () => { h.tickStarts.push(Date.now()); throw new Error('store down'); },
217+
});
218+
h.loop.start();
219+
await vi.advanceTimersByTimeAsync(MINUTE);
220+
await h.loop.stop();
221+
222+
expect(h.tickStarts).toEqual([0, 30_000, 60_000]);
223+
expect(h.errors).toHaveLength(3);
224+
// A rejected tick asks for no due time.
225+
expect(h.dueReads).toBe(0);
226+
});
227+
228+
it('a rejected tick after a due time was learned still sleeps until that due time, not 0 ms', async () => {
229+
useClockAtZero();
230+
const tickStarts: number[] = [];
231+
const loop = new DispatchLoop({
232+
intervalMs: BASE,
233+
maxIdleIntervalMs: FIVE_MINUTES,
234+
soleWriter: true,
235+
// The cold-start tick finds nothing; every later tick rejects.
236+
runTick: async () => {
237+
tickStarts.push(Date.now());
238+
if (tickStarts.length > 1) throw new Error('store down');
239+
return 0;
240+
},
241+
nextDueAt: async () => 20_000,
242+
onTickError: () => {},
243+
});
244+
loop.start();
245+
await vi.advanceTimersByTimeAsync(MINUTE);
246+
await loop.stop();
247+
248+
// The due tick at 20 s rejects; the next is a ceiling away.
249+
expect(tickStarts).toEqual([0, 20_000]);
250+
});
251+
252+
it('a due time at or before the tick that found nothing is not chased (visible, but not this loop\'s to claim)', async () => {
253+
useClockAtZero();
254+
const h = harness({
255+
soleWriter: true,
256+
maxIdleIntervalMs: 30_000,
257+
// Another node holds the row's partition: it is due, and never claimable here.
258+
runTick: async () => { h.tickStarts.push(Date.now()); return 0; },
259+
nextDueAt: async () => 0,
260+
});
261+
h.loop.start();
262+
await vi.advanceTimersByTimeAsync(MINUTE);
263+
await h.loop.stop();
264+
265+
expect(h.tickStarts).toEqual([0, 30_000, 60_000]);
266+
});
267+
268+
it('a rejected due read is reported, and the loop sleeps by its backoff alone', async () => {
269+
useClockAtZero();
270+
const h = harness({
271+
soleWriter: true,
272+
maxIdleIntervalMs: 30_000,
273+
nextDueAt: async () => { throw new Error('read failed'); },
274+
});
275+
h.loop.start();
276+
await vi.advanceTimersByTimeAsync(MINUTE);
277+
await h.loop.stop();
278+
279+
expect(h.tickStarts).toEqual([0, 30_000, 60_000]);
280+
expect(h.errors).toHaveLength(3);
281+
});
282+
283+
it('stop() is final: a wakeAt() after it is a no-op', async () => {
284+
useClockAtZero();
285+
const h = harness({ soleWriter: true, maxIdleIntervalMs: FIVE_MINUTES });
286+
h.loop.start();
287+
await vi.advanceTimersByTimeAsync(1);
288+
await h.loop.stop();
289+
h.loop.wakeAt(10_000);
290+
await vi.advanceTimersByTimeAsync(MINUTE);
291+
292+
expect(h.tickStarts).toEqual([0]);
293+
});
294+
});

0 commit comments

Comments
 (0)