From 8c523c413b99332d13aa085794ead8ab308ec7f2 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Sun, 6 Sep 2026 02:09:10 +0800 Subject: [PATCH] fix(storage): reclaim the bytes the v1 upgrade orphaned The v1 upgrade rebuilds the artifact catalog by dropping the table and re-inserting what still decodes. A record naming a retired source, one naming no source, the tombstone a user's deletion left behind, and any row too malformed or self-inconsistent to carry over all fail that gate -- and every one of them leaves its file on disk, unreachable from the catalog that used to name it. Dropping a row is the last moment anything knows where its bytes are, so the upgrade now records every path it does not carry over, in the same transaction. Unlinking there is not an option: a rollback after one would be unrecoverable. The store drains that list on recovery, and a path some record has since claimed keeps its bytes. Recording what the upgrade drops is exact where a file-name pattern could only be a guess, so there is nothing to scan and nothing to keep in step with the writers that produced these files. The cost is that a store converted by an earlier build keeps its residue: its rows are already gone. Generated-by: Claude Code --- .../src/server/execution-composition.ts | 9 ++ .../src/__tests__/artifact-stores.test.ts | 108 ++++++++++++++++++ packages/storage/src/artifact-store.ts | 42 ++++++- packages/storage/src/artifact-stores.ts | 2 + .../storage/src/sqlite-artifact-metadata.ts | 18 +++ .../storage/src/sqlite-artifact-schema.ts | 39 ++++++- 6 files changed, 208 insertions(+), 10 deletions(-) diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 27f8c4bc59..7c0f8f90e0 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -1755,6 +1755,15 @@ export async function createExecutionRuntimeHostComposition( recovery: { state: async () => { await skills.recover(); + try { + await openedArtifactStore.reclaimUpgradeResidue(); + } catch (error) { + // Leftover bytes are not worth refusing to start over; the next + // start tries again. + console.error( + `[runtime-host] upgrade residue could not be reclaimed: ${generalizedErrorMessage(error)}`, + ); + } }, }, drain: [ diff --git a/packages/storage/src/__tests__/artifact-stores.test.ts b/packages/storage/src/__tests__/artifact-stores.test.ts index 1965bbcc63..2bb2a687fa 100644 --- a/packages/storage/src/__tests__/artifact-stores.test.ts +++ b/packages/storage/src/__tests__/artifact-stores.test.ts @@ -135,6 +135,114 @@ describe('interactive artifact store authority', () => { }); }); + test('reclaims the bytes the v1 upgrade orphaned, including a user-deleted upload', async () => { + await withInteractiveOwner(async (owner, root, track) => { + const initial = await openInteractiveArtifactStoreForWrite(owner.lease); + initial.close(); + const db = new DatabaseSync(join(root, 'runtime.sqlite')); + db.exec(` + DROP TABLE artifact_records; + CREATE TABLE artifact_records ( + storage_key TEXT PRIMARY KEY, artifact_id TEXT NOT NULL, + session_id TEXT NOT NULL, created_at INTEGER NOT NULL CHECK(created_at >= 0), + status TEXT NOT NULL CHECK(status IN ('live', 'deleted')), + relative_path TEXT NOT NULL, record_json TEXT NOT NULL + ); + CREATE UNIQUE INDEX artifact_records_relative_path ON artifact_records(relative_path); + UPDATE operational_schema_migrations SET version = 1 WHERE scope = 'artifact'; + `); + // One row per reason the upgrade drops one, each with bytes on disk. + const rows = [ + { id: 'live', name: 'quarterly numbers.csv', status: 'live', source: 'user_upload' }, + { id: 'erased', name: 'passport scan.pdf', status: 'deleted', source: 'user_upload' }, + { + id: 'retired', + name: 'provider-request-step-4-cap.json', + status: 'live', + source: 'provider_request_capture', + }, + { id: 'sourceless', name: 'recap-request.json', status: 'live', source: undefined }, + { id: 'broken', name: 'unreadable.txt', status: 'live', source: 'tool_result' }, + { id: 'mismatched', name: 'inconsistent.txt', status: 'live', source: 'tool_result' }, + // Sorts first, and a directory cannot be unlinked, so it stands in for + // any leftover the store cannot remove. + { id: 'aborted', name: 'stuck', status: 'live', source: 'provider_request_capture' }, + ]; + await mkdir(join(root, 'artifacts', 'session-1'), { recursive: true }); + for (const row of rows) { + const relativePath = `session-1/${row.id}-${row.name}`; + if (row.id === 'aborted') await mkdir(join(root, 'artifacts', relativePath)); + else await writeFile(join(root, 'artifacts', relativePath), `bytes of ${row.id}`); + db.prepare('INSERT INTO artifact_records VALUES (?, ?, ?, ?, ?, ?, ?)').run( + row.id, + row.id, + 'session-1', + 1, + row.status, + relativePath, + row.id === 'broken' + ? '{' + : JSON.stringify({ + id: row.id === 'mismatched' ? 'other-id' : row.id, + sessionId: 'session-1', + turnId: 'turn-1', + createdAt: 1, + name: row.name, + kind: 'file', + sizeBytes: `bytes of ${row.id}`.length, + relativePath, + ...(row.source ? { source: row.source } : {}), + status: row.status, + }), + ); + } + db.close(); + + const store = track(await openInteractiveArtifactStoreForWrite(owner.lease)); + // Re-creating a dropped record's exact path before the reclamation runs + // must keep the new bytes, not honour the note. + await store.create({ + id: 'erased', + sessionId: 'session-1', + turnId: 'turn-2', + name: 'passport scan.pdf', + kind: 'file', + content: 'uploaded again', + source: 'user_upload', + }); + await store.reclaimUpgradeResidue(); + await store.reclaimUpgradeResidue(); + + const path = (id: string) => + join(root, 'artifacts', `session-1/${id}-${rows.find((row) => row.id === id)!.name}`); + for (const id of ['retired', 'sourceless', 'broken', 'mismatched']) { + await assert.rejects(() => stat(path(id)), { code: 'ENOENT' }); + } + assert.deepEqual(await store.readTextInSession('session-1', 'live'), { + ok: true, + text: 'bytes of live', + }); + assert.deepEqual(await store.readTextInSession('session-1', 'erased'), { + ok: true, + text: 'uploaded again', + }); + assert.equal((await stat(path('aborted'))).isDirectory(), true); + store.close(); + + // Everything behind the one that would not go was still reclaimed, and + // only its own note survives for a later attempt. + const remaining = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true }); + assert.deepEqual( + remaining + .prepare('SELECT relative_path FROM artifact_upgrade_orphan_paths') + .all() + .map((row) => (row as { relative_path: string }).relative_path), + ['session-1/aborted-stuck'], + ); + remaining.close(); + }); + }); + test('requires authentic leases and writer facades', async () => { await assert.rejects( () => diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index c2c5519f04..45f565193b 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -182,6 +182,7 @@ export interface ArtifactAuthorityStore extends DurableArtifactAttachmentReader input: ConversationArtifactCopyInput, ): Promise; purgeSessionArtifacts(sessionId: string): Promise; + reclaimUpgradeResidue(): Promise; deleteOwnedArtifactInSession( sessionId: string, artifactId: string, @@ -473,6 +474,43 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { }); } + /** + * Deletes the files the v1 upgrade recorded as no longer named by any record. + * + * A path some record has since claimed keeps its bytes. A note is discharged + * once its file is gone, and a file that will not go keeps only its own note + * rather than holding up the ones behind it. + */ + async reclaimUpgradeResidue(): Promise { + await this.enqueueMutation(async () => { + await this.prepareMutationUnlocked(); + const recorded = this.metadataRepository.readUpgradeOrphanPaths(); + if (recorded.length === 0) return; + const claimed = new Set(this.records.map((record) => record.relativePath)); + const directories = new Set(); + const discharged: string[] = []; + try { + for (const relativePath of recorded) { + if (claimed.has(relativePath) || !isSafeRelativeArtifactPath(relativePath)) { + discharged.push(relativePath); + continue; + } + const target = join(this.artifactRoot, relativePath); + try { + await unlink(target); + directories.add(dirname(target)); + } catch (error) { + if (!isNotFound(error)) continue; + } + discharged.push(relativePath); + } + } finally { + for (const directory of directories) await syncDirectory(directory); + } + if (discharged.length > 0) this.metadataRepository.forgetUpgradeOrphanPaths(discharged); + }); + } + private async replayExistingArtifactUnlocked( existing: ArtifactRecord, input: CreateArtifactInput, @@ -1088,10 +1126,6 @@ function assertArtifactTurnKey(value: unknown): asserts value is string { const ARTIFACT_KIND_SET = new Set(ARTIFACT_KINDS); const ARTIFACT_SOURCE_SET = new Set(ARTIFACT_SOURCES); -function isRecord(value: unknown): value is Record { - return typeof value === 'object' && value !== null && !Array.isArray(value); -} - async function assertArtifactDirectory(artifactRoot: string, directory: string): Promise { const root = await ensureRealDirectory(artifactRoot); const resolvedDirectory = await realpath(directory); diff --git a/packages/storage/src/artifact-stores.ts b/packages/storage/src/artifact-stores.ts index 58d82b3d73..b49470e9d4 100644 --- a/packages/storage/src/artifact-stores.ts +++ b/packages/storage/src/artifact-stores.ts @@ -72,6 +72,7 @@ export interface InteractiveArtifactStoreWriter extends DurableArtifactAttachmen input: ConversationArtifactCopyInput, ): Promise; purgeSessionArtifacts(sessionId: string): Promise; + reclaimUpgradeResidue(): Promise; listPage: ArtifactAuthorityStore['listPage']; listTurnArtifacts: ArtifactAuthorityStore['listTurnArtifacts']; getInSession: ArtifactAuthorityStore['getInSession']; @@ -177,6 +178,7 @@ function createWriterFacade( return run(() => store.copyConversationArtifacts(acceptedInput)); }, purgeSessionArtifacts: (sessionId) => run(() => store.purgeSessionArtifacts(sessionId)), + reclaimUpgradeResidue: () => run(() => store.reclaimUpgradeResidue()), deleteUserArtifactInSession: (sessionId, artifactId) => run(() => store.deleteUserArtifactInSession(sessionId, artifactId)), close: () => { diff --git a/packages/storage/src/sqlite-artifact-metadata.ts b/packages/storage/src/sqlite-artifact-metadata.ts index 08c75685e2..643ad80db1 100644 --- a/packages/storage/src/sqlite-artifact-metadata.ts +++ b/packages/storage/src/sqlite-artifact-metadata.ts @@ -92,6 +92,24 @@ class SqliteArtifactMetadataRepository { }); } + readUpgradeOrphanPaths(): string[] { + this.assertOpen(); + const rows = this.#lease.database + .prepare('SELECT relative_path FROM artifact_upgrade_orphan_paths ORDER BY relative_path') + .all() as Array<{ relative_path: string }>; + return rows.map((row) => row.relative_path); + } + + forgetUpgradeOrphanPaths(relativePaths: readonly string[]): void { + this.assertOpen(); + this.#lease.transaction('write', () => { + const forget = this.#lease.database.prepare( + 'DELETE FROM artifact_upgrade_orphan_paths WHERE relative_path = ?', + ); + for (const relativePath of relativePaths) forget.run(relativePath); + }); + } + close(): void { if (this.#closed) return; this.#closed = true; diff --git a/packages/storage/src/sqlite-artifact-schema.ts b/packages/storage/src/sqlite-artifact-schema.ts index f8d3f3e733..84a880ed8b 100644 --- a/packages/storage/src/sqlite-artifact-schema.ts +++ b/packages/storage/src/sqlite-artifact-schema.ts @@ -18,7 +18,10 @@ */ import type { DatabaseSync } from 'node:sqlite'; -import { decodeArtifactRecordJsons } from './artifact-metadata-codec.js'; +import { + decodeArtifactRecordJsons, + isSafeRelativeArtifactPath, +} from './artifact-metadata-codec.js'; export const SQLITE_ARTIFACT_SCHEMA_VERSION = 3; @@ -27,15 +30,21 @@ export function migrateSqliteArtifactDatabase(db: DatabaseSync): void { name?: unknown; }>; const retained: string[] = []; - if (columns.some(({ name }) => name === 'status' || name === 'storage_key')) { + // Every path the old table named. Whatever is not carried over is a file no + // catalog will name again, and this is the last moment anything knows it is + // there. Unlinking here is not an option: a rollback after one would be + // unrecoverable, so the paths are recorded for the store to reclaim later. + const scanned: string[] = []; + const hasStatusColumn = columns.some(({ name }) => name === 'status'); + if (hasStatusColumn || columns.some(({ name }) => name === 'storage_key')) { const rows = db.prepare('SELECT * FROM artifact_records').all(); for (const row of rows) { - if (columns.some(({ name }) => name === 'status') && row.status !== 'live') continue; + if (typeof row.relative_path === 'string' && isSafeRelativeArtifactPath(row.relative_path)) { + scanned.push(row.relative_path); + } try { const parsed = JSON.parse(String(row.record_json)); if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) continue; - if (parsed.status !== undefined && parsed.status !== 'live') continue; - delete parsed.status; if ( parsed.id !== row.artifact_id || parsed.sessionId !== row.session_id || @@ -43,6 +52,13 @@ export function migrateSqliteArtifactDatabase(db: DatabaseSync): void { parsed.relativePath !== row.relative_path ) continue; + if ( + [hasStatusColumn ? row.status : undefined, parsed.status].some( + (value) => value !== undefined && value !== null && value !== 'live', + ) + ) + continue; + delete parsed.status; retained.push(JSON.stringify(parsed)); } catch {} } @@ -62,11 +78,22 @@ export function migrateSqliteArtifactDatabase(db: DatabaseSync): void { CREATE UNIQUE INDEX IF NOT EXISTS artifact_records_relative_path ON artifact_records(relative_path); + + CREATE TABLE IF NOT EXISTS artifact_upgrade_orphan_paths ( + relative_path TEXT PRIMARY KEY + ); + `); + const carried = decodeArtifactRecordJsons(retained); + const kept = new Set(carried.map((record) => record.relativePath)); + const orphan = db.prepare(` + INSERT INTO artifact_upgrade_orphan_paths VALUES (?) + ON CONFLICT(relative_path) DO NOTHING `); + for (const relativePath of scanned) if (!kept.has(relativePath)) orphan.run(relativePath); const insert = db.prepare(` INSERT INTO artifact_records VALUES (?, ?, ?, ?, ?) `); - for (const record of decodeArtifactRecordJsons(retained)) { + for (const record of carried) { insert.run( record.id, record.sessionId,