From 3734f57e9bb8fc8d9eb7f6c942f72f7952a7aee3 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 20:44:30 +0000 Subject: [PATCH] fix(state): preserve legacy migration semantics Adopt legacy SQLite databases with their WAL sidecars, retain canonical reset inputs while migrating committed results, make memory migrations atomic, and surface successful-scope close failures. --- .changeset/state-migration-compatibility.md | 9 + .../rsc-runtime/src/state/memory-driver.ts | 10 +- packages/rsc-runtime/src/state/sqlite.ts | 129 ++++++++++--- .../rsc-runtime/tests/state-kernel.test.ts | 37 ++++ .../rsc-runtime/tests/state-sqlite.test.ts | 169 +++++++++++++++++- 5 files changed, 320 insertions(+), 34 deletions(-) create mode 100644 .changeset/state-migration-compatibility.md diff --git a/.changeset/state-migration-compatibility.md b/.changeset/state-migration-compatibility.md new file mode 100644 index 000000000..0ab2c0434 --- /dev/null +++ b/.changeset/state-migration-compatibility.md @@ -0,0 +1,9 @@ +--- +"@agent-bundle/runtime": patch +--- + +Preserve durable state across the sqlite filename transition, recover legacy +journal results before schema migrations rebase history, keep reset +idempotency inputs unchanged while migrating their committed results, make +in-memory migrations atomic, and surface sqlite close failures on otherwise +successful shutdown. diff --git a/packages/rsc-runtime/src/state/memory-driver.ts b/packages/rsc-runtime/src/state/memory-driver.ts index 9b626d0eb..a40469fe0 100644 --- a/packages/rsc-runtime/src/state/memory-driver.ts +++ b/packages/rsc-runtime/src/state/memory-driver.ts @@ -83,7 +83,7 @@ interface MemoryStoreInternals { definition: AgentStateDefinition; head: AgentStateSnapshot; readonly journal: AgentStateJournalRecord[]; - readonly keys: Map>; + keys: Map>; } interface MemoryStoreEntry { @@ -319,17 +319,19 @@ const migrateOpenStore = ( state: migrated, toVersion: definition.version, }; - internals.journal.push(record); + const keys = new Map>(); // Committed results replay across migrations: every stored result sits at // `fromVersion` (this loop maintains that inductively), so each one rides // the same migration chain as the head. for (const [key, entry] of internals.keys) { - internals.keys.set(key, { + keys.set(key, { record: entry.record, state: runStateMigrations(definition, fromVersion, entry.state), }); } - internals.keys.set(record.idempotencyKey, { record, state: migrated }); + keys.set(record.idempotencyKey, { record, state: migrated }); + internals.keys = keys; + internals.journal.push(record); internals.head = Object.freeze({ revision: record.revision, state: migrated }); internals.definition = definition; }; diff --git a/packages/rsc-runtime/src/state/sqlite.ts b/packages/rsc-runtime/src/state/sqlite.ts index 1b465b487..971e400da 100644 --- a/packages/rsc-runtime/src/state/sqlite.ts +++ b/packages/rsc-runtime/src/state/sqlite.ts @@ -1,5 +1,9 @@ import { createHash } from 'node:crypto'; -import { mkdirSync } from 'node:fs'; +import { + existsSync, + mkdirSync, + renameSync, +} from 'node:fs'; import { dirname, join, resolve } from 'node:path'; // node:sqlite emits an ExperimentalWarning on load (documented in the README): // the module is Node's built-in SQLite binding, stable enough for Node >= 22.13 @@ -27,6 +31,7 @@ import type { AgentStateDefinition, AgentStateDispatchOptions, AgentStateDriver, + AgentStateEvent, AgentStateEventSchemas, AgentStateJournalRecord, AgentStateReadOptions, @@ -197,6 +202,7 @@ interface JournalRow { readonly kind: string; readonly name: string | null; readonly payload: string | null; + readonly result_state: string | null; readonly revision: number; readonly state: string | null; readonly to_version: number | null; @@ -230,6 +236,9 @@ const recordFromRow = (definitionId: string, row: JournalRow): AgentStateJournal const sanitizedFileName = (definitionId: string): string => `${definitionId.replace(/[^a-zA-Z0-9._-]+/gu, '-')}-${createHash('sha256').update(definitionId, 'utf8').digest('hex').slice(0, 16)}.sqlite`; +const legacySanitizedFileName = (definitionId: string): string => + `${definitionId.replace(/[^a-zA-Z0-9._-]+/gu, '-')}-${Buffer.from(definitionId, 'utf8').toString('hex').slice(0, 12)}.sqlite`; + class SqliteConnection extends Context.Service()( '@agent-bundle/runtime/state/SqliteConnection', ) {} @@ -341,11 +350,13 @@ class SqliteStore implements Age #committedByKey( db: DatabaseSync, key: string, - ): { readonly record: AgentStateJournalRecord; readonly stateText: string | null } | undefined { + ): { readonly record: AgentStateJournalRecord; readonly resultStateText: string | null } | undefined { const row = db.prepare('SELECT * FROM agent_state_journal WHERE idempotency_key = ?').get(key) as | JournalRow | undefined; - return row === undefined ? undefined : { record: recordFromRow(this.#definition.id, row), stateText: row.state }; + return row === undefined + ? undefined + : { record: recordFromRow(this.#definition.id, row), resultStateText: row.result_state ?? row.state }; } /** @@ -356,11 +367,11 @@ class SqliteStore implements Age */ #committedState( db: DatabaseSync, - committed: { readonly record: AgentStateJournalRecord; readonly stateText: string | null }, + committed: { readonly record: AgentStateJournalRecord; readonly resultStateText: string | null }, ): TState { const raw = - committed.stateText !== null - ? parseStoredJson(this.#definition.id, 'state', committed.record.revision, committed.stateText) + committed.resultStateText !== null + ? parseStoredJson(this.#definition.id, 'result state', committed.record.revision, committed.resultStateText) : this.#replayTo(db, committed.record.revision); const parsed = this.#definition.schema.safeParse(raw); if (!parsed.success) { @@ -387,7 +398,7 @@ class SqliteStore implements Age const stateText = canonicalJson(state); db .prepare( - 'INSERT INTO agent_state_journal (revision, kind, name, payload, state, to_version, idempotency_key, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)', + 'INSERT INTO agent_state_journal (revision, kind, name, payload, state, result_state, to_version, idempotency_key, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', ) .run( record.revision, @@ -397,6 +408,7 @@ class SqliteStore implements Age // Event rows store their post-commit state too, so idempotent replay // survives migrations without exact-revision replay. stateText, + stateText, record.kind === 'migrate' ? record.toVersion : null, record.idempotencyKey, record.committedAt, @@ -417,7 +429,7 @@ class SqliteStore implements Age const committedByKey = (db: DatabaseSync, key: string) => this.#committedByKey(db, key); const committedState = ( db: DatabaseSync, - committed: { readonly record: AgentStateJournalRecord; readonly stateText: string | null }, + committed: { readonly record: AgentStateJournalRecord; readonly resultStateText: string | null }, ) => this.#committedState(db, committed); const headState = (db: DatabaseSync) => this.#headState(db, 'commit'); const now = this.#now; @@ -601,6 +613,7 @@ class SqliteStore implements Age name TEXT, payload TEXT, state TEXT, + result_state TEXT, to_version INTEGER, idempotency_key TEXT NOT NULL UNIQUE, committed_at TEXT NOT NULL @@ -611,6 +624,12 @@ class SqliteStore implements Age state TEXT NOT NULL ); `); + const journalColumns = transactionDb.prepare('PRAGMA table_info(agent_state_journal)').all() as unknown as { + readonly name: string; + }[]; + if (!journalColumns.some((column) => column.name === 'result_state')) { + transactionDb.exec('ALTER TABLE agent_state_journal ADD COLUMN result_state TEXT'); + } const definition = this.#definition; const meta = transactionDb .prepare('SELECT definition_id, schema_version, kernel_format FROM agent_state_meta WHERE id = 1') @@ -678,14 +697,43 @@ class SqliteStore implements Age } } const migrated = runStateMigrations(definition, meta.schema_version, rawHead); - // Stored post-commit states ride the same chain so committed - // idempotency keys keep replaying after the migration; every stored - // state sits at `meta.schema_version` (maintained inductively here). - const updateState = transactionDb.prepare('UPDATE agent_state_journal SET state = ? WHERE revision = ?'); - for (const row of rows) { - if (row.state === null) continue; - const rawState = parseStoredJson(definition.id, 'state', row.revision, row.state); - updateState.run(canonicalJson(runStateMigrations(definition, meta.schema_version, rawState)), row.revision); + // Journal records retain the original commit input for dedupe. Their + // committed results migrate separately, matching the memory driver's + // `{ record, state }` split. Legacy event rows without a result are + // replayed before the new migration baseline makes old revisions + // unavailable. + const updateResult = transactionDb.prepare( + 'UPDATE agent_state_journal SET result_state = ? WHERE revision = ?', + ); + let replayState: unknown = definition.initial; + for (const [index, row] of rows.entries()) { + const record = records[index] as AgentStateJournalRecord; + const storedResultText = row.result_state ?? row.state; + let migratedResult: TState; + if (storedResultText !== null) { + replayState = parseStoredJson(definition.id, 'result state', row.revision, storedResultText); + migratedResult = runStateMigrations(definition, meta.schema_version, replayState); + } else if (record.kind === 'event') { + try { + replayState = definition.reduce( + replayState as TState, + { name: record.name, payload: record.payload } as AgentStateEvent, + ); + } catch (error) { + throw new AgentStateError( + 'migration-failure', + `State '${definition.id}' could not recover legacy result at revision ${String(record.revision)}`, + { cause: error }, + ); + } + migratedResult = runStateMigrations(definition, meta.schema_version, replayState); + } else { + throw new AgentStateError( + 'corrupt', + `State '${definition.id}' journal row at revision ${String(record.revision)} has no committed result`, + ); + } + updateResult.run(canonicalJson(migratedResult), row.revision); } const record: AgentStateJournalRecord = { committedAt: this.#now().toISOString(), @@ -763,25 +811,48 @@ export const createSqliteStateDriver = (options: SqliteStateDriverOptions): Agen `State '${definition.id}' declares lifetime '${definition.lifetime}' but this driver provides 'workspace-durable'`, ); } - return resolve( - options.file !== undefined ? options.file : join(options.root as string, sanitizedFileName(definition.id)), - ); - }), + if (options.file !== undefined) return resolve(options.file); + const root = options.root as string; + const currentFile = resolve(join(root, sanitizedFileName(definition.id))); + const legacyFile = resolve(join(root, legacySanitizedFileName(definition.id))); + mkdirSync(dirname(currentFile), { recursive: true }); + if (!existsSync(currentFile) && existsSync(legacyFile)) { + for (const suffix of ['-wal', '-shm']) { + const legacySidecar = `${legacyFile}${suffix}`; + if (!existsSync(legacySidecar)) continue; + try { + renameSync(legacySidecar, `${currentFile}${suffix}`); + } catch (error) { + // A concurrent adopter may have moved this sidecar after + // the existence check. Other failures must remain visible. + if ((error as SqliteErrorShape).code !== 'ENOENT') throw error; + } + } + try { + renameSync(legacyFile, currentFile); + } catch (error) { + // Another opener may have atomically adopted the same + // legacy file after both observed it. The winner's current + // path is authoritative; otherwise preserve the failure. + if (!existsSync(currentFile)) throw error; + } + } + return currentFile; + }, true), ); const connection = Effect.acquireRelease( sqliteEffect(definition.id, 'open database', () => { mkdirSync(dirname(file), { recursive: true }); return new DatabaseSync(file); }, true), - (db) => - Effect.sync(() => { - try { - db.close(); - } catch { - // Closing an already-broken connection must not mask the - // caller's path (the original failure carries the cause). - } - }), + (db, exit) => { + const close = sqliteEffect(definition.id, 'close database', () => { + db.close(); + }, true); + return Exit.isFailure(exit) + ? close.pipe(Effect.catch(() => Effect.void)) + : close; + }, ); const runtime = makeScopedEffectRuntime( Layer.effect(SqliteConnection, connection), diff --git a/packages/rsc-runtime/tests/state-kernel.test.ts b/packages/rsc-runtime/tests/state-kernel.test.ts index 691f16b74..f760aaa59 100644 --- a/packages/rsc-runtime/tests/state-kernel.test.ts +++ b/packages/rsc-runtime/tests/state-kernel.test.ts @@ -361,6 +361,43 @@ describe('explicit migrations', () => { }); }); + it('leaves the process store unchanged when a historical result migration throws', async () => { + const driver = createMemoryStateDriver(); + const storeV1 = await driver.open(v1()); + await storeV1.dispatch('incremented', { by: 1 }, { idempotencyKey: 'i1' }); + await storeV1.dispatch('incremented', { by: 2 }, { idempotencyKey: 'i2' }); + await storeV1.dispatch('incremented', { by: 3 }, { idempotencyKey: 'i3' }); + + await expect( + driver.open( + v2((persisted) => { + const state = persisted as CounterState; + if (state.count === 3) throw new Error('cannot migrate historical result'); + return { ...state, unit: 'edits' }; + }), + ), + ).rejects.toMatchObject({ code: 'migration-failure' }); + + expect(await storeV1.read()).toEqual({ revision: 3, state: { count: 6 } }); + expect((await storeV1.changes({ afterRevision: 0 })).changes.map((change) => change.kind)).toEqual([ + 'event', + 'event', + 'event', + ]); + await expect( + storeV1.dispatch('incremented', { by: 1 }, { idempotencyKey: 'i1' }), + ).resolves.toEqual({ replayed: true, revision: 1, state: { count: 1 } }); + + const storeV2 = await driver.open(v2()); + expect(await storeV2.read()).toEqual({ revision: 4, state: { count: 6, unit: 'edits' } }); + expect((await storeV2.changes({ afterRevision: 0 })).changes.map((change) => change.kind)).toEqual([ + 'event', + 'event', + 'event', + 'migrate', + ]); + }); + it('rejects opening a persisted-newer store with an older definition', async () => { const driver = createMemoryStateDriver(); await driver.open(v2()); diff --git a/packages/rsc-runtime/tests/state-sqlite.test.ts b/packages/rsc-runtime/tests/state-sqlite.test.ts index 09a0f4a2c..306fe9f7c 100644 --- a/packages/rsc-runtime/tests/state-sqlite.test.ts +++ b/packages/rsc-runtime/tests/state-sqlite.test.ts @@ -1,4 +1,5 @@ -import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { createHash } from 'node:crypto'; +import { access, mkdtemp, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { DatabaseSync } from 'node:sqlite'; @@ -76,6 +77,89 @@ const counterDefinition = ( schema: z.object({ count: z.number().int() }).strict(), }); +const migratingCounterDefinition = ( + id = 'state-sqlite-test/migrating-counter', +): AgentStateDefinition => + defineState({ + events: counterEvents, + id, + initial: { count: 0 }, + lifetime: 'workspace-durable', + migrations: { + 2: (persisted) => ({ count: (persisted as CounterState).count * 10 }), + }, + reduce: (state, event) => ({ count: state.count + event.payload.by }), + schema: z.object({ count: z.number().int() }).strict(), + version: 2, + }); + +const legacyFileName = (definitionId: string): string => + `${definitionId.replace(/[^a-zA-Z0-9._-]+/gu, '-')}-${Buffer.from(definitionId, 'utf8').toString('hex').slice(0, 12)}.sqlite`; + +const currentFileName = (definitionId: string): string => + `${definitionId.replace(/[^a-zA-Z0-9._-]+/gu, '-')}-${createHash('sha256').update(definitionId, 'utf8').digest('hex').slice(0, 16)}.sqlite`; + +const createLegacyMigrationDatabase = (file: string, definitionId: string): void => { + const db = new DatabaseSync(file); + try { + db.exec(` + CREATE TABLE agent_state_meta ( + id INTEGER PRIMARY KEY CHECK (id = 1), + definition_id TEXT NOT NULL, + schema_version INTEGER NOT NULL, + kernel_format INTEGER NOT NULL + ); + CREATE TABLE agent_state_journal ( + revision INTEGER PRIMARY KEY, + kind TEXT NOT NULL CHECK (kind IN ('event', 'reset', 'migrate')), + name TEXT, + payload TEXT, + state TEXT, + to_version INTEGER, + idempotency_key TEXT NOT NULL UNIQUE, + committed_at TEXT NOT NULL + ); + CREATE TABLE agent_state_head ( + id INTEGER PRIMARY KEY CHECK (id = 1), + revision INTEGER NOT NULL, + state TEXT NOT NULL + ); + `); + db.prepare( + 'INSERT INTO agent_state_meta (id, definition_id, schema_version, kernel_format) VALUES (1, ?, 1, 1)', + ).run(definitionId); + const insert = db.prepare( + 'INSERT INTO agent_state_journal (revision, kind, name, payload, state, to_version, idempotency_key, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)', + ); + insert.run(1, 'event', 'bumped', '{"by":2}', null, null, 'legacy:event', '2026-01-01T00:00:00.000Z'); + insert.run(2, 'reset', null, null, '{"count":5}', null, 'legacy:reset', '2026-01-01T00:00:01.000Z'); + insert.run(3, 'event', 'bumped', '{"by":1}', null, null, 'legacy:event-2', '2026-01-01T00:00:02.000Z'); + db.prepare('INSERT INTO agent_state_head (id, revision, state) VALUES (1, 3, ?)').run('{"count":6}'); + } finally { + db.close(); + } +}; + +const holdUncheckpointedLegacyEvent = (file: string): DatabaseSync => { + const keeper = new DatabaseSync(file); + keeper.exec('PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; BEGIN DEFERRED'); + keeper.prepare('SELECT revision FROM agent_state_head WHERE id = 1').get(); + const writer = new DatabaseSync(file); + try { + writer.exec('PRAGMA wal_autocheckpoint = 0; BEGIN IMMEDIATE'); + writer + .prepare( + 'INSERT INTO agent_state_journal (revision, kind, name, payload, state, to_version, idempotency_key, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)', + ) + .run(4, 'event', 'bumped', '{"by":1}', null, null, 'legacy:event-3', '2026-01-01T00:00:03.000Z'); + writer.prepare('UPDATE agent_state_head SET revision = 4, state = ? WHERE id = 1').run('{"count":7}'); + writer.exec('COMMIT'); + } finally { + writer.close(); + } + return keeper; +}; + const otherDefinition = (): AgentStateDefinition => defineState({ events: counterEvents, @@ -128,6 +212,71 @@ describe('sqlite driver storage behavior', () => { await expect(first.read()).rejects.toMatchObject({ code: 'store-closed' }); })); + it('adopts the legacy root database and live WAL sidecars without data loss', () => + withRoot(async (root) => { + const definition = migratingCounterDefinition(); + const legacyFile = join(root, legacyFileName(definition.id)); + const currentFile = join(root, currentFileName(definition.id)); + createLegacyMigrationDatabase(legacyFile, definition.id); + const keeper = holdUncheckpointedLegacyEvent(legacyFile); + try { + await access(`${legacyFile}-wal`); + await access(`${legacyFile}-shm`); + + const driver = createSqliteStateDriver({ root }); + const store = await driver.open(definition); + + expect(store.location).toBe(currentFile); + expect(await store.read()).toEqual({ revision: 5, state: { count: 70 } }); + await expect(access(legacyFile)).rejects.toMatchObject({ code: 'ENOENT' }); + await expect(access(`${legacyFile}-wal`)).rejects.toMatchObject({ code: 'ENOENT' }); + await expect(access(`${legacyFile}-shm`)).rejects.toMatchObject({ code: 'ENOENT' }); + await access(`${currentFile}-wal`); + await access(`${currentFile}-shm`); + await driver.close(); + } finally { + keeper.exec('ROLLBACK'); + keeper.close(); + } + })); + + it('backfills legacy NULL event results before migration for idempotent replay', () => + withRoot(async (root) => { + const definition = migratingCounterDefinition(); + const file = join(root, 'legacy-null-event.sqlite'); + createLegacyMigrationDatabase(file, definition.id); + + const store = await createSqliteStateDriver({ file }).open(definition); + await expect( + store.dispatch('bumped', { by: 2 }, { idempotencyKey: 'legacy:event' }), + ).resolves.toEqual({ replayed: true, revision: 1, state: { count: 20 } }); + await expect( + store.dispatch('bumped', { by: 1 }, { idempotencyKey: 'legacy:event-2' }), + ).resolves.toEqual({ replayed: true, revision: 3, state: { count: 60 } }); + await store.close(); + })); + + it('preserves legacy reset input while migrating its idempotent result', () => + withRoot(async (root) => { + const definition = migratingCounterDefinition(); + const file = join(root, 'legacy-reset.sqlite'); + createLegacyMigrationDatabase(file, definition.id); + + const store = await createSqliteStateDriver({ file }).open(definition); + await expect( + store.reset({ idempotencyKey: 'legacy:reset', seed: { count: 5 } }), + ).resolves.toEqual({ replayed: true, revision: 2, state: { count: 50 } }); + const db = new DatabaseSync(file); + try { + expect(db.prepare('SELECT state FROM agent_state_journal WHERE revision = 2').get()).toEqual({ + state: '{"count":5}', + }); + } finally { + db.close(); + } + await store.close(); + })); + it('rejects a pending open when the driver closes before initialization resumes', () => withRoot(async (root) => { const driver = createSqliteStateDriver({ root }); @@ -205,6 +354,24 @@ describe('sqlite driver storage behavior', () => { await driver.close(); })); + it('surfaces database close failures when the store scope otherwise succeeds', () => + withRoot(async (root) => { + const closeFailure = new Error('database close failed'); + const originalClose = DatabaseSync.prototype.close; + let failClose = false; + DatabaseSync.prototype.close = function close(this: DatabaseSync): void { + originalClose.call(this); + if (failClose) throw closeFailure; + }; + try { + const store = await createSqliteStateDriver({ root }).open(counterDefinition()); + failClose = true; + await expect(store.close()).rejects.toBe(closeFailure); + } finally { + DatabaseSync.prototype.close = originalClose; + } + })); + it('fails closed with a typed corrupt error when the file is not a database', () => withRoot(async (root) => { const file = join(root, 'state.sqlite');