レッスン4 — Wallet Backend を Rust で作る
問い
ユーザーが 1 分で 50 回送信しても、nonce を衝突させずに署名・送信し続けたい。さらにガス急騰時には stuck tx を置換してセッションの詰まりを防ぐ。並行送信・単調増加 nonce・stuck 処理を同時に扱うバックエンドをどう組むか？（Rust ~250 行: signer pool + nonce manager + send queue + replace-on-stuck + confirm watcher。）
公開する HTTP API はこの形:
$ curl -X POST http://localhost:7000/send \
-H "Content-Type: application/json" \
-d '{
"from": "0xAlice...",
"to": "0xBob...",
"value": "0x16345785d8a0000",
"data": "0x"
}'
{ "tx_hash": "0xabc...", "queued_at": "2026-05-04T12:34:56Z" }
1 回の POST で signer 取得 → nonce 予約 → ガス見積もり → 署名 → 送信 → 監視まで行う。30 秒で着地しなければ watcher が fee を引き上げて再送する。
原理（最小モデル）
- 各アドレスは next nonce の真の source を 1 つだけ持つ — in-process 状態であって新しい RPC コールではない。 初回だけ
provider.get_transaction_count(addr).pending()で初期化し、以降の予約はローカルArc<Mutex<HashMap<Address, u64>>>で進む。forgetでキャッシュ破棄 → 次回 RPC で再同期（回復手段）。 - 送信と確認を分離する。
send_raw_transaction後は即 return し、確認は別タスクの watcher に任せる。tx hash は「受け付けた」であって「着地した」ではない。 - stuck は同一 nonce + 高 fee で置換する（待機でない）。 先行 nonce が詰まると後続が全部止まる。大半のノードは fee を ≥10% 上げない置換を拒否するので、bump は 25%（最低置換幅 10% より余裕）。
- RPC 中はロックを保持しない。 watcher はキューをスナップショットしてから RPC を回す（キュー全体の直列化を避ける）。
データ経路を 1 枚で:
flowchart TB
Client["HTTP client"] -->|POST /send| API["axum handler"]
API -->|reserve nonce| NM["NonceManager<br/>(per-address)"]
API -->|build & sign| Signer["PrivateKeySigner<br/>(env からロード)"]
API -->|broadcast| Provider["Alloy Provider"]
API -->|track| Q["pending queue<br/>(tx_hash, deadline, fee)"]
Q -->|every 5s| Watcher["confirm watcher"]
Watcher -->|landed?| Provider
Watcher -->|stuck > 30s| Bump["bump fee 1.25x<br/>+ resubmit"]
Bump --> Q
具体例
signer pool + nonce manager（コア不変条件: 各アドレスの next nonce はローカル 1 source）:
use alloy_primitives::Address;
use alloy_signer_local::PrivateKeySigner;
use std::{collections::HashMap, sync::Arc};
use tokio::sync::Mutex;
pub struct SignerPool {
inner: HashMap<Address, PrivateKeySigner>,
}
impl SignerPool {
pub fn from_env() -> eyre::Result<Self> {
// SIGNERS env 変数: comma 区切りの 0x プレフィクス付き秘密鍵
let mut inner = HashMap::new();
for hex in std::env::var("SIGNERS")?.split(',') {
let signer: PrivateKeySigner = hex.trim().parse()?;
inner.insert(signer.address(), signer);
}
Ok(Self { inner })
}
pub fn get(&self, addr: &Address) -> Option<&PrivateKeySigner> {
self.inner.get(addr)
}
}
#[derive(Clone)]
pub struct NonceManager {
state: Arc<Mutex<HashMap<Address, u64>>>,
}
impl NonceManager {
pub fn new() -> Self {
Self { state: Arc::new(Mutex::new(HashMap::new())) }
}
/// `addr` の next nonce を予約、初回は RPC で初期化
pub async fn reserve<P: alloy_provider::Provider>(
&self,
addr: Address,
provider: &P,
) -> eyre::Result<u64> {
let mut state = self.state.lock().await;
let nonce = match state.get(&addr) {
Some(&n) => n,
None => provider.get_transaction_count(addr).pending().await?,
};
state.insert(addr, nonce + 1);
Ok(nonce)
}
/// `addr` のキャッシュ nonce をリセット (回復不能な submission 失敗後に呼ぶ)
pub async fn forget(&self, addr: Address) {
self.state.lock().await.remove(&addr);
}
}
ガス見積もり（EIP-1559、推定を手書きせず provider に）+ bump:
use alloy_eips::eip1559::Eip1559Estimation;
#[derive(Clone, Copy, Debug)]
pub struct GasParams {
pub max_fee_per_gas: u128,
pub max_priority_fee_per_gas: u128,
}
pub async fn estimate_gas<P: alloy_provider::Provider>(provider: &P) -> eyre::Result<GasParams> {
let est: Eip1559Estimation = provider.estimate_eip1559_fees().await?;
Ok(GasParams {
max_fee_per_gas: est.max_fee_per_gas,
max_priority_fee_per_gas: est.max_priority_fee_per_gas,
})
}
pub fn bump(params: GasParams) -> GasParams {
// 25% 引き上げ — 大半のクライアントの mempool 最低値 10% より余裕を持たせる
GasParams {
max_fee_per_gas: params.max_fee_per_gas * 125 / 100,
max_priority_fee_per_gas: params.max_priority_fee_per_gas * 125 / 100,
}
}
送信パス（nonce は署名前に予約、送信後は即 return）:
use alloy_consensus::{TxEip1559, SignableTransaction};
use alloy_network::{TxSignerSync, TransactionBuilder};
use alloy_primitives::{Bytes, U256};
use alloy_rpc_types::TransactionRequest;
use std::time::{Duration, Instant};
#[derive(Clone)]
pub struct PendingTx {
pub from: Address,
pub nonce: u64,
pub current_hash: alloy_primitives::B256,
pub gas_params: GasParams,
pub deadline: Instant,
pub original_request: TransactionRequest,
}
pub async fn send_one<P: alloy_provider::Provider>(
provider: &P,
pool: &SignerPool,
nm: &NonceManager,
req: TransactionRequest,
) -> eyre::Result<PendingTx> {
let from = req.from.ok_or_else(|| eyre::eyre!("from required"))?;
let signer = pool.get(&from).ok_or_else(|| eyre::eyre!("unknown signer: {from}"))?;
let nonce = nm.reserve(from, provider).await?;
let gas = estimate_gas(provider).await?;
let chain_id = provider.get_chain_id().await?;
let req = req
.with_nonce(nonce)
.with_chain_id(chain_id)
.with_gas_limit(req.gas.unwrap_or(100_000))
.with_max_fee_per_gas(gas.max_fee_per_gas)
.with_max_priority_fee_per_gas(gas.max_priority_fee_per_gas);
let tx = req.clone().build(&signer.clone().into()).await?;
let raw = tx.encoded_2718();
let pending = provider.send_raw_transaction(&raw).await?;
Ok(PendingTx {
from,
nonce,
current_hash: *pending.tx_hash(),
gas_params: gas,
deadline: Instant::now() + Duration::from_secs(30),
original_request: req,
})
}
replace-on-stuck 付き confirm watcher（1 タスクが全 queued tx を監視、deadline 超過は bump + resubmit）:
use std::collections::HashMap;
use tokio::sync::RwLock;
use tokio::time::sleep;
#[derive(Clone)]
pub struct PendingQueue {
inner: Arc<RwLock<HashMap<alloy_primitives::B256, PendingTx>>>,
}
impl PendingQueue {
pub fn new() -> Self { Self { inner: Arc::new(RwLock::new(HashMap::new())) } }
pub async fn insert(&self, ptx: PendingTx) {
self.inner.write().await.insert(ptx.current_hash, ptx);
}
}
pub async fn watcher<P: alloy_provider::Provider + Clone>(
provider: P,
pool: SignerPool,
queue: PendingQueue,
) {
loop {
sleep(Duration::from_secs(5)).await;
// RPC 中に read lock を持たないようスナップショット
let snapshot: Vec<PendingTx> = queue.inner.read().await.values().cloned().collect();
for mut ptx in snapshot {
// Inclusion チェック
if let Ok(Some(_receipt)) = provider.get_transaction_receipt(ptx.current_hash).await {
queue.inner.write().await.remove(&ptx.current_hash);
tracing::info!(hash = ?ptx.current_hash, "landed");
continue;
}
// Stuck? bump して resubmit
if Instant::now() >= ptx.deadline {
let bumped = bump(ptx.gas_params);
let signer = pool.get(&ptx.from).expect("signer missing");
let req = ptx.original_request
.clone()
.with_max_fee_per_gas(bumped.max_fee_per_gas)
.with_max_priority_fee_per_gas(bumped.max_priority_fee_per_gas);
match req.build(&signer.clone().into()).await {
Ok(tx) => {
let raw = tx.encoded_2718();
if let Ok(p) = provider.send_raw_transaction(&raw).await {
let new_hash = *p.tx_hash();
let mut w = queue.inner.write().await;
w.remove(&ptx.current_hash);
ptx.current_hash = new_hash;
ptx.gas_params = bumped;
ptx.deadline = Instant::now() + Duration::from_secs(30);
w.insert(new_hash, ptx);
tracing::warn!("bumped + resubmitted");
}
}
Err(e) => tracing::error!(?e, "rebuild failed"),
}
}
}
}
}
HTTP API スケルトン（axum）— watcher を spawn して /send を配線:
use axum::{extract::State, routing::post, Json, Router};
use serde::{Deserialize, Serialize};
#[derive(Clone)]
pub struct AppState<P: alloy_provider::Provider + Clone + 'static> {
pub provider: P,
pub signers: Arc<SignerPool>,
pub nonces: NonceManager,
pub queue: PendingQueue,
}
#[derive(Deserialize)]
pub struct SendRequest {
from: Address,
to: Address,
value: U256,
data: Option<Bytes>,
gas: Option<u64>,
}
#[derive(Serialize)]
pub struct SendResponse {
tx_hash: alloy_primitives::B256,
}
async fn handle_send<P: alloy_provider::Provider + Clone + 'static>(
State(state): State<AppState<P>>,
Json(req): Json<SendRequest>,
) -> Result<Json<SendResponse>, (axum::http::StatusCode, String)> {
let tx_req = TransactionRequest::default()
.with_from(req.from)
.with_to(req.to)
.with_value(req.value)
.with_input(req.data.unwrap_or_default())
.with_gas_limit(req.gas.unwrap_or(100_000));
let pending = send_one(&state.provider, &state.signers, &state.nonces, tx_req)
.await
.map_err(|e| (axum::http::StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
let hash = pending.current_hash;
state.queue.insert(pending).await;
Ok(Json(SendResponse { tx_hash: hash }))
}
#[tokio::main]
async fn main() -> eyre::Result<()> {
tracing_subscriber::fmt::init();
// Provider 例: QuickNode、Alchemy、Infura、または自前 Reth ノード。
let provider = alloy_provider::ProviderBuilder::new()
.connect(&std::env::var("RPC_URL")?)
.await?;
let state = AppState {
provider: provider.clone(),
signers: Arc::new(SignerPool::from_env()?),
nonces: NonceManager::new(),
queue: PendingQueue::new(),
};
// Watcher を spawn
{
let p = provider.clone();
let s = (*state.signers).clone(); // Note: SignerPool に Clone が必要
let q = state.queue.clone();
tokio::spawn(async move { watcher(p, s, q).await });
}
let app = Router::new()
.route("/send", post(handle_send))
.with_state(state);
let listener = tokio::net::TcpListener::bind("0.0.0.0:7000").await?;
axum::serve(listener, app).await?;
Ok(())
}
これで最小サービスは完成（インポート込みで約 250 行）。
失敗例（誤解）
「送信ごとに provider.get_transaction_count(from).await で nonce を取ればいい」は誤り — 並行する 2 つの送信が両方 nonce N を読み、両方 N で署名し、1 つしか着地しない（もう 1 つは mempool で拒否）。同様に「同じ nonce で再送信するだけ」も誤り（fee を ≥10% 上げない置換はノードが拒否、サイレントに失敗）。「eth_sendRawTransaction の戻り値を信じる」も誤り（tx hash は受け付けで、着地でない）。
ここまでで「ローカル nonce 1 source + 送信/確認の分離 + 同一 nonce 高 fee 置換」は着地した。ここから 5 ステップで組み立てる。コードは抜粋（実行時は補助コードが必要）。
🛑 予測。 「nonce 取得→署名→送信」を同一アドレスで 100ms 以内に 2 回実行すると何が壊れるか。（答え: 両方が同じ nonce N を読んで N で署名し、1 つだけ着地する。もう 1 つは「nonce too low / already known」で拒否され、ユーザの 2 件目の送信がサイレントに消える。ローカル nonce 予約はこの競合を mutex 下の
nonce + 1で潰す。）
ステップで組み立てる
Step 0: プロジェクトと依存
# Cargo.toml
[package]
name = "wallet-backend"
version = "0.1.0"
edition = "2021"
[dependencies]
alloy-primitives = "1.5"
alloy-provider = "1.0"
alloy-rpc-types = "1.0"
alloy-network = "1.0"
alloy-signer = "1.0"
alloy-signer-local = "1.0"
alloy-consensus = "2.0"
alloy-eips = "1.0"
axum = "0.7"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
eyre = "0.6"
tracing = "0.1"
tracing-subscriber = "0.3"
秘密鍵は環境変数 SIGNERS にカンマ区切りで渡す（SIGNERS=0xabc...,0xdef...）。RPC エンドポイントは RPC_URL で渡す。
Step 1-5
上の 5 ブロック: ① signer pool + nonce manager（pending() を使う、初回だけ RPC、forget は回復手段）→ ② ガス見積もり（estimate_eip1559_fees()、bump 25%）→ ③ 送信パス（nonce を署名前に予約、送信と確認を分離）→ ④ watcher（先にスナップショット、receipt=Some で着地判定、deadline 超過で bump+resubmit）→ ⑤ axum API（watcher を tokio::spawn）。署名層は差し替え可能（local / keystore / mnemonic / KMS、send フロー本体は保てる）。
production gap: 鍵カストディ（KMS/HSM/MPC へ）/ 冪等性（request_id）/ 鍵単位レート制限 / 永続キュー（DB/Redis）/ マルチ RPC ファンアウト / nonce ギャップ検出 / observability（pending_count 等）。
Step 6: 起動と疎通
$ RPC_URL=https://sepolia.infura.io/v3/$KEY \
SIGNERS=0xabc...,0xdef... \
cargo run --release
# 別ターミナルで送信テスト
$ curl -s -X POST http://localhost:7000/send \
-H "Content-Type: application/json" \
-d '{"from":"0xAlice...","to":"0xBob...","value":"0x16345785d8a0000","data":"0x"}'
{ "tx_hash": "0x...", "queued_at": "..." }
答え合わせ（Test gate）
最低ラインは不変条件 2 つ（roundtrip + 並行 nonce 単調性）:
// tests/wallet_invariants.rs
use alloy::consensus::TxEnvelope;
use alloy::eips::Decodable2718;
#[tokio::test]
async fn signed_tx_roundtrips() {
let svc = test_service().await;
let req = SendRequest { from: ALICE, to: BOB, value: U256::from(1), data: vec![] };
let signed_bytes = svc.sign_only(&req).await.unwrap();
let decoded = TxEnvelope::decode_2718(&mut signed_bytes.as_slice()).unwrap();
assert_eq!(decoded.recover_signer().unwrap(), ALICE);
assert_eq!(decoded.tx().to(), Some(BOB));
assert_eq!(decoded.tx().value(), U256::from(1));
}
#[tokio::test]
async fn no_nonce_gaps_under_concurrent_send() {
let svc = test_service().await;
let base = svc.next_nonce(ALICE).await;
let handles: Vec<_> = (0..50)
.map(|_| {
let svc = svc.clone();
tokio::spawn(async move { svc.send(stub_request(ALICE)).await.unwrap() })
})
.collect();
let mut nonces: Vec<u64> = futures::future::try_join_all(handles).await.unwrap()
.into_iter().map(|r| r.nonce).collect();
nonces.sort();
let expected: Vec<u64> = (base..base + 50).collect();
assert_eq!(nonces, expected, "nonce は連続かつユニークである必要がある");
}
両方 pass まで未完了。前者が崩れると不正 tx を作り、後者が崩れると nonce 詰まりで送信が停止する。
合格基準
- 上記 2 テスト（roundtrip + 50 並行送信で nonce
base..base+50に欠損/重複なし）が green。 - なぜローカル nonce 状態が核か（nonce ごとの RPC 往復なしで並行送信）を 1 文で言える。
- bump 幅を 25% にする理由（最低置換幅 10% より余裕）を即答できる。
Drill
SendRequestにrequest_idを足しrequest_id → tx_hashを 1 時間キャッシュ（重複 POST にキャッシュ hash を返す）（30 分）。/sendを per-from semaphore（max 4 並行）でラップ、超過は 429（30 分）。PendingTxを insert 時に Redis、着地時に削除、起動時に復元（1.5 時間）。- 2 provider に
send_raw_transactionをブロードキャストし最初のOkを返すMultiProvider（1 時間）。 POST /cancel { from, nonce }を追加（同 nonce で 50% 引き上げた 0-value 自送）（1 時間）。
📺 関連動画
wJnywGB33O4 | Georgios Konstantopoulos — Foundry, a portable, fast and modular toolkit (Foundry の tx パイプライン内で使われている同じ Alloy + Rust signer 機構)
まとめ（3行）
- wallet backend の核はローカル nonce 状態 — 各アドレスの next nonce を in-process 1 source に置き、mutex 下の
nonce + 1で並行送信の競合を潰す（送信ごとの RPC 往復なし）。 - 送信と確認を分離:
send_oneは即 return、別タスクの watcher が receipt をポーリングし deadline 超過で同一 nonce + 25% bump で置換（待機でなく置換 — 先行詰まりが後続を止めるから）。 - Test gate: tx エンコード roundtrip + 並行下の nonce 単調性。次は認証層の EIP-7702 sponsor。
次のレッスン（レッスン5）
最小限の EIP-7702 Sponsor サービス（Type 4 tx + paymaster パターン）。replay 防止（同じ auth tuple は 2 度 sponsor できない）と gas 会計（sponsor が払い、ユーザは 0）で担保する。