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

.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 } } });
  }