fix(portal): review round — G2 propagation, typed-token row accounting, INV-9 assert

- portal-discovery: a predecessor extension's failure now PROPAGATES (await earlier
  without swallowing) so a successor can never confirm coverage over an unscanned
  gap (B1/G2 — silent factory-child loss reproduced in review).
- portal.ts: per-fetch RowToken accounting (S1 — a stale evicted fetch can't
  register/free rows against its replacement); coveredTo recorded per cache entry;
  INV-9 runtime assert at request build; INV-12 stash assert now strict-mode-only
  (production keeps pre-refactor overwrite semantics + debug log, S2);
  refreshPortalHead retry via client (injectable sleep, S5k).
- portal-client: finalizedHeadRetry; countRows includes headers (S5g); dead
  NoProgressError guard removed — progress is by construction (B4c).
- portal-errors: NoProgressError deleted; InvariantViolation message points at
  INVARIANTS.md + PORTAL_CHECKS escape hatch (S5b).
- portal/config.ts restored (repo-side compat-harness utility — B2); PUBLISHING.md
  stale sentence fixed.

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

Author: dzhelezov

PUBLISHING.md

@@ -11,9 +11,11 @@ field per chain that routes the historical backfill through SQD Portal (realtime
   number, so a plain mirror (`0.16.6`) can't be re-published after a bad build. Each release is
   published with `--tag latest`, so `npm i @subsquid/ponder` resolves it (npm doesn't auto-pick a
   prerelease otherwise); an exact build can be pinned as `@subsquid/ponder@0.16.6-sqd.1`.
-- **We don't hand-maintain a fork.** This repo holds only the **Portal layer** (`portal/`): two modules
-  (`portal.ts`, `portal-transform.ts`) + a per-version `wiring/<ver>.patch` (the 4 one-line touch-points)
-  + `config.ts` (`withPortal`). That's the entire diff against upstream.
+- **We don't hand-maintain a fork.** This repo holds only the **Portal layer** (`portal/`): the
+  `portal-*.ts` modules (an invariant-first functional core behind the `portal.ts` shell — see
+  `portal/INVARIANTS.md`) + a per-version `wiring/<ver>.patch` (the 4 one-line touch-points). That's
+  the entire diff against upstream. (`portal/config.ts` is a repo-side compat-harness utility — not
+  part of the published build.)
 - **The fork is generated.** `scripts/sync-upstream.sh <ver>` clones `ponder@<ver>`, applies the layer,
   renames the package, and builds — producing the publishable package. Tracking a new ponder version is
   "author one small patch + run the script", not "merge a fork".

portal/portal-client.ts

@@ -15,7 +15,6 @@
  * real network or timers.
  */
 import {
-  NoProgressError,
   PortalDatasetStartError,
   PortalHttpError,
   PortalQueryTooLargeError,
@@ -74,9 +73,12 @@ const stripFields = (q: PortalQuery, dropped: Set<string>): PortalQuery => {
   return { ...q, fields };
 };
 
+// Row accounting is CONSERVATIVE (INV-7): each block counts its header too, so header-only batches
+// (block-interval includeAllBlocks scans, whose retained headers are real buffered memory) register
+// against the budget instead of registering ~0.
 const countRows = (blocks: RawBlock[]): number => {
   let n = 0;
-  for (const b of blocks) n += (b.logs?.length ?? 0) + (b.transactions?.length ?? 0) + (b.traces?.length ?? 0);
+  for (const b of blocks) n += 1 + (b.logs?.length ?? 0) + (b.transactions?.length ?? 0) + (b.traces?.length ?? 0);
   return n;
 };
 
@@ -90,6 +92,8 @@ export type StreamOpts = {
 export interface PortalClient {
   /** GET /finalized-head → the finalized block number (undefined on any failure). */
   finalizedHead(): Promise<number | undefined>;
+  /** finalizedHead() with a bounded retry (backoff via the injectable sleep). */
+  finalizedHeadRetry(attempts?: number): Promise<number | undefined>;
   /** POST /finalized-stream and yield parsed NDJSON batches over [from, to]. */
   stream(query: PortalQuery, from: number, to: number, opts?: StreamOpts): AsyncGenerator<RawBlock[]>;
 }
@@ -215,7 +219,9 @@ export function createPortalClient(deps: PortalClientDeps): PortalClient {
       if (batch === "done") return;
       if (onRows) onRows(countRows(batch.blocks));
       yield batch.blocks;
-      if (batch.last < cursor) throw new NoProgressError(cursor);
+      // Progress by construction (INV-13): fetchBatch initialises `last = cursor` and only ever raises
+      // it, so `cursor = last + 1 ≥ cursor + 1` — the cursor strictly advances on every yielded batch,
+      // and a 204 terminates. No runtime guard is needed (and none could ever fire).
       cursor = batch.last + 1;
     }
   }
@@ -228,5 +234,16 @@ export function createPortalClient(deps: PortalClientDeps): PortalClient {
     return undefined;
   };
 
-  return { finalizedHead, stream };
+  // The head probe is cheap and load-bearing for the finality-gap decision, so callers retry it.
+  // `sleepImpl` injection keeps the retry cadence testable (no real timers in unit tests).
+  const finalizedHeadRetry = async (attempts = 3): Promise<number | undefined> => {
+    for (let attempt = 0; attempt < attempts; attempt++) {
+      const h = await finalizedHead();
+      if (h !== undefined) return h;
+      await sleep(200 * (attempt + 1));
+    }
+    return undefined;
+  };
+
+  return { finalizedHead, finalizedHeadRetry, stream };
 }

portal/portal-discovery.ts

@@ -110,10 +110,16 @@ export function createDiscovery(deps: DiscoveryDeps): Discovery {
     through = to; // optimistic advance so concurrent ensures dedup onto one scan
     status = "scanning";
     const p = (async () => {
-      await earlier.catch(() => {}); // G2: a prior failure doesn't poison this extension
+      // G2/INV-3: a predecessor extension's failure MUST propagate. This extension only scans
+      // [through+1..to] on the assumption the predecessor covered everything below it; swallowing the
+      // failure and then confirming `to` would certify coverage over the predecessor's unscanned gap —
+      // permanently losing any child created inside it. By rethrowing, this extension rejects BEFORE
+      // scanning, its catch below rolls `through` back to `confirmed`, and the awaiting data chunks
+      // reject (G1-evicted) → a later interval replans contiguously from confirmed+1 / the floor.
+      await earlier;
       stats.discChunks += windows.length;
       await Promise.all(windows.map(([lo, hi]) => scanWindow(lo, hi)));
-      confirmed = to; // INV-3: advance the confirmed watermark ONLY on success
+      confirmed = to; // INV-3: advance the confirmed watermark ONLY on success (never past a gap)
       status = "idle";
     })();
     inflight = p;

portal/portal-errors.ts

@@ -86,16 +86,6 @@ export class PortalQueryTooLargeError extends Error {
   }
 }
 
-/** The stream failed to advance past the cursor (a Portal that returned a lower `last`). */
-export class NoProgressError extends Error {
-  readonly cursor: number;
-  constructor(cursor: number) {
-    super(`Portal no progress @ ${cursor}`);
-    this.name = "NoProgressError";
-    this.cursor = cursor;
-  }
-}
-
 /**
  * A violated runtime invariant (see portal-invariant.ts + INVARIANTS.md). The repo's
  * philosophy: a loud crash beats silent corruption. Carries the invariant `id` and a
@@ -105,7 +95,10 @@ export class InvariantViolation extends Error {
   readonly id: string;
   readonly context: Record<string, unknown> | undefined;
   constructor(id: string, msg: string, context?: Record<string, unknown>) {
-    super(`Invariant ${id} violated: ${msg}${context ? ` — ${safeJson(context)}` : ""}`);
+    super(
+      `Invariant ${id} violated: ${msg}${context ? ` — ${safeJson(context)}` : ""} ` +
+        `(see portal/INVARIANTS.md ${id}; set PORTAL_CHECKS=off to bypass runtime checks)`,
+    );
     this.name = "InvariantViolation";
     this.id = id;
     this.context = context;

portal/portal.ts

@@ -65,11 +65,17 @@ export const createPortalHistoricalSync = (args: CreateHistoricalSyncParameters)
   const discovery = createDiscovery({ client, childAddresses: args.childAddresses, factories: spec.factories, discoveryWindows: cfg.discoveryWindows, stats });
 
   // ── mutable shell state ──────────────────────────────────────────────────────────────────────────
-  const dataCache = new Map<number, Promise<ChunkData>>(); // idx → chunk (INV-1: filter-complete)
-  const chunkSpecId = new Map<number, symbol>(); //          idx → fetch-spec identity (INV-1)
-  const chunkRows = new Map<number, number>(); //            idx → registered rows (freed on evict/reject)
-  const stash = new Map<string, StashEntry>(); //            interval → block-data, consumed by syncBlockData
-  const delegated = new Set<string>(); //                    interval keys routed to RPC (finality gap)
+  // One cache entry per chunk idx. `token` keys the row accounting to THIS fetch (not the idx): a stale
+  // in-flight fetch evicted and replaced at the same idx can neither register rows into nor free rows
+  // from the replacement's budget — its own token is freed exactly once (idempotent). `coveredTo` is the
+  // upper bound this fetch actually covered (head-clamped at fetch time), revalidated on every hit
+  // (INV-13: a chunk truncated at a then-lower finalized head is refetched, never served stale).
+  type RowToken = { rows: number; freed: boolean };
+  type CacheEntry = { promise: Promise<ChunkData>; specId: symbol; coveredTo: number; token: RowToken };
+  const dataCache = new Map<number, CacheEntry>();
+  const freeToken = (t: RowToken): void => { if (!t.freed) { t.freed = true; gate.freeRows(t.rows); } };
+  const stash = new Map<string, StashEntry>(); // interval → block-data, consumed by syncBlockData
+  const delegated = new Set<string>(); //        interval keys routed to RPC (finality gap)
   let chunkBlocks = cfg.chunkBlocks;
   let chunkSizeP: Promise<void> | undefined;
   let discStartIdx: number | undefined; // factory-deploy chunk = discovery floor (C4: clamps DOWNWARD)
@@ -89,11 +95,8 @@ export const createPortalHistoricalSync = (args: CreateHistoricalSyncParameters)
 
   const refreshPortalHead = async (): Promise<number | undefined> => {
     if (cfg.finalizedHead !== undefined) return (portalHead = cfg.finalizedHead);
-    for (let attempt = 0; attempt < 3; attempt++) {
-      const h = await client.finalizedHead();
-      if (h !== undefined) return (portalHead = h);
-      await new Promise((r) => setTimeout(r, 200 * (attempt + 1)));
-    }
+    const h = await client.finalizedHeadRetry(3); // retry lives in the client (injectable sleep)
+    if (h !== undefined) return (portalHead = h);
     return portalHead; // may be a kept-prior value, or undefined if never probed successfully
   };
 
@@ -106,21 +109,35 @@ export const createPortalHistoricalSync = (args: CreateHistoricalSyncParameters)
     })());
 
   // ── one data chunk: gated on discovery-through-this-chunk, then the source streams ──────────────────
-  const dataChunk = (idx: number): Promise<ChunkData> => {
+  // `needTo` = the highest block the CALLER requires this chunk to cover (interval end / current grid
+  // end). A cached entry is served only when its fetch actually covered that far (INV-13).
+  const dataChunk = (idx: number, needTo: number): Promise<ChunkData> => {
     const cached = dataCache.get(idx);
     if (cached) {
-      stats.cacheHits++;
-      invariant("INV-1", chunkSpecId.get(idx) === spec.id, "cached chunk built under a different fetch-spec", () => ({ idx }));
-      return cached;
+      invariant("INV-1", cached.specId === spec.id, "cached chunk built under a different fetch-spec", () => ({ idx }));
+      if (cached.coveredTo >= needTo) { stats.cacheHits++; return cached.promise; }
+      // Head-boundary staleness (INV-13, bug found relative to main): this chunk was fetched while the
+      // Portal head sat inside its grid range, so it was truncated at the then-head. The head has since
+      // advanced past what the caller needs — serving the cache would silently omit (fetchTimeHead,
+      // needTo]. Evict + refetch the whole chunk under the current head.
+      dataCache.delete(idx);
+      freeToken(cached.token);
+      log.debug({ service: "portal", msg: `Portal ${chain.name}: chunk ${idx} covered to ${cached.coveredTo} < needed ${needTo} (head advanced) → refetch` });
     }
+    const [from, to] = chunkRange(idx, chunkBlocks, spec.backfillStart, dataEnd());
+    // INV-9: a Portal data request never targets past the known finalized head (an explicit backfill
+    // toBlock may exceed it by configuration — the delegation branch already guards served intervals).
+    invariant("INV-9", portalHead === undefined || to <= portalHead || spec.backfillEnd !== undefined, "data request targets past the Portal finalized head", () => ({ idx, to, portalHead }));
+    const token: RowToken = { rows: 0, freed: false };
+    // G3: register rows as batches ARRIVE. Guarded by the token so a stale stream that outlives its
+    // eviction cannot register orphaned rows (S1: accounting is per-fetch, not per-idx).
+    const onRows = (n: number): void => { if (token.freed) return; token.rows += n; gate.addRows(n); };
     const p = (async (): Promise<ChunkData> => {
-      const [from, to] = chunkRange(idx, chunkBlocks, spec.backfillStart, dataEnd());
       await discovery.ensure(to, { chunkBlocks, endHint: spec.backfillEnd ?? portalHead ?? to }); // children ≤ this chunk are known
       invariant("INV-3", spec.factories.length === 0 || discStartIdx === undefined || discovery.through() >= to, "data fetch under a stale discovery watermark", () => ({ idx, through: discovery.through(), to }));
       stats.dataChunks++;
       const cd = createChunkData();
       const neededMissing = new Set<string>();
-      const onRows = (n: number): void => { chunkRows.set(idx, (chunkRows.get(idx) ?? 0) + n); gate.addRows(n); }; // G3: register as batches arrive
 
       const lq = spec.logQuery();
       if (lq) for await (const blocks of client.stream(lq, from, to, { neededMissing, onRows })) {
@@ -158,20 +175,22 @@ export const createPortalHistoricalSync = (args: CreateHistoricalSyncParameters)
       }
       return cd;
     })();
-    chunkSpecId.set(idx, spec.id);
-    dataCache.set(idx, p);
-    // G1 (INV-13): a rejected chunk promise is evicted immediately (and its registered rows freed) so a
-    // later interval RETRIES rather than replaying the cached rejection forever.
+    const entry: CacheEntry = { promise: p, specId: spec.id, coveredTo: to, token };
+    dataCache.set(idx, entry);
+    // G1 (INV-13): a rejected chunk promise is evicted immediately (and its registered rows freed — via
+    // ITS token, exactly once) so a later interval RETRIES rather than replaying the cached rejection.
     p.catch(() => {
-      if (dataCache.get(idx) === p) { dataCache.delete(idx); chunkSpecId.delete(idx); }
-      gate.freeRows(chunkRows.get(idx) ?? 0); chunkRows.delete(idx);
+      if (dataCache.get(idx) === entry) dataCache.delete(idx);
+      freeToken(token);
     });
     return p;
   };
 
   const evictBehind = (intervalStart: number): void => {
     for (const i of evictionPlan(dataCache.keys(), chunkBlocks, intervalStart)) {
-      dataCache.delete(i); chunkSpecId.delete(i); gate.freeRows(chunkRows.get(i) ?? 0); chunkRows.delete(i);
+      const entry = dataCache.get(i)!;
+      dataCache.delete(i);
+      freeToken(entry.token);
     }
   };
 
@@ -199,8 +218,8 @@ export const createPortalHistoricalSync = (args: CreateHistoricalSyncParameters)
       const capped = traceSafeChunkBlocks(chunkBlocks, spec.needTraces || spec.needBlocks, cfg.traceChunkBlocks);
       if (capped !== chunkBlocks) {
         chunkBlocks = capped;
-        dataCache.clear(); chunkSpecId.clear();
-        for (const r of chunkRows.values()) gate.freeRows(r); chunkRows.clear();
+        for (const entry of dataCache.values()) freeToken(entry.token);
+        dataCache.clear();
         discStartIdx = undefined; discovery.reset();
         log.debug({ service: "portal", msg: `Portal ${chain.name}: dense sources → chunkBlocks capped to ${chunkBlocks} (grid reset)` });
       }
@@ -215,11 +234,16 @@ export const createPortalHistoricalSync = (args: CreateHistoricalSyncParameters)
       const startIdx = idxOf(interval[0], chunkBlocks), endIdx = idxOf(interval[1], chunkBlocks);
       const idxs: number[] = [];
       for (let i = startIdx; i <= endIdx; i++) idxs.push(i);
-      const data = await Promise.all(idxs.map(dataChunk));
+      // needTo per chunk: the interval's requirement clamped to the chunk's current grid end — a cached
+      // entry truncated at an older, lower head fails the coveredTo check and is refetched (INV-13).
+      const needToOf = (i: number): number => Math.min(chunkRange(i, chunkBlocks, spec.backfillStart, dataEnd())[1], interval[1]);
+      const data = await Promise.all(idxs.map((i) => dataChunk(i, needToOf(i))));
 
       // PARALLEL read-ahead: prefetch ahead (never past the backfill end) — depth bounded by the shared
       // memory budget, not a fixed count (always lead-1; deeper only while unsaturated).
-      for (const d of readAheadPlan(endIdx, chunkBlocks, dataEnd(), cfg.readahead, gate.saturated())) void dataChunk(d).catch(() => {});
+      for (const d of readAheadPlan(endIdx, chunkBlocks, dataEnd(), cfg.readahead, gate.saturated())) {
+        void dataChunk(d, chunkRange(d, chunkBlocks, spec.backfillStart, dataEnd())[1]).catch(() => {});
+      }
 
       const tXform = Date.now(); // decode/transform time: Portal NDJSON → Ponder Sync* shapes
       const assembled = assembleRange(data, interval, spec, args.childAddresses);
@@ -228,7 +252,11 @@ export const createPortalHistoricalSync = (args: CreateHistoricalSyncParameters)
       evictBehind(interval[0]); // free chunks fully behind the cursor + their memory budget
 
       await syncStore.insertLogs({ logs: assembled.logs, chainId: chain.id });
-      invariant("INV-12", !stash.has(ikey(interval)) && !delegated.has(ikey(interval)), "stash entry created twice / for a delegated interval", () => ({ key: ikey(interval) }));
+      // INV-12: a stash entry is created then consumed exactly once. An upstream retry can legitimately
+      // re-issue a range, so production (`on`) keeps the pre-refactor OVERWRITE semantics with a debug
+      // log; `strict` (tests/CI) makes the double-set loud.
+      invariantStrict("INV-12", () => !stash.has(ikey(interval)) && !delegated.has(ikey(interval)), "stash entry created twice / for a delegated interval", () => ({ key: ikey(interval) }));
+      if (stash.has(ikey(interval))) log.debug({ service: "portal", msg: `Portal ${chain.name} [${interval[0]},${interval[1]}]: stash entry overwritten (upstream range retry)` });
       stash.set(ikey(interval), { blocks: assembled.blocks, txs: assembled.txs, receipts: assembled.receipts, traces: assembled.traces, closest: assembled.closest });
 
       log.debug({ service: "portal", msg: `Portal ${chain.name} [${interval[0]},${interval[1]}]: ${assembled.logs.length} logs (dataChunks=${stats.dataChunks} discChunks=${stats.discChunks} http=${stats.http} hits=${stats.cacheHits} inflight=${stats.maxInflight} err=${stats.errors})` });