diff --git a/chain/ethereum/src/env.rs b/chain/ethereum/src/env.rs index b15ee0ee018..c3a60133d62 100644 --- a/chain/ethereum/src/env.rs +++ b/chain/ethereum/src/env.rs @@ -14,8 +14,16 @@ pub struct EnvVars { /// Additional deterministic errors that have not yet been hardcoded. /// /// Set by the environment variable `GRAPH_RPC_ETH_CALL_ERRORS`, separated - /// by `;`. + /// by `;`. Matched case-insensitively as substrings of the error message. pub rpc_eth_call_errors: Vec, + /// Additional `eth_getLogs` error messages that mean the request exceeded + /// the provider's block-range or result cap, so the range should shrink + /// rather than the request be retried. Matched case-insensitively as + /// substrings of the error message. + /// + /// Set by the environment variable `GRAPH_RPC_ETH_GET_LOGS_ERRORS`, + /// separated by `;`. + pub rpc_eth_get_logs_errors: Vec, /// Set by the environment variable `GRAPH_ETH_GET_LOGS_MAX_CONTRACTS`. The /// default value is 2000. pub get_logs_max_contracts: usize, @@ -110,16 +118,23 @@ impl EnvVars { } } +/// Splits a `;`-separated list of error message fragments, trimmed and +/// lowercased so they can be matched against a lowercased message. +fn split_error_messages(value: &str) -> Vec { + value + .split(';') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_lowercase) + .collect() +} + impl From for EnvVars { fn from(x: Inner) -> Self { Self { get_logs_max_contracts: x.get_logs_max_contracts, - rpc_eth_call_errors: x - .rpc_eth_call_errors - .split(';') - .filter(|s| !s.is_empty()) - .map(str::to_string) - .collect(), + rpc_eth_call_errors: split_error_messages(&x.rpc_eth_call_errors), + rpc_eth_get_logs_errors: split_error_messages(&x.rpc_eth_get_logs_errors), trace_stream_step_size: x.trace_stream_step_size, max_event_only_range: x.max_event_only_range, block_batch_size: x.block_batch_size, @@ -160,6 +175,8 @@ impl Default for EnvVars { struct Inner { #[envconfig(from = "GRAPH_RPC_ETH_CALL_ERRORS", default = "")] rpc_eth_call_errors: String, + #[envconfig(from = "GRAPH_RPC_ETH_GET_LOGS_ERRORS", default = "")] + rpc_eth_get_logs_errors: String, #[envconfig(from = "GRAPH_ETH_GET_LOGS_MAX_CONTRACTS", default = "2000")] get_logs_max_contracts: usize, diff --git a/chain/ethereum/src/ethereum_adapter.rs b/chain/ethereum/src/ethereum_adapter.rs index 9c85f0ac551..3d48b533f80 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,73 @@ 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; operators can add their own through +/// `GRAPH_RPC_ETH_GET_LOGS_ERRORS`. +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 +]; + +/// Whether an `eth_getLogs` request failed because it exceeded the provider's +/// cap. Such a request is not retried as-is; `log_stream` shrinks the range +/// instead. Anything else is retried and then surfaced, and the polling block +/// stream lowers its own range in response. +fn is_too_many_logs_err(err: &RpcError) -> bool { + match err { + RpcError::ErrorResp(payload) => is_too_many_logs_message(&payload.message), + // 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. + RpcError::Transport(TransportErrorKind::HttpError(http_err)) => { + error_message_in_body(&http_err.body).is_some_and(|m| is_too_many_logs_message(&m)) + } + // A 200 with an error object alloy could not parse, e.g. missing `id`. + RpcError::DeserError { text, .. } => { + error_message_in_body(text).is_some_and(|m| is_too_many_logs_message(&m)) + } + _ => false, + } +} + +fn is_too_many_logs_message(message: &str) -> bool { + let message = message.to_lowercase(); + TOO_MANY_LOGS_MESSAGES + .iter() + .copied() + .chain(ENV_VARS.rpc_eth_get_logs_errors.iter().map(String::as_str)) + .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) +} + impl EthereumAdapter { pub fn is_call_only(&self) -> bool { self.call_only @@ -395,7 +463,6 @@ impl EthereumAdapter { from: BlockNumber, to: BlockNumber, filter: Arc, - too_many_logs_fingerprints: &'static [&'static str], ) -> Result< Vec, TimeoutError>, @@ -407,11 +474,9 @@ 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)), + Err(e) => !is_too_many_logs_err(e), }, ) .limit(self.settings.request_retries) @@ -507,15 +572,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 +615,17 @@ 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 - { + // A client-side timeout (`Elapsed`) never shrinks the range. + let too_many_logs = + matches!(&e, TimeoutError::Inner(e) if is_too_many_logs_err(e)); + if too_many_logs && 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 +633,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 +2746,217 @@ 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, TOO_MANY_LOGS_MESSAGES, + block_trigger_types_from_intervals, check_block_receipt_support, is_too_many_logs_err, + parse_block_triggers, }; use graph::blockchain::BlockPtr; use graph::components::ethereum::AnyNetworkBare; 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(), + })) + } + + /// 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!( + is_too_many_logs_err(&error_resp(-32602, fragment)), + "fragment {fragment:?} did not match on its own" + ); + } + } + + /// Verbatim cap messages, under the code each provider uses. + #[test] + fn too_many_logs_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. + ( + -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. + ( + -32602, + "eth_getLogs and eth_newFilter are limited to a 10,000 blocks range", + ), + // Dwellir. + ( + -32005, + "Plan block-range limit exceeded, or a stricter native limit", + ), + ]; + for (code, message) in caps { + assert!( + is_too_many_logs_err(&error_resp(code, message)), + "{message:?}" + ); + } + } + + /// Codes alone never match, including the ones the old fingerprints + /// matched on: providers use them for transient and unrelated failures too. + #[test] + fn too_many_logs_ignores_codes_without_a_cap_message() { + let unrelated = [ + error_resp(-32000, "header not found"), + error_resp(-32002, "request timed out"), + error_resp(-32003, ""), + error_resp(-32005, "Filter address count exceeds limit"), + 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, ""), + http_err(503, ""), + http_err( + 503, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32055,"message":"No nodes available"}}"#, + ), + RpcError::NullResp, + ]; + for err in unrelated { + assert!(!is_too_many_logs_err(&err), "{err:?}"); + } + } + + /// A cap in an HTTP error body, or in a 200 body alloy could not parse. + #[test] + fn too_many_logs_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. + let ankr = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32062,"message":"Block range is too large","data":{"trace_id":"4b1f"}}}"#; + assert!(is_too_many_logs_err(&http_err(400, full))); + assert!(is_too_many_logs_err(&http_err(413, ankr))); + assert!(is_too_many_logs_err(&http_err(503, full))); + assert!(is_too_many_logs_err(&deser_err(full))); + assert!(is_too_many_logs_err(&deser_err(bare))); + // 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!(is_too_many_logs_err(&http_err(403, body)), "{body}"); + } + + // Bodies that are not a cap, or not a JSON-RPC error at all. + assert!(!is_too_many_logs_err(&http_err( + 400, + r#"{"error":{"code":-32602,"message":"invalid block range params"}}"# + ))); + assert!(!is_too_many_logs_err(&http_err( + 400, + "Bad Request" + ))); + assert!(!is_too_many_logs_err(&deser_err("not json"))); + } + + /// Infura rate-limits with a 429 and -32005, the code of its result cap. + #[test] + fn too_many_logs_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] { + assert!( + !is_too_many_logs_err(&http_err(status, rate_limited)), + "{status}" + ); + assert!( + !is_too_many_logs_err(&http_err(status, backend)), + "{status}" + ); + } + assert!(!is_too_many_logs_err(&deser_err(rate_limited))); + assert!(!is_too_many_logs_err(&deser_err(backend))); + } + + /// The message is the source of truth; the status does not gate it, so + /// an operator-supplied message works whatever status its provider uses. + #[test] + fn too_many_logs_bodies_are_read_on_any_status() { + let cap = + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32602,"message":"block range too large"}}"#; + for status in [400, 401, 403, 413, 422, 429, 500, 502, 503, 504] { + assert!(is_too_many_logs_err(&http_err(status, cap)), "{status}"); + } + } + #[test] fn parse_block_triggers_every_block() { let block = create_minimal_block_for_test(2, hash(2)); diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 7a601e32e64..4bfdd29b540 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -60,6 +60,13 @@ those. is not set, the default value will be `0`. - `GRAPH_ETH_GET_LOGS_MAX_CONTRACTS`: Maximum number of contracts to query in a single `eth_getLogs` request. Defaults to 2000. +- `GRAPH_RPC_ETH_GET_LOGS_ERRORS`: Additional `eth_getLogs` error messages that mean the + request exceeded the provider's block-range or result cap, so graph-node shrinks the + block range instead of retrying. A `;`-separated list of substrings, matched + case-insensitively against the error message, for providers not already recognised. +- `GRAPH_RPC_ETH_CALL_ERRORS`: Additional `eth_call` error messages to treat as a + deterministic revert rather than a transient failure. A `;`-separated list of substrings, + matched case-insensitively against the error message. ## Firehose configuration