diff --git a/.changeset/lifecycle-unguarded-reap-batching.md b/.changeset/lifecycle-unguarded-reap-batching.md new file mode 100644 index 0000000000..20473d7e4b --- /dev/null +++ b/.changeset/lifecycle-unguarded-reap-batching.md @@ -0,0 +1,37 @@ +--- +'@objectstack/objectql': patch +--- + +fix(objectql): bound every lifecycle reap, not just the guarded ones + +`LifecycleService.reap()` issued a single unlimited `delete(..., { multi: true })` +per sweep for any object without a registered reap guard — no limit, no paging. +Batching existed only on the two side paths (`guardedReap` and the Archiver, +both 500 × 20 per sweep), whose comment gives the reason plainly: "bound one +sweep's work, drain the backlog across sweeps". That reason never depended on a +guard being registered. + +Steady state was fine — an hourly sweep deletes a small increment. The cost +landed exactly once per table, on the first sweep after `retention` is declared +on a table that already holds history: one DELETE scanning every historical row, +which on SQLite holds the whole database's write lock for its duration and on +Postgres arrives later as autovacuum debt. + +Unguarded reaps now run the same batched machinery the guarded ones do — an +object with no guard is simply the empty guard intersection, which confirms +every candidate row — so there is one reap path rather than two parallel ones. +Candidates are read a page at a time and deleted by id, at most 500 × 20 rows +per object per sweep, with the remainder draining on later sweeps. Cutoff and +`retention.onlyWhen` predicates are unchanged; they now select the candidate +read. The sweep report's `deleted` count reflects rows actually deleted this +sweep. + +Two consequences worth knowing: + +- A reap fires one `afterDelete` hook per reaped row instead of one per object + per sweep. Every lifecycle-declaring platform object is in the audit writer's + `SKIP_OBJECTS`, and `sys_file` already reaped per id via its guards, so no + object in the platform changes its audit output. +- An engine that does not implement `find` cannot page and keeps the previous + single bulk DELETE, so retention enforcement never silently stops on it. Every + real engine implements `find`. diff --git a/packages/objectql/src/lifecycle/lifecycle-service.test.ts b/packages/objectql/src/lifecycle/lifecycle-service.test.ts index f31b6cf90e..c5adf8aa3b 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.test.ts +++ b/packages/objectql/src/lifecycle/lifecycle-service.test.ts @@ -361,23 +361,35 @@ describe('LifecycleService.sweep — reap guard', () => { expect(report.errors).toEqual([{ object: 'sys_file', error: 'storage unreachable' }]); }); - it('a guard on one object never changes the blind reap of others (regression pin)', async () => { - const { engine, deletes } = captureEngine( + it('a guard on one object is never consulted for the reap of others (regression pin)', async () => { + // [#5194] This pin used to read "sys_job_run: classic blind reap" and + // assert the unbounded `multi: true` DELETE — the exact limb #5194 removed, + // so keeping the assertion would have kept it passing on an empty result. + // The property it exists for is untouched and now sharper: an UNGUARDED + // object runs the same batched machinery with an empty guard list, so + // "can another object's guard reach this reap?" is a live question rather + // than a moot one. It must not, and it is never even consulted. + const guard = vi.fn(async () => []); + const { engine, deletes, finds } = captureEngine( [ ...guarded, { name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }, ], - { findImpl: () => [] }, + { findImpl: (object) => (object === 'sys_job_run' ? [{ id: 'j1' }] : []) }, ); const svc = service(engine); - svc.registerReapGuard('sys_file', async () => []); + svc.registerReapGuard('sys_file', guard); const report = await svc.sweep(); - // sys_file: no candidates → no delete. sys_job_run: classic blind reap. - expect(deletes).toHaveLength(1); - expect(deletes[0].object).toBe('sys_job_run'); - expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('30d') } }); + // sys_file: no candidates → nothing to confirm, nothing deleted. + expect(guard).not.toHaveBeenCalled(); + // sys_job_run: read on its own cutoff, deleted by id, guard-free. + expect(finds.map((f) => f.object)).toEqual(['sys_file', 'sys_job_run']); + expect(finds[1].where).toEqual({ created_at: { $lt: isoCutoff('30d') } }); + expect(deletes).toEqual([ + { object: 'sys_job_run', where: { id: 'j1' }, multi: true, context: { isSystem: true, positions: [], permissions: [] } }, + ]); expect(report.errors).toEqual([]); }); @@ -625,6 +637,141 @@ describe('LifecycleService.sweep — reap guard composition', () => { }); }); +describe('LifecycleService.sweep — unguarded reap batching (#5194)', () => { + /** An ordinary lifecycle object: retention declared, NO reap guard. */ + const plain: LifecycleObjectLike[] = [ + { name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }, + ]; + + /** + * An engine whose rows really disappear, so "what one sweep did" and "what + * the next sweep still finds" are both observable. Deletion is by id through + * a Map (O(1)) — the ceiling case drives 10k deletes and an array scan per + * delete would make this test quadratic. + */ + function storeEngine(objects: LifecycleObjectLike[], rowCount: number) { + const store = new Map>(); + for (let i = 0; i < rowCount; i++) store.set(`r${i}`, { id: `r${i}`, created_at: '2020-01-01T00:00:00Z' }); + const captured = captureEngine(objects, { + findImpl: (_object, options) => { + const limit = (options?.limit as number) ?? store.size; + const page: Array> = []; + for (const row of store.values()) { + if (page.length >= limit) break; + page.push(row); + } + return page; + }, + deleteImpl: (_object, options) => { + const id = options?.where?.id as string | undefined; + const existed = id !== undefined && store.delete(id); + return { deletedCount: existed ? 1 : 0 }; + }, + }); + return { ...captured, store }; + } + + it('bounds the FIRST sweep over a large backlog, and drains the rest on later sweeps', async () => { + // The #5194 scenario: retention declared on a table that already holds + // history. Before this change that first sweep was ONE unbounded DELETE + // over every historical row — a long write transaction (SQLite locks the + // whole database for it). It is now 20 pages of 500, by id, and the + // remainder waits for the next sweep. + const CEILING = 500 * 20; + const { engine, deletes, finds, store } = storeEngine(plain, CEILING + 7); + const svc = service(engine); + + const first = await svc.sweep(); + + expect(finds).toHaveLength(20); // exactly the per-sweep page budget + expect(finds.every((f) => f.limit === 500)).toBe(true); + expect(deletes).toHaveLength(CEILING); // …and not one row more + expect(first.swept[0].deleted).toBe(CEILING); // the report says what really happened + expect(store.size).toBe(7); // backlog left standing, not lost + + // Every delete named ONE row by scalar id — never a predicate, so each is a + // short transaction and each gets the by-id cascade path. + expect(deletes.every((d) => typeof d.where?.id === 'string')).toBe(true); + expect(deletes.every((d) => Object.keys(d.where).length === 1)).toBe(true); + + const second = await svc.sweep(); + + expect(second.swept[0].deleted).toBe(7); // the remainder drains next sweep + expect(store.size).toBe(0); + expect(finds).toHaveLength(21); // 20 + one short page, which ends the pass + }); + + it('leaves the steady state alone: a small increment is one page and one report line', async () => { + const { engine, deletes, finds, store } = storeEngine(plain, 3); + + const report = await service(engine).sweep(); + + expect(finds).toHaveLength(1); // short page ⇒ no second read + expect(finds[0].where).toEqual({ created_at: { $lt: isoCutoff('30d') } }); + expect(finds[0].context).toEqual({ isSystem: true, positions: [], permissions: [] }); + expect(deletes.map((d) => d.where)).toEqual([{ id: 'r0' }, { id: 'r1' }, { id: 'r2' }]); + expect(store.size).toBe(0); + expect(report.swept).toEqual([ + { object: 'sys_job_run', class: 'telemetry', policy: 'retention', cutoff: isoCutoff('30d'), deleted: 3 }, + ]); + expect(report.errors).toEqual([]); + }); + + it('still honours retention.onlyWhen — the predicate moves to the candidate READ', async () => { + const onlyWhen: LifecycleObjectLike[] = [ + { + name: 'sys_automation_run', + lifecycle: { + class: 'telemetry', + retention: { maxAge: '30d', onlyWhen: { status: { $in: ['completed', 'failed'] } } }, + } as unknown as LifecycleObjectLike['lifecycle'], + }, + ]; + const { engine, finds, deletes } = storeEngine(onlyWhen, 2); + + await service(engine).sweep(); + + // The scope narrows which rows are CANDIDATES; it is not silently dropped + // now that the delete addresses rows by id. + expect(finds[0].where).toEqual({ + created_at: { $lt: isoCutoff('30d') }, + status: { $in: ['completed', 'failed'] }, + }); + expect(deletes.map((d) => d.where)).toEqual([{ id: 'r0' }, { id: 'r1' }]); + }); + + it('an engine that cannot read rows keeps the single bulk DELETE — retention never just stops', async () => { + // `find` is optional on LifecycleEngineLike. Batching needs it; enforcement + // does not. An engine without it degrades to the pre-#5194 unbounded delete + // rather than losing the policy — the fail-safe for guards (skip) would be + // the wrong trade here, because no guard is waiting to confirm anything. + const { engine, deletes } = captureEngine(plain); // no findImpl → no engine.find + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(1); + expect(deletes[0].multi).toBe(true); + expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('30d') } }); + expect(report.skipped).toEqual([]); + expect(report.swept[0].deleted).toBe(3); // whatever the driver reported + }); + + it('never turns an id-less row into a predicate delete', async () => { + // `where: { id: undefined }` + `multi: true` is NOT a by-id delete: the + // engine finds no scalar id, routes to deleteMany, and runs the batch's + // whole cutoff predicate. With zero guards nothing else narrows the page, + // so this is the one place that invariant can be enforced. + const { engine, deletes } = captureEngine(plain, { + findImpl: () => [{ id: 'r0' }, { id: null }, { created_at: '2020-01-01T00:00:00Z' }], + }); + + const report = await service(engine).sweep(); + + expect(deletes.map((d) => d.where)).toEqual([{ id: 'r0' }]); + expect(report.swept[0].deleted).toBe(1); + }); +}); + describe('LifecycleService.sweep — Archiver (P3)', () => { const AUDIT_OBJ: LifecycleObjectLike = { name: 'sys_audit_log', @@ -816,9 +963,15 @@ describe('LifecycleService.sweep — governance (P4)', () => { }); it('tenant-scoped overrides sweep each tenant on its own window and everyone else globally', async () => { - const { engine, deletes } = captureEngine([TELEMETRY_OBJ]); - (engine as any).find = async (object: string) => - object === 'sys_organization' ? [{ id: 'org_reg' }, { id: 'org_plain' }] : []; + // [#5194] One candidate row per pass, so each pass is observable both in + // the predicate it reads with and in the by-id delete it then issues. + const { engine, deletes, finds } = captureEngine([TELEMETRY_OBJ], { + findImpl: (object, options) => { + if (object === 'sys_organization') return [{ id: 'org_reg' }, { id: 'org_plain' }]; + const org = (options?.where as Record | undefined)?.organization_id; + return [{ id: org === 'org_reg' ? 'tenant_row' : 'global_row' }]; + }, + }); const settings = fakeSettings( {}, { org_reg: { retention_overrides: { sys_job_run: { maxAge: '2y' } } } }, @@ -826,31 +979,42 @@ describe('LifecycleService.sweep — governance (P4)', () => { await service(engine, { getSettings: () => settings }).sweep(); - // One tenant-scoped delete on the regulated tenant's 2y window… - expect(deletes[0].where).toEqual({ + // The per-pass predicate is carried by the candidate READ now — the passes + // themselves, and their windows, are unchanged. + const reaps = finds.filter((f) => f.object === 'sys_job_run'); + // One tenant-scoped pass on the regulated tenant's 2y window… + expect(reaps[0].where).toEqual({ created_at: { $lt: isoCutoff('2y') }, organization_id: 'org_reg', }); // …then the global 30d pass excluding it but INCLUDING NULL-org rows. - expect(deletes[1].where).toEqual({ + expect(reaps[1].where).toEqual({ created_at: { $lt: isoCutoff('30d') }, $or: [{ organization_id: { $nin: ['org_reg'] } }, { organization_id: null }], }); - expect(deletes).toHaveLength(2); + expect(reaps).toHaveLength(2); + expect(deletes.map((d) => d.where)).toEqual([{ id: 'tenant_row' }, { id: 'global_row' }]); }); it('retention.onlyWhen survives tenant-scoped overrides on every pass', async () => { - const { engine, deletes } = captureEngine([ + const { engine, deletes, finds } = captureEngine( + [ + { + name: 'sys_automation_run', + lifecycle: { + class: 'telemetry', + retention: { maxAge: '30d', onlyWhen: { status: { $in: ['completed', 'failed'] } } }, + } as any, + }, + ], { - name: 'sys_automation_run', - lifecycle: { - class: 'telemetry', - retention: { maxAge: '30d', onlyWhen: { status: { $in: ['completed', 'failed'] } } }, - } as any, + findImpl: (object, options) => { + if (object === 'sys_organization') return [{ id: 'org_reg' }]; + const org = (options?.where as Record | undefined)?.organization_id; + return [{ id: org === 'org_reg' ? 'tenant_row' : 'global_row' }]; + }, }, - ]); - (engine as any).find = async (object: string) => - object === 'sys_organization' ? [{ id: 'org_reg' }] : []; + ); const settings = fakeSettings( {}, { org_reg: { retention_overrides: { sys_automation_run: { maxAge: '2y' } } } }, @@ -858,18 +1022,23 @@ describe('LifecycleService.sweep — governance (P4)', () => { await service(engine, { getSettings: () => settings }).sweep(); + // [#5194] `onlyWhen` narrows which rows are CANDIDATES; it rides the read + // that selects them, on every pass, and is not dropped now that the delete + // addresses rows by id. const predicate = { status: { $in: ['completed', 'failed'] } }; - expect(deletes[0].where).toEqual({ + const reaps = finds.filter((f) => f.object === 'sys_automation_run'); + expect(reaps[0].where).toEqual({ created_at: { $lt: isoCutoff('2y') }, organization_id: 'org_reg', ...predicate, }); - expect(deletes[1].where).toEqual({ + expect(reaps[1].where).toEqual({ created_at: { $lt: isoCutoff('30d') }, $or: [{ organization_id: { $nin: ['org_reg'] } }, { organization_id: null }], ...predicate, }); - expect(deletes).toHaveLength(2); + expect(reaps).toHaveLength(2); + expect(deletes.map((d) => d.where)).toEqual([{ id: 'tenant_row' }, { id: 'global_row' }]); }); it('raises quota and growth alerts (observe-only — no extra deletes)', async () => { @@ -1035,9 +1204,9 @@ describe('LifecycleService — retention floors (#5195)', () => { }); it('floors a TENANT-scoped override too — the same door one scope down', async () => { - const { engine, deletes } = captureEngine([QUEUE_LIKE]); - (engine as any).find = async (object: string) => - object === 'sys_organization' ? [{ id: 'org_fast' }] : []; + const { engine, finds } = captureEngine([QUEUE_LIKE], { + findImpl: (object) => (object === 'sys_organization' ? [{ id: 'org_fast' }] : []), + }); const settings = fakeSettings( {}, { org_fast: { retention_overrides: { app_work_queue: { maxAge: '1h' } } } }, @@ -1049,7 +1218,9 @@ describe('LifecycleService — retention floors (#5195)', () => { // The tenant pass falls back to the window that DID pass the floor (7d), // so no tenant can shorten its way past another package's contract. - expect(deletes[0].where).toEqual({ + // [#5194] The floored window is observable on the candidate read. + const reaps = finds.filter((f) => f.object === 'app_work_queue'); + expect(reaps[0].where).toEqual({ created_at: { $lt: isoCutoff('7d') }, organization_id: 'org_fast', status: 'done', @@ -1397,6 +1568,50 @@ describe('LifecycleService teardown (#4747)', () => { expect(report.danglingReferences?.aborted).toBe(true); }); + it('stop() mid-reap ends the paging loop instead of running out the page budget', async () => { + // [#5194] `sweep()` checks the abort bit between OBJECTS. That used to be + // the whole story for an unguarded reap, which was a single `await` between + // two such checks; it is now up to 20 pages of reads and deletes, so the + // loop has to check too — otherwise teardown keeps pushing writes at a + // datasource the host is closing, which is the #4747 defect itself. + const store = new Map>(); + for (let i = 0; i < 5000; i++) store.set(`r${i}`, { id: `r${i}`, created_at: '2020-01-01T00:00:00Z' }); + const { engine, deletes } = captureEngine( + [{ name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }], + { + findImpl: (_object, options) => { + const limit = (options?.limit as number) ?? store.size; + const page: Array> = []; + for (const row of store.values()) { + if (page.length >= limit) break; + page.push(row); + } + return page; + }, + deleteImpl: (_object, options) => { + const id = options?.where?.id as string | undefined; + if (id !== undefined) store.delete(id); + return { deletedCount: 1 }; + }, + }, + ); + const svc = service(engine); + // Teardown lands while the first page is being deleted. + const original = engine.delete.bind(engine); + engine.delete = async (object: string, options: unknown) => { + svc.stop(); + return original(object, options); + }; + + await svc.sweep(); + + // The page in flight finishes (it is already confirmed work), and the loop + // stops there rather than reading and deleting 19 more pages. + expect(deletes).toHaveLength(500); + expect(store.size).toBe(4500); + expect(svc.stopped).toBe(true); + }); + 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 44b0779dea..862d0519b6 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.ts +++ b/packages/objectql/src/lifecycle/lifecycle-service.ts @@ -25,11 +25,24 @@ import type { * * Design constraints (ADR-0057 §3.3): * - One implementation, owned here — not N per-plugin sweepers. - * - Sweeps run under a system context (cross-tenant operator policy) and - * use bulk `multi: true` deletes, so at most ONE afterDelete hook fires - * per object per sweep — audit sees an aggregate, never per-row noise - * (telemetry-class sys_* objects are additionally in the audit writer's - * SKIP_OBJECTS, so they produce no audit rows at all). + * - Sweeps run under a system context (cross-tenant operator policy). + * - [#5194] Every reap is BOUNDED. Candidates are read a page at a time and + * deleted by id — at most {@link REAP_BATCH_SIZE} × + * {@link REAP_MAX_BATCHES_PER_SWEEP} rows per object per sweep, with the + * remainder draining across later sweeps. Unguarded objects used to issue + * one `multi: true` DELETE with no limit instead: invisible in the steady + * state (hourly sweeps delete a small increment), and a table-scanning long + * write transaction exactly once per table — the first sweep after a + * retention is declared on an already-large table. SQLite holds the whole + * database's write lock for the duration of that one statement; Postgres + * takes it as autovacuum debt. + * + * Stated plainly because it is the cost of that bound: a reap now fires one + * afterDelete hook PER REAPED ROW, not one per object per sweep. That is + * free for today's population — every lifecycle-declaring platform object + * is in the audit writer's SKIP_OBJECTS (telemetry/transient plumbing), so + * it produces no audit rows either way, and `sys_file`, the one that is + * audited, already reaped per id because it carries reap guards. * - A sweep failure is logged and isolated; it never throws into the * scheduler and never blocks other objects' policies. */ @@ -300,10 +313,16 @@ interface ArchiveCapableDriver { const ARCHIVE_BATCH_SIZE = 500; const ARCHIVE_MAX_BATCHES_PER_SWEEP = 20; -/** Guarded reap batching — same posture as the Archiver: bound one sweep's - * work, drain the backlog across sweeps. */ -const REAP_GUARD_BATCH_SIZE = 500; -const REAP_GUARD_MAX_BATCHES_PER_SWEEP = 20; +/** + * Reap batching — same posture as the Archiver: bound one sweep's work, drain + * the backlog across sweeps. + * + * [#5194] These govern EVERY reap, not just guarded ones. The reasoning the + * Archiver's constants carry ("bound one sweep's work") never depended on a + * guard being registered; the unguarded path simply had not had it applied. + */ +const REAP_BATCH_SIZE = 500; +const REAP_MAX_BATCHES_PER_SWEEP = 20; /** * Reap guard (ADR-0057 amendment): a domain callback consulted by the Reaper @@ -1071,7 +1090,8 @@ export class LifecycleService { // A guarded object is NEVER blind-deleted: without row reads the guard // cannot confirm, so the reap is skipped (fail-safe), not degraded. const guards = this.reapGuardsFor(object); - if (guards.length > 0 && typeof engine.find !== 'function') { + const canReadRows = typeof engine.find === 'function'; + if (guards.length > 0 && !canReadRows) { if (!report.skipped.some((s) => s.object === object && s.reason === 'reap-guard-unsupported')) { report.skipped.push({ object, reason: 'reap-guard-unsupported' }); } @@ -1083,9 +1103,21 @@ export class LifecycleService { if (n === undefined) total = undefined; else if (total !== undefined) total += n; }; + // [#5194] One reap path for every object. Zero guards is not a different + // algorithm — it is the empty intersection, i.e. "every candidate row is + // confirmed" — so the batching, the per-sweep ceiling and the by-id deletes + // are identical either way, and there is exactly one place where a reap + // decides what to delete. + // + // The fallback below is NOT the unguarded path; it is the no-`find` path. + // `LifecycleEngineLike.find` is optional (a two-method test double is a + // legal engine here), and an engine that cannot read rows cannot page + // through them — so it keeps the pre-#5194 single unbounded DELETE rather + // than losing retention enforcement entirely. Every real engine has `find` + // (`ObjectQL.find`, wired in `plugin.ts`), so production always batches. const reapWhere = async (where: Record): Promise => - guards.length > 0 - ? this.guardedReap(engine, object, guards, where) + canReadRows + ? this.batchedReap(engine, object, guards, where) : countDeleted(await engine.delete(object, { where, multi: true, context: { ...SYSTEM_CTX } })); if (tenantWindows.length === 0) { @@ -1126,13 +1158,24 @@ export class LifecycleService { } /** - * Guarded reap: fetch candidate rows in batches, let every guard confirm - * (after performing external cleanup) or veto each, delete only the ids - * ALL of them confirmed. A guard error propagates to the per-object handler - * in `sweep()` — an erroring guard must never fail open into deletion. A - * batch that isn't fully confirmed ends the pass: vetoed rows still match - * the cutoff filter and would be re-fetched forever; the next sweep retries - * them. + * Batched reap: fetch candidate rows a page at a time, let every registered + * guard confirm (after performing external cleanup) or veto each, delete + * only the ids ALL of them confirmed — by id, page after page, up to + * {@link REAP_MAX_BATCHES_PER_SWEEP} pages. A guard error propagates to the + * per-object handler in `sweep()` — an erroring guard must never fail open + * into deletion. A batch that isn't fully confirmed ends the pass: vetoed + * rows still match the cutoff filter and would be re-fetched forever; the + * next sweep retries them. + * + * [#5194] `guards` MAY BE EMPTY, and that is the ordinary case — an object + * with no guard registered is the empty intersection, which confirms every + * candidate. The guard loop then simply does not execute, and what remains is + * exactly the bound this method exists to impose: read ≤ {@link + * REAP_BATCH_SIZE} rows, delete them by id, stop after + * {@link REAP_MAX_BATCHES_PER_SWEEP} pages and let the next sweep continue. + * The alternative — a second, guard-free batching routine beside this one — + * would be two implementations of one policy, drifting apart at the first + * change to either. * * [#5535] The intersection is computed as a narrowing pipeline rather than * N independent verdicts unioned at the end, because a guard's confirmation @@ -1143,21 +1186,38 @@ export class LifecycleService { * everything ends the batch before the rest are called at all. The delete * set is the same whatever the registration order. */ - private async guardedReap( + private async batchedReap( engine: LifecycleEngineLike, object: string, guards: readonly LifecycleReapGuard[], where: Record, ): Promise { let total = 0; - for (let batch = 0; batch < REAP_GUARD_MAX_BATCHES_PER_SWEEP; batch++) { + for (let batch = 0; batch < REAP_MAX_BATCHES_PER_SWEEP; batch++) { + // [#4747] Leg boundary, per page. `sweep()` checks the abort bit between + // OBJECTS; before #5194 an unguarded reap was a single `await` between + // two such checks, so that was the whole story. A reap is now up to + // REAP_MAX_BATCHES_PER_SWEEP pages of reads and deletes, and pushing them + // at a datasource the host is closing is precisely what #4747 stopped. + if (this.abort.aborted) break; const rows = await engine.find!(object, { where, - limit: REAP_GUARD_BATCH_SIZE, + limit: REAP_BATCH_SIZE, context: { ...SYSTEM_CTX }, }); if (!rows?.length) break; - let confirmed = rows; + // [#5194] A row with no usable id is dropped before anything is asked + // about it or done to it. It cannot be deleted by id, and it must never + // reach the delete below: `where: { id: undefined }` with `multi: true` + // is not a by-id delete at all — the engine's dispatch reads no scalar + // id, routes to `deleteMany`, and the predicate it would run is the + // batch's whole cutoff filter. The guard intersection used to drop such + // rows as a side effect of matching ids (see {@link idKey}); with zero + // guards nothing narrows, so the invariant is stated here instead of + // being an emergent property of a loop that may not run. Dropping them + // pre-guard also keeps a guard from reclaiming bytes for a row that was + // never deletable — the same reason #5535 narrows before it asks. + let confirmed = rows.filter((row) => idKey(row?.id) !== undefined); for (const guard of guards) { const ids = new Set( (await guard(object, confirmed)).map(idKey).filter((k): k is string => k !== undefined), @@ -1179,7 +1239,7 @@ export class LifecycleService { }); } total += confirmed.length; - if (confirmed.length < rows.length || rows.length < REAP_GUARD_BATCH_SIZE) break; + if (confirmed.length < rows.length || rows.length < REAP_BATCH_SIZE) break; } return total; }