From b2eaf69cb063ed44375a5315d85f0e2699fc707b Mon Sep 17 00:00:00 2001 From: Derek Guenther Date: Thu, 5 Nov 2026 17:25:23 -0500 Subject: [PATCH 1/5] feat(internal-explorer): stream new blocks over newHeads With 200ms Denim blocks the 2s poll of the ten-block home list replaced the whole list on every refresh. Follow the chain head over a server-side newHeads subscription instead, delivered to the page as Server-Sent Events, and fall back to the existing poll when a chain has no TIPS__WS_URL or the stream is reconnecting. Block summaries now carry Denim's timestampMs, and the home list shows sub-second ages. --- .env.example | 3 + app/api/internal-explorer/block-list.test.ts | 35 +++- app/api/internal-explorer/block-list.ts | 27 ++- app/api/internal-explorer/config.ts | 6 + app/api/internal-explorer/head-stream.test.ts | 156 ++++++++++++++++++ app/api/internal-explorer/head-stream.ts | 155 +++++++++++++++++ app/api/internal-explorer/heads/route.ts | 72 ++++++++ .../library/block-window.test.ts | 47 ++++++ app/internal-explorer/library/block-window.ts | 18 ++ app/internal-explorer/library/client.ts | 2 + app/internal-explorer/library/format.ts | 7 + app/internal-explorer/page.tsx | 63 ++++++- 12 files changed, 577 insertions(+), 14 deletions(-) create mode 100644 app/api/internal-explorer/head-stream.test.ts create mode 100644 app/api/internal-explorer/head-stream.ts create mode 100644 app/api/internal-explorer/heads/route.ts create mode 100644 app/internal-explorer/library/block-window.test.ts create mode 100644 app/internal-explorer/library/block-window.ts diff --git a/.env.example b/.env.example index f566e358..96402769 100644 --- a/.env.example +++ b/.env.example @@ -30,6 +30,9 @@ NEXT_PUBLIC_VIBENET_RPC_URL=https://rpc.vibes.base.org # One deployment serves all chains; provide a set per chain # (TIPS_MAINNET_, TIPS_SEPOLIA_, TIPS_ZERONET_). Secrets → Vercel env, not git. # TIPS_MAINNET_RPC_URL= +# Execution WebSocket for the live newHeads block stream. Optional per chain; +# without it the block list polls TIPS__RPC_URL. +# TIPS_MAINNET_WS_URL= # Audit events JSON-RPC endpoint (Postgres-backed transaction observability). # Opt-in per chain: the only source for bundle and rejected views, and the # enrichment source for block and transaction views. diff --git a/app/api/internal-explorer/block-list.test.ts b/app/api/internal-explorer/block-list.test.ts index 9fa21983..521b02db 100644 --- a/app/api/internal-explorer/block-list.test.ts +++ b/app/api/internal-explorer/block-list.test.ts @@ -2,7 +2,40 @@ import assert from 'node:assert/strict'; import { describe, test } from 'vitest'; -import { blockNumbersForPage, nextBlockCursor, parseBlockListQuery } from './block-list'; +import { + blockNumbersForPage, + nextBlockCursor, + parseBlockListQuery, + parseBlockSummary, +} from './block-list'; + +describe('parseBlockSummary', () => { + const rpcBlock = { + hash: `0x${'ab'.repeat(32)}`, + number: '0x10', + timestamp: '0x6b49d200', + transactions: [`0x${'01'.repeat(32)}`, `0x${'02'.repeat(32)}`], + gasUsed: '0x5208', + gasLimit: '0x7270e00', + baseFeePerGas: '0x1', + }; + + test('reads the Denim millisecond timestamp when present', () => { + const summary = parseBlockSummary({ ...rpcBlock, timestampMs: '0x1a3185c50c8' }); + assert.equal(summary?.timestampMs, 1_800_000_000_200); + assert.equal(summary?.timestamp, 1_800_000_000); + assert.equal(summary?.transactionCount, 2); + }); + + test('leaves timestampMs null for blocks without it', () => { + assert.equal(parseBlockSummary(rpcBlock)?.timestampMs, null); + }); + + test('rejects malformed blocks', () => { + assert.equal(parseBlockSummary(null), null); + assert.equal(parseBlockSummary({ ...rpcBlock, number: undefined }), null); + }); +}); describe('block list pagination', () => { test('starts at the inclusive cursor', () => { diff --git a/app/api/internal-explorer/block-list.ts b/app/api/internal-explorer/block-list.ts index 7410277a..e92b0653 100644 --- a/app/api/internal-explorer/block-list.ts +++ b/app/api/internal-explorer/block-list.ts @@ -11,6 +11,8 @@ export interface BlockSummary { hash: string; number: number; timestamp: number; + /** Unix milliseconds from Denim's `timestampMs`; null on blocks or nodes without it. */ + timestampMs: number | null; transactionCount: number; gasUsed: string | null; gasLimit: string | null; @@ -95,12 +97,8 @@ async function fetchLatestBlockNumber(rpcUrl: string): Promise { return hexToNumber(result); } -async function fetchBlockByNumber( - rpcUrl: string, - blockNumber: number, -): Promise { - const result = await rpcCall(rpcUrl, 'eth_getBlockByNumber', [toBlockTag(blockNumber), false]); - +/** Map a raw `eth_getBlockBy*` result (hashes only) to a summary, or null if malformed. */ +export function parseBlockSummary(result: unknown): BlockSummary | null { if (!result || typeof result !== 'object') return null; const block = result as Record; const number = hexToNumber(block.number); @@ -113,6 +111,7 @@ async function fetchBlockByNumber( hash: String(block.hash), number, timestamp, + timestampMs: hexToNumber(block.timestampMs), transactionCount: Array.isArray(block.transactions) ? block.transactions.length : 0, gasUsed: decimalQuantity(block.gasUsed), gasLimit: decimalQuantity(block.gasLimit), @@ -120,6 +119,22 @@ async function fetchBlockByNumber( }; } +async function fetchBlockByNumber( + rpcUrl: string, + blockNumber: number, +): Promise { + return parseBlockSummary( + await rpcCall(rpcUrl, 'eth_getBlockByNumber', [toBlockTag(blockNumber), false]), + ); +} + +export async function fetchBlockSummaryByHash( + rpcUrl: string, + hash: string, +): Promise { + return parseBlockSummary(await rpcCall(rpcUrl, 'eth_getBlockByHash', [hash, false])); +} + export async function fetchBlocksByNumbers( rpcUrl: string, numbers: number[], diff --git a/app/api/internal-explorer/config.ts b/app/api/internal-explorer/config.ts index 661bcebf..b5816745 100644 --- a/app/api/internal-explorer/config.ts +++ b/app/api/internal-explorer/config.ts @@ -19,6 +19,12 @@ export function getRpcUrl(chain: ExplorerChain): string { return envValue([`TIPS_${ENV_PREFIX[chain]}_RPC_URL`]) ?? 'http://localhost:8545'; } +// Execution WebSocket endpoint used for the live newHeads stream. Opt-in per +// chain via TIPS__WS_URL; when unset the block list polls instead. +export function getWsUrl(chain: ExplorerChain): string | undefined { + return envValue([`TIPS_${ENV_PREFIX[chain]}_WS_URL`]); +} + // Audit events JSON-RPC endpoint (Postgres-backed) for a chain. Unlike the // execution RPC there is no default: audit is opt-in per chain via // TIPS__AUDIT_RPC_URL. When unset, the audit source is treated as diff --git a/app/api/internal-explorer/head-stream.test.ts b/app/api/internal-explorer/head-stream.test.ts new file mode 100644 index 00000000..7397fa00 --- /dev/null +++ b/app/api/internal-explorer/head-stream.test.ts @@ -0,0 +1,156 @@ +import assert from 'node:assert/strict'; + +import { describe, test, vi } from 'vitest'; + +import type { BlockSummary } from './block-list'; +import { HeadHub, type HeadSocket } from './head-stream'; + +const HASH = `0x${'ab'.repeat(32)}`; + +class FakeSocket implements HeadSocket { + readyState = 0; + readonly sent: unknown[] = []; + closed = false; + private readonly listeners = new Map void>>(); + + send(data: string) { + this.sent.push(JSON.parse(data)); + } + + close() { + this.closed = true; + } + + addEventListener(type: string, listener: (event?: never) => void) { + const list = this.listeners.get(type) ?? []; + list.push(listener as (event?: unknown) => void); + this.listeners.set(type, list); + } + + emit(type: 'open' | 'close' | 'error'): void; + emit(type: 'message', payload: unknown): void; + emit(type: string, payload?: unknown) { + const event = type === 'message' ? { data: JSON.stringify(payload) } : undefined; + for (const listener of this.listeners.get(type) ?? []) listener(event); + } +} + +function summary(hash: string): BlockSummary { + return { + hash, + number: 16, + timestamp: 1_800_000_000, + timestampMs: 1_800_000_000_200, + transactionCount: 2, + gasUsed: null, + gasLimit: null, + baseFeePerGas: null, + }; +} + +function setup() { + const sockets: FakeSocket[] = []; + const fetchBlock = vi.fn(async (hash: string) => summary(hash)); + const onEnded = vi.fn(); + const hub = new HeadHub({ + wsUrl: 'wss://node.example:8546', + fetchBlock, + onEnded, + connect: () => { + const socket = new FakeSocket(); + sockets.push(socket); + return socket; + }, + }); + return { hub, sockets, fetchBlock, onEnded }; +} + +const headNotification = (hash: string) => ({ + jsonrpc: '2.0', + method: 'eth_subscription', + params: { subscription: '0x1', result: { number: '0x10', hash } }, +}); + +describe('HeadHub', () => { + test('shares one upstream subscription across subscribers', () => { + const { hub, sockets } = setup(); + hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + sockets[0].emit('open'); + + assert.equal(sockets.length, 1); + assert.deepEqual(sockets[0].sent, [ + { jsonrpc: '2.0', id: 1, method: 'eth_subscribe', params: ['newHeads'] }, + ]); + }); + + test('delivers each head as the block re-read by hash', async () => { + const { hub, sockets, fetchBlock } = setup(); + const received: BlockSummary[][] = [[], []]; + hub.subscribe({ onBlock: (block) => received[0].push(block), onEnd: () => {} }); + hub.subscribe({ onBlock: (block) => received[1].push(block), onEnd: () => {} }); + sockets[0].emit('open'); + sockets[0].emit('message', headNotification(HASH)); + + await vi.waitFor(() => assert.equal(received[1].length, 1)); + assert.deepEqual(fetchBlock.mock.calls, [[HASH]]); + assert.equal(received[0][0].timestampMs, 1_800_000_000_200); + }); + + test('ignores notifications without a valid block hash', async () => { + const { hub, sockets, fetchBlock } = setup(); + hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + sockets[0].emit('open'); + sockets[0].emit('message', headNotification('not-a-hash')); + sockets[0].emit('message', { jsonrpc: '2.0', id: 1, result: '0x1' }); + + await Promise.resolve(); + assert.equal(fetchBlock.mock.calls.length, 0); + }); + + test('closes the upstream when the last subscriber leaves', () => { + const { hub, sockets, onEnded } = setup(); + const first = hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + const second = hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + + first(); + assert.equal(sockets[0].closed, false); + second(); + assert.equal(sockets[0].closed, true); + assert.equal(onEnded.mock.calls.length, 1); + }); + + test('ends every subscriber when the upstream drops', () => { + const { hub, sockets, onEnded } = setup(); + const ended = [vi.fn(), vi.fn()]; + hub.subscribe({ onBlock: () => {}, onEnd: ended[0] }); + hub.subscribe({ onBlock: () => {}, onEnd: ended[1] }); + sockets[0].emit('close'); + + assert.equal(ended[0].mock.calls.length, 1); + assert.equal(ended[1].mock.calls.length, 1); + assert.equal(onEnded.mock.calls.length, 1); + }); + + test('ends subscribers when the subscription is rejected', () => { + const { hub, sockets } = setup(); + const onEnd = vi.fn(); + hub.subscribe({ onBlock: () => {}, onEnd }); + sockets[0].emit('open'); + sockets[0].emit('message', { jsonrpc: '2.0', id: 1, error: { message: 'not supported' } }); + + assert.equal(onEnd.mock.calls.length, 1); + assert.equal(sockets[0].closed, true); + }); + + test('a subscriber added after the hub ended is ended immediately', () => { + const { hub, sockets } = setup(); + hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + sockets[0].emit('error'); + + const onEnd = vi.fn(); + hub.subscribe({ onBlock: () => {}, onEnd }); + assert.equal(onEnd.mock.calls.length, 1); + assert.equal(sockets.length, 1); + }); +}); diff --git a/app/api/internal-explorer/head-stream.ts b/app/api/internal-explorer/head-stream.ts new file mode 100644 index 00000000..c05b9349 --- /dev/null +++ b/app/api/internal-explorer/head-stream.ts @@ -0,0 +1,155 @@ +// Live block heads for the Internal Explorer. One upstream `eth_subscribe` +// newHeads WebSocket per endpoint is shared by every connected browser, and +// each head is re-read over HTTP by hash: newHeads carries no transaction list, +// and RPC proxies that re-encode headers can drop Base's `timestampMs`. +// Server-only. +import { fetchBlockSummaryByHash, type BlockSummary } from './block-list'; + +const CONNECT_TIMEOUT_MS = 10_000; +const SUBSCRIBE_REQUEST_ID = 1; +const BLOCK_HASH = /^0x[0-9a-f]{64}$/i; + +/** The subset of the WHATWG WebSocket the hub uses; injectable for tests. */ +export interface HeadSocket { + readonly readyState: number; + send(data: string): void; + close(): void; + addEventListener(type: 'open' | 'close' | 'error', listener: () => void): void; + addEventListener(type: 'message', listener: (event: { data: unknown }) => void): void; +} + +export interface HeadSubscriber { + onBlock: (block: BlockSummary) => void; + /** Called once when the upstream stream ends; the subscriber is already removed. */ + onEnd: () => void; +} + +export interface HeadHubOptions { + wsUrl: string; + fetchBlock: (hash: string) => Promise; + connect?: (url: string) => HeadSocket; + /** Called once the hub has ended, so a registry can drop it. */ + onEnded?: () => void; +} + +export class HeadHub { + private readonly subscribers = new Set(); + private socket: HeadSocket | null = null; + private connectTimer: ReturnType | undefined; + private ended = false; + + constructor(private readonly options: HeadHubOptions) {} + + /** Add a subscriber, opening the upstream on first use. Returns an unsubscribe function. */ + subscribe(subscriber: HeadSubscriber): () => void { + if (this.ended) { + subscriber.onEnd(); + return () => {}; + } + this.subscribers.add(subscriber); + if (!this.socket) this.open(); + return () => { + this.subscribers.delete(subscriber); + if (this.subscribers.size === 0) this.end(); + }; + } + + private open() { + const connect = this.options.connect ?? ((url: string) => new WebSocket(url)); + let socket: HeadSocket; + try { + socket = connect(this.options.wsUrl); + } catch (error) { + console.error('Internal Explorer head stream failed to connect:', error); + this.end(); + return; + } + this.socket = socket; + this.connectTimer = setTimeout(() => this.end(), CONNECT_TIMEOUT_MS); + + socket.addEventListener('open', () => { + clearTimeout(this.connectTimer); + socket.send( + JSON.stringify({ + jsonrpc: '2.0', + id: SUBSCRIBE_REQUEST_ID, + method: 'eth_subscribe', + params: ['newHeads'], + }), + ); + }); + socket.addEventListener('message', (event) => this.handleMessage(event.data)); + socket.addEventListener('error', () => this.end()); + socket.addEventListener('close', () => this.end()); + } + + private handleMessage(data: unknown) { + let message: { + id?: unknown; + error?: unknown; + method?: unknown; + params?: { result?: { hash?: unknown } }; + }; + try { + message = JSON.parse(String(data)); + } catch { + return; + } + + if (message.id === SUBSCRIBE_REQUEST_ID && message.error) { + console.error('Internal Explorer head stream subscription rejected:', message.error); + this.end(); + return; + } + if (message.method !== 'eth_subscription') return; + + const hash = message.params?.result?.hash; + if (typeof hash !== 'string' || !BLOCK_HASH.test(hash)) return; + + void this.options + .fetchBlock(hash) + .then((block) => { + if (!block || this.ended) return; + for (const subscriber of this.subscribers) subscriber.onBlock(block); + }) + .catch(() => {}); + } + + private end() { + if (this.ended) return; + this.ended = true; + clearTimeout(this.connectTimer); + try { + this.socket?.close(); + } catch { + // Already closed. + } + const subscribers = [...this.subscribers]; + this.subscribers.clear(); + this.options.onEnded?.(); + for (const subscriber of subscribers) subscriber.onEnd(); + } +} + +const hubs = new Map(); + +/** Subscribe to live heads for a chain, sharing one upstream socket per WebSocket URL. */ +export function subscribeToHeads( + wsUrl: string, + rpcUrl: string, + subscriber: HeadSubscriber, +): () => void { + let hub = hubs.get(wsUrl); + if (!hub) { + const created = new HeadHub({ + wsUrl, + fetchBlock: (hash) => fetchBlockSummaryByHash(rpcUrl, hash), + onEnded: () => { + if (hubs.get(wsUrl) === created) hubs.delete(wsUrl); + }, + }); + hubs.set(wsUrl, created); + hub = created; + } + return hub.subscribe(subscriber); +} diff --git a/app/api/internal-explorer/heads/route.ts b/app/api/internal-explorer/heads/route.ts new file mode 100644 index 00000000..66c5e725 --- /dev/null +++ b/app/api/internal-explorer/heads/route.ts @@ -0,0 +1,72 @@ +import { resolveExplorerChainFromRequest } from '../chain'; +import { getRpcUrl, getWsUrl } from '../config'; +import { explorerDisabledResponse } from '../guard'; +import { subscribeToHeads } from '../head-stream'; + +export const runtime = 'nodejs'; +export const dynamic = 'force-dynamic'; + +// Comment frames keep idle proxies from closing the stream between heads. +const KEEPALIVE_MS = 15_000; +// EventSource waits this long before reconnecting after the stream ends. +const RECONNECT_MS = 3_000; + +// Server-Sent Events stream of new canonical blocks (BlockSummary JSON per +// `data:` frame), fed by the chain's newHeads WebSocket. 404 when the chain +// has no TIPS__WS_URL, so the client falls back to polling /blocks. +export async function GET(request: Request) { + const disabled = explorerDisabledResponse(); + if (disabled) return disabled; + const chain = resolveExplorerChainFromRequest(request); + const wsUrl = getWsUrl(chain); + if (!wsUrl) { + return Response.json({ error: 'Head stream not configured' }, { status: 404 }); + } + const rpcUrl = getRpcUrl(chain); + + const encoder = new TextEncoder(); + let finish = () => {}; + + const stream = new ReadableStream({ + start(controller) { + let done = false; + let unsubscribe = () => {}; + const send = (frame: string) => { + if (!done) controller.enqueue(encoder.encode(frame)); + }; + + const keepalive = setInterval(() => send(': keepalive\n\n'), KEEPALIVE_MS); + finish = () => { + if (done) return; + done = true; + clearInterval(keepalive); + request.signal.removeEventListener('abort', finish); + unsubscribe(); + try { + controller.close(); + } catch { + // Already closed by the client. + } + }; + + send(`retry: ${RECONNECT_MS}\n\n`); + request.signal.addEventListener('abort', finish); + unsubscribe = subscribeToHeads(wsUrl, rpcUrl, { + onBlock: (block) => send(`data: ${JSON.stringify(block)}\n\n`), + onEnd: () => finish(), + }); + }, + cancel() { + finish(); + }, + }); + + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + }, + }); +} diff --git a/app/internal-explorer/library/block-window.test.ts b/app/internal-explorer/library/block-window.test.ts new file mode 100644 index 00000000..acc38cee --- /dev/null +++ b/app/internal-explorer/library/block-window.test.ts @@ -0,0 +1,47 @@ +import assert from 'node:assert/strict'; + +import { describe, test } from 'vitest'; + +import { mergeBlockWindow } from './block-window'; +import type { BlockSummary } from './types'; + +function block(number: number, hash = `0x${number.toString(16)}`): BlockSummary { + return { + hash, + number, + timestamp: 1_800_000_000, + timestampMs: null, + transactionCount: 2, + gasUsed: null, + gasLimit: null, + baseFeePerGas: null, + }; +} + +const numbers = (blocks: BlockSummary[]) => blocks.map((b) => b.number); + +describe('mergeBlockWindow', () => { + test('keeps the newest blocks first and trims to the limit', () => { + const current = [block(12), block(11), block(10)]; + assert.deepEqual(numbers(mergeBlockWindow(current, [block(13)], 3)), [13, 12, 11]); + }); + + test('places out-of-order heads by height', () => { + const merged = mergeBlockWindow([block(13), block(11)], [block(12)], 10); + assert.deepEqual(numbers(merged), [13, 12, 11]); + }); + + test('replaces a block at the same height with the incoming one', () => { + const merged = mergeBlockWindow([block(12, '0xold'), block(11)], [block(12, '0xnew')], 10); + assert.deepEqual( + merged.map((b) => b.hash), + ['0xnew', '0xb'], + ); + }); + + test('keeps streamed heads newer than a late polled snapshot', () => { + const streamed = [block(20), block(19)]; + const snapshot = [block(18), block(17), block(16)]; + assert.deepEqual(numbers(mergeBlockWindow(streamed, snapshot, 4)), [20, 19, 18, 17]); + }); +}); diff --git a/app/internal-explorer/library/block-window.ts b/app/internal-explorer/library/block-window.ts new file mode 100644 index 00000000..f06b4d9e --- /dev/null +++ b/app/internal-explorer/library/block-window.ts @@ -0,0 +1,18 @@ +// Newest-first window of recent blocks shared by the live head stream and the +// polling fallback. Client-safe. +import type { BlockSummary } from './types'; + +/** + * Merge `incoming` into `current`, keeping one block per height (the incoming + * one wins, so a reorged head replaces the old block) and the newest `limit`. + */ +export function mergeBlockWindow( + current: readonly BlockSummary[], + incoming: readonly BlockSummary[], + limit: number, +): BlockSummary[] { + const byNumber = new Map(); + for (const block of current) byNumber.set(block.number, block); + for (const block of incoming) byNumber.set(block.number, block); + return [...byNumber.values()].sort((lhs, rhs) => rhs.number - lhs.number).slice(0, limit); +} diff --git a/app/internal-explorer/library/client.ts b/app/internal-explorer/library/client.ts index 9533372d..be5f3358 100644 --- a/app/internal-explorer/library/client.ts +++ b/app/internal-explorer/library/client.ts @@ -64,6 +64,8 @@ function withQuery(path: string, params: Record get('/api/internal-explorer/blocks', chain, signal), + /** Server-Sent Events URL for live BlockSummary heads; 404s when the chain has no WS URL. */ + headStreamUrl: (chain: ExplorerChain) => `/api/internal-explorer/heads?chain=${chain}`, blocksPage: ( chain: ExplorerChain, options?: { cursor?: number; limit?: number }, diff --git a/app/internal-explorer/library/format.ts b/app/internal-explorer/library/format.ts index 9f41b55b..70a63dc7 100644 --- a/app/internal-explorer/library/format.ts +++ b/app/internal-explorer/library/format.ts @@ -41,6 +41,13 @@ export function timeAgoFromSeconds(ts: number): string { return `${Math.floor(seconds / 86400)}d ago`; } +/** Like timeAgoFromSeconds, with tenths of a second under 10s for 200ms blocks. */ +export function timeAgoFromMilliseconds(ts: number, now = Date.now()): string { + const milliseconds = Math.max(0, Math.floor(now - ts)); + if (milliseconds < 10_000) return `${(Math.floor(milliseconds / 100) / 10).toFixed(1)}s ago`; + return timeAgoFromSeconds(ts / 1000); +} + /** `0x1234…abcd` abbreviation for a hash/address. */ export function shortHash(value: string, lead = 10, tail = 8): string { if (value.length <= lead + tail + 1) return value; diff --git a/app/internal-explorer/page.tsx b/app/internal-explorer/page.tsx index 9019239b..686c5020 100644 --- a/app/internal-explorer/page.tsx +++ b/app/internal-explorer/page.tsx @@ -24,7 +24,14 @@ import { formatSignedInteger, formatSignedPct, } from './library/explorer-format'; -import { formatGasPrice, formatHexValue, shortHash, timeAgoFromSeconds } from './library/format'; +import { mergeBlockWindow } from './library/block-window'; +import { + formatGasPrice, + formatHexValue, + shortHash, + timeAgoFromMilliseconds, + timeAgoFromSeconds, +} from './library/format'; import { explorerHref } from './library/links'; import { formatRejectionReason, @@ -162,7 +169,11 @@ function BlockRow({ #{block.number.toLocaleString()} - {timeAgoFromSeconds(block.timestamp)} + + {block.timestampMs !== null + ? timeAgoFromMilliseconds(block.timestampMs) + : timeAgoFromSeconds(block.timestamp)} +
{block.hash} @@ -219,13 +230,16 @@ function BlocksTab({ const { showShadowDelta, setShowShadowDelta } = useShadowDelta(); const [shadowCandidates, setShadowCandidates] = useState>({}); - // Canonical-block view: poll the chain tip directly. The shadow-delta view - // has its own source below, so this fetch stands down when it is on. + // Canonical-block view: load the chain tip once, then follow the live head + // stream. Poll instead while the stream is unavailable (not configured for + // the chain, or reconnecting). The shadow-delta view has its own source + // below, so this stands down when it is on. useEffect(() => { if (showShadowDelta) return undefined; let cancelled = false; const controller = new AbortController(); + let pollTimer: number | undefined; setLoading(true); setBlocks([]); setFetchFailed(false); @@ -235,7 +249,7 @@ function BlocksTab({ try { const data = await explorerApi.blocks(chain, controller.signal); if (!cancelled) { - setBlocks(data.blocks); + setBlocks((current) => mergeBlockWindow(current, data.blocks, HOME_BLOCK_LIMIT)); setFetchFailed(false); } } catch { @@ -245,12 +259,47 @@ function BlocksTab({ } } + function startPolling() { + if (pollTimer === undefined) pollTimer = window.setInterval(() => void load(), 2000); + } + + function stopPolling() { + window.clearInterval(pollTimer); + pollTimer = undefined; + } + void load(); - const timer = window.setInterval(() => void load(), 2000); + + const source = + typeof EventSource === 'undefined' ? null : new EventSource(explorerApi.headStreamUrl(chain)); + if (source) { + source.onmessage = (event) => { + let head: BlockSummary; + try { + head = JSON.parse(event.data) as BlockSummary; + } catch { + return; + } + stopPolling(); + setBlocks((current) => mergeBlockWindow(current, [head], HOME_BLOCK_LIMIT)); + setFetchFailed(false); + setLoading(false); + }; + // EventSource retries on its own after a dropped stream; it only gives + // up (CLOSED) on a non-stream response such as the not-configured 404. + source.onerror = () => { + startPolling(); + if (source.readyState === EventSource.CLOSED) source.close(); + }; + } else { + startPolling(); + } + return () => { cancelled = true; controller.abort(); - window.clearInterval(timer); + source?.close(); + stopPolling(); }; }, [chain, showShadowDelta]); From fa9a2c618448620a66c3c97f66cdfceec29e8ced Mon Sep 17 00:00:00 2001 From: Derek Guenther Date: Thu, 5 Nov 2026 17:48:07 -0500 Subject: [PATCH 2/5] feat(internal-explorer): forward heads straight from newHeads headers Drop the per-head eth_getBlockByHash on the server. Streamed blocks use the header's timestampMs when the upstream sends it and fall back to the second-resolution timestamp otherwise. The page fills transaction counts (and any missing timestampMs) by batching streamed heights into one /blocks-by-numbers request every 500ms, matched by hash so a reorg can't swap in the wrong block. --- app/api/internal-explorer/block-list.test.ts | 5 ++ app/api/internal-explorer/block-list.ts | 18 +++--- app/api/internal-explorer/head-stream.test.ts | 63 ++++++++++--------- app/api/internal-explorer/head-stream.ts | 31 ++++----- app/api/internal-explorer/heads/route.ts | 10 ++- .../library/block-window.test.ts | 32 +++++++++- app/internal-explorer/library/block-window.ts | 28 ++++++++- app/internal-explorer/page.tsx | 39 +++++++++++- 8 files changed, 158 insertions(+), 68 deletions(-) diff --git a/app/api/internal-explorer/block-list.test.ts b/app/api/internal-explorer/block-list.test.ts index 521b02db..5703aae7 100644 --- a/app/api/internal-explorer/block-list.test.ts +++ b/app/api/internal-explorer/block-list.test.ts @@ -31,6 +31,11 @@ describe('parseBlockSummary', () => { assert.equal(parseBlockSummary(rpcBlock)?.timestampMs, null); }); + test('leaves transactionCount null for a header without a transaction list', () => { + const { transactions: _transactions, ...header } = rpcBlock; + assert.equal(parseBlockSummary(header)?.transactionCount, null); + }); + test('rejects malformed blocks', () => { assert.equal(parseBlockSummary(null), null); assert.equal(parseBlockSummary({ ...rpcBlock, number: undefined }), null); diff --git a/app/api/internal-explorer/block-list.ts b/app/api/internal-explorer/block-list.ts index e92b0653..6c136fa8 100644 --- a/app/api/internal-explorer/block-list.ts +++ b/app/api/internal-explorer/block-list.ts @@ -13,7 +13,8 @@ export interface BlockSummary { timestamp: number; /** Unix milliseconds from Denim's `timestampMs`; null on blocks or nodes without it. */ timestampMs: number | null; - transactionCount: number; + /** Null for streamed heads until the client fills it from /blocks-by-numbers. */ + transactionCount: number | null; gasUsed: string | null; gasLimit: string | null; baseFeePerGas: string | null; @@ -97,7 +98,11 @@ async function fetchLatestBlockNumber(rpcUrl: string): Promise { return hexToNumber(result); } -/** Map a raw `eth_getBlockBy*` result (hashes only) to a summary, or null if malformed. */ +/** + * Map a raw `eth_getBlockBy*` result (hashes only) or a newHeads header to a + * summary, or null if malformed. Headers carry no transaction list, so their + * `transactionCount` is null. + */ export function parseBlockSummary(result: unknown): BlockSummary | null { if (!result || typeof result !== 'object') return null; const block = result as Record; @@ -112,7 +117,7 @@ export function parseBlockSummary(result: unknown): BlockSummary | null { number, timestamp, timestampMs: hexToNumber(block.timestampMs), - transactionCount: Array.isArray(block.transactions) ? block.transactions.length : 0, + transactionCount: Array.isArray(block.transactions) ? block.transactions.length : null, gasUsed: decimalQuantity(block.gasUsed), gasLimit: decimalQuantity(block.gasLimit), baseFeePerGas: decimalQuantity(block.baseFeePerGas), @@ -128,13 +133,6 @@ async function fetchBlockByNumber( ); } -export async function fetchBlockSummaryByHash( - rpcUrl: string, - hash: string, -): Promise { - return parseBlockSummary(await rpcCall(rpcUrl, 'eth_getBlockByHash', [hash, false])); -} - export async function fetchBlocksByNumbers( rpcUrl: string, numbers: number[], diff --git a/app/api/internal-explorer/head-stream.test.ts b/app/api/internal-explorer/head-stream.test.ts index 7397fa00..68b7de1d 100644 --- a/app/api/internal-explorer/head-stream.test.ts +++ b/app/api/internal-explorer/head-stream.test.ts @@ -35,26 +35,11 @@ class FakeSocket implements HeadSocket { } } -function summary(hash: string): BlockSummary { - return { - hash, - number: 16, - timestamp: 1_800_000_000, - timestampMs: 1_800_000_000_200, - transactionCount: 2, - gasUsed: null, - gasLimit: null, - baseFeePerGas: null, - }; -} - function setup() { const sockets: FakeSocket[] = []; - const fetchBlock = vi.fn(async (hash: string) => summary(hash)); const onEnded = vi.fn(); const hub = new HeadHub({ wsUrl: 'wss://node.example:8546', - fetchBlock, onEnded, connect: () => { const socket = new FakeSocket(); @@ -62,13 +47,16 @@ function setup() { return socket; }, }); - return { hub, sockets, fetchBlock, onEnded }; + return { hub, sockets, onEnded }; } -const headNotification = (hash: string) => ({ +const headNotification = (hash: string, extra: Record = {}) => ({ jsonrpc: '2.0', method: 'eth_subscription', - params: { subscription: '0x1', result: { number: '0x10', hash } }, + params: { + subscription: '0x1', + result: { number: '0x10', hash, timestamp: '0x6b49d200', gasUsed: '0x5208', ...extra }, + }, }); describe('HeadHub', () => { @@ -84,28 +72,47 @@ describe('HeadHub', () => { ]); }); - test('delivers each head as the block re-read by hash', async () => { - const { hub, sockets, fetchBlock } = setup(); + test('delivers each head to every subscriber straight from the header', () => { + const { hub, sockets } = setup(); const received: BlockSummary[][] = [[], []]; hub.subscribe({ onBlock: (block) => received[0].push(block), onEnd: () => {} }); hub.subscribe({ onBlock: (block) => received[1].push(block), onEnd: () => {} }); sockets[0].emit('open'); + sockets[0].emit('message', headNotification(HASH, { timestampMs: '0x1a3185c50c8' })); + + assert.equal(received[1].length, 1); + assert.deepEqual(received[0][0], { + hash: HASH, + number: 16, + timestamp: 1_800_000_000, + timestampMs: 1_800_000_000_200, + transactionCount: null, + gasUsed: '21000', + gasLimit: null, + baseFeePerGas: null, + }); + }); + + test('leaves timestampMs null when the header lacks it', () => { + const { hub, sockets } = setup(); + const received: BlockSummary[] = []; + hub.subscribe({ onBlock: (block) => received.push(block), onEnd: () => {} }); + sockets[0].emit('open'); sockets[0].emit('message', headNotification(HASH)); - await vi.waitFor(() => assert.equal(received[1].length, 1)); - assert.deepEqual(fetchBlock.mock.calls, [[HASH]]); - assert.equal(received[0][0].timestampMs, 1_800_000_000_200); + assert.equal(received[0].timestampMs, null); + assert.equal(received[0].timestamp, 1_800_000_000); }); - test('ignores notifications without a valid block hash', async () => { - const { hub, sockets, fetchBlock } = setup(); - hub.subscribe({ onBlock: () => {}, onEnd: () => {} }); + test('ignores notifications without a valid block hash', () => { + const { hub, sockets } = setup(); + const onBlock = vi.fn(); + hub.subscribe({ onBlock, onEnd: () => {} }); sockets[0].emit('open'); sockets[0].emit('message', headNotification('not-a-hash')); sockets[0].emit('message', { jsonrpc: '2.0', id: 1, result: '0x1' }); - await Promise.resolve(); - assert.equal(fetchBlock.mock.calls.length, 0); + assert.equal(onBlock.mock.calls.length, 0); }); test('closes the upstream when the last subscriber leaves', () => { diff --git a/app/api/internal-explorer/head-stream.ts b/app/api/internal-explorer/head-stream.ts index c05b9349..9839f25d 100644 --- a/app/api/internal-explorer/head-stream.ts +++ b/app/api/internal-explorer/head-stream.ts @@ -1,9 +1,9 @@ // Live block heads for the Internal Explorer. One upstream `eth_subscribe` -// newHeads WebSocket per endpoint is shared by every connected browser, and -// each head is re-read over HTTP by hash: newHeads carries no transaction list, -// and RPC proxies that re-encode headers can drop Base's `timestampMs`. +// newHeads WebSocket per endpoint is shared by every connected browser. Heads +// are forwarded straight from the header: `timestampMs` when the upstream sends +// it, and no transaction count, which the client fills in separately. // Server-only. -import { fetchBlockSummaryByHash, type BlockSummary } from './block-list'; +import { parseBlockSummary, type BlockSummary } from './block-list'; const CONNECT_TIMEOUT_MS = 10_000; const SUBSCRIBE_REQUEST_ID = 1; @@ -26,7 +26,6 @@ export interface HeadSubscriber { export interface HeadHubOptions { wsUrl: string; - fetchBlock: (hash: string) => Promise; connect?: (url: string) => HeadSocket; /** Called once the hub has ended, so a registry can drop it. */ onEnded?: () => void; @@ -88,7 +87,7 @@ export class HeadHub { id?: unknown; error?: unknown; method?: unknown; - params?: { result?: { hash?: unknown } }; + params?: { result?: Record }; }; try { message = JSON.parse(String(data)); @@ -103,16 +102,13 @@ export class HeadHub { } if (message.method !== 'eth_subscription') return; - const hash = message.params?.result?.hash; + const header = message.params?.result; + const hash = header?.hash; if (typeof hash !== 'string' || !BLOCK_HASH.test(hash)) return; - void this.options - .fetchBlock(hash) - .then((block) => { - if (!block || this.ended) return; - for (const subscriber of this.subscribers) subscriber.onBlock(block); - }) - .catch(() => {}); + const block = parseBlockSummary(header); + if (!block) return; + for (const subscriber of this.subscribers) subscriber.onBlock(block); } private end() { @@ -134,16 +130,11 @@ export class HeadHub { const hubs = new Map(); /** Subscribe to live heads for a chain, sharing one upstream socket per WebSocket URL. */ -export function subscribeToHeads( - wsUrl: string, - rpcUrl: string, - subscriber: HeadSubscriber, -): () => void { +export function subscribeToHeads(wsUrl: string, subscriber: HeadSubscriber): () => void { let hub = hubs.get(wsUrl); if (!hub) { const created = new HeadHub({ wsUrl, - fetchBlock: (hash) => fetchBlockSummaryByHash(rpcUrl, hash), onEnded: () => { if (hubs.get(wsUrl) === created) hubs.delete(wsUrl); }, diff --git a/app/api/internal-explorer/heads/route.ts b/app/api/internal-explorer/heads/route.ts index 66c5e725..d1b7d433 100644 --- a/app/api/internal-explorer/heads/route.ts +++ b/app/api/internal-explorer/heads/route.ts @@ -1,5 +1,5 @@ import { resolveExplorerChainFromRequest } from '../chain'; -import { getRpcUrl, getWsUrl } from '../config'; +import { getWsUrl } from '../config'; import { explorerDisabledResponse } from '../guard'; import { subscribeToHeads } from '../head-stream'; @@ -11,8 +11,8 @@ const KEEPALIVE_MS = 15_000; // EventSource waits this long before reconnecting after the stream ends. const RECONNECT_MS = 3_000; -// Server-Sent Events stream of new canonical blocks (BlockSummary JSON per -// `data:` frame), fed by the chain's newHeads WebSocket. 404 when the chain +// Server-Sent Events stream of new canonical blocks (header-only BlockSummary +// JSON per `data:` frame), fed by the chain's newHeads WebSocket. 404 when the chain // has no TIPS__WS_URL, so the client falls back to polling /blocks. export async function GET(request: Request) { const disabled = explorerDisabledResponse(); @@ -22,8 +22,6 @@ export async function GET(request: Request) { if (!wsUrl) { return Response.json({ error: 'Head stream not configured' }, { status: 404 }); } - const rpcUrl = getRpcUrl(chain); - const encoder = new TextEncoder(); let finish = () => {}; @@ -51,7 +49,7 @@ export async function GET(request: Request) { send(`retry: ${RECONNECT_MS}\n\n`); request.signal.addEventListener('abort', finish); - unsubscribe = subscribeToHeads(wsUrl, rpcUrl, { + unsubscribe = subscribeToHeads(wsUrl, { onBlock: (block) => send(`data: ${JSON.stringify(block)}\n\n`), onEnd: () => finish(), }); diff --git a/app/internal-explorer/library/block-window.test.ts b/app/internal-explorer/library/block-window.test.ts index acc38cee..2b90dc8b 100644 --- a/app/internal-explorer/library/block-window.test.ts +++ b/app/internal-explorer/library/block-window.test.ts @@ -2,7 +2,7 @@ import assert from 'node:assert/strict'; import { describe, test } from 'vitest'; -import { mergeBlockWindow } from './block-window'; +import { fillBlockDetails, mergeBlockWindow } from './block-window'; import type { BlockSummary } from './types'; function block(number: number, hash = `0x${number.toString(16)}`): BlockSummary { @@ -44,4 +44,34 @@ describe('mergeBlockWindow', () => { const snapshot = [block(18), block(17), block(16)]; assert.deepEqual(numbers(mergeBlockWindow(streamed, snapshot, 4)), [20, 19, 18, 17]); }); + + test('a repeated head keeps details already filled for that block', () => { + const filled = { ...block(12), timestampMs: 1_800_000_000_400 }; + const header = { ...block(12), transactionCount: null }; + const [merged] = mergeBlockWindow([filled], [header], 10); + assert.equal(merged.transactionCount, 2); + assert.equal(merged.timestampMs, 1_800_000_000_400); + }); +}); + +describe('fillBlockDetails', () => { + test('replaces header-only rows with their fetched summaries', () => { + const header = { ...block(12), transactionCount: null }; + const fetched = { ...block(12), transactionCount: 7, timestampMs: 1_800_000_000_400 }; + const [filled, untouched] = fillBlockDetails([header, block(11)], [fetched]); + assert.equal(filled.transactionCount, 7); + assert.equal(filled.timestampMs, 1_800_000_000_400); + assert.equal(untouched.transactionCount, 2); + }); + + test('ignores a fetched block that no longer matches the row after a reorg', () => { + const header = { ...block(12, '0xnew'), transactionCount: null }; + const [row] = fillBlockDetails([header], [{ ...block(12, '0xold'), transactionCount: 7 }]); + assert.equal(row.hash, '0xnew'); + assert.equal(row.transactionCount, null); + }); + + test('does not add blocks that have left the window', () => { + assert.deepEqual(numbers(fillBlockDetails([block(20)], [block(5)])), [20]); + }); }); diff --git a/app/internal-explorer/library/block-window.ts b/app/internal-explorer/library/block-window.ts index f06b4d9e..b68fdf08 100644 --- a/app/internal-explorer/library/block-window.ts +++ b/app/internal-explorer/library/block-window.ts @@ -5,6 +5,7 @@ import type { BlockSummary } from './types'; /** * Merge `incoming` into `current`, keeping one block per height (the incoming * one wins, so a reorged head replaces the old block) and the newest `limit`. + * A repeat of the same block keeps details the incoming copy lacks. */ export function mergeBlockWindow( current: readonly BlockSummary[], @@ -13,6 +14,31 @@ export function mergeBlockWindow( ): BlockSummary[] { const byNumber = new Map(); for (const block of current) byNumber.set(block.number, block); - for (const block of incoming) byNumber.set(block.number, block); + for (const block of incoming) { + const existing = byNumber.get(block.number); + byNumber.set( + block.number, + existing?.hash === block.hash + ? { + ...block, + timestampMs: block.timestampMs ?? existing.timestampMs, + transactionCount: block.transactionCount ?? existing.transactionCount, + } + : block, + ); + } return [...byNumber.values()].sort((lhs, rhs) => rhs.number - lhs.number).slice(0, limit); } + +/** + * Replace rows with their fetched full summaries. Only rows whose hash still + * matches are replaced, so a fetch that raced a reorg can't swap in a + * different block; the reorged head arrives on the stream on its own. + */ +export function fillBlockDetails( + current: readonly BlockSummary[], + fetched: readonly BlockSummary[], +): BlockSummary[] { + const byHash = new Map(fetched.map((block) => [block.hash, block])); + return current.map((block) => byHash.get(block.hash) ?? block); +} diff --git a/app/internal-explorer/page.tsx b/app/internal-explorer/page.tsx index 686c5020..05562ced 100644 --- a/app/internal-explorer/page.tsx +++ b/app/internal-explorer/page.tsx @@ -24,7 +24,7 @@ import { formatSignedInteger, formatSignedPct, } from './library/explorer-format'; -import { mergeBlockWindow } from './library/block-window'; +import { fillBlockDetails, mergeBlockWindow } from './library/block-window'; import { formatGasPrice, formatHexValue, @@ -181,7 +181,7 @@ function BlockRow({
- {block.transactionCount} + {block.transactionCount ?? '—'} txns
@@ -216,6 +216,11 @@ function BlockRow({ } const HOME_BLOCK_LIMIT = 10; +// Streamed heads have no transaction count; collect their heights for this long +// and fill them with one /blocks-by-numbers request. +const DETAIL_FILL_MS = 500; +// The WebSocket node can be ahead of the HTTP one, so a fill may miss a block. +const DETAIL_FILL_ATTEMPTS = 3; function BlocksTab({ chain, @@ -268,6 +273,34 @@ function BlocksTab({ pollTimer = undefined; } + const pendingDetails = new Map(); + let fillTimer: number | undefined; + + function queueDetails(blockNumber: number, attempts = 0) { + if (attempts >= DETAIL_FILL_ATTEMPTS) return; + pendingDetails.set(blockNumber, attempts); + fillTimer ??= window.setTimeout(() => void fillDetails(), DETAIL_FILL_MS); + } + + async function fillDetails() { + fillTimer = undefined; + const batch = [...pendingDetails]; + pendingDetails.clear(); + if (batch.length === 0) return; + try { + const filled = await explorerApi.blocksByNumbers( + chain, + batch.map(([blockNumber]) => blockNumber), + controller.signal, + ); + if (!cancelled) setBlocks((current) => fillBlockDetails(current, filled)); + } catch { + if (!cancelled) { + for (const [blockNumber, attempts] of batch) queueDetails(blockNumber, attempts + 1); + } + } + } + void load(); const source = @@ -282,6 +315,7 @@ function BlocksTab({ } stopPolling(); setBlocks((current) => mergeBlockWindow(current, [head], HOME_BLOCK_LIMIT)); + if (head.transactionCount === null) queueDetails(head.number); setFetchFailed(false); setLoading(false); }; @@ -300,6 +334,7 @@ function BlocksTab({ controller.abort(); source?.close(); stopPolling(); + window.clearTimeout(fillTimer); }; }, [chain, showShadowDelta]); From 77b5312d65774ca3302641706be05b8839b4b109 Mon Sep 17 00:00:00 2001 From: Derek Guenther Date: Thu, 5 Nov 2026 18:02:14 -0500 Subject: [PATCH 3/5] feat(internal-explorer): show streamed heads after a fixed delay Batching fills every 500ms left 0-3 header-only rows at the top, so the dashed transaction counts crept toward the head and fell back. Hold each streamed head for 1s before adding it to the list; the fill lands in that window, so rows appear complete and the list stays a steady distance behind the chain head. A late fill still shows the row on schedule and fills it in place. --- app/internal-explorer/page.tsx | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/app/internal-explorer/page.tsx b/app/internal-explorer/page.tsx index 05562ced..bb75e20c 100644 --- a/app/internal-explorer/page.tsx +++ b/app/internal-explorer/page.tsx @@ -219,6 +219,9 @@ const HOME_BLOCK_LIMIT = 10; // Streamed heads have no transaction count; collect their heights for this long // and fill them with one /blocks-by-numbers request. const DETAIL_FILL_MS = 500; +// Each streamed head is shown this long after it arrives, giving the fill time +// to land first and keeping the list a steady distance behind the chain head. +const HEAD_DISPLAY_DELAY_MS = 1_000; // The WebSocket node can be ahead of the HTTP one, so a fill may miss a block. const DETAIL_FILL_ATTEMPTS = 3; @@ -275,6 +278,19 @@ function BlocksTab({ const pendingDetails = new Map(); let fillTimer: number | undefined; + const heldHeads = new Map(); + const holdTimers = new Set(); + + function holdHead(head: BlockSummary) { + heldHeads.set(head.hash, head); + const timer = window.setTimeout(() => { + holdTimers.delete(timer); + const block = heldHeads.get(head.hash) ?? head; + heldHeads.delete(head.hash); + setBlocks((current) => mergeBlockWindow(current, [block], HOME_BLOCK_LIMIT)); + }, HEAD_DISPLAY_DELAY_MS); + holdTimers.add(timer); + } function queueDetails(blockNumber: number, attempts = 0) { if (attempts >= DETAIL_FILL_ATTEMPTS) return; @@ -293,7 +309,11 @@ function BlocksTab({ batch.map(([blockNumber]) => blockNumber), controller.signal, ); - if (!cancelled) setBlocks((current) => fillBlockDetails(current, filled)); + if (cancelled) return; + for (const block of filled) { + if (heldHeads.has(block.hash)) heldHeads.set(block.hash, block); + } + setBlocks((current) => fillBlockDetails(current, filled)); } catch { if (!cancelled) { for (const [blockNumber, attempts] of batch) queueDetails(blockNumber, attempts + 1); @@ -314,7 +334,7 @@ function BlocksTab({ return; } stopPolling(); - setBlocks((current) => mergeBlockWindow(current, [head], HOME_BLOCK_LIMIT)); + holdHead(head); if (head.transactionCount === null) queueDetails(head.number); setFetchFailed(false); setLoading(false); @@ -335,6 +355,7 @@ function BlocksTab({ source?.close(); stopPolling(); window.clearTimeout(fillTimer); + for (const timer of holdTimers) window.clearTimeout(timer); }; }, [chain, showShadowDelta]); From 68ac2de260a64bdb0715c42d10919d5dfe3f0b0f Mon Sep 17 00:00:00 2001 From: Derek Guenther Date: Thu, 5 Nov 2026 18:04:15 -0500 Subject: [PATCH 4/5] feat(internal-explorer): reveal streamed tx counts at a fixed age Show streamed heads immediately again and delay only their transaction count: a count is revealed 600ms after its head arrived, so the dashed rows stay a fixed depth below the top. Fills now batch every 200ms so they land inside that window, and timestampMs from a fill is applied as soon as it lands. --- app/internal-explorer/page.tsx | 51 ++++++++++++++++++++-------------- 1 file changed, 30 insertions(+), 21 deletions(-) diff --git a/app/internal-explorer/page.tsx b/app/internal-explorer/page.tsx index bb75e20c..67400498 100644 --- a/app/internal-explorer/page.tsx +++ b/app/internal-explorer/page.tsx @@ -218,10 +218,12 @@ function BlockRow({ const HOME_BLOCK_LIMIT = 10; // Streamed heads have no transaction count; collect their heights for this long // and fill them with one /blocks-by-numbers request. -const DETAIL_FILL_MS = 500; -// Each streamed head is shown this long after it arrives, giving the fill time -// to land first and keeping the list a steady distance behind the chain head. -const HEAD_DISPLAY_DELAY_MS = 1_000; +const DETAIL_FILL_MS = 200; +// A streamed head's transaction count is revealed this long after the head +// arrives, so counts appear a steady number of rows below the top instead of +// whenever each fill happens to land. Must exceed DETAIL_FILL_MS plus a fill +// round trip. +const TX_COUNT_REVEAL_MS = 600; // The WebSocket node can be ahead of the HTTP one, so a fill may miss a block. const DETAIL_FILL_ATTEMPTS = 3; @@ -278,18 +280,20 @@ function BlocksTab({ const pendingDetails = new Map(); let fillTimer: number | undefined; - const heldHeads = new Map(); - const holdTimers = new Set(); + // Fetched summaries for streamed heads whose transaction count is not yet + // due; null until the fill lands. + const heldCounts = new Map(); + const revealTimers = new Set(); - function holdHead(head: BlockSummary) { - heldHeads.set(head.hash, head); + function holdCount(head: BlockSummary) { + heldCounts.set(head.hash, null); const timer = window.setTimeout(() => { - holdTimers.delete(timer); - const block = heldHeads.get(head.hash) ?? head; - heldHeads.delete(head.hash); - setBlocks((current) => mergeBlockWindow(current, [block], HOME_BLOCK_LIMIT)); - }, HEAD_DISPLAY_DELAY_MS); - holdTimers.add(timer); + revealTimers.delete(timer); + const filled = heldCounts.get(head.hash); + heldCounts.delete(head.hash); + if (filled) setBlocks((current) => fillBlockDetails(current, [filled])); + }, TX_COUNT_REVEAL_MS); + revealTimers.add(timer); } function queueDetails(blockNumber: number, attempts = 0) { @@ -310,10 +314,12 @@ function BlocksTab({ controller.signal, ); if (cancelled) return; - for (const block of filled) { - if (heldHeads.has(block.hash)) heldHeads.set(block.hash, block); - } - setBlocks((current) => fillBlockDetails(current, filled)); + const visible = filled.map((block) => { + if (!heldCounts.has(block.hash)) return block; + heldCounts.set(block.hash, block); + return { ...block, transactionCount: null }; + }); + setBlocks((current) => fillBlockDetails(current, visible)); } catch { if (!cancelled) { for (const [blockNumber, attempts] of batch) queueDetails(blockNumber, attempts + 1); @@ -334,8 +340,11 @@ function BlocksTab({ return; } stopPolling(); - holdHead(head); - if (head.transactionCount === null) queueDetails(head.number); + setBlocks((current) => mergeBlockWindow(current, [head], HOME_BLOCK_LIMIT)); + if (head.transactionCount === null) { + holdCount(head); + queueDetails(head.number); + } setFetchFailed(false); setLoading(false); }; @@ -355,7 +364,7 @@ function BlocksTab({ source?.close(); stopPolling(); window.clearTimeout(fillTimer); - for (const timer of holdTimers) window.clearTimeout(timer); + for (const timer of revealTimers) window.clearTimeout(timer); }; }, [chain, showShadowDelta]); From 201e5d3479f524697069a45bda77623ee8697968 Mon Sep 17 00:00:00 2001 From: Derek Guenther Date: Thu, 5 Nov 2026 18:20:04 -0500 Subject: [PATCH 5/5] feat(internal-explorer): count streamed transactions on the server Replace the per-page /blocks-by-numbers fill with one eth_getBlockTransactionCountByHash per head in the shared server subscription. Blocks reach the page complete and in arrival order, so the list has no placeholder rows and pages make no requests while the stream is live. A lookup that fails or exceeds 1s sends the block without a count rather than stalling the stream. --- app/api/internal-explorer/block-list.ts | 9 ++- app/api/internal-explorer/head-stream.test.ts | 68 ++++++++++++++++--- app/api/internal-explorer/head-stream.ts | 38 +++++++++-- app/api/internal-explorer/heads/route.ts | 8 +-- .../library/block-window.test.ts | 24 +------ app/internal-explorer/library/block-window.ts | 15 +--- app/internal-explorer/page.tsx | 67 +----------------- 7 files changed, 106 insertions(+), 123 deletions(-) diff --git a/app/api/internal-explorer/block-list.ts b/app/api/internal-explorer/block-list.ts index 6c136fa8..5dc56d1d 100644 --- a/app/api/internal-explorer/block-list.ts +++ b/app/api/internal-explorer/block-list.ts @@ -13,7 +13,7 @@ export interface BlockSummary { timestamp: number; /** Unix milliseconds from Denim's `timestampMs`; null on blocks or nodes without it. */ timestampMs: number | null; - /** Null for streamed heads until the client fills it from /blocks-by-numbers. */ + /** Null when unknown, such as a streamed head whose count lookup failed. */ transactionCount: number | null; gasUsed: string | null; gasLimit: string | null; @@ -124,6 +124,13 @@ export function parseBlockSummary(result: unknown): BlockSummary | null { }; } +export async function fetchTransactionCountByHash( + rpcUrl: string, + hash: string, +): Promise { + return hexToNumber(await rpcCall(rpcUrl, 'eth_getBlockTransactionCountByHash', [hash])); +} + async function fetchBlockByNumber( rpcUrl: string, blockNumber: number, diff --git a/app/api/internal-explorer/head-stream.test.ts b/app/api/internal-explorer/head-stream.test.ts index 68b7de1d..95c19b39 100644 --- a/app/api/internal-explorer/head-stream.test.ts +++ b/app/api/internal-explorer/head-stream.test.ts @@ -35,11 +35,14 @@ class FakeSocket implements HeadSocket { } } -function setup() { +function setup( + fetchTransactionCount: (hash: string) => Promise = async () => 7, +) { const sockets: FakeSocket[] = []; const onEnded = vi.fn(); const hub = new HeadHub({ wsUrl: 'wss://node.example:8546', + fetchTransactionCount, onEnded, connect: () => { const socket = new FakeSocket(); @@ -72,46 +75,95 @@ describe('HeadHub', () => { ]); }); - test('delivers each head to every subscriber straight from the header', () => { - const { hub, sockets } = setup(); + test('delivers each head to every subscriber with its header and transaction count', async () => { + const fetchCount = vi.fn(async () => 7); + const { hub, sockets } = setup(fetchCount); const received: BlockSummary[][] = [[], []]; hub.subscribe({ onBlock: (block) => received[0].push(block), onEnd: () => {} }); hub.subscribe({ onBlock: (block) => received[1].push(block), onEnd: () => {} }); sockets[0].emit('open'); sockets[0].emit('message', headNotification(HASH, { timestampMs: '0x1a3185c50c8' })); - assert.equal(received[1].length, 1); + await vi.waitFor(() => assert.equal(received[1].length, 1)); + assert.deepEqual(fetchCount.mock.calls, [[HASH]]); assert.deepEqual(received[0][0], { hash: HASH, number: 16, timestamp: 1_800_000_000, timestampMs: 1_800_000_000_200, - transactionCount: null, + transactionCount: 7, gasUsed: '21000', gasLimit: null, baseFeePerGas: null, }); }); - test('leaves timestampMs null when the header lacks it', () => { + test('leaves timestampMs null when the header lacks it', async () => { const { hub, sockets } = setup(); const received: BlockSummary[] = []; hub.subscribe({ onBlock: (block) => received.push(block), onEnd: () => {} }); sockets[0].emit('open'); sockets[0].emit('message', headNotification(HASH)); + await vi.waitFor(() => assert.equal(received.length, 1)); assert.equal(received[0].timestampMs, null); assert.equal(received[0].timestamp, 1_800_000_000); }); - test('ignores notifications without a valid block hash', () => { - const { hub, sockets } = setup(); + test('delivers heads in arrival order even when lookups finish out of order', async () => { + const resolvers = new Map void>(); + const { hub, sockets } = setup( + (hash) => new Promise((resolve) => resolvers.set(hash, resolve)), + ); + const received: string[] = []; + hub.subscribe({ onBlock: (block) => received.push(block.hash), onEnd: () => {} }); + sockets[0].emit('open'); + const first = `0x${'01'.repeat(32)}`; + const second = `0x${'02'.repeat(32)}`; + sockets[0].emit('message', headNotification(first)); + sockets[0].emit('message', headNotification(second)); + + resolvers.get(second)?.(1); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.deepEqual(received, []); + resolvers.get(first)?.(1); + await vi.waitFor(() => assert.deepEqual(received, [first, second])); + }); + + test('delivers a head without a count when the lookup fails or stalls', async () => { + vi.useFakeTimers(); + try { + const stalled = `0x${'03'.repeat(32)}`; + const { hub, sockets } = setup((hash) => + hash === stalled ? new Promise(() => {}) : Promise.reject(new Error('rpc down')), + ); + const received: BlockSummary[] = []; + hub.subscribe({ onBlock: (block) => received.push(block), onEnd: () => {} }); + sockets[0].emit('open'); + sockets[0].emit('message', headNotification(HASH)); + sockets[0].emit('message', headNotification(stalled)); + + await vi.advanceTimersByTimeAsync(1_000); + assert.deepEqual( + received.map((block) => block.transactionCount), + [null, null], + ); + } finally { + vi.useRealTimers(); + } + }); + + test('ignores notifications without a valid block hash', async () => { + const fetchCount = vi.fn(async () => 7); + const { hub, sockets } = setup(fetchCount); const onBlock = vi.fn(); hub.subscribe({ onBlock, onEnd: () => {} }); sockets[0].emit('open'); sockets[0].emit('message', headNotification('not-a-hash')); sockets[0].emit('message', { jsonrpc: '2.0', id: 1, result: '0x1' }); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(fetchCount.mock.calls.length, 0); assert.equal(onBlock.mock.calls.length, 0); }); diff --git a/app/api/internal-explorer/head-stream.ts b/app/api/internal-explorer/head-stream.ts index 9839f25d..0e9d6a76 100644 --- a/app/api/internal-explorer/head-stream.ts +++ b/app/api/internal-explorer/head-stream.ts @@ -1,11 +1,13 @@ // Live block heads for the Internal Explorer. One upstream `eth_subscribe` -// newHeads WebSocket per endpoint is shared by every connected browser. Heads -// are forwarded straight from the header: `timestampMs` when the upstream sends -// it, and no transaction count, which the client fills in separately. -// Server-only. -import { parseBlockSummary, type BlockSummary } from './block-list'; +// newHeads WebSocket per endpoint is shared by every connected browser. Each +// head is summarized from its header (`timestampMs` when the upstream sends +// it) plus one transaction-count lookup, since newHeads carries no transaction +// list. Blocks are delivered complete and in arrival order. Server-only. +import { fetchTransactionCountByHash, parseBlockSummary, type BlockSummary } from './block-list'; const CONNECT_TIMEOUT_MS = 10_000; +// Delivery is ordered, so a stuck lookup would hold back every later head. +const COUNT_TIMEOUT_MS = 1_000; const SUBSCRIBE_REQUEST_ID = 1; const BLOCK_HASH = /^0x[0-9a-f]{64}$/i; @@ -26,6 +28,7 @@ export interface HeadSubscriber { export interface HeadHubOptions { wsUrl: string; + fetchTransactionCount: (hash: string) => Promise; connect?: (url: string) => HeadSocket; /** Called once the hub has ended, so a registry can drop it. */ onEnded?: () => void; @@ -36,6 +39,7 @@ export class HeadHub { private socket: HeadSocket | null = null; private connectTimer: ReturnType | undefined; private ended = false; + private delivery: Promise = Promise.resolve(); constructor(private readonly options: HeadHubOptions) {} @@ -108,7 +112,22 @@ export class HeadHub { const block = parseBlockSummary(header); if (!block) return; - for (const subscriber of this.subscribers) subscriber.onBlock(block); + + const count = this.lookUpTransactionCount(hash); + this.delivery = this.delivery.then(async () => { + const transactionCount = await count; + if (this.ended) return; + for (const subscriber of this.subscribers) subscriber.onBlock({ ...block, transactionCount }); + }); + } + + private lookUpTransactionCount(hash: string): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(null), COUNT_TIMEOUT_MS); + }); + const lookup = this.options.fetchTransactionCount(hash).catch(() => null); + return Promise.race([lookup, timeout]).finally(() => clearTimeout(timer)); } private end() { @@ -130,11 +149,16 @@ export class HeadHub { const hubs = new Map(); /** Subscribe to live heads for a chain, sharing one upstream socket per WebSocket URL. */ -export function subscribeToHeads(wsUrl: string, subscriber: HeadSubscriber): () => void { +export function subscribeToHeads( + wsUrl: string, + rpcUrl: string, + subscriber: HeadSubscriber, +): () => void { let hub = hubs.get(wsUrl); if (!hub) { const created = new HeadHub({ wsUrl, + fetchTransactionCount: (hash) => fetchTransactionCountByHash(rpcUrl, hash), onEnded: () => { if (hubs.get(wsUrl) === created) hubs.delete(wsUrl); }, diff --git a/app/api/internal-explorer/heads/route.ts b/app/api/internal-explorer/heads/route.ts index d1b7d433..1ba367e8 100644 --- a/app/api/internal-explorer/heads/route.ts +++ b/app/api/internal-explorer/heads/route.ts @@ -1,5 +1,5 @@ import { resolveExplorerChainFromRequest } from '../chain'; -import { getWsUrl } from '../config'; +import { getRpcUrl, getWsUrl } from '../config'; import { explorerDisabledResponse } from '../guard'; import { subscribeToHeads } from '../head-stream'; @@ -11,8 +11,8 @@ const KEEPALIVE_MS = 15_000; // EventSource waits this long before reconnecting after the stream ends. const RECONNECT_MS = 3_000; -// Server-Sent Events stream of new canonical blocks (header-only BlockSummary -// JSON per `data:` frame), fed by the chain's newHeads WebSocket. 404 when the chain +// Server-Sent Events stream of new canonical blocks (BlockSummary JSON per +// `data:` frame), fed by the chain's newHeads WebSocket. 404 when the chain // has no TIPS__WS_URL, so the client falls back to polling /blocks. export async function GET(request: Request) { const disabled = explorerDisabledResponse(); @@ -49,7 +49,7 @@ export async function GET(request: Request) { send(`retry: ${RECONNECT_MS}\n\n`); request.signal.addEventListener('abort', finish); - unsubscribe = subscribeToHeads(wsUrl, { + unsubscribe = subscribeToHeads(wsUrl, getRpcUrl(chain), { onBlock: (block) => send(`data: ${JSON.stringify(block)}\n\n`), onEnd: () => finish(), }); diff --git a/app/internal-explorer/library/block-window.test.ts b/app/internal-explorer/library/block-window.test.ts index 2b90dc8b..da16230a 100644 --- a/app/internal-explorer/library/block-window.test.ts +++ b/app/internal-explorer/library/block-window.test.ts @@ -2,7 +2,7 @@ import assert from 'node:assert/strict'; import { describe, test } from 'vitest'; -import { fillBlockDetails, mergeBlockWindow } from './block-window'; +import { mergeBlockWindow } from './block-window'; import type { BlockSummary } from './types'; function block(number: number, hash = `0x${number.toString(16)}`): BlockSummary { @@ -53,25 +53,3 @@ describe('mergeBlockWindow', () => { assert.equal(merged.timestampMs, 1_800_000_000_400); }); }); - -describe('fillBlockDetails', () => { - test('replaces header-only rows with their fetched summaries', () => { - const header = { ...block(12), transactionCount: null }; - const fetched = { ...block(12), transactionCount: 7, timestampMs: 1_800_000_000_400 }; - const [filled, untouched] = fillBlockDetails([header, block(11)], [fetched]); - assert.equal(filled.transactionCount, 7); - assert.equal(filled.timestampMs, 1_800_000_000_400); - assert.equal(untouched.transactionCount, 2); - }); - - test('ignores a fetched block that no longer matches the row after a reorg', () => { - const header = { ...block(12, '0xnew'), transactionCount: null }; - const [row] = fillBlockDetails([header], [{ ...block(12, '0xold'), transactionCount: 7 }]); - assert.equal(row.hash, '0xnew'); - assert.equal(row.transactionCount, null); - }); - - test('does not add blocks that have left the window', () => { - assert.deepEqual(numbers(fillBlockDetails([block(20)], [block(5)])), [20]); - }); -}); diff --git a/app/internal-explorer/library/block-window.ts b/app/internal-explorer/library/block-window.ts index b68fdf08..4ebe8ca4 100644 --- a/app/internal-explorer/library/block-window.ts +++ b/app/internal-explorer/library/block-window.ts @@ -28,17 +28,4 @@ export function mergeBlockWindow( ); } return [...byNumber.values()].sort((lhs, rhs) => rhs.number - lhs.number).slice(0, limit); -} - -/** - * Replace rows with their fetched full summaries. Only rows whose hash still - * matches are replaced, so a fetch that raced a reorg can't swap in a - * different block; the reorged head arrives on the stream on its own. - */ -export function fillBlockDetails( - current: readonly BlockSummary[], - fetched: readonly BlockSummary[], -): BlockSummary[] { - const byHash = new Map(fetched.map((block) => [block.hash, block])); - return current.map((block) => byHash.get(block.hash) ?? block); -} +} \ No newline at end of file diff --git a/app/internal-explorer/page.tsx b/app/internal-explorer/page.tsx index 67400498..c54cd452 100644 --- a/app/internal-explorer/page.tsx +++ b/app/internal-explorer/page.tsx @@ -24,7 +24,7 @@ import { formatSignedInteger, formatSignedPct, } from './library/explorer-format'; -import { fillBlockDetails, mergeBlockWindow } from './library/block-window'; +import { mergeBlockWindow } from './library/block-window'; import { formatGasPrice, formatHexValue, @@ -216,16 +216,6 @@ function BlockRow({ } const HOME_BLOCK_LIMIT = 10; -// Streamed heads have no transaction count; collect their heights for this long -// and fill them with one /blocks-by-numbers request. -const DETAIL_FILL_MS = 200; -// A streamed head's transaction count is revealed this long after the head -// arrives, so counts appear a steady number of rows below the top instead of -// whenever each fill happens to land. Must exceed DETAIL_FILL_MS plus a fill -// round trip. -const TX_COUNT_REVEAL_MS = 600; -// The WebSocket node can be ahead of the HTTP one, so a fill may miss a block. -const DETAIL_FILL_ATTEMPTS = 3; function BlocksTab({ chain, @@ -278,55 +268,6 @@ function BlocksTab({ pollTimer = undefined; } - const pendingDetails = new Map(); - let fillTimer: number | undefined; - // Fetched summaries for streamed heads whose transaction count is not yet - // due; null until the fill lands. - const heldCounts = new Map(); - const revealTimers = new Set(); - - function holdCount(head: BlockSummary) { - heldCounts.set(head.hash, null); - const timer = window.setTimeout(() => { - revealTimers.delete(timer); - const filled = heldCounts.get(head.hash); - heldCounts.delete(head.hash); - if (filled) setBlocks((current) => fillBlockDetails(current, [filled])); - }, TX_COUNT_REVEAL_MS); - revealTimers.add(timer); - } - - function queueDetails(blockNumber: number, attempts = 0) { - if (attempts >= DETAIL_FILL_ATTEMPTS) return; - pendingDetails.set(blockNumber, attempts); - fillTimer ??= window.setTimeout(() => void fillDetails(), DETAIL_FILL_MS); - } - - async function fillDetails() { - fillTimer = undefined; - const batch = [...pendingDetails]; - pendingDetails.clear(); - if (batch.length === 0) return; - try { - const filled = await explorerApi.blocksByNumbers( - chain, - batch.map(([blockNumber]) => blockNumber), - controller.signal, - ); - if (cancelled) return; - const visible = filled.map((block) => { - if (!heldCounts.has(block.hash)) return block; - heldCounts.set(block.hash, block); - return { ...block, transactionCount: null }; - }); - setBlocks((current) => fillBlockDetails(current, visible)); - } catch { - if (!cancelled) { - for (const [blockNumber, attempts] of batch) queueDetails(blockNumber, attempts + 1); - } - } - } - void load(); const source = @@ -341,10 +282,6 @@ function BlocksTab({ } stopPolling(); setBlocks((current) => mergeBlockWindow(current, [head], HOME_BLOCK_LIMIT)); - if (head.transactionCount === null) { - holdCount(head); - queueDetails(head.number); - } setFetchFailed(false); setLoading(false); }; @@ -363,8 +300,6 @@ function BlocksTab({ controller.abort(); source?.close(); stopPolling(); - window.clearTimeout(fillTimer); - for (const timer of revealTimers) window.clearTimeout(timer); }; }, [chain, showShadowDelta]);