レッスン2 — 本物の Production Indexer を読む — Tempo の tidx

問い

本番 indexer が壊れるのは、異なるクエリ形が同時に来る瞬間だ — 「アドレス X の最新 10 件」（point lookup）と「過去 1 年の日次集計」（range scan）。1 つのストレージエンジンに両方を無理に載せると片方が死ぬ。どう設計するか？（読む対象は公開実装 tidx、Tempo の EVM L1 で実運用。）

原理（最小モデル）

OLTP + OLAP のデュアルストレージ。 PostgreSQL の row store は point lookup に強い、ClickHouse の column store は広い期間集計に強い。互いに相手の質問をすると死ぬ（PG に日次 volume = 全ページ read で数十秒〜数分、CH に最新 10 件 = point index がなくスキャン）。tidx は両方に書き、read で振り分ける。
dual sink: 1 reader、2 write。 SinkSet { pool, ch: Option<...> }、write_all が tokio::try_join! で PG/CH に同時書き込み（待ち時間 ≈ max(pg, ch)）。PG は 4 テーブルを 1 トランザクション、CH は append-only で部分失敗は retry 回復、CH は任意（未設定なら PG-only で OLAP だけ無効）。
lazy event decoding。 event の事前登録を要求しない。logs に生バイト（selector/topics/data）を保持し、デコードはクエリ時に EventSignature::parse が行う（signature 文字列 → topic0 計算 → フィルタ + 射影 CTE 合成）。代償は保存量 5-10 倍、得るものは事前登録なしで新しい質問に即応（Subgraph は manifest で事前デコード、tidx は query 時）。
realtime + gap sync の 2 ループ + 別 RPC client 2 本。 realtime は低遅延重視、backfill は帯域消費型 — 共有すると backfill が realtime を詰まらせる。sync_state の 4 カーソル（head/tip/synced/backfill）で 2 ループの干渉を防ぐ。

具体例

「2 回書く」抽象（src/sync/sink.rs）:

#[derive(Clone)]
pub struct SinkSet {
    pool: Pool,                 // PostgreSQL (常に存在)
    ch: Option<ClickHouseSink>, // ClickHouse (オプショナル)
}

impl SinkSet {
    pub async fn write_all(
        &self,
        blocks: &[BlockRow], txs: &[TxRow],
        logs: &[LogRow], receipts: &[ReceiptRow],
    ) -> Result<()> {
        if let Some(ch) = &self.ch {
            tokio::try_join!(
                writer::write_batch(&self.pool, blocks, txs, logs, receipts),
                ch.write_blocks(blocks),
                ch.write_txs(txs),
                ch.write_logs(logs),
                ch.write_receipts(receipts),
            )?;
        } else {
            writer::write_batch(&self.pool, blocks, txs, logs, receipts).await?;
        }
        Ok(())
    }
}

sync engine はチェーンを 1 回 読んで結果を write_all に渡す（src/sync/engine.rs）:

pub struct SyncEngine {
    throttled_pool: ThrottledPool,
    sinks: SinkSet,             // ← ファンアウトはここ
    realtime_rpc: RpcClient,    // ← tip 追跡用の別 RPC client
    backfill_rpc: RpcClient,    // ← gap 埋め用の別 RPC client
    chain_id: u64,
    ...
}

query routing の契約（src/query/router.rs）— エンジン 2 つ、HTTP API は ?engine= 省略時に router が選ぶ:

pub enum QueryEngine {
    ClickHouse,  // OLAP
    Postgres,    // OLTP
}

SELECT * FROM blocks WHERE num = 12345 → point lookup → PG。SELECT type, COUNT(*) FROM txs GROUP BY type → 集計 → CH。振り分けはヒューリスティック（用途明確なら ?engine= で明示）。CH の materialized view は insert 時に集計更新（read 時の重い再集計を避ける）。

失敗例（誤解）

「PostgreSQL だけ / ClickHouse だけで十分」は誤り — PG only は「過去 1 年の日次 volume」で死に（ディスク律速）、CH only は「最新 10 件」で死ぬ（point index なしでスキャン、答え 10 行のためのムダ IO）。それぞれが相手の得意クエリで律速する。tidx は両方に書き read で振り分ける — エンジン分離が本質、自動振り分けは利便機能にすぎない。

ここまでで「OLTP/OLAP デュアルストレージ + dual sink」は着地した。ここから tidx を読み、自分の indexer に落とす。コードは抜粋（実行時は repo 全体を参照）。

🛑 予測。 dual-sink で write_all はどこまで順序を保証するか。ブロック N が PG にだけあり CH にない状態は起こり得るか。（答え: 起き得るが短時間。try_join! は両成功時に return するため、途中で CH を先に読むと古い値を見うる。tidx はこれを許容し ch_backfill_block で後から埋める。整合性モデルが PG=トランザクション・CH=append-only retry で違うのが前提。）

ステップで組み立てる

Step 1-7: tidx を読む

Dual Sink（sink.rs、上の SinkSet）— try_join! で PG/CH 同時書き込み。
Sync Engine（engine.rs、上の struct）— RPC client 2 本の理由（realtime 低遅延 vs backfill 帯域消費、並行度予算を個別管理）。backfill_first/trust_rpc を読む。
スキーマ（db/*.sql と db/clickhouse/*.sql）— 同じカラム、別のエンジン/索引戦略。txs/logs/receipts は block_timestamp を非正規化（JOIN 削減）。
Lazy event decoding（query/parser.rs の EventSignature::parse）— 生バイト保持、query 時デコード。CLI から signature を渡せば event 名がテーブルとして見える:
```
tidx query \
  --signature "Transfer(address indexed from, address indexed to, uint256 value)" \
  "SELECT * FROM Transfer WHERE from = '\\xAlice...' LIMIT 10"
```
Query routing（router.rs、上の QueryEngine）— point lookup→PG / 集計→CH。

Materialized views（api/views.rs）— insert 時集計。認可は require_admin_mutation（trusted IP かつ x-tidx-admin: 1 ヘッダ = defense in depth）。HTTP で view を作る:

curl -X POST "https://tidx.example.com/views" -d '{
  "chainId": 4217,
  "name": "top_holders",
  "sql": "SELECT token, holder, sum(balance) AS balance
          FROM token_balances GROUP BY token, holder HAVING balance > 0",
  "orderBy": ["token", "holder"]
}'

Sync アーキ — realtime（~0 lag）+ gap sync（最新→最古に埋める、利用頻度の高い新データを先に）。

Step 8: 読みから書きへ — 自分の indexer

そのまま採用: Ethereum JSON-RPC を話すなら rpc_url を変えて tidx up（テーブルはチェーン非依存、chain_id はデータ列）。
Fork: 独自フィールド（fee payer 等）が要るなら 3 箇所を拡張 — ① db/*.sql + db/clickhouse/*.sql（追加カラム）② src/sync/decoder.rs（追加フィールド抽出）③ src/types.rs（*Row struct 拡張）。sync engine / dual sink / query router / views API はそのまま。

答え合わせ（Test gate）

最低ラインは fixture chain replay。reorg ケースは必須（ChainCommitted だけ処理する indexer は reorg で状態を壊す）:

// tests/fixture_replay.rs
use reth_exex::ExExNotification;

#[tokio::test]
async fn replays_committed_then_reverted() {
    let pg = test_pg_pool().await;            // テストごとに ephemeral schema
    let mut indexer = Indexer::new(pg.clone());

    // ブロック N..N+5 を適用
    for n in N..N+5 {
        indexer.handle(committed_fixture(n)).await.unwrap();
    }
    assert_eq!(pg.tx_count_at_block(N+4).await, GOLDEN_TX_COUNT);

    // N+3..N+5 を reorg
    indexer.handle(reverted_fixture(N+3..=N+5)).await.unwrap();
    assert_eq!(pg.tx_count_at_block(N+2).await, GOLDEN_TX_COUNT_AT_N2);
    assert_eq!(pg.tx_count_at_block(N+5).await, 0, "reorg されたブロックは消えるはず");
}

#[tokio::test]
async fn idempotent_under_replay() {
    // 同じ通知を 2 回流しても no-op になる必要がある（クラッシュ復旧シナリオ）
}

fixture は tests/fixtures/ に ExExNotification をシリアライズして置く。2 テスト green まで未完了。

合格基準

上記 2 テスト（reorg 巻き戻し + idempotent replay）が green。
PG が殺すクエリクラスと CH が殺すクエリクラスを 1 つずつ言える。
tidx を fork するとき触る 3 ファイル（schema / decoder / types）を即答できる。

Drill

sink.rs で PG 成功・CH 失敗時に何が起きるかトレース（PG はロールバックする? 次の iteration は?）（30 分）。
router.rs で ?engine= なしの振り分けルールを 1 文で特定（45 分）。
両スキーマ + TxRow + decoder に L2 用 l1_origin カラムを追加して cargo build（2 時間）。
実 analytics 質問を 1 つ選び POST /views ボディを書く（1 時間）。
tidx init → rpc_url を testnet に → tidx up → tidx status --watch で realtime が追いつくのを見る（1 時間）。

📺 関連動画

GhEhzE9SFqY | Alexey Shekhirin — Using Reth Execution Extensions for next generation indexing (Devcon 2024)

まとめ（3行）

本番 indexer は OLTP（PG、point lookup）+ OLAP（ClickHouse、期間集計）のデュアルストレージ — 片方だけだと相手の得意クエリで律速する。tidx は両方に書き read で振り分ける。
dual sink が tokio::try_join! で同時書き込み（一時的に PG/CH がずれるのは許容し後埋め）。lazy event decoding で事前登録不要（保存量 5-10 倍と引き換え）。RPC client を realtime/backfill で分離。
Test gate: fixture chain replay で reorg 巻き戻し + idempotent replay。ChainReverted が実際に巻き戻すことを証明する（reorg 無視は典型的本番障害）。

次のレッスン（レッスン3）

Reth にカスタム RPC エンドポイントを足す（extend_rpc_modules）。ノードを in-process で起動し、新メソッドを HTTP で叩いて JSON レスポンスを assert、エラーパス（不正パラメータ・欠損ブロック）も網羅する。