From 190f0aa4f875a662a0d5cca40e150c2245976e76 Mon Sep 17 00:00:00 2001 From: Vincent Taglia Date: Mon, 19 Oct 2026 10:55:38 -0500 Subject: [PATCH] chain/ethereum: Restore eth_getLogs range reduction after alloy migration log_stream shrinks the block range when a provider rejects eth_getLogs as too heavy. It recognised that by substring-matching the rendered error, including "ServerError(-32005)" and "ServerError(-32000)", which are rust-web3's Debug rendering. Since the alloy migration in v0.42.0 those strings never appear, so a provider enforcing a range or result cap got ten identical retries, "Unexpected RPC error", and a block stream restart loop instead of a smaller range. Match the alloy error structurally instead. A known cap message shrinks the range at once, without retrying. Caps are matched by message because most providers report them under a generic code such as -32602. Everything else is retried and surfaced as before, and the polling block stream lowers its own range in response, so an unknown cap still converges. The message is also read out of HTTP error bodies and out of 200 bodies alloy could not deserialize, on any status: some providers answer a cap with an error status and the reason in the body, e.g. Ankr with a 413 and GetBlock with a 403, and the transport otherwise hides it. Only the message is consulted, never the code, since a code in an error body is as likely a rate limit. Add GRAPH_RPC_ETH_GET_LOGS_ERRORS, a `;`-separated list of extra cap messages mirroring GRAPH_RPC_ETH_CALL_ERRORS, so operators can cover providers that are not hardcoded. Both variables are documented in docs/environment-variables.md. Behaviour change for GRAPH_RPC_ETH_CALL_ERRORS: its entries are now trimmed and lowercased when parsed. Until now they were compared as-is against a lowercased RPC error message, so an entry only matched if the operator had written it in lower case, and an entry with a stray space around a `;` either matched every message (a whitespace-only entry) or missed messages that began or ended with the phrase. Entries that were already lower case and unpadded behave exactly as before. --- chain/ethereum/src/env.rs | 31 ++- chain/ethereum/src/ethereum_adapter.rs | 300 ++++++++++++++++++++++--- docs/environment-variables.md | 7 + 3 files changed, 306 insertions(+), 32 deletions(-) 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