レッスン5 — ExEx API をステップで組み立てる

問い

ExEx（Execution Extension）= Reth が提供する「実行ループに Rust コードを注入する」仕組み。ノード速度のインデクサ・MEV ボット・リアルタイムリスクエンジンを チェーン本体と同じプロセス内で 構築可能。素朴な「RPC ポーリング」インデクサから組み立てると、API の 4 要素の理由が見えるか？

原理（最小モデル）

素朴な RPC ポーリングが劣る 3 理由. レイテンシ（数秒遅れ）+ アトミック性（Reth コミット → インデクサ間の race）+ Reorg 弱い情報。修正方針 = 同プロセスで動かす。
ExExNotification の 3 バリアント. ChainCommitted { new }（追加）+ ChainReorged { old, new }（置換、アトミックスワップ）+ ChainReverted { old }（削除のみ）。
3 アーム必須が ExEx #1 バグ防止. ChainReorged 欠落 → 派生状態に古いチェーンのデータ phantom 化 + 新チェーンのデータ欠落。
FinishedHeight で「終わった」を Reth に伝える. Pruner が安全に prune できる最下位を集約。忘れると Reth がフルアーカイブに変質。
ストリーム pull（コールバックではない）. ctx.notifications.try_next().await でハンドラ自分のペース、Reth はハンドラ速度に縛られない。
init/run 分割. exex_init（同期セットアップ + future 返す）+ exex（永続 future、ストリーム poll）。Reth が「起動失敗」と「動いた後クラッシュ」を区別可能。
install_exex で複数拡張. 独立通知ストリーム、独立 FinishedHeight、互いに干渉しない。Pruner は 最遅の 1 つ に合わせる。

具体例

素朴な RPC ポーリング:

fn main() {
    let rpc = HttpProvider::new("http://localhost:8545");
    let mut last_block = 0;
    loop {
        let head = rpc.get_block_number().unwrap();
        for n in (last_block+1)..=head {
            let block = rpc.get_block(n).unwrap();
            index(block);
        }
        last_block = head;
        sleep(Duration::from_secs(1));
    }
}

劣る 3 理由:

レイテンシ — RPC ポーリング = req/res オーバーヘッド、tip から数秒遅れ → MEV / リスク / リアルタイム UX に使えない
アトミック性 — Reth が新ブロックをコミット → インデクサが見るまでに期間 = race condition
Reorg — ポーリングで head = N を 2 回見るが中身違うブロック = インデクサが外側で reorg 検出する羽目、しかも Reth より弱い情報

最初の試案（コールバック）:

fn on_new_block<F: Fn(&Block)>(reth: &mut Reth, callback: F) {
    reth.add_listener(callback);
}

足りないもの 2 つ:

Reorg は append-only ではない — 「新ブロック追加」だけのコールバックでは「ブロック N のハッシュ X が Y に置換」を表現不能 → 派生状態が静かに壊れる
Reth に「終わった」を伝える方法なし — N-100,000 のデータを prune してよいか Reth は判断不可

3 バリアントの通知:

enum ExExNotification {
    ChainCommitted { new: Chain },             // canonical ブロック追加
    ChainReorged   { old: Chain, new: Chain }, // old が new に置換
    ChainReverted  { old: Chain },             // 削除（置換なし）
}

各バリアントの扱い:

ChainCommitted { new } — 新ブロック状態をインデックスに追記
ChainReorged { old, new } — old の状態を undo し、new の状態を apply（アトミックスワップ）
ChainReverted { old } — old の状態を undo して待つ、Reth は新 tip を選んだら後続 ChainCommitted を送る

ChainReorged 欠落の失敗モード（ExEx #1 バグ）:

HashMap に 古いチェーン の tx + canonical チェーンは 新しいチェーン
インデックスを後で読むと canonical 上に存在しない tx が返る = phantom-data バグ
さらに悪い: 新しいチェーンの tx はインデックスされていない（ChainCommitted 未受信、無視した ChainReorged 受信）

FinishedHeight:

ctx.events.send(ExExEvent::FinishedHeight(block_number_hash))?;

ctx.events = Reth への書き込み専用チャンネル。ブロック終了ごと送信、Reth が全 ExEx の最小値を集約してその下を prune。

忘れたディスク帰結: Reth が prune すべき履歴を全保持 → 半年後にディスク使用量複利膨張 → 「無害なインデクサ」がノードをフルアーカイブに変質。

ストリーム pull:

while let Some(notification) = ctx.notifications.try_next().await? {
    // 自分のペースで処理
}

非同期 + ハンドラのペース + Reth がハンドラ速度に縛られない + イベント順序保証。

init/run 分割:

async fn exex_init<Node: FullNodeComponents>(
    ctx: ExExContext<Node>,
) -> eyre::Result<impl Future<Output = eyre::Result<()>>> {
    // 同期的セットアップはここ
    Ok(exex(ctx))  // 長時間動く future を返す
}

2 関数:

exex_init — ノード起動時に 1 度だけ、同期セットアップ、future を返す
exex — 永続 future、通知ストリームを poll

分割理由: ファイル open 等を exex に入れると Reth が通知バッファ後に init 失敗 → ExEx 健全と誤認しつつ通知積上り。分割で「起動失敗」と「動いた後クラッシュ」を区別可能。

複数拡張:

.install_exex("MyIndexer", exex_init)
.install_exex("MevWatcher", mev_init)
.install_exex("RiskEngine", risk_init)

実用的含意:

互いに干渉しない — 各ストリーム独立バッファ
Pruner は最遅の 1 つに合わせる — 全 ExEx の FinishedHeight の 最小値 より下しか prune しない
クラッシュは独立 — 1 つ panic でも他 ExEx + Reth は動き続ける
メトリクスは ExEx ごと — 第 1 引数の名前が reth_exex_<name>_* メトリクスラベル

失敗例（誤解）

「ChainCommitted だけ扱えば十分」— 間違い（ExEx #1 バグ）。Reorg 5 ブロック深く → HashMap に phantom data + 新チェーン未インデックス。3 アーム全部必要。

「FinishedHeight 送信は任意」— 間違い。忘れると Pruner が止まる → ディスク膨張複利 → 半年後にフルアーカイブ。毎 commit ごとに必須。

「同プロセスならパフォーマンスは同じ」— 間違い。RPC ポーリング = req/res オーバーヘッド + tip から数秒遅れ + race。ExEx = ゼロレイテンシ + アトミック + 構造化 reorg 情報。MEV / リアルタイムに必要。

ステップで組み立てる

Step 1: RPC ポーリングの 3 失敗

レイテンシ + アトミック性 + Reorg 弱情報 → 同プロセス（ExEx）で全解決。

Step 2: 3 バリアント通知

ChainCommitted（追加）+ ChainReorged（置換）+ ChainReverted（削除のみ）。全 3 アーム必須。

Step 3: `FinishedHeight` の役割

Reth に「終わった」を伝え、Pruner が安全 prune。忘れるとディスク膨張。

Step 4: ストリーム pull の利点

ハンドラのペース、Reth がブロックされない、順序保証。

Step 5: init/run 分割の理由

「起動失敗」と「動いた後クラッシュ」を区別可能、ファイル open は exex_init で。

Step 6: `install_exex` で複数拡張

独立通知 + Pruner は最遅に合わせる + クラッシュ独立 + メトリクス別。

答え合わせ

3 アーム必須が ExEx #1 バグ防止である理由: ChainReorged 欠落 → HashMap に古いチェーン phantom + 新チェーン未インデックス。任意の通知シーケンスの後、各 Address のカウントが正しい不変条件 を保つには、reorg で old 全 undo → new 全 apply を 1 通知でアトミックに必要。
FinishedHeight を忘れたディスク帰結: Reth が prune すべき履歴を「ExEx が後で読みたいかも」で全保持 → 半年でディスク複利膨張 → ノードがフルアーカイブに変質。「無害なインデクサ」の典型本番事故。
クラッシュした ExEx の Pruner への影響: 落ちた時点の FinishedHeight で Pruner 固まる → 以降ブロック全部「再起動したら読みたい」扱い → 全履歴蓄積。再起動して FinishedHeight 進めるまで蓄積継続。

合格基準

3 バリアント（ChainCommitted / ChainReorged / ChainReverted）を即答できる。
3 アーム必須の理由を ExEx #1 バグで説明できる。
FinishedHeight の役割と忘却時のディスク帰結を言える。
init/run 分割の理由を「起動失敗 vs 後でクラッシュ」で説明できる。
複数 ExEx の Pruner ルール（最遅に合わせる）を即答できる。

まとめ（3行）

ExEx = 同プロセス Rust 注入、RPC ポーリングの 3 失敗（レイテンシ + アトミック性 + Reorg 弱情報）を全解決。
3 バリアント通知（ChainCommitted / ChainReorged / ChainReverted）の 全アーム必須（ExEx #1 バグ防止）+ FinishedHeight で Pruner 制御（忘却 → ディスク複利膨張）。
init/run 分割で起動失敗を検出 + install_exex で複数拡張独立、Pruner は 最遅の 1 つに合わせる。