From 6e0c47910d0afe1923f5b226718b4b152550ded7 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 08:26:36 +0000 Subject: [PATCH 1/2] =?UTF-8?q?fix(service-queue,platform-objects):=20boun?= =?UTF-8?q?d=20sys=5Fjob=5Fqueue=20=E2=80=94=20completed=20rows=20expire?= =?UTF-8?q?=20on=20a=20declared=20ADR-0057=20retention=20(#5179)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DbQueueAdapter marked delivered messages `completed` and nothing ever touched the row again: `purge()` had zero production callers, `purgeFailed()` is a manual dead-letter API, and the object declared no lifecycle policy — so the queue table only ever grew (one permanent row per queued email since #5160). sys_job_queue now declares `lifecycle: { class: 'transient', retention: { maxAge: '7d', onlyWhen: { status: 'completed' } } }`, enforced by the one platform-owned LifecycleService reaper (ADR-0057 §3.3) on its existing hourly sweep — no new sweeper in the adapter's poll loop, no new configuration. `pending`/`running` (live work) and `failed`/`dlq` (the dead-letter queue) are never swept at any age. The dedup window becomes an enforced invariant rather than a coincidence: publish dedups terminal rows by `created_at` against `idempotencyWindowMs`, the reaper cuts off on the same axis, and DbQueueAdapter now reads the declared window (`completedRetentionWindowMs()`) and throws at construction if the idempotency window is configured longer than it. `class: 'transient'` and not `telemetry`: per ADR-0057 §3.6 a telemetry/event/audit class relocates the table to the dedicated `telemetry` datasource wherever one is registered, and moving a live work queue's storage would be a migration, not a cleanup. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017MCKJaEomEqg4tvz4SzdNd --- .changeset/job-queue-completed-retention.md | 52 +++ .../src/audit/sys-job-queue.object.ts | 53 +++ packages/services/service-queue/README.md | 23 ++ packages/services/service-queue/src/common.ts | 33 ++ .../service-queue/src/db-queue-adapter.ts | 74 ++++- packages/services/service-queue/src/index.ts | 2 +- .../src/job-queue-retention.test.ts | 307 ++++++++++++++++++ 7 files changed, 541 insertions(+), 3 deletions(-) create mode 100644 .changeset/job-queue-completed-retention.md create mode 100644 packages/services/service-queue/src/job-queue-retention.test.ts diff --git a/.changeset/job-queue-completed-retention.md b/.changeset/job-queue-completed-retention.md new file mode 100644 index 0000000000..752976f664 --- /dev/null +++ b/.changeset/job-queue-completed-retention.md @@ -0,0 +1,52 @@ +--- +"@objectstack/platform-objects": minor +"@objectstack/service-queue": minor +--- + +fix(service-queue): `sys_job_queue` no longer grows forever — `completed` rows expire on a declared 7-day retention (#5179) + +`DbQueueAdapter` marked a delivered message `status: 'completed'` and then +**nothing ever touched that row again**. `purge()` had zero production callers +(tests only), `purgeFailed()` is a manual dead-letter API, and the object +declared no lifecycle policy at all — so every queue delivery left a permanent +row, which since #5160 means one permanent row per queued email. + +`sys_job_queue` now declares an ADR-0057 policy and the platform +`LifecycleService` enforces it on its existing hourly sweep: + +```ts +lifecycle: { + class: 'transient', + retention: { maxAge: '7d', onlyWhen: { status: 'completed' } }, +} +``` + +**Only `completed` rows are swept.** `pending` / `running` are live work, and +`failed` / `dlq` are the dead-letter queue — they exist to wait for a human, so +they are never deleted automatically at any age. `listFailed()` / `replay()` / +`purgeFailed()` remain the only way a dead letter leaves the table. This is +also why the policy is `retention` (age + row filter) rather than a `ttl` on +`completed_at`: TTL has no row filter, and `dlq` rows stamp `completed_at` too. + +**No new configuration, and no new sweeper.** ADR-0057 §3.3 puts one reaper in +the platform rather than one per plugin — the same call the sibling +`sys_job_run` (30d) already makes. Any kernel with a data engine already runs +it, its per-sweep `[lifecycle] sweep: … ~N rows reaped` line now accounts for +this table too, and the window is overridable per environment through the +`lifecycle` settings namespace without touching code. + +**The dedup window is now an enforced invariant, not a coincidence.** Publish +dedups against a terminal row by comparing its `created_at` to +`idempotencyWindowMs` (default 24h), and the reaper cuts off on that same +`created_at` axis — so retention (7d) ≥ dedup window is what keeps "duplicate +publishes inside the window are suppressed" true. `DbQueueAdapter` reads the +declared window (new export `completedRetentionWindowMs()`) and **throws at +construction** if `idempotencyWindowMs` is configured longer than it, instead of +silently degrading into duplicate deliveries days later. If you raise +`idempotencyWindowMs` past 7 days, raise the object's declared retention (or the +`lifecycle` settings override) to match — the error message names both numbers. + +`class: 'transient'` is deliberate: `telemetry`/`event`/`audit` classes +relocate their table to the dedicated `telemetry` datasource wherever one is +registered (ADR-0057 §3.6), and moving a live work queue's storage would be a +migration, not a cleanup. diff --git a/packages/platform-objects/src/audit/sys-job-queue.object.ts b/packages/platform-objects/src/audit/sys-job-queue.object.ts index b808f9132c..f17171b875 100644 --- a/packages/platform-objects/src/audit/sys-job-queue.object.ts +++ b/packages/platform-objects/src/audit/sys-job-queue.object.ts @@ -22,6 +22,9 @@ import { ObjectSchema, Field } from '@objectstack/spec/data'; * Writers: `DbQueueAdapter` (publish/lease/complete/fail). * Readers: Studio DLQ view, ops dashboards, the adapter's worker loop. * + * Retention: `completed` rows are swept by the platform LifecycleService — + * see the `lifecycle` block below (#5179). + * * @namespace sys */ export const SysJobQueue = ObjectSchema.create({ @@ -31,6 +34,56 @@ export const SysJobQueue = ObjectSchema.create({ icon: 'inbox', isSystem: true, managedBy: 'engine-owned', + + /** + * [ADR-0057 §3.1/§3.3, #5179] The queue table only ever GREW: the adapter + * marks a delivered message `completed` and nothing ever touched the row + * again (`purge()` had zero production callers, `purgeFailed()` is a manual + * dead-letter API). Since #5160 that is one permanent row per email. + * + * Bounded declaratively rather than by a sweeper inside `DbQueueAdapter`: + * ADR-0057 §3.3 puts ONE reaper in the platform (`LifecycleService`), not N + * per-plugin ones — the same call the sibling `sys_job_run` already makes. + * That the writer is the adapter itself (never user data) is what makes an + * unattended delete safe here; the declaration is where an operator can see + * the window, and `lifecycle` settings can override it per environment + * without a code change. + * + * `onlyWhen: { status: 'completed' }` is the whole safety story: + * - `pending` / `running` are LIVE work — reaping them would drop + * undelivered messages; + * - `dlq` / `failed` are the dead-letter surface and exist precisely to + * wait for a human (`listFailed` / `replay` / `purgeFailed`), so they + * are never swept automatically, at any age. + * This is also why the policy is `retention` (age by `created_at` + row + * filter) and not `ttl` on `completed_at`: TTL has no row filter, and `dlq` + * rows stamp `completed_at` too — a TTL would eat the dead-letter queue. + * + * Window = 7d, and it MUST stay ≥ the adapter's idempotency window + * (`DbQueueAdapterOptions.idempotencyWindowMs`, default 24h): publish + * dedups against terminal rows by comparing `created_at` to that window + * (`db-queue-adapter.ts`), and the Reaper cuts off on the very same + * `created_at` axis — so a retention ≥ the dedup window means a row the + * dedup check still needs can never have been reaped, with no clock skew + * between the two rules. 7d gives a week of delivery history for debugging + * and 7× headroom over the default dedup window. `DbQueueAdapter` reads + * this declaration and refuses to start when the two are configured the + * wrong way round, so the invariant cannot drift apart silently. + * + * `class: 'transient'` ("workflow / ephemeral state" — ADR-0057 §3.1), not + * `telemetry`: this is live work state, not a log, and per §3.6 a + * `telemetry`/`event`/`audit` class RELOCATES the table to the dedicated + * `telemetry` datasource wherever one is registered. Moving a live queue's + * store is a migration, not a cleanup — `transient` deliberately stays on + * the primary. + */ + lifecycle: { + class: 'transient', + retention: { + maxAge: '7d', + onlyWhen: { status: 'completed' }, + }, + }, description: 'Durable job/message queue including dead letters', displayNameField: 'queue', nameField: 'queue', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) diff --git a/packages/services/service-queue/README.md b/packages/services/service-queue/README.md index c9f46044d8..720b32c67e 100644 --- a/packages/services/service-queue/README.md +++ b/packages/services/service-queue/README.md @@ -70,6 +70,29 @@ new QueueServicePlugin({ new QueueServicePlugin({ adapter: 'memory' }); ``` +### Retention — how `sys_job_queue` stays bounded + +Delivered messages are not kept forever. `sys_job_queue` declares an ADR-0057 +lifecycle policy and the platform `LifecycleService` (shipped with +`@objectstack/objectql`, armed on every kernel that has data) enforces it — no +configuration, no extra scheduler: + +| Row state | What happens | +|---|---| +| `completed` | deleted **7 days** after `created_at` | +| `pending` / `running` | never swept — live work | +| `failed` / `dlq` | never swept — the dead-letter queue waits for a human (`listFailed` / `replay` / `purgeFailed`) | + +Two consequences worth knowing: + +- **`idempotencyWindowMs` must not exceed the retention window.** Dedup against + a terminal message compares its `created_at` to that window, so a longer + setting would start accepting duplicates the moment the row was swept. The + `db` adapter throws at construction instead of degrading quietly. +- **The window is overridable per environment** through the `lifecycle` + settings namespace (`maxAge` per object), like every other ADR-0057 policy. + Keep it ≥ your idempotency window. + ## Service API Implements `IQueueService` from `@objectstack/spec/contracts`: diff --git a/packages/services/service-queue/src/common.ts b/packages/services/service-queue/src/common.ts index 74d139a5f0..f01767e7da 100644 --- a/packages/services/service-queue/src/common.ts +++ b/packages/services/service-queue/src/common.ts @@ -36,6 +36,39 @@ export function nowIso(clock?: JobClock): string { return (clock?.now() ?? new Date()).toISOString(); } +/** + * Milliseconds per ADR-0057 lifecycle duration unit. Mirrors + * `parseLifecycleDuration` in `@objectstack/objectql` (the canonical runtime + * consumer), reproduced here rather than imported because the queue adapters + * deliberately do not depend on the engine package — they duck-type + * {@link JobEngine} so they stay testable without booting a kernel. Both + * tables are fixed by the ADR (coarse operational bounds: `y` is 365 days), + * and `job-queue-retention.test.ts` pins this one against them. + */ +const LIFECYCLE_UNIT_MS: Record = { + h: 3_600_000, + d: 86_400_000, + w: 7 * 86_400_000, + y: 365 * 86_400_000, +}; + +/** + * Parse an ADR-0057 duration literal (`'6h'`, `'7d'`, `'12w'`, `'7y'`) into + * milliseconds. Throws on anything else: declarations reach this code already + * validated by `LifecycleSchema`, so a failure here is a broken declaration, + * not user input — and a queue that silently guessed a window would be exactly + * the silent behaviour #5179 is about. + */ +export function lifecycleDurationMs(literal: string): number { + const m = /^(\d+)(h|d|w|y)$/.exec(literal); + if (!m) { + throw new Error( + `[service-queue] invalid lifecycle duration literal '${literal}' — expected with unit h|d|w|y (e.g. '7d')`, + ); + } + return Number(m[1]) * LIFECYCLE_UNIT_MS[m[2]!]!; +} + export function parseJson(raw: unknown, fallback?: T): T | undefined { if (raw == null) return fallback; if (typeof raw === 'string') { diff --git a/packages/services/service-queue/src/db-queue-adapter.ts b/packages/services/service-queue/src/db-queue-adapter.ts index 53eab76c70..1afc51cbf6 100644 --- a/packages/services/service-queue/src/db-queue-adapter.ts +++ b/packages/services/service-queue/src/db-queue-adapter.ts @@ -7,11 +7,13 @@ import type { QueueMessageRecord, QueueHandler, } from '@objectstack/spec/contracts'; +import { SysJobQueue } from '@objectstack/platform-objects/audit'; import { SYSTEM_CTX, uid, nowIso, parseJson, + lifecycleDurationMs, type JobEngine, type JobClock, type JobLogger, @@ -19,6 +21,29 @@ import { const QUEUE_TABLE = 'sys_job_queue'; +/** + * How long a `completed` row survives before the platform Reaper deletes it. + * + * Read from the object's own ADR-0057 declaration + * (`sys_job_queue.lifecycle.retention`, #5179) instead of being a second + * number here: the declaration is what actually runs (LifecycleService sweeps + * every registered object hourly), so a copy in this file could only ever be + * a copy that drifts. A missing or unparseable declaration throws: the queue's + * dedup contract below is defined against this window, so "no window" is not a + * state the adapter can run in. + */ +export function completedRetentionWindowMs(): number { + const maxAge = SysJobQueue.lifecycle?.retention?.maxAge; + if (!maxAge) { + throw new Error( + '[service-queue] sys_job_queue no longer declares lifecycle.retention — DbQueueAdapter dedups against ' + + 'terminal rows by `created_at` window and relies on that declared retention to keep them (ADR-0057, #5179). ' + + 'Restore the declaration in @objectstack/platform-objects rather than sweeping the table from here.', + ); + } + return lifecycleDurationMs(maxAge); +} + export interface DbQueueAdapterOptions { /** Polling interval for the worker loop (ms, default 1000) */ pollIntervalMs?: number; @@ -26,7 +51,15 @@ export interface DbQueueAdapterOptions { batchSize?: number; /** Lease duration before another worker may reclaim (ms, default 30000) */ leaseMs?: number; - /** Idempotency window — how long the same key blocks re-publish (ms, default 24h) */ + /** + * Idempotency window — how long the same key blocks re-publish (ms, default 24h). + * + * Must not exceed `sys_job_queue`'s declared retention for `completed` rows + * ({@link completedRetentionWindowMs}, 7d): the window is evaluated against + * rows that are still in the table, so a longer window would silently start + * accepting duplicates as soon as the Reaper swept the row it dedups + * against. The constructor rejects that configuration (#5179). + */ idempotencyWindowMs?: number; /** Default maxAttempts when publish doesn't specify (default 3) */ defaultMaxAttempts?: number; @@ -52,6 +85,15 @@ interface RegisteredHandler { * Idempotency: publish suppresses duplicates within a configurable * window when `(queue, idempotencyKey)` is non-null. * + * Retention: this adapter does NOT sweep the table. `completed` rows are + * bounded by `sys_job_queue`'s declared ADR-0057 retention (7d, filtered to + * `status='completed'`), enforced by the one platform-owned + * `LifecycleService` reaper — see the object definition in + * `@objectstack/platform-objects` and {@link completedRetentionWindowMs}. + * `dlq`/`failed` rows are never swept; they are the dead-letter surface + * ({@link DbQueueAdapter.listFailed} / {@link DbQueueAdapter.replay} / + * {@link DbQueueAdapter.purgeFailed}). + * * Designed for SQLite and Postgres alike — uses CAS via WHERE-clauses, * not row-level locking. */ @@ -84,6 +126,25 @@ export class DbQueueAdapter implements IQueueService { autoStart: o.autoStart ?? true, workerId: o.workerId ?? uid('worker'), }; + + // [#5179] The dedup window only means anything while the row it dedups + // against still exists. `completed` rows now expire on the declared + // retention window, so an idempotency window LONGER than it would quietly + // degrade into "dedup for as long as the Reaper happens not to have run" — + // duplicate deliveries appearing days later, with nothing in any log. The + // two windows are ordered here, at construction, rather than tolerated at + // publish time: the fix is a config or declaration change, and both are + // named in the message. + const retentionMs = completedRetentionWindowMs(); + if (this.opts.idempotencyWindowMs > retentionMs) { + throw new Error( + `[service-queue] idempotencyWindowMs (${this.opts.idempotencyWindowMs}ms) exceeds the retention window ` + + `sys_job_queue declares for completed rows (${retentionMs}ms, lifecycle.retention.maxAge — ADR-0057). ` + + 'Terminal-row dedup is evaluated by `created_at` against that same window, so the longer setting would ' + + 'silently accept duplicates once a row is reaped. Lower idempotencyWindowMs, or raise the declared ' + + 'retention (both windows are measured from `created_at`).', + ); + } } // ── IQueueService ──────────────────────────────────────────────── @@ -96,7 +157,16 @@ export class DbQueueAdapter implements IQueueService { const opts = options ?? {}; const now = this.now(); - // Idempotency check + // Idempotency check. + // + // [#5179] This is the reason `sys_job_queue`'s retention is filtered and + // generous rather than aggressive: a terminal (`completed`/`dlq`) row + // blocks a re-publish only while its `created_at` is inside the + // idempotency window, so the row must SURVIVE that long. The declared + // retention (7d on `completed`, nothing on `dlq`) is measured on the very + // same `created_at` axis and is ≥ this window — enforced in the + // constructor — which makes "the reaper deleted a row the dedup check + // needed" unrepresentable rather than merely unlikely. if (opts.idempotencyKey) { const windowStart = new Date(now.getTime() - this.opts.idempotencyWindowMs).toISOString(); const existing = await this.engine.find(QUEUE_TABLE, { diff --git a/packages/services/service-queue/src/index.ts b/packages/services/service-queue/src/index.ts index 94a5cee63c..cdaae1de6b 100644 --- a/packages/services/service-queue/src/index.ts +++ b/packages/services/service-queue/src/index.ts @@ -4,6 +4,6 @@ export { QueueServicePlugin } from './queue-service-plugin.js'; export type { QueueServicePluginOptions } from './queue-service-plugin.js'; export { MemoryQueueAdapter } from './memory-queue-adapter.js'; export type { MemoryQueueAdapterOptions } from './memory-queue-adapter.js'; -export { DbQueueAdapter } from './db-queue-adapter.js'; +export { DbQueueAdapter, completedRetentionWindowMs } from './db-queue-adapter.js'; export type { DbQueueAdapterOptions } from './db-queue-adapter.js'; export type { JobEngine, JobClock, JobLogger } from './common.js'; diff --git a/packages/services/service-queue/src/job-queue-retention.test.ts b/packages/services/service-queue/src/job-queue-retention.test.ts new file mode 100644 index 0000000000..7ab9d7d590 --- /dev/null +++ b/packages/services/service-queue/src/job-queue-retention.test.ts @@ -0,0 +1,307 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +// #5179 — `sys_job_queue` only ever grew: the adapter marked a delivered +// message `completed` and nothing ever touched the row again. The fix is the +// object's own ADR-0057 `lifecycle.retention` declaration, swept by the ONE +// platform-owned Reaper (`LifecycleService`, @objectstack/objectql) — not a +// second sweeper inside the adapter's poll loop. +// +// What this file pins is therefore the CONTRACT the two sides share: +// 1. the declaration itself (window, and that it only ever names `completed`); +// 2. that the declared window can never be shorter than the adapter's +// idempotency window — the dedup rule reads terminal rows that the Reaper +// is allowed to delete, so ordering those two windows is what keeps +// "dedup inside the window" true; +// 3. that applying the declared policy repeatedly bounds the table while +// leaving live work (`pending`/`running`) and the dead-letter queue +// (`dlq`/`failed`) alone at ANY age. +// +// The Reaper's own semantics (that it merges `retention.onlyWhen` into the +// delete filter) are pinned upstream in +// `packages/objectql/src/lifecycle/lifecycle-service.test.ts` ("merges +// retention.onlyWhen into the reap filter (mixed tables, #2834)"); `sweep()` +// below is a faithful mirror of that same where-clause construction +// (`{ created_at: { $lt: cutoff }, ...onlyWhen }`, `multi: true`, system +// context) so this package can assert the consequences without depending on +// the engine. + +import { describe, it, expect, beforeEach } from 'vitest'; +import { SysJobQueue } from '@objectstack/platform-objects/audit'; +import { DbQueueAdapter, completedRetentionWindowMs } from './db-queue-adapter.js'; +import { lifecycleDurationMs } from './common.js'; + +const QUEUE_TABLE = 'sys_job_queue'; +const DEFAULT_IDEMPOTENCY_WINDOW_MS = 24 * 60 * 60 * 1000; +const HOUR = 60 * 60 * 1000; +const DAY = 24 * HOUR; + +type Row = Record; + +/** + * In-memory engine mirroring the two call shapes this test needs: + * - the adapter's `where:` equality find / `(table, {id, ...patch})` update + * (same contract as `db-queue-adapter.test.ts`); + * - the Reaper's `delete(table, { where, multi: true })` with a `$lt` + * operator, which the row-by-row `where.id` delete of the adapter's own + * `purge()` does not exercise. + */ +function makeFakeEngine() { + const tables = new Map(); + function matches(row: Row, where: Record): boolean { + for (const [k, v] of Object.entries(where)) { + if (v && typeof v === 'object' && !Array.isArray(v)) { + // NULL-safe like SQL: a row with no value never satisfies `$lt`. + if ('$lt' in v && (row[k] == null || !(String(row[k]) < String(v.$lt)))) return false; + if ('$in' in v && !(v.$in as unknown[]).includes(row[k])) return false; + continue; + } + if (row[k] !== v) return false; + } + return true; + } + return { + tables, + rows(): Row[] { + return tables.get(QUEUE_TABLE) ?? []; + }, + async find(table: string, opts: any = {}) { + 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; + }); + } + } + if (opts.offset) out = out.slice(opts.offset); + if (opts.limit) out = out.slice(0, opts.limit); + return out; + }, + async insert(table: string, data: Row) { + const t = tables.get(table) ?? []; + t.push({ ...data }); + tables.set(table, t); + return { id: data.id }; + }, + async update(table: string, patch: Row) { + 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) { + const t = tables.get(table) ?? []; + if (opts?.multi) { + const where = opts.where ?? {}; + const keep = t.filter((r) => !matches(r, where)); + tables.set(table, keep); + return t.length - keep.length; // drivers report a deleted count + } + const id = opts?.where?.id; + if (id == null) throw new Error('Delete requires an ID or options.multi=true'); + tables.set(table, t.filter((r) => r.id !== id)); + return { id }; + }, + }; +} + +/** Mutable clock — the adapter stamps `created_at` from it. */ +function makeClock(startMs: number) { + let ms = startMs; + return { + now: () => new Date(ms), + advance(byMs: number) { ms += byMs; }, + get ms() { return ms; }, + }; +} + +/** + * Apply `sys_job_queue`'s DECLARED retention exactly as `LifecycleService` + * would (single bulk delete, system context) and return the row count deleted. + * Reads the declaration rather than restating it, so a change to the object + * definition changes what this sweep does — the same coupling production has. + */ +async function sweep(engine: ReturnType, nowMs: number): Promise { + const lc = SysJobQueue.lifecycle!; + const retention = lc.retention!; + const cutoff = new Date(nowMs - lifecycleDurationMs(retention.maxAge)).toISOString(); + const deleted = await engine.delete(QUEUE_TABLE, { + where: { created_at: { $lt: cutoff }, ...(retention.onlyWhen ?? {}) }, + multi: true, + context: { isSystem: true, positions: [], permissions: [] }, + }); + return deleted as number; +} + +describe('sys_job_queue retention declaration (#5179)', () => { + it('declares an ADR-0057 retention window that only ever names completed rows', () => { + const lc = SysJobQueue.lifecycle; + expect(lc).toBeDefined(); + // `transient` (workflow / ephemeral state) — NOT telemetry/event/audit, + // which ADR-0057 §3.6 relocates to the dedicated `telemetry` datasource. + // A live work queue must not change stores as a side effect of cleanup. + expect(lc?.class).toBe('transient'); + expect(lc?.retention?.maxAge).toBe('7d'); + // The dead-letter queue is the reason this filter exists: `dlq`/`failed` + // wait for a human, `pending`/`running` are undelivered work. + expect(lc?.retention?.onlyWhen).toEqual({ status: 'completed' }); + expect(lc?.ttl).toBeUndefined(); // TTL has no row filter — it would eat the DLQ + expect(lc?.archive).toBeUndefined(); // and archive is incompatible with onlyWhen + }); + + it('keeps the retention window ≥ the adapter default idempotency window', () => { + // The invariant the dedup rule rests on. If someone shortens the declared + // window below the dedup window, this fails before a duplicate delivery + // ever teaches anyone about it in production. + expect(completedRetentionWindowMs()).toBeGreaterThanOrEqual(DEFAULT_IDEMPOTENCY_WINDOW_MS); + expect(completedRetentionWindowMs()).toBe(7 * DAY); + }); + + it('parses lifecycle duration literals the way @objectstack/objectql does', () => { + // Mirror of `parseLifecycleDuration` (coarse bounds: y = 365d). + expect(lifecycleDurationMs('6h')).toBe(6 * HOUR); + expect(lifecycleDurationMs('7d')).toBe(7 * DAY); + expect(lifecycleDurationMs('12w')).toBe(12 * 7 * DAY); + expect(lifecycleDurationMs('1y')).toBe(365 * DAY); + expect(() => lifecycleDurationMs('7 days')).toThrow(/invalid lifecycle duration/); + }); +}); + +describe('DbQueueAdapter window ordering (#5179)', () => { + it('refuses to construct with an idempotency window longer than the retention', () => { + expect(() => new DbQueueAdapter({ + engine: makeFakeEngine(), + options: { autoStart: false, idempotencyWindowMs: 8 * DAY }, + })).toThrow(/exceeds the retention window/); + }); + + it('accepts an idempotency window equal to the retention window', () => { + expect(() => new DbQueueAdapter({ + engine: makeFakeEngine(), + options: { autoStart: false, idempotencyWindowMs: completedRetentionWindowMs() }, + })).not.toThrow(); + }); +}); + +describe('declared retention bounds sys_job_queue (#5179)', () => { + let engine: ReturnType; + let clock: ReturnType; + let adapter: DbQueueAdapter; + + beforeEach(() => { + engine = makeFakeEngine(); + clock = makeClock(Date.parse('2026-01-01T00:00:00.000Z')); + adapter = new DbQueueAdapter({ + engine, + clock, + options: { pollIntervalMs: 60_000, autoStart: false, defaultMaxAttempts: 3 }, + }); + }); + + it('bounds the table over a long publish→completed run', async () => { + await adapter.subscribe('email.send.async', async () => { /* delivered */ }); + + // 60 days, one message a day, a sweep after each — the shape #5160 gives + // this table once every queued email lands here. + for (let day = 0; day < 60; day++) { + await adapter.publish('email.send.async', { day }); + await adapter.pollOnce(); + await sweep(engine, clock.ms); + clock.advance(DAY); + // Bounded at every point in the run, not just at the end: the window + // holds 7 days of completed rows plus the one published today. + expect(engine.rows().length).toBeLessThanOrEqual(8); + } + + const rows = engine.rows(); + expect(rows.every((r) => r.status === 'completed')).toBe(true); + // Without the policy this run would leave 60 permanent rows. + expect(rows.length).toBeLessThanOrEqual(8); + expect(rows.length).toBeGreaterThan(0); + }); + + it('never sweeps dlq, failed, pending or running rows however old', async () => { + const ancient = new Date(clock.ms - 3650 * DAY).toISOString(); + engine.tables.set(QUEUE_TABLE, [ + { id: 'm_dlq', queue: 'q', status: 'dlq', payload_json: '{}', created_at: ancient }, + { id: 'm_failed', queue: 'q', status: 'failed', payload_json: '{}', created_at: ancient }, + { id: 'm_pending', queue: 'q', status: 'pending', payload_json: '{}', created_at: ancient }, + { id: 'm_running', queue: 'q', status: 'running', payload_json: '{}', created_at: ancient }, + { id: 'm_done', queue: 'q', status: 'completed', payload_json: '{}', created_at: ancient }, + ]); + + const deleted = await sweep(engine, clock.ms); + + expect(deleted).toBe(1); + expect(engine.rows().map((r) => r.id).sort()) + .toEqual(['m_dlq', 'm_failed', 'm_pending', 'm_running']); + + // A ten-year-old dead letter is still listable and replayable — that is + // what the dead-letter queue IS. + const failed = await adapter.listFailed('q'); + expect(failed.map((f) => f.id)).toEqual(['m_dlq']); + }); + + it('leaves completed rows inside the dedup window alone, so dedup still holds', async () => { + await adapter.subscribe('billing', async () => { /* delivered */ }); + const first = await adapter.publish('billing', { invoice: 1 }, { idempotencyKey: 'inv-1' }); + await adapter.pollOnce(); + expect(engine.rows()[0]!.status).toBe('completed'); + + // Late in the 24h dedup window — well inside the 7d retention window, so + // the sweep must not take it. + clock.advance(23 * HOUR); + expect(await sweep(engine, clock.ms)).toBe(0); + expect(engine.rows()).toHaveLength(1); + + const again = await adapter.publish('billing', { invoice: 1 }, { idempotencyKey: 'inv-1' }); + expect(again).toBe(first); + expect(engine.rows()).toHaveLength(1); + }); + + it('sweeps a completed row only long after its dedup window has expired', async () => { + await adapter.subscribe('billing', async () => { /* delivered */ }); + const first = await adapter.publish('billing', { invoice: 2 }, { idempotencyKey: 'inv-2' }); + await adapter.pollOnce(); + + // Past the retention window (and therefore, by construction, long past + // the dedup window): the row goes, and the key is publishable again — + // which is the dedup contract, not a violation of it. + clock.advance(7 * DAY + HOUR); + expect(await sweep(engine, clock.ms)).toBe(1); + expect(engine.rows()).toHaveLength(0); + + const again = await adapter.publish('billing', { invoice: 2 }, { idempotencyKey: 'inv-2' }); + expect(again).not.toBe(first); + expect(engine.rows()).toHaveLength(1); + expect(engine.rows()[0]!.status).toBe('pending'); + }); + + it('a sweep mid-flight cannot take a message that has not been delivered yet', async () => { + // Old row, still `pending` because its handler was never registered: + // exactly the row an age-only policy would have thrown away. + engine.tables.set(QUEUE_TABLE, [{ + id: 'm_stuck', + queue: 'slow', + status: 'pending', + payload_json: JSON.stringify({ v: 1 }), + attempts: 0, + max_attempts: 3, + scheduled_for: new Date(clock.ms - 30 * DAY).toISOString(), + created_at: new Date(clock.ms - 30 * DAY).toISOString(), + }]); + + expect(await sweep(engine, clock.ms)).toBe(0); + + const seen: unknown[] = []; + await adapter.subscribe('slow', async (msg) => { seen.push(msg.data); }); + await adapter.pollOnce(); + + expect(seen).toEqual([{ v: 1 }]); + expect(engine.rows()[0]!.status).toBe('completed'); + }); +}); From 7d43e4d8652d6bc3a68e352cc4a6ccf93ed7d707 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 08:31:52 +0000 Subject: [PATCH 2/2] test(service-queue): pin the retention fake engine's delete() to ObjectQL's own dispatch (#4550) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `check:engine-double-contract` flagged the new fake engine in job-queue-retention.test.ts: its `delete()` hand-mirrored the engine's guard (`if (opts?.where?.id == null) throw`) instead of routing through `assertEngineDeleteDispatch`. A mirror is looser than the producer on exactly the shape a copy always drops — `where: { id: { $in: [...] } }` reads as an id and is a multi-row predicate the real engine rejects without `multi` — and a double looser than the engine it stands in for is how #4434 shipped a dead REST route with its suite green. Routes through the producer's predicate, same shape as the other 14 pinned doubles, and adds the `@objectstack/objectql` devDependency the import needs (the precedent set in plugin-email by b169f217, and in plugin-approvals / plugin-sharing before it). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017MCKJaEomEqg4tvz4SzdNd --- packages/services/service-queue/package.json | 1 + .../src/job-queue-retention.test.ts | 18 +++++++++++------- pnpm-lock.yaml | 3 +++ 3 files changed, 15 insertions(+), 7 deletions(-) diff --git a/packages/services/service-queue/package.json b/packages/services/service-queue/package.json index db5e842c7e..b0d4034f96 100644 --- a/packages/services/service-queue/package.json +++ b/packages/services/service-queue/package.json @@ -24,6 +24,7 @@ "@objectstack/spec": "workspace:*" }, "devDependencies": { + "@objectstack/objectql": "workspace:*", "@types/node": "^26.1.2", "typescript": "^6.0.3", "vitest": "^4.1.10" diff --git a/packages/services/service-queue/src/job-queue-retention.test.ts b/packages/services/service-queue/src/job-queue-retention.test.ts index 7ab9d7d590..ebb5a7459b 100644 --- a/packages/services/service-queue/src/job-queue-retention.test.ts +++ b/packages/services/service-queue/src/job-queue-retention.test.ts @@ -26,6 +26,7 @@ // the engine. import { describe, it, expect, beforeEach } from 'vitest'; +import { assertEngineDeleteDispatch } from '@objectstack/objectql'; import { SysJobQueue } from '@objectstack/platform-objects/audit'; import { DbQueueAdapter, completedRetentionWindowMs } from './db-queue-adapter.js'; import { lifecycleDurationMs } from './common.js'; @@ -94,17 +95,20 @@ function makeFakeEngine() { 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, + // and the case a mirror always drops is the one that only LOOKS like an + // id (`where: { id: { $in: [...] } }` without `multi`). + const dispatch = assertEngineDeleteDispatch(opts); const t = tables.get(table) ?? []; - if (opts?.multi) { - const where = opts.where ?? {}; - const keep = t.filter((r) => !matches(r, where)); + if (dispatch.kind === 'multi') { + const keep = t.filter((r) => !matches(r, opts?.where ?? {})); tables.set(table, keep); return t.length - keep.length; // drivers report a deleted count } - const id = opts?.where?.id; - if (id == null) throw new Error('Delete requires an ID or options.multi=true'); - tables.set(table, t.filter((r) => r.id !== id)); - return { id }; + tables.set(table, t.filter((r) => r.id !== dispatch.id)); + return { id: dispatch.id }; }, }; } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 523ea07444..a0a14b50c1 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2198,6 +2198,9 @@ importers: specifier: workspace:* version: link:../../spec devDependencies: + '@objectstack/objectql': + specifier: workspace:* + version: link:../../objectql '@types/node': specifier: ^26.1.2 version: 26.1.2