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

.gitignore

@@ -8,3 +8,4 @@ examples/*/package-lock.json
 examples/*/ponder-env.d.ts
 examples/*/.ponder/
 examples/*/generated/
+.idea/

portal/INVARIANTS.md

@@ -24,8 +24,8 @@ Legacy `C#` tags from the pre-refactor comments are mapped where they existed.
 | **INV-6** | **RPC-shape byte-identity of transforms.** Transform outputs are byte-equal to the RPC path's encodings (hex quantities, casing, null/undefined, accessList normalization, status mapping). | The sync store must be byte-identical whether backfilled from Portal or RPC. | Pure `portal-transform.ts` functions calibrated against captured fixtures. | — (external differential gate) | `portal-transform.test.ts` (golden fixtures + `hx` totality); external `harness/diff`. |
 | **INV-7** | **Bounded buffered rows.** Row-count is conservative — rows are registered as NDJSON batches ARRIVE (fixes **G3**), freed exactly once on eviction/rejection; `rows ≥ 0` always; read-ahead beyond depth-1 admits only when not saturated. | In-flight NDJSON arrays were invisible to backpressure; the memory cap must engage before the heap dies. | `portal-client.ts` `onRows` hook per batch; `portal.ts` keys accounting to a per-fetch `RowToken` freed exactly once (idempotent — a stale evicted fetch can neither register nor free against its replacement); the gate reducer clamps `rows ≥ 0`. | `gateReduce` asserts `rows ≥ 0`. | `portal-gate.test.ts` — "INV-8/INV-7: bounds hold under arbitrary event sequences"; `portal-client.test.ts` — "G3: onRows is called once per yielded batch, mid-stream"; `portal-shell.test.ts` — "G3: buffered rows are visible to the gate MID-CHUNK", "S1: rows return to baseline after chunks settle and are evicted". |
 | **INV-8** | **AIMD bounds.** `MIN ≤ limit ≤ MAX`; `active ≥ 0`; admission only while `active < limit`; waiters are FIFO and drained whenever capacity exists (no starvation) — including when a RAMP raises the limit. | The shared endpoint's live capacity is discovered by AIMD; the controller must never exceed its own bounds or starve a waiter. | Pure reducer `gateReduce(state, event) → state`; the shell pumps FIFO on release AND onOk. | `gateReduce` asserts `MIN ≤ limit ≤ MAX` and `active ≥ 0`. | `portal-gate.test.ts` — "INV-8/INV-7: bounds hold under arbitrary event sequences", "createGate: FIFO admission, no starvation while capacity exists", "createGate: an AIMD ramp PUMPS parked waiters". |
-| **INV-9** (was C3) | **Finality safety.** A Portal data request never targets past the known finalized head (an explicit backfill `toBlock` may exceed the head by configuration); an interval ending past the head — or an unknown head — delegates the WHOLE interval to RPC (stream-realtime mode: warn + empty). The head is monotonic upstream, so staleness errs safe (stale-LOW). | Portal serves only finalized data; its head can lag Ponder's target. Serving the missing tail would 204 and mark it synced (silent hole). | `portal.ts` finality/delegation branch; `chunkRange` clamps `to ≤ dataEnd`. | `portal.ts` `dataChunk` asserts `desiredTo ≤ portalHead` at request build (when the head is known and no explicit backfill end overrides). | `portal-shell.test.ts` — "INV-9: an interval past the Portal head delegates WHOLE to RPC; the delegated key is consumed exactly once", "INV-9: an unknown head (probe persistently failing) delegates to RPC", "INV-9: stream-realtime mode suppresses the RPC fallback (warn + empty, NOT delegated)"; `portal-transform.test.ts` — "isFinalityGap…". |
-| **INV-10** | **Realtime chain linkage.** The unfinalized list is strictly increasing in number and parentHash-linked; `reconcile` yields reorgs whose `reorgedBlocks` are exactly the trimmed suffix; `gap` resets; `finalize` is monotonic and never at/below the startup boundary. | Realtime reorg/finalize handling must be provably correct to never regress persisted checkpoints. | Pure `reconcile`/`takeFinalized` in `portal-realtime.ts`; the wire suppresses finalize ≤ startup/lastFinalized. | `portalRealtimeEvents` asserts the parentHash link + increasing number on every append (O(1) tripwire — reconcile guarantees it). | `portal-realtime.test.ts` (reconcile/takeFinalized cases); `portal-realtime-wire.test.ts` (finalize monotonic + generator flows). |
+| **INV-9** (was C3) | **Finality safety.** A Portal data request never targets past the known finalized head (an explicit backfill `toBlock` may exceed the head by configuration); an interval ending past the head — or an unknown head — delegates the WHOLE interval to RPC. In stream-realtime mode the RPC fallback is suppressed: a KNOWN head past the interval is by design (realtime `/stream` covers it → debug + empty), but an UNKNOWN head is FATAL at both the clamp and delegation seams — the historical↔realtime boundary can't be located, so failing loud beats marking the range synced with no data (fixes **G4**). The head is monotonic upstream, so staleness errs safe (stale-LOW). | Portal serves only finalized data; its head can lag Ponder's target. Serving the missing tail would 204 and mark it synced (silent hole). | `portal.ts` finality/delegation branch (stream mode throws on an UNKNOWN head, not warn+empty); `clampFinalizedToPortalHead` retries the probe then throws on a persistently unknown head; `chunkRange` clamps `to ≤ dataEnd`. | `portal.ts` `dataChunk` asserts `desiredTo ≤ portalHead` at request build (when the head is known and no explicit backfill end overrides). | `portal-shell.test.ts` — "INV-9: an interval past the Portal head delegates WHOLE to RPC; the delegated key is consumed exactly once", "INV-9: an unknown head (probe persistently failing) delegates to RPC", "INV-9: stream-realtime mode suppresses the RPC fallback (empty, NOT delegated) — known head past the interval", "INV-9: stream-realtime mode with an UNKNOWN head is FATAL — refuses to mark the range synced with no data (finding 6)"; `portal-realtime-wire.test.ts` — "clampFinalizedToPortalHead: Portal head unknown in stream mode → FATAL"; `portal-transform.test.ts` — "isFinalityGap…". |
+| **INV-10** | **Realtime chain linkage.** The unfinalized list is strictly increasing in number and parentHash-linked; `reconcile` yields reorgs whose `reorgedBlocks` are exactly the trimmed suffix; an unknown-parent `gap` is **FATAL** (throws — never silently resets, which would leave already-indexed unfinalized blocks on the wrong fork with no rollback event, fixes **G5**); `finalize` is monotonic and never at/below the startup boundary. | Realtime reorg/finalize handling must be provably correct to never regress persisted checkpoints; a gap we can't reconcile has no safe silent recovery. | Pure `reconcile`/`takeFinalized` in `portal-realtime.ts`; `portalRealtimeEvents` throws on a `gap`; the wire suppresses finalize ≤ startup/lastFinalized. | `portalRealtimeEvents` asserts the parentHash link + increasing number on every append (O(1) tripwire — reconcile guarantees it; a `gap` throws before it). | `portal-realtime.test.ts` — reconcile/takeFinalized cases, "portalRealtimeEvents: an unknown-parent gap is FATAL, not silently skipped (finding 7)"; `portal-realtime-wire.test.ts` (finalize monotonic + generator flows). |
 | **INV-11** | **Merge equivalence.** `mergeLogRequests(reqs)` matches exactly the same logs as `reqs` (∀log: matches(reqs) ⟺ matches(merged)); the merge key includes the FULL (address set, topic1, topic2, topic3) tuple; batches respect `PORTAL_MAX_ADDRESSES`; body-size guard at the client. | An N-event contract emits N requests repeating the child list; merging shrinks the body without changing the result set. | Pure `mergeLogRequests` (topic0 union, undefined-absorbs-all). | Body-size guard in `portal-client`. | `portal-filters.test.ts` — "INV-11: mergeLogRequests matches exactly the same logs as the originals" (the model covers address + topic0..3), "INV-11: requests differing ONLY in topic2/topic3 are NOT merged (key includes every topic)". |
 | **INV-12** | **Exactly-once insertion / stash lifecycle.** Per interval: logs inserted once (range-data), blocks/txs/receipts/traces once (block-data); a stash entry is created then consumed exactly once; `delegated` entries consumed once; stash stays bounded. An upstream range RETRY legitimately re-creates an entry: production (`on`) keeps the pre-refactor overwrite semantics (debug-logged); `strict` makes the double-set loud. | Double insertion corrupts; a leaked stash entry leaks memory; but a benign upstream retry must not crash production. | `portal.ts` inserts logs in range-data, block-data in block-data; stash `set` then `delete`. | `invariantStrict("INV-12", …)` on set (strict mode only); `delete` on take. | `portal-shell.test.ts` — "INV-12: the stash is consumed exactly once — a second syncBlockData returns undefined without re-inserting", "INV-12: 'on' mode keeps overwrite semantics for an upstream range retry; 'strict' makes it loud"; `portal.test.ts` (insert-exactly-once round-trips). |
 | **INV-13** | **Progress & cache bounds.** The stream cursor strictly advances BY CONSTRUCTION (`cursor = last + 1 ≥ cursor + 1` per batch; a 204 terminates — no runtime guard could ever fire, so none exists); the chunk cache never exceeds the in-service span + READAHEAD; **a rejected chunk promise is evicted immediately** so a later interval retries instead of replaying the cached rejection (fixes **G1**); **a FRONTIER chunk truncated at a then-lower finalized head is EXTENDED (tail-only stream + merge) before being served past its `coveredTo`** — never served stale (head-boundary bug found in review; fixed on main by PR #5/802ceed and RE-IMPLEMENTED here with behavior parity — main's own #5 regression test passes unchanged). A failed extend evicts the whole entry — never leaves the optimistic high-water in place. | A cached rejection permanently poisons a chunk; an unbounded cache OOMs; a blind idx cache hit over a head-truncated chunk marks an interval synced over a silent gap. | `readAheadPlan`/`evictionPlan` bound the cache (pure); `dataChunk` records `coveredTo` per cache entry and extends via the shared `runStreams` (append-only merge; rows accounted through the same per-fetch token); `p.catch`/`extended.catch` evict + free exactly once. | `dataChunk`'s INV-1/INV-9/INV-3 asserts bracket every fetch; `portal-client.ts` documents progress-by-construction at the cursor advance. | `portal-chunks.test.ts` — "INV-13: read-ahead plan is bounded (≤ readahead), within raEnd, and depth-1-only when saturated", "eviction plan…"; `portal-client.test.ts` — "INV-13: the cursor strictly advances and the stream terminates"; `portal.test.ts` — "regression: frontier chunk truncated at a lagging Portal head is EXTENDED…" (main's #5 test, verbatim); `portal-shell.test.ts` — "G1: a failed chunk is evicted (rows freed) and a later call refetches…" and "frontier extend streams ONLY the newly-finalized tail…". |
@@ -41,3 +41,5 @@ Legacy `C#` tags from the pre-refactor comments are mapped where they existed.
 - **Head-boundary staleness → INV-13** (found in review; fixed on main by PR #5, merged as 802ceed — re-implemented here with behavior parity incl. the `extends` metrics counter; main's #5 regression test passes unchanged). A frontier chunk fetched while the Portal head sat inside its grid range was cached truncated forever; when the head advanced, later intervals in the same chunk were served from the truncated cache and marked synced over a silent gap. Fixed with per-entry `coveredTo` + tail-only EXTEND. One deliberate divergence: on a FAILED extend, main leaves a cached rejection (its pre-existing G1 gap); here the entry is evicted and its rows freed — the G1-consistent recovery this PR introduces.
 - **Restart child loss → INV-15** (fixed on main by PR #2; ported into this architecture with all six of its regression tests passing unchanged).
 - **Backfill-floor silent gap (C10) → INV-16** (found + fixed by PR #8, @mo4islona; ported here onto `compileFetchSpec`). `backfillStart` took `Math.min` over only the DEFINED `fromBlock`s, so one bounded source (15M) raised the floor past an unbounded source and `chunkRange` clamped its `[0, 15M)` history out — assembled empty and marked synced. Now any undefined `fromBlock` ⇒ floor `0` (symmetric with `backfillEnd`). Mutation-verified: reverting the floor to `Math.min` over the defined-only list fails both the `portal-filters.test.ts` unit assertion (mixed ⇒ 0) and the `portal.test.ts` C10 regression (the unbounded source's block-100 log goes un-inserted).
+- **Stream-mode finalized-head probe (C11) → INV-9** (PR #8, @mo4islona). In `PORTAL_REALTIME=stream` the RPC finality-gap fallback is suppressed and the Portal head IS the historical↔realtime boundary. A persistently failing `/finalized-head` probe left the clamp passing the RPC finalized block through (historical then targeted `(portalHead, rpcFinalized]`, which stream mode serves nothing for) and the delegation seam returning `[]` — both marking the range synced while realtime streamed only ABOVE the head we couldn't find: a permanent silent gap. Now BOTH seams fail loud — `clampFinalizedToPortalHead` retries then throws, `portal.ts` throws on an unknown head. Mutation-verified: reverting either throw fails its stream-mode test. NOTE: this changes the pre-existing INV-9 "warn + empty" / clamp "conservative passthrough" contract — deliberately (see INV-9).
+- **Unknown-parent realtime gap (C-gap) → INV-10** (PR #8, @mo4islona). A streamed block whose parent is unknown to the unfinalized window (a reorg deeper than the window, or a skipped block) was silently cleared-and-continued, leaving the already-indexed unfinalized blocks on the WRONG fork with no rollback event and dropping the skipped span. Now `portalRealtimeEvents` throws (a rejection escaping the realtime generator hits ponder's `unhandledRejection` → `exit(75)`; a restart re-derives the window cleanly from the finalized head). Mutation-verified: restoring `unfinalized.length = 0` fails "an unknown-parent gap is FATAL". NOTE: changes the pre-existing INV-10 "`gap` resets" contract — deliberately (see INV-10). A walk-parents-and-refetch auto-recovery is a follow-up.

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-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,30 @@ 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([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([logFilter(), 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(
+      [logFilter({ hasTransactionReceipt: true })],
+      'mainnet',
+    ),
+  ).toThrow(/transaction receipts/);
+});
+
 // ─────────────────────────────── light-block conversion ───────────────────────────────
 
 test('lightToLightBlock: Portal decimal number/timestamp → ponder hex, hashes passthrough', () => {
@@ -289,7 +314,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 +325,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 () => {