diff --git a/.changeset/lifecycle-archive-abort-check.md b/.changeset/lifecycle-archive-abort-check.md new file mode 100644 index 0000000000..6d41ce83b9 --- /dev/null +++ b/.changeset/lifecycle-archive-abort-check.md @@ -0,0 +1,20 @@ +--- +'@objectstack/objectql': patch +--- + +lifecycle: the Archiver's batch loop now honours the teardown abort bit (#4747) + +`LifecycleService.stop()` raises an abort bit that the sweep checks at each leg +boundary — between objects, and (since #5753) between reap pages. The Archiver's +own batch loop did not check it, so teardown landing mid-archive still ran the +remaining batches out: up to 20 × 500 rows of `find` + per-row `upsert` + +`bulkDelete` issued across two datasources the host is already closing. + +It now breaks at the batch boundary like the reap loop does. The "hot-delete only +what the cold store took" safety rule is unaffected: the batch in flight finishes +its `upsert` → `bulkDelete` pair, and batches not yet begun are left for the next +sweep to re-read. + +Not reachable in any of today's deployments — the Archiver only engages once an +object declares `archive` with a provisioned cold datasource, and no platform +object does. This is prevention for the first deployment that declares one. diff --git a/packages/objectql/src/lifecycle/lifecycle-service.test.ts b/packages/objectql/src/lifecycle/lifecycle-service.test.ts index c5adf8aa3b..f0fa3a905d 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.test.ts +++ b/packages/objectql/src/lifecycle/lifecycle-service.test.ts @@ -1612,6 +1612,121 @@ describe('LifecycleService teardown (#4747)', () => { expect(svc.stopped).toBe(true); }); + /** + * [#5755] The Archiver's batch loop is the reap loop's twin: same 20-page + * budget, same "teardown lands mid-loop" exposure — except every page costs + * a read AND up to 500 writes on a SECOND datasource. These two tests pin the + * loop's two ends: the budget it runs when nothing calls it off, and the stop + * it makes when something does. + */ + const ARCHIVED_OBJ: LifecycleObjectLike = { + name: 'sys_audit_log', + lifecycle: { + class: 'audit', + retention: { maxAge: '90d' }, + // No `keep`: the cold-side prune is a single call AFTER the loop, so + // declaring it here would assert a leg these tests do not govern. + archive: { after: '90d', to: 'archive' }, + } as any, + }; + + /** + * A counting hot/cold pair. Every leg the archive loop can issue — the page + * read, the per-row copy, the hot delete — is recorded, so "the loop stopped" + * is observable per leg instead of inferred from a row count. + */ + function archivePair(rowCount: number, onCopy?: (copied: number) => void) { + let remaining: Array> = []; + for (let i = 0; i < rowCount; i++) { + remaining.push({ id: `r${i}`, created_at: '2020-01-01T00:00:00.000Z' }); + } + const pageReads: number[] = []; + const copied: string[] = []; + const hotDeleted: Array> = []; + return { + pageReads, + copied, + hotDeleted, + remaining: () => remaining, + hot: { + name: 'default', + find: async (_object: string, query: any) => { + const limit = (query?.limit as number) ?? remaining.length; + pageReads.push(limit); + return remaining.slice(0, limit); + }, + upsert: async () => ({}), + bulkDelete: async (_object: string, ids: Array) => { + hotDeleted.push(ids); + const gone = new Set(ids); + remaining = remaining.filter((r) => !gone.has(r.id as string)); + }, + deleteMany: async () => 0, + }, + cold: { + name: 'archive', + find: async () => [], + upsert: async (_object: string, row: Record) => { + copied.push(row.id as string); + onCopy?.(copied.length); + return row; + }, + bulkDelete: async () => {}, + deleteMany: async () => 0, + }, + }; + } + + it('an archive nobody calls off runs its whole 20-batch budget', async () => { + // The control for the test below: with the abort bit down, the loop is + // bounded only by ARCHIVE_MAX_BATCHES_PER_SWEEP (20 × 500), and the rest of + // the backlog waits for the next sweep. + const pair = archivePair(10_500); + const { engine } = captureEngine([ARCHIVED_OBJ], { + driver: pair.hot, + datasources: { archive: pair.cold }, + }); + + const report = await service(engine).sweep(); + + expect(pair.pageReads).toHaveLength(20); + expect(pair.copied).toHaveLength(10_000); + expect(pair.hotDeleted).toHaveLength(20); + expect(pair.remaining()).toHaveLength(500); + expect(report.swept.find((e) => e.policy === 'archive')?.archived).toBe(10_000); + }); + + it('stop() mid-archive ends the batch loop instead of running out the page budget', async () => { + // The #4747 shape on the archive side: teardown lands while batch 2 is + // copying, so batches 3…20 must issue nothing at all — no hot read, no cold + // copy, no hot delete — at two datasources 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([ARCHIVED_OBJ], { + driver: pair.hot, + datasources: { archive: pair.cold }, + }); + svc = service(engine); + + const report = await svc.sweep(); + + expect(svc.stopped).toBe(true); + // Two pages read, not 20 — the loop stopped at the boundary after the batch + // that was in flight, and never asked the hot store for page 3. + expect(pair.pageReads).toHaveLength(2); + expect(pair.copied).toHaveLength(1000); + // The safety rule survives the interruption: the batch in flight completed + // its pair, so every id the cold store took was hot-deleted and no id was + // hot-deleted that the cold store had not taken — no half batch either way. + expect(pair.hotDeleted.map((ids) => ids.length)).toEqual([500, 500]); + expect(pair.hotDeleted.flat()).toEqual(pair.copied); + // …and the batches never begun are still hot, for the next sweep to move. + expect(pair.remaining()).toHaveLength(9500); + expect(report.swept.find((e) => e.policy === 'archive')?.archived).toBe(1000); + }); + 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 862d0519b6..dc762d229d 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.ts +++ b/packages/objectql/src/lifecycle/lifecycle-service.ts @@ -1045,6 +1045,19 @@ export class LifecycleService { const cutoff = new Date(this.now() - parseLifecycleDuration(archive.after)).toISOString(); let archived = 0; for (let batch = 0; batch < ARCHIVE_MAX_BATCHES_PER_SWEEP; batch++) { + // [#4747] Leg boundary, per batch — the same check the reap loop makes + // (see {@link batchedReap}). `sweep()` checks the abort bit between + // OBJECTS, which leaves one archive round as up to 20 pages of reads and + // writes across TWO datasources (a hot page read, a cold upsert per row, + // a hot bulk delete), i.e. ~10k operations issued at stores the host may + // already be closing — precisely what #4747 stopped. + // + // Breaking BETWEEN batches keeps the Archiver's safety rule intact + // ("hot-delete only what the cold store has taken"): a batch already in + // flight finishes its upsert → bulkDelete pair, and batches not yet begun + // are simply left for the next sweep, which re-reads them from the hot + // store unchanged. + if (this.abort.aborted) break; const rows = await hot.find(object, { where: { created_at: { $lt: cutoff } }, limit: ARCHIVE_BATCH_SIZE,