Skip to content

Commit 920f887

Browse files
fix(service-queue,core): the idle job-queue poll backs off, on the one shared DispatchLoop (#18134)
Fixes #17612 Scope item 3 (空闲轮询) — the last open item on this card. Items 1, 2 and 4 landed in #18105 and are re-verified untouched on `origin/main` here. Clause-②: yes ## What the card asked, and what was measured first The card says item 3 should **reuse** #17610's backoff rather than write a second one, and the dispatching seat pre-ruled out the two cheap shapes: option D (a second backoff inside `service-queue`) by the card's own text, option B (exporting the loop from `service-messaging` and depending on it from `service-queue`) as an inverted dependency direction. The remaining direction is A — lift the loop somewhere both can depend on. The escape hatch was to stop and report if A turned out disproportionate to the remaining benefit, since items 1/2 already removed the per-tick **cost** and only the statement **count** while idle was left. Both sides were measured before any code moved: | | reads per idle hour, one registered queue | |---|---| | before (flat 1 s `setInterval`) | **3601** | | after (`DispatchLoop`, 1 s base / 30 s ceiling) | **124** | That is a **29x** reduction, linear in the number of registered queues, and on a remote driver every one of those reads is an HTTP round trip. The cost of A turned out to be small: `service-queue` and `service-messaging` **already both depend on `@objectstack/core`**, so the lift adds **zero new dependency edges** — it is one file move, one export, and two import re-points. A is not disproportionate to a 29x cut, so A was built rather than reported as a stop. ## What changed - **`packages/services/service-messaging/src/dispatch-loop.ts` moves to `packages/core/src/dispatch-loop.ts`** and is exported from core's index. It is a timing primitive owned by neither the messaging domain nor the queue domain; core is the package all three consumers already depend on. `service-messaging` publishes only its `index`, which never carried the loop, so **nothing published moved** — its two dispatchers just import from `@objectstack/core` now. - **`DbQueueAdapter.start()` runs that loop** instead of a bare `setInterval`. The loop also subsumes the old `running` re-entrancy flag, which was doing tick coalescing by hand. - **New option `DbQueueAdapterOptions.maxIdleIntervalMs`** (default 30 s). At or below `pollIntervalMs` it disables the backoff and restores the flat poll exactly. - **`publish()` and `replay()` wake the loop**, so the ceiling is never on the latency path for work this process was told about. A *deferred* `publish()` deliberately does not wake it — that tick would claim nothing and would throw the backoff away for free. ## Evidence Tests: `service-queue` 78 passed (70 before + 8 new legs), `service-messaging` 460, `core` 1316, and the downstream consumers `plugin-email` 468 and `plugin-audit` 334 — all on the final head. `typecheck` green for all three edited packages, and it demonstrably reaches `*.test.ts` (it caught a real error there during this work). **The flat-poll leg is a negative control, not decoration.** An upper bound on statements is satisfied by a worker that stopped ticking altogether — the one failure a low number cannot distinguish. So the same clock, engine and counter are run with the backoff disabled and pinned at 3601; that is what makes 124 a reading about the backoff rather than about a dead loop. **Reverse verification** — two legs, each mutated on disk (anchor count 1 before, injected marker 1 / removed anchor 0 after), run, restored with `git checkout HEAD --`, and proved byte-identical by blob hash with `git diff HEAD` empty: | ablation | predicted | observed | |---|---|---| | the backoff itself, in `core/src/dispatch-loop.ts` | RED | RED — 1 failed / 7 passed, `expected 3601 to be 124` | | the wake seam in `DbQueueAdapter.publish()` | RED on the wake leg | RED — 2 failed / 6 passed (wake leg + the drain leg) | The first ablation reads **3601** — independently reproducing the before number from a second direction. It also proves the new vitest alias really reaches core's **source**: no rebuild happened anywhere in that leg, and the verdict still moved. **Clause-② re-derivation, from the delivered diff.** Reachability was taken as the two discriminating reads, not the word `export` and not a bundle grep: - `.d.ts` export list — core's `dist/index.d.ts` (the file `exports['.'].types` names) carries `DispatchLoop`, `type DispatchLoopOptions` and `DEFAULT_MAX_IDLE_INTERVAL_MS` inside its single `export { ... }` statement. - Runtime `in await import(entry)` through the package entry (resolved to `packages/core/dist/index.js` via the exports map), with controls both ways: | symbol | role | result | |---|---|---| | `DispatchLoop` | subject | REACHABLE | | `DEFAULT_MAX_IDLE_INTERVAL_MS` | subject | REACHABLE | | `DispatchLoopOptions` | subject, type-only | not reachable at runtime, by design; present in the `.d.ts` export list | | `ObjectKernel`, `LiteKernel` | positive controls, published before this diff | REACHABLE | | `DispatchLoopTimer` | negative control, plausible sibling | not reachable | | `dispatchLoop` | negative control, wrong case | not reachable | `files` is `["dist", ...]`, so that entry ships. A new symbol is reachable from a shared package's published entry ⇒ **Clause-②** is yes, and the changeset grades `@objectstack/core` **minor** accordingly (never `patch`). **Gate denominator**, reconciled against `scripts/pm/dispatch-gates.mjs --ran` on the final head `95070c921`: **73 derived, 70 run, 3 NOT MEASURED, 0 UNRUN**. The three are `check:dual-build-cjs-loads`, `check:i18n` and `check:type-check-debt`, each of which exited **3** — the code those gates use for "prerequisite not met, nothing was measured", all three wanting a whole-repo `pnpm build`. They are declared NOT MEASURED rather than read as green; CI builds fresh and runs them. Three gates went red against my own diff during this work and were fixed, not routed around: `check:engine-double-contract` (the new double's `update()` now opens with `assertEngineUpdateDispatch`), `check:objectql-double-limit` (the double applies the caller's `limit` by presence, so `limit: 0` is not silently widened to the whole table), and `check:test-source-alias` (the package now aliases `@objectstack/core` to source). `scripts/engine-double-contract.pinned.json` grew by **2 rows** — one `delete`, one `update`, both for the new test file — from 800 entries / 875 pinned-sum on `main` to 802 / 877. ## Acceptance notes - The consumer radius was swept by construction site rather than by edited package: every `new DbQueueAdapter` in the repo was enumerated, and every test site passes `autoStart: false` and drives `pollOnce()` by hand, so the timer change reaches none of them. The one production construction is `QueueServicePlugin`, whose `destroy()` already awaits `stop()`. - `start()` now runs its first tick immediately, where the old `setInterval` waited one interval. That is strictly lower latency and no test depended on the delay. - Observation, not filed: the sibling engine doubles in `db-queue-adapter.test.ts` and `job-queue-retention.test.ts` apply `limit` by truthiness (`if (opts.limit)`), the same shape `check:objectql-double-limit` rejected in the new file. They are inside that gate's existing baseline, so it already knows; a fixture that asked for `limit: 0` would read the whole table. --- _Generated by [Claude Code](https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj)_ --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 93272fc commit 920f887

10 files changed

Lines changed: 415 additions & 20 deletions

File tree

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
---
2+
"@objectstack/core": minor
3+
"@objectstack/service-queue": minor
4+
"@objectstack/service-messaging": patch
5+
---
6+
7+
`DbQueueAdapter` backs off while `sys_job_queue` is idle instead of polling flat at 1 s, and the loop that does it is now published from `@objectstack/core` as `DispatchLoop` (#17612).
8+
9+
A registered-but-idle queue issued **3600 candidate reads an hour, per queue**, whatever was in the table — on a remote driver, 3600 HTTP round trips an hour of pure idle cost. Measured over one simulated idle hour on the engine boundary the adapter really talks to: **3601 reads before, 124 after**, with the flat-poll number re-measured on the same harness as a control so the new one is a reading about the backoff rather than about a loop that stopped ticking.
10+
11+
- **One mechanism, not a third copy.** The idle-backoff loop was written for `NotificationDispatcher` (#17610), shared with `HttpDispatcher` (#17623), and lived unexported inside `@objectstack/service-messaging`. `DbQueueAdapter` was the third polling worker needing it. It moves to `@objectstack/core` — the package all three already depend on — because it is a timing primitive owned by neither the messaging domain nor the queue domain, and having `service-queue` depend on `service-messaging` to reach it would invert the dependency direction. **New export from `@objectstack/core`: `DispatchLoop`, `DispatchLoopOptions`, `DEFAULT_MAX_IDLE_INTERVAL_MS`.**
12+
- **Nothing published moved.** `@objectstack/service-messaging` exports only its `index`, which never carried the loop; its two dispatchers now import it from `@objectstack/core` and its own surface is byte-unchanged.
13+
- **New option `DbQueueAdapterOptions.maxIdleIntervalMs`** (default 30 s). Each tick that claims nothing doubles the delay to the next from `pollIntervalMs` up to this ceiling; anything claimed, and every wake, snaps it straight back. **Setting it at or below `pollIntervalMs` restores the flat poll exactly.**
14+
- ⚠️ **What the backoff costs, and what it does not.** Work published through this adapter now wakes the loop, so a due `publish()` and `replay()` are picked up at the base interval as before — the ceiling is never on their latency path. What it does cost is up to `maxIdleIntervalMs` of extra latency on work this process was never told about: a row another node wrote, a deferred row coming due, a crashed worker's lease expiring. A deferred `publish()` deliberately does **not** wake the loop, since that tick would claim nothing and would throw the backoff away.

packages/services/service-messaging/src/dispatch-loop.ts renamed to packages/core/src/dispatch-loop.ts

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

33
/**
4-
* The timer loop both outbox dispatchers run — `NotificationDispatcher` over
5-
* `sys_notification_delivery` and `HttpDispatcher` over `sys_http_delivery`.
4+
* The timer loop every polling worker in the platform runs —
5+
* `NotificationDispatcher` over `sys_notification_delivery`, `HttpDispatcher`
6+
* over `sys_http_delivery`, and `DbQueueAdapter` over `sys_job_queue`.
67
*
78
* #17610 wrote this loop inside `NotificationDispatcher`. #17623 found
89
* `HttpDispatcher` still on the fixed 500 ms `setInterval` the notification side
9-
* had just left, and moved the loop here, so the two dispatchers run one
10-
* implementation of these rules instead of two copies that can drift:
10+
* had just left, and pulled the loop out into one implementation of these rules
11+
* instead of two copies that can drift. #17612 found the third copy —
12+
* `DbQueueAdapter` on a flat 1 s `setInterval` — and that is why the loop
13+
* lives HERE, in `@objectstack/core`, rather than in either service: it is a
14+
* timing primitive, owned by neither the messaging domain nor the queue
15+
* domain, and a service-to-service dependency between them to share it would
16+
* invert the direction (a queue service depending on a messaging service).
17+
* `@objectstack/core` is the package all three already depend on.
18+
*
19+
* The rules, one implementation:
1120
*
1221
* - **Never two ticks at once.** A tick asked for while one is running (a
1322
* {@link DispatchLoop.wake}, typically) becomes ONE follow-up tick, run the

‎packages/core/src/index.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,12 @@ export * from './dependency-resolver.js';
128128
// Export Phase 3 components - Package lifecycle management
129129
export * from './namespace-resolver.js';
130130

131+
// [#17612] The one polling-worker timer loop — tick coalescing, idle backoff
132+
// and a final stop() — shared by every worker that drains a table
133+
// (service-messaging's two outbox dispatchers, service-queue's DbQueueAdapter).
134+
// It lives here because it is a timing primitive belonging to neither domain.
135+
export * from './dispatch-loop.js';
136+
131137
// Re-export contracts from @objectstack/spec for backward compatibility
132138
export type {
133139
Logger,

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import type { MessagingChannel, MessagingChannelContext, Notification, SendResul
44
import type { AckResult, ClaimedDeliveryRecord, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js';
55
import { classifyDeliveryAttempt } from './backoff.js';
66
import { renderDigest } from './digest-render.js';
7-
import { DispatchLoop } from './dispatch-loop.js';
7+
import { DispatchLoop } from '@objectstack/core';
88

99
/** Minimal channel-registry surface the dispatcher needs (MessagingService satisfies it). */
1010
export interface ChannelRegistry {
@@ -49,7 +49,7 @@ export interface NotificationDispatcherLogger {
4949
* {@link NotificationDispatcherOptions.maxIdleIntervalMs}. It lives with the
5050
* loop both dispatchers run (#17623) and stays exported from here.
5151
*/
52-
export { DEFAULT_MAX_IDLE_INTERVAL_MS } from './dispatch-loop.js';
52+
export { DEFAULT_MAX_IDLE_INTERVAL_MS } from '@objectstack/core';
5353

5454
export interface NotificationDispatcherOptions {
5555
nodeId: string;

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
*
77
* The loop used to tick every `intervalMs` (500 ms) forever, whatever it found —
88
* the shape #17610 removed from `NotificationDispatcher`. Both dispatchers now
9-
* run the one `DispatchLoop` (`dispatch-loop.ts`), whose mechanics are pinned in
9+
* run the one `DispatchLoop` (`@objectstack/core`), whose mechanics are pinned in
1010
* full through the notification side in `dispatcher-idle-backoff.test.ts`. This
1111
* file pins the same behaviour through `HttpDispatcher`, plus what only the HTTP
1212
* side has: a retry schedule the backoff must not starve, and its own ingress —

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
22

33
import type { DispatchCluster, DispatchLockHandle } from './dispatcher.js';
4-
import { DispatchLoop } from './dispatch-loop.js';
4+
import { DispatchLoop } from '@objectstack/core';
55
import { classifyAttempt, sendOnce, type FetchImpl } from './http-sender.js';
66
import type { HttpAckResult, HttpClaimCredential, HttpDelivery, IHttpOutbox } from './http-outbox.js';
77

‎packages/services/service-queue/src/db-queue-adapter.ts‎

Lines changed: 57 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import type {
88
QueueHandler,
99
} from '@objectstack/spec/contracts';
1010
import { SysJobQueue } from '@objectstack/platform-objects/audit';
11+
import { DispatchLoop, DEFAULT_MAX_IDLE_INTERVAL_MS } from '@objectstack/core';
1112
import {
1213
SYSTEM_CTX,
1314
uid,
@@ -135,6 +136,21 @@ export interface LifecycleFloorRegistrar {
135136
export interface DbQueueAdapterOptions {
136137
/** Polling interval for the worker loop (ms, default 1000) */
137138
pollIntervalMs?: number;
139+
/**
140+
* [#17612] Ceiling of the IDLE backoff, in ms (default
141+
* {@link DEFAULT_MAX_IDLE_INTERVAL_MS}, 30 s). Each tick that claims nothing
142+
* doubles the delay to the next from `pollIntervalMs` up to this value; any
143+
* tick that claims work, and every {@link DbQueueAdapter.wake}, snaps it
144+
* straight back to `pollIntervalMs`.
145+
*
146+
* A value at or below `pollIntervalMs` disables the backoff and restores the
147+
* flat poll. What the backoff costs is latency on work this process was not
148+
* told about — a row another node wrote, a deferred row coming due, a
149+
* crashed worker's lease expiring: never noticed more than this long after it
150+
* became claimable. Work published through THIS adapter wakes the loop, so it
151+
* is unaffected.
152+
*/
153+
maxIdleIntervalMs?: number;
138154
/** Max messages claimed per poll tick (default 10) */
139155
batchSize?: number;
140156
/** Lease duration before another worker may reclaim (ms, default 30000) */
@@ -192,8 +208,7 @@ export class DbQueueAdapter implements IQueueService {
192208
private readonly opts: Required<Omit<DbQueueAdapterOptions, 'workerId'>> & { workerId: string };
193209

194210
private readonly handlers = new Map<string, RegisteredHandler[]>();
195-
private timer?: ReturnType<typeof setInterval>;
196-
private running = false;
211+
private loop?: DispatchLoop;
197212

198213
constructor(args: {
199214
engine: JobEngine;
@@ -207,6 +222,7 @@ export class DbQueueAdapter implements IQueueService {
207222
const o = args.options ?? {};
208223
this.opts = {
209224
pollIntervalMs: o.pollIntervalMs ?? 1000,
225+
maxIdleIntervalMs: o.maxIdleIntervalMs ?? DEFAULT_MAX_IDLE_INTERVAL_MS,
210226
batchSize: o.batchSize ?? 10,
211227
leaseMs: o.leaseMs ?? 30_000,
212228
idempotencyWindowMs: o.idempotencyWindowMs ?? 24 * 60 * 60 * 1000,
@@ -353,6 +369,13 @@ export class DbQueueAdapter implements IQueueService {
353369
updated_at: now.toISOString(),
354370
}, { context: SYSTEM_CTX });
355371

372+
// [#17612] Tick now rather than wait out the idle backoff — but only for a
373+
// row that is claimable THIS instant. A deferred row is exactly the case
374+
// the loop's contract already covers (noticed within one backed-off
375+
// interval of coming due), and waking for it would reset the backoff to
376+
// `pollIntervalMs` for a tick guaranteed to claim nothing.
377+
if (Date.parse(scheduledFor) <= now.getTime()) this.wake();
378+
356379
return id;
357380
}
358381

@@ -426,6 +449,9 @@ export class DbQueueAdapter implements IQueueService {
426449
scheduled_for: now.toISOString(),
427450
updated_at: now.toISOString(),
428451
}, { context: SYSTEM_CTX });
452+
// [#17612] `replay` re-arms the row for RIGHT NOW, so it wakes the loop
453+
// unconditionally — there is no deferred case here to discriminate.
454+
this.wake();
429455
}
430456

431457
async purgeFailed(messageId: string): Promise<void> {
@@ -440,19 +466,38 @@ export class DbQueueAdapter implements IQueueService {
440466
// ── Worker lifecycle ─────────────────────────────────────────────
441467

442468
start(): void {
443-
if (this.timer) return;
444-
this.timer = setInterval(() => {
445-
if (this.running) return;
446-
this.running = true;
447-
this.pollOnce()
448-
.catch((err) => { this.logger?.warn?.('DbQueueAdapter: poll tick failed', err); })
449-
.finally(() => { this.running = false; });
450-
}, this.opts.pollIntervalMs);
451-
(this.timer as any)?.unref?.();
469+
if (this.loop) return;
470+
// [#17612] The shared polling loop (`@objectstack/core`), not a bare
471+
// `setInterval`: it coalesces overlapping ticks — the job the `running`
472+
// flag here used to do — AND backs off while the queue is idle. A flat 1 s
473+
// poll cost 3600 candidate SELECTs an hour per registered queue whatever
474+
// was in the table; on a remote driver every one of those is an HTTP round
475+
// trip. The same shape #17610 removed from `NotificationDispatcher` and
476+
// #17623 from `HttpDispatcher`; this was the third copy.
477+
this.loop = new DispatchLoop({
478+
intervalMs: this.opts.pollIntervalMs,
479+
maxIdleIntervalMs: this.opts.maxIdleIntervalMs,
480+
runTick: () => this.pollOnce(),
481+
onTickError: (err) => { this.logger?.warn?.('DbQueueAdapter: poll tick failed', err); },
482+
});
483+
this.loop.start();
452484
}
453485

454486
async stop(): Promise<void> {
455-
if (this.timer) { clearInterval(this.timer); this.timer = undefined; }
487+
if (!this.loop) return;
488+
const loop = this.loop;
489+
this.loop = undefined;
490+
await loop.stop();
491+
}
492+
493+
/**
494+
* [#17612] Work is claimable now: run a tick at once and reset the idle
495+
* backoff. A no-op while the worker is stopped — `start()` ticks immediately
496+
* anyway. This is what keeps {@link DbQueueAdapterOptions.maxIdleIntervalMs}
497+
* off the latency path for anything published through this process.
498+
*/
499+
wake(): void {
500+
this.loop?.wake();
456501
}
457502

458503
/** Test-friendly synchronous poll. */

0 commit comments

Comments
 (0)