diff --git a/apps/indexer/src/provider.rs b/apps/indexer/src/provider.rs index 90681bc1..7cb300a9 100644 --- a/apps/indexer/src/provider.rs +++ b/apps/indexer/src/provider.rs @@ -14,6 +14,7 @@ use decode::parse_hex_i64; mod block_transaction; mod code; mod decode; +mod error; mod http_client; mod logs_receipts; mod ops; diff --git a/apps/indexer/src/provider/block_transaction.rs b/apps/indexer/src/provider/block_transaction.rs index f9865088..ba1510e4 100644 --- a/apps/indexer/src/provider/block_transaction.rs +++ b/apps/indexer/src/provider/block_transaction.rs @@ -8,6 +8,7 @@ use super::{ ProviderHeadHashSnapshot, ProviderHeadSnapshot, ProviderResolvedBlock, RAW_PAYLOAD_KIND_FULL_BLOCK, decode::{block_hash_from_value, normalize_hash}, + error::format_provider_error, logs_receipts::ProviderBlockLogFetch, provider_batch_item_limit, request::{JsonRpcBatchCall, is_retryable_provider_error}, @@ -138,7 +139,7 @@ impl JsonRpcProvider { service = "indexer", component = "provider", tag, - error = %format!("{error:#}"), + error = %format_provider_error(&error), "provider checkpoint tag is unavailable; degrading to absent optional head" ); Ok(None) diff --git a/apps/indexer/src/provider/error.rs b/apps/indexer/src/provider/error.rs new file mode 100644 index 00000000..6491a410 --- /dev/null +++ b/apps/indexer/src/provider/error.rs @@ -0,0 +1,146 @@ +pub(super) fn format_provider_error(error: &anyhow::Error) -> String { + let mut rendered = format!("{error:#}"); + for cause in error.chain() { + if let Some(error) = cause.downcast_ref::() { + redact_reqwest_url(&mut rendered, error); + } + } + rendered +} + +pub(super) fn format_provider_transport_error(error: &reqwest::Error) -> String { + let mut rendered = error.to_string(); + redact_reqwest_url(&mut rendered, error); + rendered +} + +pub(super) fn redact_provider_transport_error_url(error: &mut reqwest::Error) { + let Some(url) = error.url_mut() else { + return; + }; + let _ = url.set_username(""); + let _ = url.set_password(None); + let _ = url.set_port(None); + url.set_path(""); + url.set_query(None); + url.set_fragment(None); +} + +fn redact_reqwest_url(rendered: &mut String, error: &reqwest::Error) { + let Some(url) = error.url() else { + return; + }; + let redacted_url = match url.host_str() { + Some(host) => format!("{}://{host}", url.scheme()), + None => format!("{}://", url.scheme()), + }; + *rendered = rendered.replace(url.as_str(), &redacted_url); +} + +#[cfg(test)] +mod tests { + use std::{future::pending, time::Duration}; + + use anyhow::{Context, Result}; + use tokio::net::TcpListener; + + use super::{format_provider_error, format_provider_transport_error}; + use crate::provider::{JsonRpcProvider, request::is_retryable_provider_error}; + + const SECRET_PATH: &str = "/provider-key-secret/v1"; + const SECRET_QUERY: &str = "api_key=query-secret"; + + #[tokio::test] + async fn transport_log_error_keeps_host_and_redacts_path_and_query() -> Result<()> { + let (error, host) = url_bearing_timeout_error().await?; + + let rendered = format_provider_transport_error(&error); + + assert!(rendered.contains(&format!("for url (http://{host})"))); + assert!(!rendered.contains(SECRET_PATH)); + assert!(!rendered.contains(SECRET_QUERY)); + Ok(()) + } + + #[tokio::test] + async fn retry_warning_error_is_redacted_and_remains_retryable() -> Result<()> { + let (error, host) = url_bearing_timeout_error().await?; + let error = anyhow::Error::new(error).context("failed to send JSON-RPC request for test"); + + let rendered = format_provider_error(&error); + + assert!(rendered.contains(&format!("for url (http://{host})"))); + assert!(!rendered.contains(SECRET_PATH)); + assert!(!rendered.contains(SECRET_QUERY)); + assert!(rendered.to_ascii_lowercase().contains("timed out")); + assert!(is_retryable_provider_error(&error)); + assert!(is_retryable_provider_error(&anyhow::anyhow!(rendered))); + Ok(()) + } + + #[tokio::test] + async fn retry_exhaustion_error_is_safe_for_debug_log_sinks() -> Result<()> { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .context("failed to bind retry exhaustion test server")?; + let address = listener + .local_addr() + .context("failed to read retry exhaustion test server address")?; + let server = tokio::spawn(async move { + while let Ok((connection, _)) = listener.accept().await { + tokio::spawn(async move { + let _connection = connection; + pending::<()>().await; + }); + } + }); + let endpoint = format!("http://{address}{SECRET_PATH}?{SECRET_QUERY}"); + let provider = + JsonRpcProvider::new_with_request_timeout(&endpoint, Duration::from_millis(25))?; + + let error = provider + .fetch_json_rpc_result("eth_chainId", Vec::new()) + .await + .expect_err("the provider request must exhaust its timeout retries"); + server.abort(); + let rendered = format!("{error:?}"); + + assert!(rendered.contains(&format!("http://{}", address.ip()))); + assert!(!rendered.contains(SECRET_PATH)); + assert!(!rendered.contains(SECRET_QUERY)); + assert!(is_retryable_provider_error(&error)); + Ok(()) + } + + async fn url_bearing_timeout_error() -> Result<(reqwest::Error, String)> { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .context("failed to bind timeout test server")?; + let address = listener + .local_addr() + .context("failed to read timeout test server address")?; + let server = tokio::spawn(async move { + let _connection = listener.accept().await; + pending::<()>().await; + }); + let endpoint = format!("http://{address}{SECRET_PATH}?{SECRET_QUERY}"); + let client = reqwest::Client::builder() + .timeout(Duration::from_millis(100)) + .build() + .context("failed to build timeout test client")?; + + let error = client + .get(&endpoint) + .send() + .await + .expect_err("the server must hold the response until the request times out"); + server.abort(); + assert!(error.is_timeout(), "expected timeout error: {error}"); + assert_eq!( + error.url().map(reqwest::Url::as_str), + Some(endpoint.as_str()) + ); + + Ok((error, address.ip().to_string())) + } +} diff --git a/apps/indexer/src/provider/request.rs b/apps/indexer/src/provider/request.rs index 7f81b271..43443825 100644 --- a/apps/indexer/src/provider/request.rs +++ b/apps/indexer/src/provider/request.rs @@ -15,6 +15,9 @@ use tracing::warn; use super::{ JsonRpcProvider, + error::{ + format_provider_error, format_provider_transport_error, redact_provider_transport_error_url, + }, http_client::JSON_RPC_POOL_RESET_TIMEOUT_THRESHOLD, payload_cache::{JsonRpcPayloadFingerprint, JsonRpcResultPayload}, }; @@ -87,7 +90,7 @@ impl JsonRpcProvider { method, attempt = attempt + 1, max_attempts = MAX_JSON_RPC_ATTEMPTS, - error = %format!("{error:#}"), + error = %format_provider_error(&error), "retrying transient JSON-RPC provider request" ); sleep_json_rpc_backoff(attempt).await; @@ -189,7 +192,7 @@ impl JsonRpcProvider { request_context = "batch", attempt = attempt + 1, max_attempts = MAX_JSON_RPC_ATTEMPTS, - error = %format!("{error:#}"), + error = %format_provider_error(&error), "retrying transient JSON-RPC provider batch" ); sleep_json_rpc_backoff(attempt).await; @@ -283,7 +286,8 @@ impl JsonRpcProvider { .await { Ok(response) => response, - Err(error) => { + Err(mut error) => { + redact_provider_transport_error_url(&mut error); self.record_json_rpc_transport_error(client_generation, request_context, &error); return Err(error).with_context(|| { format!("failed to send JSON-RPC request for {request_context}") @@ -293,7 +297,8 @@ impl JsonRpcProvider { let status = response.status(); let body = match response.bytes().await { Ok(body) => body, - Err(error) => { + Err(mut error) => { + redact_provider_transport_error_url(&mut error); self.record_json_rpc_transport_error(client_generation, request_context, &error); return Err(error).context("failed to read JSON-RPC response body"); } @@ -336,7 +341,7 @@ impl JsonRpcProvider { timeout_threshold = JSON_RPC_POOL_RESET_TIMEOUT_THRESHOLD, previous_client_generation = reset.previous_generation, new_client_generation = reset.new_generation, - error = %error, + error = %format_provider_transport_error(error), "rebuilt JSON-RPC HTTP client after a transport timeout" ), Ok(None) => {} @@ -345,8 +350,8 @@ impl JsonRpcProvider { component = "provider", request_context, client_generation, - error = %error, - reset_error = %format!("{reset_error:#}"), + error = %format_provider_transport_error(error), + reset_error = %format_provider_error(&reset_error), "failed to rebuild JSON-RPC HTTP client after a transport timeout" ), } diff --git a/apps/indexer/src/provider/transaction_receipts.rs b/apps/indexer/src/provider/transaction_receipts.rs index b498dc50..5a5224f3 100644 --- a/apps/indexer/src/provider/transaction_receipts.rs +++ b/apps/indexer/src/provider/transaction_receipts.rs @@ -9,7 +9,8 @@ use tracing::warn; use super::{ JsonRpcProvider, ProviderBlockBundle, ProviderReceipt, ProviderResolvedBlock, ProviderTransaction, ProviderTransactionReceiptBundle, ProviderTransactionReceiptRequest, - provider_batch_item_limit, provider_batch_request_concurrency, request::JsonRpcBatchCall, + error::format_provider_error, provider_batch_item_limit, provider_batch_request_concurrency, + request::JsonRpcBatchCall, }; use validation::{fallback_receipts_by_key, fallback_transactions_by_key}; @@ -139,7 +140,7 @@ impl JsonRpcProvider { Ok(block_payloads) => block_payloads, Err(error) => { warn!( - error = %format!("{error:#}"), + error = %format_provider_error(&error), selected_transaction_receipt_fallback_count = requests.len(), "block-scoped selected transaction/receipt fallback failed; retrying direct lookup" ); @@ -374,7 +375,7 @@ impl JsonRpcProvider { Ok(fallback_bundles) => fallback_bundles, Err(error) => { warn!( - error = %format!("{error:#}"), + error = %format_provider_error(&error), "fallback JSON-RPC provider failed selected transaction/receipt lookup" ); return bundles;