> ## 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

> Tracing and distribution library embedded in the execution client: writes block execution data to S3 and Kafka

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](https://github.com/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

<Tabs>
  <Tab title="Live 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.

    ```text theme={null}
    Block execution → PipelineTracer (EVM hooks) → CallTracer + PrestateTracer
        → Processor (JSON+gzip / RLP) → S3 dual buckets + 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">
    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).

    ```text theme={null}
    RPC request → Block replay (RPCTracer) → CallTracer + PrestateTracer
        → Returns DebankOutPut
    ```

    ```go theme={null}
    rpcTracer := tracer.NewRPCTracer(configJSON)
    // After replaying the block with the tracer's hooks
    output := rpcTracer.GetOutPut(originRoot, root, destructs, accounts, storages, codes)
    ```
  </Tab>
</Tabs>

## 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:

```text theme={null}
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.

<Tabs>
  <Tab title="etcd automatic election">
    Configure `etcd_endpoints` and leave `is_backup` empty.

    ```go theme={null}
    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.
  </Tab>

  <Tab title="Manual assignment">
    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`.
  </Tab>
</Tabs>

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

```bash theme={null}
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
```

## Related documents

The repository has three more detailed documents:

* [`docs/architecture.md`](https://github.com/Chaintable/pipeline/blob/main/docs/architecture.md) — module details, data flow, deployment
* [`docs/protocol.md`](https://github.com/Chaintable/pipeline/blob/main/docs/protocol.md) — field-level definitions of all data types
* [`docs/integration-modes.md`](https://github.com/Chaintable/pipeline/blob/main/docs/integration-modes.md) — comparison of the two integration modes

<Columns cols={2}>
  <Card title="Interface contracts" icon="plug" href="/en/architecture/interfaces">
    The exact format of Kafka messages and S3 keys.
  </Card>

  <Card title="Adding a new chain" icon="git-branch-plus" href="/en/guides/new-chain">
    Adapt pipeline to Geth, legacy Geth, or Reth forks.
  </Card>
</Columns>