pipeline

pipeline 是一个 Go 库，不是独立进程。它被编译进执行客户端，通过 EVM 追踪钩子采集区块执行数据，序列化后写入 S3，并向 Kafka 发布区块变更通知。

项	值
仓库	Chaintable/pipeline
语言	Go 1.23+
许可证	Apache-2.0
主要依赖	go-ethereum、AWS SDK v2、kafka-go、etcd client v3

模块

目录	职责	关键文件
`tracer/`	EVM 钩子与数据采集	`pipeline_tracer.go`、`call_tracer.go`、`prestate_tracer.go`、`rpc_tracer.go`
`processor/`	序列化、S3 上传、Kafka 发布	`push.go`、`serializer.go`
`types/`	数据结构定义	`Block`、`Transaction`、`Trace`、`Event`、`BlockStorageDiff`
`leader/`	Leader 选举（手动 / etcd）	`manager.go`、`leader_failover.go`
`writer/`	写节点注册（etcd 租约）	`registry.go`
`util/`	S3、Kafka、编解码	`s3.go`、`kafka.go`、`codec.go`
`metrics/`	可观测性指标	`metrics.go`

两种集成模式

Live Tracer
RPC Tracer

追踪器嵌入区块执行流程，实时采集并上传。零额外延迟，但需要修改执行客户端的核心代码。

区块执行 → PipelineTracer（EVM 钩子）→ CallTracer + PrestateTracer
    → Processor（JSON+gzip / RLP）→ S3 双桶 + Kafka

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

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

pipelineTracer := tracer.NewPipelineTracer(configJSON)

通过 trace_debankBlock 按需重放区块，返回完整输出。不需要改动核心 EVM，适合无法接入 live tracer 的客户端（如 Reth）。

RPC 请求 → 区块重放（RPCTracer）→ CallTracer + PrestateTracer
    → 返回 DebankOutPut

rpcTracer := tracer.NewRPCTracer(configJSON)
// 用 tracer 的钩子重放区块后
output := rpcTracer.GetOutPut(originRoot, root, destructs, accounts, storages, codes)

采集机制

CallTracer

维护调用栈并构建调用树。OnEnter 压栈，OnExit 出栈并挂到父帧下。

存储变更标记：识别 SSTORE 指令，把 StorageChange 向上传播到祖先调用
失败隔离：调用失败时标记自身及所有子调用的 ParentFailed，展平后进入 ErrorTraces / ErrorEvents
ID 生成：trace ID 为 hash(tx_id, parent_trace_id, position)，event ID 为 hash(parent_trace_id, position)，保证跨区块可复现

StateDiff 的两条路径

方式	触发	说明
Commit 集转换（默认）	`OnCommit` 拿到 StateDB 的提交集	直接、准确，开销最低
PrestateTracer	配置 `enable_prestate_tracer`	在 `OnOpcode` 预加载 `SLOAD` / `SSTORE` / `BALANCE` 涉及的状态，`OnTxEnd` 计算差异

第二种用于没有 commit hook 的客户端。originRoot == root 时不生成差异对象。

上传与发布

OnCommit 里并发执行四路上传：

uploadBlockHeader     → 内部桶  {chainID}[/{version}]/{blockHash}/block       JSON+gzip
uploadBlockDiff       → 内部桶  {chainID}[/{version}]/{stateRoot}/stateDiff   RLP
uploadBlockFile       → 外部桶  {chainID}[/{version}]/{blockHash}             JSON+gzip
uploadBlockValidation → 外部桶  {chainID}[/{version}]/{height}/{blockHash}    JSON+gzip

编码按消费者选择：

StateDiff 用 RLP：只被内部组件消费，体积比 JSON 小得多。
BlockFile 用 JSON：外部消费者需要能直接解析。

配置 s3_temp_dir 后启用本地缓存：先落盘再异步上传，上传成功后删除本地文件。 S3 返回 5xx 时按退避重试，重试次数计入 pipeline/s3_upload_retry。

Leader 选举

多个写节点可以同时运行 pipeline，但只有 Leader 向 Kafka 发布通知，所有实例都上传 S3。S3 对象以哈希和 state root 为键，重复上传是幂等的。

etcd 自动选举
手动指定

配置 etcd_endpoints，把 is_backup 留空。

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

键不存在时抢占，存在时 watch。
键被删除后随机退避再抢，避免惊群。
成为 Leader 前等待 grace_period（默认 10 秒），让上一任 Leader 完成收尾。

配置 is_backup 为 true 或 false，不配置 etcd_endpoints。角色固定不变。两个参数互斥：同时配置或都不配置会在启动时 log.Crit 退出。

配置了 etcd 时还会启用写节点注册：以租约方式写入 {chainID}[/{version}]/writers/{nodeID}，节点故障时自动清理。nodex-proxy 的 /{chainId}/writers 管理接口读的就是这份数据。

配置

传给 NewPipelineTracer 的 JSON（即 geth 的 --vmtrace.jsonconfig）：

字段	默认	说明
`region`	-	AWS 区域
`node_x_bucket`	-	内部桶名
`chain_table_bucket`	-	外部桶名
`brokers`	-	Kafka broker 列表
`topic`	`nodex_pipeline_{chainID}[_{version}]`	内部 topic
`version`	空	版本命名空间，影响 topic、S3 键和 etcd 键
`s3_temp_dir`	空	本地上传缓存目录，留空则直传
`enable_prestate_tracer`	`false`	改用 prestate tracer 采集状态差异
`is_backup`	`nil`	`nil` = etcd 自动选举；`true`/`false` = 手动模式
`etcd_endpoints`	-	etcd 端点，与 `is_backup` 互斥
`election_key`	`{chainID}[/{version}]/writers/leader`	选举键
`node_id`	主机名	节点唯一标识
`grace_period`	`10`	成为 Leader 前的等待秒数
`writer_registry_ttl`	`10`	写节点注册租约秒数

指标

通过 go-ethereum 的 metrics 体系暴露。

指标	类型	用途
`pipeline/block_num`	Gauge	最新处理的区块高度
`pipeline/block_time`	Gauge	最新区块时间戳
`pipeline/latest_uploaded_block_number`	Gauge	最新已上传高度，与 `block_num` 的差值反映上传积压
`pipeline/block_process`	Timer	单块处理总耗时
`pipeline/tx_execution`	Timer	交易执行耗时
`pipeline/block_header_upload` 等	Timer	各类对象的上传耗时
`pipeline/block_push`	Timer	Kafka 发布耗时
`pipeline/s3_upload_retry`	Counter	S3 重试次数
`pipeline/kafka_write/{topic}`	Timer	按 topic 的写入耗时

开发

make build   # go build ./...
make test    # go test -count=1 -shuffle=on ./...
make race    # 竞态检测
make lint    # golangci-lint
make ci      # 提交前跑这个

接口契约

Kafka 消息与 S3 键的精确格式。

接入新链

把 pipeline 适配到 Geth、旧版 Geth 或 Reth 分叉。

开始使用

核心概念

架构

组件

指南

参考

模块

两种集成模式

采集机制

CallTracer

StateDiff 的两条路径

上传与发布

Leader 选举

配置

指标

开发

相关文档

接口契约

接入新链

​模块

​两种集成模式

​采集机制

​CallTracer

​StateDiff 的两条路径

​上传与发布

​Leader 选举

​配置

​指标

​开发

​相关文档

接口契约

接入新链

模块

两种集成模式

采集机制

CallTracer

StateDiff 的两条路径

上传与发布

Leader 选举

配置

指标

开发

相关文档