fix(portal): wave-4 review — five medium fixes (store parity, monotonic head, schema-degradation bounds, bounded realtime probe) (#44)

* fix(portal): wave-4 review — five medium fixes (log re-match parity, monotonic head, per-field schema bound, stream block-field degradation, bounded realtime probe)

Five fixes from a four-lens validation review of the portal layer, each
mutation-verified (reverting the fix hunk fails the named regression test):

1. assembleRange re-matches logs per-filter (isLogFilterMatched +
   isAddressMatched) — factory pre-creation logs and a bounded filter's
   out-of-range logs no longer diverge the sync store from the RPC path
   (INV-6/INV-2); a dropped log's parent tx/block no longer rides in.
2. The cached Portal head is monotonic (LB replicas answer stale-LOW), and
   stream mode's known-head-below-interval branch is FATAL instead of
   returning [] — closing the reopened G4/C11 silent-gap class (INV-9).
3. stream()'s schema-degradation bound keys on the table-qualified fieldKey,
   not the bare column name — a dataset missing the same column in two
   tables (logs_bloom in blocks AND transactions) no longer crash-loops.
4. /stream 400 degradation covers droppable block.* fields, not just
   transaction.* — completing review B3 (a mix_hash-less dataset no longer
   refuses stream mode after backfilling fine).
5. portalFinalizedHead delegates to the shared bounded probe
   (probeFinalizedHead, issue #14/PR #16 hardening) — a hung
   /finalized-head can no longer freeze finalize emission or startup;
   deletes the duplicated weak probe.

Suite green on 0.16.6 and 0.15.17 grafts (197 tests, 7 new).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(portal): wave-4 review round 2 — matched-log needed-field gate + #47 INVARIANTS candor

Addresses the CHANGES_REQUESTED on #44 (two narrow deltas).

2 (medium) — the needed-field fatal counted raw `cd.logs.size`, but the new log
re-match (INV-6 store parity) can drop every log a chunk added (a factory child's
pre-creation logs / a bounded filter's out-of-range logs). A tail of all-dropped
logs over a dataset missing a needed field armed a false `neededMissing` fatal →
G1 evict → crash-loop, for data the indexer never keeps — the exact class of #20's
trace/transfer residual, created here by the new re-match boundary. `runStreams`
now counts only logs surviving `logMatched` toward the growth check (trace/tx/block
sources keep the raw map-size delta; the #20 trace/transfer residual is unchanged).
New `buildRawLogMatcher` reuses `buildMatchers` + `toSyncLog`, so the seam count and
`assembleRange` agree by construction over the same discovery-complete childAddresses.
Mutation-verified: reverting the seam to the raw `cd.logs.size` delta fails exactly
the new `portal.test.ts` regression.

1 (INVARIANTS candor) — reworded the INV-9 row + the wave-4 monotonic bullet: the
`max`-keep guarantees only that the cached head never exceeds the true finalized head;
range completeness against the *serving* replica is not verified (a phantom-coverage
hole pre-existing on main per PR #15 FIX 1, widened to process-lifetime by max-keeping).
Both now reference #47; the client-side range-completeness guard is deferred there
pending one empirical check of Portal 204 semantics. INV-13 residual note updated
(logs now matched-precise; trace/transfer the remaining #20 slice).

Gates: 0.16.6 + 0.15.17 `--test` green (198 each, +1 test); biome clean.

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

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: mo4islona

portal/INVARIANTS.md

@@ -167,6 +167,89 @@ test('a tx matched by BOTH a log filter and a tx filter keeps one tx row AND get
   expect((out.receipts[0] as any).transactionHash).toBe('0xBOTH');
 });
 
+// ── log re-match: parity with the RPC path's per-filter matching (wave 4) ───────────────────────────
+
+test("wave 4: a factory child's PRE-CREATION log is excluded — and its parent tx/block with it (parity with isAddressMatched)", () => {
+  // The Portal log request carries the child address with NO per-address creation floor, so a child that
+  // emitted filter-matching logs BEFORE its creation event comes back "matched". Upstream's RPC path
+  // stores a child's logs only from its creation block on (isAddressMatched: creation ≤ blockNumber);
+  // skipping the re-match stored the pre-creation log + its tx + its block — an INV-6 store divergence.
+  const CHILD = '0x000000000000000000000000000000000c0ffee1';
+  const factory: any = {
+    id: 'f1',
+    type: 'log',
+    chainId: 1,
+    sourceId: 's',
+    address: '0x00000000000000000000000000000000000fac70',
+    eventSelector: '0xsel',
+    childAddressLocation: 'topic1',
+    fromBlock: undefined,
+    toBlock: undefined,
+  };
+  const factoryLogFilter: any = { ...logFilter, address: factory };
+  const childAddresses = new Map([['f1', new Map([[CHILD, 100]])]]); // created at block 100
+  const cd = createChunkData();
+  cd.headers.set(50, header(50));
+  cd.logs.set(50, [rawLog(CHILD, '0xtxPRE')]); // emitted BEFORE creation → must be dropped
+  cd.txs.set(50, [rawTx('0xtxPRE')]);
+  cd.headers.set(150, header(150));
+  cd.logs.set(150, [rawLog(CHILD, '0xtxPOST')]); // after creation → kept
+  cd.txs.set(150, [rawTx('0xtxPOST')]);
+  const spec = compileFetchSpec(
+    [{ filter: factoryLogFilter }],
+    childAddresses as any,
+  );
+  const out = assembleRange([cd], [0, 200], spec, childAddresses as any);
+  expect(out.logs).toHaveLength(1);
+  expect(Number(BigInt((out.logs[0] as any).blockNumber))).toBe(150);
+  expect(out.txs).toHaveLength(1); // the dropped log's parent tx must not ride in
+  expect((out.txs[0] as any).hash).toBe('0xtxPOST');
+  expect(out.blocks.map((b) => Number(BigInt((b as any).number)))).toEqual([
+    150,
+  ]); // block 50 held ONLY the dropped log → not stored
+});
+
+test("wave 4: a bounded filter's below-fromBlock log is excluded even when another filter's chunk fetched it (per-filter range re-match)", () => {
+  // Chunk fetches are bounded by the SPEC-global backfillStart/dataEnd, so with filter A unbounded and
+  // filter B fromBlock:200, chunks below 200 are fetched with B's topics too (merged server request).
+  // Upstream's isLogFilterMatched rejects a B-matching log below 200; without the re-match it was
+  // stored and returned as matched — an INV-6 store divergence.
+  const filterA: any = { ...logFilter, topic0: '0xaaaa' }; // unbounded
+  const filterB: any = { ...logFilter, topic0: '0xbbbb', fromBlock: 200 };
+  const logWithTopic = (topic0: string, tx: string) => ({
+    ...rawLog('0xV', tx),
+    topics: [topic0],
+  });
+  const cd = createChunkData();
+  cd.headers.set(100, header(100));
+  // block 100: an A log (kept) and a B log BELOW B's fromBlock (dropped)
+  cd.logs.set(100, [
+    logWithTopic('0xaaaa', '0xtxA100'),
+    logWithTopic('0xbbbb', '0xtxB100'),
+  ]);
+  cd.txs.set(100, [rawTx('0xtxA100'), rawTx('0xtxB100')]);
+  cd.headers.set(250, header(250));
+  cd.logs.set(250, [logWithTopic('0xbbbb', '0xtxB250')]); // in B's range → kept
+  cd.txs.set(250, [rawTx('0xtxB250')]);
+  const spec = compileFetchSpec(
+    [{ filter: filterA }, { filter: filterB }],
+    new Map(),
+  );
+  const out = assembleRange([cd], [0, 300], spec, new Map());
+  const keptBlocks = out.logs.map((l) =>
+    Number(BigInt((l as any).blockNumber)),
+  );
+  expect(keptBlocks.sort((x, y) => x - y)).toEqual([100, 250]);
+  expect(out.logs.map((l) => (l as any).topics[0]).sort()).toEqual([
+    '0xaaaa',
+    '0xbbbb',
+  ]); // the below-range B log at 100 is gone
+  expect(out.txs.map((t) => (t as any).hash).sort()).toEqual([
+    '0xtxA100',
+    '0xtxB250',
+  ]); // 0xtxB100 (parent of the dropped log) must not be inserted
+});
+
 // ── INV-5: trace ranking = index in the cmpTraceAddr-sorted full list ────────────────────────────────
 
 const arbTraceAddr = fc.array(fc.integer({ min: 0, max: 3 }), { maxLength: 4 });

portal/portal-assemble.test.ts

@@ -25,6 +25,7 @@ import type {
 import {
   isAddressFactory,
   isAddressMatched,
+  isLogFilterMatched,
   isTraceFilterMatched,
   isTransactionFilterMatched,
   isTransferFilterMatched,
@@ -105,6 +106,7 @@ export function rankTraces(traces: RawTrace[]): RankedTrace[] {
 }
 
 type Matchers = {
+  logMatched: (log: SyncLog, bn: number) => boolean;
   traceMatched: (frame: CallFrame, bn: number) => boolean;
   txFilterMatched: (tx: SyncTransaction, bn: number) => boolean;
 };
@@ -125,6 +127,19 @@ const buildMatchers = (
       childAddresses: childAddresses.get((filterAddr as { id: string }).id)!,
     });
   return {
+    // Log re-match (wave 4 review): the Portal's server-side log filter is the MERGED request set with no
+    // per-filter fromBlock/toBlock and no per-child creation floor, so it legitimately returns (a) a
+    // factory child's logs from BEFORE the child's creation block and (b) a bounded filter's logs outside
+    // its own range when another filter's chunk fetched them. Upstream's RPC path re-matches every log
+    // (isLogFilterMatched + isAddressMatched) before insertion; skipping that here stored/returned rows
+    // the RPC path excludes — an INV-6 store divergence. buildEvents re-filters, so no wrong events fired,
+    // but the sync store must be byte-identical across both paths.
+    logMatched: (log, bn) =>
+      spec.logFilters.some(
+        (f) =>
+          isLogFilterMatched({ filter: f, log }) &&
+          factoryAddrOk(f.address, log.address, bn),
+      ),
     traceMatched: (frame, bn) => {
       const blk = { number: BigInt(bn) } as never;
       for (const f of spec.transferFilters)
@@ -165,6 +180,24 @@ const buildMatchers = (
   };
 };
 
+/**
+ * Stream-seam parity for `runStreams`' needed-field growth check (wave-4 log re-match). The Portal's
+ * server-side log filter over-returns rows that assembly then DROPS — a factory child's PRE-CREATION logs
+ * and a bounded filter's OUT-OF-RANGE logs (see `logMatched` above) — so counting raw returned logs toward
+ * "matched data this call added" would arm the needed-field fatal for data the indexer never keeps → G1
+ * evict → crash-loop. This returns a predicate over RAW logs that mirrors `assembleRange`'s per-log
+ * re-match EXACTLY (same `buildMatchers` + `toSyncLog`, over the same discovery-complete `childAddresses`),
+ * so the seam count and assembly agree by construction.
+ */
+export const buildRawLogMatcher = (
+  spec: FetchSpec,
+  childAddresses: ChildAddresses,
+): ((raw: RawLog, hdr: RawHeader, bn: number) => boolean) => {
+  const { logMatched } = buildMatchers(spec, childAddresses);
+
+  return (raw, hdr, bn) => logMatched(toSyncLog(raw, hdr), bn);
+};
+
 /** Per-chunk trace assembly: full-tree ranking then client-side filtering (INV-5). `onReceipt` (FIX 4)
  * is invoked with each MATCHED trace's parent transaction so the caller can emit a receipt when a
  * trace/transfer filter has `hasTransactionReceipt` — the log/tx branches never see these txs. */
@@ -260,20 +293,34 @@ export function assembleRange(
   // INV-2 is enforced BY CONSTRUCTION here: every emitting branch below sits behind the single
   // `inRange` predicate (there is deliberately no redundant per-row assert — it could never fire),
   // and the exactness property is proven against a brute-force model in portal-assemble.test.ts.
+  // Each log is re-matched against the ACTUAL log filters (see logMatched above); a dropped log's
+  // parent tx/block must not ride in either, so txs are keyed to the KEPT logs' transactionHash.
   for (const cd of chunks)
     for (const [bn, hdr] of cd.headers) {
       if (!inRange(bn)) continue;
-      const logs = cd.logs.get(bn) ?? [];
-      if (logs.length) {
-        blocksByNumber.set(bn, toSyncBlockHeader(hdr));
-        for (const raw of logs) syncLogs.push(toSyncLog(raw, hdr));
-        for (const tx of cd.txs.get(bn) ?? [])
-          if (tx.hash && !seenTx.has(tx.hash)) {
-            seenTx.add(tx.hash);
-            syncTxs.push(toSyncTransaction(tx, hdr));
-            if (spec.needReceipts) pushReceipt(tx, hdr);
-          }
+
+      const rawLogs = cd.logs.get(bn) ?? [];
+      if (rawLogs.length === 0) continue;
+
+      const keptTxHashes = new Set<string>();
+      let keptAny = false;
+      for (const raw of rawLogs) {
+        const log = toSyncLog(raw, hdr);
+        if (!matchers.logMatched(log, bn)) continue;
+
+        keptAny = true;
+        syncLogs.push(log);
+        if (raw.transactionHash) keptTxHashes.add(raw.transactionHash);
       }
+      if (!keptAny) continue;
+
+      blocksByNumber.set(bn, toSyncBlockHeader(hdr));
+      for (const tx of cd.txs.get(bn) ?? [])
+        if (tx.hash && keptTxHashes.has(tx.hash) && !seenTx.has(tx.hash)) {
+          seenTx.add(tx.hash);
+          syncTxs.push(toSyncTransaction(tx, hdr));
+          if (spec.needReceipts) pushReceipt(tx, hdr);
+        }
     }
 
   // block-interval sources: ensure each matched block is in the blocks table (cd.blockHeaders already

portal/portal-assemble.ts

@@ -212,6 +212,46 @@ test('field degradation: a NEEDED field (logsBloom) missing is recorded in neede
   expect([...neededMissing]).toEqual(['transaction.logsBloom (logs_bloom)']);
 });
 
+test('field degradation: the SAME column missing in TWO tables drops each independently (bound keyed per-FIELD, not per-column — wave 4)', async () => {
+  // A dataset lacking logs_bloom in BOTH blocks and transactions (the Monad-style shape FIX 3 exists
+  // for; every query projects block.logsBloom, and the tx query always projects RECEIPT_FIELDS'
+  // logsBloom). The "dropping didn't help → real error" bound used to key on the BARE column name, so
+  // the second table's 400 hit the bound and threw a fatal PortalSchemaFieldError → the chunk rejected,
+  // was G1-evicted, refetched, and failed identically: a deterministic crash-loop. Keying per
+  // table-qualified field drops each table's column independently while keeping the bound for a field
+  // whose OWN drop didn't fix its own 400.
+  const bodies: any[] = [];
+  const client = mk({
+    fetchImpl: (async (_u: string, init: any) => {
+      const q = JSON.parse(init.body);
+      bodies.push(q);
+      if (q.fields?.block?.logsBloom !== undefined)
+        return badRes("column 'logs_bloom' is not found in 'blocks'");
+      if (q.fields?.transaction?.logsBloom !== undefined)
+        return badRes("column 'logs_bloom' is not found in 'transactions'");
+      return ndjsonRes([{ header: { number: 10 } }]);
+    }) as any,
+  });
+  const neededMissing = new Set<string>();
+  const q: PortalQuery = {
+    type: 'evm',
+    fields: {
+      block: { logsBloom: true, number: true },
+      transaction: { logsBloom: true, hash: true },
+    },
+  };
+  const out = await collect(client.stream(q, 0, 10, { neededMissing })); // no throw — the old code threw here
+  expect(out).toHaveLength(1);
+  const last = bodies[bodies.length - 1];
+  expect(last.fields.block.logsBloom).toBeUndefined(); // first drop
+  expect(last.fields.transaction.logsBloom).toBeUndefined(); // second drop (used to be the fatal)
+  expect(last.fields.block.number).toBe(true); // other fields kept (locality)
+  expect(last.fields.transaction.hash).toBe(true);
+  // both are NEEDED (non-droppable) fields → each recorded for the FIX-3 seam to judge, not silent
+  expect(neededMissing.has('block.logsBloom (logs_bloom)')).toBe(true);
+  expect(neededMissing.has('transaction.logsBloom (logs_bloom)')).toBe(true);
+});
+
 test('dataset-start 400 clamps the cursor forward, not a crash', async () => {
   let clampedFrom = -1;
   const client = mk({