diff --git a/apps/api/src/indexer/protocol-checkpoint.ts b/apps/api/src/indexer/protocol-checkpoint.ts new file mode 100644 index 0000000..778f31a --- /dev/null +++ b/apps/api/src/indexer/protocol-checkpoint.ts @@ -0,0 +1,49 @@ +import { counterpartyJson } from "#api/integrations/counterparty"; +import type { ReplayIdentity } from "#api/queries/replay-checkpoints"; + +export const PENDING_PROTOCOL_KEY = "replay_protocol_identities"; + +export async function fetchReplayIdentity(api: string, height: number, parsed = false): Promise { + const { result } = await counterpartyJson<{ result?: Partial }>(api, `/blocks/${height}`); + if (!result?.block_hash || (result.block_index !== undefined && result.block_index !== height)) + throw new Error("Cannot verify applied block hash"); + const identity = { + block_index: height, + block_hash: result.block_hash, + ledger_hash: result.ledger_hash ?? null, + messages_hash: result.messages_hash ?? null, + }; + if (parsed) requireParsedIdentity(identity); + return identity; +} + +export function requireParsedIdentity(identity: ReplayIdentity): void { + if ( + ![identity.block_hash, identity.ledger_hash, identity.messages_hash].every( + (hash) => typeof hash === "string" && /^[a-f0-9]{64}$/i.test(hash), + ) + ) + throw new Error("Parsed block protocol hashes unavailable"); +} + +export function sameReplayIdentity(stored: ReplayIdentity, fresh: ReplayIdentity): boolean { + if ((stored.ledger_hash && !fresh.ledger_hash) || (stored.messages_hash && !fresh.messages_hash)) + throw new Error("Parsed block protocol hashes unavailable"); + return ( + stored.block_hash === fresh.block_hash && + (!stored.ledger_hash || stored.ledger_hash === fresh.ledger_hash) && + (!stored.messages_hash || stored.messages_hash === fresh.messages_hash) + ); +} + +export function pendingReplayIdentities(raw: string | null): ReplayIdentity[] { + if (raw === null) return []; + const rows: ReplayIdentity[] = JSON.parse(raw); + if (!Array.isArray(rows) || rows.length > 50) throw new Error("Invalid pending protocol identities"); + for (const row of rows) { + if (!Number.isSafeInteger(row.block_index) || row.block_index < 0) + throw new Error("Invalid pending protocol height"); + requireParsedIdentity(row); + } + return rows; +} diff --git a/apps/api/src/indexer/sync.ts b/apps/api/src/indexer/sync.ts index 5cad9d1..768917a 100644 --- a/apps/api/src/indexer/sync.ts +++ b/apps/api/src/indexer/sync.ts @@ -1,4 +1,11 @@ import { initializeUndo, pruneUndo, restoreUndo, writeWithUndo } from "#api/indexer/reorg-undo"; +import { + fetchReplayIdentity, + sameReplayIdentity, + pendingReplayIdentities, + PENDING_PROTOCOL_KEY, +} from "#api/indexer/protocol-checkpoint"; +import { storedReplayIdentity, lastParsedIdentity } from "#api/queries/replay-checkpoints"; /** * Counterparty mirror engine — chronological event replay into normalized D1. * @@ -230,10 +237,11 @@ export async function syncCoreEvents( cursorBlock > rollbackTo ) throw new Error("Invalid pending rollback"); - const local = await env.CORE_DB.prepare("SELECT lower(hex(block_hash)) hash FROM blocks WHERE block_index=?") - .bind(rollbackTo) - .first<{ hash: string }>(); - if (!local?.hash || local.hash !== (await blockHash(env.COUNTERPARTY_API_BASE, rollbackTo))) { + const local = await storedReplayIdentity(env.CORE_DB, rollbackTo); + if ( + !local?.block_hash || + !sameReplayIdentity(local, await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, rollbackTo)) + ) { throw new Error("Pending rollback no longer has a verified common ancestor"); } await rollbackCoreDatabase(env.CORE_DB, rollbackTo, rollbackIndex); @@ -246,6 +254,7 @@ export async function syncCoreEvents( env.CORE_DB.prepare( "DELETE FROM core_state WHERE key IN ('rollback_to','rollback_event_index','rollback_cursor_block')", ), + env.CORE_DB.prepare("DELETE FROM core_state WHERE key=?").bind(PENDING_PROTOCOL_KEY), ]); } @@ -256,6 +265,19 @@ export async function syncCoreEvents( // Persist the independently observed source height before replay. Even when a later event handler fails, // readiness must report the lag instead of comparing the local mirror to itself and claiming "synced". await setCoreStateStmt(env.CORE_DB, "source_tip_block", String(sourceTipBlock)).run(); + // This fixed parsed tip commits to protocol history before event pages are + // fetched. A later descendant is fine; changing this identity is not. + const sourceAnchor = await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, sourceTipBlock, true); + let pendingMismatch: number | null = null; + let retainedIdentities = pendingReplayIdentities(await getCoreState(env.CORE_DB, PENDING_PROTOCOL_KEY)); + for (const identity of retainedIdentities) { + if ( + identity.block_index > sourceTipBlock || + !sameReplayIdentity(identity, await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, identity.block_index, true)) + ) { + pendingMismatch = Math.min(pendingMismatch ?? identity.block_index, identity.block_index); + } + } if (lastBlock > 0) { // Raw page writes can precede the durable cursor. Verify those headers // too, or a crash followed by a fork could retain orphan INSERT OR IGNORE rows. @@ -263,26 +285,34 @@ export async function syncCoreEvents( "SELECT block_index FROM blocks ORDER BY block_index DESC LIMIT 1", ).first<{ block_index: number }>(); const verifyBlock = Math.max(lastBlock, rawTip?.block_index ?? lastBlock); - const [storedHash, actualHash] = await Promise.all([ - env.CORE_DB.prepare(`SELECT lower(hex(block_hash)) hash FROM blocks WHERE block_index=?`) - .bind(verifyBlock) - .first<{ hash: string }>() - .then((row) => row?.hash || null), - sourceTipBlock < verifyBlock - ? Promise.resolve("past-source-tip") - : blockHash(env.COUNTERPARTY_API_BASE, verifyBlock), - ]); - if (!storedHash || !actualHash) throw new Error("Cannot verify applied block hash"); - if (storedHash !== actualHash) { - let rollbackTo = verifyBlock - 1; + const stored = await storedReplayIdentity(env.CORE_DB, verifyBlock); + if (!stored?.block_hash) throw new Error("Cannot verify applied block hash"); + let changedBlock = pendingMismatch; + if ( + sourceTipBlock < verifyBlock || + !sameReplayIdentity(stored, await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, verifyBlock)) + ) + changedBlock = Math.min(changedBlock ?? verifyBlock, verifyBlock); + // The cursor can be inside an unfinished block. Its completed predecessor + // still commits to protocol history and must be checked every invocation. + const parsed = await lastParsedIdentity(env.CORE_DB, verifyBlock); + if ( + parsed && + parsed.block_index !== verifyBlock && + !sameReplayIdentity(parsed, await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, parsed.block_index, true)) + ) + changedBlock = Math.min(changedBlock ?? parsed.block_index, parsed.block_index); + if (changedBlock !== null) { + let rollbackTo = changedBlock - 1; let found = false; - for (; rollbackTo >= Math.max(0, verifyBlock - 24); rollbackTo--) { - const retained = await env.CORE_DB.prepare( - `SELECT lower(hex(block_hash)) hash FROM blocks WHERE block_index=?`, - ) - .bind(rollbackTo) - .first<{ hash: string }>(); - if (retained?.hash && retained.hash === (await blockHash(env.COUNTERPARTY_API_BASE, rollbackTo))) { + const protocolRequired = pendingMismatch !== null || !!parsed; + for (; rollbackTo >= Math.max(0, changedBlock - 24); rollbackTo--) { + const retained = await storedReplayIdentity(env.CORE_DB, rollbackTo); + if ( + retained?.block_hash && + (!protocolRequired || (retained.ledger_hash && retained.messages_hash)) && + sameReplayIdentity(retained, await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, rollbackTo)) + ) { found = true; break; } @@ -296,6 +326,7 @@ export async function syncCoreEvents( setCoreStateStmt(env.CORE_DB, "rollback_cursor_block", String(cursorBlock)), ]); await rollbackCoreDatabase(env.CORE_DB, rollbackTo, rollbackIndex); + retainedIdentities = []; lastIndex = rollbackIndex; lastBlock = cursorBlock; await env.CORE_DB.batch([ @@ -305,9 +336,10 @@ export async function syncCoreEvents( env.CORE_DB.prepare( "DELETE FROM core_state WHERE key IN ('rollback_to','rollback_event_index','rollback_cursor_block')", ), + env.CORE_DB.prepare("DELETE FROM core_state WHERE key=?").bind(PENDING_PROTOCOL_KEY), ]); } - } + } else if (pendingMismatch !== null) throw new Error("Protocol changed before first checkpoint; rebuild required"); const cap = Math.min(opts.maxEvents ?? MAX_EVENTS_PER_RUN, MAX_EVENTS_PER_RUN); let applied = 0; let chunk = CHUNK; @@ -328,6 +360,19 @@ export async function syncCoreEvents( if (events.length === 0) break; for (let offset = 0; offset < events.length; offset += APPLY_CHUNK) { const slice = events.slice(offset, offset + APPLY_CHUNK); + if (slice.some((event) => event.block_index > sourceTipBlock)) + throw new Error("Events advanced beyond this pass's fixed anchor; retry"); + const identities = await Promise.all( + [...new Set(slice.map((event) => event.block_index))].map((height) => + fetchReplayIdentity(env.COUNTERPARTY_API_BASE, height, true), + ), + ); + for (const retained of retainedIdentities) { + const fresh = + identities.find((row) => row.block_index === retained.block_index) ?? + (await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, retained.block_index, true)); + if (!sameReplayIdentity(retained, fresh)) throw new Error("Protocol changed between replay slices"); + } const ctx: Ctx = { stmts: [], identities: createIdentitySet(), @@ -362,18 +407,50 @@ export async function syncCoreEvents( previousHash = hash; } dispatch(event, ctx); + if (event.event === "BLOCK_PARSED") { + const identity = identities.find((row) => row.block_index === event.block_index)!; + if ( + event.params.ledger_hash !== identity.ledger_hash || + event.params.messages_hash !== identity.messages_hash + ) + throw new Error("Parsed event protocol identity changed"); + } } // A fork during page retrieval must not overwrite the checkpoint with // a child from the new branch and hide an orphaned parent forever. + if (lastBlock > 0) { + const anchor = await storedReplayIdentity(env.CORE_DB, lastBlock); + if (!anchor || !sameReplayIdentity(anchor, await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, lastBlock))) + throw new Error("Applied chain changed while fetching events"); + } + for (const identity of identities) { + if ( + !sameReplayIdentity( + identity, + await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, identity.block_index, true), + ) + ) + throw new Error("Protocol changed while fetching events"); + } + for (const retained of retainedIdentities.filter( + (row) => !identities.some((identity) => identity.block_index === row.block_index), + )) { + if ( + !sameReplayIdentity( + retained, + await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, retained.block_index, true), + ) + ) + throw new Error("Protocol anchor changed while fetching events"); + } if ( - lastBlock > 0 && - (await blockHash(env.COUNTERPARTY_API_BASE, lastBlock)) !== - (await env.CORE_DB.prepare("SELECT lower(hex(block_hash)) hash FROM blocks WHERE block_index=?") - .bind(lastBlock) - .first<{ hash: string }>() - .then((row) => row?.hash)) + !sameReplayIdentity( + sourceAnchor, + await fetchReplayIdentity(env.COUNTERPARTY_API_BASE, sourceAnchor.block_index, true), + ) ) - throw new Error("Applied chain changed while fetching events"); + throw new Error("Protocol source anchor changed while fetching events"); + await setCoreStateStmt(env.CORE_DB, PENDING_PROTOCOL_KEY, JSON.stringify(identities)).run(); await batchAll(env.CORE_DB, [...dictionaryStatements(ctx.identities), ...ctx.stmts]); const dispenseTxs = slice .filter((event) => event.event === "DISPENSE") @@ -393,6 +470,7 @@ export async function syncCoreEvents( setCoreStateStmt(env.CORE_DB, "last_block_index", String(lastBlock)), checkpointHashStmt(env.CORE_DB, lastBlock), ]); + retainedIdentities = identities; } } diff --git a/apps/api/src/queries/replay-checkpoints.ts b/apps/api/src/queries/replay-checkpoints.ts new file mode 100644 index 0000000..8337b2b --- /dev/null +++ b/apps/api/src/queries/replay-checkpoints.ts @@ -0,0 +1,29 @@ +import { one } from "#api/db"; + +export interface ReplayIdentity { + block_index: number; + block_hash: string; + ledger_hash: string | null; + messages_hash: string | null; +} + +export function storedReplayIdentity(db: D1Database, height: number): Promise { + return one( + db, + `SELECT block_index,lower(hex(block_hash)) block_hash, + nullif(lower(hex(ledger_hash)),'') ledger_hash,nullif(lower(hex(messages_hash)),'') messages_hash + FROM blocks WHERE block_index=?`, + height, + ); +} + +export function lastParsedIdentity(db: D1Database, height: number): Promise { + return one( + db, + `SELECT block_index,lower(hex(block_hash)) block_hash, + lower(hex(ledger_hash)) ledger_hash,lower(hex(messages_hash)) messages_hash + FROM blocks WHERE block_index<=? AND length(ledger_hash)=32 AND length(messages_hash)=32 + ORDER BY block_index DESC LIMIT 1`, + height, + ); +} diff --git a/apps/api/tests/core-foundation-writes.test.ts b/apps/api/tests/core-foundation-writes.test.ts index a65648b..3a9fbf8 100644 --- a/apps/api/tests/core-foundation-writes.test.ts +++ b/apps/api/tests/core-foundation-writes.test.ts @@ -306,7 +306,10 @@ test("a stale state hash cannot trigger a false rollback after a partial page", const url = parsed.toString(); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result: [{ event_index: 10 }] })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 101 } })); - if (url.endsWith("/blocks/101")) return new Response(JSON.stringify({ result: { block_hash: hash } })); + if (url.endsWith("/blocks/101")) + return new Response( + JSON.stringify({ result: { block_hash: hash, ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) } }), + ); throw new Error(`Unexpected rollback request: ${url}`); }; try { @@ -367,8 +370,16 @@ test("a fork after raw writes but before the cursor commit discards the orphan p const url = parsed.toString(); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result: [{ event_index: 9 }] })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 101 } })); - if (url.endsWith("/blocks/101")) return new Response(JSON.stringify({ result: { block_hash: "cc".repeat(32) } })); - if (url.endsWith("/blocks/100")) return new Response(JSON.stringify({ result: { block_hash: hash } })); + if (url.endsWith("/blocks/101")) + return new Response( + JSON.stringify({ + result: { block_hash: "cc".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }), + ); + if (url.endsWith("/blocks/100")) + return new Response( + JSON.stringify({ result: { block_hash: hash, ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) } }), + ); if (url.includes("/events?cursor=9")) return new Response(JSON.stringify({ result: [{ event_index: 9, block_index: 100 }] })); throw new Error(`unexpected request: ${url}`); @@ -394,8 +405,14 @@ test("a missing event is never skipped even if a later page contains valid event const parsed = new URL(String(input)); parsed.searchParams.delete("verbose"); const url = parsed.toString(); - if (url.endsWith("/blocks/100")) return Response.json({ result: { block_hash: "a0".repeat(32) } }); - if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "a1".repeat(32) } }); + if (url.endsWith("/blocks/100")) + return Response.json({ + result: { block_hash: "a0".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); + if (url.endsWith("/blocks/101")) + return Response.json({ + result: { block_hash: "a1".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result: [{ event_index: 2 }] })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 100 } })); if (url.endsWith("/blocks/100")) return new Response(JSON.stringify({ result: {} })); @@ -451,7 +468,10 @@ test("pending rollback survives a process exit after branch deletion before the const url = parsed.toString(); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result: [{ event_index: 9 }] })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 100 } })); - if (url.endsWith("/blocks/100")) return new Response(JSON.stringify({ result: { block_hash: hash } })); + if (url.endsWith("/blocks/100")) + return new Response( + JSON.stringify({ result: { block_hash: hash, ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) } }), + ); throw new Error(`unexpected request: ${url}`); }; try { @@ -1694,8 +1714,14 @@ test("replay advances its durable cursor", async () => { const parsed = new URL(String(input)); parsed.searchParams.delete("verbose"); const url = parsed.toString(); - if (url.endsWith("/blocks/100")) return Response.json({ result: { block_hash: "a0".repeat(32) } }); - if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "a1".repeat(32) } }); + if (url.endsWith("/blocks/100")) + return Response.json({ + result: { block_hash: "a0".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); + if (url.endsWith("/blocks/101")) + return Response.json({ + result: { block_hash: "a1".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); if (url.endsWith("/events?limit=1")) { // The event row is authoritative even if a node's result_count uses // different indexing semantics. @@ -1785,8 +1811,14 @@ test("replay checkpoints a dense fetched page in durable slices", async () => { const parsed = new URL(String(input)); parsed.searchParams.delete("verbose"); const url = parsed.toString(); - if (url.endsWith("/blocks/100")) return Response.json({ result: { block_hash: "a0".repeat(32) } }); - if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "a1".repeat(32) } }); + if (url.endsWith("/blocks/100")) + return Response.json({ + result: { block_hash: "a0".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); + if (url.endsWith("/blocks/101")) + return Response.json({ + result: { block_hash: "a1".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); if (url.endsWith("/events?limit=1")) { return new Response(JSON.stringify({ result_count: 251, result: [{ event_index: 251 }] })); } @@ -1836,8 +1868,18 @@ test("replay clamps the events page to the pending window and shrinks it when a const parsed = new URL(String(input)); parsed.searchParams.delete("verbose"); const url = parsed.toString(); - if (url.endsWith("/blocks/100")) return Response.json({ result: { block_hash: "a0".repeat(32) } }); - if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "a1".repeat(32) } }); + if (url.endsWith("/blocks/100")) + return Response.json({ + result: { block_hash: "a0".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); + if (url.endsWith("/blocks/102")) + return Response.json({ + result: { block_hash: "a2".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); + if (url.endsWith("/blocks/101")) + return Response.json({ + result: { block_hash: "a1".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); if (url.endsWith("/events?limit=1")) { return new Response(JSON.stringify({ result_count: 3 }), { status: 200 }); } @@ -1897,7 +1939,12 @@ test("compact replay rolls back a mismatched checkpoint before accepting the rep const url = parsed.toString(); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result_count: 10 })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 101 } })); - if (url.endsWith("/blocks/101")) return new Response(JSON.stringify({ result: { block_hash: "replacement-101" } })); + if (url.endsWith("/blocks/101")) + return new Response( + JSON.stringify({ + result: { block_hash: "dd".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }), + ); if (url.includes("/events?cursor=10")) return new Response( JSON.stringify({ @@ -1907,7 +1954,12 @@ test("compact replay rolls back a mismatched checkpoint before accepting the rep ], }), ); - if (url.endsWith("/blocks/100")) return new Response(JSON.stringify({ result: { block_hash: "a0".repeat(32) } })); + if (url.endsWith("/blocks/100")) + return new Response( + JSON.stringify({ + result: { block_hash: "a0".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }), + ); throw new Error(`unexpected Counterparty request: ${url}`); }; try { @@ -2079,8 +2131,14 @@ test("a stale parent height probe cannot splice a new event header onto the old globalThis.fetch = async (input) => { const url = new URL(String(input)); if (url.pathname === "/blocks/last") return Response.json({ result: { block_index: 101 } }); - if (url.pathname === "/blocks/100") return Response.json({ result: { block_hash: "aa".repeat(32) } }); - if (url.pathname === "/blocks/101") return Response.json({ result: { block_hash: "cc".repeat(32) } }); + if (url.pathname === "/blocks/100") + return Response.json({ + result: { block_hash: "aa".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); + if (url.pathname === "/blocks/101") + return Response.json({ + result: { block_hash: "cc".repeat(32), ledger_hash: "bb".repeat(32), messages_hash: "cc".repeat(32) }, + }); if (!url.searchParams.has("cursor")) return Response.json({ result: [{ event_index: 1 }] }); return Response.json({ result: [ @@ -2155,3 +2213,115 @@ test("mutable undo rolls back atomically on failure and rejects pre-deployment h await assert.rejects(restoreUndo(db, 99), /predates exact undo/); database.close(); }); + +for (const messagesOnly of [false, true]) + for (const interrupted of [false, true]) + test(`protocol-only reparse restores exact state (interrupted=${interrupted}, messagesOnly=${messagesOnly})`, async () => { + const database = new DatabaseSync(":memory:"); + database.exec(CORE_DDL); + const hash = "aa".repeat(32), + ledger = "bb".repeat(32), + messages = "cc".repeat(32), + replacement = "dd".repeat(32); + database.exec(`INSERT INTO reorg_undo_state VALUES(1,100); + INSERT INTO blocks(block_index,block_hash,ledger_hash,messages_hash) VALUES(100,X'${hash}',X'${ledger}',X'${messages}'),(101,X'${hash}',${interrupted ? "NULL" : `X'${ledger}'`},${interrupted ? "NULL" : `X'${messages}'`}); + INSERT INTO core_state VALUES('last_block_index','${interrupted ? 100 : 101}'),('last_event_index','${interrupted ? 0 : 2}'); + INSERT INTO orders(tx_index,tx_hash,block_index,status,give_remaining) VALUES(1,zeroblob(32),100,'open','75');`); + const db = d1(database); + await writeWithUndo(db, [ + { block: 101, statement: db.prepare("UPDATE orders SET status='filled',give_remaining='0'") }, + ]); + if (interrupted) + database + .prepare("INSERT INTO core_state VALUES(?,?)") + .run( + "replay_protocol_identities", + JSON.stringify([{ block_index: 101, block_hash: hash, ledger_hash: ledger, messages_hash: messages }]), + ); + const original = globalThis.fetch; + let latestReads = 0; + const events = [ + { + event_index: 0, + block_index: 100, + event: "BLOCK_PARSED", + params: { ledger_hash: ledger, messages_hash: messages }, + }, + { + event_index: 1, + block_index: 101, + event: "NEW_BLOCK", + params: { block_hash: hash, previous_block_hash: hash }, + }, + { + event_index: 2, + block_index: 101, + event: "BLOCK_PARSED", + params: { ledger_hash: messagesOnly ? ledger : replacement, messages_hash: replacement }, + }, + ]; + globalThis.fetch = async (input) => { + const u = new URL(String(input)); + if (u.pathname === "/blocks/last") { + latestReads++; + return Response.json({ result: { block_index: latestReads === 1 ? 101 : 102 } }); + } + if (u.pathname.startsWith("/blocks/")) { + const height = Number(u.pathname.split("/").at(-1)); + return Response.json({ + result: { + block_index: height, + block_hash: hash, + ledger_hash: height === 101 && !messagesOnly ? replacement : ledger, + messages_hash: height === 101 ? replacement : messages, + }, + }); + } + const cursor = u.searchParams.has("cursor") ? Number(u.searchParams.get("cursor")) : Infinity; + return Response.json({ + result: events + .filter((e) => e.event_index <= cursor) + .reverse() + .slice(0, Number(u.searchParams.get("limit"))), + }); + }; + try { + const result = await syncCoreEvents({ CORE_DB: db, COUNTERPARTY_API_BASE: "https://core.test" }); + assert.equal(result.caught_up, true); + assert.equal(result.applied, 2); + assert.equal(latestReads, 1); + assert.equal(database.prepare("SELECT give_remaining FROM orders").get()?.give_remaining, "75"); + assert.equal( + database.prepare("SELECT lower(hex(ledger_hash)) h FROM blocks WHERE block_index=101").get()?.h, + messagesOnly ? ledger : replacement, + ); + assert.equal(database.prepare("SELECT value FROM core_state WHERE key='rollback_to'").get(), undefined); + assert.ok(database.prepare("SELECT value FROM core_state WHERE key='replay_protocol_identities'").get()); + } finally { + globalThis.fetch = original; + database.close(); + } + }); + +test("missing protocol evidence leaves existing state and cursors intact", async () => { + const database = new DatabaseSync(":memory:"); + database.exec(CORE_DDL); + database.exec(`INSERT INTO blocks(block_index,block_hash,ledger_hash,messages_hash) VALUES(100,zeroblob(32),zeroblob(32),zeroblob(32)); + INSERT INTO core_state VALUES('last_block_index','100'),('last_event_index','0');`); + const original = globalThis.fetch; + globalThis.fetch = async (input) => + new URL(String(input)).pathname === "/events" + ? Response.json({ result: [{ event_index: 0 }] }) + : Response.json({ result: { block_index: 100, block_hash: "00".repeat(32) } }); + try { + await assert.rejects( + syncCoreEvents({ CORE_DB: d1(database), COUNTERPARTY_API_BASE: "https://core.test" }), + /protocol hashes unavailable/, + ); + assert.equal(database.prepare("SELECT value FROM core_state WHERE key='last_event_index'").get()?.value, "0"); + assert.equal(database.prepare("SELECT COUNT(*) n FROM blocks").get()?.n, 1); + } finally { + globalThis.fetch = original; + database.close(); + } +}); diff --git a/apps/api/tests/replay-verbose-parity.test.ts b/apps/api/tests/replay-verbose-parity.test.ts index 9be481f..34180c3 100644 --- a/apps/api/tests/replay-verbose-parity.test.ts +++ b/apps/api/tests/replay-verbose-parity.test.ts @@ -374,10 +374,18 @@ async function replay(pages: FixtureEvent[][]): Promise { const parsed = new URL(String(input)); parsed.searchParams.delete("verbose"); const url = parsed.toString(); - if (url.endsWith("/blocks/101")) return Response.json({ result: { block_hash: "aa".repeat(32) } }); + if (url.endsWith("/blocks/101")) + return Response.json({ + result: { block_hash: "aa".repeat(32), ledger_hash: "b1".repeat(32), messages_hash: "b3".repeat(32) }, + }); if (url.endsWith("/events?limit=1")) return new Response(JSON.stringify({ result_count: tipIndex + 1 })); if (url.endsWith("/blocks/last")) return new Response(JSON.stringify({ result: { block_index: 102 } })); - if (url.endsWith("/blocks/102")) return new Response(JSON.stringify({ result: { block_hash: "ab".repeat(32) } })); + if (url.endsWith("/blocks/102")) + return new Response( + JSON.stringify({ + result: { block_hash: "ab".repeat(32), ledger_hash: "c1".repeat(32), messages_hash: "c3".repeat(32) }, + }), + ); const cursor = /\/events\?cursor=(\d+)&limit=(\d+)/.exec(url); if (cursor) { // fetchAsc requests cursor=from+chunk-1&limit=chunk; recover `from` and serve its page. diff --git a/docs/reorg-recovery.md b/docs/reorg-recovery.md index 63c4c5c..8218912 100644 --- a/docs/reorg-recovery.md +++ b/docs/reorg-recovery.md @@ -28,3 +28,23 @@ Tests exercise a retained URL cache, stale parent probes, catch-up verification, unknown hashes, exact partial-fill rollback, interrupted undo and missing history. This change does not certify historical imported state or all asynchronous derived projections; those retain their existing repair queues and maintenance behavior. + +Replay also compares stored `ledger_hash` and `messages_hash`, so Counterparty +reparses are detected even when the Bitcoin block hash stays unchanged. The last +completed block is verified when the cursor is inside a partially applied block. +Before writes, each replay slice saves its verified block identities in +`core_state.replay_protocol_identities`. They survive cursor commits and restarts, +covering partial blocks whose `BLOCK_PARSED` event has not yet been applied. + +A fixed parsed source-tip identity is captured before event retrieval and checked +again before writes. A newly mined descendant does not change that fixed anchor. +Missing protocol hashes pause replay; a changed anchor during fetching retries +without resetting application data. Confirmed mismatches use the existing exact +undo path and require a protocol-verified ancestor within retained history. + +No additional migration is needed. Blocks already store completed protocol +identities; the new state key records future slices without inventing hashes for +historical incomplete blocks. Administrative imports must establish a verified +baseline and clear obsolete slice identities while replay is paused. Each slice +adds bounded header reads and one small durable state write. Following mode +normally needs one slice; backlog processing costs more verification requests.