fix(portal): close silent-gap paths in backfill floor + stream realtime

Five correctness fixes, each a path where the fork could mark a block range
"synced" while serving no (or wrong-fork) data. Every fix ships a regression
test verified to fail without it.

Historical backfill:
- backfillStartBlock: an undefined fromBlock is genesis (the runtime reads
  `filter.fromBlock ?? 0`), so if ANY source omits it the floor is 0. The old
  Math.min-over-defined-fromBlocks let a bounded source (e.g. 15M) clamp every
  chunk fetch past the [0, 15M) history of an unbounded source, marking that
  prefix synced over a silent gap. (finding 3)

Portal-native stream realtime (PORTAL_REALTIME=stream, experimental):
- Refuse to start when a chain has non-log sources (trace/transfer/transaction/
  block) or receipt-requiring log sources: stream mode emits only log-bearing
  block events, so those events were silently skipped while their intervals were
  finalized as cached. Fail loud instead. (finding 5)
- A failed /finalized-head probe is now fatal at both seams (clampFinalized-
  ToPortalHead retries then throws; the historical seam throws on an unknown
  head) instead of passing the RPC finalized block through / returning [] —
  which left (portalHead, rpcFinalized] permanently skipped. (finding 6)
- Force a /stream reconnect the moment a factory child is discovered (a bumped
  logsRevision streamHotBlocks watches), so the widened server-side filter takes
  effect on the next block instead of only after an unrelated reconnect. Same-
  block child logs remain a documented limitation. (finding 4)
- An unknown-parent gap (a reorg deeper than the unfinalized window) is now
  fatal instead of silently clearing the window and continuing — which left the
  pre-fork blocks indexed on the wrong fork with no rollback event. (finding 7)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mo4islona

portal/portal-realtime-wire.test.ts

@@ -8,6 +8,7 @@ import type {
 } from "@/internal/types.js";
 import type { PortalRealtimeEvent } from "./portal-realtime.js";
 import {
+  assertStreamModeSupported,
   buildPortalLogRequests,
   clampFinalizedToPortalHead,
   discoverChildAddresses,
@@ -90,6 +91,31 @@ test("isPortalRealtime: only when a chain has a Portal source AND the flag is 's
   expect(isPortalRealtime({ portal: "http://p" })).toBe(false); // unset → A-path
 });
 
+test("assertStreamModeSupported: log-only sources are accepted", () => {
+  expect(() =>
+    assertStreamModeSupported([{ filter: logFilter() }], "mainnet"),
+  ).not.toThrow();
+});
+
+test("assertStreamModeSupported: a non-log source is refused — it would be silently skipped while marked synced (finding 5)", () => {
+  const trace = { type: "trace" } as any;
+  expect(() =>
+    assertStreamModeSupported(
+      [{ filter: logFilter() }, { filter: trace }],
+      "mainnet",
+    ),
+  ).toThrow(/only log sources, but this chain has trace/);
+});
+
+test("assertStreamModeSupported: a log source that needs transaction receipts is refused (finding 5)", () => {
+  expect(() =>
+    assertStreamModeSupported(
+      [{ filter: logFilter({ hasTransactionReceipt: true }) }],
+      "mainnet",
+    ),
+  ).toThrow(/transaction receipts/);
+});
+
 // ─────────────────────────────── light-block conversion ───────────────────────────────
 
 test("lightToLightBlock: Portal decimal number/timestamp → ponder hex, hashes passthrough", () => {
@@ -289,7 +315,7 @@ test("clampFinalizedToPortalHead: Portal at/ahead of RPC finalized → no clamp
   expect(out).toBe(finalized);
 });
 
-test("clampFinalizedToPortalHead: Portal head unknown (probe fails) → conservative passthrough", async () => {
+test("clampFinalizedToPortalHead: Portal head unknown in stream mode → FATAL (never silently passes the RPC finalized through) (finding 6)", async () => {
   process.env.PORTAL_REALTIME = "stream";
   const fetchImpl = (async () => {
     throw new Error("down");
@@ -300,13 +326,16 @@ test("clampFinalizedToPortalHead: Portal head unknown (probe fails) → conserva
     parentHash: "0xp",
     timestamp: "0x1",
   } as LightBlock;
-  const out = await clampFinalizedToPortalHead({
-    chain: { portal: "http://p", name: "c" } as any,
-    rpc: {} as any,
-    finalizedBlock: finalized,
-    fetchImpl,
-  });
-  expect(out).toBe(finalized);
+  // Old behavior passed `finalized` through — leaving historical targeting (portalHead, rpcFinalized] while
+  // realtime starts above it: a permanent silent gap. In stream mode a head we can't probe is fatal.
+  await expect(
+    clampFinalizedToPortalHead({
+      chain: { portal: "http://p", name: "c" } as any,
+      rpc: {} as any,
+      finalizedBlock: finalized,
+      fetchImpl,
+    }),
+  ).rejects.toThrow(/finalized-head probe failed/);
 });
 
 test("clampFinalizedToPortalHead: Portal head BELOW RPC finalized → refetch the block at the Portal head", async () => {

portal/portal-realtime-wire.ts

@@ -55,6 +55,37 @@ export const isPortalRealtime = (chain: {
 }): boolean =>
   typeof chain.portal === "string" && process.env.PORTAL_REALTIME === "stream";
 
+/**
+ * Stream mode only emits `block` events carrying LOGS — no transactions, receipts, or traces (see
+ * `toRealtimeSyncEvent`). So a non-log source (trace/transfer/transaction/block filter) would receive NO
+ * realtime events, yet ponder still finalizes its intervals as cached → a permanent, SILENT gap. Likewise
+ * a log source that requested transaction receipts (`hasTransactionReceipt`) can't be served. Refuse to
+ * start rather than corrupt. The historical Portal backfill supports every source type up to the finalized
+ * head, so this rejects only PORTAL_REALTIME=stream — not the chain. (Finding 5)
+ */
+export function assertStreamModeSupported(
+  eventCallbacks: { filter: Filter }[],
+  chainName: string,
+): void {
+  const filters = eventCallbacks.map((e) => e.filter);
+  const nonLog = [
+    ...new Set(filters.filter((f) => f.type !== "log").map((f) => f.type)),
+  ].sort();
+  if (nonLog.length > 0)
+    throw new Error(
+      `Portal ${chainName}: PORTAL_REALTIME=stream serves only log sources, but this chain has ${nonLog.join(", ")} source(s). Realtime would silently skip their events while marking the range synced. Use the default RPC realtime (unset PORTAL_REALTIME) or remove the non-log sources.`,
+    );
+
+  const needsReceipts = filters.some(
+    (f) =>
+      (f as { hasTransactionReceipt?: boolean }).hasTransactionReceipt === true,
+  );
+  if (needsReceipts)
+    throw new Error(
+      `Portal ${chainName}: PORTAL_REALTIME=stream cannot serve transaction receipts (a log source sets hasTransactionReceipt), so realtime would drop them. Use the default RPC realtime (unset PORTAL_REALTIME).`,
+    );
+}
+
 const portalHeaders = (): Record<string, string> => {
   const h: Record<string, string> = {
     "content-type": "application/json",
@@ -65,6 +96,8 @@ const portalHeaders = (): Record<string, string> => {
 };
 
 const cleanUrl = (portal: string): string => portal.replace(/\/$/, "");
+const sleep = (ms: number): Promise<void> =>
+  new Promise((r) => setTimeout(r, ms));
 
 // ─────────────────────────────── Portal finalized head + finality clamp ───────────────────────────────
 
@@ -102,15 +135,30 @@ export async function clampFinalizedToPortalHead(params: {
   if (isPortalRealtime(chain) === false) return finalizedBlock;
 
   const portalUrl = cleanUrl(chain.portal!);
-  const head = await portalFinalizedHead(
-    portalUrl,
-    portalHeaders(),
-    params.fetchImpl,
-  );
-  // Head unknown → stay conservative and keep the RPC finalized block (historical's own RPC finality-gap
-  // fallback remains the safety net). Portal at/ahead of RPC finalized → nothing to clamp.
-  if (head === undefined || head >= hexToNumber(finalizedBlock.number))
-    return finalizedBlock;
+  // The Portal finalized head IS the finality boundary in stream mode — load-bearing, so retry the cheap
+  // probe (transient blips are routine under multichain load) before deciding.
+  let head: number | undefined;
+  for (let attempt = 0; attempt < 3; attempt++) {
+    head = await portalFinalizedHead(
+      portalUrl,
+      portalHeaders(),
+      params.fetchImpl,
+    );
+    if (head !== undefined) break;
+
+    await sleep(200 * (attempt + 1));
+  }
+  // Head UNKNOWN after retries: FATAL in stream mode. The old behavior passed the RPC finalized block
+  // through, which leaves historical targeting (portalHead, rpcFinalized] — a range stream mode's
+  // historical seam serves nothing for — while realtime starts ABOVE it: a permanent silent gap. There is
+  // no safe conservative default here (the RPC fallback is suppressed in stream mode), so fail loud at
+  // startup rather than run with a hole. (Finding 6)
+  if (head === undefined)
+    throw new Error(
+      `Portal ${chain.name}: /finalized-head probe failed in stream mode (PORTAL_REALTIME=stream) — cannot establish the finality boundary. Check Portal connectivity for ${portalUrl}.`,
+    );
+  // Portal at/ahead of RPC finalized → nothing to clamp (never RAISE the boundary).
+  if (head >= hexToNumber(finalizedBlock.number)) return finalizedBlock;
 
   const clamped = (await eth_getBlockByNumber(rpc, [numberToHex(head), false], {
     retryNullBlockRequest: true,
@@ -309,6 +357,9 @@ export function toRealtimeSyncEvent(
         hasMatchedFilter: ev.hasMatchedFilter,
         block: ev.block,
         logs: ev.logs,
+        // LOG-ONLY: stream mode serves logs, not txs/receipts/traces, so `event.transaction` is undefined
+        // in realtime handlers (the historical Portal backfill DOES populate it) — a known gap.
+        // assertStreamModeSupported() refuses non-log sources and receipt-requiring log sources up front. (F5)
         transactions: [],
         transactionReceipts: [],
         traces: [],
@@ -345,6 +396,9 @@ export async function* getPortalRealtimeEventGenerator(params: {
 }) {
   const { common, chain, eventCallbacks, syncProgress, childAddresses } =
     params;
+  // Fail loud BEFORE streaming if this chain has sources stream mode can't serve (non-log / receipts) —
+  // else their events are silently skipped while their intervals are marked synced. (Finding 5)
+  assertStreamModeSupported(eventCallbacks, chain.name);
   const portalUrl = cleanUrl(chain.portal!);
   const headers = portalHeaders();
   const factories = uniqueFactories(eventCallbacks);
@@ -355,9 +409,11 @@ export async function* getPortalRealtimeEventGenerator(params: {
     ? hexToNumber(syncProgress.end.number)
     : undefined;
 
-  // Mutable: rebuilt (in place) whenever a new child is discovered so the next stream reconnection filters
-  // the new child's logs too (portal-realtime.ts re-reads this array when it re-opens the stream).
+  // Mutable: rebuilt (in place) whenever a new child is discovered so the stream filters the new child's
+  // logs too. `logsRevision` bumps on every rebuild; streamHotBlocks re-opens the /stream the moment it
+  // advances (the Portal filters server-side, so an open connection won't pick up the widened filter). (F4)
   const logs = buildPortalLogRequests(eventCallbacks, childAddresses);
+  let logsRevision = 0;
 
   let childCount = 0;
   for (const [, m] of childAddresses) childCount += m.size;
@@ -380,6 +436,7 @@ export async function* getPortalRealtimeEventGenerator(params: {
       logs,
       blockFields: BLOCK_FIELDS,
       logFields: LOG_FIELDS,
+      getLogsRevision: () => logsRevision,
       finalizedHead: () =>
         portalFinalizedHead(portalUrl, headers, params.fetchImpl),
       finalizePollMs: params.finalizePollMs,
@@ -407,7 +464,8 @@ export async function* getPortalRealtimeEventGenerator(params: {
       if (applyDiscovered(discovered, childAddresses, blockNumber)) {
         const next = buildPortalLogRequests(eventCallbacks, childAddresses);
         logs.length = 0;
-        logs.push(...next); // mutate in place — picked up on the next `/stream` reconnection
+        logs.push(...next); // mutate in place
+        logsRevision++; // force streamHotBlocks to re-open with the widened filter now (Finding 4)
       }
 
       yield { chain, event: toRealtimeSyncEvent(ev, discovered) };

portal/portal-realtime.test.ts

@@ -3,6 +3,7 @@ import {
   type Light,
   portalRealtimeEvents,
   reconcile,
+  streamHotBlocks,
   takeFinalized,
 } from "./portal-realtime.js";
 
@@ -162,3 +163,109 @@ test("portalRealtimeEvents: a re-streamed fork emits a reorg to the common ances
   expect(reorg.block.hash).toBe("a"); // common ancestor = block 10
   expect(reorg.reorgedBlocks.map((b: Light) => b.hash)).toEqual(["b"]);
 });
+
+test("portalRealtimeEvents: an unknown-parent gap is FATAL, not silently skipped (finding 7)", async () => {
+  const batches = [
+    {
+      header: { number: 10, hash: "a", parentHash: "z", timestamp: 10 },
+      logs: [],
+    },
+    // block 12's parent is unknown to our window ([10]) — a reorg deeper than the window (e.g. one that
+    // landed while disconnected, past the resume cursor). The old code silently cleared and continued.
+    {
+      header: { number: 12, hash: "c", parentHash: "unknown", timestamp: 12 },
+      logs: [],
+    },
+  ];
+  const ac = new AbortController();
+  const iter = portalRealtimeEvents({
+    portalUrl: "http://portal",
+    headers: {},
+    fromBlock: 10,
+    logs: [],
+    fetchImpl: mockFetch(batches, () => ac.abort()),
+    signal: ac.signal,
+    finalizedHead: async () => 0,
+    finalizePollMs: 999999,
+  });
+  const seen: any[] = [];
+  await expect(
+    (async () => {
+      for await (const e of iter) seen.push(e);
+    })(),
+  ).rejects.toThrow(/unknown parent/i);
+  // block 10 was delivered before the gap; the gap block was NOT swallowed into a silent resync
+  expect(seen.some((e) => e.type === "block" && e.block.number === "0xa")).toBe(
+    true,
+  );
+});
+
+test("streamHotBlocks: re-opens the /stream with the widened filter the moment the logs revision advances (finding 4)", async () => {
+  const enc = new TextEncoder();
+  const streamOf = (block: any, close: boolean) =>
+    new ReadableStream({
+      start(c) {
+        c.enqueue(enc.encode(JSON.stringify(block) + "\n"));
+        if (close) c.close(); // else leave open — only the revision change tears it down
+      },
+    });
+  const bodies: any[] = [];
+  let conn = 0;
+  const fetchImpl = (async (_url: string, init: any) => {
+    bodies.push(JSON.parse(init.body));
+    conn += 1;
+    if (conn === 1)
+      // connection 1: deliver block 100, then stay OPEN (no more data) so nothing but a rev change reopens it
+      return {
+        status: 200,
+        ok: true,
+        body: streamOf(
+          { header: { number: 100, hash: "h100", parentHash: "h99", timestamp: 1 }, logs: [] },
+          false,
+        ),
+      };
+    if (conn === 2)
+      // connection 2 (after the reopen): deliver block 101 and close
+      return {
+        status: 200,
+        ok: true,
+        body: streamOf(
+          { header: { number: 101, hash: "h101", parentHash: "h100", timestamp: 2 }, logs: [] },
+          true,
+        ),
+      };
+    return { status: 204, ok: false, body: null };
+  }) as any;
+
+  const logs: any[] = [{ address: ["0xfactory"], topic0: ["0xproxycreated"] }];
+  let rev = 0;
+  const ac = new AbortController();
+  const gen = streamHotBlocks({
+    portalUrl: "http://portal",
+    headers: {},
+    fromBlock: 100,
+    logs,
+    getLogsRevision: () => rev,
+    fetchImpl,
+    signal: ac.signal,
+  });
+
+  const first = await gen.next(); // block 100 from connection 1
+  expect(first.value?.header.number).toBe(100);
+  expect(bodies).toHaveLength(1);
+  expect(bodies[0].fromBlock).toBe(100);
+
+  // a newly-discovered factory child widens the filter and bumps the revision
+  logs.length = 0;
+  logs.push({ address: ["0xfactory", "0xnewchild"], topic0: ["0xproxycreated"] });
+  rev = 1;
+
+  const second = await gen.next(); // rev advanced → connection 1 torn down, reopen resumes from cursor 101
+  expect(second.value?.header.number).toBe(101);
+  expect(bodies).toHaveLength(2);
+  expect(bodies[1].fromBlock).toBe(101); // resumed PAST block 100 (no re-delivery / spurious reorg)
+  expect(bodies[1].logs[0].address).toContain("0xnewchild"); // reopened with the widened filter
+
+  await gen.return(undefined); // stop the generator
+  ac.abort();
+});