|
| 1 | +/** |
| 2 | + * Portal-native realtime — serve `[portal-finalized-head → tip]` from the Portal `/stream` (fork-aware |
| 3 | + * hot-blocks) instead of RPC. |
| 4 | + * |
| 5 | + * WHY: in realtime mode the fork currently backfills `[deploy → portal-finalized-head]` from the Portal, |
| 6 | + * then fetches `[portal-finalized-head → chain-finalized-head]` over RPC (the "finality gap") and runs |
| 7 | + * ponder's RPC RealtimeSync for the tip. For dense chains under multichain load that RPC gap-fetch stalls |
| 8 | + * — ponder's single-thread global ordering blocks on the slowest chain's gap. The Portal already holds |
| 9 | + * these hot-blocks; streaming them removes the RPC gap at its source. |
| 10 | + * |
| 11 | + * HOW: `/stream` with `includeAllBlocks:true` yields EVERY block header (a consecutive parentHash chain, |
| 12 | + * for reorg tracking) plus the filtered euler logs, in one request. We reconcile each block into a local |
| 13 | + * unfinalized chain and emit ponder's own `RealtimeSyncEvent`s (`block` / `reorg` / `finalize`) reusing |
| 14 | + * `portal-transform` — so the downstream indexing + checkpointing in runtime/realtime.ts is UNCHANGED. |
| 15 | + * |
| 16 | + * The reorg/finalize reconciliation is pure + unit-tested; the `/stream` read and `/finalized-head` poll |
| 17 | + * are the I/O shell. |
| 18 | + */ |
| 19 | +import type { SyncBlockHeader, SyncLog } from "@/internal/types.js"; |
| 20 | +import { hexToNumber, type Hex } from "viem"; |
| 21 | +import { hx, toSyncBlockHeader, toSyncLog, type RawHeader } from "./portal-transform.js"; |
| 22 | + |
| 23 | +// ─────────────────────────────── pure reorg / finalize core (unit-tested) ─────────────────────────────── |
| 24 | + |
| 25 | +/** The minimum a block needs for chain reconciliation. `number` is a decimal string (Portal-native). */ |
| 26 | +export type Light = { number: number; hash: string; parentHash: string; timestamp: number }; |
| 27 | + |
| 28 | +export type Reconcile = |
| 29 | + | { kind: "append" } // extends the tip (normal case) |
| 30 | + | { kind: "duplicate" } // already the tip (idempotent re-delivery) |
| 31 | + | { kind: "reorg"; commonAncestor: Light; reorgedBlocks: Light[] } // forks off an earlier block |
| 32 | + | { kind: "gap" }; // parent is unknown (beyond our window / a skipped block) → caller must re-sync |
| 33 | + |
| 34 | +/** |
| 35 | + * Reconcile a newly-streamed block against the local unfinalized chain (oldest→newest, linked |
| 36 | + * parentHash→hash). Pure. With `includeAllBlocks` the happy path is always `append`; a reorg re-streams |
| 37 | + * from the fork point, so the new block's `parentHash` matches an EARLIER block — everything after that |
| 38 | + * common ancestor is reorged. |
| 39 | + */ |
| 40 | +export function reconcile(unfinalized: Light[], next: Light): Reconcile { |
| 41 | + if (unfinalized.length === 0) return { kind: "append" }; |
| 42 | + const tip = unfinalized[unfinalized.length - 1]!; |
| 43 | + if (next.hash === tip.hash) return { kind: "duplicate" }; |
| 44 | + if (next.parentHash === tip.hash) return { kind: "append" }; |
| 45 | + const idx = unfinalized.findIndex((b) => b.hash === next.parentHash); |
| 46 | + if (idx === -1) return { kind: "gap" }; |
| 47 | + return { kind: "reorg", commonAncestor: unfinalized[idx]!, reorgedBlocks: unfinalized.slice(idx + 1) }; |
| 48 | +} |
| 49 | + |
| 50 | +/** Split the unfinalized chain at a newly-finalized block number. Pure. */ |
| 51 | +export function takeFinalized( |
| 52 | + unfinalized: Light[], |
| 53 | + finalizedNumber: number, |
| 54 | +): { finalizedTip: Light | undefined; remaining: Light[] } { |
| 55 | + let finalizedTip: Light | undefined; |
| 56 | + const remaining: Light[] = []; |
| 57 | + for (const b of unfinalized) { |
| 58 | + if (b.number <= finalizedNumber) finalizedTip = b; |
| 59 | + else remaining.push(b); |
| 60 | + } |
| 61 | + return { finalizedTip, remaining }; |
| 62 | +} |
| 63 | + |
| 64 | +export const toLight = (h: RawHeader): Light => ({ |
| 65 | + number: h.number, |
| 66 | + hash: h.hash, |
| 67 | + parentHash: h.parentHash, |
| 68 | + timestamp: h.timestamp, |
| 69 | +}); |
| 70 | + |
| 71 | +// ─────────────────────────────── /stream I/O shell ─────────────────────────────── |
| 72 | + |
| 73 | +export type PortalRealtimeArgs = { |
| 74 | + portalUrl: string; |
| 75 | + headers: Record<string, string>; |
| 76 | + /** first block to stream (exclusive of already-indexed finalized head): syncProgress.finalized.number + 1 */ |
| 77 | + fromBlock: number; |
| 78 | + /** euler log filters (address/topics), already merged — passed straight into the Portal query */ |
| 79 | + logs: Array<Record<string, unknown>>; |
| 80 | + /** the block header fields ponder needs */ |
| 81 | + blockFields?: Record<string, boolean>; |
| 82 | + logFields?: Record<string, boolean>; |
| 83 | + signal?: AbortSignal; |
| 84 | + fetchImpl?: typeof fetch; |
| 85 | +}; |
| 86 | + |
| 87 | +/** |
| 88 | + * Stream fork-aware hot-blocks with `includeAllBlocks` (every header + filtered logs). Yields raw Portal |
| 89 | + * batch objects `{ header, logs }`. Re-opens the stream from the last seen block on disconnect so it runs |
| 90 | + * continuously; a reorg simply re-streams from the fork point (reconcile() detects it). |
| 91 | + */ |
| 92 | +export async function* streamHotBlocks( |
| 93 | + args: PortalRealtimeArgs, |
| 94 | +): AsyncGenerator<{ header: RawHeader; logs: any[] }> { |
| 95 | + const fetchImpl = args.fetchImpl ?? fetch; |
| 96 | + let cursor = args.fromBlock; |
| 97 | + for (;;) { |
| 98 | + if (args.signal?.aborted) return; |
| 99 | + const body = JSON.stringify({ |
| 100 | + type: "evm", |
| 101 | + fromBlock: cursor, |
| 102 | + includeAllBlocks: true, |
| 103 | + fields: { |
| 104 | + block: args.blockFields ?? { number: true, hash: true, parentHash: true, timestamp: true }, |
| 105 | + log: args.logFields ?? { address: true, topics: true, data: true, logIndex: true, transactionHash: true, transactionIndex: true }, |
| 106 | + }, |
| 107 | + logs: args.logs, |
| 108 | + }); |
| 109 | + let res: Response; |
| 110 | + try { |
| 111 | + res = await fetchImpl(`${args.portalUrl}/stream`, { method: "POST", headers: { "content-type": "application/json", ...args.headers }, body, signal: args.signal }); |
| 112 | + } catch { |
| 113 | + await sleep(1000, args.signal); |
| 114 | + continue; |
| 115 | + } |
| 116 | + if (res.status === 204 || !res.body) { await sleep(500, args.signal); continue; } // no hot data yet; re-poll |
| 117 | + if (!res.ok) { await sleep(1000, args.signal); continue; } |
| 118 | + const reader = res.body.getReader(); |
| 119 | + const dec = new TextDecoder(); |
| 120 | + let buf = ""; |
| 121 | + try { |
| 122 | + for (;;) { |
| 123 | + const { done, value } = await reader.read(); |
| 124 | + if (done) break; |
| 125 | + buf += dec.decode(value, { stream: true }); |
| 126 | + let nl: number; |
| 127 | + while ((nl = buf.indexOf("\n")) >= 0) { |
| 128 | + const line = buf.slice(0, nl); |
| 129 | + buf = buf.slice(nl + 1); |
| 130 | + if (!line) continue; |
| 131 | + const batch = JSON.parse(line); |
| 132 | + if (batch?.header?.number != null) { |
| 133 | + cursor = batch.header.number + 1; // resume past this block on reconnect |
| 134 | + yield { header: batch.header, logs: batch.logs ?? [] }; |
| 135 | + } |
| 136 | + } |
| 137 | + } |
| 138 | + } catch { |
| 139 | + /* stream cut — reconnect from cursor */ |
| 140 | + } |
| 141 | + await sleep(200, args.signal); |
| 142 | + } |
| 143 | +} |
| 144 | + |
| 145 | +const sleep = (ms: number, signal?: AbortSignal) => |
| 146 | + new Promise<void>((resolve) => { |
| 147 | + if (signal?.aborted) return resolve(); |
| 148 | + const t = setTimeout(resolve, ms); |
| 149 | + signal?.addEventListener("abort", () => { clearTimeout(t); resolve(); }, { once: true }); |
| 150 | + }); |
| 151 | + |
| 152 | +// ─────────────────────────────── event producer ─────────────────────────────── |
| 153 | +// Emits ponder RealtimeSyncEvent-shaped objects. `finalizedHead()` polls the Portal finalized head so we |
| 154 | +// can emit `finalize` events (the RPC RealtimeSync derives finality from confirmations; the Portal tells |
| 155 | +// us directly). NB: the exact ponder BlockWithEventData construction (transactions/receipts/traces/ |
| 156 | +// childAddresses) is completed at the wiring step in runtime/realtime.ts, which already has the factory |
| 157 | +// child-address maps + buildEvents; here we surface the block header + logs + reorg/finalize control flow. |
| 158 | + |
| 159 | +export type PortalRealtimeEvent = |
| 160 | + | { type: "block"; block: SyncBlockHeader; logs: SyncLog[]; hasMatchedFilter: boolean } |
| 161 | + | { type: "reorg"; block: Light; reorgedBlocks: Light[] } |
| 162 | + | { type: "finalize"; block: Light }; |
| 163 | + |
| 164 | +export async function* portalRealtimeEvents( |
| 165 | + args: PortalRealtimeArgs & { finalizedHead: () => Promise<number | undefined>; finalizePollMs?: number }, |
| 166 | +): AsyncGenerator<PortalRealtimeEvent> { |
| 167 | + const unfinalized: Light[] = []; |
| 168 | + let lastFinalizePoll = 0; |
| 169 | + const pollMs = args.finalizePollMs ?? 4000; |
| 170 | + |
| 171 | + for await (const { header, logs } of streamHotBlocks(args)) { |
| 172 | + const light = toLight(header); |
| 173 | + const r = reconcile(unfinalized, light); |
| 174 | + if (r.kind === "duplicate") continue; |
| 175 | + if (r.kind === "gap") { |
| 176 | + // parent unknown: our window is behind a deeper reorg. Drop to a resync by clearing and continuing |
| 177 | + // from this block (the finalized floor still protects already-committed data). |
| 178 | + unfinalized.length = 0; |
| 179 | + } else if (r.kind === "reorg") { |
| 180 | + // trim the local chain and surface the rollback to the common ancestor |
| 181 | + while (unfinalized.length && unfinalized[unfinalized.length - 1]!.number > r.commonAncestor.number) unfinalized.pop(); |
| 182 | + yield { type: "reorg", block: r.commonAncestor, reorgedBlocks: r.reorgedBlocks }; |
| 183 | + } |
| 184 | + unfinalized.push(light); |
| 185 | + |
| 186 | + const block = toSyncBlockHeader(header); |
| 187 | + const syncLogs = logs.map((l) => toSyncLog(l, header)); |
| 188 | + yield { type: "block", block, logs: syncLogs, hasMatchedFilter: syncLogs.length > 0 }; |
| 189 | + |
| 190 | + // finalize on a cadence (cheap head probe), not every block |
| 191 | + const now = Date.now(); |
| 192 | + if (now - lastFinalizePoll >= pollMs) { |
| 193 | + lastFinalizePoll = now; |
| 194 | + const fh = await args.finalizedHead().catch(() => undefined); |
| 195 | + if (fh !== undefined) { |
| 196 | + const { finalizedTip, remaining } = takeFinalized(unfinalized, fh); |
| 197 | + if (finalizedTip) { |
| 198 | + unfinalized.length = 0; |
| 199 | + unfinalized.push(...remaining); |
| 200 | + yield { type: "finalize", block: finalizedTip }; |
| 201 | + } |
| 202 | + } |
| 203 | + } |
| 204 | + } |
| 205 | +} |
| 206 | + |
| 207 | +export { hx as _hx }; |
| 208 | +export type { SyncBlockHeader, SyncLog, Hex }; |
| 209 | +export { hexToNumber }; |
0 commit comments