refactor(realtime): make the lib RPC-agnostic — no "Portal-backed RPC" concept

We test the RPC implementation through a generic interface the lib is blind to. Removed
portalRpc/portalRealtime from portal/realtime.ts — the realtime source is just an RPC (authenticated
RPCs are plain viem http with headers; nothing special). rpcRealtime() composes URL strings or viem
Transports into a latency-ranked fallback. The Euler eval config/probe use generic naming
(REALTIME_RPC_URL/KEY, "realtime-rpc") — the Portal-backed implementation detail stays out of the repo.
29 tests pass.

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

Author: dzhelezov

harness/euler-multichain/ponder.config.ts

@@ -9,44 +9,43 @@ import chainsData from "./chains.json";
 //
 // Historical backfill is ALWAYS the Portal. The realtime source is a config choice:
 //   • bench mode (default): bounded [deploy, head] — a clean fixed-range benchmark; RPC only for setup.
-//   • EULER_REALTIME=true:  unbounded → live, realtime served by the Portal-backed RPC — the service
-//     offered to clients (Portal-backed under the hood, designed for the recent tip).
+//   • EULER_REALTIME=true:  unbounded → live, realtime served by an authenticated RPC (URL + x-api-key
+//     from env). Ponder owns reorg handling and is agnostic to what the RPC is or how it's implemented.
 //
 // Portal base URLs are deployment config (from env); relative to a base the documented Portal API paths
 // apply — /datasets/<>, /metadata, … (see docs.sqd.dev). Public examples default to the public Portal.
-//   PORTAL_URL      = Portal base (default https://portal.sqd.dev — the public, rate-limited Portal)
-//   PORTAL_RPC_URL  = Portal-backed RPC base (from env; provisioned per client)
-// One config serves any client by swapping env. `rpc.subsquid.io` is a generic proxied RPC (like Alchemy)
-// — NOT the Portal-backed product — and is deliberately not used; keyless public RPCs (chains.json
-// `freeRpcs`) are only for setup / finality tail.
+//   PORTAL_URL        = Portal base (default https://portal.sqd.dev — the public, rate-limited Portal)
+//   REALTIME_RPC_URL  = authenticated realtime RPC base (with REALTIME_RPC_KEY, from env)
+// One config serves any deployment by swapping env. Keyless public RPCs (chains.json `freeRpcs`) are only
+// for setup / finality tail — excluded from the realtime path (403/timeout cascades stall it).
 //
-// No secrets/tenancy in the repo: PORTAL_API_KEY, PORTAL_URL, PORTAL_RPC_KEY, PORTAL_RPC_URL come from env.
+// No secrets/tenancy in the repo: PORTAL_API_KEY, PORTAL_URL, REALTIME_RPC_KEY, REALTIME_RPC_URL come from env.
 const proxyCreated = parseAbiItem(
   "event ProxyCreated(address indexed proxy, bool upgradeable, address implementation, bytes trailingData)",
 );
 
 type ChainRow = { id: number; name: string; ds: string; factory: string; head: number; deploy: number; sqdSlug: string | null; freeRpcs: string[] };
 const rows = chainsData as ChainRow[];
 const PORTAL_URL = process.env.PORTAL_URL ?? "https://portal.sqd.dev";
-const PORTAL_RPC_KEY = process.env.PORTAL_RPC_KEY;
-const PORTAL_RPC_URL = process.env.PORTAL_RPC_URL; // Portal-backed RPC base (from env; provisioned per client)
+const REALTIME_RPC_KEY = process.env.REALTIME_RPC_KEY;
+const REALTIME_RPC_URL = process.env.REALTIME_RPC_URL; // realtime RPC base (from env; provisioned per client)
 const REALTIME = process.env.EULER_REALTIME === "true";
 
-// Chains served by the Portal-backed RPC. Realtime uses it (alone) on these.
-const PORTAL_RPC_CHAINS = new Set([1, 42161, 8453, 43114, 137, 56, 9745, 143]);
-const portalRpc = (chainId: number) =>
-  http(`${PORTAL_RPC_URL}/${chainId}`, { fetchOptions: { headers: { "x-api-key": PORTAL_RPC_KEY ?? "" } } });
+// Chains served by the realtime RPC. Realtime uses it (alone) on these.
+const REALTIME_CHAINS = new Set([1, 42161, 8453, 43114, 137, 56, 9745, 143]);
+const realtimeRpc = (chainId: number) =>
+  http(`${REALTIME_RPC_URL}/${chainId}`, { fetchOptions: { headers: { "x-api-key": REALTIME_RPC_KEY ?? "" } } });
 
-// realtime → the Portal-backed RPC ALONE (the endpoint under evaluation): no generic proxy, no flaky
+// realtime → the realtime RPC ALONE (the endpoint under evaluation): no generic proxy, no flaky
 // keyless public fallback (their 403/timeout cascades stall the single-thread). bench → public RPCs only.
 const rpcFor = (c: ChainRow) =>
-  REALTIME && PORTAL_RPC_KEY && PORTAL_RPC_URL && PORTAL_RPC_CHAINS.has(c.id) ? portalRpc(c.id) : c.freeRpcs;
+  REALTIME && REALTIME_RPC_KEY && REALTIME_RPC_URL && REALTIME_CHAINS.has(c.id) ? realtimeRpc(c.id) : c.freeRpcs;
 
 // optional subset: EULER_CHAINS=ethereum,base limits the run. In realtime mode with no explicit subset,
-// default to the Portal-backed-RPC chains (the 8 that have a realtime source).
+// default to the realtime-RPC chains (the 8 that have a realtime source).
 const only = (process.env.EULER_CHAINS ?? "").split(",").map((s) => s.trim()).filter(Boolean);
 let active = only.length ? rows.filter((c) => only.includes(c.name)) : rows;
-if (REALTIME && !only.length) active = rows.filter((c) => PORTAL_RPC_CHAINS.has(c.id));
+if (REALTIME && !only.length) active = rows.filter((c) => REALTIME_CHAINS.has(c.id));
 
 export default createConfig({
   // Postgres (production shape, separate process → own memory) when DATABASE_URL is set; else pglite.
@@ -61,7 +60,7 @@ export default createConfig({
       chain: Object.fromEntries(
         active.map((c) => [
           c.name,
-          // realtime: no endBlock → backfill to finalized head, then go live on the Portal-backed RPC.
+          // realtime: no endBlock → backfill to finalized head, then go live on the realtime RPC.
           { address: factory({ address: c.factory as `0x${string}`, event: proxyCreated, parameter: "proxy" }), startBlock: c.deploy, ...(REALTIME ? {} : { endBlock: c.head }) },
         ]),
       ),

harness/euler-multichain/probe.mjs

@@ -3,28 +3,28 @@
 // responsiveness. Standalone (no imports beyond node) so it survives an overnight run alongside ponder.
 // Mirrors portal/realtime.ts (unit-tested). Writes PROBE_METRICS_FILE (rolling JSON) + a .log trail.
 //
-//   PORTAL_RPC_KEY=... [SQD_RPC_KEY=...] [PROBE_INTERVAL_MS=15000] node probe.mjs
+//   REALTIME_RPC_KEY=... [SQD_RPC_KEY=...] [PROBE_INTERVAL_MS=15000] node probe.mjs
 import fs from "node:fs";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 
 const dir = path.dirname(fileURLToPath(import.meta.url));
 const chains = JSON.parse(fs.readFileSync(path.join(dir, "chains.json"), "utf8"));
-const KEY = process.env.PORTAL_RPC_KEY;
-const RPC_URL = process.env.PORTAL_RPC_URL; // Portal-backed RPC base (from env; provisioned per client)
-const PORTAL_RPC_CHAINS = new Set([1, 42161, 8453, 43114, 137, 56, 9745, 143]);
+const KEY = process.env.REALTIME_RPC_KEY;
+const RPC_URL = process.env.REALTIME_RPC_URL; // realtime RPC base (from env; provisioned per client)
+const REALTIME_CHAINS = new Set([1, 42161, 8453, 43114, 137, 56, 9745, 143]);
 const METRICS = process.env.PROBE_METRICS_FILE || path.join(dir, "probe-metrics.json");
 const INTERVAL = Number(process.env.PROBE_INTERVAL_MS || 15000);
 const WINDOW = Number(process.env.PROBE_WINDOW || 40); // rolling ticks
 
-if (!KEY || !RPC_URL) { console.error("PORTAL_RPC_KEY + PORTAL_RPC_URL required (the Portal-backed RPC base + x-api-key)"); process.exit(1); }
+if (!KEY || !RPC_URL) { console.error("REALTIME_RPC_KEY + REALTIME_RPC_URL required (the realtime RPC base + x-api-key)"); process.exit(1); }
 
-// per chain, probe the Portal-backed RPC (the product) alongside a keyless public RPC (freshness baseline).
-// rpc.subsquid.io is a generic proxy, not the Portal-backed product — deliberately excluded.
-const targets = chains.filter((c) => PORTAL_RPC_CHAINS.has(c.id)).map((c) => ({
+// per chain, probe the realtime RPC alongside a keyless public RPC (freshness baseline).
+// rpc.subsquid.io is a generic proxy — deliberately excluded.
+const targets = chains.filter((c) => REALTIME_CHAINS.has(c.id)).map((c) => ({
   chainId: c.id, chain: c.name,
   endpoints: [
-    { name: "portal-rpc", url: `${RPC_URL}/${c.id}`, headers: { "x-api-key": KEY } },
+    { name: "realtime-rpc", url: `${RPC_URL}/${c.id}`, headers: { "x-api-key": KEY } },
     ...(c.freeRpcs || []).slice(0, 1).map((u, i) => ({ name: `public${i}`, url: u })),
   ],
 }));
@@ -67,8 +67,8 @@ async function tick() {
       avgLagBlk: lags.length ? Math.round(lags.reduce((a, b) => a + b, 0) / lags.length) : null, lastError: [...ss].reverse().find((s) => s.error)?.error };
   }
   try { fs.writeFileSync(METRICS, JSON.stringify({ ts: new Date().toISOString(), intervalMs: INTERVAL, endpoints: summary }, null, 2)); } catch { /* best-effort */ }
-  const line = Object.values(summary).filter((v) => v.endpoint === "portal-rpc").map((v) => `${v.chain} p50=${v.latP50} p95=${v.latP95} ok=${(v.okRate * 100).toFixed(0)}% lag=${v.avgLagBlk ?? "?"}`).join(" | ");
-  try { fs.appendFileSync(METRICS + ".log", `${new Date().toISOString()} [portal-rpc] ${line}\n`); } catch { /* best-effort */ }
+  const line = Object.values(summary).filter((v) => v.endpoint === "realtime-rpc").map((v) => `${v.chain} p50=${v.latP50} p95=${v.latP95} ok=${(v.okRate * 100).toFixed(0)}% lag=${v.avgLagBlk ?? "?"}`).join(" | ");
+  try { fs.appendFileSync(METRICS + ".log", `${new Date().toISOString()} [realtime-rpc] ${line}\n`); } catch { /* best-effort */ }
 }
 
 console.log(`probe: ${targets.length} chains × endpoints, interval ${INTERVAL}ms -> ${METRICS}`);

portal/realtime.test.ts

@@ -1,26 +1,12 @@
 import { expect, test } from "vitest";
-import { portalRpc, portalRealtime, rpcRealtime, freshnessLag, summarize, probeOnce, type ProbeSample } from "./realtime.js";
+import { http } from "viem";
+import { rpcRealtime, freshnessLag, summarize, probeOnce, type ProbeSample } from "./realtime.js";
 
-test("portalRpc: appends chainId to the client-specific base + x-api-key header", () => {
-  const t = portalRpc(42161, "prt_test", { baseUrl: "https://portal.example/rpc" });
-  expect(typeof t).toBe("function");
-  // viem http transport exposes its url on the instantiated value
-  const inst: any = t({} as any);
-  expect(inst.value?.url).toBe("https://portal.example/rpc/42161");
-  const t2: any = portalRpc(1, "k", { baseUrl: "http://local/rpc" })({} as any);
-  expect(t2.value?.url).toBe("http://local/rpc/1");
-});
-
-test("portalRpc: throws without a base (never hardcodes a client domain)", () => {
-  const saved = process.env.PORTAL_RPC_URL;
-  delete process.env.PORTAL_RPC_URL;
-  expect(() => portalRpc(1, "k")).toThrow(/baseUrl|PORTAL_RPC_URL/);
-  if (saved !== undefined) process.env.PORTAL_RPC_URL = saved;
-});
-
-test("portalRealtime / rpcRealtime return composed viem transports", () => {
-  expect(typeof portalRealtime(1, "k", ["http://a", "http://b"], { baseUrl: "https://portal.example/rpc" })).toBe("function");
-  expect(typeof rpcRealtime(["http://a", "http://b"])).toBe("function");
+test("rpcRealtime: composes URL strings and authed transports into a latency-ranked fallback", () => {
+  // an authenticated RPC is just a viem transport — the lib is agnostic to what backs it
+  const authed = http("http://b", { fetchOptions: { headers: { "x-api-key": "k" } } });
+  expect(typeof rpcRealtime(["http://a", authed])).toBe("function");
+  expect(typeof rpcRealtime(["http://a", "http://b"], { rank: false })).toBe("function");
 });
 
 test("freshnessLag: blocks behind the freshest endpoint, per chain", () => {

portal/realtime.ts

@@ -1,50 +1,30 @@
 /**
  * Realtime source helpers + endpoint instrumentation.
  *
- * Ponder owns realtime and reorg handling natively (RealtimeSync: parent-hash tracking + rollback to
- * the common ancestor, bounded by the finalized block). The fork only swaps the HISTORICAL sync to
- * the Portal. So "which realtime source" is purely a `rpc` config choice — ponder polls whatever
- * transport(s) you give it and reorg-handles the result identically:
+ * Ponder owns realtime and reorg handling natively (RealtimeSync: parent-hash tracking + rollback to the
+ * common ancestor, bounded by the finalized block). The fork only swaps the HISTORICAL sync to the Portal.
+ * So the realtime source is purely a `rpc` config choice — ponder polls whatever transport(s) you give it
+ * and reorg-handles the result identically. The lib is RPC-agnostic: an authenticated RPC (an x-api-key
+ * header, a keyed proxy, …) is just `http(url, { fetchOptions: { headers } })` — nothing special.
  *
- *   import { createConfig } from "@subsquid/ponder";
  *   import { http, fallback } from "viem";
- *   import { portalRpc } from "@subsquid/ponder/realtime";
- *
- *   // (A) Portal realtime, RPC fallback — Portal-backed RPC leads; public RPCs cover downtime/lag.
- *   //     base URL is provisioned per client → set PORTAL_RPC_URL (or pass { baseUrl }).
- *   rpc: fallback([ portalRpc(1, process.env.PORTAL_RPC_KEY!), http(process.env.RPC_1), http(process.env.RPC_1B) ]),
- *
- *   // (B) RPC(s) realtime — a plain Ponder-style list, or latency-ranked for fastest-tip.
+ *   // a plain list (Ponder-style), or latency-ranked for fastest-tip:
  *   rpc: [process.env.RPC_1!, process.env.RPC_1B!],
  *   rpc: fallback([http(process.env.RPC_1!), http(process.env.RPC_1B!)], { rank: true }),
+ *   // an authenticated RPC + fallback:
+ *   rpc: fallback([ http(process.env.RPC_URL!, { fetchOptions: { headers: { "x-api-key": process.env.RPC_KEY! } } }), http(process.env.RPC_1!) ]),
  *
- * `fallback` gives smooth failover; `{ rank: true }` probes latency and routes to the fastest, so the
- * freshest tip lands. Both (A) and (B) get ponder's reorg safety for free.
+ * `fallback` gives smooth failover; `{ rank: true }` probes latency and routes to the fastest tip. Either
+ * way ponder's reorg safety comes for free.
  */
 import { http, fallback, type Transport } from "viem";
 
 /**
- * Portal-backed EVM RPC transport (auth via `x-api-key`). The base URL is provisioned per client, so
- * pass `opts.baseUrl` or set `PORTAL_RPC_URL` — never hardcode a client domain.
+ * Compose realtime RPC(s) into a latency-ranked fallback so the fastest tip lands. Accepts URL strings or
+ * viem Transports — pass a Transport for an authenticated RPC: `http(url, { fetchOptions: { headers } })`.
  */
-export function portalRpc(chainId: number, apiKey: string, opts?: { baseUrl?: string; timeout?: number; retryCount?: number }): Transport {
-  const base = opts?.baseUrl ?? process.env.PORTAL_RPC_URL;
-  if (!base) throw new Error("portalRpc: pass opts.baseUrl or set PORTAL_RPC_URL (the Portal-backed RPC base, provisioned per client)");
-  return http(`${base}/${chainId}`, {
-    fetchOptions: { headers: { "x-api-key": apiKey } },
-    timeout: opts?.timeout ?? 10_000,
-    retryCount: opts?.retryCount ?? 2,
-  });
-}
-
-/** Portal realtime with RPC fallback(s): Portal-backed RPC preferred, public RPCs as fallback. */
-export function portalRealtime(chainId: number, apiKey: string, fallbackRpcs: string[] = [], opts?: { baseUrl?: string; rank?: boolean }): Transport {
-  return fallback([portalRpc(chainId, apiKey, { baseUrl: opts?.baseUrl }), ...fallbackRpcs.map((u) => http(u))], { rank: opts?.rank ?? false });
-}
-
-/** RPC(s)-only realtime: a list of RPC URLs, latency-ranked so the fastest tip lands. */
-export function rpcRealtime(rpcs: string[], opts?: { rank?: boolean }): Transport {
-  return fallback(rpcs.map((u) => http(u)), { rank: opts?.rank ?? true });
+export function rpcRealtime(rpcs: Array<string | Transport>, opts?: { rank?: boolean }): Transport {
+  return fallback(rpcs.map((r) => (typeof r === "string" ? http(r) : r)), { rank: opts?.rank ?? true });
 }
 
 // ─────────────────────────────── endpoint instrumentation ───────────────────────────────