| name | etl-rs |
| description | Rust ETL workspace under etl-rs/ — workspace crates (types, config, sources with BlockSource trait, sinks for Redis/Neo4j/Postgres/ClickHouse, pipeline retry/DLQ, consumer, ingest, worker, clickhouse), data-flow diagram (Etherscan → Redis streams → Neo4j+PG), `ingest` one-shot CLI subcommands (block, address, reprocess-failed, targeted addresses/neighborhood), `worker` long-running binary three tasks (A targeted_queue BRPOP, B refresh known_labels, C XREADGROUP stream consumer → graph+PG), build commands, and how to reset a stuck consumer group via XGROUP SETID. Invoke when touching anything under etl-rs/, modifying the ingest or worker binaries, changing Redis stream consumer logic, or debugging the ETL pipeline. |
Rust ETL Workers (etl-rs/)
Two long-running binaries plus one CLI, communicating via Redis:
[Etherscan API]
│ ← ingest binary (one-shot CLI) / worker task A (continuous)
▼
[Redis] streams: ingested_txs / ingested_traces / ingested_transfers
list: ingest:targeted_queue (BRPOPped by worker task A)
│ ← worker task C (stream consumer)
▼
[Neo4j + PostgreSQL]
Workspace crates:
| Crate | Role |
|---|
types | Shared value types (Transaction, Trace, Transfer) |
config | Env-based Config + ProcessConfig |
sources | BlockSource trait + Etherscan / Alchemy / Mock impls |
sinks | Redis Streams writer/consumer, Neo4j/Postgres/ClickHouse writers |
pipeline | Retry, DLQ, shutdown handle, progress reporters |
consumer | read_batch + process_read_batch (used by worker task C) |
ingest | ingest binary — one-shot CLI for backfills / ad-hoc targeted fetches |
worker | worker binary — long-running, three tokio tasks (targeted queue + refresh + stream consumer) |
clickhouse | clickhouse-process binary — independent OLAP consumer group |
ingest CLI:
ingest block --start N --end M [--with-traces] [--with-transfers]
ingest address 0x... [--with-traces] [--with-transfers]
ingest reprocess-failed --source {etherscan,alchemy}
ingest targeted addresses --addrs 0xaaa,0xbbb
ingest targeted neighborhood 0xseed --hops 1
worker binary:
worker # continuous — task A (BRPOP targeted_queue),
# task B (refresh known_labels + high-risk),
# task C (stream consumer → Neo4j + Postgres)
Build:
cd etl-rs
cargo build --release
Reset stuck consumer group:
docker exec chain-analysis-redis redis-cli XGROUP SETID ingested_txs chain-analysis-process 0
(The consumer group name is still chain-analysis-process for
backwards compatibility — only the binary name changed.)
