レッスン4 — Reth 上で最小 sequencer を作る

問い

動く L2 sequencer は Rust 約 270 行 で書ける。Reth が難しい部分（revm 実行 + MDBX ストレージ + state 管理 + P2P）をすべて引き受けてくれる から。Sequencer の仕事は Engine API 経由で Reth を駆動し L1 に投稿することだけ。その 270 行は何か？

注: 以下のコードは教材向けの概念実装（最小骨格）です。実運用にはエラーハンドリング、型定義、再試行、監視、完全な P2P 実装が追加で必要です。

原理（最小モデル）

• 1 プロセスに 4 コンポーネント. Sequencer loop（Engine API 駆動）+ Mempool（user tx 受け入れ + fee 優先）+ L1 inbox watcher（deposit subscribe + force-include）+ Batcher（定期 L1 投稿）。
• Sequencer loop 9 ステップ. Head 取得 → Payload attrs 計算 → forkchoiceUpdated → 500ms 待つ → getPayload → 署名 → newPayload → forkchoiceUpdated（finalize）→ broadcast。
• Mempool は優先キュー + 検証 + eviction. Gas tip ソート + timeout / fullness eviction + reorg 時 tx 戻す + sanity 検証。
• L1 inbox watcher は event subscribe. Deposit イベント → L2 deposit tx エンコード → mempool で force-include 優先。
• Batcher は定期投稿. ~60s ごと → fetch_blocks + zlib + blob チャンク + blob tx 提出。
• 本番落とし穴 6 つ. Liveness alarm / L1 reorg / L2 reorg / Pre-confirmation / Mempool DOS / DB 成長。

具体例

アーキテクチャ:

flowchart TB
    Users["L2 Users (HTTP RPC)"] -->|tx| Mempool["Mempool"]
    L1Sub["L1 Inbox サブスクリプション"] -->|deposit イベント| Mempool
    Mempool -->|pending txs| Loop["Sequencer Loop<br/>(2s ごとにブロック生成)"]
    Loop -->|forkchoiceUpdated + getPayload| Reth["Reth EL"]
    Loop -->|ブロック署名| Signer
    Loop -->|new payload| Reth
    Loop -->|broadcast| P2P["P2P Network"]
    Loop -->|60s ごと| Batcher["Batcher"]
    Batcher -->|blob tx| L1["L1 (Ethereum)"]

Sequencer loop（コア生成ループ）:

use alloy_provider::{Provider, ProviderBuilder};
use alloy_signer_local::PrivateKeySigner;
use reth_rpc_engine_api::EngineApiClient;
use std::time::Duration;
use tokio::time::interval;

pub struct MinimalSequencer {
    signer: PrivateKeySigner,
    engine: EngineApiClient,
    mempool: Arc<Mempool>,
    chain_id: u64,
    block_period: Duration,
}

impl MinimalSequencer {
    pub async fn run(self) -> eyre::Result<()> {
        let mut ticker = interval(self.block_period);
        loop {
            ticker.tick().await;
            if let Err(e) = self.produce_block().await {
                tracing::error!(?e, "block production failed");
            }
        }
    }

    async fn produce_block(&self) -> eyre::Result<()> {
        // 1. Reth から現 head 取得
        let parent_hash = self.current_head().await?;

        // 2. Payload 属性計算
        let attrs = PayloadAttributes {
            timestamp: now_seconds(),
            prev_randao: B256::random(),
            suggested_fee_recipient: self.signer.address(),
            withdrawals: vec![],
            parent_beacon_block_root: None,
        };

        // 3. Reth に構築開始を伝える
        let forkchoice = ForkchoiceState {
            head_block_hash: parent_hash,
            safe_block_hash: parent_hash,
            finalized_block_hash: parent_hash,
        };
        let resp = self.engine
            .fork_choice_updated_v4(forkchoice, Some(attrs))
            .await?;
        let payload_id = resp.payload_id.ok_or_else(|| eyre!("no payload id"))?;

        // 4. Reth が構築する時間を少し待つ
        tokio::time::sleep(Duration::from_millis(500)).await;

        // 5. 構築済 payload 取得
        let payload = self.engine
            .get_payload_v4(payload_id)
            .await?;

        // 6. Payload hash 署名
        let signature = self.signer
            .sign_hash(&payload.execution_payload.block_hash())
            .await?;

        // 7. 署名済 payload を Reth に提出
        self.engine
            .new_payload_v4(payload.execution_payload.clone())
            .await?;

        // 8. Forkchoice 更新 (新 head を最終としてマーク)
        let new_head = payload.execution_payload.block_hash();
        self.engine
            .fork_choice_updated_v4(
                ForkchoiceState {
                    head_block_hash: new_head,
                    safe_block_hash: new_head,
                    finalized_block_hash: new_head,
                },
                None,
            )
            .await?;

        // 9. Peer に broadcast (P2P、省略)

        tracing::info!(
            block = new_head.to_string(),
            "produced block"
        );
        Ok(())
    }
}

Mempool:

use alloy_consensus::TxEnvelope;
use std::sync::{Arc, RwLock};

pub struct Mempool {
    pending: Arc<RwLock<Vec<TxEnvelope>>>,
}

impl Mempool {
    pub fn submit(&self, tx: TxEnvelope) -> eyre::Result<TxHash> {
        validate_tx(&tx)?;
        let hash = tx.hash();
        self.pending.write().unwrap().push(tx);
        Ok(hash)
    }

    pub fn submit_deposit(&self, tx: TxEnvelope) -> eyre::Result<TxHash> {
        // 実装上は通常 tx と同じ受け口に流し込む。
        self.submit(tx)
    }

    pub fn drain_pending(&self, limit: usize) -> Vec<TxEnvelope> {
        let mut pending = self.pending.write().unwrap();
        let len = pending.len().min(limit);
        pending.drain(..len).collect()
    }
}

L1 inbox watcher:

pub struct L1InboxWatcher {
    l1_provider: Box<dyn Provider>,
    inbox_address: Address,
    mempool: Arc<Mempool>,
}

impl L1InboxWatcher {
    pub async fn run(self) -> eyre::Result<()> {
        let mut stream = self.l1_provider
            .subscribe_logs(&Filter::new()
                .address(self.inbox_address)
                .event("DepositInitiated(...)"))
            .await?;

        while let Some(log) = stream.next().await {
            let deposit_tx = self.encode_l2_deposit_tx(&log)?;
            self.mempool.submit_deposit(deposit_tx)?;
        }
        Ok(())
    }
}

Batcher:

pub struct Batcher {
    l1_provider: Box<dyn Provider>,
    l2_provider: Box<dyn Provider>,
    batcher_signer: PrivateKeySigner,
    last_batch_block: AtomicU64,
}

impl Batcher {
    pub async fn run(self) -> eyre::Result<()> {
        let mut ticker = interval(Duration::from_secs(60));
        loop {
            ticker.tick().await;
            if let Err(e) = self.post_batch().await {
                tracing::error!(?e, "batch posting failed");
            }
        }
    }

    async fn post_batch(&self) -> eyre::Result<()> {
        let from = self.last_batch_block.load(Ordering::SeqCst);
        let to = self.l2_provider.get_block_number().await?;

        // 1. `from` から `to` の全 L2 ブロック取得
        let blocks = self.fetch_blocks(from, to).await?;

        // 2. 圧縮
        let data = zlib_compress(rlp_encode(&blocks))?;

        // 3. Blob サイズチャンクに分割
        let chunks = chunk(data, MAX_BLOB_SIZE);

        // 4. 各々を blob tx として提出
        for chunk in chunks {
            let blob_tx = build_blob_tx(chunk, self.batcher_signer.address());
            self.l1_provider.send_raw_transaction(blob_tx).await?;
        }

        // 5. Last batch 更新
        self.last_batch_block.store(to, Ordering::SeqCst);
        Ok(())
    }
}

ファイル構成:

~/my-sequencer/
├── src/
│   ├── main.rs           ← 20 行 (全部配線)
│   ├── sequencer.rs      ← 80 行
│   ├── mempool.rs        ← 50 行
│   ├── l1_watcher.rs     ← 40 行
│   ├── batcher.rs        ← 50 行
│   └── rpc.rs            ← 30 行 (ユーザ提出用 HTTP server)
├── Cargo.toml
└── README.md

本番落とし穴:

失敗例（誤解）

「Reth なしで sequencer を書くと数行で済む」— 間違い。Reth が引き受けている revm 実行 / MDBX / state 管理 / P2P を自前で書くと数万行。270 行は Reth に乗る ことで成立する数字。

「mempool は単純な Vec で十分」— MVP では Yes、本番では No。本番は ① 優先キュー（gas tip ソート）+ ② timeout / fullness eviction + ③ reorg 時 tx 戻す + ④ DOS 防御。MVP から始めて足していく。

「batcher は失敗しない」— 間違い。L1 ガス急騰 / nonce conflict / blob 容量超過 / L1 RPC エラー、すべて起こる。再試行戦略（exponential backoff + ガス価格 bump）が必要。

🛑 予測。 Sequencer は ~2s ごとに L2 ブロック構築。本番で最初に起きやすい失敗モードは？（ヒント: コンセンサスでも暗号でもない）（答え: L1 reorg 連鎖。L1 が短期 reorg → batcher が投稿した batch tx が orphan → 再 batch 必要 + 同時に L1 inbox watcher が観察した deposit イベントが orphan → mempool に偽 deposit が残る → state 不整合。または mempool DOS = 攻撃者が無効 tx を spam → mempool 膨らみ + drain が遅くなり ブロック生成が遅れる。両者とも consensus / crypto と無関係、操作上の堅牢性問題。）

ステップで組み立てる

Step 1: 4 コンポーネントを即答

Sequencer loop / Mempool / L1 inbox watcher / Batcher。

Step 2: Sequencer loop の 9 ステップ

Head 取得 → Attrs 計算 → forkchoiceUpdated → 500ms 待つ → getPayload → 署名 → newPayload → forkchoiceUpdated（finalize）→ broadcast。

Step 3: 各コンポーネントの行数感覚

Sequencer 80 / Mempool 50 / L1 watcher 40 / Batcher 50 / main 20 / rpc 30 = ~270 行。

Step 4: MVP → 本番のステップ

落とし穴 6 つ（Liveness alarm / L1 reorg / L2 reorg / Pre-conf / DOS / DB 成長）を 1 つずつ実装。

Step 5: なぜ 270 行で済むか

Reth が引き受けるもの: revm 実行 / MDBX / state 管理 / P2P / staged sync / ExEx。substrate に乗る = 自分のコードは差分のみ。

Step 6: Tempo Moderato に当てはめる

Paradigm 製 sequencer は ~300-500 行 + merchant 認可 + 規制 monitoring + payment 優先順序付け。新規性はアプリケーションロジック、コンセンサスメカニクスではない。

答え合わせ

• 270 行で sequencer が成立する構造的理由: Reth が引き受けている重い部分（revm + MDBX + state 管理 + P2P + staged sync）= 数万行 → 自分はオーケストレーション層（Engine API 駆動 + mempool + L1 watcher + batcher）= ~270 行。substrate に乗る差分のみ書く のが extension model の本質。
• 最初に起きやすい失敗モード: ① L1 reorg 連鎖（batcher の tx orphan + deposit イベント orphan → mempool 不整合）、② mempool DOS（spam → ブロック生成遅延）、③ DB 成長（pruning なしで disk full）、④ Pre-conf 違反（reorg で約束破る → 評判 / slash）。コンセンサス / 暗号でなく操作上の堅牢性問題。
• Tempo Moderato の構造予測: ~300-500 行 Reth ベース + ① merchant 認可フィルタ（builder レベル）+ ② 不正検知用緊急停止権限 + ③ payment-tx 優先 + ④ 規制 monitoring。アプリケーションロジック新規性 + コンセンサスメカニクスは OP Stack 系。

合格基準

• 4 コンポーネントを即答できる。
• Sequencer loop の 9 ステップを順に言える。
• 各コンポーネント行数感覚（270 行合計）を言える。
• 本番落とし穴 6 つを即答できる。
• 「Reth が substrate」の構造的意味を 1 文で説明できる。

まとめ（3行）

• 最小 sequencer = Rust 約 270 行（Sequencer loop 80 + Mempool 50 + L1 watcher 40 + Batcher 50 + main + rpc）、Reth が重い部分を引き受ける。
• Sequencer loop 9 ステップ（forkchoiceUpdated + 500ms 待つ + getPayload + 署名 + newPayload + finalize + broadcast）= Engine API 駆動 + 4 コンポーネント協調。
• MVP → 本番は 6 落とし穴（Liveness / L1 reorg / L2 reorg / Pre-conf / DOS / DB 成長）を 1 つずつ実装、Tempo Moderato も同パターン + アプリケーションロジック差分。