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,