Fix cached reorg misses and restore exact replay state

apps/api/src/indexer/events/context.ts

@@ -6,7 +6,7 @@
  * one file per message type, each self-contained, routed by dispatch.ts.
  */
 // A statement is a thunk so it can be bound against the live DB inside the batcher.
-export type Stmt = (db: D1Database) => D1PreparedStatement;
+export type Stmt = ((db: D1Database) => D1PreparedStatement) & { blockIndex?: number };
 // One Counterparty event from the event stream.
 export interface Ev {
   event_index: number;

apps/api/src/indexer/events/dispatch.ts

@@ -45,8 +45,10 @@ export function dispatch(ev: Ev, ctx: Ctx): void {
   // The chronological stream states block_time once per block (NEW_BLOCK); every later event in the
   // block inherits it. Divisibility comes from the chunk's local map, never from verbose enrichment.
   if (ev.event === "NEW_BLOCK" && p.block_time != null) ctx.blockTime = p.block_time;
+  const start = ctx.stmts.length;
   const h = HANDLERS[ev.event];
   if (h) h({ ev, p, b, bt: p.block_time ?? ctx.blockTime, div: assetDivisible(ctx, p.asset) }, ctx);
+  for (const stmt of ctx.stmts.slice(start)) stmt.blockIndex = b;
 }
 // Every event name we have a handler for — used to build the stream fetch filter (single source of truth).
 export const HANDLED_EVENTS = Object.keys(HANDLERS);

apps/api/src/indexer/reorg-undo.ts

@@ -0,0 +1,126 @@
+/** Exact, bounded before-images. A batch's context and writes commit together;
+ * concurrent readers/writers never observe an active journal context. */
+export async function initializeUndo(db: D1Database, height: number): Promise<void> {
+  await db.prepare("INSERT OR IGNORE INTO reorg_undo_state(singleton,floor) VALUES(1,?)").bind(height).run();
+}
+
+export async function pruneUndo(db: D1Database, height: number): Promise<void> {
+  const floor = Math.max(0, height - 24);
+  await db.batch([
+    db.prepare("DELETE FROM reorg_undo WHERE block_index<=?").bind(floor),
+    db.prepare("UPDATE reorg_undo_state SET floor=max(floor,?) WHERE singleton=1").bind(floor),
+  ]);
+}
+
+export async function writeWithUndo(
+  db: D1Database,
+  writes: { statement: D1PreparedStatement; block?: number }[],
+): Promise<void> {
+  // At most two statements per write plus context cleanup, below D1's batch bound.
+  for (let offset = 0; offset < writes.length; offset += 40) {
+    const statements: D1PreparedStatement[] = [db.prepare("DELETE FROM reorg_undo_context")];
+    let active: number | undefined;
+    for (const write of writes.slice(offset, offset + 40)) {
+      if (write.block !== active) {
+        statements.push(
+          write.block === undefined
+            ? db.prepare("DELETE FROM reorg_undo_context")
+            : db
+                .prepare(
+                  "INSERT INTO reorg_undo_context(singleton,block_index) VALUES(1,?) ON CONFLICT(singleton) DO UPDATE SET block_index=excluded.block_index",
+                )
+                .bind(write.block),
+        );
+        active = write.block;
+      }
+      statements.push(write.statement);
+    }
+    statements.push(db.prepare("DELETE FROM reorg_undo_context"));
+    await db.batch(statements);
+  }
+}
+
+const TABLES = new Set([
+  "assets",
+  "orders",
+  "order_matches",
+  "dispensers",
+  "fairminters",
+  "pools",
+  "bets",
+  "bet_matches",
+  "rps",
+  "rps_matches",
+]);
+interface Column {
+  name: string;
+  type: string;
+  pk: number;
+}
+interface Undo {
+  seq: number;
+  table_name: string;
+  row_key: string;
+  before_row: string | null;
+}
+type StoredRow = Record<string, string | number | null>;
+
+/** The sync entry point initializes the floor before checking for reorgs,
+ * so pre-deployment history fails closed.
+ * Each inverse and deletion of its journal row share one transaction, making
+ * interrupted rollback resumable without double undo. */
+export async function restoreUndo(db: D1Database, height: number): Promise<boolean> {
+  const state = await db.prepare("SELECT floor FROM reorg_undo_state WHERE singleton=1").first<{ floor: number }>();
+  if (!state) throw new Error("Undo history is not initialized; rebuild required");
+  if (height < state.floor) throw new Error("Reorg predates exact undo history; rebuild required");
+  const schemas = new Map<string, Column[]>();
+  for (;;) {
+    const rows = await db
+      .prepare(
+        "SELECT seq,table_name,row_key,before_row FROM reorg_undo WHERE block_index>? ORDER BY seq DESC LIMIT 40",
+      )
+      .bind(height)
+      .all<Undo>();
+    if (!rows.results.length) return true;
+    const statements = [db.prepare("DELETE FROM reorg_undo_context")];
+    for (const row of rows.results) {
+      if (!TABLES.has(row.table_name)) throw new Error("Unknown undo table");
+      let columns = schemas.get(row.table_name);
+      if (!columns) {
+        columns = (await db.prepare(`PRAGMA table_info(${row.table_name})`).all<Column>()).results;
+        schemas.set(row.table_name, columns);
+      }
+      const keys = columns.filter((column) => column.pk).sort((a, b) => a.pk - b.pk);
+      const decode = (column: Column, value: string | number | null) => {
+        if (column.type.toUpperCase() !== "BLOB" || value === null) return value;
+        if (typeof value !== "string" || !/^(?:[0-9a-f]{2})*$/i.test(value)) throw new Error("Invalid undo blob");
+        return Uint8Array.from(value.match(/../g) ?? [], (byte) => parseInt(byte, 16));
+      };
+      if (row.before_row === null) {
+        const key = JSON.parse(row.row_key) as StoredRow;
+        statements.push(
+          db
+            .prepare(
+              `DELETE FROM ${row.table_name} WHERE ${keys.map((column) => `"${column.name}" IS ?`).join(" AND ")}`,
+            )
+            .bind(...keys.map((column) => decode(column, key[column.name]))),
+        );
+      } else {
+        const before = JSON.parse(row.before_row) as StoredRow;
+        const retained = columns.filter((column) => Object.hasOwn(before, column.name));
+        const names = retained.map((column) => `"${column.name}"`);
+        const mutable = retained.filter((column) => !column.pk);
+        statements.push(
+          db
+            .prepare(
+              `INSERT INTO ${row.table_name}(${names.join(",")}) VALUES(${names.map(() => "?").join(",")})
+          ON CONFLICT(${keys.map((column) => `"${column.name}"`).join(",")}) DO UPDATE SET ${mutable.map((column) => `"${column.name}"=excluded."${column.name}"`).join(",")}`,
+            )
+            .bind(...retained.map((column) => decode(column, before[column.name]))),
+        );
+      }
+      statements.push(db.prepare("DELETE FROM reorg_undo WHERE seq=?").bind(row.seq));
+    }
+    await db.batch(statements);
+  }
+}

apps/api/src/indexer/sync.ts

@@ -1,3 +1,4 @@
+import { initializeUndo, pruneUndo, restoreUndo, writeWithUndo } from "#api/indexer/reorg-undo";
 /**
  * Counterparty mirror engine — chronological event replay into normalized D1.
  *
@@ -64,7 +65,10 @@ function setCoreStateStmt(db: D1Database, key: string, value: string): D1Prepare
 // without advancing the cursor; the chunk re-runs and idempotency re-applies it exactly. Never drops.
 async function batchAll(db: D1Database, stmts: Stmt[]): Promise<void> {
   for (let i = 0; i < stmts.length; i += DB_BATCH) {
-    await db.batch(stmts.slice(i, i + DB_BATCH).map((f) => f(db)));
+    await writeWithUndo(
+      db,
+      stmts.slice(i, i + DB_BATCH).map((f) => ({ statement: f(db), block: f.blockIndex })),
+    );
   }
 }
 
@@ -208,6 +212,7 @@ export async function syncCoreEvents(
     let lastBlock = Number.parseInt(blockValue ?? "0", 10);
     if (!Number.isSafeInteger(lastIndex) || lastIndex < -1) throw new Error("replay cursor is invalid");
 
+    await initializeUndo(env.CORE_DB, lastBlock);
     const rollbackValue = await getCoreState(env.CORE_DB, "rollback_to");
     if (rollbackValue !== null) {
       const rollbackTo = Number(rollbackValue);
@@ -251,8 +256,7 @@ 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();
-    const followingWindow = tip - lastIndex < 5 * CHUNK;
-    if (followingWindow && lastBlock > 0) {
+    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.
       const rawTip = await env.CORE_DB.prepare(
@@ -264,9 +268,12 @@ export async function syncCoreEvents(
           .bind(verifyBlock)
           .first<{ hash: string }>()
           .then((row) => row?.hash || null),
-        blockHash(env.COUNTERPARTY_API_BASE, verifyBlock),
+        sourceTipBlock < verifyBlock
+          ? Promise.resolve("past-source-tip")
+          : blockHash(env.COUNTERPARTY_API_BASE, verifyBlock),
       ]);
-      if (storedHash && actualHash && storedHash !== actualHash) {
+      if (!storedHash || !actualHash) throw new Error("Cannot verify applied block hash");
+      if (storedHash !== actualHash) {
         let rollbackTo = verifyBlock - 1;
         let found = false;
         for (; rollbackTo >= Math.max(0, verifyBlock - 24); rollbackTo--) {
@@ -330,7 +337,43 @@ export async function syncCoreEvents(
           assetDivisibility: await chunkAssetDivisibility(env.CORE_DB, slice),
           blockTime,
         };
-        for (const event of slice) dispatch(event, ctx);
+        // Validate the actual event header, not just a separately cached height lookup.
+        let previousHeight = lastBlock;
+        let previousHash =
+          lastBlock > 0
+            ? 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)
+            : undefined;
+        for (const event of slice) {
+          if (event.event === "NEW_BLOCK") {
+            const height = event.block_index;
+            const hash = event.params.block_hash;
+            if (
+              typeof hash !== "string" ||
+              !/^[0-9a-f]{64}$/i.test(hash) ||
+              hash !== (await blockHash(env.COUNTERPARTY_API_BASE, height))
+            )
+              throw new Error(`Unverified event block ${height}`);
+            if (previousHash && (height !== previousHeight + 1 || event.params.previous_block_hash !== previousHash))
+              throw new Error(`Event block ${height} does not extend the applied chain`);
+            previousHeight = height;
+            previousHash = hash;
+          }
+          dispatch(event, ctx);
+        }
+        // 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 &&
+          (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))
+        )
+          throw new Error("Applied chain changed while fetching events");
         await batchAll(env.CORE_DB, [...dictionaryStatements(ctx.identities), ...ctx.stmts]);
         const dispenseTxs = slice
           .filter((event) => event.event === "DISPENSE")
@@ -353,6 +396,7 @@ export async function syncCoreEvents(
       }
     }
 
+    await pruneUndo(env.CORE_DB, lastBlock);
     const caughtUp = lastIndex >= tip;
     if (caughtUp) {
       // Gated, not every tick. The prune retains the newest snapshot per
@@ -442,6 +486,7 @@ async function deleteAbove(db: D1Database, table: string, block: number): Promis
 
 /** Reverse committed orphan balance deltas before deleting their provenance. */
 export async function rollbackCoreDatabase(db: D1Database, rollbackTo: number, rollbackIndex: number): Promise<void> {
+  await restoreUndo(db, rollbackTo);
   const restores = await balanceRollbackStatements(db, rollbackTo, rollbackIndex);
   for (let offset = 0; offset < restores.length; offset += DB_BATCH) {
     await db.batch(restores.slice(offset, offset + DB_BATCH));
@@ -483,12 +528,6 @@ export async function rollbackCoreDatabase(db: D1Database, rollbackTo: number, r
        AND NOT EXISTS (SELECT 1 FROM trades WHERE trades.venue=trade_legs.venue AND trades.ref=trade_legs.trade_ref)`,
     )
     .run();
-  await db
-    .prepare(`UPDATE orders SET status='open',closed_block_index=NULL WHERE closed_block_index>?`)
-    .bind(rollbackTo)
-    .run();
-  await db.prepare(`UPDATE dispensers SET closed_block_index=NULL WHERE closed_block_index>?`).bind(rollbackTo).run();
-
   await db.prepare("DELETE FROM balance_snapshots WHERE block_index>?").bind(rollbackTo).run();
   // Orphan ledger stays until all restores commit, so a crashed rollback can retry.
   await deleteAbove(db, "ledger_events", rollbackTo);

apps/api/src/integrations/counterparty.ts

@@ -1,3 +1,4 @@
+import { freshCounterpartyUrl } from "#api/integrations/fresh-read";
 /** Counterparty HTTP client — fetch + decode (precision-safe) with good-citizen backoff. */
 import { parseCounterpartyJson } from "#api/indexer/codec";
 import { discard } from "#api/lib/net";
@@ -42,7 +43,9 @@ export async function counterpartyJson<T = unknown>(
   for (let attempt = 0; ; attempt++) {
     const remaining = deadline - Date.now();
     if (remaining <= 0) throw new Error(`Counterparty ${path} request deadline exceeded`);
-    const response = await fetch(`${api}${path}`, { signal: AbortSignal.timeout(Math.min(timeoutMs, remaining)) });
+    const response = await fetch(freshCounterpartyUrl(`${api}${path}`), {
+      signal: AbortSignal.timeout(Math.min(timeoutMs, remaining)),
+    });
     if (response.ok) {
       try {
         return parseCounterpartyResponse<T>(await response.text());

apps/api/src/integrations/fresh-read.ts

@@ -0,0 +1,11 @@
+/** Core caches by URL across rollbacks. Its scalar arguments and verbose
+ * enrichment use the first value; the second value only changes the cache key.
+ * Keep pagination, limits and enrichment identical on every attempt. */
+export function freshCounterpartyUrl(input: string): string {
+  const url = new URL(input);
+  const verbose = url.searchParams.get("verbose") ?? "false";
+  url.searchParams.delete("verbose");
+  url.searchParams.append("verbose", verbose);
+  url.searchParams.append("verbose", crypto.randomUUID());
+  return url.toString();
+}