diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index c95550c9..d25ad58a 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -1,9 +1,8 @@ -//! AlloyChainClient implementation. -//! -//! This is the production implementation of the `ChainClient` trait using +//! AlloyChainClient: the production implementation of the `ChainClient` trait, using //! alloy for Ethereum interactions. use std::{ + future::Future, sync::{ Arc, atomic::{AtomicU64, Ordering}, @@ -14,13 +13,14 @@ use std::{ use async_trait::async_trait; use dipper_rpc::indexer::indexer_client::sol::RecurringCollectionAgreement; use thegraph_core::alloy::{ - eips::BlockNumberOrTag, + eips::{BlockNumberOrTag, eip2718::Encodable2718}, network::{EthereumWallet, TransactionBuilder}, primitives::{Address, B256, FixedBytes}, - providers::{Provider, ProviderBuilder}, + providers::Provider, rpc::types::TransactionRequest, signers::local::PrivateKeySigner, sol_types::{SolCall, SolValue}, + transports::TransportError, }; use tokio::sync::Mutex; @@ -32,19 +32,17 @@ use super::{ use crate::{ chain_client::{ChainClient, ChainClientError}, config::ChainClientConfig, + worker::service::PROCESS_JOB_TIMEOUT, }; -/// OFFER_TYPE_NEW from `RecurringCollector.sol`. Used when submitting a new -/// agreement offer on-chain. The contract defines OFFER_TYPE_NONE=0, -/// OFFER_TYPE_NEW=1, OFFER_TYPE_UPDATE=2; passing 0 reverts with +/// OFFER_TYPE_NEW from `RecurringCollector.sol`, used when submitting a new agreement +/// offer on-chain. The contract defines NONE=0, NEW=1, UPDATE=2; passing 0 reverts with /// RecurringCollectorInvalidOfferType(0). const OFFER_TYPE_NEW: u8 = 1; -/// Time to wait for a tx receipt to appear before declaring the tx dropped -/// from the mempool. On hardhat this is ~15 blocks at 1s each; on Arbitrum -/// at 0.25s block time this is 60 confirmations. Short enough that the -/// pgmq retry budget can recover within the 300s RCA deadline, long enough -/// to tolerate typical network glitches. +/// Time to wait for a tx receipt before declaring the tx dropped from the mempool: ~15 +/// blocks on hardhat at 1s each, 60 confirmations on Arbitrum at 0.25s. Short enough that +/// the pgmq retry budget recovers inside the RCA deadline `deadline_seconds` sets. const RECEIPT_POLL_TIMEOUT: Duration = Duration::from_secs(15); /// Interval between `eth_getTransactionReceipt` polls while waiting for a @@ -70,6 +68,8 @@ const NONCE_ERROR_PATTERNS: &[&str] = &[ "nonce is too low", "invalid nonce", "replacement transaction underpriced", + // A backstop only. `is_already_broadcast` claims this wording first and reports the + // broadcast as the success it is, so a send never reaches here saying it. "already known", ]; @@ -79,6 +79,60 @@ fn is_nonce_error(error: &str) -> bool { NONCE_ERROR_PATTERNS.iter().any(|p| lower.contains(p)) } +/// Number of attempts `sign_and_send` makes at finding a nonce the chain will take. +const MAX_NONCE_RETRIES: u32 = 2; + +/// How long one submission may hold `submit_lock` before giving up, derived from the retry +/// schedule so the cut-off can never pre-empt a retry the config allows: each nonce attempt +/// is at worst one full walk of the endpoint ring to read the nonce and another to send. +fn derive_submit_deadline(pool: &RpcProviderPool) -> Duration { + let budget = pool.worst_case_walk() * (2 * MAX_NONCE_RETRIES); + // The rest of the worker job's budget stays reserved for what follows the broadcast: + // the receipt poll and the nonce-gap fill. + let cap = PROCESS_JOB_TIMEOUT / 5 * 4; + if budget > cap { + tracing::warn!( + budget_secs = budget.as_secs(), + cap_secs = cap.as_secs(), + "retry schedule wants more time than a worker job allows; submissions may give \ + up before trying every endpoint" + ); + return cap; + } + budget +} + +/// Run a submission under `deadline`, reporting a failed submission if it runs over. +/// Giving up releases the lock, and the caller re-runs the job rather than losing the work. +async fn under_submit_deadline( + deadline: Duration, + work: F, +) -> Result +where + F: Future>, +{ + tokio::time::timeout(deadline, work) + .await + .unwrap_or_else(|_| { + Err(ChainClientError::SubmitFailed(anyhow::anyhow!( + "gave up submitting after {}s holding the submission lock", + deadline.as_secs() + ))) + }) +} + +/// How nodes say they are already holding the transaction being offered to them. +const ALREADY_BROADCAST_PATTERNS: &[&str] = + &["already known", "already imported", "known transaction"]; + +/// Whether a rejection means this exact transaction is already in a mempool. Offering the +/// same signed bytes to a second endpoint is expected to land here, and it means the +/// broadcast succeeded, so it must not be mistaken for a reason to send another. +fn is_already_broadcast(error: &TransportError) -> bool { + let lower = error.to_string().to_lowercase(); + ALREADY_BROADCAST_PATTERNS.iter().any(|p| lower.contains(p)) +} + /// Classify the outcome of a nonce-gap fill submission. Pure so the /// swallow rule (`is_nonce_error` → `Ok(())`) is unit-testable without /// an RPC mock. @@ -117,10 +171,8 @@ fn classify_fill_nonce_gap_outcome( } } -/// Production implementation of `ChainClient` using alloy. -/// -/// This struct is `Clone` via internal `Arc` wrapping, allowing it to be shared -/// across async task contexts. +/// Production implementation of `ChainClient` using alloy. `Clone` via internal `Arc` +/// wrapping, so it can be shared across async task contexts. #[derive(Clone)] pub struct AlloyChainClient { /// Inner state wrapped in Arc for Clone support @@ -158,26 +210,21 @@ struct AlloyChainClientInner { gas_price_multiplier: f64, /// Maximum gas price in gwei max_gas_price_gwei: u64, - /// In-memory nonce counter. Concurrent callers atomically increment this - /// to get unique nonces without querying the chain, avoiding - /// "replacement transaction underpriced" errors when multiple offer() - /// transactions are submitted in parallel. + /// The next nonce a submission should use. Read under `submit_lock` and advanced only + /// once a broadcast succeeds, so a failed submission cannot skip a slot the chain + /// would then sit waiting on while later transactions queue behind the gap. nonce: AtomicU64, - /// Serializes nonce reservation through mempool submission so the RPC - /// sees txs in nonce order and the wallet's queue can never strand a - /// higher-nonce tx behind an unfilled gap. Released before receipt - /// polling so multiple confirmations pipeline concurrently. + /// Serializes reading the nonce counter through committing it after the mempool + /// submission, so two submissions can never take the same slot. Released before + /// receipt polling so multiple confirmations pipeline concurrently. submit_lock: Mutex<()>, + /// How long one submission may hold `submit_lock`; see `derive_submit_deadline`. + submit_deadline: Duration, } impl AlloyChainClient { - /// Create a new AlloyChainClient from configuration. - /// - /// # Errors - /// - /// Returns an error if: - /// - No RPC providers are configured - /// - The signer cannot be constructed + /// Create a new AlloyChainClient from configuration. Errors if no RPC providers are + /// configured or the signer cannot be constructed. pub fn new( config: &ChainClientConfig, chain_id: u64, @@ -208,6 +255,8 @@ impl AlloyChainClient { "AlloyChainClient initialized" ); + let submit_deadline = derive_submit_deadline(&rpc_pool); + Ok(Self { inner: Arc::new(AlloyChainClientInner { rpc_pool, @@ -220,14 +269,13 @@ impl AlloyChainClient { max_gas_price_gwei: config.max_gas_price_gwei, nonce: AtomicU64::new(NONCE_UNINITIALIZED), submit_lock: Mutex::new(()), + submit_deadline, }), }) } - /// Build, gas-estimate, and send a call to any contract. - /// - /// Shared entry point for the manager-routed offer and cancel calls. - /// `log_agreement_id` is used only for structured logging. + /// Build, gas-estimate, and send a call to any contract. Shared entry point for the + /// manager-routed offer and cancel calls; `log_agreement_id` is only for logging. async fn build_and_send_call( &self, to: Address, @@ -240,12 +288,9 @@ impl AlloyChainClient { .to(to) .input(calldata.into()); - // 2. Estimate gas with safety bounds. - // - // The estimator may surface a structured contract revert (e.g. an - // already-canceled agreement). Box the typed error through alloy's - // Custom transport variant so `rpc_pool.execute` can hand it back - // without losing the selector and revert payload. + // 2. Estimate gas with safety bounds. The estimator may surface a structured + // contract revert (e.g. an already-canceled agreement); box it through alloy's + // Custom transport variant so the pool hands back the selector and payload intact. let gas_limit = self .inner .rpc_pool @@ -307,49 +352,36 @@ impl AlloyChainClient { self.sign_and_send(tx, log_agreement_id).await } - /// Get the next nonce, initializing from chain on first call. - /// - /// Concurrent callers each get a unique nonce via atomic - /// fetch-and-increment, avoiding the "replacement transaction - /// underpriced" race when multiple offer() calls fire in parallel. + /// The slot the next submission should use, read from the chain on first call. Only + /// `commit_nonce` moves the counter on, so a submission that never got a transaction + /// out leaves its slot to the next caller instead of leaving the chain waiting on it. async fn next_nonce(&self) -> Result { let current = self.inner.nonce.load(Ordering::SeqCst); - if current == NONCE_UNINITIALIZED { - let chain_nonce = self.fetch_chain_nonce().await?; - // CAS: if another task already initialized, use its value - match self.inner.nonce.compare_exchange( - NONCE_UNINITIALIZED, - chain_nonce + 1, - Ordering::SeqCst, - Ordering::SeqCst, - ) { - Ok(_) => return Ok(chain_nonce), - // Another task already initialized; get next unique nonce - Err(_) => return Ok(self.inner.nonce.fetch_add(1, Ordering::SeqCst)), - } + if current != NONCE_UNINITIALIZED { + return Ok(current); } - Ok(self.inner.nonce.fetch_add(1, Ordering::SeqCst)) + let chain_nonce = self.fetch_chain_nonce().await?; + self.inner.nonce.store(chain_nonce, Ordering::SeqCst); + Ok(chain_nonce) + } + + /// Record that `nonce` carried a successful broadcast, so the next submission moves on + /// to the following slot. + fn commit_nonce(&self, nonce: u64) { + self.inner.nonce.fetch_max(nonce + 1, Ordering::SeqCst); } - /// Ratchet the in-memory nonce counter up to `chain_pending + 1`. - /// Never decreases the counter, so it is safe to call from any - /// context without `submit_lock` held: an in-flight reservation - /// from another caller cannot be invalidated. Callers needing a - /// reservation call `next_nonce()` after. + /// Ratchet the in-memory counter up to the next slot the chain will accept. Never + /// decreases it, so an endpoint reporting a stale pending count cannot wind the + /// counter back onto a slot a broadcast has already spent. async fn resync_nonce(&self) -> Result<(), ChainClientError> { let chain_nonce = self.fetch_chain_nonce().await?; - self.inner - .nonce - .fetch_max(chain_nonce + 1, Ordering::SeqCst); + self.inner.nonce.fetch_max(chain_nonce, Ordering::SeqCst); Ok(()) } - /// Fetch the pending transaction count from chain. - /// - /// Uses the "pending" block tag so the count includes transactions - /// sitting in the mempool from our wallet. Querying "latest" would - /// return a stale count when prior txs are awaiting confirmation, - /// causing the next tx to reuse a nonce that's already in-flight. + /// Fetch the pending transaction count from chain. The "pending" tag counts our own + /// mempool transactions too; "latest" would miss them and reuse an in-flight nonce. async fn fetch_chain_nonce(&self) -> Result { self.inner .rpc_pool @@ -360,21 +392,30 @@ impl AlloyChainClient { .await } - /// Sign and send a transaction with nonce error handling. - /// - /// Uses the in-memory nonce counter for the first attempt. On nonce - /// errors, re-syncs from chain and retries. `submit_lock` spans the - /// retry loop so concurrent reservations cannot interleave with each - /// other's submissions; released before receipt polling. + /// Sign and send a transaction, re-syncing the nonce from chain and retrying on nonce + /// errors. `submit_lock` spans the retry loop so two submissions can never interleave; + /// released before receipt polling so confirmations pipeline concurrently. async fn sign_and_send( &self, - mut tx: TransactionRequest, + tx: TransactionRequest, agreement_id: &[u8; 16], ) -> Result { - const MAX_NONCE_RETRIES: u32 = 2; - let _submit_guard = self.inner.submit_lock.lock().await; + under_submit_deadline( + self.inner.submit_deadline, + self.sign_and_send_locked(tx, agreement_id), + ) + .await + } + /// The work `sign_and_send` does while holding `submit_lock`, split out so the deadline + /// wraps the work rather than the wait for the lock: a caller queueing politely behind + /// someone else should not be charged for the time it spent waiting. + async fn sign_and_send_locked( + &self, + mut tx: TransactionRequest, + agreement_id: &[u8; 16], + ) -> Result { for attempt in 0..MAX_NONCE_RETRIES { if attempt > 0 { self.resync_nonce().await?; @@ -387,6 +428,7 @@ impl AlloyChainClient { match result { Ok(tx_hash) => { + self.commit_nonce(nonce); tracing::info!( agreement_id = %format_args!("0x{}", agreement_id.iter().map(|b| format!("{b:02x}")).collect::()), tx_hash = %tx_hash, @@ -418,12 +460,9 @@ impl AlloyChainClient { ))) } - /// Submit a self-transfer of 0 wei at `nonce` so the chain has - /// something to mine in a slot left empty by an evicted tx, - /// releasing higher-nonce txs from the same wallet that were - /// stuck behind the gap. Best-effort: an `is_nonce_error` - /// rejection means the original is still in flight or the gap is - /// already filled, so we treat that as success. + /// Submit a self-transfer of 0 wei at `nonce` so the chain has something to mine in a + /// slot left empty by an evicted tx, releasing higher-nonce txs stuck behind the gap. + /// Best-effort: an `is_nonce_error` rejection means the slot is spoken for, so success. async fn fill_nonce_gap(&self, nonce: u64) -> Result<(), ChainClientError> { let _submit_guard = self.inner.submit_lock.lock().await; @@ -455,35 +494,62 @@ impl AlloyChainClient { classify_fill_nonce_gap_outcome(nonce, self.send_transaction(&tx).await) } + /// Broadcast a transaction, retrying and rotating endpoints the way every read call + /// already does. Signing happens once, up front, so every endpoint is offered the same + /// bytes under one hash and the hash is known before anyone is asked to accept them. async fn send_transaction(&self, tx: &TransactionRequest) -> Result { - let wallet = EthereumWallet::from(self.inner.signer.clone()); - let url = self.inner.rpc_pool.current_url().clone(); - - // Build HTTP client with timeout - let client = reqwest::Client::builder() - .timeout(self.inner.rpc_pool.request_timeout()) - .build() - .map_err(|e| { - ChainClientError::ConfigError(format!("Failed to build HTTP client: {e}")) - })?; + // Nothing fills a field in on this path any more, and a request that names no chain is + // signed for chain 1 rather than refused, so check before the signature exists. Not a + // `ConfigError`: the cancel path reads that as the chain client being switched off. + if tx.chain_id() != Some(self.inner.chain_id) { + return Err(ChainClientError::SubmitFailed(anyhow::anyhow!( + "refusing to sign for chain {:?} while configured for chain {}", + tx.chain_id(), + self.inner.chain_id + ))); + } - // Build wallet-enabled provider - let provider = ProviderBuilder::new() - .wallet(wallet) - .connect_reqwest(client, url); + let wallet = EthereumWallet::from(self.inner.signer.clone()); + // A caller leaving out a field fails here rather than having it filled in, so this is + // as likely to be an incomplete request as a signing fault. Alloy's own wording names + // which, so leave the reason to it. + let signed = tx.clone().build(&wallet).await.map_err(|e| { + ChainClientError::SubmitFailed(anyhow::anyhow!("Transaction not ready to send: {e}")) + })?; + let signed_hash = *signed.tx_hash(); + let raw = signed.encoded_2718(); - let pending = provider - .send_transaction(tx.clone()) + self.inner + .rpc_pool + .execute("send_transaction", |provider| { + let raw = raw.clone(); + async move { + match provider.send_raw_transaction(&raw).await { + // The hash follows from the bytes, so report the one we signed + // instead of what this endpoint echoed back. A wrong hash sends the + // receipt poll after a transaction that never mines. + Ok(_) => Ok(signed_hash), + // This endpoint already holds these exact bytes, which is the + // outcome we were after. Answer with the hash we signed rather + // than reporting a failure that would send a second transaction. + Err(e) if is_already_broadcast(&e) => Ok(signed_hash), + Err(e) => Err(e), + } + } + }) .await - .map_err(|e| ChainClientError::SubmitFailed(anyhow::anyhow!("Send failed: {e}")))?; - - Ok(*pending.tx_hash()) + .map_err(|e| match e { + // A submission that got nowhere is a failed submission, whatever the + // endpoints happened to say. Anything the pool could name precisely, such + // as a rejection from the contract, keeps the name it already has. + ChainClientError::RpcError(cause) => ChainClientError::SubmitFailed(cause), + other => other, + }) } - /// Poll `eth_getTransactionReceipt` until the tx has mined or the timeout - /// elapses. `Ok(Some(status))` reports the receipt's success flag; - /// `Ok(None)` signals the tx never appeared in time (dropped from the - /// mempool). Transient RPC errors are retried silently until timeout. + /// Poll `eth_getTransactionReceipt` until the tx has mined or the timeout elapses. + /// `Ok(Some(status))` reports the receipt's success flag; `Ok(None)` says the tx never + /// appeared in time (dropped from the mempool). Transient RPC errors keep polling. async fn wait_for_receipt( &self, tx_hash: B256, @@ -503,10 +569,8 @@ impl AlloyChainClient { Ok(Some(r)) => return Ok(Some(r.status())), Ok(None) => {} // not mined yet Err(e) => { - // Transient RPC error — log and keep polling. If the - // error is persistent, the outer handler will see the - // eventual timeout as `Ok(None)` and resubmit, which is - // the safe default. + // Transient RPC error: log and keep polling. If it persists, the outer + // handler sees the timeout as `Ok(None)` and resubmits, the safe default. tracing::debug!( tx_hash = %tx_hash, error = %e, @@ -712,15 +776,652 @@ impl ChainClient for AlloyChainClient { #[cfg(test)] mod tests { + use url::Url; + use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate, matchers::method}; + use super::*; + /// Answers a send with a fixed transaction hash, echoing the request id so alloy's + /// transport accepts the response. Any other call is a mistake in the test rather than + /// something to answer with a hash, so say so instead of returning nonsense. + struct SendResponder { + tx_hash: B256, + } + + impl Respond for SendResponder { + fn respond(&self, request: &Request) -> ResponseTemplate { + let body: serde_json::Value = + serde_json::from_slice(&request.body).expect("JSON-RPC request body"); + let method = body["method"].as_str().unwrap_or_default(); + assert!( + method.starts_with("eth_send"), + "this mock only answers sends, got {method}" + ); + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": format!("{tx_hash:#x}", tx_hash = self.tx_hash), + })) + } + } + + async fn server_answering_with(tx_hash: B256) -> MockServer { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(SendResponder { tx_hash }) + .mount(&server) + .await; + server + } + + async fn server_answering_500() -> MockServer { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500).set_body_string( + r#"{"id":0,"jsonrpc":"2.0","error":{"message":"Temporary internal error. Please retry","code":19}}"#, + )) + .mount(&server) + .await; + server + } + + /// Answers HTTP 200 carrying a JSON-RPC error, which is how a chain reports a rejection + /// such as a stale nonce and how some providers report being overloaded. + async fn server_answering_rpc_error(code: i64, message: &str) -> MockServer { + let server = MockServer::start().await; + let body = serde_json::json!({ + "id": 0, + "jsonrpc": "2.0", + "error": { "code": code, "message": message }, + }); + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(body)) + .mount(&server) + .await; + server + } + + fn client_over(providers: Vec) -> AlloyChainClient { + client_over_retrying(providers, 0) + } + + fn client_over_retrying(providers: Vec, max_retries: u32) -> AlloyChainClient { + let config = ChainClientConfig { + enabled: true, + providers, + request_timeout: Duration::from_secs(5), + max_retries, + domain_refresh_interval: Duration::from_secs(3600), + gas_price_multiplier: 1.2, + max_gas_price_gwei: 100, + gas_buffer_multiplier: 2.0, + gas_floor: 100_000, + gas_max_addition: 200_000, + }; + + AlloyChainClient::new( + &config, + 1337, + Address::repeat_byte(0x11), + Address::repeat_byte(0x22), + &[0x42; 32], + ) + .expect("chain client") + } + + /// A transaction with everything filled, so no filler needs to reach the network and + /// the bytes signed are identical whichever provider receives them. + fn ready_to_send_tx(from: Address) -> TransactionRequest { + TransactionRequest::default() + .from(from) + .to(Address::repeat_byte(0x33)) + .value(thegraph_core::alloy::primitives::U256::ZERO) + .with_gas_limit(21_000) + .with_max_fee_per_gas(2_000_000_000) + .with_max_priority_fee_per_gas(1_000_000_000) + .with_chain_id(1337) + .with_nonce(7) + } + + /// The hash the signed bytes carry, which is what a send reports. + async fn signed_hash_of(client: &AlloyChainClient, tx: &TransactionRequest) -> B256 { + let wallet = EthereumWallet::from(client.inner.signer.clone()); + *tx.clone() + .build(&wallet) + .await + .expect("sign the transaction") + .tx_hash() + } + + /// Submitting a transaction must fail over to the next provider, exactly as every read + /// call does. On 2026-07-29 the send bypassed the pool, so one endpoint answering 500 + /// stranded an accepted agreement while a healthy second endpoint was never tried. + #[tokio::test] + async fn send_transaction_rotates_to_the_next_provider_on_server_fault() { + let sick = server_answering_500().await; + let healthy = server_answering_with(B256::repeat_byte(0xab)).await; + + let client = client_over(vec![ + sick.uri().parse().expect("sick provider URL"), + healthy.uri().parse().expect("healthy provider URL"), + ]); + let tx = ready_to_send_tx(client.inner.signer.address()); + + let tx_hash = client + .send_transaction(&tx) + .await + .expect("send should succeed on the second provider"); + + assert_eq!( + tx_hash, + signed_hash_of(&client, &tx).await, + "the hash reported must be the one we signed, not the one the endpoint made up" + ); + assert!( + !sick + .received_requests() + .await + .unwrap_or_default() + .is_empty(), + "the failing provider should have been tried first" + ); + assert!( + !healthy + .received_requests() + .await + .unwrap_or_default() + .is_empty(), + "the healthy provider should have been tried after rotation" + ); + } + + /// With a single provider there is nowhere to rotate to, so the fault must surface rather + /// than be retried forever or silently swallowed. It surfaces as a failed submission rather + /// than the pool's generic RPC fault, which is what says the transaction went nowhere. + #[tokio::test] + async fn send_transaction_reports_failure_when_every_provider_is_sick() { + let sick = server_answering_500().await; + let client = client_over(vec![sick.uri().parse().expect("sick provider URL")]); + let tx = ready_to_send_tx(client.inner.signer.address()); + + let err = client + .send_transaction(&tx) + .await + .expect_err("a 500 from the only provider must error"); + + assert!( + matches!(err, ChainClientError::SubmitFailed(_)), + "got {err}" + ); + } + + /// What the chain said has to survive the pool, because `sign_and_send` reads this text to + /// recognise a stale nonce and resync from the chain. Routing the send through the pool + /// once replaced the reason with a generic summary, which silently disabled that recovery. + #[tokio::test] + async fn send_transaction_surfaces_what_the_provider_said() { + let rejecting = server_answering_rpc_error(-32000, "nonce too low: next nonce 12").await; + let client = client_over(vec![rejecting.uri().parse().expect("provider URL")]); + let tx = ready_to_send_tx(client.inner.signer.address()); + + let err = client + .send_transaction(&tx) + .await + .expect_err("a rejected transaction must error"); + + let text = err.to_string(); + assert!( + is_nonce_error(&text), + "the nonce reason must still be recognisable, got: {text}" + ); + } + + /// Answers the nonce lookup that precedes a send, then reports every send as already + /// held. Counting the sends is how a duplicate submission shows up. + struct AlreadyHeldResponder { + pending_nonce: u64, + } + + impl Respond for AlreadyHeldResponder { + fn respond(&self, request: &Request) -> ResponseTemplate { + let body: serde_json::Value = + serde_json::from_slice(&request.body).expect("JSON-RPC request body"); + match body["method"].as_str().unwrap_or_default() { + "eth_getTransactionCount" => { + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": format!("{:#x}", self.pending_nonce), + })) + } + "eth_sendRawTransaction" => { + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": { "code": -32000, "message": "already known" }, + })) + } + other => panic!("unexpected method {other}"), + } + } + } + + fn sends_among(requests: &[wiremock::Request]) -> usize { + requests + .iter() + .filter(|r| { + let body: serde_json::Value = + serde_json::from_slice(&r.body).expect("JSON-RPC request body"); + body["method"] == "eth_sendRawTransaction" + }) + .count() + } + + /// The raw transaction each endpoint was asked to accept. + fn broadcast_bytes(requests: &[wiremock::Request]) -> Vec { + requests + .iter() + .filter_map(|r| { + let body: serde_json::Value = + serde_json::from_slice(&r.body).expect("JSON-RPC request body"); + (body["method"] == "eth_sendRawTransaction") + .then(|| body["params"][0].as_str().expect("raw tx").to_string()) + }) + .collect() + } + + /// Rotating between endpoints is only safe while they are all offered the same bytes: two + /// different transactions would mean two chances of both being mined and paid for. This + /// is what makes a rebroadcast a retry of one transaction rather than a second one. + #[tokio::test] + async fn every_endpoint_is_offered_the_same_bytes() { + let sick = server_answering_500().await; + let healthy = server_answering_with(B256::repeat_byte(0xab)).await; + let client = client_over(vec![ + sick.uri().parse().expect("sick provider URL"), + healthy.uri().parse().expect("healthy provider URL"), + ]); + let tx = ready_to_send_tx(client.inner.signer.address()); + + client.send_transaction(&tx).await.expect("send"); + + let offered = [ + broadcast_bytes(&sick.received_requests().await.unwrap_or_default()), + broadcast_bytes(&healthy.received_requests().await.unwrap_or_default()), + ] + .concat(); + assert_eq!(offered.len(), 2, "both endpoints should have been asked"); + assert_eq!( + offered[0], offered[1], + "the two endpoints were offered different transactions" + ); + } + + /// Recovering from a stale nonce has to land on the slot the chain will actually take. + /// Aiming one past it leaves that slot empty, and a transaction behind an empty slot never + /// gets mined, so a wallet could stay stuck until something else happened to fill it. + #[tokio::test] + async fn resync_lands_on_the_next_slot_the_chain_will_accept() { + let pending = Arc::new(AtomicU64::new(3)); + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(PendingNonceResponder { + pending: pending.clone(), + }) + .mount(&server) + .await; + let client = client_over(vec![server.uri().parse().expect("provider URL")]); + + assert_eq!( + client.next_nonce().await.expect("first reservation"), + 3, + "the first reservation takes the slot the chain reports" + ); + + // Something else spent slots 3 to 9, so 10 is now the next one free. + pending.store(10, Ordering::SeqCst); + client.resync_nonce().await.expect("resync"); + + assert_eq!( + client.next_nonce().await.expect("reservation after resync"), + 10, + "the reservation after a resync must not skip the free slot" + ); + } + + /// Reports whatever the pending count currently is, so a test can move the chain on. + struct PendingNonceResponder { + pending: Arc, + } + + impl Respond for PendingNonceResponder { + fn respond(&self, request: &Request) -> ResponseTemplate { + let body: serde_json::Value = + serde_json::from_slice(&request.body).expect("JSON-RPC request body"); + assert_eq!( + body["method"], "eth_getTransactionCount", + "this mock only answers nonce lookups" + ); + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": format!("{:#x}", self.pending.load(Ordering::SeqCst)), + })) + } + } + + /// The first endpoint takes the bytes but its reply is lost, so the same bytes go to the + /// second, which already holds them. That is the outcome we wanted, so it has to be + /// reported with the hash rather than as a failure. + #[tokio::test] + async fn send_transaction_accepts_a_transaction_already_in_a_mempool() { + let sick = server_answering_500().await; + let holding = server_answering_rpc_error(-32000, "already known").await; + let client = client_over(vec![ + sick.uri().parse().expect("sick provider URL"), + holding.uri().parse().expect("holding provider URL"), + ]); + let tx = ready_to_send_tx(client.inner.signer.address()); + + let tx_hash = client + .send_transaction(&tx) + .await + .expect("a transaction already in a mempool is a successful broadcast"); + + assert_eq!( + tx_hash, + signed_hash_of(&client, &tx).await, + "the hash reported must be the one we signed" + ); + } + + /// Treating "we already have it" as a stale nonce made the caller resync and send a + /// second transaction at a different nonce, which for an agreement offer means paying + /// twice for one offer. One broadcast has to stay one broadcast. + #[tokio::test] + async fn sign_and_send_does_not_resend_a_transaction_already_in_a_mempool() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(AlreadyHeldResponder { pending_nonce: 6 }) + .mount(&server) + .await; + let client = client_over(vec![server.uri().parse().expect("provider URL")]); + + client + .sign_and_send(ready_to_send_tx(client.inner.signer.address()), &[0u8; 16]) + .await + .expect("a transaction already in a mempool is a successful broadcast"); + + let requests = server.received_requests().await.unwrap_or_default(); + assert_eq!( + sends_among(&requests), + 1, + "the transaction must be broadcast once, not resent at a new nonce" + ); + } + + /// Nothing fills a field in on the send path, and a request naming no chain is signed for + /// chain 1 rather than refused, so a transaction has to say which chain it is for. It has + /// to read as a failed submission: a config fault means "chain client off" to the caller. + #[tokio::test] + async fn send_transaction_refuses_a_transaction_that_names_another_chain() { + let server = server_answering_with(B256::repeat_byte(0xab)).await; + let client = client_over(vec![server.uri().parse().expect("provider URL")]); + let from = client.inner.signer.address(); + + let mut names_no_chain = ready_to_send_tx(from); + names_no_chain.chain_id = None; + + for tx in [ready_to_send_tx(from).with_chain_id(1), names_no_chain] { + let err = client + .send_transaction(&tx) + .await + .expect_err("a transaction for another chain must not be signed"); + assert!( + matches!(err, ChainClientError::SubmitFailed(_)), + "got {err}" + ); + } + + assert!( + server + .received_requests() + .await + .unwrap_or_default() + .is_empty(), + "no endpoint should have been asked to accept either transaction" + ); + } + + /// A field a caller leaves out is no longer filled in for them, so the send stops rather + /// than putting an incomplete transaction on the wire. What it says has to name the field, + /// because a caller reading only "signing failed" would go looking at the wrong thing. + #[tokio::test] + async fn send_transaction_refuses_a_transaction_missing_a_field() { + let server = server_answering_with(B256::repeat_byte(0xab)).await; + let client = client_over(vec![server.uri().parse().expect("provider URL")]); + + let mut no_gas_limit = ready_to_send_tx(client.inner.signer.address()); + no_gas_limit.gas = None; + + let err = client + .send_transaction(&no_gas_limit) + .await + .expect_err("a transaction with no gas limit must not be sent"); + + let text = err.to_string(); + assert!( + text.contains("gas_limit"), + "the failure should name the missing field, got: {text}" + ); + assert!( + server + .received_requests() + .await + .unwrap_or_default() + .is_empty(), + "no endpoint should have been asked to accept it" + ); + } + + /// The nonce reason has to survive a second provider failing a different way. Reporting + /// only the last provider's reason hid the rejection behind a transport fault, and + /// `sign_and_send` then skipped the resync that would have unstuck the wallet. + #[tokio::test] + async fn send_transaction_surfaces_a_nonce_reason_from_any_provider() { + let rejecting = server_answering_rpc_error(-32000, "nonce too low: next nonce 12").await; + let sick = server_answering_500().await; + let client = client_over(vec![ + rejecting.uri().parse().expect("rejecting provider URL"), + sick.uri().parse().expect("sick provider URL"), + ]); + let tx = ready_to_send_tx(client.inner.signer.address()); + + let err = client + .send_transaction(&tx) + .await + .expect_err("both providers refused, so the send must error"); + + let text = err.to_string(); + assert!( + is_nonce_error(&text), + "the nonce reason must survive the later transport fault, got: {text}" + ); + } + + /// An endpoint saying it is overloaded is asked again before the submission gives up on + /// it, because that complaint often clears on its own and rotating away costs a provider. + /// One retry rather than the usual several, to keep the backoff this waits out short. + #[tokio::test] + async fn send_transaction_retries_a_struggling_endpoint_before_rotating() { + let overloaded = + server_answering_rpc_error(-32005, "project ID request rate exceeded").await; + let healthy = server_answering_with(B256::repeat_byte(0xcd)).await; + let client = client_over_retrying( + vec![ + overloaded.uri().parse().expect("overloaded provider URL"), + healthy.uri().parse().expect("healthy provider URL"), + ], + 1, + ); + let tx = ready_to_send_tx(client.inner.signer.address()); + + client + .send_transaction(&tx) + .await + .expect("send should succeed on the healthy provider"); + + assert_eq!( + overloaded + .received_requests() + .await + .unwrap_or_default() + .len(), + 2, + "the struggling endpoint should get its retry before the rotation" + ); + assert_eq!( + healthy.received_requests().await.unwrap_or_default().len(), + 1, + "the healthy endpoint should answer on the first ask" + ); + } + + /// Every other submission queues behind the one holding the lock, so a submission that + /// never finishes has to be cut off rather than waited out. Paused time so the deadline + /// is reached without the test spending it. + #[tokio::test(start_paused = true)] + async fn a_submission_that_never_finishes_is_given_up_on() { + let err = under_submit_deadline(Duration::from_secs(60), std::future::pending()) + .await + .expect_err("a submission that never finishes must not be waited out"); + + assert!( + matches!(err, ChainClientError::SubmitFailed(_)), + "got {err}" + ); + assert!( + err.to_string().contains("60"), + "the failure should say how long it waited, got: {err}" + ); + } + + /// A submission that finishes inside the deadline is left alone, so the cap only ever + /// catches the case it is there for. + #[tokio::test(start_paused = true)] + async fn a_submission_that_finishes_in_time_is_left_alone() { + let submitted = under_submit_deadline(Duration::from_secs(60), async { + tokio::time::sleep(Duration::from_secs(30)).await; + Ok(SubmittedTx { + hash: B256::repeat_byte(0x77), + nonce: 3, + }) + }) + .await + .expect("a submission inside the deadline should stand"); + + assert_eq!(submitted.hash, B256::repeat_byte(0x77)); + } + + /// The deadline exists to stop one submission starving the queue, not to cut off retries + /// the config asks for, so it is derived from the schedule: each nonce attempt walks the + /// whole ring twice, once reading the chain's nonce and once broadcasting. #[test] - fn test_is_nonce_error() { + fn the_submit_deadline_covers_the_retry_schedule() { + let client = client_over_retrying( + vec![ + "http://one.invalid".parse().expect("first URL"), + "http://two.invalid".parse().expect("second URL"), + ], + 1, + ); + + // Per endpoint: 2 attempts of 5s plus 1s of backoff; 2 endpoints make one walk of + // 22s; 2 walks for each of the 2 nonce attempts. + assert_eq!(client.inner.submit_deadline, Duration::from_secs(88)); + } + + /// A schedule that wants more time than a worker job has is capped rather than obeyed, + /// leaving room after the broadcast for the receipt poll and the nonce-gap fill. + #[test] + fn the_submit_deadline_stays_inside_a_worker_job() { + let providers = (0..20) + .map(|i| { + format!("http://rpc{i}.invalid") + .parse() + .expect("provider URL") + }) + .collect(); + let client = client_over_retrying(providers, 3); + + assert_eq!(client.inner.submit_deadline, PROCESS_JOB_TIMEOUT / 5 * 4); + } + + /// Offering the same bytes twice is what makes a retry safe: an endpoint that took them + /// and then failed to say so recognises them the second time, and reports the broadcast + /// that already happened rather than accepting a second transaction. + #[tokio::test] + async fn a_retry_that_lands_on_bytes_already_held_is_a_success() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(AlreadyHeldResponder { pending_nonce: 6 }) + .mount(&server) + .await; + let client = client_over_retrying(vec![server.uri().parse().expect("provider URL")], 1); + let tx = ready_to_send_tx(client.inner.signer.address()); + + let tx_hash = client + .send_transaction(&tx) + .await + .expect("bytes already held are a successful broadcast"); + + assert_eq!(tx_hash, signed_hash_of(&client, &tx).await); + assert_eq!( + sends_among(&server.received_requests().await.unwrap_or_default()), + 1, + "an accepted broadcast should not be offered again" + ); + } + + /// A chain rejection is the chain's answer, not one endpoint's, so asking the same + /// endpoint again would only collect the same refusal at the cost of the delay. + #[tokio::test] + async fn send_transaction_does_not_repeat_a_chain_rejection() { + let rejecting = server_answering_rpc_error(-32000, "nonce too low: next nonce 12").await; + let spare = server_answering_with(B256::repeat_byte(0xef)).await; + let client = client_over_retrying( + vec![ + rejecting.uri().parse().expect("rejecting provider URL"), + spare.uri().parse().expect("spare provider URL"), + ], + 3, + ); + let tx = ready_to_send_tx(client.inner.signer.address()); + + client + .send_transaction(&tx) + .await + .expect("the spare endpoint accepts, so the send succeeds"); + + assert_eq!( + rejecting + .received_requests() + .await + .unwrap_or_default() + .len(), + 1, + "a chain rejection should not be retried against the same endpoint" + ); + } + + #[test] + fn nonce_rejections_are_told_apart_from_other_refusals() { // Nonce errors assert!(is_nonce_error("nonce too low")); assert!(is_nonce_error("Nonce Too Low for account")); assert!(is_nonce_error("invalid nonce: expected 5, got 3")); assert!(is_nonce_error("replacement transaction underpriced")); + // A backstop, not a live path: a send reports this as the successful broadcast it is. assert!(is_nonce_error("transaction already known")); // Non-nonce errors @@ -731,16 +1432,15 @@ mod tests { } #[test] - fn test_classify_fill_nonce_gap_outcome_success_returns_ok() { + fn a_nonce_gap_fill_that_is_accepted_is_a_success() { let result = classify_fill_nonce_gap_outcome(42, Ok(B256::ZERO)); assert!(result.is_ok()); } #[test] - fn test_classify_fill_nonce_gap_outcome_swallows_nonce_error() { - // Each of these strings flips `is_nonce_error` to true; the gap - // fill must treat them as success because the original tx is - // either still in flight or the slot is already filled. + fn a_nonce_gap_fill_refused_on_the_nonce_is_still_a_success() { + // Each of these strings flips `is_nonce_error` to true; the gap fill must treat + // them as success because they all mean the slot is already spoken for. for msg in [ "nonce too low", "replacement transaction underpriced", @@ -757,7 +1457,7 @@ mod tests { } #[test] - fn test_classify_fill_nonce_gap_outcome_propagates_other_error() { + fn a_nonce_gap_fill_that_fails_for_another_reason_is_reported() { // Errors that don't match `is_nonce_error` mean the noop tx itself // failed for a real reason (RPC down, gas estimation broken, etc.), // so the wallet may stay wedged. Surface to the caller. @@ -769,84 +1469,75 @@ mod tests { ); } - #[tokio::test] - async fn test_nonce_reservation_unique_under_concurrent_callers() { - use std::collections::HashSet; - - let counter = Arc::new(AtomicU64::new(NONCE_UNINITIALIZED)); - let lock = Arc::new(Mutex::new(())); - let chain_pending = Arc::new(AtomicU64::new(100)); - let reserved: Arc>> = Arc::new(Mutex::new(HashSet::new())); - - const TASKS: usize = 50; - let mut handles = Vec::with_capacity(TASKS); - for i in 0..TASKS { - let counter = counter.clone(); - let lock = lock.clone(); - let chain_pending = chain_pending.clone(); - let reserved = reserved.clone(); - - handles.push(tokio::spawn(async move { - let _guard = lock.lock().await; - - // Mirror the entry shape of `next_nonce`/`resync_nonce`: - // - First caller initializes via fetch+CAS. - // - Every seventh subsequent caller hits a "nonce error" - // path: ratchets the counter via fetch_max(chain + 1) - // then reserves via fetch_add. The ratchet never lowers - // the counter, so an in-flight reservation cannot be - // invalidated even if the chain reports a lower pending. - // - Everyone else takes the next slot via fetch_add. - let nonce = if counter.load(Ordering::SeqCst) == NONCE_UNINITIALIZED { - let chain_nonce = chain_pending.load(Ordering::SeqCst); - match counter.compare_exchange( - NONCE_UNINITIALIZED, - chain_nonce + 1, - Ordering::SeqCst, - Ordering::SeqCst, - ) { - Ok(_) => chain_nonce, - Err(_) => counter.fetch_add(1, Ordering::SeqCst), - } - } else if i % 7 == 0 { - let chain_nonce = chain_pending.load(Ordering::SeqCst); - counter.fetch_max(chain_nonce + 1, Ordering::SeqCst); - counter.fetch_add(1, Ordering::SeqCst) - } else { - counter.fetch_add(1, Ordering::SeqCst) - }; - - // Simulate the time spent signing and submitting to the - // mempool while the lock is held; this is the window the - // bug exploited when the lock was missing. - tokio::time::sleep(Duration::from_micros(50)).await; - - // Simulate the chain accepting the tx into pending: the - // pending count cannot decrease, so use fetch_max. - let _ = chain_pending.fetch_update( - Ordering::SeqCst, - Ordering::SeqCst, - |cur| Some(cur.max(nonce + 1)), - ); + /// Answers nonce lookups with a fixed pending count, refuses the first send with a + /// server fault, and accepts every send after it: the shape of a submission failing + /// outright and the job being re-run. + struct FailsFirstSendResponder { + pending_nonce: u64, + sends: AtomicU64, + } - let mut reserved = reserved.lock().await; - assert!( - reserved.insert(nonce), - "nonce {nonce} reissued to a concurrent caller (counter rewound past in-flight reservation)" - ); - })); + impl Respond for FailsFirstSendResponder { + fn respond(&self, request: &Request) -> ResponseTemplate { + let body: serde_json::Value = + serde_json::from_slice(&request.body).expect("JSON-RPC request body"); + match body["method"].as_str().unwrap_or_default() { + "eth_getTransactionCount" => { + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": format!("{:#x}", self.pending_nonce), + })) + } + "eth_sendRawTransaction" if self.sends.fetch_add(1, Ordering::SeqCst) == 0 => { + ResponseTemplate::new(500).set_body_string("Temporary internal error") + } + "eth_sendRawTransaction" => { + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": format!("{:#x}", B256::repeat_byte(0xaa)), + })) + } + other => panic!("unexpected method {other}"), + } } + } - for h in handles { - h.await.expect("worker task panicked"); - } + /// A submission that never got a transaction out must leave its nonce for the next one. + /// Spending it anyway left a slot the chain kept waiting on: every later transaction + /// queued behind the empty slot, and nothing filled it short of a restart. + #[tokio::test] + async fn a_failed_submission_leaves_its_nonce_to_the_next() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(FailsFirstSendResponder { + pending_nonce: 5, + sends: AtomicU64::new(0), + }) + .mount(&server) + .await; + let client = client_over(vec![server.uri().parse().expect("provider URL")]); + let tx = ready_to_send_tx(client.inner.signer.address()); + + client + .sign_and_send(tx.clone(), &[0u8; 16]) + .await + .expect_err("the only endpoint refused, so the submission fails"); - let reserved = reserved.lock().await; + let retry = client + .sign_and_send(tx.clone(), &[0u8; 16]) + .await + .expect("the endpoint accepts the re-run"); assert_eq!( - reserved.len(), - TASKS, - "expected {TASKS} unique nonces, got {}", - reserved.len() + retry.nonce, 5, + "the re-run must take the slot the failure never spent" ); + + let next = client + .sign_and_send(tx, &[0u8; 16]) + .await + .expect("a further submission succeeds"); + assert_eq!(next.nonce, 6, "a successful broadcast spends its slot"); } } diff --git a/bin/dipper-service/src/chain_client/rpc_provider.rs b/bin/dipper-service/src/chain_client/rpc_provider.rs index 5dd7bf84..0d038539 100644 --- a/bin/dipper-service/src/chain_client/rpc_provider.rs +++ b/bin/dipper-service/src/chain_client/rpc_provider.rs @@ -31,9 +31,30 @@ fn extract_chain_client_error(err: TransportError) -> Option { } } -/// Error patterns that indicate a transient failure worth retrying. -/// -/// These patterns are matched case-insensitively against the error message. +/// How an endpoint is named in logs and errors. Hosted RPC endpoints carry their API key in +/// the path or the query, so naming one by host says which endpoint it was without the key. +fn endpoint_name(url: &Url) -> String { + match (url.host_str(), url.port()) { + (Some(host), Some(port)) => format!("{host}:{port}"), + (Some(host), None) => host.to_string(), + (None, _) => "unnamed endpoint".to_string(), + } +} + +/// How a failure reads once the endpoint's URL is taken out of it. A connection that never +/// got a reply is described by the HTTP client, which names the URL it was reaching for, and +/// that is where a hosted endpoint carries the API key that gets us in. +fn describe_failure(url: &Url, error: &TransportError) -> String { + let name = endpoint_name(url); + // A URL is as likely to be printed with its trailing slash as without, so take out both. + error + .to_string() + .replace(url.as_str(), &name) + .replace(url.as_str().trim_end_matches('/'), &name) +} + +/// Error text that indicates a transient failure worth retrying, used only for faults +/// that arrive as prose rather than as a status code or JSON-RPC error object. const RETRYABLE_ERROR_PATTERNS: &[&str] = &[ "connection refused", "connection reset", @@ -42,11 +63,9 @@ const RETRYABLE_ERROR_PATTERNS: &[&str] = &[ "timed out", "rate limit", "too many requests", - "503", - "502", - "429", "service unavailable", "bad gateway", + "temporary internal error", ]; /// Type alias for the provider with default fillers. @@ -58,28 +77,26 @@ pub type HttpProvider = FillProvider< RootProvider, >; -/// RPC provider pool with automatic rotation and retry. -/// -/// Manages multiple RPC provider URLs and automatically rotates between them -/// on failure. Uses exponential backoff for retries within a single provider. +/// Several RPC endpoints treated as one, retried with exponential backoff on the current +/// endpoint and rotated through on failure. #[derive(Debug)] pub struct RpcProviderPool { /// Provider URLs (primary first, then fallbacks) providers: Vec, /// Current provider index (atomic for thread-safety) current_index: AtomicUsize, - /// Request timeout per RPC call - request_timeout: Duration, + /// Shared across every endpoint, which is what lets connections be pooled and reused + /// rather than reopened per call. Built once, so a call can never fail for lack of one. + http: reqwest::Client, /// Maximum retries per provider before rotating max_retries: u32, + /// The longest one walk of the ring can take; computed once here because the pool is + /// what knows the schedule. The submission deadline is derived from it. + worst_case_walk: Duration, } impl RpcProviderPool { - /// Create a new RPC provider pool. - /// - /// # Errors - /// - /// Returns an error if no providers are configured. + /// Create a new RPC provider pool. Errors if no providers are configured. pub fn new( providers: Vec, request_timeout: Duration, @@ -91,47 +108,38 @@ impl RpcProviderPool { )); } + let http = reqwest::Client::builder() + .timeout(request_timeout) + .build() + .map_err(|e| { + ChainClientError::ConfigError(format!("Failed to build HTTP client: {e}")) + })?; + tracing::info!( provider_count = providers.len(), - primary = %providers[0], + primary = %endpoint_name(&providers[0]), "RPC provider pool initialized" ); + // Every endpoint spending the full request timeout on every attempt, plus the + // backoff waited out between attempts, across one visit to each endpoint. + let backoff: Duration = (0..max_retries).map(Self::backoff_delay).sum(); + let worst_case_walk = + (request_timeout * (max_retries + 1) + backoff) * providers.len() as u32; + Ok(Self { providers, current_index: AtomicUsize::new(0), - request_timeout, + http, max_retries, + worst_case_walk, }) } - /// Get the configured request timeout. - pub fn request_timeout(&self) -> Duration { - self.request_timeout - } - - /// Get the current provider URL. - pub fn current_url(&self) -> &Url { - let idx = self.current_index.load(Ordering::Relaxed) % self.providers.len(); - &self.providers[idx] - } - - /// Build a provider for the current URL. - pub fn get_provider(&self) -> Result { - let url = self.current_url(); - - // Build HTTP client with timeout - let client = reqwest::Client::builder() - .timeout(self.request_timeout) - .build() - .map_err(|e| { - ChainClientError::ConfigError(format!("Failed to build HTTP client: {e}")) - })?; - - // Build alloy provider using the connect_reqwest method - let provider = ProviderBuilder::new().connect_reqwest(client, url.clone()); - - Ok(provider) + /// The longest [`execute`](Self::execute) can spend before giving up: every retry the + /// schedule allows, on every endpoint, with the backoff between them all waited out. + pub fn worst_case_walk(&self) -> Duration { + self.worst_case_walk } /// Rotate to the next provider. @@ -139,100 +147,150 @@ impl RpcProviderPool { /// Returns the new provider URL after rotation. pub fn rotate(&self) -> &Url { let old_idx = self.current_index.fetch_add(1, Ordering::Relaxed); - let new_idx = (old_idx + 1) % self.providers.len(); - &self.providers[new_idx] + self.url_at(old_idx + 1) } - /// Execute an RPC operation with retry and provider rotation. - /// - /// The closure receives a provider and should return a `Result`. On retryable - /// errors, the operation is retried with exponential backoff. After exhausting - /// retries, the pool rotates to the next provider and continues. - /// - /// # Arguments - /// - /// * `operation` - Name of the operation (for logging) - /// * `f` - Closure that performs the RPC call - /// - /// # Type Parameters - /// - /// * `F` - Closure type - /// * `Fut` - Future type returned by the closure - /// * `T` - Success type + /// The endpoint an unbounded ring position lands on. + fn url_at(&self, position: usize) -> &Url { + &self.providers[position % self.providers.len()] + } + + /// Run an RPC call, retrying the current endpoint with backoff and then rotating on to + /// the next. `operation` names the call for logging. pub async fn execute(&self, operation: &str, f: F) -> Result where F: Fn(HttpProvider) -> Fut, Fut: Future>, { - let mut last_error: Option = None; + self.execute_with_retries(operation, self.max_retries, f) + .await + } + + async fn execute_with_retries( + &self, + operation: &str, + max_retries: u32, + f: F, + ) -> Result + where + F: Fn(HttpProvider) -> Fut, + Fut: Future>, + { + // What each endpoint said, in the order they were tried. `sign_and_send` matches + // this text to tell a stale nonce from a transport fault, so a rejection from the + // first endpoint has to survive a different kind of failure on the next. + let mut reasons: Vec = Vec::with_capacity(self.providers.len()); + // The first endpoint to name its refusal precisely, such as a contract rejecting the + // call during gas estimation. Callers act on a named refusal and only log a generic + // one, so a later endpoint merely being unreachable must not bury it. + let mut named_refusal: Option = None; let mut providers_tried = 0; + // Walk the ring by local offset from wherever the pool points. Re-reading the shared + // index each time lets a concurrent rotation send this call back to a provider it + // already tried, so it can give up without ever reaching the healthy one. + let start = self.current_index.load(Ordering::Relaxed); + loop { - // Get current provider - let provider = self.get_provider()?; - let current_url = self.current_url().clone(); + let current_url = self.url_at(start + providers_tried).clone(); + let endpoint = endpoint_name(¤t_url); + + // Reused across this endpoint's attempts, so a retry does not pay for a fresh + // TLS handshake on the path that is already running out of time. + let provider = + ProviderBuilder::new().connect_reqwest(self.http.clone(), current_url.clone()); // Retry loop for current provider - for attempt in 0..=self.max_retries { + let mut endpoint_error: Option = None; + for attempt in 0..=max_retries { match f(provider.clone()).await { Ok(result) => return Ok(result), - Err(e) if Self::is_retryable(&e) && attempt < self.max_retries => { + Err(e) if Self::is_retryable(&e) && attempt < max_retries => { let delay = Self::backoff_delay(attempt); tracing::warn!( operation, - provider = %current_url, + provider = %endpoint, attempt = attempt + 1, - max_retries = self.max_retries, + max_retries, delay_ms = delay.as_millis(), - error = %e, + error = %describe_failure(¤t_url, &e), "Retryable RPC error, backing off" ); tokio::time::sleep(delay).await; - last_error = Some(e); + endpoint_error = Some(e); } Err(e) => { - last_error = Some(e); + endpoint_error = Some(e); break; } } } + // Every attempt records why it failed before stopping, so the fallback only + // covers a configuration that somehow allows no attempt at all. + let endpoint_error = endpoint_error + .unwrap_or_else(|| TransportErrorKind::custom_str("no attempt was made")); + let reason = describe_failure(¤t_url, &endpoint_error); + reasons.push(format!("{endpoint}: {reason}")); providers_tried += 1; + // Structured errors reach us boxed in through `TransportErrorKind::custom`, which + // is how gas estimation hands back a contract rejection. Keep the first one. + named_refusal = named_refusal.or_else(|| extract_chain_client_error(endpoint_error)); + // Check if we've tried all providers if providers_tried >= self.providers.len() { - let final_err = - last_error.unwrap_or_else(|| TransportErrorKind::custom_str("unknown error")); - - // Preserve structured ChainClientError instances boxed in via - // TransportErrorKind::custom (e.g. ContractRevert from gas - // estimation). Otherwise fall back to the generic wrap. - if let Some(typed) = extract_chain_client_error(final_err) { - return Err(typed); + if let Some(named) = named_refusal { + return Err(named); } return Err(ChainClientError::RpcError(anyhow::anyhow!( - "All {} RPC providers failed for '{}'", + "All {} RPC providers failed for '{}': {}", self.providers.len(), operation, + reasons.join("; "), ))); } - // Rotate to next provider - let new_url = self.rotate(); + // Move the shared start on, so a call beginning after this one skips the endpoint + // that just failed. Counting failures this way lets concurrent callers wind it + // back round to a failing endpoint, costing them the one wasted first ask. + self.rotate(); + let next_url = self.url_at(start + providers_tried); tracing::warn!( operation, - old_provider = %current_url, - new_provider = %new_url, + old_provider = %endpoint, + new_provider = %endpoint_name(next_url), providers_tried, total_providers = self.providers.len(), + error = %reason, "Rotating RPC provider after failures" ); } } - /// Check if an error is retryable. + /// Whether an error is worth trying again rather than giving up on. Each check can only + /// say yes, so a fault the status and the error code both miss still gets read as text. fn is_retryable(error: &TransportError) -> bool { + // A 5xx is the server failing for its own reasons and 429 is it declining; either + // can succeed on a retry or another endpoint. A status is worth reading before the + // text, because the same digits inside a revert reason mean nothing. + if let RpcError::Transport(kind) = error + && let Some(http) = kind.as_http_error() + && (http.status >= 500 || http.status == 429) + { + return true; + } + + // Some providers answer 200 and report being overloaded in the JSON-RPC error + // instead, each with its own code. Alloy knows those codes, and reports a genuine + // execution error such as a revert as not worth retrying. + if let RpcError::ErrorResp(payload) = error + && payload.is_retry_err() + { + return true; + } + let error_str = error.to_string().to_lowercase(); RETRYABLE_ERROR_PATTERNS .iter() @@ -253,10 +311,326 @@ impl RpcProviderPool { #[cfg(test)] mod tests { + use thegraph_core::alloy::providers::Provider; + use wiremock::{Mock, MockServer, ResponseTemplate, matchers::method}; + use super::*; + /// Refuses every call for a reason no retry would clear, so the pool moves straight on. + async fn server_refusing() -> MockServer { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "error": { "code": -32000, "message": "refused" }, + }))) + .mount(&server) + .await; + server + } + + /// Answers a block-number lookup, which is the shape of the read calls the service makes. + async fn server_answering_block(number: u64) -> MockServer { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "result": format!("{number:#x}"), + }))) + .mount(&server) + .await; + server + } + + /// Every read goes through this path, so a read has to fail over the way a submission does, + /// and a fault worth another go has to earn one before it rotates. One retry keeps the + /// backoff short; real time, since pausing it jumps the clock to the request timeout. + #[tokio::test] + async fn a_read_retries_a_sick_endpoint_then_rotates() { + let sick = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(503).set_body_string("service unavailable")) + .mount(&sick) + .await; + let healthy = server_answering_block(0x2a).await; + + let pool = RpcProviderPool::new( + vec![ + sick.uri().parse().expect("sick URL"), + healthy.uri().parse().expect("healthy URL"), + ], + Duration::from_secs(5), + 1, + ) + .expect("pool"); + + let block = pool + .execute("get_block_number", |provider| async move { + provider.get_block_number().await + }) + .await + .expect("the read should succeed on the second endpoint"); + + assert_eq!(block, 0x2a); + assert_eq!( + sick.received_requests().await.unwrap_or_default().len(), + 2, + "a read should use its retry budget before rotating" + ); + assert_eq!( + healthy.received_requests().await.unwrap_or_default().len(), + 1, + "the healthy endpoint should answer on the first ask" + ); + } + + /// A call walks its own way round the ring, so it reaches every endpoint even while other + /// calls move the shared starting point underneath it. Reading that shared point afresh + /// each time let a call revisit one endpoint and give up without trying another. + #[tokio::test] + async fn a_call_reaches_every_endpoint_even_while_others_rotate() { + let servers = [ + server_refusing().await, + server_refusing().await, + server_refusing().await, + ]; + let pool = RpcProviderPool::new( + servers + .iter() + .map(|s| s.uri().parse().expect("server URL")) + .collect(), + Duration::from_secs(5), + 0, + ) + .expect("pool"); + + let call = pool.execute("probe", |provider| async move { + tokio::task::yield_now().await; + provider.get_block_number().await + }); + let others_rotating = async { + for _ in 0..64 { + pool.rotate(); + tokio::task::yield_now().await; + } + }; + let (result, ()) = tokio::join!(call, others_rotating); + + assert!(result.is_err(), "every endpoint refused, so the call fails"); + for (i, server) in servers.iter().enumerate() { + assert_eq!( + server.received_requests().await.unwrap_or_default().len(), + 1, + "endpoint {i} should have been tried exactly once" + ); + } + } + + /// Hosted RPC endpoints carry their API key in the path, and this failure text reaches the + /// logs and every error built from it, so it has to say which endpoint refused without + /// repeating the key that gets it in. + #[tokio::test] + async fn a_failure_names_the_endpoint_without_its_api_key() { + let refusing = server_refusing().await; + let keyed: Url = format!("{}/v2/super-secret-key", refusing.uri()) + .parse() + .expect("keyed endpoint URL"); + + let pool = RpcProviderPool::new(vec![keyed], Duration::from_secs(5), 0).expect("pool"); + let err = pool + .execute("probe", |provider| async move { + provider.get_block_number().await + }) + .await + .expect_err("the endpoint refused, so the call fails"); + + let text = err.to_string(); + assert!( + !text.contains("super-secret-key"), + "the API key must not appear in the failure: {text}" + ); + assert!( + text.contains("127.0.0.1"), + "the failure should still say which endpoint refused: {text}" + ); + } + + /// A contract refusing the call is the chain's answer, not one endpoint's, and the caller + /// reads it to decide whether to give up rather than retry. A later endpoint being merely + /// unreachable must not turn that into a generic fault that reads as worth another go. + #[tokio::test] + async fn a_named_refusal_outlives_a_later_endpoint_going_dark() { + let pool = RpcProviderPool::new( + vec![ + Url::parse("http://refusing.invalid").expect("refusing endpoint URL"), + Url::parse("http://dark.invalid").expect("dark endpoint URL"), + ], + Duration::from_secs(5), + 0, + ) + .expect("pool"); + + let calls = AtomicUsize::new(0); + let err = pool + .execute("probe", |_provider| { + let refusing = calls.fetch_add(1, Ordering::Relaxed) == 0; + async move { + let outcome: Result<(), TransportError> = Err(if refusing { + TransportErrorKind::custom(ChainClientError::ContractRevert { + selector: [0xde, 0xad, 0xbe, 0xef], + data: Default::default(), + }) + } else { + TransportErrorKind::custom_str("connection refused") + }); + outcome + } + }) + .await + .expect_err("both endpoints failed, so the call fails"); + + assert!( + matches!(err, ChainClientError::ContractRevert { .. }), + "the contract's refusal should have survived, got {err}" + ); + } + + /// An endpoint that answers can only describe its own refusal, so it never repeats the + /// URL. One that never answers is described by the HTTP client instead, which says which + /// URL it was reaching for, and that is where the key sits. Nothing listens on port 1. + #[tokio::test] + async fn a_connection_that_gets_no_answer_hides_the_api_key() { + let keyed: Url = "http://127.0.0.1:1/v2/super-secret-key" + .parse() + .expect("keyed endpoint URL"); + + let pool = RpcProviderPool::new(vec![keyed], Duration::from_secs(5), 0).expect("pool"); + let err = pool + .execute("probe", |provider| async move { + provider.get_block_number().await + }) + .await + .expect_err("nothing is listening, so the call fails"); + + let text = err.to_string(); + assert!( + !text.contains("super-secret-key"), + "the API key must not appear in the failure: {text}" + ); + assert!( + text.contains("127.0.0.1:1"), + "the failure should still say which endpoint was unreachable: {text}" + ); + } + + /// Reporting only the last endpoint's reason hid what the earlier ones said, and a caller + /// reading this text to tell a chain rejection from a transport fault would then miss the + /// rejection and skip the recovery it calls for. + #[tokio::test] + async fn a_failure_names_every_endpoint_that_was_tried() { + let servers = [server_refusing().await, server_refusing().await]; + let pool = RpcProviderPool::new( + servers + .iter() + .map(|s| s.uri().parse().expect("server URL")) + .collect(), + Duration::from_secs(5), + 0, + ) + .expect("pool"); + + let err = pool + .execute("probe", |provider| async move { + provider.get_block_number().await + }) + .await + .expect_err("every endpoint refused, so the call fails"); + + let text = err.to_string(); + for server in &servers { + let named = endpoint_name(&server.uri().parse().expect("server URL")); + assert!(text.contains(&named), "{named} is missing from: {text}"); + } + assert_eq!( + text.matches("refused").count(), + servers.len(), + "every endpoint's own reason should appear: {text}" + ); + } + + /// 500 is the status a provider answered on 2026-07-29 while an accepted agreement went + /// unfunded. Reading it here is what earns a retry on that provider before rotating; the + /// rotation itself is unconditional, so this decides attempts rather than failover. + #[test] + fn server_faults_are_retryable_by_status() { + for status in [500, 502, 503, 504, 429] { + let err = TransportErrorKind::http_error(status, "provider fault".to_string()); + assert!( + RpcProviderPool::is_retryable(&err), + "HTTP {status} should be retryable" + ); + } + } + + /// A 4xx other than 429 means the request itself is wrong, so resending it unchanged + /// to the same or another provider cannot succeed. #[test] - fn test_backoff_delay_calculation() { + fn client_faults_are_not_retryable_by_status() { + for status in [400, 401, 403, 404] { + let err = TransportErrorKind::http_error(status, "bad request".to_string()); + assert!( + !RpcProviderPool::is_retryable(&err), + "HTTP {status} should not be retryable" + ); + } + } + + /// Some providers and the proxies in front of them report throttling under a 4xx rather + /// than a 429. The status alone reads as "your request is wrong", so the wording is what + /// tells this apart from a request that will be refused however often it is sent. + #[test] + fn throttling_described_in_a_client_fault_body_is_retryable() { + let err = TransportErrorKind::http_error(403, "rate limit exceeded".to_string()); + assert!( + RpcProviderPool::is_retryable(&err), + "a 403 that explains it is throttling should be retryable" + ); + } + + /// Alloy recognises the error codes providers use for throttling, but not every + /// transient fault has one. A gateway that describes a timeout in an otherwise ordinary + /// response is still worth another go, and only the wording says so. + #[test] + fn transient_faults_described_in_a_json_rpc_error_are_retryable() { + let payload = serde_json::from_str(r#"{"code":-32603,"message":"connection reset"}"#) + .expect("JSON-RPC error payload"); + let err: TransportError = RpcError::ErrorResp(payload); + assert!( + RpcProviderPool::is_retryable(&err), + "a reset described in the error body should be retryable" + ); + } + + /// The provider that stranded an accepted agreement on 2026-07-29 answered `code 19 + /// Temporary internal error. Please retry`, a code alloy does not know. Sent under a 200 + /// the status says nothing either, so the wording is the only thing left to read. + #[test] + fn a_temporary_internal_error_without_a_status_is_retryable() { + let payload = serde_json::from_str( + r#"{"code":19,"message":"Temporary internal error. Please retry"}"#, + ) + .expect("JSON-RPC error payload"); + let err: TransportError = RpcError::ErrorResp(payload); + assert!( + RpcProviderPool::is_retryable(&err), + "the fault behind the outage should be retryable however it is reported" + ); + } + + #[test] + fn each_retry_waits_twice_as_long_up_to_a_ceiling() { // 1s, 2s, 4s, 8s, 16s, 32s->30s assert_eq!(RpcProviderPool::backoff_delay(0), Duration::from_secs(1)); assert_eq!(RpcProviderPool::backoff_delay(1), Duration::from_secs(2)); @@ -267,29 +641,28 @@ mod tests { assert_eq!(RpcProviderPool::backoff_delay(10), Duration::from_secs(30)); // stays capped } + /// Faults that arrive with no status and no error code, only a description, which is + /// what a connection that never got a reply looks like. #[test] - fn test_retryable_error_detection() { - // Test retryable patterns + fn faults_described_only_in_words_are_read_from_the_text() { let retryable_errors = [ "connection refused by remote host", "Connection Reset by peer", "request TIMEOUT exceeded", - "HTTP 429 Too Many Requests", - "503 Service Unavailable", - "502 Bad Gateway", + "Service Unavailable", + "Bad Gateway", "rate limit exceeded", + "too many requests, slow down", ]; for err_str in retryable_errors { - // Create a mock transport error by using the error message - // In practice, TransportError wraps various error types - let is_match = RETRYABLE_ERROR_PATTERNS - .iter() - .any(|p| err_str.to_lowercase().contains(p)); - assert!(is_match, "Expected '{}' to be retryable", err_str); + let err = TransportErrorKind::custom_str(err_str); + assert!( + RpcProviderPool::is_retryable(&err), + "expected '{err_str}' to be retryable" + ); } - // Test non-retryable patterns let non_retryable_errors = [ "nonce too low", "insufficient funds", @@ -298,15 +671,16 @@ mod tests { ]; for err_str in non_retryable_errors { - let is_match = RETRYABLE_ERROR_PATTERNS - .iter() - .any(|p| err_str.to_lowercase().contains(p)); - assert!(!is_match, "Expected '{}' to NOT be retryable", err_str); + let err = TransportErrorKind::custom_str(err_str); + assert!( + !RpcProviderPool::is_retryable(&err), + "expected '{err_str}' to not be retryable" + ); } } #[test] - fn test_provider_pool_requires_at_least_one_provider() { + fn a_pool_with_no_endpoints_is_refused() { let result = RpcProviderPool::new(vec![], Duration::from_secs(30), 3); assert!(result.is_err()); @@ -319,29 +693,34 @@ mod tests { } } + /// The submission deadline is derived from this figure, so it has to count every + /// attempt the schedule allows and the backoff waited out between them. #[test] - fn test_provider_pool_rotation() { + fn the_worst_case_walk_counts_every_attempt_and_backoff() { let providers = vec![ Url::parse("https://rpc1.example.com").unwrap(), Url::parse("https://rpc2.example.com").unwrap(), - Url::parse("https://rpc3.example.com").unwrap(), ]; + let pool = RpcProviderPool::new(providers, Duration::from_secs(10), 3).unwrap(); - let pool = RpcProviderPool::new(providers.clone(), Duration::from_secs(30), 3).unwrap(); + // Per endpoint: 4 attempts of 10s plus 1+2+4s of backoff, across 2 endpoints. + assert_eq!(pool.worst_case_walk(), Duration::from_secs(94)); + } - // Initially at index 0 - assert_eq!(pool.current_url().as_str(), "https://rpc1.example.com/"); + #[test] + fn rotating_walks_the_endpoints_and_wraps_round() { + let providers = vec![ + Url::parse("https://rpc1.example.com").unwrap(), + Url::parse("https://rpc2.example.com").unwrap(), + Url::parse("https://rpc3.example.com").unwrap(), + ]; - // Rotate to index 1 - pool.rotate(); - assert_eq!(pool.current_url().as_str(), "https://rpc2.example.com/"); + let pool = RpcProviderPool::new(providers.clone(), Duration::from_secs(30), 3).unwrap(); - // Rotate to index 2 - pool.rotate(); - assert_eq!(pool.current_url().as_str(), "https://rpc3.example.com/"); + assert_eq!(pool.rotate().as_str(), "https://rpc2.example.com/"); + assert_eq!(pool.rotate().as_str(), "https://rpc3.example.com/"); - // Rotate wraps back to index 0 - pool.rotate(); - assert_eq!(pool.current_url().as_str(), "https://rpc1.example.com/"); + // Wraps back round rather than running off the end. + assert_eq!(pool.rotate().as_str(), "https://rpc1.example.com/"); } } diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 9be975c8..ccee80c5 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -663,7 +663,8 @@ pub struct ChainClientConfig { /// rotating to the next on persistent failures. pub providers: Vec, - /// Request timeout per RPC call in seconds (default: 30s) + /// Request timeout per RPC call in seconds (default: 10s). Every call here is a small one, + /// so this is headroom rather than a working limit, and it multiplies with `max_retries`. #[serde(default = "default_chain_client_request_timeout")] #[serde_as(as = "serde_with::DurationSeconds")] pub request_timeout: Duration, @@ -717,7 +718,7 @@ fn default_chain_client_enabled() -> bool { } fn default_chain_client_request_timeout() -> Duration { - Duration::from_secs(30) + Duration::from_secs(10) } fn default_domain_refresh_interval() -> Duration { diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml index 5d48fc60..6b857d13 100644 --- a/k8s/configmap-example.yaml +++ b/k8s/configmap-example.yaml @@ -94,7 +94,7 @@ data: "https://arb1.arbitrum.io/rpc", "https://arbitrum-one.publicnode.com" ], - "request_timeout": 30, + "request_timeout": 10, "max_retries": 3, "gas_price_multiplier": 1.2, "max_gas_price_gwei": 100,