From ce2c27b5259ec199ff37cdcd813fa4769401e3a6 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 12 Sep 2026 15:20:45 +0000 Subject: [PATCH 1/4] fix(service-messaging): bound terminal sys_notification_delivery rows at 7d MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WIP — declaration + sweep pin, verification pending. Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- ...ification-delivery-retention-sweep.test.ts | 297 ++++++++++++++++++ .../objects/notification-delivery.object.ts | 43 ++- 2 files changed, 338 insertions(+), 2 deletions(-) create mode 100644 packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts diff --git a/packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts b/packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts new file mode 100644 index 0000000000..af823330dc --- /dev/null +++ b/packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts @@ -0,0 +1,297 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// #17611 — the ruled declaration (director seat, decision batch #116 item 3, +// maintainer verbatim and untranslated: 「17611 同意」 to 「席位推荐 C 现在做,A +// 另立一卡」), driven end to end. +// +// The card: a tenant with no email transport dead-letters the `email` row of +// every fanned-out `notify` on its first attempt, and those rows then sat in +// the claim query's table for 90 days — 2,876 of them at +316/day on the +// reported production tenant, carrying no work anybody would ever do. +// +// The fix under test is a DECLARATION, so this suite is driven exactly as the +// two sibling sweeps next door (`plugin-auth/sys-session-ttl-sweep.test.ts`, +// `service-storage/sys-upload-session-ttl-sweep.test.ts`): the REAL +// declaration (`objects/notification-delivery.object.ts`) through the REAL +// provisioning pass (`applySystemFields`) through the REAL Reaper +// (`@objectstack/objectql` `LifecycleService`) against a REAL SQL backend +// (`@objectstack/driver-sql`, live better-sqlite3), over a table this driver +// created from that same declaration. Nothing here restates a window by hand — +// the policy under test is `NotificationDelivery.lifecycle` itself, so an edit +// that drops or narrows it reddens this file rather than silently voiding the +// ruling. +// +// ## The two sides the ruling asked to see pinned, and the third this adds +// +// 1. TERMINAL rows past 7d ARE selected — `dead` and `suppressed`. +// 2. `pending` / `success` under 90d are NOT — the acceptance's own words. +// 3. ⚠️ …and a `success` row past 90d IS still reaped. Scoping `retention` +// to terminal statuses would otherwise have UNBOUNDED the non-terminal +// rows entirely (`retention` is one block), so this third leg is what +// distinguishes "the table window was kept" from "the table window was +// traded away on the card that exists to tighten it". +// +// ## Why the reading is discriminating +// +// `d_dead_30d` and `d_success_30d` carry the IDENTICAL `created_at`, so no age +// rule can separate their fates — only the `onlyWhen` status filter can. And +// `d_dead_2d` is terminal but inside the 7d window, so "everything terminal is +// reaped" fails too: both halves of the scoped policy are load-bearing. +// +// ## The counterfactual +// +// `seeded({ lifecycle: … })` hands the SAME service the same rows under an +// ABLATED declaration — the 7d retention with its `onlyWhen` removed — and the +// non-terminal rows are reaped along with the terminal ones. That is what +// rules out "the sweep would have produced this result anyway". ⚠️ It is +// deliberately NOT described as a dist-level ablation: the declaration under +// test is this package's own source, imported directly, so no rebuild is +// involved and none is claimed. The on-disk ablation of the shipped file is +// recorded in the PR body. + +import { describe, it, expect, afterEach } from 'vitest'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { LifecycleService, applySystemFields, assertEngineDeleteDispatch } from '@objectstack/objectql'; +import type { LifecycleEngineLike, LifecycleObjectLike } from '@objectstack/objectql'; +import type { DriverQuery } from '@objectstack/spec/contracts'; +import { NotificationDelivery } from './objects/notification-delivery.object.js'; + +const DAY_MS = 86_400_000; +/** The instant the sweep runs. Every age below is expressed against it. */ +const SWEEP_AT_MS = Date.parse('2026-09-12T00:00:00.000Z'); +const agedDays = (days: number) => new Date(SWEEP_AT_MS - days * DAY_MS).toISOString(); + +/** Past the 7d terminal window, far inside the 90d table window. */ +const AGE_30D = agedDays(30); +/** Inside BOTH windows. */ +const AGE_2D = agedDays(2); +/** Past the 90d table window. */ +const AGE_100D = agedDays(100); + +const openDrivers: SqlDriver[] = []; +afterEach(async () => { + while (openDrivers.length) { + const d = openDrivers.pop(); + try { + await d?.disconnect(); + } catch { + /* noop */ + } + } +}); + +const silentLogger = { info: () => {}, warn: () => {}, debug: () => {}, error: () => {} }; + +/** + * `LifecycleEngineLike` over a live `SqlDriver`. `delete` opens with ObjectQL's + * own dispatch predicate so this double refuses exactly what the real engine + * refuses (#4550) rather than re-deriving the rule. + */ +function sweepEngine(driver: SqlDriver, objects: LifecycleObjectLike[]): LifecycleEngineLike { + return { + registry: { getAllObjects: () => objects }, + getDriverForObject: () => driver, + async find(object: string, options: any) { + // Typed rather than erased to `any`: the driver silently DROPS an + // unrecognised query key, so `tsc` is the only channel that can reject a + // misspelt one here (#4918). + const query: DriverQuery = { where: options?.where, limit: options?.limit }; + return driver.find(object, query); + }, + async delete(object: string, options: any) { + const dispatch = assertEngineDeleteDispatch(options); + if (dispatch.kind === 'by-id') { + const id = typeof dispatch.id === 'bigint' ? dispatch.id.toString() : dispatch.id; + return (await driver.delete(object, id)) ? 1 : 0; + } + const query: DriverQuery = { where: options?.where }; + return driver.deleteMany(object, query); + }, + } as LifecycleEngineLike; +} + +/** The object as the platform actually registers it (tenant field: organization_id). */ +function provisioned() { + return applySystemFields(NotificationDelivery as any, { multiTenant: true }) as any; +} + +/** + * The six rows the policy has to tell apart. Every one of them is a shape the + * ack paths really write: `SqlNotificationOutbox.ack` / `MemoryNotificationOutbox.ack` + * produce exactly `success | suppressed | dead | pending`, and `claim()` writes + * `in_flight`. + */ +const ROWS: Array<{ id: string; status: string; created_at: string; channel: string }> = [ + // ── past the 7d terminal window ────────────────────────────────────────── + { id: 'd_dead_30d', status: 'dead', created_at: AGE_30D, channel: 'email' }, + { id: 'd_suppressed_30d', status: 'suppressed', created_at: AGE_30D, channel: 'email' }, + // ── terminal, but INSIDE the 7d window ─────────────────────────────────── + { id: 'd_dead_2d', status: 'dead', created_at: AGE_2D, channel: 'email' }, + // ── non-terminal, same age as the reaped terminal rows ─────────────────── + { id: 'd_success_30d', status: 'success', created_at: AGE_30D, channel: 'inbox' }, + { id: 'd_pending_30d', status: 'pending', created_at: AGE_30D, channel: 'inbox' }, + { id: 'd_in_flight_30d', status: 'in_flight', created_at: AGE_30D, channel: 'inbox' }, + // ── non-terminal, past the 90d TABLE window ────────────────────────────── + { id: 'd_success_100d', status: 'success', created_at: AGE_100D, channel: 'inbox' }, +]; + +async function seeded(opts?: { lifecycle?: unknown }) { + const schema = provisioned(); + const driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + openDrivers.push(driver); + await driver.initObjects([schema]); + + for (const r of ROWS) { + await driver.create('sys_notification_delivery', { + id: r.id, + notification_id: 'n_1', + recipient_id: 'usr_1', + channel: r.channel, + status: r.status, + attempts: r.status === 'pending' ? 0 : 1, + partition_key: 0, + created_at: r.created_at, + updated_at: r.created_at, + organization_id: 'org_A', + }); + } + + const object: LifecycleObjectLike = { + name: NotificationDelivery.name, + lifecycle: + opts && 'lifecycle' in opts ? (opts.lifecycle as any) : (NotificationDelivery as any).lifecycle, + fields: schema.fields, + } as LifecycleObjectLike; + + const service = new LifecycleService({ + getEngine: () => sweepEngine(driver, [object]), + logger: silentLogger, + now: () => SWEEP_AT_MS, + initialDelayMs: 1, + sweepIntervalMs: 10, + } as any); + + return { driver, service }; +} + +const ALL_ROWS: DriverQuery = {}; +const survivors = async (driver: SqlDriver) => + (await driver.find('sys_notification_delivery', ALL_ROWS)).map((r: any) => r.id).sort(); + +describe('[#17611] sys_notification_delivery terminal retention — real declaration, real Reaper, live SQL', () => { + it('is exactly the ruled declaration', () => { + expect((NotificationDelivery as any).lifecycle).toEqual({ + class: 'telemetry', + ttl: { field: 'created_at', expireAfter: '90d' }, + retention: { + maxAge: '7d', + onlyWhen: { status: { $in: ['dead', 'suppressed'] } }, + }, + }); + }); + + it('scopes on statuses the field declares AND the ack paths really write', () => { + // A filter naming a value the writers never produce would compile to a + // predicate matching nothing — the sweep would silently reap nothing. + const scoped: string[] = (NotificationDelivery as any).lifecycle.retention.onlyWhen.status.$in; + const declared: string[] = (NotificationDelivery.fields as any).status.options.map( + (o: any) => (typeof o === 'string' ? o : o.value), + ); + for (const s of scoped) expect(declared).toContain(s); + // `success` is terminal too, and is deliberately OUT of the scope: the + // ruling keeps delivery history at the table window. + expect(scoped).not.toContain('success'); + }); + + it('POSITIVE — terminal rows past 7d are reaped, and the retention leg is recorded', async () => { + const { driver, service } = await seeded(); + + const report = await service.sweep(); + + const left = await survivors(driver); + expect(left).not.toContain('d_dead_30d'); + expect(left).not.toContain('d_suppressed_30d'); + expect(report.errors).toEqual([]); + const retention = report.swept.find( + (e: any) => e.object === 'sys_notification_delivery' && e.policy === 'retention', + ); + expect(retention).toBeTruthy(); + }); + + it('NEGATIVE — pending / success / in_flight rows of the SAME age are untouched', async () => { + const { driver, service } = await seeded(); + + await service.sweep(); + + const left = await survivors(driver); + // Identical `created_at` to the two rows just reaped, so nothing but the + // `onlyWhen` status filter can be what spared them. + expect(left).toContain('d_success_30d'); + expect(left).toContain('d_pending_30d'); + expect(left).toContain('d_in_flight_30d'); + }); + + it('NEGATIVE — a terminal row INSIDE the 7d window is untouched', async () => { + const { driver, service } = await seeded(); + + await service.sweep(); + + // Without this the suite would pass for a policy that reaped every + // terminal row at any age. + expect(await survivors(driver)).toContain('d_dead_2d'); + }); + + it('the TABLE window survived — a non-terminal row past 90d is still reaped by the ttl leg', async () => { + const { driver, service } = await seeded(); + + const report = await service.sweep(); + + expect(await survivors(driver)).not.toContain('d_success_100d'); + const ttl = report.swept.find( + (e: any) => e.object === 'sys_notification_delivery' && e.policy === 'ttl', + ); + expect(ttl).toBeTruthy(); + }); + + it('one sweep, seven rows, the whole verdict in one assertion', async () => { + const { driver, service } = await seeded(); + + await service.sweep(); + + expect(await survivors(driver)).toEqual([ + 'd_dead_2d', + 'd_in_flight_30d', + 'd_pending_30d', + 'd_success_30d', + ]); + }); + + it('COUNTERFACTUAL — drop `onlyWhen` and the same sweep takes the non-terminal rows too', async () => { + // The naive policy: a 7d table-wide retention. This is what the negative + // pins above have to discriminate against, so they are not vacuous. + const { driver, service } = await seeded({ + lifecycle: { + class: 'telemetry', + ttl: { field: 'created_at', expireAfter: '90d' }, + retention: { maxAge: '7d' }, + }, + }); + + await service.sweep(); + + expect(await survivors(driver)).toEqual(['d_dead_2d']); + }); + + it('COUNTERFACTUAL — with no lifecycle declaration the same sweep reaps nothing', async () => { + const { driver, service } = await seeded({ lifecycle: null }); + + const report = await service.sweep(); + + expect(await survivors(driver)).toEqual(ROWS.map((r) => r.id).sort()); + expect(report.swept.filter((e: any) => e.object === 'sys_notification_delivery')).toEqual([]); + }); +}); diff --git a/packages/services/service-messaging/src/objects/notification-delivery.object.ts b/packages/services/service-messaging/src/objects/notification-delivery.object.ts index 0c9b406cf1..1a4df87496 100644 --- a/packages/services/service-messaging/src/objects/notification-delivery.object.ts +++ b/packages/services/service-messaging/src/objects/notification-delivery.object.ts @@ -24,10 +24,49 @@ export const NotificationDelivery = ObjectSchema.create({ icon: 'send', isSystem: true, managedBy: 'engine-owned', - // ADR-0057: pipeline telemetry — same 90d window as sys_notification. + // ADR-0057: pipeline telemetry. TWO windows, because this table + // interleaves rows that still carry work with rows that never will. + // + // [#17611] A tenant with no transport configured for a fanned-out channel + // dead-letters that channel's row on its FIRST attempt (`attempts: 1`), + // and every `notify` writes one such row forever after. Measured on a + // production tenant: 2,876 `email`/`dead` rows at +316/day, zero pending. + // Those rows carry no work — nothing ever claims, retries or acks them + // again — yet they sat in the claim query's table for the full 90d. + // + // `retention.onlyWhen` scopes the SHORT window to the terminal-FAILURE + // statuses, the same shape `sys_job_queue`, `sys_automation_run` and + // `sys_upload_session` already declare. `success` is deliberately NOT in + // the scope: the ruling keeps delivery history at the table window. + // + // ⚠️ The 7d scope does not REPLACE the table's bound, it sits under it — + // `retention` is a single block, so scoping it would have left every + // non-terminal row (`pending`, `in_flight`, `success`) with no age bound + // at all, unbounding the larger half of this table's growth on the very + // card that exists to bound it. The `ttl` leg restates the 90d window the + // object has always declared, on the same `created_at` clock `retention` + // reaps by, so non-terminal rows keep exactly today's behaviour. Both legs + // run: `LifecycleService.reapObject` takes `ttl` and `retention` in + // independent `if`s, not an either/or. + // + // ⛔ The `$in` list is a third copy of a vocabulary the WRITERS own + // (`SqlNotificationOutbox.ack` / `MemoryNotificationOutbox.ack`), so it + // must be widened in the same change as a writer — the `sys_automation_run` + // lesson: a widened writer against a narrow sweep scope means the new + // status is simply never aged out, silently, forever. Today those two are + // the only terminal-failure statuses either ack path can produce. + // `failed` is a legal member of the `status` field below but NO writer of + // THIS object ever sets it (it is `sys_http_delivery`'s terminal status), + // so naming it here would scope the sweep on a value that cannot occur. lifecycle: { class: 'telemetry', - retention: { maxAge: '90d' }, + // The table window — every row, same 90d as sys_notification. + ttl: { field: 'created_at', expireAfter: '90d' }, + // The terminal-failure window — rows that will never carry work again. + retention: { + maxAge: '7d', + onlyWhen: { status: { $in: ['dead', 'suppressed'] } }, + }, }, description: 'Durable per-recipient × channel delivery outbox (ADR-0030 Layer 4).', titleFormat: '{channel} → {recipient_id}', From 32e8f0603a80e424a76185cda31d45f166165810 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 12 Sep 2026 15:31:28 +0000 Subject: [PATCH 2/4] test(service-messaging): pin the terminal-row retention sweep end to end Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- .../src/notification-delivery-retention-sweep.test.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts b/packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts index af823330dc..152b3d688d 100644 --- a/packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts +++ b/packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts @@ -148,7 +148,11 @@ async function seeded(opts?: { lifecycle?: unknown }) { for (const r of ROWS) { await driver.create('sys_notification_delivery', { id: r.id, - notification_id: 'n_1', + // Distinct per row: the object declares `(notification_id, recipient_id, + // channel)` UNIQUE, which is the real dedup key — one delivery per + // (event × recipient × channel). Reusing one event id here would be a + // shape production cannot produce. + notification_id: `n_${r.id}`, recipient_id: 'usr_1', channel: r.channel, status: r.status, From 0d64ad5cffbd1fa9fbdd2ce01daaa1529bb86423 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 12 Sep 2026 15:33:01 +0000 Subject: [PATCH 3/4] chore: changeset for the terminal delivery retention window Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- .../17611-terminal-delivery-retention.md | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 .changeset/17611-terminal-delivery-retention.md diff --git a/.changeset/17611-terminal-delivery-retention.md b/.changeset/17611-terminal-delivery-retention.md new file mode 100644 index 0000000000..af9c4126e6 --- /dev/null +++ b/.changeset/17611-terminal-delivery-retention.md @@ -0,0 +1,30 @@ +--- +'@objectstack/service-messaging': minor +--- + +`sys_notification_delivery` reaps its terminal-failure rows after **7 days** instead of 90 (#17611) + +**⚠️ Operational consequence, stated plainly: `dead` and `suppressed` delivery rows are now deleted 7 days after they were created.** Any report, SLA reading, dashboard or manual investigation that consulted them — "which notifications failed to send, and why" — must now read inside that window. Before this change those rows survived for 90 days. Nothing else about the table changes: `pending`, `in_flight` and `success` rows keep the same 90-day window they have always had, and no row is reaped sooner than before except the two terminal-failure statuses. + +**What was wrong.** Fan-out writes one delivery row per `(event × recipient × channel)`. A tenant with no transport configured for one of those channels dead-letters that channel's row on its **first** attempt, and every `notify` writes another one. Measured on a production tenant: 2,876 `email`/`dead` rows against 2,876 `inbox`/`success` rows, `max(attempts) = 1`, zero pending, growing +316 rows/day. Those rows carry no work — nothing ever claims, retries or acks them again — but they sat in the table the dispatcher's claim query reads on every hop for the full 90-day window, so the cost of every claim rose linearly with time. + +**The change** is one declaration on the object, using spec keys that already ship and are already consumed by the platform Reaper: + +```ts +lifecycle: { + class: 'telemetry', + ttl: { field: 'created_at', expireAfter: '90d' }, + retention: { + maxAge: '7d', + onlyWhen: { status: { $in: ['dead', 'suppressed'] } }, + }, +}, +``` + +`retention.onlyWhen` scopes the short window to the terminal-failure statuses — the same shape `sys_job_queue`, `sys_automation_run` and `sys_upload_session` already declare. No channel interface member, no new status value, no change to fan-out. + +The `ttl` leg is not new behaviour: it restates the 90-day bound the object has always declared. `lifecycle.retention` is a single block, so scoping it to terminal rows would otherwise have left `pending` / `in_flight` / `success` with **no age bound at all** — unbounding the larger half of this table's growth on the very change that exists to bound it. Both legs run: `LifecycleService.reapObject` takes `ttl` and `retention` in independent branches. `success` is deliberately outside the scope; delivery history stays at the table window. + +**If you override this object's lifecycle windows through the `lifecycle` settings namespace, re-read your configuration.** `retention_overrides.maxAge` for `sys_notification_delivery` used to move the whole table's window; it now moves the **terminal-failure** window only, and `expireAfter` moves the table window. An override left in place keeps parsing and keeps applying — to a narrower set of rows than it did before. + +**⚠️ This is worth nothing where the Reaper does not run.** The whole benefit is delivered by `LifecycleService`, which `OS_LIFECYCLE_DISABLED=1` or the plugin switch turns off. A deployment with lifecycle disabled kept these rows forever before this change and keeps them forever after it; a declaration is not a sweeper. Check that the Reaper is enabled before reading this entry as a bound on your table. From a6c8cd7f55a2687d91a9cb392ad55fc04f0fe068 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 12 Sep 2026 15:48:29 +0000 Subject: [PATCH 4/4] chore: record the new engine double in the coverage ledger Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- scripts/engine-double-contract.pinned.json | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index b888922a21..061e161b39 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3726,6 +3726,11 @@ "verb": "findOne", "pinned": 6 }, + { + "file": "packages/services/service-messaging/src/notification-delivery-retention-sweep.test.ts", + "verb": "delete", + "pinned": 1 + }, { "file": "packages/services/service-messaging/src/notification-schema-conformance.test.ts", "verb": "findOne",