> ## Documentation Index
> Fetch the complete documentation index at: https://docs.leafage.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.

# pipeline

> 嵌入执行客户端的追踪与分发库：把区块执行数据写入 S3 和 Kafka

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

| 项    | 值                                                             |
| ---- | ------------------------------------------------------------- |
| 仓库   | [Chaintable/pipeline](https://github.com/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`                                                               |

## 两种集成模式

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

    ```text theme={null}
    区块执行 → PipelineTracer（EVM 钩子）→ CallTracer + PrestateTracer
        → Processor（JSON+gzip / RLP）→ S3 双桶 + Kafka
    ```

    ```go theme={null}
    tracer.InitPipeline(region, nodeXBucket, chainTableBucket,
        brokers, topic, bizChainID, version, s3TmpDir)

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

    pipelineTracer := tracer.NewPipelineTracer(configJSON)
    ```
  </Tab>

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

    ```text theme={null}
    RPC 请求 → 区块重放（RPCTracer）→ CallTracer + PrestateTracer
        → 返回 DebankOutPut
    ```

    ```go theme={null}
    rpcTracer := tracer.NewRPCTracer(configJSON)
    // 用 tracer 的钩子重放区块后
    output := rpcTracer.GetOutPut(originRoot, root, destructs, accounts, storages, codes)
    ```
  </Tab>
</Tabs>

## 采集机制

### 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` 里并发执行四路上传：

```text theme={null}
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 为键，重复上传是幂等的。

<Tabs>
  <Tab title="etcd 自动选举">
    配置 `etcd_endpoints`，把 `is_backup` 留空。

    ```go theme={null}
    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 完成收尾。
  </Tab>

  <Tab title="手动指定">
    配置 `is_backup` 为 `true` 或 `false`，不配置 `etcd_endpoints`。角色固定不变。

    两个参数互斥：同时配置或都不配置会在启动时 `log.Crit` 退出。
  </Tab>
</Tabs>

配置了 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 的写入耗时                   |

## 开发

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

## 相关文档

仓库内还有两份更细的文档：

* [`docs/architecture.md`](https://github.com/Chaintable/pipeline/blob/main/docs/architecture.md) — 模块细节、数据流、部署
* [`docs/protocol.md`](https://github.com/Chaintable/pipeline/blob/main/docs/protocol.md) — 所有数据类型的字段级定义
* [`docs/integration-modes.md`](https://github.com/Chaintable/pipeline/blob/main/docs/integration-modes.md) — 两种集成模式的对比

<Columns cols={2}>
  <Card title="接口契约" icon="plug" href="/architecture/interfaces">
    Kafka 消息与 S3 键的精确格式。
  </Card>

  <Card title="接入新链" icon="git-branch-plus" href="/guides/new-chain">
    把 pipeline 适配到 Geth、旧版 Geth 或 Reth 分叉。
  </Card>
</Columns>