From 000356bde9f4eb14d4e4f3f98ee469d4f8423aab Mon Sep 17 00:00:00 2001 From: Jack Zhuang <50353452+hotlong@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:58:27 +0800 Subject: [PATCH 1/4] wip(service-messaging): reap once per dispatcher tick, back off while idle, wake on enqueue Claude-Session: https://claude.ai/code/session_c5c0ce54-bb9c-478c-9e5b-cf44b80d4569 Co-authored-by: Claude --- .../service-messaging/src/dispatcher.ts | 170 ++++++++++++++++-- .../services/service-messaging/src/index.ts | 2 + .../service-messaging/src/memory-outbox.ts | 37 ++-- .../src/messaging-service-plugin.ts | 23 ++- .../src/messaging-service.ts | 14 +- .../services/service-messaging/src/outbox.ts | 34 ++++ .../service-messaging/src/sql-outbox.ts | 48 +++-- 7 files changed, 268 insertions(+), 60 deletions(-) diff --git a/packages/services/service-messaging/src/dispatcher.ts b/packages/services/service-messaging/src/dispatcher.ts index 81b4cdb5d9..1ada48da55 100644 --- a/packages/services/service-messaging/src/dispatcher.ts +++ b/packages/services/service-messaging/src/dispatcher.ts @@ -43,6 +43,12 @@ export interface NotificationDispatcherLogger { info?: (msg: string, meta?: any) => void; } +/** + * [#17610] Default ceiling of the idle backoff, in ms — see + * {@link NotificationDispatcherOptions.maxIdleIntervalMs}. + */ +export const DEFAULT_MAX_IDLE_INTERVAL_MS = 30_000; + export interface NotificationDispatcherOptions { nodeId: string; outbox: INotificationOutbox; @@ -53,7 +59,22 @@ export interface NotificationDispatcherOptions { cluster?: DispatchCluster; partitionCount?: number; batchSize?: number; + /** Tick interval in ms while ticks find work (default 500). */ intervalMs?: number; + /** + * [#17610] Idle backoff ceiling in ms (default {@link DEFAULT_MAX_IDLE_INTERVAL_MS}). + * Each loop tick that claims nothing doubles the delay before the next one, + * from `intervalMs` up to this; a tick that claims anything, or a + * {@link NotificationDispatcher.wake} call, snaps it back to `intervalMs`. + * A value at or below `intervalMs` disables the backoff. + * + * While idle this bounds how late the loop notices work nobody woke it for: + * a deferred row coming due (retry schedule, quiet hours, a digest window), + * a row enqueued by a process this dispatcher does not serve, and a crashed + * node's `in_flight` rows — reaped at most `claimTtlMs` + this after their + * claim, where the fixed interval gave `claimTtlMs` + `intervalMs`. + */ + maxIdleIntervalMs?: number; lockTtlMs?: number; claimTtlMs?: number; rng?: () => number; @@ -68,22 +89,39 @@ export interface NotificationDispatcherOptions { * NotificationDispatcher (ADR-0030 P1) — drains the `sys_notification_delivery` * outbox and sends each row through its channel, retrying with backoff and * dead-lettering once the budget is exhausted. Structurally mirrors - * `WebhookDispatcher`: an interval loop walks `partitionCount` partitions, each + * `WebhookDispatcher`: a timer loop walks `partitionCount` partitions, each * guarded by a per-partition cluster lock; within a held partition it claims a * batch (`pending → in_flight`), sends, and acks. * * At-least-once: if a channel send succeeds but the ack write fails, the row * reverts to pending after the claim TTL and is re-sent — the inbox channel's * receipt write is idempotent-friendly, and downstream channels should be too. + * + * ## What an idle tick costs (#17610) + * + * Against an empty outbox one tick is `1 + 2 × partitionCount` store round + * trips: ONE visibility-timeout reap for the whole environment, then a claim + * probe and a digest probe per partition. The reap used to run inside both + * claims of every partition — `2 × partitionCount` identical environment-wide + * UPDATEs a tick, 16 of the 32 statements a tick issued with the default 8 + * partitions — on a fixed 500 ms interval that never let up, one loop per warm + * kernel. The loop now also backs off while idle + * ({@link NotificationDispatcherOptions.maxIdleIntervalMs}), and + * {@link NotificationDispatcher.wake} — which the messaging service calls when + * `emit()` enqueues deliveries — runs the next tick at once. */ export class NotificationDispatcher { private readonly opts: Required< Omit > & Pick & { cluster: DispatchCluster }; - private timer: ReturnType | undefined; + private timer: ReturnType | undefined; private running = false; private inflightTick: Promise | undefined; + /** [#17610] Consecutive loop ticks that claimed nothing — the idle backoff's exponent. */ + private idleTicks = 0; + /** [#17610] A tick was asked for while one was running: run one more the moment it settles. */ + private tickRequested = false; constructor(options: NotificationDispatcherOptions) { const intervalMs = options.intervalMs ?? 500; @@ -97,6 +135,8 @@ export class NotificationDispatcher { partitionCount: options.partitionCount ?? 8, batchSize: options.batchSize ?? 32, intervalMs, + // A ceiling below the base interval just means "no backoff". + maxIdleIntervalMs: Math.max(intervalMs, options.maxIdleIntervalMs ?? DEFAULT_MAX_IDLE_INTERVAL_MS), lockTtlMs, claimTtlMs: options.claimTtlMs ?? lockTtlMs * 2, rng: options.rng, @@ -106,66 +146,156 @@ export class NotificationDispatcher { }; } - /** Begin the periodic loop. Idempotent. */ + /** Begin the loop; the first tick runs immediately. Idempotent. */ start(): void { if (this.running) return; this.running = true; - this.scheduleTick(); - this.timer = setInterval(() => this.scheduleTick(), this.opts.intervalMs); - // Don't keep the event loop alive solely for the dispatcher. - (this.timer as { unref?: () => void })?.unref?.(); + this.idleTicks = 0; + this.loopTick(); } /** Stop the loop and drain the in-flight tick. */ async stop(): Promise { if (!this.running) return; this.running = false; - if (this.timer) { - clearInterval(this.timer); - this.timer = undefined; - } + this.tickRequested = false; + this.clearTimer(); if (this.inflightTick) { try { await this.inflightTick; } catch { /* already logged */ } } } - /** Run one full tick (all partitions). Exposed for deterministic tests. */ + /** + * [#17610] Work was just enqueued: tick now and reset the idle backoff. + * + * The messaging service calls this after `emit()` enqueues deliveries, so a + * notification raised in this process never waits out a backed-off + * interval. A wake that lands while a tick is running queues ONE follow-up + * tick for the moment it settles — the running tick may already be past the + * partition the new row hashed into — and every wake in that window + * collapses into that one. No-op while stopped. + */ + wake(): void { + if (!this.running) return; + this.idleTicks = 0; + this.loopTick(); + } + + /** Run one full tick (the reap, then all partitions). Exposed for deterministic tests. */ async tick(): Promise { await this.runTick(); } - private scheduleTick(): void { - if (this.inflightTick) return; + /** + * One tick of the loop, then the timer for the next. Never two at once: a + * call that finds a tick in flight becomes a follow-up request instead. + */ + private loopTick(): void { + if (!this.running) return; + this.clearTimer(); + if (this.inflightTick) { + this.tickRequested = true; + return; + } + const startedAt = Date.now(); this.inflightTick = this.runTick() + .then((claimed) => { + this.idleTicks = claimed > 0 ? 0 : this.idleTicks + 1; + }) .catch((err) => { + // A failing store is not work: back off rather than hammer it. + this.idleTicks += 1; this.opts.logger?.warn?.('notification-dispatcher: tick failed', { nodeId: this.opts.nodeId, error: (err as Error)?.message ?? String(err), }); }) - .finally(() => { this.inflightTick = undefined; }); + .finally(() => { + this.inflightTick = undefined; + if (!this.running) return; + if (this.tickRequested) { + this.tickRequested = false; + this.idleTicks = 0; + this.loopTick(); + return; + } + this.schedule(Math.max(0, this.nextIntervalMs() - (Date.now() - startedAt))); + }); + } + + /** + * [#17610] Delay before the next loop tick, measured from the START of the + * last one: `intervalMs` while ticks claim work, doubled for every + * consecutive empty tick after that, capped at `maxIdleIntervalMs`. + */ + private nextIntervalMs(): number { + const { intervalMs, maxIdleIntervalMs } = this.opts; + if (this.idleTicks === 0) return intervalMs; + // Exponent clamped so the product stays finite long after the cap wins. + return Math.min(maxIdleIntervalMs, intervalMs * 2 ** Math.min(this.idleTicks, 30)); } - private async runTick(): Promise { + private schedule(delayMs: number): void { + this.clearTimer(); + this.timer = setTimeout(() => { + this.timer = undefined; + this.loopTick(); + }, delayMs); + // Don't keep the event loop alive solely for the dispatcher. + (this.timer as { unref?: () => void })?.unref?.(); + } + + private clearTimer(): void { + if (this.timer === undefined) return; + clearTimeout(this.timer); + this.timer = undefined; + } + + /** One full pass: the reap, then every partition. Resolves to the rows claimed. */ + private async runTick(): Promise { + // [#17610] Visibility-timeout recovery ONCE per tick, BEFORE any claim. + // Its predicate names no partition, so this one run hands every claim + // below each row that had already expired when the tick began — what + // reaping inside every claim achieved, less the rows that expire DURING + // this tick, which the next tick's reap returns. An abandoned claim is + // still recovered within one tick of `claimTtlMs` passing (one backed-off + // tick while idle, see `maxIdleIntervalMs`), and the TTL keeps its + // meaning: a claim is never re-taken before it. + // + // No partition lock is needed, and none was ever in force: the reap only + // moves rows already past their timeout, a claim only takes `pending` + // rows, and an ack whose claim was reaped matches nothing (#11859) — while + // the per-claim reap, run under partition p's lock, was already rewriting + // rows in every other partition. + await this.opts.outbox.reap({ claimTtlMs: this.opts.claimTtlMs }); + const count = this.opts.partitionCount; const offset = stableNodeOffset(this.opts.nodeId, count); + let claimed = 0; for (let step = 0; step < count; step++) { - await this.runPartition((offset + step) % count); + claimed += await this.runPartition((offset + step) % count); } + return claimed; } - private async runPartition(index: number): Promise { + /** + * Claim and send within one partition's lock. Resolves to the number of rows + * claimed — 0 when another node holds the lock. + */ + private async runPartition(index: number): Promise { const handle = await this.opts.cluster.lock.acquire(`notify.dispatcher.partition.${index}`, { ttlMs: this.opts.lockTtlMs, waitMs: 0, }); - if (!handle) return; + if (!handle) return 0; try { const claimed = await this.opts.outbox.claim({ nodeId: this.opts.nodeId, limit: this.opts.batchSize, partition: { index, count: this.opts.partitionCount }, claimTtlMs: this.opts.claimTtlMs, + // [#17610] Reaped once for the whole tick in runTick(). + skipReap: true, }); if (claimed.length > 0) { await handle.renew?.(this.opts.lockTtlMs); @@ -183,6 +313,7 @@ export class NotificationDispatcher { limit: this.opts.batchSize, partition: { index, count: this.opts.partitionCount }, claimTtlMs: this.opts.claimTtlMs, + skipReap: true, }); if (digestRows.length > 0) { await handle.renew?.(this.opts.lockTtlMs); @@ -191,6 +322,7 @@ export class NotificationDispatcher { await this.processDigestGroup(group); } } + return claimed.length + digestRows.length; } finally { await handle.release(); } diff --git a/packages/services/service-messaging/src/index.ts b/packages/services/service-messaging/src/index.ts index 6edfd49fef..d1ba1dabb9 100644 --- a/packages/services/service-messaging/src/index.ts +++ b/packages/services/service-messaging/src/index.ts @@ -104,6 +104,8 @@ export type { DeliveryPayload, EnqueueDeliveryInput, ClaimOptions, + // [#17610] The dispatcher's once-per-tick visibility-timeout recovery. + ReapOptions, AckResult, } from './outbox.js'; // [#11453] `ack()`'s status precondition refuses with this, so a caller that diff --git a/packages/services/service-messaging/src/memory-outbox.ts b/packages/services/service-messaging/src/memory-outbox.ts index f2a9467861..3bbd19f17f 100644 --- a/packages/services/service-messaging/src/memory-outbox.ts +++ b/packages/services/service-messaging/src/memory-outbox.ts @@ -9,6 +9,7 @@ import type { EnqueueDeliveryInput, INotificationOutbox, NotificationDeliveryRecord, + ReapOptions, } from './outbox.js'; import { NotificationAckError, @@ -67,17 +68,14 @@ export class MemoryNotificationOutbox implements INotificationOutbox { return id; } + async reap(opts: ReapOptions): Promise { + this.reapExpired(opts.now ?? this.clock(), opts.claimTtlMs); + } + async claim(opts: ClaimOptions): Promise { const now = opts.now ?? this.clock(); - // Reap stale in_flight. - for (const r of this.rows.values()) { - if (r.status === 'in_flight' && (r.claimedAt ?? 0) < now - opts.claimTtlMs) { - r.status = 'pending'; - r.claimedBy = undefined; - r.claimedAt = undefined; - r.updatedAt = now; - } - } + // Reap stale in_flight — unless the caller already reaped this pass. + if (!opts.skipReap) this.reapExpired(now, opts.claimTtlMs); const out: ClaimedDeliveryRecord[] = []; for (const r of this.rows.values()) { if (out.length >= opts.limit) break; @@ -99,14 +97,7 @@ export class MemoryNotificationOutbox implements INotificationOutbox { async claimDigest(opts: ClaimOptions): Promise { const now = opts.now ?? this.clock(); // Reap stale in_flight (same as claim). - for (const r of this.rows.values()) { - if (r.status === 'in_flight' && (r.claimedAt ?? 0) < now - opts.claimTtlMs) { - r.status = 'pending'; - r.claimedBy = undefined; - r.claimedAt = undefined; - r.updatedAt = now; - } - } + if (!opts.skipReap) this.reapExpired(now, opts.claimTtlMs); // Claim every DUE batched row in the partition — a window must be taken // whole, so `limit` does not truncate a group here. const out: ClaimedDeliveryRecord[] = []; @@ -187,6 +178,18 @@ export class MemoryNotificationOutbox implements INotificationOutbox { } } + /** Visibility-timeout recovery: every expired `in_flight` claim reverts to `pending`. */ + private reapExpired(now: number, claimTtlMs: number): void { + for (const r of this.rows.values()) { + if (r.status === 'in_flight' && (r.claimedAt ?? 0) < now - claimTtlMs) { + r.status = 'pending'; + r.claimedBy = undefined; + r.claimedAt = undefined; + r.updatedAt = now; + } + } + } + async list(filter?: { status?: DeliveryStatus; notificationId?: string }): Promise { let rows = [...this.rows.values()]; if (filter?.status) rows = rows.filter((r) => r.status === filter.status); diff --git a/packages/services/service-messaging/src/messaging-service-plugin.ts b/packages/services/service-messaging/src/messaging-service-plugin.ts index ef33fa83b1..9454859202 100644 --- a/packages/services/service-messaging/src/messaging-service-plugin.ts +++ b/packages/services/service-messaging/src/messaging-service-plugin.ts @@ -7,7 +7,7 @@ import { MessagingService } from './messaging-service.js'; import { createInboxChannel } from './inbox-channel.js'; import { SqlNotificationOutbox } from './sql-outbox.js'; import { SqlHttpOutbox } from './sql-http-outbox.js'; -import { NotificationDispatcher, type DispatchCluster } from './dispatcher.js'; +import { DEFAULT_MAX_IDLE_INTERVAL_MS, NotificationDispatcher, type DispatchCluster } from './dispatcher.js'; import { HttpDispatcher } from './http-dispatcher.js'; import { createEmailChannel } from './email-channel.js'; import { createSmsChannel } from './sms-channel.js'; @@ -45,8 +45,19 @@ export interface MessagingServicePluginOptions { reliableDelivery?: boolean; /** Outbox/dispatcher partition count (default 8). */ partitionCount?: number; - /** Dispatcher tick interval in ms (default 500). */ + /** Dispatcher tick interval in ms while there is work (default 500). */ dispatchIntervalMs?: number; + /** + * [#17610] Ceiling in ms for the notification dispatcher's idle backoff + * (default 30000). Consecutive ticks that claim nothing double the interval + * from `dispatchIntervalMs` up to this; a tick that claims work snaps it + * back, and an `emit()` that enqueues deliveries wakes the dispatcher at + * once. While idle it bounds how late the dispatcher notices work nobody + * woke it for: a deferred delivery coming due (retry, quiet hours, digest + * window), a row enqueued by another process, a crashed node's claim passing + * its timeout. A value at or below `dispatchIntervalMs` disables the backoff. + */ + dispatchMaxIdleIntervalMs?: number; /** * Topics that bypass the per-user preference matrix (ADR-0030 P2) — e.g. * security/system alerts users must not be able to mute. Exact match, or a @@ -99,6 +110,7 @@ export class MessagingServicePlugin implements Plugin { reliableDelivery: true, partitionCount: 8, dispatchIntervalMs: 500, + dispatchMaxIdleIntervalMs: DEFAULT_MAX_IDLE_INTERVAL_MS, mandatoryTopics: [], ...options, }; @@ -281,7 +293,9 @@ export class MessagingServicePlugin implements Plugin { return; } const outbox = new SqlNotificationOutbox(engine, { partitionCount: this.options.partitionCount }); - service.setOutbox(outbox); + // [#17610] Resolved at call time: the dispatcher is constructed + // below, and after destroy() the hook is a no-op. + service.setOutbox(outbox, { onEnqueued: () => this.dispatcher?.wake() }); let cluster: DispatchCluster | undefined; try { @@ -298,11 +312,12 @@ export class MessagingServicePlugin implements Plugin { cluster, partitionCount: this.options.partitionCount, intervalMs: this.options.dispatchIntervalMs, + maxIdleIntervalMs: this.options.dispatchMaxIdleIntervalMs, logger: ctx.logger, }); this.dispatcher.start(); ctx.logger.info( - `[messaging] reliable delivery on (outbox + dispatcher, ${this.options.partitionCount} partitions${cluster ? ', clustered' : ', single-node'})`, + `[messaging] reliable delivery on (outbox + dispatcher, ${this.options.partitionCount} partitions${cluster ? ', clustered' : ', single-node'}, idle backoff up to ${Math.max(this.options.dispatchIntervalMs, this.options.dispatchMaxIdleIntervalMs)}ms)`, ); // ADR-0018 M3: generic outbound-HTTP outbox + dispatcher. Backs diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index 6ff39483a4..b3a0ba77f4 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -181,6 +181,8 @@ export class MessagingService { private readonly resolver: RecipientResolver; private readonly preferences: PreferenceResolver; private outbox?: INotificationOutbox; + /** [#17610] Fired after an `emit()` enqueues deliveries — see {@link setOutbox}. */ + private onDeliveriesEnqueued?: () => void; private httpOutbox?: IHttpOutbox; /** [#8069] Producer vetoes over redelivery, keyed by `HttpDelivery.source`. */ private readonly redeliverGuards = new Map(); @@ -204,9 +206,17 @@ export class MessagingService { * Attach the durable delivery outbox after construction. The plugin wires * this once the data engine is resolvable (kernel:ready), switching `emit()` * from inline fan-out to the reliable enqueue → dispatcher path. + * + * [#17610] `onEnqueued` fires once per `emit()` that enqueued at least one + * delivery. The plugin points it at `NotificationDispatcher.wake()`: the + * dispatcher backs its tick interval off while the outbox is idle, and rows + * this process just wrote should go out on the next tick, not the next + * backed-off one. It belongs to the outbox it was attached with — attaching + * another outbox replaces (or clears) it. */ - setOutbox(outbox: INotificationOutbox): void { + setOutbox(outbox: INotificationOutbox, options: { onEnqueued?: () => void } = {}): void { this.outbox = outbox; + this.onDeliveriesEnqueued = options.onEnqueued; } /** @@ -904,6 +914,8 @@ export class MessagingService { // retroactively. `failed` keeps its meaning — an enqueue that threw // never reached the outbox at all. const enqueued = deliveries.filter((d) => d.ok).length; + // [#17610] Wake the dispatcher — see `setOutbox`. + if (enqueued > 0) this.onDeliveriesEnqueued?.(); return { notificationId, deduped: false, deliveries, delivered: 0, enqueued, failed: deliveries.length - enqueued, diff --git a/packages/services/service-messaging/src/outbox.ts b/packages/services/service-messaging/src/outbox.ts index 3ca0c1a9a2..697d9916af 100644 --- a/packages/services/service-messaging/src/outbox.ts +++ b/packages/services/service-messaging/src/outbox.ts @@ -102,6 +102,27 @@ export interface ClaimOptions { claimTtlMs: number; /** "Now" reference, ms. Defaults to Date.now(). */ now?: number; + /** + * [#17610] Skip the visibility-timeout reap this call otherwise runs before + * claiming. Default `false`: a direct `claim()` stays self-contained — the + * call that wants stale `in_flight` rows back is the one that recovers them. + * + * A caller that already ran {@link INotificationOutbox.reap} for this pass + * passes `true`. `NotificationDispatcher` reaps once per tick and then claims + * every partition twice (normal + digest); the reap is environment-wide — its + * predicate names no partition — so repeating it per claim recovers nothing + * the first run did not, and on an idle outbox with the default 8 partitions + * it was 16 of a tick's 32 statements. + */ + skipReap?: boolean; +} + +/** [#17610] Options for {@link INotificationOutbox.reap}. */ +export interface ReapOptions { + /** Visibility timeout — `in_flight` rows claimed longer ago than this revert to `pending`. */ + claimTtlMs: number; + /** "Now" reference, ms. Defaults to the store's clock. */ + now?: number; } export interface AckSuccess { @@ -209,6 +230,19 @@ export function notificationAckNoCredentialMessage(id: string): string { */ export interface INotificationOutbox { enqueue(input: EnqueueDeliveryInput): Promise; + /** + * [#17610] The visibility-timeout recovery on its own: every `in_flight` row + * whose claim is older than `claimTtlMs` reverts to `pending` with its claim + * credential cleared — across the WHOLE store, every partition and every + * organization. It is the step {@link claim} / {@link claimDigest} run first + * unless told `skipReap`, exposed so a dispatcher can run it once per pass + * instead of once per (partition × claim kind). + * + * Safe at any moment and from any number of nodes: it moves only rows already + * past their timeout, {@link claim} takes only `pending` rows, and an + * {@link ack} whose claim was reaped matches nothing and is refused (#11859). + */ + reap(opts: ReapOptions): Promise; claim(opts: ClaimOptions): Promise; /** * Record the outcome of ONE dispatch attempt on a row this caller claimed, diff --git a/packages/services/service-messaging/src/sql-outbox.ts b/packages/services/service-messaging/src/sql-outbox.ts index 72cfa57419..5ea002b8f2 100644 --- a/packages/services/service-messaging/src/sql-outbox.ts +++ b/packages/services/service-messaging/src/sql-outbox.ts @@ -10,6 +10,7 @@ import type { EnqueueDeliveryInput, INotificationOutbox, NotificationDeliveryRecord, + ReapOptions, } from './outbox.js'; import { hashPartition } from './backoff.js'; import { toEpochMs } from './audit-timestamp.js'; @@ -66,10 +67,11 @@ interface DeliveryRow { * `sys_stamp_audit_update` hook stamps it on every update unconditionally, and * `updated_at` is `readonly`, so a caller-supplied value is stripped by * `stripReadonlyFields` (#2948) before it reaches the driver — with a WARN per - * call. `claim()` / `claimDigest()` start with an unconditional reap UPDATE that - * runs whether or not a row is stale, so on an idle dev server the three claim - * paths × 8 partitions × a 500 ms dispatcher tick spammed 48 identical warnings - * a second and drowned the console. Writing the column was already a no-op + * call. The visibility-timeout reap is an unconditional UPDATE that runs whether + * or not a row is stale — `reap()`, and `claim()` / `claimDigest()` unless told + * `skipReap` — and while every claim path ran it per partition, an idle dev + * server's three claim paths × 8 partitions × a 500 ms dispatcher tick spammed + * 48 identical warnings a second and drowned the console. Writing the column was already a no-op * (stripped, then re-stamped); passing it as epoch-ms would also have been the * wrong shape for a native TIMESTAMP column (see `toEpochMs`) had it ever * survived the strip. Leave it to the platform. @@ -128,17 +130,16 @@ export class SqlNotificationOutbox implements INotificationOutbox { } } + async reap(opts: ReapOptions): Promise { + await this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); + } + async claim(opts: ClaimOptions): Promise { const now = opts.now ?? Date.now(); - // 1. Reap stale in_flight rows (visibility-timeout recovery). - await this.engine.update( - this.objectName, - { status: 'pending', claimed_by: null, claimed_at: null }, - // Environment-wide by design: recovers rows a crashed node abandoned, - // for every organization. Warrant in `outbox-dispatcher-scope.ts`. - dispatcherSweepOptions({ status: 'in_flight', claimed_at: { $lt: now - opts.claimTtlMs } }), - ); + // 1. Reap stale in_flight rows (visibility-timeout recovery) — unless the + // caller already ran `reap()` for this pass (#17610). + if (!opts.skipReap) await this.reapExpired(now, opts.claimTtlMs); // 2. Candidate ids: ready pending rows in our partition. Batched (digest) // rows are excluded — they drain via claimDigest so they collapse. @@ -179,13 +180,7 @@ export class SqlNotificationOutbox implements INotificationOutbox { const now = opts.now ?? Date.now(); // 1. Reap stale in_flight (same as claim). - await this.engine.update( - this.objectName, - { status: 'pending', claimed_by: null, claimed_at: null }, - // Environment-wide by design: recovers rows a crashed node abandoned, - // for every organization. Warrant in `outbox-dispatcher-scope.ts`. - dispatcherSweepOptions({ status: 'in_flight', claimed_at: { $lt: now - opts.claimTtlMs } }), - ); + if (!opts.skipReap) await this.reapExpired(now, opts.claimTtlMs); // 2. All DUE batched rows in our partition — a window is claimed whole, so // we don't apply `limit` (a generous cap guards a pathological backlog). @@ -336,6 +331,21 @@ export class SqlNotificationOutbox implements INotificationOutbox { } } + /** + * The visibility-timeout reap: ONE predicate UPDATE returning every expired + * `in_flight` claim to `pending`. No partition in the predicate — it spans + * the whole table by construction. + */ + private async reapExpired(now: number, claimTtlMs: number): Promise { + await this.engine.update( + this.objectName, + { status: 'pending', claimed_by: null, claimed_at: null }, + // Environment-wide by design: recovers rows a crashed node abandoned, + // for every organization. Warrant in `outbox-dispatcher-scope.ts`. + dispatcherSweepOptions({ status: 'in_flight', claimed_at: { $lt: now - claimTtlMs } }), + ); + } + async list(filter?: { status?: DeliveryStatus; notificationId?: string }): Promise { const where: Record = {}; if (filter?.status) where.status = filter.status; From 6821a0cce0dedc259392f3075993922eaba2318e Mon Sep 17 00:00:00 2001 From: Jack Zhuang <50353452+hotlong@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:05:39 +0800 Subject: [PATCH 2/4] test(service-messaging): pin idle tick cost, idle backoff, wake-on-emit and once-per-tick reap recovery Claude-Session: https://claude.ai/code/session_c5c0ce54-bb9c-478c-9e5b-cf44b80d4569 Co-authored-by: Claude --- ...17610-notification-dispatcher-idle-cost.md | 35 +++ ...ery-claim-tenant-audit.integration.test.ts | 26 ++ .../src/dispatcher-idle-backoff.test.ts | 256 ++++++++++++++++++ .../dispatcher-idle-cost.integration.test.ts | 183 +++++++++++++ .../plugin-enqueue-wakes-dispatcher.test.ts | 146 ++++++++++ 5 files changed, 646 insertions(+) create mode 100644 .changeset/17610-notification-dispatcher-idle-cost.md create mode 100644 packages/services/service-messaging/src/dispatcher-idle-backoff.test.ts create mode 100644 packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts create mode 100644 packages/services/service-messaging/src/plugin-enqueue-wakes-dispatcher.test.ts diff --git a/.changeset/17610-notification-dispatcher-idle-cost.md b/.changeset/17610-notification-dispatcher-idle-cost.md new file mode 100644 index 0000000000..be4cb94ebd --- /dev/null +++ b/.changeset/17610-notification-dispatcher-idle-cost.md @@ -0,0 +1,35 @@ +--- +'@objectstack/service-messaging': minor +--- + +`NotificationDispatcher` reaps once per tick instead of once per claim, backs off while the outbox is idle, and `emit()` wakes it (#17610) + +**What an idle dispatcher cost.** Against an EMPTY `sys_notification_delivery` outbox every tick walked `partitionCount` partitions (default 8) and ran `claim()` and `claimDigest()` in each — and each of those opened with the environment-wide visibility-timeout reap before its candidate SELECT. Measured on a real `ObjectQL` + `SqlDriver`: **32 statements a tick, 16 of them the identical reap UPDATE**, on a fixed 500 ms interval that never let up, one loop per warm kernel. On remote Turso every statement is an HTTP round trip. + +**Now:** + +- **The reap runs once per tick**, before any claim — an idle tick is `1 + 2 × partitionCount` = 17 statements. Its predicate names no partition, so one run returns every claim that had expired when the tick began; a claim that expires during the tick is returned by the next one. A crashed node's `in_flight` rows are still recovered within one tick of `claimTtlMs` passing, and a claim is still never re-taken before its TTL. +- **The loop backs off while idle.** Every tick that claims nothing doubles the delay to the next, from `intervalMs` up to `maxIdleIntervalMs` (default 30 s; `MessagingServicePlugin` option `dispatchMaxIdleIntervalMs`). A tick that claims work snaps back to `intervalMs`. With the defaults, ten idle minutes are 24 ticks instead of 1,201. +- **`emit()` wakes the dispatcher.** `MessagingService.setOutbox(outbox, { onEnqueued })` fires once per `emit()` that enqueued at least one delivery; the plugin points it at the new `NotificationDispatcher.wake()`, which ticks immediately — or once more, right after a tick already in flight. + +**Latency bound.** A notification emitted in the process that runs the dispatcher goes out on the tick `wake()` starts, no later than before. While idle, work nobody announces is noticed within one backed-off interval, at most `maxIdleIntervalMs` (30 s by default): a deferred delivery coming due (retry schedule, quiet hours, digest window), a row enqueued by a process that does not run this dispatcher, and a crashed node's expired claim (recovered within `claimTtlMs` + `maxIdleIntervalMs`). Set `dispatchMaxIdleIntervalMs` to `dispatchIntervalMs` to keep the fixed interval. + +**BREAKING (interface member added):** `INotificationOutbox` gains `reap(opts: ReapOptions)`, and `ClaimOptions` gains an optional `skipReap`. A custom outbox implementation must add `reap()` — the visibility-timeout recovery it already runs at the top of `claim()`, as a method of its own (both built-in stores, `SqlNotificationOutbox` and `MemoryNotificationOutbox`, factor it out exactly that way). Callers of `claim()` / `claimDigest()` are unaffected: without `skipReap` they reap as before. + +FROM → TO, for an implementer: + +```ts +// FROM +class MyOutbox implements INotificationOutbox { + async claim(opts: ClaimOptions) { await this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); /* … */ } +} +// TO +class MyOutbox implements INotificationOutbox { + async reap(opts: ReapOptions) { await this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); } + async claim(opts: ClaimOptions) { if (!opts.skipReap) await this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); /* … */ } +} +``` + +Breaking ships as `minor` per the launch-window convention (`scripts/check-changeset-no-major.mjs`). + + diff --git a/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts b/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts index baf3b7c13d..281829c059 100644 --- a/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts +++ b/packages/services/service-messaging/src/delivery-claim-tenant-audit.integration.test.ts @@ -219,4 +219,30 @@ describe('sys_notification_delivery — the dispatcher claim path is a classifie await controlUnscopedUpdateMany(DELIVERY_OBJECT); }); + + it("reap() recovers a crashed node's claims in every organization, without a finding", async () => { + const outbox = new SqlNotificationOutbox(engine as any, { partitionCount: 1 }); + const idA = await outbox.enqueue({ + notificationId: 'n_a', recipientId: 'u_a', channel: 'inbox', organizationId: 'org_a', payload: {}, + } as any); + const idB = await outbox.enqueue({ + notificationId: 'n_b', recipientId: 'u_b', channel: 'inbox', organizationId: 'org_b', payload: {}, + } as any); + // A node claims both rows and dies: its claim is stamped ten minutes ago. + await outbox.claim({ nodeId: 'dead_node', limit: 10, claimTtlMs: 60_000, now: Date.now() - 10 * 60_000 }); + + // [#17610] The dispatcher's once-per-tick reap, on its own — a third + // predicate write on the claim path, classified by the same warrant. + await outbox.reap({ claimTtlMs: 60_000 }); + + expect(auditedUpdateMany(DELIVERY_OBJECT)).toBe(false); + // Both organizations' abandoned rows are back in the queue with the claim + // credential cleared — a per-organization reap would have stranded one. + const rows = await outbox.list(); + expect(rows.map((r) => `${r.id}:${r.organizationId}:${r.status}:${r.claimedBy ?? '-'}`).sort()).toEqual( + [`${idA}:org_a:pending:-`, `${idB}:org_b:pending:-`].sort(), + ); + + await controlUnscopedUpdateMany(DELIVERY_OBJECT); + }); }); diff --git a/packages/services/service-messaging/src/dispatcher-idle-backoff.test.ts b/packages/services/service-messaging/src/dispatcher-idle-backoff.test.ts new file mode 100644 index 0000000000..8ec2e0217a --- /dev/null +++ b/packages/services/service-messaging/src/dispatcher-idle-backoff.test.ts @@ -0,0 +1,256 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #17610 — the `NotificationDispatcher` loop backs off while the outbox is + * idle, and new work wakes it. + * + * The loop used to tick every `intervalMs` (500 ms) forever, whatever it found. + * Now each tick that claims nothing doubles the delay to the next one, capped at + * `maxIdleIntervalMs`; a tick that claims work snaps it back to `intervalMs`; + * and `wake()` — called by `MessagingService` whenever `emit()` enqueues + * deliveries — ticks at once. + * + * Every leg runs on vitest's fake timers, so "when did a tick start" is an + * exact reading rather than a sleep-and-hope. A tick is timestamped at its + * `reap()`, which the dispatcher issues exactly once, first, per tick. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { MemoryNotificationOutbox } from './memory-outbox.js'; +import { NotificationDispatcher } from './dispatcher.js'; +import { MessagingService } from './messaging-service.js'; +import type { MessagingChannel } from './channel.js'; +import type { EnqueueDeliveryInput, ReapOptions } from './outbox.js'; + +const BASE = 500; +const CAP = 30_000; +const MINUTE = 60_000; + +const silentLogger = { info() {}, warn() {}, error() {} }; + +/** A memory outbox that records the (fake) instant every tick starts. */ +class TickRecordingOutbox extends MemoryNotificationOutbox { + readonly tickStarts: number[] = []; + override async reap(opts: ReapOptions): Promise { + this.tickStarts.push(Date.now()); + return super.reap(opts); + } +} + +function row(recipientId: string): EnqueueDeliveryInput { + return { notificationId: `n_${recipientId}`, recipientId, channel: 'inbox', payload: { title: 'hi' } }; +} + +function gaps(ts: readonly number[]): number[] { + return ts.slice(1).map((t, i) => t - ts[i]); +} + +/** Resolves the next time `recipient` is sent to. */ +interface SendProbe { + readonly sent: string[]; + sentTo(recipient: string): Promise; + channel: MessagingChannel; +} + +function sendProbe(hold?: { recipient: string; until: Promise }): SendProbe { + const sent: string[] = []; + const waiters = new Map void>(); + return { + sent, + sentTo(recipient) { + if (sent.includes(recipient)) return Promise.resolve(); + return new Promise((resolve) => waiters.set(recipient, resolve)); + }, + channel: { + id: 'inbox', + async send(_ctx, delivery) { + sent.push(delivery.recipient); + waiters.get(delivery.recipient)?.(); + if (hold && delivery.recipient === hold.recipient) await hold.until; + return { ok: true }; + }, + }, + }; +} + +function setup(options: { maxIdleIntervalMs?: number; probe?: SendProbe } = {}) { + const outbox = new TickRecordingOutbox(1); + const probe = options.probe ?? sendProbe(); + const dispatcher = new NotificationDispatcher({ + nodeId: 'node-test', + outbox, + channels: { getChannel: (id) => (id === probe.channel.id ? probe.channel : undefined) }, + channelContext: { logger: silentLogger }, + partitionCount: 1, + intervalMs: BASE, + maxIdleIntervalMs: options.maxIdleIntervalMs, + }); + return { outbox, probe, dispatcher }; +} + +beforeEach(() => { vi.useFakeTimers(); }); +afterEach(() => { vi.useRealTimers(); }); + +describe('#17610 NotificationDispatcher — idle backoff', () => { + it('doubles the interval on every empty tick, up to maxIdleIntervalMs', async () => { + const { outbox, dispatcher } = setup({ maxIdleIntervalMs: CAP }); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(10 * MINUTE); + await dispatcher.stop(); + + const g = gaps(outbox.tickStarts); + expect(g.slice(0, 6)).toEqual([1_000, 2_000, 4_000, 8_000, 16_000, CAP]); + expect(g.slice(5).every((gap) => gap === CAP)).toBe(true); + // Ten idle minutes: 24 ticks, where the fixed 500 ms loop ran 1,201. + expect(outbox.tickStarts).toHaveLength(24); + }); + + it('defaults the ceiling to 30 s', async () => { + const { outbox, dispatcher } = setup(); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5 * MINUTE); + await dispatcher.stop(); + expect(Math.max(...gaps(outbox.tickStarts))).toBe(30_000); + }); + + it('a ceiling at or below intervalMs disables the backoff', async () => { + const { outbox, dispatcher } = setup({ maxIdleIntervalMs: BASE }); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5_000); + await dispatcher.stop(); + expect(outbox.tickStarts).toHaveLength(11); + expect(new Set(gaps(outbox.tickStarts))).toEqual(new Set([BASE])); + }); + + it('a row nobody announced waits at most one backed-off interval, and the tick that claims it resets the backoff', async () => { + const { outbox, probe, dispatcher } = setup({ maxIdleIntervalMs: CAP }); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5 * MINUTE); // fully backed off + + // Written without a wake() — as a row enqueued by another process is. + const enqueuedAt = Date.now(); + await outbox.enqueue(row('u1')); + await vi.advanceTimersByTimeAsync(CAP + BASE); + await dispatcher.stop(); + + expect(probe.sent).toEqual(['u1']); + const claimedAt = outbox.tickStarts.findIndex((t) => t >= enqueuedAt); + // The latency bound #17610 trades for the idle savings: one ceiling. + expect(outbox.tickStarts[claimedAt] - enqueuedAt).toBeLessThanOrEqual(CAP); + // …and work found means the very next tick is back on the base interval. + expect(outbox.tickStarts[claimedAt + 1] - outbox.tickStarts[claimedAt]).toBe(BASE); + }); +}); + +describe('#17610 NotificationDispatcher — wake()', () => { + it('ticks immediately, not at the next backed-off slot, and restarts from intervalMs', async () => { + const { outbox, probe, dispatcher } = setup({ maxIdleIntervalMs: CAP }); + dispatcher.start(); + // 5 min 10 s: between two backed-off ticks, the next one 20 s away. + await vi.advanceTimersByTimeAsync(5 * MINUTE + 10_000); + const ticksBefore = outbox.tickStarts.length; + + await outbox.enqueue(row('u1')); + dispatcher.wake(); + // No timer time passes here — only the woken tick's own promise chain. + await probe.sentTo('u1'); + expect(outbox.tickStarts).toHaveLength(ticksBefore + 1); + expect(outbox.tickStarts[outbox.tickStarts.length - 1]).toBe(Date.now()); + + await vi.advanceTimersByTimeAsync(BASE); + await dispatcher.stop(); + expect(outbox.tickStarts).toHaveLength(ticksBefore + 2); + expect(outbox.tickStarts[ticksBefore + 1] - outbox.tickStarts[ticksBefore]).toBe(BASE); + }); + + it('wakes during a running tick collapse into ONE follow-up tick, run the moment it settles', async () => { + let release!: () => void; + const until = new Promise((resolve) => { release = resolve; }); + const probe = sendProbe({ recipient: 'u1', until }); + const { outbox, dispatcher } = setup({ maxIdleIntervalMs: CAP, probe }); + + await outbox.enqueue(row('u1')); + dispatcher.start(); + await probe.sentTo('u1'); // tick 1 is now parked inside u1's send + + // u2 lands in the partition tick 1 has already claimed past. + await outbox.enqueue(row('u2')); + dispatcher.wake(); + dispatcher.wake(); + dispatcher.wake(); + expect(outbox.tickStarts).toHaveLength(1); // never two ticks at once + + release(); + await probe.sentTo('u2'); + // Exactly one follow-up, at the same instant tick 1 settled. + expect(outbox.tickStarts).toHaveLength(2); + expect(outbox.tickStarts[1]).toBe(outbox.tickStarts[0]); + + // Three wakes did not queue three ticks: the next one is a base interval out. + await vi.advanceTimersByTimeAsync(BASE - 1); + expect(outbox.tickStarts).toHaveLength(2); + await dispatcher.stop(); + }); + + it('stop() cancels the pending backed-off tick, and a wake() after stop is a no-op', async () => { + const { outbox, dispatcher } = setup({ maxIdleIntervalMs: CAP }); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(MINUTE); + await dispatcher.stop(); + const ticksAtStop = outbox.tickStarts.length; + + dispatcher.wake(); + await vi.advanceTimersByTimeAsync(10 * MINUTE); + expect(outbox.tickStarts).toHaveLength(ticksAtStop); + }); +}); + +describe('#17610 MessagingService.emit() wakes the dispatcher its outbox is wired to', () => { + function wiredStack() { + const outbox = new TickRecordingOutbox(1); + const probe = sendProbe(); + const service = new MessagingService({ logger: silentLogger }); + service.registerChannel(probe.channel); + const dispatcher = new NotificationDispatcher({ + nodeId: 'node-test', + outbox, + channels: service, + channelContext: { logger: silentLogger }, + partitionCount: 1, + intervalMs: BASE, + maxIdleIntervalMs: CAP, + }); + // The seam MessagingServicePlugin wires. + service.setOutbox(outbox, { onEnqueued: () => dispatcher.wake() }); + return { outbox, probe, service, dispatcher }; + } + + it('an emit() that enqueues a delivery is sent at once by a backed-off dispatcher', async () => { + const { outbox, probe, service, dispatcher } = wiredStack(); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5 * MINUTE + 10_000); + const ticksBefore = outbox.tickStarts.length; + + const result = await service.emit({ topic: 'deal.won', audience: ['user_1'], payload: { title: 'Won' } }); + expect(result.enqueued).toBe(1); + + await probe.sentTo('user_1'); + expect(outbox.tickStarts).toHaveLength(ticksBefore + 1); + expect(outbox.tickStarts[outbox.tickStarts.length - 1]).toBe(Date.now()); + await dispatcher.stop(); + }); + + it('an emit() that enqueues nothing does not wake it', async () => { + const { outbox, service, dispatcher } = wiredStack(); + dispatcher.start(); + await vi.advanceTimersByTimeAsync(5 * MINUTE + 10_000); + const ticksBefore = outbox.tickStarts.length; + + const result = await service.emit({ topic: 'deal.won', audience: [], payload: { title: 'Won' } }); + expect(result.enqueued).toBe(0); + + await vi.advanceTimersByTimeAsync(0); + expect(outbox.tickStarts).toHaveLength(ticksBefore); + await dispatcher.stop(); + }); +}); diff --git a/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts b/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts new file mode 100644 index 0000000000..6626df1b3d --- /dev/null +++ b/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts @@ -0,0 +1,183 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #17610 — what an IDLE `NotificationDispatcher` tick costs the store, and that + * cutting it keeps every delivery guarantee. + * + * ## The measurement this pins + * + * Before #17610 one tick over an EMPTY outbox issued 32 statements with the + * default 8 partitions: `claim()` and `claimDigest()` each opened with the + * environment-wide visibility-timeout reap, in every partition — 16 identical + * UPDATEs — plus one candidate SELECT each (16). On remote Turso every + * statement is an HTTP round trip, and the loop never stopped. The reap now + * runs once per tick, so an idle tick is `1 + 2 × partitionCount` = 17. + * + * ## Why a real engine + * + * The count is taken on the `IDataEngine` the outbox talks to — the boundary + * this package controls — over a real `ObjectQL` + `SqlDriver` (better-sqlite3 + * `:memory:`), so every counted call is a statement the production outbox + * really issues. On this harness one engine call is one SQL statement (measured + * with the driver's query events while writing this file: 32 and 32 before, + * 17 and 17 after). + * + * ## The vacuity traps closed here + * + * - **An upper bound alone is satisfied by a dispatcher that does nothing.** + * The reap count is pinned EXACTLY — once per tick, because zero would + * strand a crashed node's rows forever — and the legs below prove the very + * same harness claims, sends, collapses digests and recovers. + * - **"Reap once" could be bought by recovering less.** The recovery leg drives + * a real expired claim through ONE tick; its negative leg proves a claim + * that has not expired is left alone, so the TTL is still a floor. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js'; +import { NotificationDelivery } from './objects/notification-delivery.object.js'; +import { NotificationDispatcher } from './dispatcher.js'; +import type { MessagingChannel } from './channel.js'; + +/** The production default. */ +const PARTITIONS = 8; +const TICKS = 10; +const TTL = 60_000; + +let engine: ObjectQL; +let outbox: SqlNotificationOutbox; +/** Engine calls against the delivery table, by method. */ +let calls: { update: number; find: number; other: number }; +/** `recipient` of every send, in order. */ +let sent: string[]; + +const recordingInbox: MessagingChannel = { + id: 'inbox', + async send(_ctx, delivery) { + sent.push(delivery.recipient); + return { ok: true }; + }, +}; + +function dispatcher(): NotificationDispatcher { + return new NotificationDispatcher({ + nodeId: 'node-live', + outbox, + channels: { getChannel: (id) => (id === recordingInbox.id ? recordingInbox : undefined) }, + channelContext: { logger: { info() {}, warn() {}, error() {} } }, + partitionCount: PARTITIONS, + claimTtlMs: TTL, + intervalMs: 10_000, // ticks are driven manually + }); +} + +beforeEach(async () => { + const driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + engine = new ObjectQL(); + engine.registerDriver(driver, true); + await engine.init(); + engine.registry.registerObject(NotificationDelivery as any, '@objectstack/service-messaging'); + await engine.syncSchemas(); + + calls = { update: 0, find: 0, other: 0 }; + sent = []; + // Count on the engine instance the outbox holds, so the tally is of real + // outbox traffic rather than a stand-in's. + type EngineCall = (name: string, ...rest: unknown[]) => unknown; + const e = engine as unknown as Record; + for (const method of ['find', 'findOne', 'update', 'insert', 'delete', 'count'] as const) { + const orig = e[method].bind(engine); + e[method] = (name: string, ...rest: unknown[]) => { + if (name === DELIVERY_OBJECT) { + if (method === 'update') calls.update++; + else if (method === 'find') calls.find++; + else calls.other++; + } + return orig(name, ...rest); + }; + } + outbox = new SqlNotificationOutbox(engine as any, { partitionCount: PARTITIONS }); +}); + +afterEach(async () => { + try { await engine?.destroy(); } catch { /* noop */ } +}); + +describe('#17610 NotificationDispatcher — idle tick cost', () => { + it('an idle tick is 1 + 2 × partitionCount statements: ONE reap, then a claim and a digest probe per partition', async () => { + const d = dispatcher(); + for (let i = 0; i < TICKS; i++) await d.tick(); + + // On an empty outbox the only UPDATE a tick issues is the reap. It is + // environment-wide, so once per tick is all it can use — and exactly + // once, never zero: it is the only crash recovery there is. + expect(calls.update).toBe(TICKS); + // Before #17610: 2 × PARTITIONS reaps + 2 × PARTITIONS probes = 32 a tick. + expect(calls.update + calls.find + calls.other).toBeLessThanOrEqual(TICKS * (1 + 2 * PARTITIONS)); + }); + + it('rows enqueued after an idle stretch go out on the very next tick — every partition, digests collapsed', async () => { + const d = dispatcher(); + for (let i = 0; i < TICKS; i++) await d.tick(); + expect(sent).toEqual([]); + + // Enough distinct notifications that they hash across several of the 8 + // partitions: the next tick must drain all of them, wherever they landed. + const recipients = Array.from({ length: 16 }, (_, i) => `u${i}`); + for (const [i, recipientId] of recipients.entries()) { + await outbox.enqueue({ notificationId: `n${i}`, recipientId, channel: 'inbox', payload: { title: 't' } }); + } + // Two batched rows of ONE digest window: the digest probe must claim them + // whole and the dispatcher must send them as one message. + for (const notificationId of ['nd1', 'nd2']) { + await outbox.enqueue({ + notificationId, recipientId: 'u_digest', channel: 'inbox', payload: { title: notificationId }, + digestKey: 'u_digest|inbox|w1', + }); + } + const partitions = new Set((await outbox.list()).map((r) => r.partitionKey)); + expect(partitions.size).toBeGreaterThan(1); + + await d.tick(); + + expect([...sent].sort()).toEqual([...recipients, 'u_digest'].sort()); + const rows = await outbox.list(); + expect(rows).toHaveLength(18); + expect(rows.every((r) => r.status === 'success' && r.attempts === 1)).toBe(true); + }); + + it("recovers a crashed node's expired claim and delivers it within ONE tick", async () => { + await outbox.enqueue({ notificationId: 'n_crashed', recipientId: 'u1', channel: 'inbox', payload: { title: 't' } }); + // A node claims the row and dies: its claim was stamped TTL + 1 ms ago. + const [abandoned] = await outbox.claim({ + nodeId: 'node-crashed', limit: 10, claimTtlMs: TTL, now: Date.now() - TTL - 1, + }); + expect(abandoned?.status).toBe('in_flight'); + + await dispatcher().tick(); + + // The tick's single reap ran BEFORE its claims, so the same tick took + // the row back and delivered it. + expect(sent).toEqual(['u1']); + const [row] = await outbox.list(); + expect(row).toMatchObject({ status: 'success', attempts: 1 }); + }); + + it('leaves a claim that has NOT expired alone — the TTL is still a floor', async () => { + await outbox.enqueue({ notificationId: 'n_live', recipientId: 'u1', channel: 'inbox', payload: { title: 't' } }); + // Another node's claim, 5 s inside its visibility timeout. + await outbox.claim({ nodeId: 'node-busy', limit: 10, claimTtlMs: TTL, now: Date.now() - TTL + 5_000 }); + + await dispatcher().tick(); + + expect(sent).toEqual([]); + const [row] = await outbox.list(); + expect(row).toMatchObject({ status: 'in_flight', claimedBy: 'node-busy', attempts: 0 }); + }); +}); diff --git a/packages/services/service-messaging/src/plugin-enqueue-wakes-dispatcher.test.ts b/packages/services/service-messaging/src/plugin-enqueue-wakes-dispatcher.test.ts new file mode 100644 index 0000000000..3c4f12a81b --- /dev/null +++ b/packages/services/service-messaging/src/plugin-enqueue-wakes-dispatcher.test.ts @@ -0,0 +1,146 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #17610 — in the COMPOSED plugin, an `emit()` that enqueues deliveries wakes + * the notification dispatcher. + * + * The dispatcher backs off while the outbox is idle, which is only harmless + * because the ingress that writes delivery rows wakes it. That wiring lives in + * `MessagingServicePlugin` (`setOutbox(outbox, { onEnqueued })`), and every + * unit test of the dispatcher would stay green if a refactor dropped it — + * notifications would quietly start going out up to `maxIdleIntervalMs` late. + * So this boots the real plugin on a real engine and measures delivery. + * + * ## Why the verdict is deterministic, not a race + * + * The plugin boots with `dispatchIntervalMs` = 60 s: after its first tick at + * `kernel:ready`, no timer-driven tick can start for a minute. A delivery that + * completes inside the few seconds this test waits can only have been started + * by `wake()`. + * + * The NEGATIVE CONTROL proves that premise on the same boot: a row written + * straight into the outbox table — no `emit()`, so no wake — is still + * `pending` after a real wait. Without it, a dispatcher ticking fast for any + * unrelated reason would pass the positive leg vacuously. + */ + +import { describe, it, expect, afterEach } from 'vitest'; +import { ObjectKernel } from '@objectstack/core'; +import { ObjectQLPlugin } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import type { ObjectQL } from '@objectstack/objectql'; +import type { IDataEngine } from '@objectstack/spec/contracts'; +import { MessagingServicePlugin } from './messaging-service-plugin.js'; +import type { MessagingService } from './messaging-service.js'; +import { DELIVERY_OBJECT } from './sql-outbox.js'; +import { hashPartition } from './backoff.js'; + +/** No timer-driven tick inside the test's lifetime. */ +const DISPATCH_INTERVAL_MS = 60_000; +const PARTITIONS = 1; +/** Long enough that a ticking dispatcher would have drained the control row many times over. */ +const CONTROL_WAIT_MS = 500; +const DELIVERY_BUDGET_MS = 5_000; + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function until(predicate: () => boolean, budgetMs: number): Promise { + const deadline = Date.now() + budgetMs; + while (!predicate()) { + if (Date.now() > deadline) throw new Error(`condition not met within ${budgetMs} ms`); + await sleep(10); + } +} + +const openKernels: ObjectKernel[] = []; +const openDrivers: Array<{ disconnect?: () => Promise }> = []; + +afterEach(async () => { + // Kernels first, drivers second: the kernel's own teardown still wants a + // live driver to drain against. + while (openKernels.length) { + try { await openKernels.pop()?.shutdown(); } catch { /* already stopped */ } + } + while (openDrivers.length) { + try { await openDrivers.pop()?.disconnect?.(); } catch { /* noop */ } + } +}); + +async function bootMessagingKernel() { + const kernel = new ObjectKernel({ logger: { level: 'silent' } } as any); + openKernels.push(kernel); + + await kernel.use(new ObjectQLPlugin()); + await kernel.use( + new MessagingServicePlugin({ dispatchIntervalMs: DISPATCH_INTERVAL_MS, partitionCount: PARTITIONS }), + ); + await kernel.bootstrap(); + + const objectql = kernel.getService('objectql'); + const driver: any = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + await driver.connect(); + objectql.registerDriver(driver, true); + openDrivers.push(driver); + await objectql.syncSchemas(); + + return { + engine: kernel.getService('data'), + messaging: kernel.getService('messaging'), + }; +} + +async function statusOf(engine: IDataEngine, id: string): Promise { + const found = await engine.findOne(DELIVERY_OBJECT, { where: { id }, fields: ['status'] }); + return found?.status; +} + +describe('#17610 MessagingServicePlugin — emit() wakes the backed-off dispatcher', () => { + it('sends an emitted notification at once, while a row nobody announced stays pending', async () => { + const { engine, messaging } = await bootMessagingKernel(); + const sent: string[] = []; + messaging.registerChannel({ + id: 'probe', + async send(_ctx, delivery) { + sent.push(delivery.recipient); + return { ok: true }; + }, + }); + + // NEGATIVE CONTROL — a ready `pending` row written straight into the + // outbox table, the way a process with no dispatcher of its own would. + const now = new Date(); + await engine.insert(DELIVERY_OBJECT, { + id: 'dlv_unannounced', + notification_id: 'evt_unannounced', + recipient_id: 'user_quiet', + channel: 'probe', + payload: { title: 'quiet' }, + partition_key: hashPartition('evt_unannounced', PARTITIONS), + status: 'pending', + attempts: 0, + created_at: now, + updated_at: now, + }); + await sleep(CONTROL_WAIT_MS); + expect(await statusOf(engine, 'dlv_unannounced')).toBe('pending'); + expect(sent).toEqual([]); + + // The ingress: emit() enqueues and wakes. The woken tick drains the whole + // partition, so the unannounced row goes out with it. + const result = await messaging.emit({ + topic: 'wake.probe', + audience: ['user_loud'], + channels: ['probe'], + payload: { title: 'loud' }, + }); + expect(result.enqueued).toBe(1); + + await until(() => sent.length === 2, DELIVERY_BUDGET_MS); + expect([...sent].sort()).toEqual(['user_loud', 'user_quiet']); + expect(await statusOf(engine, 'dlv_unannounced')).toBe('success'); + }, 20_000); +}); From f368fd0aac872d071919a9fb6f01f23cc5c8e3b8 Mon Sep 17 00:00:00 2001 From: Jack Zhuang <50353452+hotlong@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:10:56 +0800 Subject: [PATCH 3/4] fix(service-messaging): make outbox reap() an optional capability the dispatcher probes for A store without reap() keeps working: its claims keep reaping as before. Claude-Session: https://claude.ai/code/session_c5c0ce54-bb9c-478c-9e5b-cf44b80d4569 Co-authored-by: Claude --- ...17610-notification-dispatcher-idle-cost.md | 20 +------------ .../dispatcher-idle-cost.integration.test.ts | 28 +++++++++++++++++-- .../service-messaging/src/dispatcher.ts | 25 ++++++++++++----- .../services/service-messaging/src/outbox.ts | 6 +++- 4 files changed, 50 insertions(+), 29 deletions(-) diff --git a/.changeset/17610-notification-dispatcher-idle-cost.md b/.changeset/17610-notification-dispatcher-idle-cost.md index be4cb94ebd..90466f3bb0 100644 --- a/.changeset/17610-notification-dispatcher-idle-cost.md +++ b/.changeset/17610-notification-dispatcher-idle-cost.md @@ -14,22 +14,4 @@ **Latency bound.** A notification emitted in the process that runs the dispatcher goes out on the tick `wake()` starts, no later than before. While idle, work nobody announces is noticed within one backed-off interval, at most `maxIdleIntervalMs` (30 s by default): a deferred delivery coming due (retry schedule, quiet hours, digest window), a row enqueued by a process that does not run this dispatcher, and a crashed node's expired claim (recovered within `claimTtlMs` + `maxIdleIntervalMs`). Set `dispatchMaxIdleIntervalMs` to `dispatchIntervalMs` to keep the fixed interval. -**BREAKING (interface member added):** `INotificationOutbox` gains `reap(opts: ReapOptions)`, and `ClaimOptions` gains an optional `skipReap`. A custom outbox implementation must add `reap()` — the visibility-timeout recovery it already runs at the top of `claim()`, as a method of its own (both built-in stores, `SqlNotificationOutbox` and `MemoryNotificationOutbox`, factor it out exactly that way). Callers of `claim()` / `claimDigest()` are unaffected: without `skipReap` they reap as before. - -FROM → TO, for an implementer: - -```ts -// FROM -class MyOutbox implements INotificationOutbox { - async claim(opts: ClaimOptions) { await this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); /* … */ } -} -// TO -class MyOutbox implements INotificationOutbox { - async reap(opts: ReapOptions) { await this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); } - async claim(opts: ClaimOptions) { if (!opts.skipReap) await this.reapExpired(opts.now ?? Date.now(), opts.claimTtlMs); /* … */ } -} -``` - -Breaking ships as `minor` per the launch-window convention (`scripts/check-changeset-no-major.mjs`). - - +**Contract additions — all optional, nothing to change on upgrade.** `INotificationOutbox` gains an optional `reap(opts: ReapOptions)` — the visibility-timeout recovery `claim()` / `claimDigest()` already open with, as a method of its own — and `ClaimOptions` gains an optional `skipReap`. Both built-in stores (`SqlNotificationOutbox`, `MemoryNotificationOutbox`) implement them. A custom outbox without `reap()` keeps working as it is: the dispatcher probes for the method and, when it is absent, lets each claim reap as before — correct, at the old per-claim cost; implementing `reap()` and honouring `skipReap` is what earns the once-per-tick cost. Direct callers of `claim()` / `claimDigest()` are unaffected: without `skipReap` they reap exactly as before. Also new: `NotificationDispatcher.wake()`, the dispatcher's `maxIdleIntervalMs` option, and `MessagingService.setOutbox`'s optional second argument. diff --git a/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts b/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts index 6626df1b3d..d54d192165 100644 --- a/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts +++ b/packages/services/service-messaging/src/dispatcher-idle-cost.integration.test.ts @@ -40,6 +40,7 @@ import { SqlNotificationOutbox, DELIVERY_OBJECT } from './sql-outbox.js'; import { NotificationDelivery } from './objects/notification-delivery.object.js'; import { NotificationDispatcher } from './dispatcher.js'; import type { MessagingChannel } from './channel.js'; +import type { INotificationOutbox } from './outbox.js'; /** The production default. */ const PARTITIONS = 8; @@ -61,10 +62,10 @@ const recordingInbox: MessagingChannel = { }, }; -function dispatcher(): NotificationDispatcher { +function dispatcher(store: INotificationOutbox = outbox): NotificationDispatcher { return new NotificationDispatcher({ nodeId: 'node-live', - outbox, + outbox: store, channels: { getChannel: (id) => (id === recordingInbox.id ? recordingInbox : undefined) }, channelContext: { logger: { info() {}, warn() {}, error() {} } }, partitionCount: PARTITIONS, @@ -180,4 +181,27 @@ describe('#17610 NotificationDispatcher — idle tick cost', () => { const [row] = await outbox.list(); expect(row).toMatchObject({ status: 'in_flight', claimedBy: 'node-busy', attempts: 0 }); }); + + it('an outbox without reap() keeps working: its claims keep reaping, and an expired claim is still recovered', async () => { + // The shape of a store written before `reap()` existed: the same SQL + // store underneath, with that one method not exposed. + const skipReapSeen: unknown[] = []; + const legacy: INotificationOutbox = { + enqueue: (input) => outbox.enqueue(input), + claim: (opts) => { skipReapSeen.push(opts.skipReap); return outbox.claim(opts); }, + claimDigest: (opts) => { skipReapSeen.push(opts.skipReap); return outbox.claimDigest(opts); }, + ack: (claimed, result) => outbox.ack(claimed, result), + list: (filter) => outbox.list(filter), + }; + await outbox.enqueue({ notificationId: 'n_legacy', recipientId: 'u1', channel: 'inbox', payload: { title: 't' } }); + await outbox.claim({ nodeId: 'node-crashed', limit: 10, claimTtlMs: TTL, now: Date.now() - TTL - 1 }); + + await dispatcher(legacy).tick(); + + // No claim was told to skip the reap it is now the only source of… + expect(skipReapSeen).toHaveLength(2 * PARTITIONS); + expect(skipReapSeen.every((skip) => skip !== true)).toBe(true); + // …so the expired claim is still recovered and delivered in the same tick. + expect(sent).toEqual(['u1']); + }); }); diff --git a/packages/services/service-messaging/src/dispatcher.ts b/packages/services/service-messaging/src/dispatcher.ts index 1ada48da55..bcd2b32267 100644 --- a/packages/services/service-messaging/src/dispatcher.ts +++ b/packages/services/service-messaging/src/dispatcher.ts @@ -267,22 +267,32 @@ export class NotificationDispatcher { // rows, and an ack whose claim was reaped matches nothing (#11859) — while // the per-claim reap, run under partition p's lock, was already rewriting // rows in every other partition. - await this.opts.outbox.reap({ claimTtlMs: this.opts.claimTtlMs }); + // + // `reap` is optional on the outbox contract, so a store written before it + // keeps working: without it every claim keeps reaping as it always did — + // correct, at the per-claim cost. + const { outbox } = this.opts; + let reapedForTick = false; + if (outbox.reap) { + await outbox.reap({ claimTtlMs: this.opts.claimTtlMs }); + reapedForTick = true; + } const count = this.opts.partitionCount; const offset = stableNodeOffset(this.opts.nodeId, count); let claimed = 0; for (let step = 0; step < count; step++) { - claimed += await this.runPartition((offset + step) % count); + claimed += await this.runPartition((offset + step) % count, reapedForTick); } return claimed; } /** * Claim and send within one partition's lock. Resolves to the number of rows - * claimed — 0 when another node holds the lock. + * claimed — 0 when another node holds the lock. `skipReap` is true when this + * tick already ran the outbox's `reap()`. */ - private async runPartition(index: number): Promise { + private async runPartition(index: number, skipReap: boolean): Promise { const handle = await this.opts.cluster.lock.acquire(`notify.dispatcher.partition.${index}`, { ttlMs: this.opts.lockTtlMs, waitMs: 0, @@ -294,8 +304,9 @@ export class NotificationDispatcher { limit: this.opts.batchSize, partition: { index, count: this.opts.partitionCount }, claimTtlMs: this.opts.claimTtlMs, - // [#17610] Reaped once for the whole tick in runTick(). - skipReap: true, + // [#17610] Reaped once for the whole tick in runTick(), when the + // outbox has a reap() to run. + skipReap, }); if (claimed.length > 0) { await handle.renew?.(this.opts.lockTtlMs); @@ -313,7 +324,7 @@ export class NotificationDispatcher { limit: this.opts.batchSize, partition: { index, count: this.opts.partitionCount }, claimTtlMs: this.opts.claimTtlMs, - skipReap: true, + skipReap, }); if (digestRows.length > 0) { await handle.renew?.(this.opts.lockTtlMs); diff --git a/packages/services/service-messaging/src/outbox.ts b/packages/services/service-messaging/src/outbox.ts index 697d9916af..60f4e0dbe1 100644 --- a/packages/services/service-messaging/src/outbox.ts +++ b/packages/services/service-messaging/src/outbox.ts @@ -241,8 +241,12 @@ export interface INotificationOutbox { * Safe at any moment and from any number of nodes: it moves only rows already * past their timeout, {@link claim} takes only `pending` rows, and an * {@link ack} whose claim was reaped matches nothing and is refused (#11859). + * + * Optional, so an outbox written before it keeps working unchanged: the + * dispatcher probes for it and, when it is absent, lets every claim reap as + * before — correct, at the per-claim cost. Both built-in stores implement it. */ - reap(opts: ReapOptions): Promise; + reap?(opts: ReapOptions): Promise; claim(opts: ClaimOptions): Promise; /** * Record the outcome of ONE dispatch attempt on a row this caller claimed, From ff0ba83a2555402db3214ca7082b0824acffbde1 Mon Sep 17 00:00:00 2001 From: Jack Zhuang <50353452+hotlong@users.noreply.github.com> Date: Fri, 11 Sep 2026 12:23:22 +0800 Subject: [PATCH 4/4] docs(permissions): re-measure the tenant-audit write-call-site census (223 -> 222) SqlNotificationOutbox's two inline reap UPDATEs became one reapExpired() helper, so the census counts one fewer write call site. Claude-Session: https://claude.ai/code/session_c5c0ce54-bb9c-478c-9e5b-cf44b80d4569 Co-authored-by: Claude --- .../docs/permissions/tenant-audit-census.mdx | 34 +++++++++---------- ...08-tenant-audit-write-call-sites.counts.md | 16 ++++----- 2 files changed, 25 insertions(+), 25 deletions(-) diff --git a/content/docs/permissions/tenant-audit-census.mdx b/content/docs/permissions/tenant-audit-census.mdx index 34614af68f..611e972c22 100644 --- a/content/docs/permissions/tenant-audit-census.mdx +++ b/content/docs/permissions/tenant-audit-census.mdx @@ -98,7 +98,7 @@ are reported as `undecidable` rather than assumed either way. The same holds twice over for the context. An options argument spelled as a literal can be read; one spelled `options`, `{ ...opts }`, or handed through a -forwarding shim cannot, and **67 of the 223 sites are spelled that way**. A +forwarding shim cannot, and **66 of the 222 sites are spelled that way**. A context resolved from an inline literal or a local `const` can be tested for `isSystem`; one arriving from a helper call cannot. @@ -126,8 +126,8 @@ now **0**: nothing on this surface threads a context that provably lacks the fla **"No tenant context" counted sites it had not read.** An options argument the walker could not parse was folded into the same bucket as one it had read and -found empty. That published **84 sites "carrying no tenant context at all"** -when 17 said so and 67 were simply unread — an over-claim in the *alarming* +found empty. That published **83 sites "carrying no tenant context at all"** +when 17 said so and 66 were simply unread — an over-claim in the *alarming* direction, on the very figure this page tells other cards to cite. `carries` is now three-valued, and an unreadable argument can never contribute to the provable count. @@ -147,10 +147,10 @@ reproduce them. Where it disagrees, it disagrees on the page: | carried figure | where it survives | this census | | :--- | :--- | ---: | -| 175 write call sites | quoted in the merged changeset | **223** | +| 175 write call sites | quoted in the merged changeset | **222** | | 24 carrying no tenant context | quoted in the merged changeset | **9** provable and tenancy-enabled; **32** more whose options argument is unreadable | -| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **149 of 223** decidable, **74** undecidable | -| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 105 decidably elevated, 0 decidably not, 101 undecidable | +| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **149 of 222** decidable, **73** undecidable | +| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 105 decidably elevated, 0 decidably not, 100 undecidable | | 141 and 132, two independent re-derivations | the card that filed this work | — | **The differences are not reconciled, and deliberately so.** The old census's @@ -167,11 +167,11 @@ would report a smaller number and would not say so. The fourth row is the one worth flagging to anyone citing it. **The 135 / 77% figure has no surviving corroboration anywhere in the tree.** This census reads -105 of 223 (47%) as decidably elevated, with 101 more whose elevation is a +105 of 222 (47%) as decidably elevated, with 100 more whose elevation is a run-time fact — so the claim is neither confirmed nor refuted, and the honest answer is that a static reading cannot settle it. -⇒ **Cite `9 / 223`, and say what it is**: the sites whose options argument was +⇒ **Cite `9 / 222`, and say what it is**: the sites whose options argument was READ and holds no tenant context, against a decidably tenancy-enabled object. That is the control's provable yield surface. ⛔ Do not cite it as "the sites without tenant context" — **32 further sites** have an options argument this @@ -183,23 +183,23 @@ cannot read, and they are neither in nor out. | what | count | | :--- | ---: | -| write call sites on the application surface | **223** | +| write call sites on the application surface | **222** | | …whose object name is statically decidable | 149 | -| …whose object name is chosen at run time | 74 | +| …whose object name is chosen at run time | 73 | | …against an object with tenancy ENABLED | 149 | | …against an object that declares tenancy off | 0 | | threading a tenant context | 139 | | PROVABLY carrying none (options read, no context key) | **17** | | …of those, against a decidably tenancy-enabled object | **9** | -| options argument UNREADABLE — may or may not carry one | 67 | +| options argument UNREADABLE — may or may not carry one | 66 | | …of those, against a decidably tenancy-enabled object | 32 | | threading a decidably ELEVATED (`isSystem`) context | 105 | | threading a context that is decidably NOT elevated | 0 | -| threading a context whose elevation is a run-time fact | 101 | +| threading a context whose elevation is a run-time fact | 100 | | how the instrument reached the site | count | | :--- | ---: | -| receiver carried a readable engine type | 178 | +| receiver carried a readable engine type | 177 | | receiver erased, placed by the object NAME | 19 | | receiver erased, placed by an `object: string` PARAMETER | 15 | | receiver erased, placed by an `UNTYPED_RECEIVERS` row | 11 | @@ -207,7 +207,7 @@ cannot read, and they are neither in nor out. | object name spelled inline | 109 | | object name spelled through a `const` | 40 | | object name is an `object: string` parameter | 19 | -| object name is some other run-time expression | 55 | +| object name is some other run-time expression | 54 | The corpus walked is every tracked non-test source under `packages/services/` and `packages/plugins/`; calls to a same-named method on something that is not @@ -224,12 +224,12 @@ holds still. They are required to be HERE and to say WHEN they were true; their values are not compared. The reasoning, and the measurement behind it, are in `scripts/check-tenant-audit-census.mjs`. -Measured on 2026-09-10 at `638d2b544`. +Measured on 2026-09-11 at `f368fd0aa`. | corpus scale (not enforced) | count | | :--- | ---: | -| tracked non-test sources scanned | 562 | -| engine-shaped types recognised | 58 | +| tracked non-test sources scanned | 563 | +| engine-shaped types recognised | 59 | | declared objects in the registry | 300 | | same-named calls subtracted as non-engine | 137 | diff --git a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md index d8c65d4c87..7f3bed3901 100644 --- a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md +++ b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md @@ -29,19 +29,19 @@ silent, and `node scripts/tenant-audit-census.mjs --write` is the resolution. | Measure | Value | |---|---:| -| Write call sites | 223 | +| Write call sites | 222 | | Object name statically decidable | 149 | -| Object name chosen at run time | 74 | +| Object name chosen at run time | 73 | | Against a tenancy-enabled object | 149 | | Against an object declaring tenancy off | 0 | | Threading a tenant context | 139 | | Provably carrying none | 17 | | …and decidably tenancy-enabled | 9 | -| Options argument unreadable | 67 | +| Options argument unreadable | 66 | | …and decidably tenancy-enabled | 32 | | Threading a decidably elevated context | 105 | | Threading a decidably non-elevated context | 0 | -| Threading a context of undecidable elevation | 101 | +| Threading a context of undecidable elevation | 100 | ## Corpus scale — present and dated, ⛔ NOT enforced @@ -52,12 +52,12 @@ holds still. They are required to be HERE and to say WHEN they were true; their values are not compared. The reasoning, and the measurement behind it, are in `scripts/check-tenant-audit-census.mjs`. -Measured on 2026-09-10 at `638d2b544`. +Measured on 2026-09-11 at `f368fd0aa`. | corpus scale (not enforced) | count | | :--- | ---: | -| tracked non-test sources scanned | 562 | -| engine-shaped types recognised | 58 | +| tracked non-test sources scanned | 563 | +| engine-shaped types recognised | 59 | | declared objects in the registry | 300 | | same-named calls subtracted as non-engine | 137 | @@ -181,7 +181,7 @@ Measured on 2026-09-10 at `638d2b544`. | `packages/services/service-messaging/src/sql-http-outbox.ts` | `insert` | `this.objectName` | undecidable | options unreadable | 1 | | `packages/services/service-messaging/src/sql-http-outbox.ts` | `update` | `this.objectName` | undecidable | options unreadable | 4 | | `packages/services/service-messaging/src/sql-outbox.ts` | `insert` | `this.objectName` | undecidable | options unreadable | 1 | -| `packages/services/service-messaging/src/sql-outbox.ts` | `update` | `this.objectName` | undecidable | options unreadable | 5 | +| `packages/services/service-messaging/src/sql-outbox.ts` | `update` | `this.objectName` | undecidable | options unreadable | 4 | | `packages/services/service-queue/src/db-queue-adapter.ts` | `delete` | `sys_job_queue` | enabled | context, elevation undecidable | 2 | | `packages/services/service-queue/src/db-queue-adapter.ts` | `insert` | `sys_job_queue` | enabled | context, elevation undecidable | 1 | | `packages/services/service-queue/src/db-queue-adapter.ts` | `update` | `sys_job_queue` | enabled | context, elevation undecidable | 6 |