fix(service-queue,platform-objects): sys_job_queue 的 completed 行按声明式 retention 到期即清(#5179)

.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.

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)

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`:

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"

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<string, number> = {
+  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 <n><unit> with unit h|d|w|y (e.g. '7d')`,
+    );
+  }
+  return Number(m[1]) * LIFECYCLE_UNIT_MS[m[2]!]!;
+}
+
 export function parseJson<T = unknown>(raw: unknown, fallback?: T): T | undefined {
   if (raw == null) return fallback;
   if (typeof raw === 'string') {

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

@@ -7,26 +7,59 @@ 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,
 } from './common.js';
 
 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;
   /** Max messages claimed per poll tick (default 10) */
   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, {

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';