fix(objectql): compose reap guards by intersection instead of last-wins (#5535) (#5708)

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


Claude-Session: https://claude.ai/code/session_01V7WetGmnfoXNn8cLieKKmx

Co-authored-by: Claude <noreply@anthropic.com>

Author: os-zhuang

.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` 上追加注册,而不必担心顶掉字节回收。

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<Record<string, unknown>>) {
+    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<Record<string, unknown>>) => 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<Record<string, unknown>>) =>
+      idsOf(r).filter((id) => id !== 'f2');
+    const vetoF3 = async (_o: string, r: Array<Record<string, unknown>>) =>
+      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<Record<string, unknown>>) =>
+      idsOf(r).filter((id) => id !== 'f2'),
+    );
+    const second = vi.fn(async (_o: string, r: Array<Record<string, unknown>>) => 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<Record<string, unknown>>) => 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<Record<string, unknown>>) => 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<Record<string, unknown>>) => 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',