Compute with BlockX

BlockDB handles storage; BlockX handles compute. When a single block carries a lot of data and blocks are produced frequently, single-process sequential handling cannot keep up with the block rate (on BSC, for example, you have only a few seconds per block). BlockX runs each block’s rows fully in parallel, keeping your business logic in sync with block production.

The most typical scenario

Most on-chain data processing follows the same pipeline:

Subscribe to an upstream block table (for example chain.trace.eth)
        │
        ▼
Run your function on every row of each block
        │
        ▼
Write the function's results to your own target block table

The most typical upstream is a block trace table such as chain.trace.eth — one block holds hundreds or thousands of traces, which is a natural fit for fully parallel compute. And this pipeline uses the same configuration and the same function for real-time subscription and historical block backfill.

Wiring the pipeline with Pipeline

BlockX’s high-level wrapper Pipeline packages “source table → function → target table” into one object and exposes only two verbs:

backfill(block_start, block_end): batch backfill of historical blocks.
update(): catch up on historical gaps first, then switch to real-time subscription once caught up.

Both reuse the same triggers, the same function, and the same target_table.

from blockx import Pipeline, Trigger, SOURCE_ROW

# 1) Write a stateless function: one source row dict in; return a dict to write it to the
#    target table, or None to drop it. (Parameter order matches the positions in params;
#    the source row is a SOURCE_ROW placeholder.)
def trace_to_token_transfer(chain_id, record):
    value = float(record.get("value") or 0)
    if value <= 0 or not record.get("tx_id"):
        return None              # filter out invalid rows
    return {
        "id":        record["id"],
        "from_addr": record["from_addr"],
        "to_addr":   record["to_addr"],
        "value":     value,
        "tx_id":     record["tx_id"],
    }

# 2) Assemble the Pipeline: subscribe to chain.trace.eth, write to token.token_transfer.eth
pipe = Pipeline(
    triggers=[
        Trigger(
            table="chain.trace.eth",
            func=trace_to_token_transfer,
            params=["eth", SOURCE_ROW],   # chain_id="eth"; the second argument is the source row
        ),
    ],
    target_table="token.token_transfer.eth",
)

# 3) Historical backfill: from 18000000 up to the upstream's current consensus height
backfill_result = pipe.backfill(block_start=18000000)

# 4) Real-time: fill gaps first, then switch to subscription and keep consuming new blocks
pipe.update()

backfill and update run on the same configuration:

	`backfill`	`update`
Purpose	Batch backfill of historical heights	Real-time consumption plus automatic gap filling
Data source	The height range `[block_start, block_end]`	The subscription stream plus catch-up
Function / Trigger / target table	Identical	Identical
Termination	Returns a summary after the range completes	Long-lived until stopped

So you maintain one function and one Pipeline configuration. How much history to backfill and how far real-time should catch up are both handled by the same object, which naturally rules out “backfill and real-time logic drifting apart”.

Low-level interfaces (advanced)

Pipeline is the high-level wrapper. If you need finer control, bypass it and compose the low-level BlockTableCallConfig (what to compute), BlockWriteHandler (where results go), and TaskBuilder (submit the task):

from blockx import BlockTableCallConfig, BlockWriteHandler, TaskBuilder
from blockdb import Block

block = Block(id="0x...", height=18000000, timestamp="2024-06-08T00:00:00Z")

config = BlockTableCallConfig(block=block, triggerSources=[
    {"table": "chain.trace.eth", "func": trace_to_token_transfer},
])
handler = BlockWriteHandler(targetTable="token.token_transfer.eth", block=block)

task = TaskBuilder.build(call_config=config, handler=handler)
result = task.submit()

For the complete API and configuration options, see the BlockX documentation. This page only covers the core path for working with BlockDB.

Getting started

Core concepts

Architecture

Guides

Reference

Compute with BlockX

Compute with BlockX

The most typical scenario

Wiring the pipeline with Pipeline

Low-level interfaces (advanced)

​Compute with BlockX

​The most typical scenario

​Wiring the pipeline with Pipeline

​Real-time and backfill share one configuration

​Low-level interfaces (advanced)

​Related topics

Compute with BlockX

The most typical scenario

Wiring the pipeline with Pipeline

Real-time and backfill share one configuration

Low-level interfaces (advanced)

Related topics