From c8dee7da3a6e9b4ee0a6254291c600f9bc2337e7 Mon Sep 17 00:00:00 2001 From: likun Date: Fri, 4 Sep 2026 11:16:02 +0800 Subject: [PATCH] perf(storage): persist artifact metadata deltas Generated-by: OpenAI Codex --- .../src/__tests__/artifact-store.test.ts | 2 +- .../sqlite-artifact-metadata.test.ts | 105 ++++++++++++++++++ .../src/artifact-metadata-repository.ts | 7 +- packages/storage/src/artifact-store.ts | 31 +++--- .../storage/src/sqlite-artifact-metadata.ts | 32 +++++- 5 files changed, 156 insertions(+), 21 deletions(-) create mode 100644 packages/storage/src/__tests__/sqlite-artifact-metadata.test.ts diff --git a/packages/storage/src/__tests__/artifact-store.test.ts b/packages/storage/src/__tests__/artifact-store.test.ts index 339c28d336..31c8c759ae 100644 --- a/packages/storage/src/__tests__/artifact-store.test.ts +++ b/packages/storage/src/__tests__/artifact-store.test.ts @@ -1599,7 +1599,7 @@ async function writeArtifactMetadata( ): Promise { const repository = createSqliteArtifactMetadataRepository(root); try { - repository.replaceAll(records); + repository.applyChanges({ upserts: records }); } finally { repository.close(); } diff --git a/packages/storage/src/__tests__/sqlite-artifact-metadata.test.ts b/packages/storage/src/__tests__/sqlite-artifact-metadata.test.ts new file mode 100644 index 0000000000..a9214898ab --- /dev/null +++ b/packages/storage/src/__tests__/sqlite-artifact-metadata.test.ts @@ -0,0 +1,105 @@ +/* + * 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 assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; +import { test } from 'node:test'; +import type { ArtifactRecord } from '@maka/core/artifacts'; +import { OPERATIONAL_STATE_DATABASE_NAME } from '../operational-state-store.js'; +import { createSqliteArtifactMetadataRepository } from '../sqlite-artifact-metadata.js'; + +test('Artifact metadata changes only write changed rows', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-artifact-metadata-delta-')); + const repository = createSqliteArtifactMetadataRepository(root); + let inspector: DatabaseSync | undefined; + try { + const unchanged = artifactRecord('unchanged'); + const updated = artifactRecord('updated'); + const removed = artifactRecord('removed'); + repository.applyChanges({ upserts: [unchanged, updated, removed] }); + + inspector = new DatabaseSync(join(root, OPERATIONAL_STATE_DATABASE_NAME)); + inspector.exec(` + CREATE TABLE artifact_write_audit(kind TEXT NOT NULL); + CREATE TRIGGER artifact_write_audit_insert AFTER INSERT ON artifact_records + BEGIN INSERT INTO artifact_write_audit VALUES ('insert'); END; + CREATE TRIGGER artifact_write_audit_update AFTER UPDATE ON artifact_records + BEGIN INSERT INTO artifact_write_audit VALUES ('update'); END; + CREATE TRIGGER artifact_write_audit_delete AFTER DELETE ON artifact_records + BEGIN INSERT INTO artifact_write_audit VALUES ('delete'); END; + `); + + repository.applyChanges({ + upserts: [unchanged, { ...updated, status: 'deleted' }, artifactRecord('added')], + deleteIds: [removed.id], + }); + + const writes = inspector + .prepare('SELECT kind, count(*) AS count FROM artifact_write_audit GROUP BY kind') + .all() as Array<{ kind: string; count: number }>; + assert.deepEqual( + writes.map(({ kind, count }) => ({ kind, count })), + [ + { kind: 'delete', count: 1 }, + { kind: 'insert', count: 1 }, + { kind: 'update', count: 1 }, + ], + ); + assert.deepEqual( + repository + .readAll() + .map(({ id, status }) => ({ id, status })) + .sort((left, right) => left.id.localeCompare(right.id)), + [ + { id: 'added', status: 'live' }, + { id: 'unchanged', status: 'live' }, + { id: 'updated', status: 'deleted' }, + ], + ); + + inspector.exec(` + DROP TRIGGER artifact_write_audit_insert; + DROP TRIGGER artifact_write_audit_update; + DROP TRIGGER artifact_write_audit_delete; + DROP TABLE artifact_write_audit; + `); + } finally { + inspector?.close(); + repository.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +function artifactRecord(id: string): ArtifactRecord { + return { + id, + sessionId: 'session-1', + turnId: 'turn-1', + createdAt: 1, + name: `${id}.txt`, + kind: 'file', + sizeBytes: id.length, + relativePath: `session-1/${id}-${id}.txt`, + source: 'fixture', + status: 'live', + }; +} diff --git a/packages/storage/src/artifact-metadata-repository.ts b/packages/storage/src/artifact-metadata-repository.ts index ac99e603cc..0ee907d385 100644 --- a/packages/storage/src/artifact-metadata-repository.ts +++ b/packages/storage/src/artifact-metadata-repository.ts @@ -19,9 +19,14 @@ import type { ArtifactRecord } from '@maka/core/artifacts'; +export interface ArtifactMetadataChanges { + readonly upserts?: readonly ArtifactRecord[]; + readonly deleteIds?: readonly string[]; +} + export interface ArtifactMetadataRepository { ready(): Promise; readAll(): ArtifactRecord[]; - replaceAll(records: readonly ArtifactRecord[]): void; + applyChanges(changes: ArtifactMetadataChanges): void; close(): void; } diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index 0ba7d9794c..2c50842ccb 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -70,7 +70,10 @@ import { } from './artifact-writer-lock.js'; import type { ArtifactWriterLockAuthority } from './root-authority.js'; import { syncDirectory, syncDirectoryChain, syncFile } from './stable-storage.js'; -import type { ArtifactMetadataRepository } from './artifact-metadata-repository.js'; +import type { + ArtifactMetadataChanges, + ArtifactMetadataRepository, +} from './artifact-metadata-repository.js'; import { createSqliteArtifactMetadataRepository } from './sqlite-artifact-metadata.js'; export { isSafeRelativeArtifactPath } from './artifact-metadata-codec.js'; @@ -557,7 +560,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { throw error; } await syncDirectory(targetDirectory); - await this.writeMetadataUnlocked(nextRecords); + await this.writeMetadataUnlocked({ upserts: [record] }); } catch (error) { if (targetLinked) { try { @@ -639,7 +642,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { const nextRecords = this.records.map((record) => record.id === canonical.id ? revived : record, ); - await this.writeMetadataUnlocked(nextRecords); + await this.writeMetadataUnlocked({ upserts: [revived] }); this.replaceRecords(nextRecords); return { ...revived }; } @@ -843,13 +846,15 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { async delete(artifactId: string): Promise { await this.enqueueMutation(async () => { await this.prepareMutationUnlocked({ kind: 'delete' }); + const existing = this.records.find( + (record) => record.id === artifactId && record.status !== 'deleted', + ); + if (!existing) return; + const tombstone: ArtifactRecord = { ...existing, status: 'deleted' }; const nextRecords: ArtifactRecord[] = this.records.map((record) => - record.id === artifactId && record.status !== 'deleted' - ? { ...record, status: 'deleted' } - : record, + record.id === artifactId ? tombstone : record, ); - if (nextRecords.every((record, index) => record === this.records[index])) return; - await this.writeMetadataUnlocked(nextRecords); + await this.writeMetadataUnlocked({ upserts: [tombstone] }); this.replaceRecords(nextRecords); }); } @@ -871,7 +876,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { const nextRecords = this.records.map((record) => record.id === existing.id ? tombstone : record, ); - await this.writeMetadataUnlocked(nextRecords); + await this.writeMetadataUnlocked({ upserts: [tombstone] }); this.replaceRecords(nextRecords); return { kind: 'deleted', record: { ...tombstone } }; }); @@ -992,7 +997,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { for (const directory of changedDirectories) await syncDirectory(directory); } const nextRecords = this.records.filter((record) => !ids.has(record.id)); - await this.writeMetadataUnlocked(nextRecords); + await this.writeMetadataUnlocked({ deleteIds: [...ids] }); this.replaceRecords(nextRecords); await this.removePurgeIntentUnlocked(); } @@ -1039,10 +1044,10 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { this.replaceRecords(this.metadataRepository.readAll()); } - private async writeMetadataUnlocked(records: readonly ArtifactRecord[]): Promise { + private async writeMetadataUnlocked(changes: ArtifactMetadataChanges): Promise { await this.metadataRepository.ready(); this.metadataReady = true; - this.metadataRepository.replaceAll(records); + this.metadataRepository.applyChanges(changes); } private async prepareMutationUnlocked( @@ -1197,7 +1202,7 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { status: 'live', }; const nextRecords = [...this.records, record]; - await this.writeMetadataUnlocked(nextRecords); + await this.writeMetadataUnlocked({ upserts: [record] }); this.replaceRecords(nextRecords); this.recoverableOrphans.delete(filesystemPathKey(candidate.relativePath)); return { ...record }; diff --git a/packages/storage/src/sqlite-artifact-metadata.ts b/packages/storage/src/sqlite-artifact-metadata.ts index 50b89e4eb4..c1678b6011 100644 --- a/packages/storage/src/sqlite-artifact-metadata.ts +++ b/packages/storage/src/sqlite-artifact-metadata.ts @@ -20,7 +20,10 @@ import { createHash } from 'node:crypto'; import { resolve } from 'node:path'; import type { ArtifactRecord } from '@maka/core/artifacts'; -import type { ArtifactMetadataRepository } from './artifact-metadata-repository.js'; +import type { + ArtifactMetadataChanges, + ArtifactMetadataRepository, +} from './artifact-metadata-repository.js'; import { decodeArtifactRecordJsons } from './artifact-metadata-codec.js'; import { acquireOperationalStateDatabase, @@ -58,11 +61,15 @@ class SqliteArtifactMetadataRepository implements ArtifactMetadataRepository { return decodeRows(rows); } - replaceAll(records: readonly ArtifactRecord[]): void { + applyChanges(changes: ArtifactMetadataChanges): void { this.assertOpen(); this.#lease.transaction('write', () => { - this.#lease.database.prepare('DELETE FROM artifact_records').run(); - const insert = this.#lease.database.prepare(` + const remove = this.#lease.database.prepare( + 'DELETE FROM artifact_records WHERE storage_key = ?', + ); + for (const id of changes.deleteIds ?? []) remove.run(artifactIdentityKey(id)); + + const upsert = this.#lease.database.prepare(` INSERT INTO artifact_records( storage_key, artifact_id, @@ -72,9 +79,22 @@ class SqliteArtifactMetadataRepository implements ArtifactMetadataRepository { relative_path, record_json ) VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(storage_key) DO UPDATE SET + artifact_id = excluded.artifact_id, + session_id = excluded.session_id, + created_at = excluded.created_at, + status = excluded.status, + relative_path = excluded.relative_path, + record_json = excluded.record_json + WHERE artifact_id IS NOT excluded.artifact_id + OR session_id IS NOT excluded.session_id + OR created_at IS NOT excluded.created_at + OR status IS NOT excluded.status + OR relative_path IS NOT excluded.relative_path + OR record_json IS NOT excluded.record_json `); - for (const record of records) { - insert.run( + for (const record of changes.upserts ?? []) { + upsert.run( artifactIdentityKey(record.id), record.id, record.sessionId,