diff --git a/.changeset/lifecycle-reclaim-rotation-abort-legs.md b/.changeset/lifecycle-reclaim-rotation-abort-legs.md new file mode 100644 index 0000000000..f6f96153b8 --- /dev/null +++ b/.changeset/lifecycle-reclaim-rotation-abort-legs.md @@ -0,0 +1,14 @@ +--- +'@objectstack/objectql': patch +--- + +LifecycleService 的空间回收(VACUUM 级)与分片轮转(DROP 过期分片)现在也尊重 #4747 的 teardown abort 位。 + +PR #5956 / #5755 / PR #6397 依次给 reap 分页循环、Archiver 批循环和冷侧 `keep` prune 补上了 abort 判定,但 `sweep()` 里还剩两条同形的腿: + +- **空间回收**:abort 判定在对象循环的**头部**,teardown 落在**最后一个**声明对象的 reap 内时,`batchedReap` 因读到 `aborted === true` 而 break,对象循环随即自然结束(不再经过那个判定),控制流直接落到回收循环 —— 向正在关闭的 datasource 发一次 VACUUM 级操作。 +- **分片轮转**:对象同时声明 `ttl` 与 `storage.strategy: 'rotation'` 时,ttl reap 已经读过位并 break,返回后轮转仅由 strategy 与驱动能力把关,无判定地 DROP 过期物理分片。 + +两条腿都是「已经拿到答案之后作出的决定」,而不是恰好横跨 teardown 的一次 await。推迟均无代价:回收是纯粹的页面归还,不删任何行;轮转是 O(1) 的窗口回收,下一轮 sweep 用同一个 `shards × unit` 推出同一个窗口、清同一批分片 —— 至多晚一个 sweep 间隔。 + +仅声明 `rotation`(无 `ttl`)的一路行为不变:该形态下 `reapObject` 在轮转之前没有任何 await,位在结构上必为 false,而非「无人读过」。 diff --git a/packages/objectql/src/lifecycle/lifecycle-service.test.ts b/packages/objectql/src/lifecycle/lifecycle-service.test.ts index 97c1d78306..6f305b5061 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.test.ts +++ b/packages/objectql/src/lifecycle/lifecycle-service.test.ts @@ -1839,6 +1839,221 @@ describe('LifecycleService teardown (#4747)', () => { expect(pair.coldPruned).toEqual([]); }); + /** + * [#6412] The two legs left over after #5966. Both are the same shape as the + * cold prune above — control flow that has ALREADY read `aborted === true` + * and issues a destructive operation anyway — but they sit at two different + * scopes, so each gets its own fixture and its own control: + * + * - the space reclaim, AFTER `sweep()`'s object loop (VACUUM-class); + * - the shard rotation, inside `reapObject` after the ttl reap (DROPs + * expired physical shards). + */ + + /** A hot store a reap can page through, with the by-id delete it issues. */ + function reapStore(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:00.000Z', + expires_at: '2020-01-01T00:00:00.000Z', + }); + } + return { + store, + findImpl: (_object: string, options: any) => { + 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: string, options: any) => { + const id = options?.where?.id as string | undefined; + if (id !== undefined) store.delete(id); + return { deletedCount: 1 }; + }, + }; + } + + const REAPED_OBJ: LifecycleObjectLike = { + name: 'sys_job_run', + lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } as any, + }; + + it('a space reclaim nobody calls off runs, once per drained datasource', async () => { + // The control: with the abort bit down the reclaim is ordinary work, and the + // guard must not cost it. 700 rows drain in two pages, the reap reports rows + // deleted, and the datasource is vacuumed exactly once. + const hot = reapStore(700); + const reclaimSpace = vi.fn(async () => {}); + const { engine, deletes } = captureEngine([REAPED_OBJ], { + driver: { name: 'default', reclaimSpace }, + findImpl: hot.findImpl, + deleteImpl: hot.deleteImpl, + }); + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(700); // by-id, two pages + expect(hot.store.size).toBe(0); + expect(reclaimSpace).toHaveBeenCalledTimes(1); + expect(report.reclaimed).toEqual(['default']); + }); + + it("stop() inside the LAST object's reap calls the space reclaim off too", async () => { + // The window the per-object check cannot see. `sweep()` reads the abort bit + // at the TOP of the object loop, so teardown landing in the LAST declared + // object's reap never meets that check again: `batchedReap` breaks BECAUSE + // it read `aborted === true`, the object loop then ends normally rather than + // through the check, and the very next statement used to send a VACUUM to + // the datasource the host is closing. Probe evidence on #6412: + // `reclaims === ['vacuum']`. + const hot = reapStore(5_000); + const reclaimSpace = vi.fn(async () => {}); + const { engine, deletes } = captureEngine([REAPED_OBJ], { + driver: { name: 'default', reclaimSpace }, + findImpl: hot.findImpl, + deleteImpl: hot.deleteImpl, + }); + 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); + }; + + const report = await svc.sweep(); + + expect(svc.stopped).toBe(true); + // #5194's guard, still holding: the page in flight finishes, page 2 is never + // read — and the reap still reports rows deleted, so the driver IS a reclaim + // candidate. The bit, not the candidacy, is what calls the VACUUM off. + expect(deletes).toHaveLength(500); + expect(hot.store.size).toBe(4_500); + // The leg this test exists for. + expect(reclaimSpace).not.toHaveBeenCalled(); + // Deferral, not loss: the pages the reap freed are returned by the next + // sweep, which re-derives the same reclaim set from the same deletes. + expect(report.reclaimed).toEqual([]); + }); + + const TTL_ROTATED_OBJ: LifecycleObjectLike = { + name: 'sys_activity', + lifecycle: { + class: 'telemetry', + // The delta that makes the rotation leg reachable with the bit already + // read: `ttl` puts a paging reap — the thing that reads the bit and breaks + // — in front of the rotation, inside one `reapObject` call. + ttl: { field: 'expires_at', expireAfter: '7d' }, + storage: { strategy: 'rotation', shards: 14, unit: 'day' }, + } as any, + }; + + const rotation = () => ({ + object: 'sys_activity', + current: 'sys_activity__r20260710', + shards: ['sys_activity__r20260710'], + dropped: ['sys_activity__r20260626'], + }); + + it('a shard rotation nobody calls off drops its expired shards', async () => { + // The control for the test below: bit down, ttl reap and rotation both run. + const hot = reapStore(700); + const rotateShards = vi.fn(async () => rotation()); + const { engine, deletes } = captureEngine([TTL_ROTATED_OBJ], { + driver: { name: 'default', supportsRotation: true, rotateShards }, + findImpl: hot.findImpl, + deleteImpl: hot.deleteImpl, + }); + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(700); // the ttl reap drained the store + expect(rotateShards).toHaveBeenCalledWith(TTL_ROTATED_OBJ, FIXED_NOW); + expect(report.swept.map((e) => e.policy)).toEqual(['ttl', 'rotation']); + expect(report.swept.find((e) => e.policy === 'rotation')?.droppedShards).toBe(1); + }); + + it('stop() inside the ttl reap calls the shard rotation off', async () => { + // ttl + rotation on ONE object: the reap runs first and `batchedReap` breaks + // because it READ `aborted === true`, then returns straight into a rotation + // guarded only by strategy and driver capability. The DROP of expired + // physical shards is the most destructive operation this service issues, and + // it went to a datasource the host is closing. Probe evidence on #6412: + // `drops === ['DROP shard']`. + const hot = reapStore(5_000); + const rotateShards = vi.fn(async () => rotation()); + const { engine, deletes } = captureEngine([TTL_ROTATED_OBJ], { + driver: { name: 'default', supportsRotation: true, rotateShards }, + findImpl: hot.findImpl, + deleteImpl: hot.deleteImpl, + }); + const svc = service(engine); + const original = engine.delete.bind(engine); + engine.delete = async (object: string, options: unknown) => { + svc.stop(); // teardown lands in the ttl reap's first page + return original(object, options); + }; + + const report = await svc.sweep(); + + expect(svc.stopped).toBe(true); + expect(deletes).toHaveLength(500); // the reap stopped at its own boundary … + // … and no shard was dropped after it had read the bit. + expect(rotateShards).not.toHaveBeenCalled(); + expect(report.swept.map((e) => e.policy)).toEqual(['ttl']); + // Suppressing the rotation must not divert the object into the age-based + // fallback reap — that branch is `!lc.ttl`-gated, so it stays shut. + expect(hot.store.size).toBe(4_500); + }); + + const ROTATED_ONLY_OBJ: LifecycleObjectLike = { + name: 'sys_activity', + lifecycle: { + class: 'telemetry', + storage: { strategy: 'rotation', shards: 14, unit: 'day' }, + } as any, + }; + + it('rotation without `ttl` is unchanged — nothing has read the bit before the DROP', async () => { + // The form #6412's triage graded benign, pinned so the guard above cannot + // silently take it as well. Without `ttl` (and without `archive`, which + // returns earlier) nothing in `reapObject` awaits before the rotation, so + // the bit is structurally false there — not merely unobserved. Teardown + // landing INSIDE the rotation is as close as this form gets, and the + // rotation must still complete and report exactly as it always did. + let svc!: LifecycleService; + const rotateShards = vi.fn(async () => { + svc.stop(); // mid-DROP: after dispatch, so no decision was made with it + return rotation(); + }); + const { engine, deletes } = captureEngine([ROTATED_ONLY_OBJ], { + driver: { name: 'default', supportsRotation: true, rotateShards }, + }); + svc = service(engine); + + const report = await svc.sweep(); + + expect(rotateShards).toHaveBeenCalledTimes(1); + // No fallback age reap took the rotation's place … + expect(deletes).toEqual([]); + // … and the entry is identical to the one the unaborted form reports. + expect(report.swept).toEqual([ + { + object: 'sys_activity', + class: 'telemetry', + policy: 'rotation', + cutoff: isoCutoff('14d'), + droppedShards: 1, + }, + ]); + }); + 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 ad75ebf4ea..4fec2a7e1a 100644 --- a/packages/objectql/src/lifecycle/lifecycle-service.ts +++ b/packages/objectql/src/lifecycle/lifecycle-service.ts @@ -612,6 +612,25 @@ export class LifecycleService { } for (const driver of reclaimable) { + // [#4747] Leg boundary, after the object loop — space reclaim is the + // last thing `sweep()` issues at the data plane, and the one the + // per-object check above cannot reach. That check sits at the TOP of the + // object loop, so a teardown landing inside the LAST declared object's + // reap never meets it again: `batchedReap` breaks BECAUSE it read + // `aborted === true`, the object loop then ends normally rather than + // through the check, and control arrives here. Sending a VACUUM-class + // operation to a datasource the host is closing is therefore not a race + // teardown lost — it is work issued by code that had already been told + // the engine is going away. + // + // Deferring costs nothing. Reclaim is pure housekeeping, not half of a + // pair: it deletes no row, and skipping it leaves nothing inconsistent — + // only pages unreturned, which the next sweep reclaims after re-deriving + // the same `reclaimable` set from the same deletes. Checking per driver + // rather than once before the loop also covers teardown landing inside + // one driver's reclaim: the datasources still queued are spared instead + // of being asked in turn. + if (this.abort.aborted) break; try { await driver.reclaimSpace!(); report.reclaimed.push(driver.name ?? 'default'); @@ -847,7 +866,31 @@ export class LifecycleService { let rotated = false; if (lc.storage?.strategy === 'rotation') { const driver = engine.getDriverForObject(object) as RotationCapableDriver | undefined; - if (driver && typeof driver.rotateShards === 'function' && driver.supportsRotation !== false) { + // [#4747] Leg boundary. `rotateShards` DROPs expired physical shards — the + // most destructive single operation this service issues, and the one the + // guards below it (strategy, driver capability) say nothing about. When + // the object declares `ttl` as well, the reap above has already run, and + // `batchedReap` may have broken BECAUSE it read `aborted === true`; it + // returns straight to here. Dropping shards at a datasource the host is + // closing is then a decision made with the answer in hand, not an await + // that merely straddled teardown. + // + // Deferring costs nothing. Rotation is an O(1) reclaim of a bound the next + // sweep re-derives from the same `storage.shards` × `unit` window and + // applies to the same shards: skipping it drops nothing early and retains + // nothing past its window by more than one sweep interval. + // + // The conjunct cannot divert the rotation-WITHOUT-`ttl` form into the + // age-based fallback further down. That branch is `!lc.ttl`-gated, and + // without `ttl` (and without `archive`, which returns earlier) nothing in + // `reapObject` has awaited before this point — so the bit here is + // structurally false in that form, not merely unobserved. + if ( + !this.abort.aborted && + driver && + typeof driver.rotateShards === 'function' && + driver.supportsRotation !== false + ) { const windowMs = lc.storage.shards * SHARD_UNIT_MS[lc.storage.unit]; const res = await driver.rotateShards(obj, this.now()); report.swept.push({