Portal batch processor

@subsquid/batch-processor connects a Portal-compatible data source to a database and an asynchronous data handler. Use it with the EVM Portal stream.

Development flow

Build a data source

Configure a chain-specific DataSourceBuilder, then call build().

const dataSource = new DataSourceBuilder()
  .setPortal('https://portal.sqd.dev/datasets/ethereum-mainnet')
  .setBlockRange({from: 18_000_000})
  .includeAllBlocks()
  .build()

Choose a database

Pass a Squid SDK database implementation such as TypeormDatabase, or implement the database interfaces.

Run the data handler

Use run() for a normal processor entry point. Use Processor when the caller needs to await completion or handle an error itself.

`run()` and `Processor`

Both APIs use the same source, database, handler, and optional RunOptions object.

Application entry point
Awaitable runner

import {run} from '@subsquid/batch-processor'

run(dataSource, db, async (ctx) => {
  for (const block of ctx.blocks) {
    // Decode and persist selected data.
  }
})

run() returns void. It installs top-level error handling and owns the processor process lifecycle, so it is intended for the main application entry point.

import {Processor} from '@subsquid/batch-processor'

const processor = new Processor(dataSource, db, async (ctx) => {
  for (const block of ctx.blocks) {
    // Decode and persist selected data.
  }
})

await processor.run()

Processor.run() returns Promise<void>. This form is useful in tests, scripts, and applications that manage their own top-level error handling.

The constructor signature is:

new Processor<Block, Store>(
  dataSource,
  database,
  dataHandler,
  options?,
)

Handler context

The handler receives this context:

interface DataHandlerContext<Block, Store> {
  store: Store
  blocks: Block[]
  isHead: boolean
}

store is supplied by the database transaction.
blocks contains the current block slice.
isHead is true when the processor has reached the current source head.

Portal-native handlers do not receive ctx.log or ctx._chain. Create a logger or RPC client explicitly when the handler needs one.

Finalized and hot-block databases

The database determines which stream the processor consumes:

Database capability	Source method	Database transaction
`supportsHotBlocks` is absent or `false`	`getFinalizedStream()`	`transact()`
`supportsHotBlocks: true`	`getStream()`	`transactHot2()`

See Custom Database for the full transaction contracts.

Prometheus

Pass a PrometheusServer in the optional fourth argument to run(), or as the fourth Processor constructor argument:

import {PrometheusServer, run} from '@subsquid/batch-processor'

const prometheus = new PrometheusServer()
prometheus.setPort(3000)

run(dataSource, db, handler, {prometheus})

Without an explicit server, PROCESSOR_PROMETHEUS_PORT takes precedence over PROMETHEUS_PORT. Metrics are disabled if neither variable is set.

Read a data source directly

The object returned by a Portal-native DataSourceBuilder.build() implements these methods:

Method	Result
`getHead()`	Latest source head, including unfinalized blocks
`getFinalizedHead()`	Latest finalized source head
`getStream(request)`	Hot-block-aware asynchronous block batches
`getFinalizedStream(request)`	Finalized asynchronous block batches
`getBlocksCountInRange?(range)`	Optional block-count helper used for progress estimates

A stream request has from, optional to, and optional parentHash fields. Each yielded batch contains blocks and may contain finalizedHead.

const finalizedHead = await dataSource.getFinalizedHead()

for await (const batch of dataSource.getFinalizedStream({
  from: 18_000_000,
  to: 18_000_100,
})) {
  for (const block of batch.blocks) {
    console.log(block.header.number)
  }
}

Direct iteration does not persist progress or open database transactions. Use run() or Processor when you need restart-safe indexing and hot-block rollback handling.

Overview

Pipes SDK

Squid SDK

Alternative clients

Development flow

`run()` and `Processor`

Handler context

Finalized and hot-block databases

Prometheus

Read a data source directly

​Development flow

​run() and Processor

​Handler context

​Finalized and hot-block databases

​Prometheus

​Read a data source directly

Development flow

`run()` and `Processor`

Handler context

Finalized and hot-block databases

Prometheus

Read a data source directly