pipeline

pipeline is a Go library, not a standalone process. It is compiled into the execution client, collects block execution data through EVM tracing hooks, serializes it and writes it to S3, and publishes block change notifications to Kafka.

Item	Value
Repository	Chaintable/pipeline
Language	Go 1.23+
License	Apache-2.0
Main dependencies	go-ethereum, AWS SDK v2, kafka-go, etcd client v3

Modules

Directory	Responsibility	Key files
`tracer/`	EVM hooks and data collection	`pipeline_tracer.go`, `call_tracer.go`, `prestate_tracer.go`, `rpc_tracer.go`
`processor/`	Serialization, S3 upload, Kafka publishing	`push.go`, `serializer.go`
`types/`	Data structure definitions	`Block`, `Transaction`, `Trace`, `Event`, `BlockStorageDiff`
`leader/`	Leader election (manual / etcd)	`manager.go`, `leader_failover.go`
`writer/`	Write node registration (etcd lease)	`registry.go`
`util/`	S3, Kafka, encoding and decoding	`s3.go`, `kafka.go`, `codec.go`
`metrics/`	Observability metrics	`metrics.go`

Two integration modes

Live Tracer
RPC Tracer

The tracer is embedded in the block execution flow and collects and uploads in real time. Zero extra latency, but it requires changes to the execution client’s core code.

Block execution → PipelineTracer (EVM hooks) → CallTracer + PrestateTracer
    → Processor (JSON+gzip / RLP) → S3 dual buckets + Kafka

tracer.InitPipeline(region, nodeXBucket, chainTableBucket,
    brokers, topic, bizChainID, version, s3TmpDir)

tracer.SetupLeaderElection(etcdEndpoints, electionKey, nodeID,
    version, isBackup, gracePeriod, writerConfig)

pipelineTracer := tracer.NewPipelineTracer(configJSON)

Replays blocks on demand through trace_debankBlock and returns the complete output. No changes to the core EVM are needed, which suits clients that cannot integrate the live tracer (such as Reth).

RPC request → Block replay (RPCTracer) → CallTracer + PrestateTracer
    → Returns DebankOutPut

rpcTracer := tracer.NewRPCTracer(configJSON)
// After replaying the block with the tracer's hooks
output := rpcTracer.GetOutPut(originRoot, root, destructs, accounts, storages, codes)

Collection mechanics

CallTracer

Maintains the call stack and builds the call tree. OnEnter pushes a frame; OnExit pops it and attaches it under the parent frame.

Storage change marking: recognizes the SSTORE instruction and propagates StorageChange up to ancestor calls
Failure isolation: when a call fails, marks itself and all sub-calls with ParentFailed; after flattening they go into ErrorTraces / ErrorEvents
ID generation: the trace ID is hash(tx_id, parent_trace_id, position) and the event ID is hash(parent_trace_id, position), which guarantees reproducibility across blocks

Two paths to StateDiff

Method	Trigger	Description
Commit set conversion (default)	`OnCommit` receives the StateDB’s commit set	Direct, accurate, lowest overhead
PrestateTracer	Configure `enable_prestate_tracer`	Preloads the state touched by `SLOAD` / `SSTORE` / `BALANCE` in `OnOpcode`, computes the diff in `OnTxEnd`

The second is for clients without a commit hook. No diff object is generated when originRoot == root.

Upload and publish

OnCommit runs four uploads concurrently:

uploadBlockHeader     → internal bucket  {chainID}[/{version}]/{blockHash}/block       JSON+gzip
uploadBlockDiff       → internal bucket  {chainID}[/{version}]/{stateRoot}/stateDiff   RLP
uploadBlockFile       → external bucket  {chainID}[/{version}]/{blockHash}             JSON+gzip
uploadBlockValidation → external bucket  {chainID}[/{version}]/{height}/{blockHash}    JSON+gzip

The encoding is chosen per consumer:

StateDiff uses RLP: it is consumed only by internal components, and is much smaller than JSON.
BlockFile uses JSON: external consumers need to parse it directly.

Configuring s3_temp_dir enables a local cache: data is written to disk first and uploaded asynchronously; the local file is deleted after a successful upload. When S3 returns 5xx, the upload is retried with backoff; the retry count goes into pipeline/s3_upload_retry.

Leader election

Multiple write nodes can run pipeline at the same time, but only the Leader publishes notifications to Kafka; all instances upload to S3. S3 objects are keyed by hash and state root, so duplicate uploads are idempotent.

etcd automatic election
Manual assignment

Configure etcd_endpoints and leave is_backup empty.

txn := client.Txn(ctx).
    If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).
    Then(clientv3.OpPut(key, nodeID)).
    Else(clientv3.OpGet(key))

Acquire the key when it does not exist; watch it when it does.
After the key is deleted, back off randomly before acquiring again, to avoid a thundering herd.
Wait grace_period (default 10 seconds) before becoming Leader, so the previous Leader can finish up.

Set is_backup to true or false and do not configure etcd_endpoints. The role is fixed.The two parameters are mutually exclusive: configuring both or neither exits at startup with log.Crit.

When etcd is configured, write node registration is also enabled: the node writes {chainID}[/{version}]/writers/{nodeID} with a lease, which is cleaned up automatically when the node fails. This is the data that nodex-proxy’s /{chainId}/writers admin endpoint reads.

Configuration

The JSON passed to NewPipelineTracer (that is, geth’s --vmtrace.jsonconfig):

Field	Default	Description
`region`	-	AWS region
`node_x_bucket`	-	Internal bucket name
`chain_table_bucket`	-	External bucket name
`brokers`	-	Kafka broker list
`topic`	`nodex_pipeline_{chainID}[_{version}]`	Internal topic
`version`	empty	Version namespace; affects the topic, S3 keys, and etcd keys
`s3_temp_dir`	empty	Local upload cache directory; leave empty to upload directly
`enable_prestate_tracer`	`false`	Use the prestate tracer to collect state diffs instead
`is_backup`	`nil`	`nil` = etcd automatic election; `true`/`false` = manual mode
`etcd_endpoints`	-	etcd endpoints, mutually exclusive with `is_backup`
`election_key`	`{chainID}[/{version}]/writers/leader`	Election key
`node_id`	hostname	Unique node identifier
`grace_period`	`10`	Seconds to wait before becoming Leader
`writer_registry_ttl`	`10`	Write node registration lease in seconds

Metrics

Exposed through go-ethereum’s metrics system.

Metric	Type	Purpose
`pipeline/block_num`	Gauge	Height of the latest processed block
`pipeline/block_time`	Gauge	Timestamp of the latest block
`pipeline/latest_uploaded_block_number`	Gauge	Latest uploaded height; the difference from `block_num` reflects the upload backlog
`pipeline/block_process`	Timer	Total processing time per block
`pipeline/tx_execution`	Timer	Transaction execution time
`pipeline/block_header_upload` etc.	Timer	Upload time for each object type
`pipeline/block_push`	Timer	Kafka publish time
`pipeline/s3_upload_retry`	Counter	S3 retry count
`pipeline/kafka_write/{topic}`	Timer	Write time per topic

Development

make build   # go build ./...
make test    # go test -count=1 -shuffle=on ./...
make race    # race detection
make lint    # golangci-lint
make ci      # run this before committing

The repository has three more detailed documents:

docs/architecture.md — module details, data flow, deployment
docs/protocol.md — field-level definitions of all data types
docs/integration-modes.md — comparison of the two integration modes

Interface contracts

The exact format of Kafka messages and S3 keys.

Adding a new chain

Adapt pipeline to Geth, legacy Geth, or Reth forks.

Getting started

Core concepts

Architecture

Components

Guides

Reference

Modules

Two integration modes

Collection mechanics

CallTracer

Two paths to StateDiff

Upload and publish

Leader election

Configuration

Metrics

Development

Interface contracts

Adding a new chain

​Modules

​Two integration modes

​Collection mechanics

​CallTracer

​Two paths to StateDiff

​Upload and publish

​Leader election

​Configuration

​Metrics

​Development

​Related documents

Interface contracts

Adding a new chain

Modules

Two integration modes

Collection mechanics

CallTracer

Two paths to StateDiff

Upload and publish

Leader election

Configuration

Metrics

Development

Related documents