diff --git a/.changeset/17612-db-queue-idle-backoff.md b/.changeset/17612-db-queue-idle-backoff.md new file mode 100644 index 0000000000..293dc86ed9 --- /dev/null +++ b/.changeset/17612-db-queue-idle-backoff.md @@ -0,0 +1,14 @@ +--- +"@objectstack/core": minor +"@objectstack/service-queue": minor +"@objectstack/service-messaging": patch +--- + +`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). + +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. + +- **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`.** +- **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. +- **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.** +- ⚠️ **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. diff --git a/packages/services/service-messaging/src/dispatch-loop.ts b/packages/core/src/dispatch-loop.ts similarity index 88% rename from packages/services/service-messaging/src/dispatch-loop.ts rename to packages/core/src/dispatch-loop.ts index 9e7946c1d3..5d063e15e0 100644 --- a/packages/services/service-messaging/src/dispatch-loop.ts +++ b/packages/core/src/dispatch-loop.ts @@ -1,13 +1,22 @@ // Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. /** - * The timer loop both outbox dispatchers run — `NotificationDispatcher` over - * `sys_notification_delivery` and `HttpDispatcher` over `sys_http_delivery`. + * The timer loop every polling worker in the platform runs — + * `NotificationDispatcher` over `sys_notification_delivery`, `HttpDispatcher` + * over `sys_http_delivery`, and `DbQueueAdapter` over `sys_job_queue`. * * #17610 wrote this loop inside `NotificationDispatcher`. #17623 found * `HttpDispatcher` still on the fixed 500 ms `setInterval` the notification side - * had just left, and moved the loop here, so the two dispatchers run one - * implementation of these rules instead of two copies that can drift: + * had just left, and pulled the loop out into one implementation of these rules + * instead of two copies that can drift. #17612 found the third copy — + * `DbQueueAdapter` on a flat 1 s `setInterval` — and that is why the loop + * lives HERE, in `@objectstack/core`, rather than in either service: it is a + * timing primitive, owned by neither the messaging domain nor the queue + * domain, and a service-to-service dependency between them to share it would + * invert the direction (a queue service depending on a messaging service). + * `@objectstack/core` is the package all three already depend on. + * + * The rules, one implementation: * * - **Never two ticks at once.** A tick asked for while one is running (a * {@link DispatchLoop.wake}, typically) becomes ONE follow-up tick, run the diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 9860d2ffe0..fc1f415cd6 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -128,6 +128,12 @@ export * from './dependency-resolver.js'; // Export Phase 3 components - Package lifecycle management export * from './namespace-resolver.js'; +// [#17612] The one polling-worker timer loop — tick coalescing, idle backoff +// and a final stop() — shared by every worker that drains a table +// (service-messaging's two outbox dispatchers, service-queue's DbQueueAdapter). +// It lives here because it is a timing primitive belonging to neither domain. +export * from './dispatch-loop.js'; + // Re-export contracts from @objectstack/spec for backward compatibility export type { Logger, diff --git a/packages/services/service-messaging/src/dispatcher.ts b/packages/services/service-messaging/src/dispatcher.ts index 083c99761d..68051b59f7 100644 --- a/packages/services/service-messaging/src/dispatcher.ts +++ b/packages/services/service-messaging/src/dispatcher.ts @@ -4,7 +4,7 @@ import type { MessagingChannel, MessagingChannelContext, Notification, SendResul import type { AckResult, ClaimedDeliveryRecord, INotificationOutbox, NotificationDeliveryRecord } from './outbox.js'; import { classifyDeliveryAttempt } from './backoff.js'; import { renderDigest } from './digest-render.js'; -import { DispatchLoop } from './dispatch-loop.js'; +import { DispatchLoop } from '@objectstack/core'; /** Minimal channel-registry surface the dispatcher needs (MessagingService satisfies it). */ export interface ChannelRegistry { @@ -49,7 +49,7 @@ export interface NotificationDispatcherLogger { * {@link NotificationDispatcherOptions.maxIdleIntervalMs}. It lives with the * loop both dispatchers run (#17623) and stays exported from here. */ -export { DEFAULT_MAX_IDLE_INTERVAL_MS } from './dispatch-loop.js'; +export { DEFAULT_MAX_IDLE_INTERVAL_MS } from '@objectstack/core'; export interface NotificationDispatcherOptions { nodeId: string; 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 1896d5422d..76123b9113 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 @@ -6,7 +6,7 @@ * * The loop used to tick every `intervalMs` (500 ms) forever, whatever it found — * the shape #17610 removed from `NotificationDispatcher`. Both dispatchers now - * run the one `DispatchLoop` (`dispatch-loop.ts`), whose mechanics are pinned in + * run the one `DispatchLoop` (`@objectstack/core`), whose mechanics are pinned in * full through the notification side in `dispatcher-idle-backoff.test.ts`. This * file pins the same behaviour through `HttpDispatcher`, plus what only the HTTP * side has: a retry schedule the backoff must not starve, and its own ingress — diff --git a/packages/services/service-messaging/src/http-dispatcher.ts b/packages/services/service-messaging/src/http-dispatcher.ts index f8e5db21f4..a9dfab4647 100644 --- a/packages/services/service-messaging/src/http-dispatcher.ts +++ b/packages/services/service-messaging/src/http-dispatcher.ts @@ -1,7 +1,7 @@ // Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. import type { DispatchCluster, DispatchLockHandle } from './dispatcher.js'; -import { DispatchLoop } from './dispatch-loop.js'; +import { DispatchLoop } from '@objectstack/core'; import { classifyAttempt, sendOnce, type FetchImpl } from './http-sender.js'; import type { HttpAckResult, HttpClaimCredential, HttpDelivery, IHttpOutbox } from './http-outbox.js'; diff --git a/packages/services/service-queue/src/db-queue-adapter.ts b/packages/services/service-queue/src/db-queue-adapter.ts index cb8ef2f62a..255e60a71f 100644 --- a/packages/services/service-queue/src/db-queue-adapter.ts +++ b/packages/services/service-queue/src/db-queue-adapter.ts @@ -8,6 +8,7 @@ import type { QueueHandler, } from '@objectstack/spec/contracts'; import { SysJobQueue } from '@objectstack/platform-objects/audit'; +import { DispatchLoop, DEFAULT_MAX_IDLE_INTERVAL_MS } from '@objectstack/core'; import { SYSTEM_CTX, uid, @@ -135,6 +136,21 @@ export interface LifecycleFloorRegistrar { export interface DbQueueAdapterOptions { /** Polling interval for the worker loop (ms, default 1000) */ pollIntervalMs?: number; + /** + * [#17612] Ceiling of the IDLE backoff, in ms (default + * {@link DEFAULT_MAX_IDLE_INTERVAL_MS}, 30 s). Each tick that claims nothing + * doubles the delay to the next from `pollIntervalMs` up to this value; any + * tick that claims work, and every {@link DbQueueAdapter.wake}, snaps it + * 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. + */ + maxIdleIntervalMs?: number; /** Max messages claimed per poll tick (default 10) */ batchSize?: number; /** Lease duration before another worker may reclaim (ms, default 30000) */ @@ -192,8 +208,7 @@ export class DbQueueAdapter implements IQueueService { private readonly opts: Required> & { workerId: string }; private readonly handlers = new Map(); - private timer?: ReturnType; - private running = false; + private loop?: DispatchLoop; constructor(args: { engine: JobEngine; @@ -207,6 +222,7 @@ export class DbQueueAdapter implements IQueueService { const o = args.options ?? {}; this.opts = { pollIntervalMs: o.pollIntervalMs ?? 1000, + maxIdleIntervalMs: o.maxIdleIntervalMs ?? DEFAULT_MAX_IDLE_INTERVAL_MS, batchSize: o.batchSize ?? 10, leaseMs: o.leaseMs ?? 30_000, idempotencyWindowMs: o.idempotencyWindowMs ?? 24 * 60 * 60 * 1000, @@ -353,6 +369,13 @@ export class DbQueueAdapter implements IQueueService { updated_at: now.toISOString(), }, { 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(); + return id; } @@ -426,6 +449,9 @@ export class DbQueueAdapter implements IQueueService { scheduled_for: now.toISOString(), updated_at: now.toISOString(), }, { context: SYSTEM_CTX }); + // [#17612] `replay` re-arms the row for RIGHT NOW, so it wakes the loop + // unconditionally — there is no deferred case here to discriminate. + this.wake(); } async purgeFailed(messageId: string): Promise { @@ -440,19 +466,38 @@ export class DbQueueAdapter implements IQueueService { // ── Worker lifecycle ───────────────────────────────────────────── start(): void { - if (this.timer) return; - this.timer = setInterval(() => { - if (this.running) return; - this.running = true; - this.pollOnce() - .catch((err) => { this.logger?.warn?.('DbQueueAdapter: poll tick failed', err); }) - .finally(() => { this.running = false; }); - }, this.opts.pollIntervalMs); - (this.timer as any)?.unref?.(); + if (this.loop) return; + // [#17612] The shared polling loop (`@objectstack/core`), not a bare + // `setInterval`: it coalesces overlapping ticks — the job the `running` + // flag here used to do — AND backs off while the queue is idle. A flat 1 s + // poll cost 3600 candidate SELECTs an hour per registered queue whatever + // was in the table; on a remote driver every one of those is an HTTP round + // trip. The same shape #17610 removed from `NotificationDispatcher` and + // #17623 from `HttpDispatcher`; this was the third copy. + this.loop = new DispatchLoop({ + intervalMs: this.opts.pollIntervalMs, + maxIdleIntervalMs: this.opts.maxIdleIntervalMs, + runTick: () => this.pollOnce(), + onTickError: (err) => { this.logger?.warn?.('DbQueueAdapter: poll tick failed', err); }, + }); + this.loop.start(); } async stop(): Promise { - if (this.timer) { clearInterval(this.timer); this.timer = undefined; } + if (!this.loop) return; + const loop = this.loop; + this.loop = undefined; + await loop.stop(); + } + + /** + * [#17612] Work is claimable now: run a tick at once and reset the idle + * backoff. A no-op while the worker is stopped — `start()` ticks immediately + * anyway. This is what keeps {@link DbQueueAdapterOptions.maxIdleIntervalMs} + * off the latency path for anything published through this process. + */ + wake(): void { + this.loop?.wake(); } /** Test-friendly synchronous poll. */ 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 new file mode 100644 index 0000000000..cf95f5f818 --- /dev/null +++ b/packages/services/service-queue/src/db-queue-idle-backoff.test.ts @@ -0,0 +1,290 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #17612 scope item 3 — what an IDLE `DbQueueAdapter` costs `sys_job_queue`, + * and that cutting it keeps every delivery guarantee. + * + * ## The measurement this pins + * + * `start()` ran a flat 1 s `setInterval`, so a registered-but-idle queue issued + * one candidate SELECT a second — 3600 an hour, per queue, forever, whatever + * was in the table. On a remote driver every one of those is an HTTP round + * trip. Items 1/2 of this card made each of those reads cheap (an indexed, + * `LIMIT`-bounded scan with no full sort); what was left was their COUNT, which + * is the subject #17610 fixed on the messaging side. `start()` now runs the one + * `DispatchLoop` (`@objectstack/core`), so the same idle hour is 124 reads. + * + * ## Why the flat-poll leg is FIRST, and not decoration + * + * An upper bound on statements is satisfied by a worker that has stopped + * ticking altogether — the exact failure this change could introduce and the + * one a low number cannot distinguish. So the harness is proved against the + * shape it replaced: with the backoff disabled the very same clock, engine and + * counter read 3600, which is what makes the 124 below a reading about the + * BACKOFF rather than about a dead loop. Every remaining leg then shows the + * same adapter still claiming, still draining and still waking. + * + * Every leg runs on vitest's fake timers, so "when did a tick happen" is an + * exact reading rather than a sleep-and-hope. + */ + +import { describe, it, expect, afterEach, vi } from 'vitest'; +import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/objectql'; +import { DEFAULT_MAX_IDLE_INTERVAL_MS } from '@objectstack/core'; +import { DbQueueAdapter } from './db-queue-adapter.js'; + +const QUEUE_TABLE = 'sys_job_queue'; +const BASE = 1000; +const HOUR = 60 * 60 * 1000; + +type Row = Record; + +/** + * Counting in-memory engine. Same `where:`-based find and `(table, {id,...})` + * update signature as the doubles in `db-queue-adapter.test.ts` / + * `job-queue-retention.test.ts`, plus a per-verb tally against + * `sys_job_queue` — the count is taken on the engine boundary the adapter + * really talks to, so a counted call is a statement production really issues. + */ +function makeCountingEngine() { + const tables = new Map(); + const calls = { find: 0, insert: 0, update: 0, delete: 0 }; + function compare(cell: any, v: any): boolean { + if (v !== null && typeof v === 'object' && !Array.isArray(v)) { + for (const [op, target] of Object.entries(v)) { + 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 '$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}`); + } + } + return true; + } + if (v === null) return cell == null; + return cell === v; + } + function matches(row: Row, where: Record): boolean { + for (const [k, v] of Object.entries(where)) { + if (k === '$or') { + if (!(v as Array>).some((leg) => matches(row, leg))) return false; + continue; + } + if (k.startsWith('$')) throw new Error(`fake driver: unsupported operator ${k}`); + if (!compare(row[k], v)) return false; + } + return true; + } + return { + tables, + calls, + async find(table: string, opts: any = {}) { + if (table === QUEUE_TABLE) calls.find++; + const t = tables.get(table) ?? []; + let out = opts.where ? t.filter((r) => matches(r, opts.where)) : [...t]; + if (opts.orderBy) { + for (const ord of [...opts.orderBy].reverse()) { + out.sort((a, b) => { + const av = a[ord.field], bv = b[ord.field]; + if (av === bv) return 0; + const cmp = av > bv ? 1 : -1; + return ord.order === 'desc' ? -cmp : cmp; + }); + } + } + // [#9540] Apply the caller's bound by PRESENCE, not truthiness: a + // `limit: 0` is a real bound meaning "no rows", and `if (opts.limit)` + // would hand back the whole table for it — a double LOOSER than the + // engine, on the one axis this suite counts. + if (typeof opts?.offset === 'number') out = out.slice(opts.offset); + if (typeof opts?.limit === 'number') out = out.slice(0, opts.limit); + return out; + }, + async insert(table: string, data: Row) { + if (table === QUEUE_TABLE) calls.insert++; + const t = tables.get(table) ?? []; + t.push({ ...data }); + tables.set(table, t); + return { id: data.id }; + }, + async update(table: string, patch: Row, options?: any) { + // [#5480] Opened with ObjectQL.update's OWN dispatch predicate, for + // the same reason `delete` below is: a predicate update rewrites + // every matching row, so a double looser than the engine here hides + // exactly the writes it was introduced to observe. + assertEngineUpdateDispatch(patch, options); + if (table === QUEUE_TABLE) calls.update++; + const r = (tables.get(table) ?? []).find((x) => x.id === patch.id); + if (!r) throw new Error(`row ${patch.id} not found in ${table}`); + Object.assign(r, patch); + return r; + }, + async delete(table: string, opts: any) { + // [#4550] Opened with ObjectQL.delete's OWN dispatch predicate rather + // than a hand-mirrored `if`: a double looser than the engine it stands + // in for is how #4434 shipped a dead REST route with its suite green. + if (table === QUEUE_TABLE) calls.delete++; + const dispatch = assertEngineDeleteDispatch(opts); + const t = tables.get(table) ?? []; + if (dispatch.kind === 'multi') { + const keep = t.filter((r) => !matches(r, opts?.where ?? {})); + tables.set(table, keep); + return t.length - keep.length; + } + tables.set(table, t.filter((r) => r.id !== dispatch.id)); + return { id: dispatch.id }; + }, + }; +} + +function adapterOn( + engine: ReturnType, + options: Record = {}, +): DbQueueAdapter { + return new DbQueueAdapter({ + engine: engine as any, + options: { pollIntervalMs: BASE, autoStart: false, ...options }, + }); +} + +afterEach(() => { vi.useRealTimers(); }); + +describe('#17612 DbQueueAdapter — idle poll cost', () => { + it('NEGATIVE CONTROL: with the backoff disabled the same harness reads a flat 3600 an idle hour', async () => { + vi.useFakeTimers(); + const engine = makeCountingEngine(); + // A ceiling at or below the base interval is the documented "no backoff" + // setting — i.e. exactly the 1 s `setInterval` this change replaced. + const adapter = adapterOn(engine, { maxIdleIntervalMs: BASE }); + await adapter.subscribe('q1', async () => {}); + adapter.start(); + await vi.advanceTimersByTimeAsync(HOUR); + await adapter.stop(); + + // 3600 s / 1 s, plus the tick `start()` runs immediately. + expect(engine.calls.find).toBe(3601); + }); + + it('backs off while idle: the same hour, same queue, is 124 reads instead of 3600', async () => { + vi.useFakeTimers(); + const engine = makeCountingEngine(); + const adapter = adapterOn(engine); + await adapter.subscribe('q1', async () => {}); + adapter.start(); + await vi.advanceTimersByTimeAsync(HOUR); + 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); + // Nothing was claimed, so nothing was written. + expect(engine.calls.update).toBe(0); + }); + + it('costs exactly one read per registered queue per tick', async () => { + for (const n of [1, 3]) { + const engine = makeCountingEngine(); + const adapter = adapterOn(engine); + for (let i = 0; i < n; i++) await adapter.subscribe(`q${i}`, async () => {}); + expect(await adapter.pollOnce()).toBe(0); + expect(engine.calls.find).toBe(n); + } + }); + + it('VACUITY TRAP: the same backed-off adapter still drains work it is given', async () => { + vi.useFakeTimers(); + const engine = makeCountingEngine(); + const adapter = adapterOn(engine); + const seen: unknown[] = []; + await adapter.subscribe('q1', async (msg: any) => { seen.push(msg.data); }); + adapter.start(); + // Idle long enough to be deep in the backoff. + await vi.advanceTimersByTimeAsync(5 * 60_000); + expect(seen).toEqual([]); + + await adapter.publish('q1', { n: 1 }); + await vi.advanceTimersByTimeAsync(BASE); + await adapter.stop(); + + expect(seen).toEqual([{ n: 1 }]); + const [row] = engine.tables.get(QUEUE_TABLE) ?? []; + expect(row?.status).toBe('completed'); + }); + + it('a due publish WAKES the loop — latency stays at the base interval, not the 30 s ceiling', async () => { + vi.useFakeTimers(); + const engine = makeCountingEngine(); + const adapter = adapterOn(engine); + const at: number[] = []; + await adapter.subscribe('q1', async () => { at.push(Date.now()); }); + adapter.start(); + await vi.advanceTimersByTimeAsync(5 * 60_000); + + const publishedAt = Date.now(); + await adapter.publish('q1', { n: 1 }); + // Run only the microtask/immediate work the wake schedules: far less + // than one backed-off interval, so a loop that merely waited fails here. + await vi.advanceTimersByTimeAsync(1); + await adapter.stop(); + + expect(at).toHaveLength(1); + expect(at[0]! - publishedAt).toBeLessThan(DEFAULT_MAX_IDLE_INTERVAL_MS); + expect(at[0]! - publishedAt).toBeLessThanOrEqual(BASE); + }); + + it('a DEFERRED publish does not wake the loop — the backoff is kept for a tick that would claim nothing', async () => { + vi.useFakeTimers(); + const engine = makeCountingEngine(); + const adapter = adapterOn(engine); + await adapter.subscribe('q1', async () => {}); + adapter.start(); + await vi.advanceTimersByTimeAsync(5 * 60_000); + const findsWhileIdle = engine.calls.find; + + // Due in ten minutes: the loop's own contract already covers noticing it + // within one backed-off interval of it coming due. + await adapter.publish('q1', { n: 1 }, { delay: 10 * 60_000 }); + await vi.advanceTimersByTimeAsync(1); + await adapter.stop(); + + expect(engine.calls.find).toBe(findsWhileIdle); + }); + + it('replay() wakes the loop — a re-armed dead letter is not left to the ceiling', async () => { + vi.useFakeTimers(); + const engine = makeCountingEngine(); + const adapter = adapterOn(engine); + await adapter.subscribe('q1', async () => { throw new Error('boom'); }); + const id = await adapter.publish('q1', { n: 1 }, { maxAttempts: 1 }); + // Drain it into the dead-letter surface without the loop running. + await adapter.pollOnce(); + expect((engine.tables.get(QUEUE_TABLE) ?? [])[0]?.status).toBe('dlq'); + + adapter.start(); + await vi.advanceTimersByTimeAsync(5 * 60_000); + const findsWhileIdle = engine.calls.find; + + await adapter.replay(id); + await vi.advanceTimersByTimeAsync(1); + await adapter.stop(); + + expect(engine.calls.find).toBeGreaterThan(findsWhileIdle); + }); + + it('stop() is final: no tick survives it, and a later wake() is a no-op', async () => { + vi.useFakeTimers(); + const engine = makeCountingEngine(); + const adapter = adapterOn(engine); + await adapter.subscribe('q1', async () => {}); + adapter.start(); + await vi.advanceTimersByTimeAsync(10 * BASE); + await adapter.stop(); + const findsAtStop = engine.calls.find; + + adapter.wake(); + await vi.advanceTimersByTimeAsync(HOUR); + + expect(engine.calls.find).toBe(findsAtStop); + }); +}); diff --git a/packages/services/service-queue/vitest.config.ts b/packages/services/service-queue/vitest.config.ts index 5e0591efc4..dac492507f 100644 --- a/packages/services/service-queue/vitest.config.ts +++ b/packages/services/service-queue/vitest.config.ts @@ -5,6 +5,7 @@ // for every test file in the package (packages/cli/vitest.config.ts's header // records the incident that taught that). import { defineConfig } from 'vitest/config'; +import path from 'node:path'; export default defineConfig({ test: { @@ -16,4 +17,24 @@ export default defineConfig({ // Enforced repo-wide by scripts/check-console-intercept-disarm.mjs. disableConsoleIntercept: true, }, + resolve: { + // ARRAY form with an ANCHORED `find`, deliberately. The object form matches + // by PREFIX, so a bare key whose replacement is a FILE also swallows that + // package's subpaths and resolves them to `.../index.ts/` — + // ENOTDIR at run time, in a config that reads as correct. + // `scripts/check-test-source-alias.mjs` is the authority on the rule. + alias: [ + // [#17612] `db-queue-adapter.ts` runs the shared `DispatchLoop` from + // @objectstack/core, and this suite's whole subject is that loop's idle + // cadence. Without this entry the specifier resolves through core's + // `exports` to its `dist/`, so these verdicts would be a function of + // another package's BUILD STATE rather than of the source in this + // checkout — and the dangerous direction is the quiet one: an unbuilt + // change to the loop leaves the backoff legs green against stale bytes. + { + find: /^@objectstack\/core$/, + replacement: path.resolve(__dirname, '../../core/src/index.ts'), + }, + ], + }, }); diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index 5f8d55bf37..49e035716d 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3841,6 +3841,16 @@ "verb": "delete", "pinned": 1 }, + { + "file": "packages/services/service-queue/src/db-queue-idle-backoff.test.ts", + "verb": "delete", + "pinned": 1 + }, + { + "file": "packages/services/service-queue/src/db-queue-idle-backoff.test.ts", + "verb": "update", + "pinned": 1 + }, { "file": "packages/services/service-queue/src/job-queue-retention.test.ts", "verb": "delete",