consistency-checker

consistency-checker is not on the leafage-evm data path. It consumes the same Kafka notifications as leafage-evm, observes replica state, and then does three things:

Publishes confirmed block notifications to external consumers.
Marks fork blocks on S3.
Writes node health state into etcd.

Item	Value
Repository	Chaintable/consistency-checker
Language	Go 1.23+
License	Apache-2.0
Dependencies	Kafka (one internal, one external), etcd v3, S3, Pebble DB

What problem it solves

At the moment the write node publishes a notification, the replicas have not yet applied that block. If external consumers subscribed to the internal topic directly, they would receive the notification while leafage-evm still cannot serve the block. The checker adds a confirmation step in between: it waits until a sufficient ratio of replicas has actually caught up to that height before publishing to the external topic. External consumers therefore get a guarantee — when the notification arrives, querying the cluster is certain to return that block. Along the way, it also solves two other things:

Fork blocks are not marked on S3, because only an observer knows which chain is canonical.
nodex-proxy needs to know whether each node is currently healthy.

Processing flow

Inner Kafka (BlockChangeNotification)
  → message validation & deduplication
  → prefetch from S3 in parallel (BlockValidation + key list at the same height)
  → poll replicas until ready_ratio is met
  → write to Pebble DB (dual index)
  → mark fork blocks at the same height in S3
  → publish to Outer Kafka (drop first, then new)
  → align the singleton topic (Leader only, version mode)
  → recheck fork marks with a fresh LIST after publishing

The core implementation is Process in check/check.go. A few design points:

Prefetch and wait in parallel

The S3 prefetch and replica polling run at the same time. Polling usually takes tens to hundreds of milliseconds, which covers the S3 round trip; the failure path does not wait for the prefetch to finish, the goroutine wraps up on its own.

Idempotency and retries

When the whole Process fails, the Kafka offset is not committed, and the next round retries the same message.
Duplicate messages, and the case of “already processed through the publish step but the whole message not committed”, short-circuit to the alignment step and advance directly, avoiding retry deadlocks.
Destinations that were delivered successfully are recorded in delivered and skipped on retry, so nothing is delivered twice or missed.

Timeout semantics of replica polling

Polls eth_blockNumber on all replicas at check_interval_ms intervals until a ready_ratio share of replicas reaches the target height. A timeout counts as failure, and Process retries as a whole.

Config	Default	Purpose
`check_interval_ms`	`20`	Polling interval (milliseconds)
`ready_ratio`	`0.8`	Ratio of replicas required to be considered ready
`check_timeout_ms`	`2000`	Upper bound on the total time to wait for replicas to catch up (milliseconds)
`rpc_node_timeout_ms`	`5000`	Upper bound for a single RPC (milliseconds)

The check_num config option is deprecated and no longer takes effect.

Post-publish recheck

The prefetched key list at the same height is a snapshot taken at the start of Process; objects uploaded while waiting for replicas are not in it. After publishing the notification, it runs a fresh LIST and checks again. This step is not on the critical path; failures are only logged, and the periodic scan serves as a fallback.

Fork marking

The BlockValidation object in the S3 external bucket carries an is_fork field, which is always false when the write node uploads it — because at the moment of upload the write node does not yet know whether the block will end up on the canonical chain. The checker is responsible for writing it back. Three trigger paths:

Path	Trigger
`dropBlocks` rewrite	A reorg notification with `changeType: 2` is received
Same-height scan	When each new block is processed, list all objects at that height and mark the non-canonical ones as fork
Periodic scan	Every `fork_scan_interval_sec` (default 60 seconds), look back `fork_scan_lookback` (default 64) heights

A non-zero pipeline_fork_scan_rewrites_total means a mark was overwritten or previously failed, and deserves attention.

leafage-evm relies on this mark during S3 catch-up to resolve heights into canonical hashes. If the checker stalls for a long time, catch-up for new nodes slows down.

Node state maintenance

After each polling round, the result is written back to etcd (committed in a single transaction):

Key	Content
`{chainID}[/{version}]/nodes/{ip}_{port}`	The node’s `stateType`: 1 caught up, 2 lagging, 3 offline
`{chainID}[/{version}]/lastBlockNumber`	Current confirmed height

When a node is offline and has no lease, the key is deleted outright. The height is written only when it changes, to avoid needless etcd write amplification. These two keys are the entire basis for nodex-proxy routing: node pool membership comes from the former, and the State / Archive selection threshold comes from the latter.

Two modes

Determined jointly by the version and outer_version_new_block_topic config options.

Version mode
Legacy mode

Writes both the version topic and the singleton topic.

etcd keys use the {chainID}/{version}/ prefix
S3 paths include a version segment
Write access to the singleton topic requires election through an etcd distributed lock ({chainID}/outer_block_notice)
The Leader periodically compares {chainID}/version and releases the lock when the version does not match

When switching versions, the new Leader needs to align the singleton topic to its own progress: fast-forward, or roll back to the common ancestor and replay. This logic lives in AlignOuterSingleton and align.

Writes only the singleton topic, etcd keys use the {chainID}/ prefix, and no Leader election is needed.

Local storage

Pebble DB stores confirmed blocks with a dual index:

h{hash}   → BlockInfo (RLP encoded)
n{number} → hash

It serves external block queries, and is also used during fork scans to determine the canonical hash at a given height.

JSON-RPC API

The listen address is set by the listen config option (default :8663); the same port serves GET /metrics.

Method	Description
`getLatestBlock`	Latest confirmed block
`getBlockByHeight`	Query by height
`getBlockById`	Query by hash
`blockIsValid`	Whether a block is on the canonical chain

curl -X POST http://localhost:8663 \
  -H 'Content-Type: application/json' \
  -d '{"jsonrpc":"2.0","method":"getBlockByHeight","params":["0x100"],"id":1}'

{
  "id": 1,
  "jsonrpc": "2.0",
  "result": { "id": "0x...", "num": 12345, "validation_hash": 67890, "is_fork": false }
}

All errors return -39005.

Metrics

Metric	Description
`pipeline_node_info`	Node role information (labels `chain_id`, `role`)
`pipeline_block_num` / `pipeline_block_time`	Height and timestamp of the latest published block
`pipeline_replica_ready_wait_seconds`	Distribution of time spent waiting for replicas to become ready
`pipeline_replica_ready_timeouts_total`	Number of times replicas did not become ready within the timeout
`pipeline_process_publish_seconds`	Time from the start of processing until the external notification is written
`pipeline_block_ingress_to_outer_kafka_seconds`	End-to-end latency from the write node receiving the block to a successful write to external Kafka
`pipeline_fork_scan_rewrites_total` / `pipeline_fork_scan_errors_total`	Fork scan
`pipeline_drop_block_rewrite_failures_total`	Number of times drop block marking ultimately failed

Running

go build -o checker cmd/checker/*.go
./checker -config config.yml -listen :8663

See the configuration reference for config options.

Development

make build
make test
make race
make ci

check/critical_path_test.go covers the idempotency and retry semantics of the critical path; read it first before changing the Process flow.

Getting started

Core concepts

Architecture

Components

Guides

Reference

What problem it solves

Processing flow

Fork marking

Node state maintenance

Two modes

Local storage

JSON-RPC API

Metrics

Running

Development

​What problem it solves

​Processing flow

​Fork marking

​Node state maintenance

​Two modes

​Local storage

​JSON-RPC API

​Metrics

​Running

​Development

What problem it solves

Processing flow

Fork marking

Node state maintenance

Two modes

Local storage

JSON-RPC API

Metrics

Running

Development