diff --git a/bin/debug-trace-server/src/data_provider.rs b/bin/debug-trace-server/src/data_provider.rs index 7eeb6057..e552e646 100644 --- a/bin/debug-trace-server/src/data_provider.rs +++ b/bin/debug-trace-server/src/data_provider.rs @@ -767,7 +767,7 @@ async fn resolve_contracts_inner( "Cache miss — fetching contracts from RPC" ); - // Per-attempt `eth_getCodeByHash` metrics land on `UpstreamMetrics` via the + // Per-attempt `eth_getCodeByHash` metrics land on the upstream attempt metrics via the // `TraceRpcMetrics` adapter inside `round_robin_with_backoff`. let fetched = rpc_client.get_codes_with_deadline(&missing, true, Some(deadline)).await?; diff --git a/bin/debug-trace-server/src/metrics.rs b/bin/debug-trace-server/src/metrics.rs index f69ae0df..a80ca845 100644 --- a/bin/debug-trace-server/src/metrics.rs +++ b/bin/debug-trace-server/src/metrics.rs @@ -8,7 +8,7 @@ use std::net::SocketAddr; use eyre::Result; -use metrics::{Counter, Gauge, Histogram}; +use metrics::{Counter, Gauge, Histogram, counter, histogram}; use metrics_derive::Metrics; use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; pub use stateless_common::{ @@ -246,32 +246,33 @@ impl SingleFlightMetrics { } } -/// Upstream RPC metrics with method label. -#[derive(Clone, Metrics)] -#[metrics(scope = "debug_trace")] -pub struct UpstreamMetrics { - /// Total upstream RPC requests - upstream_requests_total: Counter, - /// Total upstream RPC errors - upstream_errors_total: Counter, - /// Duration of upstream RPC requests in seconds - upstream_duration_seconds: Histogram, -} +/// Total upstream RPC attempts, labeled `(method, provider, outcome)`. +const UPSTREAM_REQUESTS_TOTAL: &str = "debug_trace_upstream_requests_total"; +/// Per-attempt upstream RPC duration in seconds, labeled `(method, provider, outcome)`. +const UPSTREAM_DURATION_SECONDS: &str = "debug_trace_upstream_duration_seconds"; +/// Logical-call deadline-exceeded ("request timed out") count, labeled `(method)`. +const UPSTREAM_DEADLINE_EXCEEDED_TOTAL: &str = "debug_trace_upstream_deadline_exceeded_total"; -impl UpstreamMetrics { - /// Creates metrics for a specific upstream RPC method. - pub fn new_for_method(method: &'static str) -> Self { - Self::new_with_labels(&[("method", method)]) - } +/// Records one upstream RPC attempt against `provider` with its `outcome` and duration. +/// +/// The `(method, provider, outcome)` labeling is what lets operators attribute latency and +/// separate a returned error from a stall-timeout, per endpoint — the collapsed method-only +/// `success` boolean could do neither. `provider` is a bounded, credential-free endpoint host. +fn record_upstream_attempt( + method: &'static str, + provider: &str, + outcome: &'static str, + duration_secs: f64, +) { + counter!(UPSTREAM_REQUESTS_TOTAL, "method" => method, "provider" => provider.to_owned(), "outcome" => outcome) + .increment(1); + histogram!(UPSTREAM_DURATION_SECONDS, "method" => method, "provider" => provider.to_owned(), "outcome" => outcome) + .record(duration_secs); +} - /// Records an upstream RPC request. - pub fn record_request(&self, success: bool, duration_secs: f64) { - self.upstream_requests_total.increment(1); - if !success { - self.upstream_errors_total.increment(1); - } - self.upstream_duration_seconds.record(duration_secs); - } +/// Records one logical upstream call giving up because its overall deadline elapsed. +fn record_upstream_deadline_exceeded(method: &'static str) { + counter!(UPSTREAM_DEADLINE_EXCEEDED_TOTAL, "method" => method).increment(1); } /// Witness fetch metrics by source. @@ -426,11 +427,15 @@ fn pre_register_all_metrics() { let _ = SingleFlightMetrics::new_for_type("coalesced"); let _ = SingleFlightMetrics::new_for_type("bypassed"); - // Data Fetch Layer: upstream RPC - let _ = UpstreamMetrics::new_for_method("eth_getHeaderByHash"); - let _ = UpstreamMetrics::new_for_method("eth_getBlockByHash"); - let _ = UpstreamMetrics::new_for_method("mega_getWitness"); - let _ = UpstreamMetrics::new_for_method("eth_getCodeByHash"); + // Data Fetch Layer: upstream RPC. The attempt series (`upstream_requests_total` / + // `upstream_duration_seconds`) carry a dynamic per-endpoint `provider` label, so they first + // appear on the initial attempt; only the method-keyed deadline-exceeded counter is + // pre-registrable here. + for method in + ["eth_getHeaderByHash", "eth_getBlockByHash", "mega_getWitness", "eth_getCodeByHash"] + { + counter!(UPSTREAM_DEADLINE_EXCEEDED_TOTAL, "method" => method).increment(0); + } // Witness Layer let _ = WitnessSourceMetrics::new_for_source("witness_generator"); @@ -495,7 +500,7 @@ pub fn record_rpc_error(method: &str) { RpcMethodMetrics::new_for_method(method).record_error(); } -/// Maps an [`RpcMethod`] to the `method` label used by [`UpstreamMetrics`]. +/// Maps an [`RpcMethod`] to the `method` label used by the upstream attempt metrics. /// /// The existing dashboard labels (`eth_getHeaderByHash`, `eth_getBlockByHash`, etc.) /// encode the trace-server-specific call flavor. Since [`RpcMethod`] is coarser @@ -513,38 +518,42 @@ fn upstream_label_for(method: stateless_common::metrics::RpcMethod) -> &'static } /// [`stateless_common::RpcMetrics`] adapter that forwards every per-attempt RPC -/// event to [`UpstreamMetrics`], keyed by method label. +/// event to the upstream metrics, keyed by `(method, provider, outcome)`. /// /// Wired via [`stateless_common::RpcClientConfig::with_metrics`] so that the -/// per-attempt duration and success/failure counters recorded inside -/// `round_robin_with_backoff` land on the same dashboards the trace server has -/// always exposed — even though `get_block` / `get_header` / `get_witness` no -/// longer surface their per-call success status at the caller level. +/// per-attempt duration and per-endpoint outcome counters recorded inside +/// `round_robin_with_backoff` reach Prometheus — including the primary-vs-failover +/// split for `mega_getWitness` and the reason (error vs timeout) for each failure. #[derive(Default)] pub struct TraceRpcMetrics; impl stateless_common::RpcMetrics for TraceRpcMetrics { - fn on_rpc_complete( + fn on_rpc_attempt( &self, method: stateless_common::metrics::RpcMethod, - success: bool, - duration_secs: Option, + provider: &str, + outcome: stateless_common::metrics::RpcAttemptOutcome, + duration_secs: f64, ) { - let label = upstream_label_for(method); - // `new_for_method` is just a label binding — no allocation beyond what - // the metrics crate deduplicates internally. - UpstreamMetrics::new_for_method(label) - .record_request(success, duration_secs.unwrap_or(0.0)); + record_upstream_attempt( + upstream_label_for(method), + provider, + outcome.as_str(), + duration_secs, + ); } - fn on_rpc_retry(&self, _method: stateless_common::metrics::RpcMethod) { - // `on_rpc_complete(_, false, _)` fires alongside `on_rpc_retry` in the - // retry loop, so the error/request counters already capture retries. + fn on_rpc_deadline_exceeded( + &self, + method: stateless_common::metrics::RpcMethod, + _elapsed_secs: f64, + ) { + record_upstream_deadline_exceeded(upstream_label_for(method)); } fn on_witness_fetch(&self, _breakdown: stateless_common::witness_size::WitnessSizeBreakdown) { - // Witness size/source metrics are recorded by `fetch_witness_with_timeout` - // at the outer timeout boundary (distinct semantics from per-attempt). + // Witness size/source metrics are recorded by `fetch_witness` at the outer + // boundary (distinct semantics from per-attempt). } } diff --git a/bin/stateless-validator/src/metrics.rs b/bin/stateless-validator/src/metrics.rs index 40762768..d3b141bb 100644 --- a/bin/stateless-validator/src/metrics.rs +++ b/bin/stateless-validator/src/metrics.rs @@ -13,7 +13,7 @@ use metrics::{counter, describe_counter, describe_gauge, describe_histogram, gau use metrics_exporter_prometheus::{Matcher, PrometheusBuilder}; pub use stateless_common::{ DEFAULT_METRICS_PORT, WitnessSizeBreakdown, - metrics::{BYTE_BUCKETS, REORG_DEPTH_BUCKETS, RpcMethod, RpcMetrics}, + metrics::{BYTE_BUCKETS, REORG_DEPTH_BUCKETS, RpcAttemptOutcome, RpcMethod, RpcMetrics}, }; use tracing::info; @@ -27,14 +27,24 @@ use crate::r2_witness::R2WitnessError; pub struct ValidatorMetrics; impl RpcMetrics for ValidatorMetrics { - fn on_rpc_complete(&self, method: RpcMethod, success: bool, duration_secs: Option) { - on_rpc_complete(method, success, duration_secs); + fn on_rpc_attempt( + &self, + method: RpcMethod, + _provider: &str, + outcome: RpcAttemptOutcome, + duration_secs: f64, + ) { + on_rpc_attempt(method, outcome, duration_secs); } fn on_rpc_retry(&self, method: RpcMethod) { on_rpc_retry(method); } + fn on_rpc_deadline_exceeded(&self, method: RpcMethod, _elapsed_secs: f64) { + on_rpc_deadline_exceeded(method); + } + fn on_witness_fetch(&self, breakdown: WitnessSizeBreakdown) { on_witness_fetch(breakdown); } @@ -73,6 +83,7 @@ pub mod names { metric!(RPC_REQUESTS_TOTAL, "rpc_requests_total"); metric!(RPC_ERRORS_TOTAL, "rpc_errors_total"); metric!(RPC_RETRY_ATTEMPTS_TOTAL, "rpc_retry_attempts_total"); + metric!(RPC_DEADLINE_EXCEEDED_TOTAL, "rpc_deadline_exceeded_total"); metric!(BLOCK_FETCH_TIME, "block_fetch_time_seconds"); metric!(CODE_FETCH_TIME, "code_fetch_time_seconds"); metric!(WITNESS_FETCH_RPC_TIME, "witness_fetch_rpc_time_seconds"); @@ -163,6 +174,10 @@ fn register_metric_descriptions() { names::RPC_RETRY_ATTEMPTS_TOTAL, "RPC transient retry attempts (before final outcome)" ); + describe_counter!( + names::RPC_DEADLINE_EXCEEDED_TOTAL, + "Logical RPC calls that gave up because their overall deadline elapsed" + ); describe_histogram!(names::BLOCK_FETCH_TIME, "Block fetch time (s)"); describe_histogram!(names::CODE_FETCH_TIME, "Code fetch time (s)"); describe_histogram!(names::WITNESS_FETCH_RPC_TIME, "Witness RPC fetch time (s)"); @@ -209,6 +224,7 @@ fn init_rpc_method_counters() { counter!(names::RPC_REQUESTS_TOTAL, "method" => method_str).increment(0); counter!(names::RPC_ERRORS_TOTAL, "method" => method_str).increment(0); counter!(names::RPC_RETRY_ATTEMPTS_TOTAL, "method" => method_str).increment(0); + counter!(names::RPC_DEADLINE_EXCEEDED_TOTAL, "method" => method_str).increment(0); } } @@ -286,29 +302,32 @@ pub fn on_rpc_retry(method: RpcMethod) { counter!(names::RPC_RETRY_ATTEMPTS_TOTAL, "method" => method.as_str()).increment(1); } -pub fn on_rpc_complete(method: RpcMethod, success: bool, duration_secs: Option) { +pub fn on_rpc_attempt(method: RpcMethod, outcome: RpcAttemptOutcome, duration_secs: f64) { let method_str = method.as_str(); counter!(names::RPC_REQUESTS_TOTAL, "method" => method_str).increment(1); - if !success { + if !outcome.is_success() { counter!(names::RPC_ERRORS_TOTAL, "method" => method_str).increment(1); } - if let Some(duration) = duration_secs { - match method { - RpcMethod::EthGetCodeByHash => { - histogram!(names::CODE_FETCH_TIME).record(duration); - } - RpcMethod::EthGetBlock => { - histogram!(names::BLOCK_FETCH_TIME).record(duration); - } - RpcMethod::MegaGetBlockWitness => { - histogram!(names::WITNESS_FETCH_RPC_TIME).record(duration); - } - _ => {} + match method { + RpcMethod::EthGetCodeByHash => { + histogram!(names::CODE_FETCH_TIME).record(duration_secs); } + RpcMethod::EthGetBlock => { + histogram!(names::BLOCK_FETCH_TIME).record(duration_secs); + } + RpcMethod::MegaGetBlockWitness => { + histogram!(names::WITNESS_FETCH_RPC_TIME).record(duration_secs); + } + _ => {} } } +/// Record a logical RPC call giving up because its overall deadline elapsed. +pub fn on_rpc_deadline_exceeded(method: RpcMethod) { + counter!(names::RPC_DEADLINE_EXCEEDED_TOTAL, "method" => method.as_str()).increment(1); +} + pub fn on_contract_cache_read(hits: u64, misses: u64) { if hits > 0 { counter!(names::CONTRACT_CACHE_HITS).increment(hits); diff --git a/crates/stateless-common/src/metrics.rs b/crates/stateless-common/src/metrics.rs index 72d16ab3..278e5c30 100644 --- a/crates/stateless-common/src/metrics.rs +++ b/crates/stateless-common/src/metrics.rs @@ -50,18 +50,74 @@ impl RpcMethod { } } +/// Outcome of a single provider attempt inside the retry loop. +/// +/// The retry loop reports one of these per provider round trip, so metrics can +/// attribute latency and failure *reason* per endpoint instead of collapsing +/// every non-success into one opaque error count. `Error` and `Timeout` are +/// both retriable failures; the split lets operators tell a provider that +/// answered with an error from one that stalled and had to be timed out. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RpcAttemptOutcome { + /// The provider returned a usable response. + Success, + /// The provider returned an error, or its response failed to decode. + Error, + /// The attempt hit the per-attempt timeout: the provider accepted the + /// request but did not answer within the budget (a stall). + Timeout, +} + +impl RpcAttemptOutcome { + /// Stable metric-label string for this outcome. + pub fn as_str(&self) -> &'static str { + match self { + RpcAttemptOutcome::Success => "success", + RpcAttemptOutcome::Error => "error", + RpcAttemptOutcome::Timeout => "timeout", + } + } + + /// Whether this attempt succeeded. + pub fn is_success(&self) -> bool { + matches!(self, RpcAttemptOutcome::Success) + } +} + /// Trait for RPC metrics callbacks. /// -/// Implement this trait to receive metrics events from the RPC client. +/// Implement this trait to receive metrics events from the RPC client. All +/// callbacks fire from inside the retry loop, so `on_rpc_attempt` is called +/// once per provider round trip (not once per logical call). pub trait RpcMetrics: Send + Sync { - /// Called when an RPC request completes (final outcome — success or permanent failure). - fn on_rpc_complete(&self, method: RpcMethod, success: bool, duration_secs: Option); + /// Called once per provider attempt, tagged with the endpoint `provider` + /// label and the attempt [`RpcAttemptOutcome`]. + /// + /// Every round trip the retry loop makes fires exactly one of these, so + /// both per-endpoint latency (via `duration_secs`) and per-reason failure + /// counts (via `outcome`) are derivable. `provider` is a bounded, + /// credential-free endpoint label (see the RPC client's `endpoint_label`). + fn on_rpc_attempt( + &self, + method: RpcMethod, + provider: &str, + outcome: RpcAttemptOutcome, + duration_secs: f64, + ); /// Called on each transient failure that will be retried (not on the final outcome). /// /// Default: no-op. Implement to track retry volume separately from logical errors. fn on_rpc_retry(&self, _method: RpcMethod) {} + /// Called when a logical call gives up because its overall deadline elapsed. + /// + /// Distinct from a per-attempt [`RpcAttemptOutcome::Timeout`]: this counts + /// the *logical call* exhausting its whole budget (e.g. a witness fetch's + /// 3s deadline), which is the operator-facing "request timed out" signal. + /// Fires at most once per logical call. Default: no-op. + fn on_rpc_deadline_exceeded(&self, _method: RpcMethod, _elapsed_secs: f64) {} + /// Called when witness data is successfully fetched. fn on_witness_fetch(&self, breakdown: WitnessSizeBreakdown); } @@ -78,4 +134,14 @@ mod tests { assert_eq!(RpcMethod::MegaGetBlockWitness.as_str(), "mega_getBlockWitness"); assert_eq!(RpcMethod::MegaSetValidatedBlocks.as_str(), "mega_setValidatedBlocks"); } + + #[test] + fn test_rpc_attempt_outcome() { + assert_eq!(RpcAttemptOutcome::Success.as_str(), "success"); + assert_eq!(RpcAttemptOutcome::Error.as_str(), "error"); + assert_eq!(RpcAttemptOutcome::Timeout.as_str(), "timeout"); + assert!(RpcAttemptOutcome::Success.is_success()); + assert!(!RpcAttemptOutcome::Error.is_success()); + assert!(!RpcAttemptOutcome::Timeout.is_success()); + } } diff --git a/crates/stateless-common/src/rpc_client.rs b/crates/stateless-common/src/rpc_client.rs index 90124e45..31d264ba 100644 --- a/crates/stateless-common/src/rpc_client.rs +++ b/crates/stateless-common/src/rpc_client.rs @@ -47,10 +47,10 @@ use salt::SaltWitness; use serde::{Deserialize, Serialize}; use stateless_core::{LightWitness, withdrawals::MptWitness}; use tokio::sync::Semaphore; -use tracing::{trace, warn}; +use tracing::{instrument, trace, warn}; use crate::{ - metrics::{RpcMethod, RpcMetrics}, + metrics::{RpcAttemptOutcome, RpcMethod, RpcMetrics}, witness_encoding::{decode_witness_response, decode_witness_response_light}, witness_size::WitnessSizeBreakdown, }; @@ -227,14 +227,20 @@ pub struct RpcClient { /// Ordered list of data providers. Data methods use round-robin load balancing: each call /// picks a starting provider via an atomic counter and cycles forward on failure. data_providers: Vec>, + /// Metric/log labels for `data_providers`, parallel by index (see [`endpoint_label`]). + data_provider_labels: Vec>, /// Round-robin counter for selecting the starting data provider on each call. /// Shared across clones so load balancing is global per logical client. data_rr_counter: Arc, /// Ordered list of witness providers. `get_witness` always starts from index 0 (primary); /// later entries are failover-only. witness_providers: Vec, + /// Metric/log labels for `witness_providers`, parallel by index (see [`endpoint_label`]). + witness_provider_labels: Vec>, /// Optional dedicated provider for reporting validated blocks. report_provider: Option, + /// Metric/log label for `report_provider` (see [`endpoint_label`]); `None` when unconfigured. + report_provider_label: Option>, /// Configuration controlling verification, retry, and concurrency behavior. config: RpcClientConfig, /// Semaphore capping concurrent in-flight data-endpoint requests @@ -292,12 +298,21 @@ impl RpcClient { }) .collect::>>()?; + // URLs above are already validated by `connect_http`, so labels only ever derive from + // well-formed endpoints; the parallel-by-index vectors feed the retry loop's per-endpoint + // metrics and logs. + let data_provider_labels = + data_apis.iter().enumerate().map(|(i, url)| endpoint_label(url, i)).collect(); + let witness_provider_labels = + witness_apis.iter().enumerate().map(|(i, url)| endpoint_label(url, i)).collect(); + let report_provider = report_api .map(|url| -> Result { Ok(ProviderBuilder::default() .connect_http(url.parse().context("Failed to parse report API URL")?)) }) .transpose()?; + let report_provider_label = report_api.map(|url| endpoint_label(url, 0)); // `.max(1)` guards against `--data-max-concurrent-requests 0` (or the witness // equivalent) silently wedging every RPC call — `Semaphore::new(0)` blocks @@ -311,9 +326,12 @@ impl RpcClient { Ok(Self { data_providers, + data_provider_labels, data_rr_counter: Arc::new(AtomicUsize::new(0)), witness_providers, + witness_provider_labels, report_provider, + report_provider_label, config, data_concurrency, witness_concurrency, @@ -372,6 +390,7 @@ impl RpcClient { if n > 1 { self.data_rr_counter.fetch_add(1, Ordering::Relaxed) % n } else { 0 }; round_robin_with_backoff( &self.data_providers, + &self.data_provider_labels, &self.data_concurrency, &self.config.rpc_retry, self.config.per_attempt_timeout, @@ -379,7 +398,7 @@ impl RpcClient { method, self.config.metrics.as_ref(), deadline, - |provider| f(provider.clone()), + |provider, _provider_label| f(provider.clone()), ) .await } @@ -616,6 +635,10 @@ impl RpcClient { /// provider 0 so the primary takes all traffic while healthy; backups are touched only /// while it is failing), each attempt one RPC round trip followed by the caller-chosen /// `decode` (see [`fetch_witness_with`]). + // A `warn`-level span (not the usual `info`) so it stays enabled at the default `warn` log + // filter: the generic retry loop's per-attempt failure logs then inherit `block_number`, + // which they cannot see otherwise, so an endpoint stall/error is traceable to its block. + #[instrument(level = "warn", skip_all, fields(block_number = number, block_hash = %hash))] async fn witness_round_robin( &self, number: u64, @@ -626,6 +649,7 @@ impl RpcClient { ) -> std::result::Result { round_robin_with_backoff( &self.witness_providers, + &self.witness_provider_labels, &self.witness_concurrency, &self.config.rpc_retry, self.config.per_attempt_timeout, @@ -633,9 +657,10 @@ impl RpcClient { RpcMethod::MegaGetBlockWitness, self.config.metrics.as_ref(), deadline, - |provider| { + |provider, provider_label| { Box::pin(async move { - fetch_witness_with(&provider, number, hash, decode, trace_msg).await + fetch_witness_with(&provider, &provider_label, number, hash, decode, trace_msg) + .await }) }, ) @@ -651,7 +676,9 @@ impl RpcClient { let provider = self.report_provider.as_ref().ok_or_else(|| eyre!("Report provider not configured"))?; let attempt_start = Instant::now(); - let result = match tokio::time::timeout( + // Single-attempt report call (no round-robin), so it classifies its own outcome the same + // way the retry loop does: returned error vs per-attempt stall. + let (result, outcome) = match tokio::time::timeout( self.config.per_attempt_timeout, provider.client().request::<_, SetValidatedBlocksResponse>( "mega_setValidatedBlocks", @@ -660,10 +687,10 @@ impl RpcClient { ) .await { - Ok(Ok(response)) => Ok(response), + Ok(Ok(response)) => (Ok(response), RpcAttemptOutcome::Success), Ok(Err(e)) => { trace!(error = %e, "mega_setValidatedBlocks failed"); - Err(eyre!("Failed to set validated blocks: {e}")) + (Err(eyre!("Failed to set validated blocks: {e}")), RpcAttemptOutcome::Error) } Err(_) => { warn!( @@ -671,17 +698,22 @@ impl RpcClient { attempt_timeout_ms = self.config.per_attempt_timeout.as_millis() as u64, "Report RPC stalled past per-attempt timeout", ); - Err(eyre!( - "mega_setValidatedBlocks timed out after {:?} (per_attempt_timeout)", - self.config.per_attempt_timeout - )) + ( + Err(eyre!( + "mega_setValidatedBlocks timed out after {:?} (per_attempt_timeout)", + self.config.per_attempt_timeout + )), + RpcAttemptOutcome::Timeout, + ) } }; if let Some(ref metrics) = self.config.metrics { - metrics.on_rpc_complete( + let provider_label = self.report_provider_label.as_deref().unwrap_or("report"); + metrics.on_rpc_attempt( RpcMethod::MegaSetValidatedBlocks, - result.is_ok(), - Some(attempt_start.elapsed().as_secs_f64()), + provider_label, + outcome, + attempt_start.elapsed().as_secs_f64(), ); } result @@ -811,6 +843,27 @@ macro_rules! log_at { }; } +/// Derives a bounded, credential-free, per-endpoint metric/log label from an endpoint URL and its +/// index in the configured list. +/// +/// Format is `{idx}:{host}`: the host (with port) stays human-readable, while the index keeps two +/// endpoints that share an authority (same SaaS host, different path or credentials) distinct — so +/// the per-endpoint metric split holds instead of aggregating them — and also encodes primary (0) +/// vs failover (1+). Any `user:pass@` userinfo is stripped so endpoint credentials never leak into +/// labels or logs; cardinality is bounded because the endpoint list is operator-configured. Falls +/// back to `provider_{idx}` when no host can be extracted (unreachable for the validated URLs the +/// constructor accepts, but kept total). +fn endpoint_label(url: &str, idx: usize) -> Arc { + let after_scheme = url.split_once("://").map_or(url, |(_, rest)| rest); + let authority = after_scheme.split(['/', '?', '#']).next().unwrap_or(""); + let host = authority.rsplit_once('@').map_or(authority, |(_, host)| host); + if host.is_empty() { + Arc::from(format!("provider_{idx}")) + } else { + Arc::from(format!("{idx}:{host}")) + } +} + /// Runs a round-robin RPC call with round-level exponential backoff and an optional deadline. /// /// Each round attempts every provider once in round-robin starting at `rr_start`. If any @@ -824,14 +877,15 @@ macro_rules! log_at { /// /// Used by both the data-method `call()` (rotates `rr_start` per call for load balancing) and /// `get_witness()` (pins `rr_start=0` for primary-failover). -// 9-argument retry primitive. Each field plays a distinct role (providers, concurrency, -// backoff policy, per-attempt timeout, starting provider, method label, metrics sink, deadline, -// per-attempt closure) and bundling them into a struct would be ceremony without encapsulation — -// there are exactly two call sites in this crate. Prefer clarity at the definition over fewer -// commas at the call. +// 10-argument retry primitive. Each field plays a distinct role (providers, their metric/log +// labels, concurrency, backoff policy, per-attempt timeout, starting provider, method label, +// metrics sink, deadline, per-attempt closure) and bundling them into a struct would be ceremony +// without encapsulation — there are exactly two call sites in this crate. Prefer clarity at the +// definition over fewer commas at the call. `provider_labels` is parallel to `providers` by index. #[allow(clippy::too_many_arguments)] async fn round_robin_with_backoff( providers: &[RootProvider], + provider_labels: &[Arc], semaphore: &Semaphore, policy: &BackoffPolicy, per_attempt_timeout: Duration, @@ -839,12 +893,28 @@ async fn round_robin_with_backoff( method: RpcMethod, metrics: Option<&Arc>, deadline: Option, - f: impl Fn(RootProvider) -> BoxFuture>, + f: impl Fn(RootProvider, Arc) -> BoxFuture>, ) -> std::result::Result where N: alloy_provider::Network, T: Send + 'static, { + debug_assert_eq!( + providers.len(), + provider_labels.len(), + "provider_labels must be parallel to providers (indexed by the same provider_idx)", + ); + + // Records the logical-call deadline give-up (once) and builds the typed error. Called from + // every site that abandons the call on a blown deadline, so the "request timed out" metric + // and the returned error stay in lockstep. + let record_deadline = |elapsed: Duration| -> RpcDeadlineExceeded { + if let Some(m) = metrics { + m.on_rpc_deadline_exceeded(method, elapsed.as_secs_f64()); + } + RpcDeadlineExceeded { method, elapsed } + }; + /// Round index from which retry logs escalate DEBUG → WARN. Short blips typically resolve /// within the first few rounds (with `initial=500ms, max=30s` defaults that's rounds 0/1/2 /// sleeping 500/1000/2000 ms + jitter, so cumulative 3.5–5.25 s before a round-3 WARN). @@ -873,9 +943,10 @@ where if let Some(d) = deadline && Instant::now() >= d { - return Err(RpcDeadlineExceeded { method, elapsed: call_start.elapsed() }); + return Err(record_deadline(call_start.elapsed())); } let provider_idx = (rr_start + offset) % n; + let provider_label: &str = &provider_labels[provider_idx]; let permit = semaphore.acquire().await.expect("semaphore closed unexpectedly"); // Per-attempt timing: record each provider call individually so histograms // and success/error counters reflect what actually happened in the retry loop @@ -889,92 +960,98 @@ where Some(d) => d.saturating_duration_since(Instant::now()).min(per_attempt_timeout), None => per_attempt_timeout, }; - let result = match tokio::time::timeout( - attempt_timeout, - f(providers[provider_idx].clone()), - ) - .await - { - Ok(r) => r, - Err(_) => { - // Attempt-level timeout fired. Distinguish two cases: - // - deadline set and now past it ⇒ overall call budget exhausted; bail with - // the typed error so the caller sees one consistent failure mode. - // - otherwise ⇒ the provider stalled but the call still has budget; - // synthesize a normal error and rotate to the next provider in this round, - // mirroring the path a returned `Err` from the closure takes. - if let Some(d) = deadline && - Instant::now() >= d - { - drop(permit); - if let Some(m) = metrics { - // Only record `on_rpc_complete(false)` here — NOT `on_rpc_retry`. - // The non-timeout failure arm fires `on_rpc_retry` because - // it's about to loop and try another provider; this arm - // gives up, so counting it as a retry would inflate the - // retry metric above the actual number of attempts made. - m.on_rpc_complete( - method, - false, - Some(attempt_start.elapsed().as_secs_f64()), - ); - } - return Err(RpcDeadlineExceeded { method, elapsed: call_start.elapsed() }); - } - // Always WARN, regardless of round — a stalled provider (TCP-accept- - // no-reply) is the failure mode this guard exists to catch, and ops - // need to see it on round 0 instead of waiting for the round-3 - // escalation that the returned-Err path goes through. - warn!( - method = method.as_str(), - provider_idx, - round, - attempt_timeout_ms = attempt_timeout.as_millis() as u64, - "RPC provider stalled past per-attempt timeout, rotating", - ); - Err(eyre!( - "{} attempt against provider {} timed out after {:?} (per_attempt_timeout)", - method.as_str(), - provider_idx, - attempt_timeout, - )) - } - }; + // Classify the attempt into a value or a (typed error, reason) pair, so the failure + // reason — a returned error vs a per-attempt stall — survives to the metrics and logs. + let attempt: std::result::Result = + match tokio::time::timeout( + attempt_timeout, + f(providers[provider_idx].clone(), Arc::clone(&provider_labels[provider_idx])), + ) + .await + { + Ok(Ok(v)) => Ok(v), + Ok(Err(e)) => Err((e, RpcAttemptOutcome::Error)), + Err(_) => Err(( + eyre!( + "{} attempt against provider {} ({}) timed out after {:?} (per_attempt_timeout)", + method.as_str(), + provider_idx, + provider_label, + attempt_timeout, + ), + RpcAttemptOutcome::Timeout, + )), + }; let attempt_duration = attempt_start.elapsed().as_secs_f64(); drop(permit); - match result { + let (err, outcome) = match attempt { Ok(v) => { if let Some(m) = metrics { - m.on_rpc_complete(method, true, Some(attempt_duration)); - } - return Ok(v); - } - Err(e) => { - if let Some(m) = metrics { - m.on_rpc_complete(method, false, Some(attempt_duration)); - m.on_rpc_retry(method); - } - // Log "trying next" only when there really is a next provider in this - // round. Suppresses two kinds of noise: - // - Single-provider config (n=1): no per-provider log at all; the - // round-summary log below is the whole story. - // - Multi-provider, last provider in a round: no "trying next" lie; the error - // is carried into the round-summary log instead. - let has_next = offset + 1 < n; - if has_next { - log_at!( - warn_level, - method = method.as_str(), - provider_idx, - round, - error = %e, - "RPC provider failed, trying next", + m.on_rpc_attempt( + method, + provider_label, + RpcAttemptOutcome::Success, + attempt_duration, ); } - last_err = Some(e); + return Ok(v); } + Err(pair) => pair, + }; + + // A per-attempt timeout that also blew the overall deadline is deadline pressure, not + // a provider stall: the attempt window was clamped to the little budget left, so the + // provider never got a fair round trip. Attribute it to the deadline (record_deadline) + // and do NOT record a per-provider `timeout` — otherwise the timeout metric blames the + // endpoint for the caller's exhausted budget. + if outcome == RpcAttemptOutcome::Timeout && + let Some(d) = deadline && + Instant::now() >= d + { + return Err(record_deadline(call_start.elapsed())); + } + + // Record the failed attempt against its endpoint with the reason (error vs a genuine + // stall — a full per-attempt window elapsed with budget still remaining). + if let Some(m) = metrics { + m.on_rpc_attempt(method, provider_label, outcome, attempt_duration); + } + + if outcome == RpcAttemptOutcome::Timeout { + // Stalled but budget remains. Always WARN, regardless of round — a stalled + // provider (TCP-accept-no-reply) is the failure mode this guard exists to catch, + // and ops need to see it on round 0 instead of waiting for the round-3 escalation. + warn!( + method = method.as_str(), + provider_idx, + provider = %provider_label, + round, + attempt_timeout_ms = attempt_timeout.as_millis() as u64, + "RPC provider stalled past per-attempt timeout, rotating", + ); + } + + // Transient failure with budget left → count the retry and rotate to the next provider. + if let Some(m) = metrics { + m.on_rpc_retry(method); + } + // Log "trying next" only for a returned error with a next provider in this round: a + // stall already logged its own WARN above, and the last provider's error is carried + // into the round-summary log below. + let has_next = offset + 1 < n; + if has_next && outcome != RpcAttemptOutcome::Timeout { + log_at!( + warn_level, + method = method.as_str(), + provider_idx, + provider = %provider_label, + round, + error = %err, + "RPC provider failed, trying next", + ); } + last_err = Some(err); } // Every provider in this round failed — summarize with the last error + sleep. @@ -991,7 +1068,7 @@ where if let Some(d) = deadline { let remaining_ms = d.saturating_duration_since(Instant::now()).as_millis() as u64; if remaining_ms == 0 { - return Err(RpcDeadlineExceeded { method, elapsed: call_start.elapsed() }); + return Err(record_deadline(call_start.elapsed())); } sleep_ms = sleep_ms.min(remaining_ms); } @@ -1107,17 +1184,20 @@ async fn do_get_header( /// multi-MB payload is CPU-bound). async fn fetch_witness_with( provider: &RootProvider, + provider_label: &str, number: u64, hash: B256, decode: fn(&str) -> std::result::Result, trace_msg: &'static str, ) -> Result { let keys = WitnessRequestKeys { block_number: U64::from(number), block_hash: hash }; + let request_start = Instant::now(); let encoded: String = provider .client() .request("mega_getBlockWitness", (keys,)) .await .map_err(|e| eyre!("mega_getBlockWitness failed for block {number}: {e}"))?; + let request_ms = request_start.elapsed().as_millis(); let decode_start = Instant::now(); let result = tokio::task::spawn_blocking(move || -> Result { @@ -1126,10 +1206,14 @@ async fn fetch_witness_with( .await .context("decode task panicked")??; + // Per-endpoint success trace: names the serving endpoint and splits the RPC round trip from the + // CPU-bound decode, so a slow witness fetch can be attributed to the right fallback endpoint. trace!( block_number = number, %hash, - decode_ms = decode_start.elapsed().as_millis(), + provider = %provider_label, + request_ms = request_ms as u64, + decode_ms = decode_start.elapsed().as_millis() as u64, "{trace_msg}", ); @@ -1876,4 +1960,190 @@ mod tests { handle.stop().unwrap(); } + + #[test] + fn test_endpoint_label_extracts_host_without_credentials() { + assert_eq!( + &*endpoint_label("http://witness.example.com:8545/rpc", 0), + "0:witness.example.com:8545" + ); + assert_eq!(&*endpoint_label("http://127.0.0.1:9000", 1), "1:127.0.0.1:9000"); + // userinfo, path, and query are all stripped — no credential can reach a label/log. + assert_eq!(&*endpoint_label("https://user:secret@host.io/v1?token=abc", 2), "2:host.io"); + // No host to extract → stable positional fallback. + assert_eq!(&*endpoint_label("", 3), "provider_3"); + // Same authority, different index → distinct labels, so per-endpoint metrics don't merge. + assert_ne!( + endpoint_label("https://gw.saas.io/key-a", 0), + endpoint_label("https://gw.saas.io/key-b", 1) + ); + } + + /// Captures [`RpcMetrics`] callbacks so tests can assert per-endpoint attribution. + #[derive(Default)] + struct CapturingMetrics { + attempts: std::sync::Mutex>, + deadlines: std::sync::Mutex>, + } + + impl RpcMetrics for CapturingMetrics { + fn on_rpc_attempt( + &self, + method: RpcMethod, + provider: &str, + outcome: RpcAttemptOutcome, + _duration_secs: f64, + ) { + self.attempts.lock().unwrap().push((method, provider.to_owned(), outcome)); + } + + fn on_rpc_deadline_exceeded(&self, method: RpcMethod, _elapsed_secs: f64) { + self.deadlines.lock().unwrap().push(method); + } + + fn on_witness_fetch(&self, _breakdown: WitnessSizeBreakdown) {} + } + + /// Each provider attempt reports its own endpoint label and outcome: a failing primary + /// records `Error` against its host, the healthy backup records `Success` against its host. + /// This is the per-endpoint attribution the collapsed method-only metric could not give. + #[tokio::test] + async fn test_metrics_record_per_provider_attempt_outcomes() { + let (ha, url_a, _) = start_counting_block_number_rpc(7, usize::MAX).await; // always errors + let (hb, url_b) = start_block_number_rpc(7).await; + + let metrics = Arc::new(CapturingMetrics::default()); + let config = RpcClientConfig { + rpc_retry: BackoffPolicy::new(Duration::from_millis(1), Duration::from_millis(2)), + ..Default::default() + } + .with_metrics(metrics.clone()); + // Data round-robin starts at index 0, so the failing A is tried before the healthy B. + let client = RpcClient::new_with_config( + &[url_a.as_str(), url_b.as_str()], + &[url_a.as_str()], + config, + None, + ) + .unwrap(); + + assert_eq!(client.get_latest_block_number().await, 7); + + let attempts = metrics.attempts.lock().unwrap(); + assert_eq!( + *attempts, + vec![ + ( + RpcMethod::EthBlockNumber, + endpoint_label(&url_a, 0).to_string(), + RpcAttemptOutcome::Error + ), + ( + RpcMethod::EthBlockNumber, + endpoint_label(&url_b, 1).to_string(), + RpcAttemptOutcome::Success + ), + ], + "expected A→Error then B→Success, each tagged with its own endpoint host", + ); + + ha.stop().unwrap(); + hb.stop().unwrap(); + } + + /// A logical call that exhausts its deadline records exactly one `on_rpc_deadline_exceeded` + /// (the operator-facing "request timed out" signal), on top of the per-attempt `Error`s. + #[tokio::test] + async fn test_metrics_record_deadline_exceeded_once() { + let (h, url, _) = start_counting_block_number_rpc(1, usize::MAX).await; // always errors + + let metrics = Arc::new(CapturingMetrics::default()); + let config = RpcClientConfig { + rpc_retry: BackoffPolicy::new(Duration::from_millis(5), Duration::from_millis(10)), + ..Default::default() + } + .with_metrics(metrics.clone()); + let client = + RpcClient::new_with_config(&[url.as_str()], &[url.as_str()], config, None).unwrap(); + + let deadline = Instant::now() + Duration::from_millis(150); + let err = client + .get_latest_block_number_with_deadline(Some(deadline)) + .await + .expect_err("always-erroring provider must exceed the deadline"); + assert_eq!(err.method, RpcMethod::EthBlockNumber); + + assert_eq!( + *metrics.deadlines.lock().unwrap(), + vec![RpcMethod::EthBlockNumber], + "exactly one give-up per logical call", + ); + let attempts = metrics.attempts.lock().unwrap(); + assert!( + !attempts.is_empty() && attempts.iter().all(|(_, _, o)| *o == RpcAttemptOutcome::Error), + "every attempt errored before the deadline: {attempts:?}", + ); + + h.stop().unwrap(); + } + + /// A stalled witness fetch's failure log must carry `block_number` even at the default `warn` + /// filter, proving the `warn`-level span propagates block context into the generic retry + /// loop's per-attempt logs (which have no block identifier of their own). + #[tokio::test] + async fn test_witness_failure_log_carries_block_number() { + #[derive(Clone)] + struct SharedBuf(Arc>>); + impl std::io::Write for SharedBuf { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for SharedBuf { + type Writer = SharedBuf; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + let buf = Arc::new(std::sync::Mutex::new(Vec::::new())); + let subscriber = tracing_subscriber::fmt() + .with_writer(SharedBuf(buf.clone())) + .with_ansi(false) + .with_max_level(tracing::Level::WARN) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + + // Stalled endpoint: accepts the TCP connection but never replies, so the per-attempt + // timeout fires and the retry loop emits its stall WARN — the log we assert on. + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let stalled_url = format!("http://{}/", listener.local_addr().unwrap()); + let _listener = listener; + let config = RpcClientConfig { + per_attempt_timeout: Duration::from_millis(50), + rpc_retry: BackoffPolicy::new(Duration::from_millis(1), Duration::from_millis(2)), + ..Default::default() + }; + let client = + RpcClient::new_with_config(&[&stalled_url], &[&stalled_url], config, None).unwrap(); + + // `deadline = None` so every per-attempt timeout emits the stall WARN (no deadline branch + // can steal it) — the assertion is independent of runner load. The outer timeout bounds + // the otherwise-unbounded retry so the test terminates. + let _ = tokio::time::timeout( + Duration::from_secs(1), + client.get_witness_light_with_deadline(4242, B256::ZERO, None), + ) + .await; + + let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap(); + assert!( + logs.contains("block_number") && logs.contains("4242"), + "witness failure log must carry the block_number span field, got:\n{logs}" + ); + } }