From 2e136845b1e024b7a127f14b2268e1bad9b603db Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:27:46 +0800 Subject: [PATCH] fix(authority): keep legacy JSON keys readable across the sqlite migration The version-2 delta codec rejected an empty path segment and rebuilt `__proto__` through the inherited accessor, so a retained version-1 projection using either key migrated "successfully" into a state log the new provider could not read. The migration also proved only copied identity and digest columns. Any string is now a legal path segment, decoded keys are written as own data properties, and the migration replays the rows it just wrote through the store's own decoder before swapping tables, so an unreadable log fails closed inside the transaction instead of after it. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- docs/reference/sqlite-authority-store.md | 17 +++- .../coordination/authority_state_log.ts | 24 ++++- .../sqlite_authority_migration.ts | 89 ++++++++++++++++++- .../authority_state_log.test.ts | 42 +++++++++ .../sqlite_authority_migration.test.ts | 68 ++++++++++++++ 5 files changed, 233 insertions(+), 7 deletions(-) diff --git a/docs/reference/sqlite-authority-store.md b/docs/reference/sqlite-authority-store.md index 09d23bc965..bdd6674e25 100644 --- a/docs/reference/sqlite-authority-store.md +++ b/docs/reference/sqlite-authority-store.md @@ -310,14 +310,21 @@ The shipped version-1 database keeps one full projection per retained row, so it cannot be read by the version-2 provider. `sqlite_authority_migration.ts` migrates one Goal database in place: it reads the frozen version-1 rows, proves every stored commit digest, writes the checkpoint/delta log, proves that each -written delta reconstructs its projection, requires the commit count to match, -swaps tables and updates the schema version inside a single +written delta reconstructs its projection, reads the not-yet-swapped tables back +and replays them through the store's own delta decoder, requires the commit +count to match, swaps tables and updates the schema version inside a single `BEGIN IMMEDIATE` transaction. Any failure rolls back and leaves version 1 untouched; a second run reports `already_current`; a rewritten proof, a mismatched goal/incarnation or an existing swap target fails closed. Cursors, operation IDs, commit digests, provider revisions, receipts, events and scan pages are byte-identical after the migration. +Retained projections keep every JSON object key the version-1 provider accepted, +including an empty key and a `__proto__` key: a database the previous provider +could read must not become one the version-2 provider cannot. The replay proof +is what keeps that promise honest, because identical identity and digest columns +alone would not show that a migrated state log is unreadable. + A version-1 database that published only its schema and metadata — the state a goal leaves behind when it selected the provider and never committed — migrates to an equally empty version-2 database instead of failing, so the operator is @@ -390,4 +397,8 @@ delta”：活跃头读取只用自己的行、对应提交和游标连续性自 才写入，`--expected-identity` 可拒绝并非操作者所指的 incarnation，失败保持 v1 原样）。只发布过 schema 与 metadata、从未提交的 v1 库会迁移成同样为空的 v2 库， 不会让操作者落在两个 provider 都不接受的状态。迁移不改 cursor、operation id、 -commit digest、provider revision、receipt、event 或 scan 页面字节。 +commit digest、provider revision、receipt、event 或 scan 页面字节。迁移在提交前 +还会把刚写入的表读回来、用 store 自己的 delta 解码器重放一遍：只核对搬过去的 +标识与摘要无法证明新的状态日志可读。v1 能接受的 JSON key（包括空字符串和 +`__proto__`）在 v2 中保持同样的数据语义，迁移不会把原本可读的库变成读不出来的 +状态。 diff --git a/loopx/control_plane/coordination/authority_state_log.ts b/loopx/control_plane/coordination/authority_state_log.ts index a49faf08ee..1fc4e97bb1 100644 --- a/loopx/control_plane/coordination/authority_state_log.ts +++ b/loopx/control_plane/coordination/authority_state_log.ts @@ -35,6 +35,10 @@ export const AUTHORITY_STATE_CHECKPOINT_INTERVAL = 64; /** * Object path inside one projection. Arrays are addressed as a whole value; * element identity is owned by the domain, not by this codec. + * + * Segments are JSON object keys, and strict JSON allows any string as a key: + * the empty string is a key, and so is `__proto__`. V1 retained whole + * projections, so both are legal retained data and stay legal here. */ export type AuthorityStatePath = readonly string[]; @@ -183,7 +187,22 @@ function applyAuthorityStateOperation(root: JsonObject, operation: AuthorityStat if (!present) protocol("authority state delta removed a value that was never stored"); delete container[last]; } - else container[last] = structuredClone(operation.value); + else setOwnJsonKey(container, last, structuredClone(operation.value)); +} + +/** + * Write one decoded key as an own data property. + * + * `container[key] = value` would run the inherited `__proto__` accessor and + * replace the reconstructed object's prototype with the stored value, so a + * retained projection carrying that key would silently lose it. Decoded + * deltas are data, so every key is created the same way the canonicalizer + * creates keys, which keeps reconstruction exact for every JSON object key. + */ +function setOwnJsonKey(container: JsonObject, key: string, value: unknown): void { + Object.defineProperty(container, key, { + value, writable: true, enumerable: true, configurable: true, + }); } function descend(container: unknown, segment: string): unknown { @@ -233,7 +252,8 @@ function decodeAuthorityStateOperation(value: unknown, index: number): Authority function decodeAuthorityStatePath(value: unknown, label: string): AuthorityStatePath { if (!Array.isArray(value) || value.length === 0) protocol(`${label} path is invalid`); return value.map(segment => { - if (typeof segment === "string" && segment.length > 0) return segment; + // Any string is a legal JSON key, including the empty string. + if (typeof segment === "string") return segment; return protocol(`${label} path segment is invalid`); }); } diff --git a/loopx/control_plane/coordination/sqlite_authority_migration.ts b/loopx/control_plane/coordination/sqlite_authority_migration.ts index 68d6fde3d1..e053876ef6 100644 --- a/loopx/control_plane/coordination/sqlite_authority_migration.ts +++ b/loopx/control_plane/coordination/sqlite_authority_migration.ts @@ -22,8 +22,9 @@ import type { DatabaseSync } from "node:sqlite"; import type { JsonObject } from "../effect_program.ts"; import { AuthorityStoreProtocolError, canonicalAuthorityObject, canonicalAuthorityObjectList, requireAuthorityStoreId } from "./authority_store_codec.ts"; -import { authorityStateCheckpointCursor, authorityStateDelta, - authorityStateDigest, isAuthorityStateCheckpoint } from "./authority_state_log.ts"; +import { applyAuthorityStateDelta, authorityStateCheckpointCursor, authorityStateDelta, + authorityStateDigest, decodeAuthorityStateDelta, + isAuthorityStateCheckpoint } from "./authority_state_log.ts"; import { SQLITE_AUTHORITY_STORE_SCHEMA, commitDigest, @@ -192,6 +193,13 @@ function executeSqliteAuthorityMigration( } const stateDigest = authorityStateDigest(projection); const delta = authorityStateDelta(previous ?? {}, projection); + // Prove the encoder before anything is written, exactly like a live + // commit: replaying the stored delta must reproduce this projection + // byte for byte. A retained projection the new format cannot read + // would otherwise be copied into V2 and only fail on a later read. + if (authorityStateDigest(applyAuthorityStateDelta(previous ?? {}, delta)) !== stateDigest) { + throw new AuthorityStoreProtocolError("V1 authority state delta does not reconstruct its commit"); + } if (isAuthorityStateCheckpoint(cursor)) { insertCheckpoint.run(cursor.toString(), JSON.stringify(projection), stateDigest); checkpoints += 1; @@ -214,6 +222,11 @@ function executeSqliteAuthorityMigration( db.prepare("INSERT INTO head_v2 VALUES (1, ?, ?, ?)").run(cursor.toString(), JSON.stringify(previous), previousDigest); } + // Prove the new format is readable before it is adopted: every retained + // row is read back from its stored text and replayed through the codec the + // store reads with, so the swap cannot publish a log only its writer can + // decode. + verifyWrittenStateLogReadable(db, commits); // Swap only after every retained row was proved and re-published. db.exec("DROP TABLE head"); db.exec("DROP TABLE commits"); @@ -243,6 +256,78 @@ function executeSqliteAuthorityMigration( } finally { db?.close(); } } +/** + * Read the not-yet-swapped V2 tables back and prove they reconstruct. + * + * The migration already proved each delta against its in-memory predecessor. + * This reads the rows it actually wrote, decodes them with the store's own + * delta decoder, replays them from the empty state, and requires every payload + * to match the digest the row published. Identical identity and digest columns + * alone would not catch a state log the new format cannot read. + */ +function verifyWrittenStateLogReadable(db: DatabaseSync, commits: number): void { + const page = db.prepare(`SELECT CAST(cursor AS TEXT) AS sequence, state_digest, parent_state_digest, + delta FROM commits_v2 WHERE cursor > ? ORDER BY cursor LIMIT ?`); + let state: JsonObject = {}; + let digest: string | null = null; + let cursor = 0n; + let read = 0; + for (;;) { + const rows = page.all(cursor.toString(), BigInt(MIGRATION_PAGE)) as unknown as + Record[]; + if (rows.length === 0) break; + for (const row of rows) { + const sequence = row.sequence; + if (typeof sequence !== "string" || !/^[1-9]\d*$/.test(sequence)) { + throw new AuthorityStoreProtocolError("migrated authority store cursor is invalid"); + } + cursor = BigInt(sequence); + read += 1; + if ((row.parent_state_digest === null ? null : String(row.parent_state_digest)) !== digest) { + throw new AuthorityStoreProtocolError("migrated authority state log parent lineage is invalid"); + } + state = applyAuthorityStateDelta(state, + decodeAuthorityStateDelta(JSON.parse(String(row.delta)) as unknown)); + if (authorityStateDigest(state) !== row.state_digest) { + throw new AuthorityStoreProtocolError("migrated authority state log digest mismatch"); + } + digest = String(row.state_digest); + } + } + if (read !== commits) { + throw new AuthorityStoreProtocolError("migrated authority state log lost retained transactions"); + } + const retainedDigest = db.prepare("SELECT state_digest FROM commits_v2 WHERE cursor = ?"); + for (const row of db.prepare(`SELECT CAST(cursor AS TEXT) AS sequence, projection, projection_digest + FROM checkpoints_v2 ORDER BY cursor`).all() as unknown as Record[]) { + const projection = canonicalAuthorityObject(JSON.parse(String(row.projection)) as unknown, + "migrated authority checkpoint projection"); + if (authorityStateDigest(projection) !== row.projection_digest) { + throw new AuthorityStoreProtocolError("migrated authority checkpoint is not readable"); + } + if (retainedDigest.get(String(row.sequence))?.state_digest !== row.projection_digest) { + throw new AuthorityStoreProtocolError("migrated authority checkpoint does not cover its cursor"); + } + } + const head = db.prepare("SELECT CAST(cursor AS TEXT) AS sequence, projection, state_digest FROM head_v2 WHERE singleton = 1") + .get() as Record | undefined; + if (read === 0) { + if (head !== undefined) { + throw new AuthorityStoreProtocolError( + "migrated authority head was published without a retained transaction"); + } + return; + } + if (head === undefined || head.sequence !== cursor.toString() || head.state_digest !== digest) { + throw new AuthorityStoreProtocolError("migrated authority head does not match its state log"); + } + const projection = canonicalAuthorityObject(JSON.parse(String(head.projection)) as unknown, + "migrated authority head projection"); + if (authorityStateDigest(projection) !== digest) { + throw new AuthorityStoreProtocolError("migrated authority head is not readable"); + } +} + /** * Read the migrated store back and prove that its retained transaction * identity sequence and authority lineage are exactly the pre-migration ones. diff --git a/tests/control_plane_ts/authority_state_log.test.ts b/tests/control_plane_ts/authority_state_log.test.ts index 2a1b0f3227..05e475f133 100644 --- a/tests/control_plane_ts/authority_state_log.test.ts +++ b/tests/control_plane_ts/authority_state_log.test.ts @@ -12,6 +12,8 @@ import { decodeAuthorityStateDelta, isAuthorityStateCheckpoint, } from "../../loopx/control_plane/coordination/authority_state_log.ts"; +import {canonicalAuthorityBytes} from + "../../loopx/control_plane/coordination/authority_store_codec.ts"; test("authority state deltas reconstruct every committed projection exactly", () => { const previous = { @@ -108,3 +110,43 @@ test("authority state digests and checkpoint windows are stable and bounded", () assert.equal(isAuthorityStateCheckpoint(64n), false); assert.throws(() => authorityStateCheckpointCursor(0n), /checkpoint cursor/u); }); + +test("authority state deltas keep every JSON key the stored projection could carry", () => { + // V1 retained a whole projection per commit, so every JSON object key was + // legal legacy input. The delta codec inherits that contract: an empty key + // is an ordinary key, and a `__proto__` key is stored data instead of the + // inherited accessor. Keys that only existed in V1 data would otherwise be + // copied into the new format and fail on the first read. + const text = (value: unknown): string => canonicalAuthorityBytes(value).toString("utf8"); + const stored = (previous: unknown, next: unknown): unknown => JSON.parse( + JSON.stringify(authorityStateDelta(previous as never, next as never))) as unknown; + + const previous = JSON.parse('{"": {"empty": true}, "__proto__": {"own": 1}, "todos": []}') as + Record; + const next = JSON.parse('{"": {"empty": false}, "__proto__": {"own": 2},' + + ' "nested": {"__proto__": {"deep": true}}, "todos": [{"todo_id": "todo-0"}]}') as + Record; + const applied = applyAuthorityStateDelta(previous, decodeAuthorityStateDelta(stored(previous, next))); + assert.equal(text(applied), text(next)); + // The reconstruction is an ordinary JSON object: `__proto__` stayed a key. + assert.equal(Object.getPrototypeOf(applied), Object.prototype); + assert.deepEqual(Object.keys(applied).sort(), ["", "__proto__", "nested", "todos"]); + assert.deepEqual(applied[""], {empty: false}); + assert.deepEqual(applied["__proto__"], {own: 2}); + assert.deepEqual(applied["nested"], {["__proto__"]: {deep: true}}); + + // Removing a legacy key is legal too, including the empty key and a key that + // the previous state only carried inside a nested object. + const trimmed = JSON.parse('{"nested": {}, "todos": []}') as Record; + const removal = applyAuthorityStateDelta(next, decodeAuthorityStateDelta(stored(next, trimmed))); + assert.equal(text(removal), text(trimmed)); + assert.deepEqual(Object.keys(removal).sort(), ["nested", "todos"]); + assert.equal(Object.hasOwn(removal, "__proto__"), false); + + // A legacy key the delta has to create is an own data property as well. + const legacyText = '{"__proto__": {"own": 1}, "": 0}'; + const created = applyAuthorityStateDelta({}, + decodeAuthorityStateDelta(stored({}, JSON.parse(legacyText) as Record))); + assert.equal(text(created), text(JSON.parse(legacyText))); + assert.equal(Object.getPrototypeOf(created), Object.prototype); +}); diff --git a/tests/control_plane_ts/sqlite_authority_migration.test.ts b/tests/control_plane_ts/sqlite_authority_migration.test.ts index 7327682863..0edd906499 100644 --- a/tests/control_plane_ts/sqlite_authority_migration.test.ts +++ b/tests/control_plane_ts/sqlite_authority_migration.test.ts @@ -232,6 +232,74 @@ function runMigrationCli(args: readonly string[]): {status: number | null; stdou "--experimental-strip-types", script, ...args], {encoding: "utf8"}); } +/** + * Retained V1 history that used only JSON object keys V1 never restricted. + * + * V1 stored a whole projection per commit, so `""` was an ordinary key and + * `__proto__` was data. The migration has to carry both into the new format and + * prove they still read back before it commits. + */ +function legacyKeySeeds(): SqliteAuthorityV1Seed[] { + return [ + {operation_id: "legacy-op-001", + projection: {authority_revision: 1, "": {empty: true}, ["__proto__"]: {own: 1}, todos: []}}, + {operation_id: "legacy-op-002", + projection: {authority_revision: 2, "": {empty: true}, ["__proto__"]: {own: 1}, + nested: {["__proto__"]: {deep: true}}, todos: [{todo_id: "todo-0", status: "open"}]}}, + {operation_id: "legacy-op-003", + projection: {authority_revision: 3, "": {empty: false}, nested: {}, + todos: [{todo_id: "todo-0", status: "done"}]}}, + ]; +} + +test("SQLite V1 migration keeps every legacy JSON key readable", async t => { + const root = await directory(t); + const v1 = createSqliteAuthorityStoreV1(root, GOAL_ID, legacyKeySeeds()); + const {DatabaseSync: Reader} = createRequire(import.meta.url)("node:sqlite"); + const legacy = new Reader(v1.path, {readOnly: true}); + try { + // The frozen V1 rows really carry the keys this test is about. + const text = String(legacy.prepare("SELECT projection FROM commits WHERE cursor = 2").get()?.projection); + assert.match(text, /"__proto__"/u); + assert.match(text, /""\s*:/u); + } finally { legacy.close(); } + const migrated = migrateSqliteAuthorityStoreV1ToV2(root, GOAL_ID, {execute: true}); + assert.equal(migrated.status, "migrated", JSON.stringify(migrated)); + assert.equal(migrated.commits, 3); + const store = new SqliteAuthorityStore(root, GOAL_ID, {existingOnly: true, expectedIdentity: v1.identity}); + // The live head reads back as the exact projection the last V1 commit published. + const head = await store.loadAuthority(); + assert.equal(head.status, "loaded", JSON.stringify(head)); + if (head.status === "loaded") { + assert.equal(canonicalAuthorityBytes(head.head).toString("utf8"), + JSON.stringify(v1.rows[2]!.operation_receipt.projection)); + // The empty key survives the migration, and the last commit really did + // drop `__proto__` rather than the codec silently dropping it for us. + assert.equal(Object.hasOwn(head.head, ""), true); + assert.equal(Object.hasOwn(head.head, "__proto__"), false); + assert.equal(Object.getPrototypeOf(head.head), Object.prototype); + } + // A retained projection keeps the legacy `__proto__` key as stored data. + const page = await store.scanCommitted(null, 6); + assert.equal(page.status, "page", JSON.stringify(page)); + if (page.status === "page") { + const retained = page.transactions[1]!.projection; + assert.equal(Object.hasOwn(retained, "__proto__"), true); + assert.equal(Object.hasOwn(retained, ""), true); + assert.equal(Object.getPrototypeOf(retained), Object.prototype); + assert.deepEqual(retained["__proto__"], {own: 1}); + } + // Every retained projection reads back byte-identically, not only the head. + const history = await logicalHistory(store); + for (const [index, row] of v1.rows.entries()) { + const parsed = JSON.parse(history[2 + index]!) as Record; + assert.equal(parsed.cursor, row.cursor); + assert.equal(parsed.operation_id, row.operation_id); + assert.deepEqual(parsed.projection, JSON.stringify(row.operation_receipt.projection)); + } + assert.equal((await store.verifyAuthorityHistory()).status, "verified"); +}); + test("SQLite V1 migration refuses a rewritten proof and leaves the database intact", async t => { const root = await directory(t); const v1 = createSqliteAuthorityStoreV1(root, GOAL_ID, seeds());