Data freshness monitoring

import { formatBlock } from '@subsquid/pipes'
import { solanaPortalSource, solanaRpcLatencyWatcher } from '@subsquid/pipes/solana'

/**
 * This example demonstrates how to track and compare block indexing latency
 * between the Subsquid Portal and external RPC providers.
 * It listens for new block heads and measures the time until blocks are
 * observed on the client side through both RPC endpoints and the Portal.
 *
 ******************************************************************************
 * ⚠️ Important:
 * - The measured values INCLUDE client-side network latency.
 * - For RPC, only the *arrival time* of the block is measured — this does NOT
 *   capture the node's internal processing or response latency if queried directly.
 *****************************************************************************
 *
 * In other words, the results represent end-to-end delays as experienced by the client,
 * not the pure Portal latency or RPC processing performance.
 */

async function main() {
  // Create a stream of new blocks from Solana mainnet portal
  const stream = solanaPortalSource({
    portal: 'https://portal.sqd.dev/datasets/solana-mainnet',
    outputs: new SolanaQueryBuilder()
      .addFields({ block: { number: true, hash: true, timestamp: true } })
      .includeAllBlocks({ from: 'latest' })
      .build(),
  }).pipe(
    solanaRpcLatencyWatcher({
      rpcUrl: ['https://api.mainnet-beta.solana.com'], // RPC endpoints to monitor
    }).pipe({
      profiler: { id: 'expose metrics' },
      transform: (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 }, data.rpc[0].portalDelayMs)
        }

        return data
      },
    }),
  )

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

    // Log block number and timestamp
    console.log(`-------------------------------------`)
    console.log(`BLOCK DATA: ${formatBlock(data.number)} / ${data.timestamp.toString()}`)
    // Log RPC latency table for the block
    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           │
└───┴─────────────────────────────────┴──────────────────────────┴───────────────┘

import { solanaPortalSource, solanaRpcLatencyWatcher } from '@subsquid/pipes/solana'
import { Registry, Counter } from 'prom-client'

const registry = new Registry()
const latencyCounter = new Counter({
  name: 'portal_latency_ms_total',
  help: 'Total Portal latency in milliseconds',
  registers: [registry],
})

const stream = solanaPortalSource({
  portal: 'https://portal.sqd.dev/datasets/solana-mainnet',
  outputs: new SolanaQueryBuilder()
    .addFields({ block: { number: true, hash: true, timestamp: true } })
    .includeAllBlocks({ from: 'latest' })
    .build(),
}).pipe(
  solanaRpcLatencyWatcher({
    rpcUrl: ['https://api.mainnet-beta.solana.com'],
  }).pipe({
    transform: (data) => {
      if (!data) return

      // Record latency metrics
      for (const rpc of data.rpc) {
        latencyCounter.inc(rpc.portalDelayMs)
      }

      return data
    },
  }),
)

// Expose metrics endpoint
import http from 'http'
const server = http.createServer(async (req, res) => {
  if (req.url === '/metrics') {
    res.setHeader('Content-Type', registry.contentType)
    res.end(await registry.metrics())
  } else {
    res.end('OK')
  }
})
server.listen(9090)

for await (const { data } of stream) {
  // Process blocks
}