レッスン1 — `Stage` トレイトをステップで組み立てる

問い

Staged Sync は Reth の背骨。本物の Stage トレイトは 6 メソッド + 非同期準備チェック + 双方向対称性 + auto_impl(Box) 属性 = 一度に概念が 6 つ降ってくる。素朴な同期ループから組み立てると、各要素の理由が見えるか？

原理（最小モデル）

素朴な「1 ブロックずつ」が遅い 3 理由. バッチなし + I/O 償却なし + 並列化なし。修正方針 = ステージに分割、ブロック範囲を端から端まで処理。
execute / unwind 対称性が要石. Reorg を特殊ケースではなく通常運用に。同じトレイト、2 方向。
ExecInput / ExecOutput で明示的再開可能. target + checkpoint で「どこで止め、どこから再開」を表現。done: bool が戻り値内にあるのは アトミック呼び出し/戻り値 のため。
poll_execute_ready で非同期準備. Rust async 形式の poll、ネットワーク I/O 待ちのステージがオーバーライド。pending を返してもパイプライン全体は止まらない。
post_*_commit で opt-in ライフサイクル. ExecutionStage が ExEx 通知を流す、Pruner が古いインデックス開放、など。
#[auto_impl(Box)] でヘテロリスト. Vec<Box<dyn Stage>> に格納可能、Box<S> への転送 impl を自動生成。

具体例

最終的に組み立てる本物のトレイト:

#[auto_impl::auto_impl(Box)]
pub trait Stage<Provider>: Send {
    fn id(&self) -> StageId;
    fn poll_execute_ready(&mut self, _cx: &mut Context<'_>, _input: ExecInput)
        -> Poll<Result<(), StageError>> { Poll::Ready(Ok(())) }
    fn execute(&mut self, provider: &Provider, input: ExecInput)
        -> Result<ExecOutput, StageError>;
    fn post_execute_commit(&mut self) -> Result<(), StageError> { Ok(()) }
    fn unwind(&mut self, provider: &Provider, input: UnwindInput)
        -> Result<UnwindOutput, StageError>;
    fn post_unwind_commit(&mut self) -> Result<(), StageError> { Ok(()) }
}

素朴な 1 ブロックずつ:

fn sync_to_tip(client: &mut RethNode) -> Result<(), Error> {
    while let Some(block) = client.next_block()? {
        let header = client.fetch_header(block)?;
        let body   = client.fetch_body(block)?;
        let senders = recover_senders(&body)?;
        let receipts = client.execute(&block, &header, &body)?;
        client.update_state(receipts)?;
        client.update_merkle_root(&block)?;
        client.write_indexes(&block)?;
        client.commit()?;
    }
    Ok(())
}

ステージのスケッチ:

let stages = vec![
    HeaderStage,       // [N..M] のヘッダーをダウンロード
    BodyStage,         // tx 本体をダウンロード
    SenderRecovery,    // ECDSA sender 復元（並列）
    Execution,         // Revm を走らせ、状態差分を蓄積
    Hashing,           // ハッシュ化されたアカウント/ストレージ変更をソート
    Merkle,            // 範囲の Merkle ルートを計算
    Indexes,           // txhash → (block, index) などのインデックス
    Finish,            // commit + 報告
];

for stage in &mut stages {
    stage.run(blocks_n_to_m)?;
}

Stage の最初の試案（メソッド 1 つだけ）:

trait Stage {
    fn execute(&mut self, blocks: BlockRange) -> Result<(), StageError>;
}

unwind 追加（reorg は通常運用）:

trait Stage {
    fn execute(&mut self, blocks: BlockRange) -> Result<(), StageError>;
    fn unwind(&mut self, blocks: BlockRange) -> Result<(), StageError>;
}

入出力 struct:

pub struct ExecInput {
    pub target: Option<BlockNumber>,
    pub checkpoint: Option<StageCheckpoint>,
}
pub struct ExecOutput {
    pub checkpoint: StageCheckpoint,
    pub done: bool,
}
pub struct UnwindInput {
    pub checkpoint: StageCheckpoint,
    pub unwind_to: BlockNumber,
    pub bad_block: Option<BlockNumber>,
}

非同期準備:

fn poll_execute_ready(&mut self, _cx: &mut Context<'_>, _input: ExecInput)
    -> Poll<Result<(), StageError>>
{
    Poll::Ready(Ok(()))  // デフォルト: 常に準備完了
}

Poll<T> の意味: Future の内部関数が Poll::Ready(value)（完了）か Poll::Pending（まだ準備中）を返す。Pending → ランタイムが脇に置き別ステージを poll → 準備完了で起こされ再 poll。スレッドをブロックせずに「待つ」を表現。

コミットフック（opt-in）:

fn post_execute_commit(&mut self) -> Result<(), StageError> { Ok(()) }
fn post_unwind_commit(&mut self) -> Result<(), StageError> { Ok(()) }

具体例: ExecutionStage が ExEx 通知を流す（subscriber が commit 済みデータを読みに来るので tx body のコミットが先に完了している保証が必要）/ Pruner 系が checkpoint 書き込み後にディスクから古いインデックスを開放。

#[auto_impl(Box)] の転送:

impl<S: Stage<P>> Stage<P> for Box<S> {
    // 全6メソッドを (**self).method(...) で転送
}

手書きせず属性で自動生成、Vec<Box<dyn Stage<...>>> に格納可能に。

失敗例（誤解）

「reorg は特殊コードパスで扱う」— 間違い。他クライアントの形 = コードベース半分が「reorg パス」化、Reth は 同じトレイトに unwind で通常運用に。

「done を has_more() メソッドで返す」— 間違い。オーケストレータが 1 ターンで 2 回呼ぶ = checkpoint と has_more が食い違うバグ余地。戻り値内のフラグでアトミック。

「同期セットアップを exex future 内に置く」— 間違い。Reth が通知バッファ後に init が失敗 → ExEx 健全と誤認しつつ通知積上り。exex_init と exex 分割で「起動できなかった」と「動いた後でクラッシュ」を区別。

ステップで組み立てる

Step 1: 素朴な「1 ブロックずつ」の 3 失敗

バッチなし（200 回 ECDSA セットアップ）+ I/O 償却なし（2000 万 commit）+ 並列化なし。ステージ分割で全部解決。

Step 2: `execute` / `unwind` 対称性

同じトレイト、2 方向。前進 = execute、後退 = unwind。Reorg は通常運用の一部に。

Step 3: `ExecInput` / `ExecOutput` で再開可能

target（どこで止め）+ checkpoint（どこから再開）+ done フラグ（戻り値内、アトミック）。

Step 4: `poll_execute_ready` で非同期準備

デフォルト Ready、ネットワーク I/O 待ちステージのみオーバーライド。

Step 5: `post_*_commit` で opt-in ライフサイクル

デフォルト no-op、必要なステージのみ（ExecutionStage が ExEx 通知、Pruner がディスク開放）。

Step 6: `#[auto_impl(Box)]` で転送自動化

Vec<Box<dyn Stage>> に格納可能、属性で 6 メソッドの転送 impl を自動生成。

答え合わせ

unwind を同トレイトにする利点: 前進と reorg が 同じ表面 を使う = コードベースが「通常パス + reorg パス」に二分されない。Reth のアーキ要石。
done が戻り値内のフラグである理由: アトミック呼び出し/戻り値。オーケストレータは 1 ターンで「checkpoint X、また呼ぶ／呼ばない」の 1 フィードバックを得る。別メソッド = 2 回呼びで食い違いバグ。
poll_execute_ready の存在理由: ネットワーク I/O 待ち（HeaderStage など）が他ステージをブロックしないように pending 返す。常時 Ready のステージはデフォルト使用。

合格基準

6 メソッドの Stage トレイトを役割で言える。
素朴 1 ブロックずつの 3 失敗（バッチ / I/O / 並列）を即答できる。
execute / unwind 対称性が要石である理由を 1 文で説明できる。
ExecInput / ExecOutput / UnwindInput の主要フィールドを言える。
#[auto_impl(Box)] が省く手書きを言える。

まとめ（3行）

Stage トレイト 6 メソッド = 素朴「1 ブロックずつ」の 3 失敗（バッチ / I/O / 並列）を解決する 6 設計判断の積み重ね。
execute / unwind 対称性 + ExecInput / ExecOutput の明示的再開 + poll_execute_ready の非同期準備 + post_*_commit の opt-in + #[auto_impl(Box)] のヘテロリスト。
次のレッスンで Reth の本物の 10 ステージパイプラインを巡る、各ステージが何をするか + なぜこの順序か。

レッスン1 — Stage トレイトをステップで組み立てる

問い