fix(objectql): archive 冷侧 keep prune 尊重 #4747 的 teardown abort 位 (#6397)

PR #5956 给 archiveObject() 的批循环补上了 abort 检查,循环之后那条腿
没跟上:批循环刚因为读到 aborted === true 而 break,紧接着仍会向正在
关闭的 cold datasource 发一次 keep 保留期的谓词 DELETE。teardown 落在
最后一批热删里时同样如此 —— 那时循环按短页正常退出,不会再读一次
abort 位。

冷侧 prune 是纯保留期回收,不受「归档成功才热删」的配对约束,推迟到
下一轮 sweep 不留任何不一致。


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

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

Author: baozhoutao

.changeset/archive-keep-prune-abort-leg.md

@@ -0,0 +1,9 @@
+---
+'@objectstack/objectql': patch
+---
+
+Lifecycle Archiver 的冷侧 `keep` prune 现在也尊重 #4747 的 teardown abort 位。
+
+PR #5956 给 `archiveObject()` 的批循环补上了 abort 检查,但循环**之后**那条腿 —— `archive.keep` 保留期在归档库上的谓词 DELETE —— 没跟上:批循环刚因为读到 `aborted === true` 而 break,紧接着仍会向正在关闭的 cold datasource 发一次 `deleteMany`。teardown 落在最后一批的热删里时同样如此,那时循环是按短页正常退出的,连再读一次 abort 位的机会都没有。
+
+冷侧 prune 是纯保留期回收,不像循环内的 `upsert` → `bulkDelete` 那样受「归档成功才热删」的配对约束,推迟到下一轮 sweep 不留任何不一致 —— 下一轮按同一个 `keep` 推出同一个 cutoff,清同一批行。

packages/objectql/src/lifecycle/lifecycle-service.test.ts

@@ -1643,10 +1643,16 @@ describe('LifecycleService teardown (#4747)', () => {
     const pageReads: number[] = [];
     const copied: string[] = [];
     const hotDeleted: Array<Array<string | number>> = [];
+    // [#5966] The cold-side `keep` prune is the loop's successor leg, so it is
+    // recorded the same way the loop's legs are — with the predicate it sent,
+    // not just a count, so "the prune ran" and "the prune ran on the right
+    // cutoff" are separable assertions.
+    const coldPruned: Array<Record<string, unknown> | undefined> = [];
     return {
       pageReads,
       copied,
       hotDeleted,
+      coldPruned,
       remaining: () => remaining,
       hot: {
         name: 'default',
@@ -1672,7 +1678,10 @@ describe('LifecycleService teardown (#4747)', () => {
           return row;
         },
         bulkDelete: async () => {},
-        deleteMany: async () => 0,
+        deleteMany: async (_object: string, query?: Record<string, unknown>) => {
+          coldPruned.push(query);
+          return 0;
+        },
       },
     };
   }
@@ -1727,6 +1736,109 @@ describe('LifecycleService teardown (#4747)', () => {
     expect(report.swept.find((e) => e.policy === 'archive')?.archived).toBe(1000);
   });
 
+  /**
+   * [#5966] The leg AFTER the loop. #5755 stopped the batch loop; the cold-side
+   * `keep` prune sits past its exit and had no check of its own, so teardown
+   * landing anywhere inside `archiveObject` still ended with one predicate
+   * DELETE at the cold datasource.
+   *
+   * The distinction that makes this worth pinning separately: by the time the
+   * prune is reached, the abort bit has already been READ — either the loop
+   * broke on it, or a leg the loop issued raised it. Continuing is a decision
+   * the code makes with the answer in hand, not an await that merely straddled
+   * teardown. The three tests below cover the prune's two ends and both ways
+   * the loop can hand control to it.
+   */
+  const KEEP_OBJ: LifecycleObjectLike = {
+    name: 'sys_audit_log',
+    lifecycle: {
+      class: 'audit',
+      retention: { maxAge: '90d' },
+      // The delta from ARCHIVED_OBJ is `keep`, and only `keep`: the prune leg
+      // does not run at all without it, which is why #5755's fixture omitted it.
+      archive: { after: '90d', to: 'archive', keep: '365d' },
+    } as any,
+  };
+
+  it('a `keep` prune nobody calls off runs, on the cutoff `keep` declares', async () => {
+    // The control: with the abort bit down the prune is ordinary work, and the
+    // guard must not cost it. 700 rows drain inside the budget, so the loop
+    // exits on its short page — not on abort — and the prune follows it.
+    const pair = archivePair(700);
+    const { engine } = captureEngine([KEEP_OBJ], {
+      driver: pair.hot,
+      datasources: { archive: pair.cold },
+    });
+
+    const report = await service(engine).sweep();
+
+    expect(pair.pageReads).toHaveLength(2);
+    expect(pair.copied).toHaveLength(700);
+    expect(pair.remaining()).toHaveLength(0);
+    // Exactly one prune, carrying the `keep` cutoff (not `after`'s).
+    expect(pair.coldPruned).toEqual([{ where: { created_at: { $lt: isoCutoff('365d') } } }]);
+    expect(report.swept.find((e) => e.policy === 'archive')?.archived).toBe(700);
+  });
+
+  it('stop() mid-archive stops the prune too, not just the batch loop', async () => {
+    // Teardown lands while batch 2 copies, so the loop breaks at its head after
+    // READING `aborted === true` — and the very next statement used to send a
+    // predicate DELETE to the cold store the host is closing.
+    let svc!: LifecycleService;
+    const pair = archivePair(10_500, (copied) => {
+      if (copied === 501) svc.stop(); // first row of batch 2
+    });
+    const { engine } = captureEngine([KEEP_OBJ], {
+      driver: pair.hot,
+      datasources: { archive: pair.cold },
+    });
+    svc = service(engine);
+
+    const report = await svc.sweep();
+
+    expect(svc.stopped).toBe(true);
+    expect(pair.pageReads).toHaveLength(2); // #5755's guard, still holding
+    expect(pair.copied).toHaveLength(1000);
+    // The leg this test exists for: nothing at all is sent to the cold store
+    // after the loop reads the bit.
+    expect(pair.coldPruned).toEqual([]);
+    // Deferral, not loss: the rows the prune would have taken are still cold,
+    // and the archiving that DID complete is still reported.
+    expect(pair.remaining()).toHaveLength(9500);
+    expect(report.swept.find((e) => e.policy === 'archive')?.archived).toBe(1000);
+  });
+
+  it('stop() after the loop has made its last check still calls the prune off', async () => {
+    // The other way in, and the one the per-batch check cannot see: the backlog
+    // drains inside the budget, so the loop exits on `rows.length <
+    // ARCHIVE_BATCH_SIZE` and never re-reads the abort bit. Teardown lands in
+    // the final hot delete — after the loop's last check, before the prune —
+    // which is precisely the window #5755 left open.
+    let svc!: LifecycleService;
+    const pair = archivePair(700);
+    const hotBulkDelete = pair.hot.bulkDelete;
+    pair.hot.bulkDelete = async (object: string, ids: Array<string | number>) => {
+      await hotBulkDelete(object, ids);
+      if (pair.remaining().length === 0) svc.stop(); // the last batch just landed
+    };
+    const { engine } = captureEngine([KEEP_OBJ], {
+      driver: pair.hot,
+      datasources: { archive: pair.cold },
+    });
+    svc = service(engine);
+
+    const report = await svc.sweep();
+
+    expect(svc.stopped).toBe(true);
+    // The archive itself completed — every row copied and hot-deleted in pairs.
+    expect(pair.pageReads).toHaveLength(2);
+    expect(pair.hotDeleted.map((ids) => ids.length)).toEqual([500, 200]);
+    expect(pair.hotDeleted.flat()).toEqual(pair.copied);
+    expect(report.swept.find((e) => e.policy === 'archive')?.archived).toBe(700);
+    // …and the prune, the one leg still owed, is left for the next sweep.
+    expect(pair.coldPruned).toEqual([]);
+  });
+
   it('stop() then start() re-arms the service — teardown is not one-way', async () => {
     const { engine } = captureEngine([]);
     let audits = 0;

packages/objectql/src/lifecycle/lifecycle-service.ts

@@ -1072,7 +1072,22 @@ export class LifecycleService {
     }
 
     // Cold-side retention: `keep` bounds the archive itself.
-    if (archive.keep && typeof cold.deleteMany === 'function') {
+    //
+    // [#4747] Leg boundary, after the batch loop — the last leg `archiveObject`
+    // can issue, and the one the per-batch check above does not reach. The loop
+    // may have just broken BECAUSE it read `aborted === true`, so firing a
+    // predicate DELETE at the cold datasource here is not a race teardown lost:
+    // it is a write issued by code that had already been told the engine is
+    // going away. That is what separates this leg from a lone `await` sitting
+    // between two checkpoints (the pre-#5194 reap shape) — there, nothing had
+    // observed the bit, and carrying on was not a decision.
+    //
+    // Deferring costs nothing. The cold prune is pure retention reclaim, not
+    // half of a pair: unlike the loop's `upsert` → `bulkDelete`, which must
+    // finish so the Archiver never hot-deletes a row the cold store has not
+    // taken, nothing is left inconsistent by skipping it. The next sweep
+    // re-derives the same cutoff from the same `keep` and prunes the same rows.
+    if (!this.abort.aborted && archive.keep && typeof cold.deleteMany === 'function') {
       const keepCutoff = new Date(this.now() - parseLifecycleDuration(archive.keep)).toISOString();
       await cold.deleteMany(object, { where: { created_at: { $lt: keepCutoff } } });
     }