レッスン3 — Reth にカスタム RPC エンドポイントを足す

問い

fee-bidding bot が pending tx の gas price ヒストグラムを 1 回の API 呼び出しで欲しい。標準の txpool_content は pending tx を 全部フルで 返す — 結局 10 個の数字にまとめるのに数百 KB を転送する。どうやって ノード内で 集計してヒストグラムだけ返すか？（読み取り専用メソッド 1 つ、Rust ~50 行、Reth fork なし。）

原理（最小モデル）

カスタム RPC = ノード内集計、Reth fork なし。 jsonrpsee の #[rpc(server, namespace=...)] が trait から server/client stub + JSON シリアライズを生成。impl は pool ハンドルを持つ struct に書き、extend_rpc_modules で既存サーバ（HTTP/WS/IPC）へ登録 — ネイティブ namespace と同じトランスポート・認証・ロギング。
server-side 集計のスイートスポット。 client 側集計（RPC ラウンドトリップ + 全 tx 転送、数百 KB）vs 外部 indexer（µs だが glue + 運用に数日）vs カスタム RPC（µs の in-process スナップショット + 集計後ペイロード + 1 回 ~50 行）。しかもノードの一部として出荷される（別サービス・別デプロイ・ポート開放なし）。
pool.pending() はスナップショット iterator で安価。 具体プール型を固定せず Pool: TransactionPool で書く。集計は O(buckets × pending)。
subscription は tokio::spawn で。 RPC ハンドラは即 return し、ストリーミングはバックグラウンドタスク（ここでブロックすると RPC サーバスレッドが止まる）。sink.send(...).is_err() で切断を検知してクリーンに return（subscription リークなし）。

データ経路を 1 枚で:

flowchart LR
    Client["RPC client<br/>(cast / dapp / dashboard)"] -->|JSON-RPC POST| Handler["jsonrpsee handler"]
    Handler -->|read snapshot| Pool["TransactionPool<br/>(in-process)"]
    Pool -->|all_transactions| Bucket["Bucket math<br/>(10 ranges)"]
    Bucket -->|JSON| Client

具体例

RPC trait（#[rpc] マクロが配管を生成）:

use jsonrpsee::{core::RpcResult, proc_macros::rpc};
use serde::Serialize;

#[derive(Debug, Clone, Serialize)]
pub struct GasBucket {
    pub min_gwei: u64,
    pub max_gwei: u64,  // 0 = 上限なし (最上位バケット用)
    pub count: usize,
}

#[rpc(server, namespace = "txpoolPlus")]
pub trait TxpoolPlusApi {
    #[method(name = "pendingByGasBucket")]
    fn pending_by_gas_bucket(&self) -> RpcResult<Vec<GasBucket>>;
}

pool アクセス付きの実装（wire 名は txpoolPlus_pendingByGasBucket）:

use reth_ethereum::pool::{PoolTransaction, TransactionPool};

pub struct TxpoolPlus<Pool> {
    pool: Pool,
}

const BUCKETS: &[(u64, u64)] = &[
    (0, 1), (1, 5), (5, 10), (10, 20), (20, 30),
    (30, 50), (50, 100), (100, 250), (250, 500), (500, 0),
];

impl<Pool> TxpoolPlusApiServer for TxpoolPlus<Pool>
where
    Pool: TransactionPool + Clone + 'static,
{
    fn pending_by_gas_bucket(&self) -> RpcResult<Vec<GasBucket>> {
        let mut counts = vec![0usize; BUCKETS.len()];

        // pending() はスナップショット iterator を返す; 呼び出しは安価
        for tx in self.pool.pending() {
            let max_priority_fee_wei = tx.max_priority_fee_per_gas().unwrap_or(0);
            let gwei = (max_priority_fee_wei / 1_000_000_000) as u64;

            for (i, &(min, max)) in BUCKETS.iter().enumerate() {
                let upper_match = max == 0 || gwei < max;
                if gwei >= min && upper_match {
                    counts[i] += 1;
                    break;
                }
            }
        }

        Ok(BUCKETS
            .iter()
            .zip(counts)
            .map(|(&(min, max), count)| GasBucket { min_gwei: min, max_gwei: max, count })
            .collect())
    }
}

NodeBuilder への統合点 extend_rpc_modules（起動時 1 回、pool 等のコンテキストを受け取る）:

fn main() {
    Cli::<EthereumChainSpecParser, Args>::parse()
        .run(async move |builder, args| {
            let handle = builder
                .node(EthereumNode::default())
                .extend_rpc_modules(move |ctx| {
                    if !args.enable_txpool_plus {
                        return Ok(());
                    }
                    let ext = TxpoolPlus { pool: ctx.pool().clone() };
                    ctx.modules.merge_configured(ext.into_rpc())?;
                    println!("txpoolPlus_pendingByGasBucket enabled");
                    Ok(())
                })
                .launch_with_debug_capabilities()
                .await?;
            handle.wait_for_node_exit().await
        })
        .unwrap();
}

メソッドはどの RPC クライアントから見てもネイティブと区別がつかない（cast rpc txpoolPlus_pendingByGasBucket や生 curl で叩ける）。これが extend_rpc_modules の約束。

失敗例（誤解）

「txpool_content を呼んで client 側で集計すればいい」は誤り — RPC ラウンドトリップ + 全 tx 転送（数百 KB）を、結局 10 個の数字のために払う。カスタム RPC は in-process スナップショットで µs + 集計後の bytes。もう 1 つの罠: 引数バリデーションで標準 JSON-RPC エラーコードを使い回す — -32603（internal error）は予約済み、不正パラメータは -32602（Invalid params）を返す。

ここまでで「カスタム RPC = ノード内集計を既存サーバに登録」は着地した。ここから trait → impl → 統合 → test の順で組み立てる。コードは抜粋（実行時は補助コードが必要）。

🛑 予測。 なぜ pool.pending() は軽く、txpool_content RPC は重いのか。（答え: pool.pending() は in-process のスナップショット iterator を読むだけ — wire 変換なし。txpool_content は全 pending tx をフルで JSON 化して wire 転送する。差は「返却データ量」と「wire 変換コスト」。）

ステップで組み立てる

Step 1-3: trait → impl → 統合

上の 3 ブロック。#[rpc(server, namespace="txpoolPlus")] が TxpoolPlusApiServer を生成 → pool ハンドルを持つ struct に impl → extend_rpc_modules で merge_configured(ext.into_rpc())。reth/examples/node-custom-rpc と並べると差分は namespace / メソッド名 / ハンドラ内部だけ。

Step 4: cast でテスト

$ cargo run --release -- node --http --enable-txpool-plus
$ cast rpc txpoolPlus_pendingByGasBucket --rpc-url http://localhost:8545
[{"min_gwei":0,"max_gwei":1,"count":12}, ...]

Step 5 (おまけ): subscription バリアント

#[subscription(...)] 属性 + tokio::spawn のバックグラウンドタスク:

fn subscribe_buckets(
    &self,
    pending: PendingSubscriptionSink,
    interval_secs: Option<u64>,
) -> SubscriptionResult {
    let pool = self.pool.clone();
    let interval = Duration::from_secs(interval_secs.unwrap_or(5));

    tokio::spawn(async move {
        let Ok(sink) = pending.accept().await else { return };
        loop {
            sleep(interval).await;
            let buckets = compute_buckets(&pool); // Step 2 から factor out
            let Ok(raw) = serde_json::value::to_raw_value(&buckets) else { continue };
            if sink.send(SubscriptionMessage::from(raw)).await.is_err() { return; }
        }
    });
    Ok(())
}

production gap: 認証（engine と同じ AUTH_SECRET、メソッドが尊重するか確認）/ レート制限（Reth はメソッド単位を同梱しない、tower で）/ バージョニング（形が変われば txpoolPlus_v2_*）/ メトリクス（ネイティブのみ自動、自分で metrics::counter!）/ 引数バリデーション（ErrorObjectOwned::owned で安定コード）。

答え合わせ（Test gate）

最低 gate は インプロセス integration テスト（success + エラーコード + subscription リーク）:

// tests/rpc_integration.rs
use jsonrpsee::{core::client::ClientT, http_client::HttpClientBuilder};
use jsonrpsee::rpc_params;

#[tokio::test]
async fn returns_buckets_for_known_state() {
    let node = boot_node_with_extension().await;       // Reth をインプロセス起動
    let client = HttpClientBuilder::default().build(node.http_url()).unwrap();

    seed_mempool(&node, &TEST_TX_FIXTURES).await;
    let buckets: Vec<Bucket> = client
        .request("txpoolPlus_pendingByGasBucket", rpc_params![10u32])
        .await.unwrap();

    assert_eq!(buckets.len(), 10);
    assert_eq!(buckets.iter().map(|b| b.count).sum::<u64>(), TEST_TX_FIXTURES.len() as u64);
}

#[tokio::test]
async fn rejects_invalid_bucket_count() {
    let node = boot_node_with_extension().await;
    let client = HttpClientBuilder::default().build(node.http_url()).unwrap();

    let err = client.request::<Vec<Bucket>, _>("txpoolPlus_pendingByGasBucket", rpc_params![0u32])
        .await.unwrap_err();
    assert_jsonrpc_error_code(&err, -32602);            // Invalid params (-32603 = internal ではない)
}

#[tokio::test]
async fn subscription_does_not_leak_on_disconnect() {
    // subscription を開く → クライアントを drop → spawn したタスクが終了することを assert
    // (クロージャ内に Drop probe を仕込んで検証)
}

完了条件: (1) success path pass、(2) 少なくとも 1 つのエラーコードテスト pass、(3) subscription cleanup テスト pass。mainnet に対する cargo run はこのどれの代替にもならない。

合格基準

上記 3 テスト（success / 不正パラメータの -32602 / subscription cleanup）が green。
なぜ server-side 集計が client 側集計より有利か（ペイロードサイズ）を 1 文で言える。
不正パラメータに使うべきエラーコード（-32602、-32603 でない）を即答できる。

Drill

pendingByNonce(address) を 2 つ目の #[method] として追加（15 分）。
priority-fee バケット化を effective-gas-price（base_fee + priority_fee）に置換 — base fee を provider から（ctx は何を公開している?）（30 分）。
engine AUTH_SECRET を提示しない呼び出しを拒否（Reth の debug メソッドを参考に）（45 分）。
レスポンスにスナップショットの timestamp + ctx.provider().best_block_number() を追加（30 分）。
レッスン1 の MEV searcher が txpoolPlus_pendingByGasBucket をクエリして 90 パーセンタイル超えで入札するクライアント（jsonrpsee::http_client）を書く（2 時間）。

まとめ（3行）

カスタム RPC = jsonrpsee の #[rpc] trait + pool ハンドルを持つ impl + extend_rpc_modules 登録。ノード内で集計して bytes だけ返す（client 側集計の数百 KB 転送を避ける）、Reth fork なし。
ネイティブと同じトランスポート・認証・ロギングで出荷される（別サービスなし）。subscription は tokio::spawn で（ハンドラをブロックしない、切断時クリーンに return）。
Test gate: in-process integration で success + エラーコード（-32602）+ subscription リーク検出。次は並行性 + 状態管理層の wallet backend。

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

Wallet Backend を Rust で作る（signer pool + nonce manager + replace-on-stuck）。roundtrip テスト（署名済み tx が decode で戻る）と nonce invariant（連続呼び出しが連続 nonce、欠損も重複もなし）で担保する。