diff --git a/chain/ethereum/src/ethereum_adapter.rs b/chain/ethereum/src/ethereum_adapter.rs index 9c85f0ac551..338b079e51f 100644 --- a/chain/ethereum/src/ethereum_adapter.rs +++ b/chain/ethereum/src/ethereum_adapter.rs @@ -40,6 +40,7 @@ use graph::prelude::{ }, transports::{RpcError, TransportErrorKind}, }, + serde_json, tokio::try_join, }; use graph::slog::o; @@ -151,6 +152,120 @@ impl std::ops::Deref for ProviderLogger { } } +/// Lowercased fragments of the messages providers return when an +/// `eth_getLogs` request exceeds a block-range or result cap. Caps are matched +/// by message because most providers report them under a generic code. Add an +/// entry only with a captured response. +const TOO_MANY_LOGS_MESSAGES: &[&str] = &[ + "exceed maximum block range", // geth + "query exceeds max block range", // reth + "query exceeds max results", // reth + "requested range exceeds maximum rpc range limit", // Besu + "query returned more than", // Infura + "this block range should work", // Alchemy + "log response size exceeded", // Alchemy + "try with this block range", // zkSync Era + "block range too large", // Monad + "block range is too large", // Ankr + "response too large", // geth + "eth_getlogs and eth_newfilter are limited to", // QuickNode + "block-range limit exceeded", // Dwellir + "block range exceeds the limit", // GetBlock +]; + +/// Codes that can mean a request was too heavy but are also used for +/// transient or unrelated failures, e.g. geth's catch-all and timeout, or +/// Infura's rate limit. +const MAYBE_TOO_HEAVY_CODES: &[i64] = &[-32005, -32003, -32002, -32000]; + +/// Why an `eth_getLogs` request failed, as far as range reduction goes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum LogsRequestFailure { + /// Over the provider's cap: shrink the range without retrying. + TooHeavy, + /// Possibly too heavy, possibly transient: retry, then shrink. + MaybeTooHeavy, + /// Retry, then surface the error. + Unrecognized, +} + +fn classify_logs_failure(err: &RpcError) -> LogsRequestFailure { + use LogsRequestFailure::*; + + match err { + RpcError::ErrorResp(payload) => { + if is_too_many_logs_message(&payload.message) { + TooHeavy + } else if MAYBE_TOO_HEAVY_CODES.contains(&payload.code) { + MaybeTooHeavy + } else { + Unrecognized + } + } + RpcError::Transport(TransportErrorKind::HttpError(http_err)) => { + // Some providers send a cap as an HTTP error with the reason in + // the body, e.g. Ankr (413) and GetBlock (403). Only the message + // is consulted: a code in an error body is as likely a rate limit. + if matches!(http_err.status, 400 | 403 | 413 | 503) + && error_message_in_body(&http_err.body) + .is_some_and(|m| is_too_many_logs_message(&m)) + { + return TooHeavy; + } + // A bare 503 can be load shedding or an unavailable backend. + if http_err.is_temporarily_unavailable() { + return MaybeTooHeavy; + } + Unrecognized + } + // A 200 with an error object alloy could not parse, e.g. missing `id`. + RpcError::DeserError { text, .. } => { + if error_message_in_body(text).is_some_and(|m| is_too_many_logs_message(&m)) { + TooHeavy + } else { + Unrecognized + } + } + _ => Unrecognized, + } +} + +fn is_too_many_logs_message(message: &str) -> bool { + let message = message.to_lowercase(); + TOO_MANY_LOGS_MESSAGES.iter().any(|m| message.contains(m)) +} + +/// Extracts the error message from a raw response body: a bare JSON-RPC error +/// object, `{"error": {"message": ..}}`, or `{"error": ".."}`. +fn error_message_in_body(body: &str) -> Option { + let body: serde_json::Value = serde_json::from_str(body).ok()?; + let error = if body.get("code").is_some() { + &body + } else { + body.get("error")? + }; + error + .as_str() + .or_else(|| error.get("message")?.as_str()) + .map(str::to_owned) +} + +/// Whether `logs_with_sigs` should retry the same request. +fn should_retry_logs_request(err: &RpcError) -> bool { + classify_logs_failure(err) != LogsRequestFailure::TooHeavy +} + +/// Whether `log_stream` should shrink the range once `logs_with_sigs` has +/// given up. A client-side timeout has never shrunk the range. +fn should_shrink_logs_range(err: &TimeoutError>) -> bool { + match err { + TimeoutError::Inner(rpc_err) => { + classify_logs_failure(rpc_err) != LogsRequestFailure::Unrecognized + } + TimeoutError::Elapsed => false, + } +} + impl EthereumAdapter { pub fn is_call_only(&self) -> bool { self.call_only @@ -395,7 +510,6 @@ impl EthereumAdapter { from: BlockNumber, to: BlockNumber, filter: Arc, - too_many_logs_fingerprints: &'static [&'static str], ) -> Result< Vec, TimeoutError>, @@ -407,11 +521,11 @@ impl EthereumAdapter { retry(retry_log_message, &logger) .redact_log_urls(true) .when( - move |res: &Result<_, RpcError>| match res { + |res: &Result<_, RpcError>| match res { Ok(_) => false, - Err(e) => !too_many_logs_fingerprints - .iter() - .any(|f| e.to_string().contains(f)), + // Retrying a request over the provider's cap cannot help; + // `log_stream` shrinks the range instead. + Err(e) => should_retry_logs_request(e), }, ) .limit(self.settings.request_retries) @@ -507,15 +621,6 @@ impl EthereumAdapter { to: BlockNumber, filter: EthGetLogsFilter, ) -> DynTryFuture<'static, Vec, Error> { - // Codes returned by Ethereum node providers if an eth_getLogs request is too heavy. - const TOO_MANY_LOGS_FINGERPRINTS: &[&str] = &[ - "ServerError(-32005)", // Infura - "503 Service Unavailable", // Alchemy - "ServerError(-32000)", // Alchemy - "Try with this block range", // zKSync era - "block range too large", // Monad - ]; - if from > to { panic!( "cannot produce a log stream on a backwards block range (from={}, to={})", @@ -559,21 +664,14 @@ impl EthereumAdapter { start, end, filter.cheap_clone(), - TOO_MANY_LOGS_FINGERPRINTS, ) .await; match res { Err(e) => { - let string_err = e.to_string(); - // If the step is already 0, the request is too heavy even for a single // block. We hope this never happens, but if it does, make sure to error. - if TOO_MANY_LOGS_FINGERPRINTS - .iter() - .any(|f| string_err.contains(f)) - && step > 0 - { + if should_shrink_logs_range(&e) && step > 0 { // The range size for a request is `step + 1`. So it's ok if the step // goes down to 0, in that case we'll request one block at a time. let new_step = step / 10; @@ -581,6 +679,7 @@ impl EthereumAdapter { "new_size" => new_step + 1); Ok(Some((vec![], (start, new_step)))) } else { + let string_err = e.to_string(); warn!(logger, "Unexpected RPC error"; "error" => &string_err); Err(anyhow!("{}", string_err)) } @@ -2693,20 +2792,274 @@ mod tests { use crate::trigger::{EthereumBlockTriggerType, EthereumTrigger}; use super::{ - EthereumBlock, EthereumBlockFilter, EthereumBlockWithCalls, - block_trigger_types_from_intervals, check_block_receipt_support, parse_block_triggers, + EthereumBlock, EthereumBlockFilter, EthereumBlockWithCalls, LogsRequestFailure, + TOO_MANY_LOGS_MESSAGES, block_trigger_types_from_intervals, check_block_receipt_support, + classify_logs_failure, parse_block_triggers, should_retry_logs_request, + should_shrink_logs_range, }; use graph::blockchain::BlockPtr; use graph::components::ethereum::AnyNetworkBare; + use graph::prelude::TimeoutError; use graph::prelude::alloy::primitives::{Address, B256, Bytes}; use graph::prelude::alloy::providers::ProviderBuilder; use graph::prelude::alloy::providers::mock::Asserter; + use graph::prelude::alloy::rpc::json_rpc::ErrorPayload; + use graph::prelude::alloy::transports::{HttpError, RpcError, TransportErrorKind}; + use graph::prelude::serde_json::{self, Value}; use graph::prelude::{EthereumCall, LightEthereumBlock, create_minimal_block_for_test}; - use jsonrpc_core::serde_json::{self, Value}; use std::collections::HashSet; use std::iter::FromIterator; use std::sync::Arc; + fn error_resp(code: i64, message: &str) -> RpcError { + RpcError::ErrorResp(ErrorPayload { + code, + message: message.to_string().into(), + data: None, + }) + } + + fn deser_err(text: &str) -> RpcError { + RpcError::DeserError { + err: serde_json::from_str::<()>("").unwrap_err(), + text: text.to_string(), + } + } + + fn http_err(status: u16, body: &str) -> RpcError { + RpcError::Transport(TransportErrorKind::HttpError(HttpError { + status, + body: body.to_string(), + })) + } + + fn classify(err: &RpcError) -> LogsRequestFailure { + classify_logs_failure(err) + } + + use LogsRequestFailure::{MaybeTooHeavy, TooHeavy, Unrecognized}; + + /// Each fragment must match on its own, under a code that never matches. + #[test] + fn every_too_many_logs_message_is_load_bearing() { + for fragment in TOO_MANY_LOGS_MESSAGES { + assert_eq!( + fragment.to_lowercase(), + *fragment, + "fragment {fragment:?} has uppercase characters and can never match" + ); + assert_eq!( + classify(&error_resp(-32602, fragment)), + TooHeavy, + "fragment {fragment:?} did not match on its own" + ); + } + } + + /// Verbatim cap messages, under the code each provider uses. + #[test] + fn caps_are_recognised_by_message() { + let caps = [ + // geth, since ethereum/go-ethereum#34647. + (-32602, "exceed maximum block range 5"), + // reth, `EthFilterError::QueryExceedsMaxBlocks` / `QueryExceedsMaxResults`. + (-32602, "query exceeds max block range 100000"), + ( + -32602, + "query exceeds max results 20000, retry with the range 100-200", + ), + // Besu, `--rpc-max-logs-range`. + (-32005, "Requested range exceeds maximum RPC range limit"), + // Infura. + (-32005, "query returned more than 10000 results"), + // Alchemy free tier, verified against eth-mainnet. + ( + -32600, + "Under the Free tier plan, you can make eth_getLogs requests with \ + up to a 10 block range. Based on your parameters, this block range \ + should work: [0x18eada4, 0x18eadad]. Upgrade to PAYG for expanded \ + block range.", + ), + // Alchemy paid tier. + ( + -32000, + "Log response size exceeded. this block range should work: [0x1, 0x2]", + ), + // zkSync Era. Mixed case, so also covers the lowercasing. + ( + -32602, + "Query returned more than 10000 results. Try with this block range [0x1, 0x2].", + ), + // Monad. + (-32602, "block range too large"), + // geth, `rpc/errors.go`. + (-32003, "response too large"), + // QuickNode, verbatim from its knowledge base. + ( + -32602, + "eth_getLogs and eth_newFilter are limited to a 10,000 blocks range", + ), + // Dwellir, verbatim from its docs; HTTP 200. + ( + -32005, + "Plan block-range limit exceeded, or a stricter native limit", + ), + ]; + for (code, message) in caps { + assert_eq!( + classify(&error_resp(code, message)), + TooHeavy, + "{message:?}" + ); + assert!(!should_retry_logs_request(&error_resp(code, message))); + assert!(should_shrink_logs_range(&TimeoutError::Inner(error_resp( + code, message + )))); + } + } + + /// An ambiguous code with no recognised message. + #[test] + fn ambiguous_codes_are_retried_then_shrunk() { + let ambiguous = [ + (-32000, "header not found"), // geth, lagging provider + (-32002, "request timed out"), // geth + (-32005, "project ID request rate exceeded"), // Infura rate limit + (-32005, "Filter address count exceeds limit"), // Besu + (-32003, ""), + ]; + for (code, message) in ambiguous { + let err = error_resp(code, message); + assert_eq!(classify(&err), MaybeTooHeavy, "{message:?}"); + assert!(should_retry_logs_request(&err)); + assert!(should_shrink_logs_range(&TimeoutError::Inner(err))); + } + // A 503 without a cap message: Alchemy load shedding, Ankr outage. + for body in [ + "", + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32055,"message":"No nodes available"}}"#, + ] { + let err = http_err(503, body); + assert_eq!(classify(&err), MaybeTooHeavy, "{body:?}"); + assert!(should_retry_logs_request(&err)); + assert!(should_shrink_logs_range(&TimeoutError::Inner(err))); + } + } + + /// The -32602 and -32600 cases are verbatim from geth, reth and Alchemy, + /// and are why those two codes cannot be matched outright. + #[test] + fn unrelated_errors_are_retried_then_surfaced() { + let unrelated = [ + error_resp(-32602, "invalid block range params"), + error_resp( + -32602, + "block range extends beyond current head block: requested 500, head 100", + ), + error_resp(-32602, "Invalid params"), + error_resp( + -32600, + "Unsupported method: eth_notARealMethod on ETH_MAINNET", + ), + error_resp(-32601, "the method eth_getLogs does not exist"), + error_resp(-32603, "Internal error"), + http_err(429, ""), + http_err(500, ""), + RpcError::NullResp, + ]; + for err in unrelated { + assert_eq!(classify(&err), Unrecognized, "{err:?}"); + assert!(should_retry_logs_request(&err)); + assert!(!should_shrink_logs_range(&TimeoutError::Inner(err))); + } + } + + /// A cap in an HTTP error body, or in a 200 body alloy could not parse. + #[test] + fn caps_buried_in_bodies_are_recognised() { + let full = + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32602,"message":"block range too large"}}"#; + let bare = r#"{"code":-32005,"message":"query returned more than 10000 results"}"#; + // Ankr, verbatim from its error reference. + let ankr = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32062,"message":"Block range is too large","data":{"trace_id":"4b1f"}}}"#; + assert_eq!(classify(&http_err(400, full)), TooHeavy); + assert_eq!(classify(&http_err(413, ankr)), TooHeavy); + assert_eq!( + classify(&error_resp(-32062, "Block range is too large")), + TooHeavy + ); + assert_eq!(classify(&deser_err(full)), TooHeavy); + assert_eq!(classify(&deser_err(bare)), TooHeavy); + // GetBlock; its docs give two spellings of the message. + for body in [ + r#"{"error":"Requested block range exceeds the limits (1500 blocks)"}"#, + r#"{"error":"Request block range exceeds the limit(1500 blocks)"}"#, + ] { + assert_eq!(classify(&http_err(403, body)), TooHeavy, "{body}"); + } + // A cap message wins over the 503 hint. + assert_eq!(classify(&http_err(503, full)), TooHeavy); + + // Bodies that are not a cap, or not a JSON-RPC error at all. + assert_eq!( + classify(&http_err( + 400, + r#"{"error":{"code":-32602,"message":"invalid block range params"}}"# + )), + Unrecognized + ); + assert_eq!( + classify(&http_err(400, "Bad Request")), + Unrecognized + ); + assert_eq!(classify(&deser_err("not json")), Unrecognized); + } + + /// Infura rate-limits with a 429 and -32005, the code of its result cap. + #[test] + fn bodies_never_match_by_code() { + let rate_limited = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32005,"message":"project ID request rate exceeded","data":{"backoff_seconds":30}}}"#; + let backend = + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"header not found"}}"#; + for status in [400, 403, 413, 429, 500, 502, 503] { + let expected = if status == 503 { + MaybeTooHeavy + } else { + Unrecognized + }; + assert_eq!( + classify(&http_err(status, rate_limited)), + expected, + "{status}" + ); + assert_eq!(classify(&http_err(status, backend)), expected, "{status}"); + } + assert_eq!(classify(&deser_err(rate_limited)), Unrecognized); + assert_eq!(classify(&deser_err(backend)), Unrecognized); + } + + /// Bodies are not read on statuses that are never caps. + #[test] + fn bodies_are_only_read_on_cap_statuses() { + let cap = + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32602,"message":"block range too large"}}"#; + for status in [401, 429, 500, 502, 504] { + assert_eq!(classify(&http_err(status, cap)), Unrecognized, "{status}"); + } + assert_eq!( + classify(&http_err( + 403, + r#"{"error":{"code":-32001,"message":"invalid project id"}}"# + )), + Unrecognized + ); + } + + #[test] + fn elapsed_timeout_never_shrinks() { + assert!(!should_shrink_logs_range(&TimeoutError::Elapsed)); + } + #[test] fn parse_block_triggers_every_block() { let block = create_minimal_block_for_test(2, hash(2));