fix(storage): retry pending context-offload checkpoints

packages/storage/src/__tests__/sqlite-context-offload-store.test.ts

@@ -1145,6 +1145,54 @@ test('reclaimFreePages returns garbage-collected pages to the file in bounded ba
   });
 });
 
+test('reclaimFreePages retries an unfinished checkpoint on the next idle tick', async (t) => {
+  const fixture = await createFixture(t);
+  const bytes = new Uint8Array(1_000_000).fill(1);
+  const stored = await fixture.store.put(putInput('session-1', 'tool-1', bytes));
+  assert.equal(stored.ok, true);
+  if (!stored.ok) return;
+  await fixture.store.releaseReference({ sessionId: 'session-1', refId: stored.record.refId });
+  await fixture.store.collectGarbage({ olderThan: 1_001, maxBlobs: 64, maxBytes: bytes.length });
+
+  const reader = new DatabaseSync(fixture.path);
+  t.after(() => reader.close());
+  reader.exec('PRAGMA wal_checkpoint(TRUNCATE)');
+  const beforeBytes = (await stat(fixture.path)).size;
+  reader.exec('BEGIN');
+  const freePages = pragmaNumber(reader, 'freelist_count');
+  assert.ok(freePages > 64);
+
+  let reclaimedPages = 0;
+  for (;;) {
+    const batch = await fixture.store.reclaimFreePages({ maxPages: 64 });
+    reclaimedPages += batch.reclaimedPages;
+    if (!batch.hasMore) break;
+  }
+  assert.equal(reclaimedPages, freePages);
+  assert.equal((await stat(fixture.path)).size, beforeBytes, 'reader pins the unshrunk main file');
+  // An incomplete checkpoint must not keep the maintenance lane spinning.
+  assert.deepEqual(await fixture.store.reclaimFreePages({ maxPages: 64 }), {
+    reclaimedPages: 0,
+    hasMore: false,
+  });
+  reader.exec('COMMIT');
+
+  // Even a pending checkpoint cannot wait for or take a writer's lock.
+  reader.exec('BEGIN IMMEDIATE');
+  assert.deepEqual(await fixture.store.reclaimFreePages({ maxPages: 64 }), {
+    reclaimedPages: 0,
+    hasMore: false,
+  });
+  reader.exec('COMMIT');
+  assert.deepEqual(await fixture.store.reclaimFreePages({ maxPages: 64 }), {
+    reclaimedPages: 0,
+    hasMore: false,
+  });
+  assert.ok((await stat(fixture.path)).size < beforeBytes, 'idle retry shrinks the main file');
+  assert.equal(pragmaNumber(reader, 'freelist_count'), 0);
+  assert.equal(pragmaText(reader, 'quick_check'), 'ok');
+});
+
 function putInput(sessionId: string, ownerId: string, bytes: Uint8Array) {
   return {
     sessionId,

packages/storage/src/sqlite-context-offload-store.ts

@@ -47,6 +47,7 @@ import {
   syncFile,
 } from './stable-storage.js';
 import { runWithContextValueMutation } from './context-value-mutation-gate.js';
+import { SqliteReclamationCheckpoint } from './sqlite-reclamation-checkpoint.js';
 
 const MAX_MEDIA_TYPE_CODE_POINTS = 256;
 const SHA256_PATTERN = /^[0-9a-f]{64}$/;
@@ -131,6 +132,7 @@ export class SqliteContextOffloadStore implements ContextOffloadStore {
   readonly #storageRoot: string | undefined;
   readonly #valueRoot: string | undefined;
   readonly #mutationGateKey: string;
+  readonly #reclamationCheckpoint = new SqliteReclamationCheckpoint();
   #closed = false;
 
   constructor(path: string, options: SqliteContextOffloadStoreOptions) {
@@ -518,6 +520,7 @@ export class SqliteContextOffloadStore implements ContextOffloadStore {
     this.#assertOpen();
     // Probe under a read transaction so an idle tick never takes the write lock.
     if (this.#readTransaction(() => this.#freelistPages()) === 0) {
+      this.#reclamationCheckpoint.run(this.#database, 0);
       return { reclaimedPages: 0, hasMore: false };
     }
     const result = this.#writeTransaction(() => {
@@ -530,7 +533,7 @@ export class SqliteContextOffloadStore implements ContextOffloadStore {
       return { reclaimedPages: before - after, hasMore: after > 0 && after < before };
     });
     // PASSIVE never waits on readers; it moves the shrink into the main file.
-    if (result.reclaimedPages > 0) this.#database.exec('PRAGMA wal_checkpoint(PASSIVE)');
+    this.#reclamationCheckpoint.run(this.#database, result.reclaimedPages);
     return result;
   }
 

packages/storage/src/sqlite-reclamation-checkpoint.ts

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { DatabaseSync } from 'node:sqlite';
+
+/** Retains an unfinished reclamation checkpoint across idle maintenance ticks. */
+export class SqliteReclamationCheckpoint {
+  #pending = false;
+
+  run(database: DatabaseSync, reclaimedPages: number): void {
+    if (reclaimedPages > 0) this.#pending = true;
+    if (!this.#pending) return;
+    // Call outside transactions. PASSIVE never waits for a reader or writer;
+    // completion affects the next idle tick, not the maintenance lane's hasMore.
+    const row = database.prepare('PRAGMA wal_checkpoint(PASSIVE)').get();
+    if (
+      typeof row?.busy !== 'number' ||
+      typeof row.log !== 'number' ||
+      typeof row.checkpointed !== 'number'
+    ) {
+      throw new Error('Invalid SQLite wal_checkpoint result');
+    }
+    this.#pending = row.busy !== 0 || row.checkpointed < row.log;
+  }
+}