diff --git a/packages/storage/src/__tests__/sqlite-context-offload-store.test.ts b/packages/storage/src/__tests__/sqlite-context-offload-store.test.ts index 0ced27f540e..4f50e6e36f8 100644 --- a/packages/storage/src/__tests__/sqlite-context-offload-store.test.ts +++ b/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, diff --git a/packages/storage/src/sqlite-context-offload-store.ts b/packages/storage/src/sqlite-context-offload-store.ts index 1eb96c88be4..8b790fb2888 100644 --- a/packages/storage/src/sqlite-context-offload-store.ts +++ b/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; } diff --git a/packages/storage/src/sqlite-reclamation-checkpoint.ts b/packages/storage/src/sqlite-reclamation-checkpoint.ts new file mode 100644 index 00000000000..f1cb3bae869 --- /dev/null +++ b/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; + } +}