perf(storage): persist artifact metadata deltas

packages/storage/src/__tests__/artifact-store.test.ts

@@ -1599,7 +1599,7 @@ async function writeArtifactMetadata(
 ): Promise<string> {
   const repository = createSqliteArtifactMetadataRepository(root);
   try {
-    repository.replaceAll(records);
+    repository.applyChanges({ upserts: records });
   } finally {
     repository.close();
   }

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',
+  };
+}

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<void>;
   readAll(): ArtifactRecord[];
-  replaceAll(records: readonly ArtifactRecord[]): void;
+  applyChanges(changes: ArtifactMetadataChanges): void;
   close(): void;
 }

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<void> {
     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<void> {
+  private async writeMetadataUnlocked(changes: ArtifactMetadataChanges): Promise<void> {
     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 };

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,