fix(portal): close silent-gap paths (backfill floor + stream realtime), rebased onto the invariant re-architecture

Re-applies three silent-gap fixes from the pre-rearchitecture branch onto the
new module layout (portal.ts monolith → functional core + shell). Each ships a
regression test; the full Portal suite stays green (138 tests).

- backfill floor (finding 3, portal-filters.ts): an undefined fromBlock is
  genesis (`filter.fromBlock ?? 0`), so if ANY source omits it the floor MUST be
  0. compileFetchSpec's Math.min over the DEFINED-only fromBlocks let a bounded
  source clamp every chunk fetch past an unbounded source's history, marking that
  prefix synced over a silent gap.

- reconnect on child discovery (finding 4, portal-realtime*.ts): stream mode
  filters /stream server-side, so a child discovered after a connection opens
  can't reach it. streamHotBlocks now snapshots a logsRevision at open and
  re-opens with the widened filter the moment it advances (bumped in the wire on
  each child); ndjsonLines cancels its reader on early break so the old
  connection closes.

- refuse unsupported stream mode (finding 5, portal-realtime-wire.ts): stream
  mode emits only log-bearing block events, so a non-log source (trace/transfer/
  transaction/block) or a receipt-requiring log source would be silently skipped
  while its intervals finalized as cached. assertStreamModeSupported() fails loud
  at startup instead.

Findings 6 (finalized-head probe) and 7 (unknown-parent gap) are intentionally
dropped: the re-architecture (#6) made the opposite, conservative choice on both
(passthrough / clear-window) as documented decisions — raise separately rather
than override here.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: mo4islona

portal/portal-client.ts

@@ -48,25 +48,33 @@ export async function* ndjsonLines(
   const reader = body.getReader();
   const dec = new TextDecoder();
   let buf = '';
-  for (;;) {
-    const { done, value } = await reader.read();
-    if (done) break;
-
-    if (value) {
-      onBytes?.(value.byteLength);
-      buf += dec.decode(value, { stream: true });
-    }
+  // A consumer may `break` early (the realtime stream re-opens the moment the logs filter revision bumps —
+  // finding 4); cancel the reader in `finally` so the underlying response body is closed rather than left
+  // locked+open. Fire-and-forget (NOT awaited): on a fully-drained stream it's a no-op, and awaiting the
+  // socket teardown on every historical fetch would add needless latency to the hot path.
+  try {
     for (;;) {
-      const nl = buf.indexOf('\n');
-      if (nl < 0) break;
+      const { done, value } = await reader.read();
+      if (done) break;
 
-      const line = buf.slice(0, nl);
-      buf = buf.slice(nl + 1);
-      if (line) yield line;
+      if (value) {
+        onBytes?.(value.byteLength);
+        buf += dec.decode(value, { stream: true });
+      }
+      for (;;) {
+        const nl = buf.indexOf('\n');
+        if (nl < 0) break;
+
+        const line = buf.slice(0, nl);
+        buf = buf.slice(nl + 1);
+        if (line) yield line;
+      }
     }
+    buf += dec.decode();
+    if (buf) yield buf;
+  } finally {
+    void reader.cancel().catch(() => {});
   }
-  buf += dec.decode();
-  if (buf) yield buf;
 }
 
 // Portal reports a missing COLUMN in a plural TABLE; map back to the field key we requested.

portal/portal-filters.test.ts

@@ -194,6 +194,26 @@ test('INV-1 support: compileFetchSpec is deterministic in structure and frozen',
   expect(a.logQuery()).toEqual(b.logQuery());
 });
 
+// ── finding 3: backfillStart floor never over-clamps an unbounded source ───────────────────────────
+
+test('finding 3: an unbounded source (fromBlock undefined) forces backfillStart 0, not Math.min over the bounded ones', () => {
+  // one bounded source at 15M, one unbounded (no fromBlock ⇒ genesis). The floor MUST be 0 — clamping to
+  // 15_000_000 would mark the unbounded source's whole [0, 15M) history synced over a silent gap.
+  const cbs = [
+    { filter: logFilter({ fromBlock: 15_000_000 }) },
+    { filter: logFilter({ fromBlock: undefined }) },
+  ];
+  expect(compileFetchSpec(cbs, new Map()).backfillStart).toBe(0);
+});
+
+test('finding 3: when EVERY source is bounded, backfillStart is the earliest fromBlock', () => {
+  const cbs = [
+    { filter: logFilter({ fromBlock: 15_000_000 }) },
+    { filter: logFilter({ fromBlock: 8_000_000 }) },
+  ];
+  expect(compileFetchSpec(cbs, new Map()).backfillStart).toBe(8_000_000);
+});
+
 test('INV-5 support: the trace request carries NO trace filter (fetch-all)', () => {
   const trace: any = {
     type: 'trace',

portal/portal-filters.ts

@@ -356,10 +356,16 @@ export function compileFetchSpec(
 
   // the chain's actual backfill window, from the filters — used to bound chunk fetches so a bounded
   // backfill (or the tail) never over-fetches. Fully automatic; no client tuning.
-  const froms = fs
-    .map((f) => f.fromBlock)
-    .filter((b): b is number => b != null);
-  const backfillStart = froms.length ? Math.min(...froms) : 0;
+  // A source with NO fromBlock starts at genesis (the runtime reads `filter.fromBlock ?? 0`); if ANY
+  // source omits it the floor MUST be 0. Taking Math.min over only the DEFINED fromBlocks would clamp
+  // every chunk fetch past the earliest DEFINED start and mark the [0, min) prefix synced over a silent
+  // gap — losing the whole history of the unbounded source. (finding 3 / C10)
+  const froms = fs.map((f) => f.fromBlock);
+  const definedFroms = froms.filter((b): b is number => b != null);
+  const backfillStart =
+    froms.some((b) => b == null) || definedFroms.length === 0
+      ? 0
+      : Math.min(...definedFroms);
   const tos = fs.map((f) => f.toBlock);
   const backfillEnd =
     tos.length && tos.every((t) => t != null)

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,33 @@ 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
 });
 
+// ─────────────────────────────── stream-mode capability gate (finding 5) ───────────────────────────────
+
+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', () => {

portal/portal-realtime-wire.ts

@@ -26,6 +26,7 @@ import type {
   EventCallback,
   Factory,
   FactoryId,
+  Filter,
   LightBlock,
   SyncLog,
 } from '@/internal/types.js';
@@ -224,6 +225,37 @@ export function toRealtimeSyncEvent(
   }
 }
 
+// ─────────────────────────────── stream-mode capability gate ───────────────────────────────
+
+/**
+ * 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.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).`,
+    );
+}
+
 // ─────────────────────────────── drop-in realtime generator ───────────────────────────────
 
 /**
@@ -243,6 +275,10 @@ 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 / receipt-
+  // requiring) — 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);
@@ -256,6 +292,9 @@ export async function* getPortalRealtimeEventGenerator(params: {
   // 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).
   const logs = buildPortalLogRequests(eventCallbacks, childAddresses);
+  // Bumps on every `logs` rebuild; streamHotBlocks watches it and re-opens the /stream the moment it
+  // advances so the widened server-side filter takes effect on the next block. (finding 4)
+  let logsRevision = 0;
 
   let childCount = 0;
   for (const [, m] of childAddresses) childCount += m.size;
@@ -278,6 +317,7 @@ export async function* getPortalRealtimeEventGenerator(params: {
       logs,
       blockFields: BLOCK_FIELDS,
       logFields: LOG_FIELDS,
+      getLogsRevision: () => logsRevision,
       finalizedHead: () =>
         portalFinalizedHead(portalUrl, headers, params.fetchImpl),
       finalizePollMs: params.finalizePollMs,
@@ -306,6 +346,7 @@ export async function* getPortalRealtimeEventGenerator(params: {
         const next = buildPortalLogRequests(eventCallbacks, childAddresses);
         logs.length = 0;
         logs.push(...next); // mutate in place — picked up on the next `/stream` reconnection
+        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,92 @@ 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('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();
+});

portal/portal-realtime.ts

@@ -97,6 +97,14 @@ export type PortalRealtimeArgs = {
   /** the block header fields ponder needs */
   blockFields?: Record<string, boolean>;
   logFields?: Record<string, boolean>;
+  /**
+   * Current logs-filter revision. The Portal filters `/stream` SERVER-side, so a `logs` change after a
+   * connection opens (a newly-discovered factory child) can't reach THAT connection. streamHotBlocks
+   * snapshots this at open and re-opens with the widened filter the moment it advances, so the child's
+   * logs are caught on the next block instead of only after an unrelated reconnect. Absent ⇒ constant 0
+   * (never reopens). (finding 4)
+   */
+  getLogsRevision?: () => number;
   signal?: AbortSignal;
   fetchImpl?: typeof fetch;
 };
@@ -155,18 +163,27 @@ export async function* streamHotBlocks(
       await sleep(1000, args.signal);
       continue;
     }
+    // Snapshot the filter revision at open; if it advances while streaming (a child was discovered), break
+    // to re-open with the widened server-side filter NOW rather than waiting for an unrelated reconnect.
+    // Breaking early cancels the ndjsonLines reader (its finally), closing this connection. (finding 4)
+    const openedRev = args.getLogsRevision?.() ?? 0;
+    let reopen = false;
     try {
       for await (const line of ndjsonLines(res.body)) {
         const batch = JSON.parse(line);
         if (batch?.header?.number != null) {
           cursor = batch.header.number + 1; // resume past this block on reconnect
           yield { header: batch.header, logs: batch.logs ?? [] };
+          if ((args.getLogsRevision?.() ?? 0) !== openedRev) {
+            reopen = true;
+            break;
+          }
         }
       }
     } catch {
       /* stream cut — reconnect from cursor */
     }
-    await sleep(200, args.signal);
+    if (!reopen) await sleep(200, args.signal); // filter changed ⇒ re-open immediately, no backoff
   }
 }