diff --git a/.changeset/archive-keep-prune-abort-leg.md b/.changeset/archive-keep-prune-abort-leg.md new file mode 100644 index 0000000000..3b48d39b00 --- /dev/null +++ b/.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,清同一批行。 diff --git a/packages/objectql/src/lifecycle/lifecycle-service.test.ts b/packages/objectql/src/lifecycle/lifecycle-service.test.ts index f0fa3a905d..97c1d78306 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.test.ts +++ b/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> = []; + // [#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 | 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) => { + 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) => { + 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; diff --git a/packages/objectql/src/lifecycle/lifecycle-service.ts b/packages/objectql/src/lifecycle/lifecycle-service.ts index dc762d229d..ad75ebf4ea 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.ts +++ b/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 } } }); }