From 32621faa2aba2b38a00e873938f18612eda56df4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 01:37:54 +0000 Subject: [PATCH] fix(objectql): compose reap guards by intersection instead of last-wins (#5535) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `registerReapGuard` was a single-slot `set()` documented as "one guard per object (last registration wins)", over a `private` registry with no probe and no warn on overwrite. A second registrant on one object therefore silently unhooked the first, and could not have noticed. That is unsafe on this seam specifically, because a guard's confirmation is not a vote but a receipt: "external cleanup is done, this row may go now". `sys_file`'s guard reclaims storage bytes before confirming, and the row is the only pointer to those bytes — so displacing it means rows deleted, bytes leaked, zero log lines. ADR-0057 §3.3 explicitly invites a second domain callback (#4672's de-indexing is one), which is what turns this from theory into the next merge. Guards now compose by intersection: registration appends, and only ids every guard confirmed are deleted; one veto keeps the row for the next sweep. Same shape as `registerRetentionFloor` one policy over, where every registrar keeps its say and the strictest wins. The intersection is a narrowing pipeline, not N verdicts unioned at the end: a guard is asked only about rows the guards before it confirmed, so it never performs irreversible cleanup for a row another guard is about to keep. The delete set does not depend on registration order. A throwing guard still aborts the object's reap with nothing deleted. Re-registering the identical function is a no-op. Single-guard behaviour is unchanged — the five existing guard tests pass untouched, and service-storage's two registrars need no change. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01V7WetGmnfoXNn8cLieKKmx --- .../reap-guard-intersection-composition.md | 40 ++++ .../src/lifecycle/lifecycle-service.test.ts | 210 ++++++++++++++++++ .../src/lifecycle/lifecycle-service.ts | 128 +++++++++-- 3 files changed, 358 insertions(+), 20 deletions(-) create mode 100644 .changeset/reap-guard-intersection-composition.md diff --git a/.changeset/reap-guard-intersection-composition.md b/.changeset/reap-guard-intersection-composition.md new file mode 100644 index 0000000000..7dd2e85919 --- /dev/null +++ b/.changeset/reap-guard-intersection-composition.md @@ -0,0 +1,40 @@ +--- +"@objectstack/objectql": minor +--- + +fix(objectql): 同一 object 的多个 reap guard 现在按交集组合,后注册者不再静默顶掉前者 (#5535) + +`LifecycleService.registerReapGuard(object, guard)` 此前是一行 `set()`,契约写明 +「One guard per object (last registration wins)」,而注册表是 `private`、无探针、 +覆盖时也不打日志。两点合起来:同一 object 上的**第二个注册方静默解除第一个**,且 +**无从察觉**。 + +这在 reap guard 这个座位上格外危险,因为 guard 的语义不是「投票」而是**回执**—— +「外部副作用已经做完,这行现在可以删了」。`sys_file` 的 guard 做的正是字节回收: +先 `storage.delete(row.key)`,失败则 veto 把行留到下一轮(行是字节的唯一指针, +先删行就永久泄漏)。任何第二个消费者在 `sys_file` 上注册,这段字节回收会被**整体 +解除**:行照删、字节泄漏、零日志。而 ADR-0057 §3.3 的 amendment 恰恰把「domain +callback」认定为 guard 的合法形态,第二个注册方是被 ADR 鼓励出来的,不是假想。 + +**新契约:交集组合。** 一个 object 可以有多个 guard,注册**追加**而非替换;只有 +**全部** guard 都确认的 id 才进删除集,任一 veto 即保留、由下一轮 sweep 重试。这与 +单 guard 时的语义完全兼容(现有五条单 guard 回归全部原样通过),也与同文件 +`registerRetentionFloor` 的「每个注册方都保留发言权,最严者胜」同构。 + +写 guard 时值得知道的两点: + +- **guard 按注册顺序执行,但只会被问到「前面的 guard 已经确认过」的行。** 这是 + 刻意的:被问到即意味着「到目前为止所有人都同意这行可以删」,于是 guard 不会 + 为一行别人正要保留的记录做不可逆的清理(否则就会出现「行还在、字节已删」或 + 「行还在、索引已删」)。删除集本身与注册顺序无关。 +- **guard 抛异常的处置不变**:异常上抛到 `sweep()` 的 per-object handler,该 + object 本轮一行不删(erroring guard 永不 fail open 进删除)。同一批里更早的 + guard 已经做掉的清理由下一轮 sweep 重试,而不会在无人完成确认的批次上兑现成 + 一次删除。 + +重复注册**同一个函数引用**是 no-op(重跑 wiring,不是第二份意见):与自己求交集 +不改变任何结果,却会让它的外部清理在每批上跑两遍。 + +无需调用方改动:`service-storage` 的两个注册方(`sys_file` / `sys_upload_session`) +一行未改,行为逐条不变。本单落地即解除 #4672(知识插件走 reap-guard 去索引化)的 +Blocked-by——它可以直接在 `sys_file` 上追加注册,而不必担心顶掉字节回收。 diff --git a/packages/objectql/src/lifecycle/lifecycle-service.test.ts b/packages/objectql/src/lifecycle/lifecycle-service.test.ts index ef707c61df..f31b6cf90e 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.test.ts +++ b/packages/objectql/src/lifecycle/lifecycle-service.test.ts @@ -415,6 +415,216 @@ describe('LifecycleService.sweep — reap guard', () => { }); }); +// [#5535] Two registrars on ONE object. Before this, `registerReapGuard` was a +// single-slot `set()`: the second registrant silently unhooked the first — on +// `sys_file` that means the byte-reclaim guard stops running while the rows +// (the only pointer to those bytes) keep being deleted. Guards now compose by +// intersection, which is what "confirm before delete" already implied. +describe('LifecycleService.sweep — reap guard composition', () => { + const guarded: LifecycleObjectLike[] = [ + { name: 'sys_file', lifecycle: { class: 'transient', ttl: { field: 'deleted_at', expireAfter: '30d' } } }, + ]; + + const rowsOf = (...ids: string[]) => ids.map((id) => ({ id, deleted_at: '2020-01-01T00:00:00Z' })); + + /** A store the sweep actually drains, so "retried next sweep" is observable. */ + function backedEngine(seed: Array>) { + const store = [...seed]; + const captured = captureEngine(guarded, { + findImpl: () => store.slice(), + deleteImpl: (_object, options) => { + const idx = store.findIndex((r) => r.id === options?.where?.id); + if (idx >= 0) store.splice(idx, 1); + return { deletedCount: idx >= 0 ? 1 : 0 }; + }, + }); + return { ...captured, store }; + } + + const idsOf = (rows: Array>) => rows.map((r) => r.id as string); + + it('deletes only ids EVERY guard confirmed; a veto by either one keeps the row', async () => { + const rows = rowsOf('f1', 'f2', 'f3'); + const { engine, deletes } = captureEngine(guarded, { findImpl: () => rows }); + const svc = service(engine); + // Each guard vetoes a different id — neither alone would keep both. + svc.registerReapGuard('sys_file', async (_o, r) => idsOf(r).filter((id) => id !== 'f2')); + svc.registerReapGuard('sys_file', async (_o, r) => idsOf(r).filter((id) => id !== 'f3')); + + const report = await svc.sweep(); + + expect(deletes.map((d) => d.where)).toEqual([{ id: 'f1' }]); + expect(report.swept[0].deleted).toBe(1); + expect(report.errors).toEqual([]); + }); + + it('is order-independent: the same two guards registered the other way round agree', async () => { + const vetoF2 = async (_o: string, r: Array>) => + idsOf(r).filter((id) => id !== 'f2'); + const vetoF3 = async (_o: string, r: Array>) => + idsOf(r).filter((id) => id !== 'f3'); + + const deleteSets: string[][] = []; + for (const order of [[vetoF2, vetoF3], [vetoF3, vetoF2]]) { + const { engine, deletes } = captureEngine(guarded, { findImpl: () => rowsOf('f1', 'f2', 'f3') }); + const svc = service(engine); + for (const g of order) svc.registerReapGuard('sys_file', g); + await svc.sweep(); + deleteSets.push(deletes.map((d) => d.where.id as string)); + } + + expect(deleteSets[0]).toEqual(['f1']); + expect(deleteSets[1]).toEqual(deleteSets[0]); + }); + + it('asks a guard only about rows the guards before it confirmed', async () => { + // The point of the narrowing: a guard's confirmation is the RECEIPT for + // cleanup it has already performed. Showing guard 2 a row guard 1 vetoed + // would have it reclaim/de-index a row that then survives the sweep. + const rows = rowsOf('f1', 'f2', 'f3'); + const { engine } = captureEngine(guarded, { findImpl: () => rows }); + const svc = service(engine); + const first = vi.fn(async (_o: string, r: Array>) => + idsOf(r).filter((id) => id !== 'f2'), + ); + const second = vi.fn(async (_o: string, r: Array>) => idsOf(r)); + svc.registerReapGuard('sys_file', first); + svc.registerReapGuard('sys_file', second); + + await svc.sweep(); + + expect(first).toHaveBeenCalledWith('sys_file', rows); // the full candidate batch + expect(second).toHaveBeenCalledTimes(1); + expect(idsOf(second.mock.calls[0][1])).toEqual(['f1', 'f3']); // f2 already vetoed + }); + + it('a guard that vetoes the whole batch spares the later guards the call', async () => { + const { engine, deletes } = captureEngine(guarded, { findImpl: () => rowsOf('f1', 'f2') }); + const svc = service(engine); + const second = vi.fn(async (_o: string, r: Array>) => idsOf(r)); + svc.registerReapGuard('sys_file', async () => []); + svc.registerReapGuard('sys_file', second); + + const report = await svc.sweep(); + + expect(second).not.toHaveBeenCalled(); + expect(deletes).toHaveLength(0); + expect(report.swept[0].deleted).toBe(0); + }); + + it('a row one guard vetoed is retried by the next sweep and deleted once both confirm', async () => { + const { engine, store } = backedEngine(rowsOf('f1', 'f2')); + const svc = service(engine); + let firstSweep = true; + svc.registerReapGuard('sys_file', async (_o, r) => + idsOf(r).filter((id) => !(firstSweep && id === 'f2')), + ); + svc.registerReapGuard('sys_file', async (_o, r) => idsOf(r)); + + const one = await svc.sweep(); + expect(idsOf(store)).toEqual(['f2']); // vetoed, still there + expect(one.swept[0].deleted).toBe(1); + + firstSweep = false; + const two = await svc.sweep(); + expect(store).toEqual([]); // retried and reaped + expect(two.swept[0].deleted).toBe(1); + }); + + it('a throwing guard deletes nothing — not even ids an earlier guard confirmed', async () => { + // Same fail-safe as the single-guard case: the error reaches the per-object + // handler in sweep() and no row is deleted. The earlier guard's external + // cleanup for this batch is simply retried next sweep — never paid out in + // a delete on a batch no one finished confirming. + const { engine, deletes } = captureEngine(guarded, { findImpl: () => rowsOf('f1', 'f2') }); + const svc = service(engine); + const first = vi.fn(async (_o: string, r: Array>) => idsOf(r)); + svc.registerReapGuard('sys_file', first); + svc.registerReapGuard('sys_file', async () => { + throw new Error('index unreachable'); + }); + + const report = await svc.sweep(); + + expect(first).toHaveBeenCalledTimes(1); + expect(deletes).toHaveLength(0); + expect(report.swept).toEqual([]); + expect(report.errors).toEqual([{ object: 'sys_file', error: 'index unreachable' }]); + }); + + it('registering the identical guard twice runs it once (re-run wiring, not a second opinion)', async () => { + const { engine, deletes } = captureEngine(guarded, { findImpl: () => rowsOf('f1') }); + const svc = service(engine); + const guard = vi.fn(async (_o: string, r: Array>) => idsOf(r)); + svc.registerReapGuard('sys_file', guard); + svc.registerReapGuard('sys_file', guard); + + await svc.sweep(); + + // Called twice, its external cleanup would run twice per batch. + expect(guard).toHaveBeenCalledTimes(1); + expect(deletes.map((d) => d.where)).toEqual([{ id: 'f1' }]); + }); + + it('multiple guards on an engine without find are skipped, never blind-deleted', async () => { + const { engine, deletes } = captureEngine(guarded); // no findImpl → no engine.find + const svc = service(engine); + svc.registerReapGuard('sys_file', async () => ['f1']); + svc.registerReapGuard('sys_file', async () => ['f1']); + + const report = await svc.sweep(); + + expect(deletes).toHaveLength(0); + expect(report.skipped).toEqual([{ object: 'sys_file', reason: 'reap-guard-unsupported' }]); + }); + + it('keeps byte reclaim running when a second consumer registers (the #5535 shape)', async () => { + // service-storage's `sys_file` guard, in the same shape but stubbed here: + // reclaim the bytes FIRST, confirm only if that succeeded (the row is the + // only pointer to the bytes, so a failed reclaim must veto). The second + // registrar is the ADR-0057 §3.3 domain callback #4672 will add — it + // de-indexes by id. Under the old single-slot registry the byte reclaim + // was unhooked wholesale by that second call: rows deleted, bytes leaked. + const bytes = new Map([ + ['f1', 'k1'], + ['f2', 'k2'], + ['f3', 'k3'], + ]); + const index = new Set(['f1', 'f2', 'f3']); + const { engine, store } = backedEngine(rowsOf('f1', 'f2', 'f3')); + const svc = service(engine); + + svc.registerReapGuard('sys_file', async (_o, rows) => { + const confirmed: string[] = []; + for (const row of rows) { + const id = row.id as string; + if (id === 'f2') continue; // storage.delete threw → veto, keep the pointer + bytes.delete(id); + confirmed.push(id); + } + return confirmed; + }); + svc.registerReapGuard('sys_file', async (_o, rows) => { + const confirmed: string[] = []; + for (const row of rows) { + index.delete(row.id as string); + confirmed.push(row.id as string); + } + return confirmed; + }); + + const report = await svc.sweep(); + + expect(idsOf(store)).toEqual(['f2']); // vetoed row retained for the next sweep + expect([...bytes.keys()]).toEqual(['f2']); // …with its bytes intact — no leak + // …and the de-indexer was never shown f2, so a surviving row keeps its + // index entry rather than silently disappearing from search. + expect([...index]).toEqual(['f2']); + expect(report.swept[0].deleted).toBe(2); + expect(report.errors).toEqual([]); + }); +}); + describe('LifecycleService.sweep — Archiver (P3)', () => { const AUDIT_OBJ: LifecycleObjectLike = { name: 'sys_audit_log', diff --git a/packages/objectql/src/lifecycle/lifecycle-service.ts b/packages/objectql/src/lifecycle/lifecycle-service.ts index ea291b7c9d..44b0779dea 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.ts +++ b/packages/objectql/src/lifecycle/lifecycle-service.ts @@ -316,6 +316,10 @@ const REAP_GUARD_MAX_BATCHES_PER_SWEEP = 20; * Guards are registered at runtime (`registerReapGuard`), not declared in the * spec: detection and scheduling stay inside the single platform sweep * (ADR-0057 §3.3 — a guard is a domain callback, not a second sweeper). + * + * [#5535] An object may carry SEVERAL guards, and they compose by + * intersection: an id is deleted only if every guard confirmed it. See + * {@link LifecycleService.registerReapGuard}. */ export type LifecycleReapGuard = ( object: string, @@ -331,8 +335,13 @@ export class LifecycleService { private lastCounts = new Map(); /** Governance snapshot for the sweep in flight. */ private governance: GovernanceSnapshot = DEFAULT_GOVERNANCE; - /** Per-object reap guards ({@link LifecycleReapGuard}). */ - private readonly reapGuards = new Map(); + /** + * [#5535] Per-object reap guards ({@link LifecycleReapGuard}), in + * registration order. A LIST, not a single slot: every registrar of an + * object keeps its say (all must confirm), the same reason + * {@link registerRetentionFloor} keys by registrar rather than overwriting. + */ + private readonly reapGuards = new Map(); /** * [#5195] Registered retention floors, keyed `object::policy::declaredBy` so * a re-registration replaces rather than accumulates, while two independent @@ -413,12 +422,56 @@ export class LifecycleService { /** * Register a {@link LifecycleReapGuard} for one object. From then on the * Reaper never blind-deletes that object's rows: candidates are fetched, - * the guard confirms (after external cleanup) or vetoes each row, and only - * confirmed ids are deleted. One guard per object (last registration wins — - * guards are platform wiring, not user surface). + * each guard confirms (after external cleanup) or vetoes each row, and only + * confirmed ids are deleted. + * + * [#5535] **Several guards may govern one object, and they COMPOSE by + * intersection** — an id is deleted only if every registered guard confirmed + * it; a single veto keeps the row for the next sweep to retry. Registering + * does not replace: the previous registrar keeps its say. + * + * The alternative — one slot, last registration wins — was how this started, + * and it is unsafe for exactly the reason the guard seam exists. A guard's + * contract is "external cleanup first, then confirm", so `sys_file`'s guard + * reclaims storage bytes before it confirms, and the row is the only pointer + * to those bytes. A second registrar (ADR-0057 §3.3 explicitly invites one — + * a domain callback such as de-indexing a derived index by id) would have + * displaced that byte reclaim wholesale: rows deleted, bytes leaked, not one + * line of log to say so, and nothing the newcomer could have read to notice + * (#5535). Intersection is the composition the "confirm before delete" + * contract already implies, and it is what {@link registerRetentionFloor} + * does one policy over (there: the strictest window wins). + * + * Two consequences worth knowing when writing a guard: + * + * - Guards run in registration order, but a guard is only ever asked about + * rows the guards before it have already confirmed. That is deliberate: + * being asked implies "everyone so far agrees this row can go", so a guard + * never performs irreversible cleanup for a row another guard is about to + * keep. The delete set itself does not depend on the order. + * - A guard that throws aborts the object's reap with nothing deleted + * (unchanged: an erroring guard must never fail open into deletion), so + * cleanup already done by an earlier guard in that batch is retried next + * sweep rather than paid out in a delete. + * + * Registering the identical function twice is a no-op — re-run wiring, not a + * second opinion: intersecting a guard's verdict with itself changes no + * outcome, while calling it twice would run its external cleanup twice. */ registerReapGuard(object: string, guard: LifecycleReapGuard): void { - this.reapGuards.set(object, guard); + const guards = this.reapGuards.get(object); + if (!guards) { + this.reapGuards.set(object, [guard]); + return; + } + if (!guards.includes(guard)) guards.push(guard); + } + + /** Snapshot of the guards governing `object`, in registration order — a copy, + * so a registration mid-sweep cannot change the set a reap in flight is + * consulting. */ + private reapGuardsFor(object: string): LifecycleReapGuard[] { + return [...(this.reapGuards.get(object) ?? [])]; } /** @@ -1017,8 +1070,8 @@ export class LifecycleService { // A guarded object is NEVER blind-deleted: without row reads the guard // cannot confirm, so the reap is skipped (fail-safe), not degraded. - const guard = this.reapGuards.get(object); - if (guard && typeof engine.find !== 'function') { + const guards = this.reapGuardsFor(object); + if (guards.length > 0 && typeof engine.find !== 'function') { if (!report.skipped.some((s) => s.object === object && s.reason === 'reap-guard-unsupported')) { report.skipped.push({ object, reason: 'reap-guard-unsupported' }); } @@ -1031,8 +1084,8 @@ export class LifecycleService { else if (total !== undefined) total += n; }; const reapWhere = async (where: Record): Promise => - guard - ? this.guardedReap(engine, object, guard, where) + guards.length > 0 + ? this.guardedReap(engine, object, guards, where) : countDeleted(await engine.delete(object, { where, multi: true, context: { ...SYSTEM_CTX } })); if (tenantWindows.length === 0) { @@ -1073,17 +1126,27 @@ export class LifecycleService { } /** - * Guarded reap: fetch candidate rows in batches, let the guard confirm - * (after performing external cleanup) or veto each, delete only confirmed - * ids. A guard error propagates to the per-object handler in `sweep()` — - * an erroring guard must never fail open into deletion. A batch that isn't - * fully confirmed ends the pass: vetoed rows still match the cutoff filter - * and would be re-fetched forever; the next sweep retries them. + * Guarded reap: fetch candidate rows in batches, let every guard confirm + * (after performing external cleanup) or veto each, delete only the ids + * ALL of them confirmed. A guard error propagates to the per-object handler + * in `sweep()` — an erroring guard must never fail open into deletion. A + * batch that isn't fully confirmed ends the pass: vetoed rows still match + * the cutoff filter and would be re-fetched forever; the next sweep retries + * them. + * + * [#5535] The intersection is computed as a narrowing pipeline rather than + * N independent verdicts unioned at the end, because a guard's confirmation + * is not an opinion — it is the *receipt* for cleanup it has already + * performed. Handing guard 2 a row guard 1 already vetoed would have it + * de-index (or otherwise reclaim) a row that then survives the sweep. So a + * guard is asked only about rows still standing, and one that vetoes + * everything ends the batch before the rest are called at all. The delete + * set is the same whatever the registration order. */ private async guardedReap( engine: LifecycleEngineLike, object: string, - guard: LifecycleReapGuard, + guards: readonly LifecycleReapGuard[], where: Record, ): Promise { let total = 0; @@ -1094,13 +1157,23 @@ export class LifecycleService { context: { ...SYSTEM_CTX }, }); if (!rows?.length) break; - const confirmed = (await guard(object, rows)).filter((id) => id !== null && id !== undefined); + let confirmed = rows; + for (const guard of guards) { + const ids = new Set( + (await guard(object, confirmed)).map(idKey).filter((k): k is string => k !== undefined), + ); + confirmed = confirmed.filter((row) => { + const k = idKey(row?.id); + return k !== undefined && ids.has(k); + }); + if (confirmed.length === 0) break; + } // Per-id deletes (NOT a `{$in}` filter): the engine's delete path reads // `where.id` as a scalar target, and by-id deletes get referential // cascade handling a filter-delete would bypass. - for (const id of confirmed) { + for (const row of confirmed) { await engine.delete(object, { - where: { id }, + where: { id: row.id }, multi: true, context: { ...SYSTEM_CTX }, }); @@ -1112,6 +1185,21 @@ export class LifecycleService { } } +/** + * [#5535] Identity key used to intersect one guard's confirmed ids with the + * rows it was handed. `undefined` for an unusable id (null/undefined), which + * never matches anything — a row with no id cannot be confirmed, and a guard + * that returns one is confirming nothing. + * + * Stringified because the two sides are typed independently: a guard returns + * `string | number` while a row's id is whatever the driver stored, so a guard + * that hands back `String(row.id)` must not silently stop reclaiming. Within + * one batch the rows' ids are distinct, so the collapse cannot merge two rows. + */ +function idKey(id: unknown): string | undefined { + return id === null || id === undefined ? undefined : String(id); +} + /** Best-effort row-count extraction from a driver's delete result. */ function countDeleted(res: unknown): number | undefined { if (typeof res === 'number') return res;