fix: keep every Portal request under MAX_RAW_QUERY_SIZE (256KB) — merge log filters + fail loud

Root cause (from sqd-network transport/src/protocol.rs): the Portal 400 "Query is too large" checks
the raw request BODY size against MAX_RAW_QUERY_SIZE = 256*1024, NOT range/data — so my earlier
range-bisect was treating the wrong variable (reverted). The 24-event EVault superset blew the body
up because ponder emits one filter per event, and the fork repeated the full child-address list once
per event (24 × ~265 addrs ≈ 270KB).

- mergeLogRequests(): fold log requests sharing an address set + topic1..3 into one, unioning topic0
  → identical results, ~N× smaller body. Euler's worst (eth: 897 children × 24 topics) ≈ 41KB, 6× under.
- Uniform proactive guard in fetchBatch (the single POST choke point) covers EVERY request type
  (logs/traces/txs/discovery): a body over 256KB throws an explicit error naming the size + the driver
  (log vs tx from/to address counts) instead of an opaque Portal 400. tx from/to sets can't be safely
  address-split, so those fail loud by design.
- 2 regression tests (merge folds N filters → 1; over-limit body fails loud). 24/24.

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

Author: dzhelezov

harness/euler-multichain/README.md

@@ -6,9 +6,15 @@ is the production shape of an Euler indexer and a stress test of the `@subsquid/
 shared Portal endpoint, one database, fifteen chains backfilling concurrently.
 
 ## What it indexes
-- The `eVaultFactory` (`GenericFactory` — `ProxyCreated` → child EVaults) per chain.
-- The 6 EVault events — `Deposit, Withdraw, Borrow, Repay, Liquidate, VaultStatus` — log-based,
-  exactly how the real euler-subgraph indexes (no receipts; `event.transaction.hash` for IDs).
+- The `eVaultFactory` (`GenericFactory` — `ProxyCreated` → child EVaults) per chain, with start blocks
+  taken from **Euler's own subgraph config** (`euler-xyz/euler-subgraph`) and every factory address
+  cross-verified against `euler-xyz/euler-interfaces` `CoreAddresses.json`.
+- The **full 24-event EVault superset** (`Deposit/Withdraw/Borrow/Repay/Liquidate/Transfer/Approval/
+  VaultStatus/InterestAccrued/DebtSocialized/PullDebt/ConvertFees/BalanceForwarderStatus/EVaultCreated`
+  + all `GovSet*`), from the euler-interfaces ABI. Euler runs **both** ponder and subgraphs in prod and
+  the public subgraph indexes only a subset (`Transfer/Borrow/Repay`), so a superset is the faithful
+  apples-to-apples target. Log-based, no receipts; `event.transaction.hash` for IDs.
+  Vault discovery is cross-checked against Euler's live Goldsky subgraph per chain (see REPORT.md).
 
 ## Chains (15)
 `ethereum · binance · unichain · polygon · monad · sonic · tac · hyperliquid · base · plasma ·

portal/portal.test.ts

@@ -376,3 +376,55 @@ test("regression (C6): deep matched trace is stored at its FULL-tree pre-order i
     expect((inserted[0].trace.trace.to as string)?.toLowerCase()).toBe(TARGET);
   } finally { srv.close(); delete process.env.PORTAL_FINALIZED_HEAD; }
 });
+
+test("merge: N same-address event filters collapse to ONE log request (unioned topic0) — keeps body small", async () => {
+  // Ponder emits one filter per event; a 24-event EVault would otherwise repeat the child-address
+  // list 24× in one body and blow past MAX_RAW_QUERY_SIZE. mergeLogRequests must fold them into one.
+  const ADDR = "0x" + "cc".repeat(20);
+  const topic0s = Array.from({ length: 6 }, (_, i) => "0x" + (i + 1).toString(16).padStart(64, "0"));
+  const dataQueries: any[] = [];
+  const srv = http.createServer((req, res) => {
+    let body = ""; req.on("data", (c) => { body += c; });
+    req.on("end", () => { const q = body ? JSON.parse(body) : {}; if (q.logs) dataQueries.push(q); res.writeHead(204).end(); });
+  });
+  const p = await new Promise<number>((r) => srv.listen(0, () => r((srv.address() as AddressInfo).port)));
+  try {
+    const filters = topic0s.map((t, i) => ({
+      type: "log", chainId: 1, sourceId: `evault:e${i}`, address: ADDR, topic0: t,
+      topic1: null, topic2: null, topic3: null, fromBlock: 0, toBlock: 100, hasTransactionReceipt: false, include: [],
+    }));
+    const sync = createPortalHistoricalSync({
+      common: { logger: { debug() {}, info() {}, warn() {}, error() {}, trace() {} } },
+      chain: { id: 1, name: "mainnet", portal: `http://localhost:${p}` },
+      childAddresses: new Map(), eventCallbacks: filters.map((f) => ({ filter: f })),
+    } as any);
+    await sync.syncBlockRangeData({
+      interval: [0, 100], requiredIntervals: filters.map((f) => ({ interval: [0, 100], filter: f })),
+      requiredFactoryIntervals: [], syncStore: { insertLogs() {}, insertBlocks() {}, insertTransactions() {}, insertTransactionReceipts() {}, insertTraces() {} },
+    } as any);
+    const merged = dataQueries.find((q) => (q.logs?.[0]?.topic0?.length ?? 0) > 1);
+    expect(merged).toBeDefined();
+    expect(merged.logs).toHaveLength(1); // ONE request, not 6
+    expect(new Set(merged.logs[0].topic0.map((x: string) => x.toLowerCase())).size).toBe(6); // all selectors unioned
+  } finally { srv.close(); }
+});
+
+test("guard: an over-limit request body fails loud with the explicit size driver (never a silent Portal 400)", async () => {
+  // 12k filter addresses → batched bodies sum > 256KB MAX_RAW_QUERY_SIZE. The proactive guard must
+  // throw a clear, actionable error BEFORE the POST, not let the Portal reject it opaquely.
+  const addrs = Array.from({ length: 12000 }, (_, i) => "0x" + i.toString(16).padStart(40, "0"));
+  const srv = http.createServer((req, res) => { let b = ""; req.on("data", (c) => { b += c; }); req.on("end", () => res.writeHead(204).end()); });
+  const p = await new Promise<number>((r) => srv.listen(0, () => r((srv.address() as AddressInfo).port)));
+  try {
+    const filter = { type: "log", chainId: 1, sourceId: "big", address: addrs, topic0: "0x" + "11".repeat(32), topic1: null, topic2: null, topic3: null, fromBlock: 0, toBlock: 100, hasTransactionReceipt: false, include: [] };
+    const sync = createPortalHistoricalSync({
+      common: { logger: { debug() {}, info() {}, warn() {}, error() {}, trace() {} } },
+      chain: { id: 1, name: "mainnet", portal: `http://localhost:${p}` },
+      childAddresses: new Map(), eventCallbacks: [{ filter }],
+    } as any);
+    await expect(sync.syncBlockRangeData({
+      interval: [0, 100], requiredIntervals: [{ interval: [0, 100], filter }], requiredFactoryIntervals: [],
+      syncStore: { insertLogs() {}, insertBlocks() {}, insertTransactions() {}, insertTransactionReceipts() {}, insertTraces() {} },
+    } as any)).rejects.toThrow(/exceeds MAX_RAW_QUERY_SIZE/);
+  } finally { srv.close(); }
+});

portal/portal.ts

@@ -72,6 +72,11 @@ type ChunkData = {
 };
 
 const PORTAL_MAX_ADDRESSES = 1000;
+// Portal rejects any request whose raw body exceeds this (sqd-network transport/src/protocol.rs:
+// `MAX_RAW_QUERY_SIZE = 256 * 1024`) with 400 "Query is too large". The body is dominated by filter
+// address lists (factory children in log/tx filters). We keep under it by merging per-event log
+// filters + batching addresses; a body that still overflows fails loud (see fetchBatch).
+const MAX_RAW_QUERY_SIZE = 256 * 1024;
 const CHUNK_BLOCKS = Number(process.env.PORTAL_CHUNK_BLOCKS ?? 500_000);
 const READAHEAD = Number(process.env.PORTAL_READAHEAD ?? 6);
 // The Portal fans ONE stream request out across up to `buffer_size` chunk-workers concurrently
@@ -217,6 +222,20 @@ export const createPortalHistoricalSync = (
 
   // one POST+drain; returns blocks or "done" (204); throws (with .retryAfterMs on 503-class).
   async function fetchBatch(body: string, cursor: number): Promise<{ blocks: { header: RawHeader; logs?: any[]; transactions?: any[]; traces?: any[] }[]; last: number } | "done"> {
+    // Proactive, uniform size guard — covers EVERY request type (logs/traces/txs/discovery) at the one
+    // POST choke point. A body over MAX_RAW_QUERY_SIZE would 400; surface it explicitly with the real
+    // driver instead. Euler's worst (eth: 897 children × 24 topics) is ~41KB, so this never fires here;
+    // it protects indexers with pathological filtered-address counts (esp. unbatched tx from/to sets).
+    if (body.length > MAX_RAW_QUERY_SIZE) {
+      const q = (() => { try { return JSON.parse(body); } catch { return {}; } })();
+      const nLog = (q.logs ?? []).reduce((s: number, r: any) => s + (r.address?.length ?? 0), 0);
+      const nTx = (q.transactions ?? []).reduce((s: number, r: any) => s + (r.from?.length ?? 0) + (r.to?.length ?? 0), 0);
+      throw new Error(
+        `Portal request body ${(body.length / 1024).toFixed(1)}KB exceeds MAX_RAW_QUERY_SIZE ${MAX_RAW_QUERY_SIZE / 1024}KB @ ${cursor}. ` +
+        `Filter addresses in this request: ${nLog} log + ${nTx} tx(from/to). ` +
+        `Log filters are already merged+batched (PORTAL_MAX_ADDRESSES=${PORTAL_MAX_ADDRESSES}); if this is a tx filter, its from/to set is too large to fit one request and cannot be safely split — narrow the filter.`,
+      );
+    }
     const tAcq = Date.now(); await portalGate.acquire(); stats.gateWaitMs += Date.now() - tAcq; // gate-wait = concurrency back-pressure
     const tFetch = Date.now();
     stats.inflight++; stats.maxInflight = Math.max(stats.maxInflight, stats.inflight);
@@ -304,17 +323,15 @@ export const createPortalHistoricalSync = (
     while (cursor <= to) {
       let attempt = 0;
       let batch: Awaited<ReturnType<typeof fetchBatch>> | undefined;
-      let hi = to; // per-cursor upper bound; bisected down on "query too large", reset each advance
       while (batch === undefined) {
-        const body = JSON.stringify({ ...stripFields(query, dropped), fromBlock: cursor, toBlock: hi });
+        const body = JSON.stringify({ ...stripFields(query, dropped), fromBlock: cursor, toBlock: to });
         try { batch = await fetchBatch(body, cursor); }
         catch (err: any) {
           if (err?.tooLarge) {
-            // dense window exceeded the Portal's query estimate → halve it and retry. Continuation
-            // then covers [hi+1, to] as usual, so no data is skipped.
-            if (hi <= cursor) throw err; // a single block still too large ⇒ a real limit, surface it
-            hi = cursor + Math.floor((hi - cursor) / 2);
-            continue;
+            // Portal caps request BYTES (MAX_RAW_QUERY_SIZE), not range — so bisecting blocks can't
+            // help. mergeLogRequests already de-dups addresses across event filters; if a body still
+            // exceeds the cap the address batch itself is too big → fail loud with the actual lever.
+            throw new Error(`Portal query body exceeds MAX_RAW_QUERY_SIZE even after merging event filters — lower PORTAL_MAX_ADDRESSES (currently ${PORTAL_MAX_ADDRESSES}) to shrink the address batch. @ ${cursor}`);
           }
           if (err?.datasetStartsAt !== undefined) {
             // dataset begins after this chunk's start (doesn't reach genesis) → skip the missing
@@ -367,6 +384,23 @@ export const createPortalHistoricalSync = (
     return out;
   }
 
+  // Ponder emits ONE filter per event, so an N-event contract (e.g. the 24-event EVault) produces N
+  // log requests that each repeat the SAME (possibly large) child-address list with a different
+  // topic0. Concatenated into one query body they can exceed the Portal's raw query-size limit
+  // (400 "Query is too large" — it caps request BYTES, not range). Collapse requests that share the
+  // same address set + topic1..3 into one, unioning topic0 — identical result set, ~N× smaller body.
+  function mergeLogRequests(reqs: PortalLogRequest[]): PortalLogRequest[] {
+    const groups = new Map<string, PortalLogRequest>();
+    for (const r of reqs) {
+      const key = JSON.stringify([r.address ? [...r.address].sort() : null, r.topic1 ?? null, r.topic2 ?? null, r.topic3 ?? null]);
+      const g = groups.get(key);
+      if (!g) { groups.set(key, { ...r, topic0: r.topic0 ? [...new Set(r.topic0)] : undefined }); continue; }
+      if (g.topic0 === undefined || r.topic0 === undefined) g.topic0 = undefined; // one wants ALL topic0 → keep the broadest
+      else { const s = new Set(g.topic0); for (const t of r.topic0) s.add(t); g.topic0 = [...s]; }
+    }
+    return [...groups.values()];
+  }
+
   // FILTER/PROJECTION STRATEGY (max Portal leverage): every row filter is pushed to
   // Portal's native server-side filters — logs by address+topics (logRequestsFor),
   // traces by callTo/callFrom/callSighash (tracePortalRequests), account txs by from/to
@@ -504,7 +538,7 @@ export const createPortalHistoricalSync = (
       await ensureDiscoveredThrough(idx, factories); // correctness: children ≤ this chunk are known
       stats.dataChunks++;
       const [from, to] = chunkRange(idx);
-      const logRequests = filters.flatMap((f) => logRequestsFor(f)).map((r) => ({ ...r, transaction: true }));
+      const logRequests = mergeLogRequests(filters.flatMap((f) => logRequestsFor(f))).map((r) => ({ ...r, transaction: true }));
       const data: ChunkData = { headers: new Map(), logs: new Map(), txs: new Map(), traceBlocks: new Map(), blockHeaders: new Map(), txBlocks: new Map() };
       const neededMissing = new Set<string>(); // needed fields the dataset lacked on THIS chunk
       if (logRequests.length > 0) {