Skip to content

Commit a727ba2

Browse files
perf(service-messaging): an idle dispatcher tick reads the database twice, not 18 and 10 times (#28403)
<!-- claude-projects-attribution: {"github_login":"os-zhuang"} --> _Requested by **Sam** · [project thread](https://claude.ai/code/project/chan_01LjZwNRFiG2fkp8SNf6rHNx?thread=cmsg_01LjZwNRFiG2fkp8SNf6rHNxECLC1evcjsCNG7ex4pEc8Z)_ Fixes #28383 Clause-②: no Metadata-protocol: none ## 维护者速读 - 改了什么:通知调度器和 HTTP 调度器在 outbox 空闲时,每次轮询只读两次库(回收过期 claim 前先读一条、再用一条跨所有分区的"有没有到期行"读决定要不要逐分区探测),不再是 18 条和 10 条。 - 为什么改:cloud 实测一个空闲租户内核每分钟读库 56 次,全部是空转;这些读计入 Turso 的行读配额。 - 风险与回滚:有活时仍走原来的逐分区 claim/digest 路径,只多一次读;回收 UPDATE 的谓词不变,只在读到过期行时才执行。outbox 接口新增的 `hasDueWork` 是可选方法,自定义 outbox 不实现就保持旧行为。回滚即 revert 本 PR。 - 先看:`packages/services/service-messaging/src/dispatcher.ts`(`runTick` 里的门)、`src/sql-outbox.ts`(`reapExpired` 与 `hasDueWork`)、`src/dispatcher-idle-cost.integration.test.ts`(语句数钉子)。 - 两分钟验证:`pnpm --filter @objectstack/service-messaging exec vitest run --maxWorkers=2 src/dispatcher-idle-cost.integration.test.ts src/http-dispatcher-idle-cost.integration.test.ts`,预期 13 个用例全绿,空闲 tick 钉在 2 条 SQL。 - 需要你做的:无,席位落地。 ## What changes An idle tick of each delivery dispatcher now costs **two SQL statements**, whatever `partitionCount` is. Measured at the client (knex `query` events) on a harness with one global `afterUpdate` hook, the kernel shape cloud measured: | Dispatcher | main `ac09e7b85` | this PR | |---|---|---| | `NotificationDispatcher` | 18 | **2** | | `HttpDispatcher` | 10 | **2** | The card's bar is ≤ 6 per dispatcher. - **Candidate 1: the reap writes only when there is something to reap.** `reapExpired` in `sql-outbox.ts` and `sql-http-outbox.ts` first reads one expired `in_flight` id (`fields: ['id'], limit: 1`). If it finds none, it returns. If it finds one, it runs the predicate `UPDATE`, which is unchanged, so every claim expired by then still goes back to `pending`, not only the one the read saw. On a kernel the old unconditional `UPDATE` was two statements (the engine reads the matched rows for the per-row `afterUpdate` dispatch, then writes); it is now one read. - **Candidates 2 and 3: one due-work read gates every partition probe, digest probes included.** `INotificationOutbox` and `IHttpOutbox` gain an optional `hasDueWork({ now })`. It asks, in one `SELECT id … LIMIT 1` across every partition and organization, whether any `pending` row is due. The predicate is the claim's own with the partition term dropped, and for notifications with the `digest_key` arm dropped, so it covers immediate and batched rows alike. `runTick` asks it after the reap. If nothing is due, the tick ends there: no partition lock is taken and no claim or digest probe runs. If something is due, the tick runs the full per-partition pass exactly as before, for one extra read. Both built-in stores (SQL and memory) implement it. A store without it keeps probing every partition. `DueWorkOptions` and `HttpDueWorkOptions` are exported. - `partitionCount` and the stored `partition_key` are unchanged. ## Why An idle hosted tenant kernel read its database 56 times a minute with nothing queued: 2 ticks × (18 + 10) at the framework's 30 s idle ceiling, plus a 140-statement backoff ramp in the first minute. Those reads count against Turso's rows-read quota, and cloud#2177 once had a staging account suspended for reads. With this PR an idle minute costs 8 statements at the 30 s ceiling. ## Tests - `dispatcher-idle-cost.integration.test.ts` and `http-dispatcher-idle-cost.integration.test.ts` now count SQL statements on the delivery table at the client as well as engine calls, with a global no-op `afterUpdate` hook registered so a predicate `UPDATE` is the read-then-write pair it is on a kernel. On this harness, main measures 18 and 10 statements per tick, the same figures cloud measured. - The idle pin: `{ update: 0, find: 2 × TICKS }` engine calls, and exactly `2 × TICKS` SQL statements, all `SELECT`. This replaces the old `≤ TICKS × (1 + 2 × partitionCount)` and `≤ TICKS × (1 + partitionCount)` ceilings. - New: a row deferred into the future (notification `notBefore`; an HTTP retry acked with `nextRetryAt`) keeps the tick at two reads and goes out on the first tick after it comes due. - New (notification): a due digest-only outbox is found by the due-work read and collapsed on that tick. - The crash-recovery legs now also assert the reap's `UPDATE` ran exactly once when a claim had expired. This guards against a reap that never writes. - The legacy-outbox legs (no `reap`, and so no `hasDueWork`) are unchanged and still see `2 × partitionCount` and `partitionCount` claims a tick. - Pins that assumed the old shape, reconciled: - `sql-outbox-audit-columns.test.ts`: its fake engine's scripted `find` queue now opens with a reap read that finds a stale row, so the reap `UPDATE` payload it checks for `updated_at` is still produced. - `system-context.pin.test.ts`: the population counts include the reap reads and one `hasDueWork` call, so the new reads are also held to the explicit system opt-in. - `http-dispatcher-idle-backoff.test.ts`: it timestamped a tick at `claim()`, which an idle tick no longer calls. It now timestamps at `reap()`, which every tick calls once first, as the notification side's file already did. ## Acceptance notes - **Deviation from the dispatch's mechanism assumption (candidate 2).** The dispatch described "one probe across the partitions this process holds". The due-work read here spans all partitions and takes no lock, and it gates the per-partition pass instead of replacing it. Lock state is unknown until each `acquire`, and on the common single-node or memory lock every partition is held anyway. A read that moves nothing needs no lock, as with the reap. Keeping the per-partition claims under their locks leaves the busy path, the batch size per partition and the digest-under-lock rule exactly as they were. On a multi-node cluster, a node whose partitions are all held elsewhere pays the same lock attempts as before whenever something is due anywhere, and two reads when nothing is. - **Candidate 3 is folded into the one read**, which is the card's first option ("fold into the claim probe"). A busy tick still runs both probes in every held partition. Narrowing the busy path is not what this card measured. - **Out of scope, as dispatched:** - Candidate 4: wake at the next due time instead of a blind ceiling (`packages/core/src/dispatch-loop.ts`). - Candidate 5: a persisted watermark for the objectql lifecycle sweep's first run 60 s after every boot. - The backoff ramp's first-minute price now falls with the per-tick price (5 ramp ticks × 4 statements instead of × 28). cloud re-measures under cloud#3413. - The new reads add `ORDER BY id` (the engine's default when a `limit` is set) over a `limit 1`. That is one row from the primary key ordering, not a scan cost this card measured. ## Verification All readings at head `95f4da30a` unless stated. <details><summary>Statement counts (before / after)</summary> A throwaway measurement on main `ac09e7b85` (not committed), with and without a global `afterUpdate` hook, one tick each over an empty outbox: - no hook: notification 17, HTTP 9; with the hook: notification 18, HTTP 10. The hook shape is `select * … where status = ? and claimed_at < ?` then the `update`, followed by one or two `select id … partition_key = ?` per partition. - After the change, both shapes: notification 2, HTTP 2. Each is `select id … where status = ? and claimed_at < ? … limit ?`, then `select id … where status = ? and ((next_… is null) or (next_… <= ?)) … limit ?`. </details> <details><summary>Package tests and typecheck</summary> - `pnpm --filter @objectstack/service-messaging test`: Test Files 62 passed (62), Tests 759 passed (759). Lock VERDICT command-exit 0 at `18e1c546b`. The later commit only adds the changeset and one `as HttpClaimCredential` cast in a test. - `pnpm --filter @objectstack/service-messaging typecheck`: exit 0 at `95f4da30a`. At `18e1c546b` the first run caught one TS2345 in the new HTTP retry leg (an `HttpDelivery` passed as the claim credential), fixed with the cast the dispatcher itself uses. - The two idle-cost files at `95f4da30a`: 13 passed (13). </details> <details><summary>Ablation (scripts/ablation-replace.mjs, restored, git diff HEAD empty, marker grep 0)</summary> Tests import `src/` relatively, so no dist rebuild sits between mutation and run. - A: the `hasDueWork` gate in `NotificationDispatcher.runTick` replaced by a marker. Notification idle-cost file: 2 failed / 5 passed (the idle pin: `find: 170` vs `20`, 17 statements vs 2; the deferred-row pin). - B: `SqlHttpOutbox.reapExpired`'s early return replaced by a marker. HTTP idle-cost file: 2 failed / 4 passed (the idle pin: `update: 10` vs `0`; the retry pin: 4 statements vs 2). </details> <details><summary>Dispatch gates and lint</summary> - `node scripts/pm/dispatch-gates.mjs --commands --repo objectstack-ai/objectstack`: 62 families derived from 15 paths vs merge base `ac09e7b85` (473 changed lines, under the 3000 threshold). Every command was run with stdin closed, and each exit code was captured to disk before any pipe. - `--ran` reconciliation: 62 derived, 62 run, 0 NOT-MEASURED, 0 UNRUN; it exits 0. - Every derived command exited 0. Three first exited 3 (PREREQUISITE NOT MET) and were re-run after their prerequisite: - `pnpm check:dual-build-cjs-loads` and `pnpm check:i18n` after a full `pnpm build` (lock VERDICT command-exit 0). - `node scripts/check-plugin-teardown-shape.mjs --self-test` after `git fetch --unshallow origin main` (48 cases pass). - `pnpm lint` (repo-wide `eslint . --no-inline-config`): lock VERDICT command-exit 0. </details> ## Human Minimum - [x] 1. Schemas start as Zod; `packages/spec` holds no runtime logic. (Not touched.) - [x] 2. Every name I added uses the casing for its kind. - [x] 3. No formatter ran; `pnpm lint` passes. (See Verification.) - [x] 4. One issue, one branch off an up-to-date `main`, one pull request. - [x] 5. A change that publishes carries a changeset, and both declaration lines are filled in here and in the changeset. - [x] 6. `packages/spec` touched: n/a, not touched. - [x] 7. No hand edits to `content/docs/references/`, `packages/*/CHANGELOG.md` or `content/docs/releases/`. - [x] 8. Title, body and comments are in English. - [x] The *Before Submitting* list in CONTRIBUTING.md is done. --- _Generated by [Claude Code](https://claude.ai/code/session_01QUs3Bvkfh86Jgvi7gewhBD)_ --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent a8ec04d commit a727ba2

15 files changed

Lines changed: 407 additions & 68 deletions
Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
---
2+
'@objectstack/service-messaging': patch
3+
---
4+
5+
perf(service-messaging): an idle notification or HTTP dispatcher tick reads the database twice, instead of 18 and 10 times
6+
7+
Clause-②: yes (widening)
8+
Metadata-protocol: none
9+
10+
A kernel that mounts messaging runs two delivery dispatchers, and each one ticked against an empty outbox at a fixed price: one claim probe per partition (and, for notifications, one digest probe per partition), plus a visibility-timeout reap that issued its predicate `UPDATE` whether or not any claim had expired. With the default 8 partitions that was 18 statements per notification tick and 10 per HTTP tick, measured at the client on a kernel. A kernel's global `afterUpdate` hook makes the engine read the matched rows before it writes, so the reap cost two statements. At the framework's 30-second idle ceiling an idle tenant kernel read its database 56 times a minute.
11+
12+
An idle tick is now two reads per dispatcher, whatever `partitionCount` is:
13+
14+
- **The reap reads before it writes.** It reads one expired `in_flight` id. Only a hit runs the `UPDATE`, whose predicate is unchanged, so it still returns every claim expired by then to `pending`.
15+
- **One due-work read gates the partition pass.** Before taking any partition lock, the dispatcher asks the outbox once, across every partition and organization, whether any `pending` row is due. For notifications this covers immediate and batched (digest) rows. If none is due, the tick ends: no lock is taken and no claim or digest probe runs. If one is due, the tick runs the full per-partition pass exactly as before, at the price of one extra read.
16+
17+
`INotificationOutbox` and `IHttpOutbox` gain an optional `hasDueWork({ now })`, with `DueWorkOptions` and `HttpDueWorkOptions` exported for it. Both built-in stores, SQL and memory, implement it. A custom outbox that does not implement it keeps the per-partition probes it had. `partitionCount` and the stored `partition_key` are unchanged.
18+
19+
The public surface widens, and that is all it does: the exported `INotificationOutbox` and `IHttpOutbox` interfaces gain one optional member each (`hasDueWork`), and the package exports two new option types (`DueWorkOptions`, `HttpDueWorkOptions`). No `packages/spec` path changes, and an existing outbox implementation compiles and behaves as before.

‎packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts‎

Lines changed: 94 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,34 @@
11
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
22

33
/**
4-
* #17610 — what an IDLE `NotificationDispatcher` tick costs the store, and that
5-
* cutting it keeps every delivery guarantee.
4+
* #17610, #28383 — what an IDLE `NotificationDispatcher` tick costs the store,
5+
* and that cutting it keeps every delivery guarantee.
66
*
77
* ## The measurement this pins
88
*
99
* Before #17610 one tick over an EMPTY outbox issued 32 statements with the
1010
* default 8 partitions: `claim()` and `claimDigest()` each opened with the
1111
* environment-wide visibility-timeout reap, in every partition — 16 identical
1212
* UPDATEs — plus one candidate SELECT each (16). On remote Turso every
13-
* statement is an HTTP round trip, and the loop never stopped. The reap now
14-
* runs once per tick, so an idle tick is `1 + 2 × partitionCount` = 17.
13+
* statement is an HTTP round trip, and the loop never stopped. #17610 ran the
14+
* reap once per tick: `1 + 2 × partitionCount` engine calls, which on a kernel
15+
* is 18 statements — the reap's predicate UPDATE is two there, because the
16+
* engine reads the matched rows for the per-row `afterUpdate` dispatch before
17+
* it writes, and it wrote with nothing to reap.
18+
*
19+
* #28383 makes an idle tick TWO statements, whatever `partitionCount` is: the
20+
* reap reads one expired id before it writes and writes only on a hit, and one
21+
* "anything due?" read across every partition gates the per-partition claim
22+
* and digest probes.
23+
*
24+
* ## Counted where cloud counted it
25+
*
26+
* Two tallies, both pinned: engine calls on the delivery table, and SQL
27+
* statements on it, read off the driver's knex `query` events — the client
28+
* boundary cloud measured 18 at (#28383). The harness registers one global
29+
* no-op `afterUpdate` hook, as a kernel's audit plugin does, so a predicate
30+
* UPDATE is the read-then-write pair it is in production; on this harness the
31+
* pre-#28383 code measured 18 statements a tick, cloud's figure.
1532
*
1633
* ## Why a real engine
1734
*
@@ -31,9 +48,16 @@
3148
* - **"Reap once" could be bought by recovering less.** The recovery leg drives
3249
* a real expired claim through ONE tick; its negative leg proves a claim
3350
* that has not expired is left alone, so the TTL is still a floor.
51+
* - **"Write only on a hit" could be bought by never writing.** The idle pin
52+
* asserts ZERO reap UPDATEs; the recovery leg asserts the UPDATE does run
53+
* when a claim has expired.
54+
* - **One "anything due?" read could be bought by a predicate that misses a
55+
* kind of row.** A due digest-only outbox, and a deferred row that comes
56+
* due, are each driven through it: both are found when due, and a row
57+
* deferred into the future keeps the tick at two reads.
3458
*/
3559

36-
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
60+
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
3761
import { ObjectQL } from '@objectstack/objectql';
3862
import { SqlDriver } from '@objectstack/driver-sql';
3963
import { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js';
@@ -52,6 +76,8 @@ let engine: ObjectQL;
5276
let outbox: SqlNotificationOutbox;
5377
/** Engine calls against the delivery table, by method. */
5478
let calls: { update: number; find: number; other: number };
79+
/** SQL statements the driver sent against the delivery table, in order. */
80+
let statements: string[];
5581
/** `recipient` of every send, in order. */
5682
let sent: string[];
5783

@@ -89,7 +115,16 @@ beforeEach(async () => {
89115
// A real `single` install's Default Organization: the system writes below
90116
// derive it (ADR-0131 D9), as a fresh NOT NULL tenant column requires (D1).
91117
await installDefaultOrganization(engine);
118+
// A kernel's audit plugin registers a global `afterUpdate`; with one
119+
// registered, a predicate UPDATE reads its matched rows before it writes —
120+
// the two-statement reap cloud counted.
121+
engine.registerHook('afterUpdate', async () => {});
92122

123+
statements = [];
124+
(driver as unknown as { knex: { on(e: 'query', cb: (q: { sql?: unknown }) => void): void } }).knex
125+
.on('query', (q) => {
126+
if (typeof q.sql === 'string' && q.sql.includes(DELIVERY_OBJECT)) statements.push(q.sql);
127+
});
93128
calls = { update: 0, find: 0, other: 0 };
94129
sent = [];
95130
// Count on the engine instance the outbox holds, so the tally is of real
@@ -114,17 +149,58 @@ afterEach(async () => {
114149
try { await engine?.destroy(); } catch { /* noop */ }
115150
});
116151

117-
describe('#17610 NotificationDispatcher — idle tick cost', () => {
118-
it('an idle tick is 1 + 2 × partitionCount statements: ONE reap, then a claim and a digest probe per partition', async () => {
152+
describe('#17610 #28383 NotificationDispatcher — idle tick cost', () => {
153+
it('an idle tick is TWO statements: the reap\'s expired-claim read and one due-work read, no UPDATE, no partition probe', async () => {
119154
const d = dispatcher();
120155
for (let i = 0; i < TICKS; i++) await d.tick();
121156

122-
// On an empty outbox the only UPDATE a tick issues is the reap. It is
123-
// environment-wide, so once per tick is all it can use — and exactly
124-
// once, never zero: it is the only crash recovery there is.
125-
expect(calls.update).toBe(TICKS);
126-
// Before #17610: 2 × PARTITIONS reaps + 2 × PARTITIONS probes = 32 a tick.
127-
expect(calls.update + calls.find + calls.other).toBeLessThanOrEqual(TICKS * (1 + 2 * PARTITIONS));
157+
// Nothing expired, so the reap reads and never writes; nothing due, so
158+
// no partition's claim or digest probe runs.
159+
expect(calls).toEqual({ update: 0, find: 2 * TICKS, other: 0 });
160+
// At the client: before #28383, 18 a tick on this harness (cloud's
161+
// figure); the card's bar is 6.
162+
expect(statements).toHaveLength(2 * TICKS);
163+
expect(statements.every((sql) => /^select /i.test(sql))).toBe(true);
164+
});
165+
166+
it('a row deferred into the future keeps the tick at two reads, and goes out on the first tick after it comes due', async () => {
167+
const d = dispatcher();
168+
const enqueuedAt = Date.now();
169+
// A retry's backoff, quiet hours and a digest window all defer a row
170+
// through `next_attempt_at`.
171+
await outbox.enqueue({
172+
notificationId: 'n_later', recipientId: 'u_later', channel: 'inbox', payload: { title: 't' },
173+
notBefore: enqueuedAt + 60_000,
174+
});
175+
statements = [];
176+
await d.tick();
177+
expect(sent).toEqual([]);
178+
expect(statements).toHaveLength(2);
179+
180+
// The store reads `Date.now()` for the due-work read and the claim alike.
181+
const clock = vi.spyOn(Date, 'now').mockReturnValue(enqueuedAt + 60_001);
182+
try {
183+
await d.tick();
184+
} finally {
185+
clock.mockRestore();
186+
}
187+
expect(sent).toEqual(['u_later']);
188+
});
189+
190+
it('a due digest-only outbox is found by the one due-work read and collapsed on that tick', async () => {
191+
const d = dispatcher();
192+
for (const notificationId of ['nd1', 'nd2', 'nd3']) {
193+
await outbox.enqueue({
194+
notificationId, recipientId: 'u_digest', channel: 'inbox', payload: { title: notificationId },
195+
digestKey: 'u_digest|inbox|w1',
196+
});
197+
}
198+
199+
await d.tick();
200+
201+
expect(sent).toEqual(['u_digest']);
202+
const rows = await outbox.list();
203+
expect(rows.every((r) => r.status === 'success')).toBe(true);
128204
});
129205

130206
it('rows enqueued after an idle stretch go out on the very next tick — every partition, digests collapsed', async () => {
@@ -165,13 +241,18 @@ describe('#17610 NotificationDispatcher — idle tick cost', () => {
165241
});
166242
expect(abandoned?.status).toBe('in_flight');
167243

244+
calls = { update: 0, find: 0, other: 0 };
245+
statements = [];
168246
await dispatcher().tick();
169247

170248
// The tick's single reap ran BEFORE its claims, so the same tick took
171249
// the row back and delivered it.
172250
expect(sent).toEqual(['u1']);
173251
const [row] = await outbox.list();
174252
expect(row).toMatchObject({ status: 'success', attempts: 1 });
253+
// The reap's read found the expired claim, so its UPDATE ran — once.
254+
const reapWrites = statements.filter((sql) => /^update .* where `status` = \? and `claimed_at` < \?$/i.test(sql));
255+
expect(reapWrites).toHaveLength(1);
175256
});
176257

177258
it('leaves a claim that has NOT expired alone — the TTL is still a floor', async () => {

‎packages/services/service-messaging/src/dispatcher.ts‎

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -107,11 +107,16 @@ export interface NotificationDispatcherOptions {
107107
* reverts to pending after the claim TTL and is re-sent — the inbox channel's
108108
* receipt write is idempotent-friendly, and downstream channels should be too.
109109
*
110-
* ## What an idle tick costs (#17610)
110+
* ## What an idle tick costs (#17610, #28383)
111111
*
112-
* Against an empty outbox one tick is `1 + 2 × partitionCount` store round
113-
* trips: ONE visibility-timeout reap for the whole environment, then a claim
114-
* probe and a digest probe per partition. The reap used to run inside both
112+
* Against an empty outbox one tick is TWO reads, whatever `partitionCount`
113+
* is: the visibility-timeout reap's "anything expired?" read, and one "anything
114+
* due?" read across every partition ({@link INotificationOutbox.hasDueWork}).
115+
* Neither finds a row, so neither the reap's UPDATE nor any partition's claim
116+
* and digest probe runs. Before #28383 it was `2 + 2 × partitionCount` on a
117+
* kernel — the reap's predicate UPDATE (a matched-row read plus the write,
118+
* unconditionally), then a claim probe and a digest probe per partition: 18
119+
* with the default 8. Before #17610 the reap ran inside both
115120
* claims of every partition — `2 × partitionCount` identical environment-wide
116121
* UPDATEs a tick, 16 of the 32 statements a tick issued with the default 8
117122
* partitions — on a fixed 500 ms interval that never let up, one loop per warm
@@ -217,6 +222,22 @@ export class NotificationDispatcher {
217222
reapedForTick = true;
218223
}
219224

225+
// [#28383] One read across every partition before any partition's two
226+
// probes. Most ticks find nothing — an idle kernel's every tick — and
227+
// they used to pay a claim probe and a digest probe per partition to
228+
// learn it (16 statements with the default 8). `false` ends the tick
229+
// here: no partition lock is taken and no claim or digest probe runs.
230+
// `true` runs the full pass below exactly as before, so a tick that has
231+
// work claims, batches and collapses digests unchanged, for one read
232+
// more. Asked AFTER the reap, so a row the reap just returned to
233+
// `pending` counts as due. The read is environment-wide and takes no
234+
// lock, like the reap: it moves nothing, and the claims it lets through
235+
// still run under their partition locks.
236+
//
237+
// Optional on the outbox contract: a store without it probes every
238+
// partition, as before.
239+
if (outbox.hasDueWork && !(await outbox.hasDueWork({}))) return 0;
240+
220241
const count = this.opts.partitionCount;
221242
const offset = stableNodeOffset(this.opts.nodeId, count);
222243
let claimed = 0;

‎packages/services/service-messaging/src/http-dispatcher-idle-backoff.test.ts‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,9 @@
1313
* `MessagingService.enqueueHttp()` and `redeliverHttp()` — waking it.
1414
*
1515
* Every leg runs on vitest's fake timers, so "when did a tick start" is an
16-
* exact reading rather than a sleep-and-hope. With one partition the dispatcher
17-
* issues exactly one `claim()` per tick, so a tick is timestamped there.
16+
* exact reading rather than a sleep-and-hope. Every tick opens with exactly one
17+
* `reap()` — an idle one claims nothing at all since #28383 — so a tick is
18+
* timestamped there, as the notification side's file does.
1819
*/
1920

2021
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
@@ -23,7 +24,7 @@ import { HttpDispatcher } from './http-dispatcher.js';
2324
import { DEFAULT_MAX_IDLE_INTERVAL_MS } from './dispatcher.js';
2425
import { MessagingService } from './messaging-service.js';
2526
import type { FetchImpl } from './http-sender.js';
26-
import type { EnqueueHttpInput, HttpAckResult, HttpClaimCredential, HttpClaimOptions, HttpDelivery } from './http-outbox.js';
27+
import type { EnqueueHttpInput, HttpAckResult, HttpClaimCredential, HttpReapOptions } from './http-outbox.js';
2728

2829
const BASE = 500;
2930
const CAP = 30_000;
@@ -35,9 +36,9 @@ const silentLogger = { info() {}, warn() {}, error() {} };
3536
class TickRecordingOutbox extends MemoryHttpOutbox {
3637
readonly tickStarts: number[] = [];
3738
readonly retriesDueAt: number[] = [];
38-
override async claim(opts: HttpClaimOptions): Promise<HttpDelivery[]> {
39+
override async reap(opts: HttpReapOptions): Promise<void> {
3940
this.tickStarts.push(Date.now());
40-
return super.claim(opts);
41+
return super.reap(opts);
4142
}
4243
override async ack(id: string, result: HttpAckResult, claimed?: HttpClaimCredential): Promise<void> {
4344
if (!result.success && result.nextRetryAt !== undefined) this.retriesDueAt.push(result.nextRetryAt);

0 commit comments

Comments
 (0)