fix(objectql): lifecycle 的空间回收与分片轮转尊重 teardown abort 位 (#6412)

.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,而非「无人读过」。

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<string, Record<string, unknown>>();
+ 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<Record<string, unknown>> = [];
+ 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;

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({