Data freshness monitoring

import { formatBlock } from "@subsquid/pipes";
import { evmPortalStream, evmRpcLatencyWatcher } from "@subsquid/pipes/evm";
import { metricsServer } from "@subsquid/pipes/metrics/node";

async function main() {
  const stream = evmPortalStream({
    id: "indexing-latency",
    portal: "https://portal.sqd.dev/datasets/base-mainnet",
    outputs: evmRpcLatencyWatcher({
      rpcUrl: ["https://base.drpc.org", "https://base-rpc.publicnode.com"], // RPC endpoints to monitor
    }).pipe((data, { metrics }) => {
      if (!data) return; // Skip if no latency data

      // For each RPC endpoint, update the latency gauge metric
      for (const rpc of data.rpc) {
        metrics
          .gauge({
            name: "rpc_latency_ms",
            help: "RPC Latency in ms",
            labelNames: ["url"],
          })
          .set({ url: rpc.url }, rpc.portalDelayMs);
      }

      return data;
    }),

    metrics: metricsServer({ port: 9090 }),
  });

  // Iterate over the stream, logging block and RPC latency data
  for await (const { data } of stream) {
    if (!data) continue; // Skip if no block data

    console.log(`Block: ${formatBlock(data.number)} / ${data.timestamp}`);
    console.table(data.rpc);
  }
}

void main()

Block: 36,046,611 / Fri Sep 26 2025 14:29:29 GMT+0400
┌───┬─────────────────────────────────┬──────────────────────────┬───────────────┐
│   │ url                             │ receivedAt               │ portalDelayMs │
├───┼─────────────────────────────────┼──────────────────────────┼───────────────┤
│ 0 │ https://base.drpc.org           │ 2025-09-26T10:29:29.134Z │ 646           │
│ 1 │ https://base-rpc.publicnode.com │ 2025-09-26T10:29:29.130Z │ 642           │
└───┴─────────────────────────────────┴──────────────────────────┴───────────────┘