subsquid-labs
diff --git a/‎portal/INVARIANTS.md‎
Lines changed: 7 additions & 1 deletion b/‎portal/INVARIANTS.md‎
Lines changed: 7 additions & 1 deletion
diff --git a/‎portal/portal-realtime-wire.test.ts‎
Lines changed: 144 additions & 0 deletions b/‎portal/portal-realtime-wire.test.ts‎
Lines changed: 144 additions & 0 deletions
@@ -86,4 +86,10 @@ Five fixes from a four-lens review (client/transport, historical data path, real
 - **Monotonic head cache + stream-mode stale-head fatal → INV-9.** `refreshPortalHead`/`ensureChunkSize` adopted ANY probe result, but load-balanced replicas answer independently — a later probe can return a LOWER head, regressing the cache. In stream mode a regressed (stale-LOW) head made an interval at/below the true head read as "past the head", and that branch returned `[]` — marked synced while realtime streams only above the boundary: the G4/C11 silent gap, reopened. Two changes: the cached head only ever rises (`max`; every observation is ≤ the true head, so the cached max is a safe UPPER bound that never exceeds the true finalized head), and the stream-mode known-head-below-interval branch is now FATAL (it never legitimately fires — the clamp bounds historical at the boundary head — so reaching it means replica lag persisted through all retries: the same "boundary cannot be located" condition as an unknown head). Mutation-verified: reverting the `max` fails `portal-shell.test.ts` "INV-9: the cached Portal head is MONOTONIC…"; the `[]`→throw change is pinned by "INV-9: stream-realtime mode with a KNOWN head below the interval is FATAL…". **Residual (tracked, #47):** the `max`-keep guarantees only that the cached head never OVER-states finality; it does not verify range COMPLETENESS against the specific replica serving a given `/finalized-stream` POST — a replica lagging the cached head can 204 a tail that `dataChunk` then records as `coveredTo` (a phantom-coverage hole pre-existing on main — PR #15 FIX 1 documents the 204/truncate shape — whose exposure `max`-keeping extends from "until the next lower probe" to the process lifetime). The honest claim is: cached max never exceeds the true finalized head; completeness against the serving replica is not yet verified. Fix is a client-side range-completeness guard (#47), deferred pending one empirical check of Portal 204 semantics.
 - **Schema-degradation bound keyed per-FIELD, not per-column → INV-13 (availability).** `stream()`'s "dropping its field didn't help → real error" bound keyed on `err.tag` — the BARE column name — so a dataset missing the same column in TWO tables (`logs_bloom` in blocks AND transactions: the Monad-style shape FIX 3 exists for; also `nonce`, `gas_used`) dropped the first, then hard-threw on the second table's 400 → chunk rejected → G1-evicted → refetched → identical failure: a deterministic crash-loop. Now keyed by the table-qualified `err.fieldKey`; a RE-thrown fieldKey still means its own drop didn't fix its own 400, so the bound survives. Mutation-verified: reverting the keying fails `portal-client.test.ts` "field degradation: the SAME column missing in TWO tables drops each independently…".
 - **Stream-mode droppable BLOCK fields → INV-10 (availability).** See the B3 completion note above. Mutation-verified: restoring the `tableKey === 'transaction'` restriction fails `portal-realtime.test.ts` "streamHotBlocks: a DROPPABLE BLOCK-field 400 (dataset lacks mix_hash) degrades too…".
-- **Bounded realtime `/finalized-head` probe → INV-10 (availability).** The wire's `portalFinalizedHead` was a bare `fetch().then(r => r.json())` — no timeout, no abort signal, no body bound — while its historical twin had been explicitly hardened against exactly this (issue #14 / PR #16). One black-holed connection froze finalize emission mid-run (`portalRealtimeEvents` awaits the probe inline in the block loop) and startup (`clampFinalizedToPortalHead`'s 3-attempt retry never reached attempt 2). The probe implementation is now SHARED: `probeFinalizedHead` in portal-client.ts (two-phase connect-abort + own-the-lock body read, hash-carrying) backs both the client's `finalizedHead()` and the wire's `portalFinalizedHead()` — deleting the duplication that let the weak copy exist. Mutation-verified: restoring the bare fetch fails `portal-realtime-wire.test.ts` "portalFinalizedHead: a HUNG probe is BOUNDED…" (times out instead of resolving `undefined`).
+- **Bounded realtime `/finalized-head` probe → INV-10 (availability).** The wire's `portalFinalizedHead` was a bare `fetch().then(r => r.json())` — no timeout, no abort signal, no body bound — while its historical twin had been explicitly hardened against exactly this (issue #14 / PR #16). One black-holed connection froze finalize emission mid-run (`portalRealtimeEvents` awaits the probe inline in the block loop) and startup (`clampFinalizedToPortalHead`'s 3-attempt retry never reached attempt 2). The probe implementation is now SHARED: `probeFinalizedHead` in portal-client.ts (two-phase connect-abort + own-the-lock body read, hash-carrying) backs both the client's `finalizedHead()` and the wire's `portalFinalizedHead()` — deleting the duplication that let the weak copy exist. Mutation-verified: restoring the bare fetch fails `portal-realtime-wire.test.ts` "portalFinalizedHead: a HUNG probe is BOUNDED…" (times out instead of resolving `undefined`).
+
+### Persisted-finality floor (stream-mode restart regression)
+
+The wave-4 monotonic head cache and the wire's `emitFinalize` gate are both IN-MEMORY: neither survives a restart. In stream mode `clampFinalizedToPortalHead` adopts the startup `/finalized-head` probe as the finality boundary, and load-balanced replicas answer independently — so a restart can land on a replica whose head Y sits BELOW finality ponder already PERSISTED (safe checkpoint X, written by `finalizeOmnichain`/`finalizeMultichain` from an earlier finalize). Ponder's own migrate-time guard ("Finalized block for chain … cannot move backwards") checks only the RPC-derived finalized block, which is fetched BEFORE the clamp — the regressed Portal head bypasses it entirely. The consequence is not a rollback (finalized history never forks) but DOUBLE-INDEXING: crash recovery reverts user tables only ABOVE X (rows at/below X lost their reorg-table rows at finalize, by design), historical yields nothing (recovery checkpoint X > boundary Y), and realtime re-streams `(Y, X]` — ponder's realtime event loop has no filter against the recovery checkpoint, so every event in the range executes AGAIN on committed rows. It compounds: the first finalize in `(Y, X]` passes `emitFinalize` (its floor was seeded from the regressed Y) and `finalizeOmnichain` writes the checkpoint VERBATIM, regressing the persisted safe checkpoint itself. The same regression exists mid-run at the backfill-cutover refetch in the omnichain/multichain orderings: `shouldCatchup` is computed per chain but the new boundaries are assigned to ALL chains, so one chain's advance adopts another chain's stale-LOW probe (the isolated ordering is immune — its per-chain `break` refuses a non-advancing boundary).
+
+The fix is the same shape as pipes-sdk's `FinalizedWatermark` (PR #88 there): a monotonic floor SEEDED FROM PERSISTED STATE. `clampFinalizedToPortalHead` takes a `floor` (block number) the boundary never goes below — a head below it is clamped UP to the floor (safe: `(head, floor]` is finalized history, identical across replicas; the lagging replica just hasn't re-marked it final) with a WARN log. The wiring passes ponder's persisted per-chain safe checkpoint (`crashRecoveryCheckpoint`, decoded via `checkpointBlockNumber`) at the startup site (`getLocalSyncProgress`) and the previously adopted boundary at the two cutover sites. `checkpointBlockNumber` is chainId-GUARDED (review, PR #55): in the omnichain ordering `finalizeOmnichain` writes the OMNICHAIN checkpoint verbatim to every chain's row (no per-chain where clause — "it is not an invariant that `chainId` and `checkpoint.chainId` are the same"), so a row's checkpoint can encode ANOTHER chain's block height; used as a floor it would either sit above the local RPC finalized block and disable the Portal-head clamp outright (the boundary lands above the Portal head — the wave-4 stream-mode stale-head FATAL, a deterministic crash-loop) or sit below the local safe point and silently under-protect. A checkpoint whose encoded chainId differs from the chain's yields NO floor (the pre-floor behavior; the multichain/isolated orderings write per-chain checkpoints and keep full protection) — mirroring upstream's own crash-recovery chainId guard in `runtime/historical.ts`, pinned by "a FOREIGN-chain checkpoint yields NO floor…". The floor also overrides a `PORTAL_FINALIZED_HEAD` pin below it — a pin below persisted finality would re-open the double-indexing hole, so correctness wins over the pin. Historical intervals at/below the floor are already cached in the sync store (they were finalized), so the floored boundary triggers no Portal data request a lagging replica would 204. Pinned by `portal-realtime-wire.test.ts` "a probed head BELOW the persisted floor is clamped UP to the floor…", "…floor at/above the RPC finalized block returns it unchanged…", "…the floor overrides a PORTAL_FINALIZED_HEAD pin below it…", and the inert-floor case "…a floor at/below the probed head is inert…".
@@ -6,10 +6,12 @@ import type {
   LightBlock,
   LogFilter,
 } from '@/internal/types.js';
+import { encodeCheckpoint, ZERO_CHECKPOINT } from '@/utils/checkpoint.js';
 import type { PortalRealtimeEvent } from './portal-realtime.js';
 import {
   assertStreamModeSupported,
   buildPortalLogRequests,
+  checkpointBlockNumber,
   clampFinalizedToPortalHead,
   discoverChildAddresses,
   getPortalRealtimeEventGenerator,
@@ -479,6 +481,148 @@ test('clampFinalizedToPortalHead: an explicit PORTAL_FINALIZED_HEAD pin below th
   expect(hexToNumber(out.number)).toBe(900);
 });
 
+// ─────────────────────────────── persisted-finality floor ───────────────────────────────
+
+// Shared scaffolding for the floor tests: RPC finalized at 1000, an rpc mock that records the block
+// request, and a floor passed in by the caller (as the wiring does from ponder's persisted checkpoint).
+const rpcFinalized1000 = {
+  number: '0x3e8',
+  hash: '0xh',
+  parentHash: '0xp',
+  timestamp: '0x1',
+} as LightBlock;
+const recordingRpc = (blockNumberHex: string) => {
+  const calls: any[] = [];
+  const rpc = {
+    request: async (req: any) => {
+      calls.push(req);
+      return {
+        number: blockNumberHex,
+        hash: '0xfloorblock',
+        parentHash: '0xpp',
+        timestamp: '0x2',
+        logsBloom: `0x${'0'.repeat(512)}`,
+        sha3Uncles: '0x0',
+        miner: '0x0',
+        stateRoot: '0x0',
+        transactionsRoot: '0x0',
+        receiptsRoot: '0x0',
+        gasUsed: '0x0',
+        gasLimit: '0x0',
+        extraData: '0x',
+        nonce: '0x0',
+        mixHash: '0x0',
+        difficulty: '0x0',
+        size: '0x0',
+        transactions: [],
+      };
+    },
+  } as any;
+
+  return { rpc, calls };
+};
+
+test('clampFinalizedToPortalHead: a probed head BELOW the persisted floor is clamped UP to the floor — a restart against a lagging replica must not re-stream already-finalized (unrevertable) blocks', async () => {
+  process.env.PORTAL_REALTIME = 'stream';
+  // Last run persisted finality at 950; this restart's probe hits a lagging replica reporting 900.
+  // Without the floor, realtime would stream from 901 and re-index (900, 950] — rows crash recovery
+  // cannot revert (their reorg-table rows were deleted at finalize).
+  const fetchImpl = (async () => ({
+    json: async () => ({ number: 900 }),
+  })) as any;
+  const { rpc, calls } = recordingRpc('0x3b6'); // 950
+  const out = await clampFinalizedToPortalHead({
+    chain: { portal: 'http://p', name: 'c' } as any,
+    rpc,
+    finalizedBlock: rpcFinalized1000,
+    floor: 950,
+    fetchImpl,
+  });
+  expect(calls[0]!.params[0]).toBe('0x3b6'); // the boundary block is fetched at the FLOOR (950), not 900
+  expect(hexToNumber(out.number)).toBe(950);
+});
+
+test('clampFinalizedToPortalHead: a floor at/below the probed head is inert — the head is adopted exactly as before', async () => {
+  process.env.PORTAL_REALTIME = 'stream';
+  const fetchImpl = (async () => ({
+    json: async () => ({ number: 900 }),
+  })) as any;
+  const { rpc, calls } = recordingRpc('0x384'); // 900
+  const out = await clampFinalizedToPortalHead({
+    chain: { portal: 'http://p', name: 'c' } as any,
+    rpc,
+    finalizedBlock: rpcFinalized1000,
+    floor: 900, // == head → no effect
+    fetchImpl,
+  });
+  expect(calls[0]!.params[0]).toBe('0x384'); // clamped to the head, same as the floorless behavior
+  expect(hexToNumber(out.number)).toBe(900);
+});
+
+test('clampFinalizedToPortalHead: a floor at/above the RPC finalized block returns it unchanged — the floor never RAISES the boundary past ponder’s own finality', async () => {
+  process.env.PORTAL_REALTIME = 'stream';
+  const fetchImpl = (async () => ({
+    json: async () => ({ number: 900 }),
+  })) as any;
+  const { rpc, calls } = recordingRpc('0x0');
+  const out = await clampFinalizedToPortalHead({
+    chain: { portal: 'http://p', name: 'c' } as any,
+    rpc,
+    finalizedBlock: rpcFinalized1000,
+    floor: 1_000, // floor == RPC finalized → pass-through, no block refetch
+    fetchImpl,
+  });
+  expect(out).toBe(rpcFinalized1000);
+  expect(calls.length).toBe(0);
+});
+
+test('clampFinalizedToPortalHead: the floor overrides a PORTAL_FINALIZED_HEAD pin below it — a pin below persisted finality must not re-open the double-indexing hole', async () => {
+  process.env.PORTAL_REALTIME = 'stream';
+  process.env.PORTAL_FINALIZED_HEAD = '900'; // operator pin BELOW the persisted floor 950
+  let probed = false;
+  const fetchImpl = (async () => {
+    probed = true;
+    return { json: async () => ({ number: 5000 }) };
+  }) as any;
+  const { rpc, calls } = recordingRpc('0x3b6'); // 950
+  const out = await clampFinalizedToPortalHead({
+    chain: { portal: 'http://p', name: 'c' } as any,
+    rpc,
+    finalizedBlock: rpcFinalized1000,
+    floor: 950,
+    fetchImpl,
+  });
+  expect(probed).toBe(false); // the pin still suppresses the live probe
+  expect(calls[0]!.params[0]).toBe('0x3b6'); // but the boundary is the FLOOR, not the pin
+  expect(hexToNumber(out.number)).toBe(950);
+});
+
+test('checkpointBlockNumber: decodes the block number out of a persisted checkpoint string; undefined stays undefined', () => {
+  const checkpoint = encodeCheckpoint({
+    ...ZERO_CHECKPOINT,
+    blockTimestamp: 1_700_000_000n,
+    chainId: 1n,
+    blockNumber: 12_345n,
+  });
+  expect(checkpointBlockNumber(checkpoint, 1)).toBe(12_345);
+  expect(checkpointBlockNumber(undefined, 1)).toBeUndefined();
+});
+
+test("checkpointBlockNumber: a FOREIGN-chain checkpoint yields NO floor — omnichain finalize writes the omnichain checkpoint verbatim to every chain's row, so its block height is another chain's", () => {
+  // finalizeOmnichain updates PONDER_CHECKPOINT with no per-chain where clause: chain 1's row can carry
+  // a checkpoint encoding chain 8453's block height. Used as chain 1's floor it either disables the
+  // Portal-head clamp (foreign height above the local RPC finalized → the stream-mode stale-head FATAL,
+  // a crash-loop) or silently under-protects (foreign height below the local safe point).
+  const foreign = encodeCheckpoint({
+    ...ZERO_CHECKPOINT,
+    blockTimestamp: 1_700_000_000n,
+    chainId: 8_453n,
+    blockNumber: 99_999_999n, // a Base-scale height, nonsense on an Ethereum-scale chain
+  });
+  expect(checkpointBlockNumber(foreign, 1)).toBeUndefined();
+  expect(checkpointBlockNumber(foreign, 8_453)).toBe(99_999_999); // same chain → still a valid floor
+});
+
 // ─────────────────────────────── end-to-end generator ───────────────────────────────
 
 // mock the Portal /stream — one NDJSON connection per entry in `connections` (then 204) — and