From c243246f145e7f14b2c8304599ad1fc740b57a8f Mon Sep 17 00:00:00 2001 From: incrypto32 Date: Wed, 10 Dec 2025 15:31:08 +0530 Subject: [PATCH 1/8] Add warning logs on block deserialization failures --- chain/ethereum/src/chain.rs | 35 ++++++++++++++++++++++---- chain/ethereum/src/ethereum_adapter.rs | 14 ++++++++++- 2 files changed, 43 insertions(+), 6 deletions(-) diff --git a/chain/ethereum/src/chain.rs b/chain/ethereum/src/chain.rs index 3f827220b93..bb444441448 100644 --- a/chain/ethereum/src/chain.rs +++ b/chain/ethereum/src/chain.rs @@ -1036,13 +1036,26 @@ impl TriggersAdapterTrait for TriggersAdapter { offset: BlockNumber, root: Option, ) -> Result, Error> { + let ptr_for_log = ptr.clone(); let block: Option = self .chain_store .cheap_clone() .ancestor_block(ptr, offset, root) .await? - .map(|x| x.0) - .map(json::from_value) + .map(|(json_value, block_ptr)| { + json::from_value(json_value.clone()).map_err(|e| { + warn!( + self.logger, + "Failed to deserialize cached ancestor block {} (offset {} from {}): {}. \ + This may indicate stale cache data from a previous version.", + block_ptr.hash_hex(), + offset, + ptr_for_log.hash_hex(), + e + ); + e + }) + }) .transpose()?; Ok(block.map(|block| { BlockFinality::NonFinal(EthereumBlockWithCalls { @@ -1060,9 +1073,21 @@ impl TriggersAdapterTrait for TriggersAdapter { let chain_store = self.chain_store.cheap_clone(); // First try to get the block from the store if let Ok(blocks) = chain_store.blocks(vec![block.hash.clone()]).await { - if let Some(block) = blocks.first() { - if let Ok(block) = json::from_value::(block.clone()) { - return Ok(block.parent_ptr()); + if let Some(cached_json) = blocks.first() { + match json::from_value::(cached_json.clone()) { + Ok(block) => { + return Ok(block.parent_ptr()); + } + Err(e) => { + warn!( + self.logger, + "Failed to deserialize cached block {}: {}. \ + This may indicate stale cache data from a previous version. \ + Falling back to Firehose.", + block.hash_hex(), + e + ); + } } } } diff --git a/chain/ethereum/src/ethereum_adapter.rs b/chain/ethereum/src/ethereum_adapter.rs index bd0febbf56c..6cc2ec674d8 100644 --- a/chain/ethereum/src/ethereum_adapter.rs +++ b/chain/ethereum/src/ethereum_adapter.rs @@ -1627,7 +1627,19 @@ impl EthereumAdapterTrait for EthereumAdapter { .map_err(|e| error!(&logger, "Error accessing block cache {}", e)) .unwrap_or_default() .into_iter() - .filter_map(|value| json::from_value(value).ok()) + .filter_map(|value| { + json::from_value(value.clone()) + .map_err(|e| { + warn!( + &logger, + "Failed to deserialize cached block: {}. \ + This may indicate stale cache data from a previous version. \ + Block will be re-fetched from RPC.", + e + ); + }) + .ok() + }) .map(|b| Arc::new(LightEthereumBlock::new(b))) .collect(); From 12a76fa05b08276036d7f60fbe798b06eccb2f0b Mon Sep 17 00:00:00 2001 From: incrypto32 Date: Wed, 24 Dec 2025 07:11:34 -0800 Subject: [PATCH 2/8] add Firehose/RPC fallback for stale block cache in ancestor_block --- chain/ethereum/src/chain.rs | 166 ++++++++++++++++++++----------- chain/ethereum/src/codec.rs | 30 +++--- graph/src/ipfs/server_address.rs | 2 +- 3 files changed, 124 insertions(+), 74 deletions(-) diff --git a/chain/ethereum/src/chain.rs b/chain/ethereum/src/chain.rs index bb444441448..4624675efb8 100644 --- a/chain/ethereum/src/chain.rs +++ b/chain/ethereum/src/chain.rs @@ -10,7 +10,7 @@ use graph::blockchain::{ use graph::components::network_provider::ChainName; use graph::components::store::{DeploymentCursorTracker, SourceableStore}; use graph::data::subgraph::UnifiedMappingApiVersion; -use graph::firehose::{FirehoseEndpoint, ForkStep}; +use graph::firehose::{FirehoseEndpoint, FirehoseEndpoints, ForkStep}; use graph::futures03::TryStreamExt; use graph::prelude::{ retry, BlockHash, ComponentLoggerConfig, ElasticComponentLoggerConfig, EthereumBlock, @@ -1037,32 +1037,62 @@ impl TriggersAdapterTrait for TriggersAdapter { root: Option, ) -> Result, Error> { let ptr_for_log = ptr.clone(); - let block: Option = self + let cached = self .chain_store .cheap_clone() .ancestor_block(ptr, offset, root) - .await? - .map(|(json_value, block_ptr)| { - json::from_value(json_value.clone()).map_err(|e| { - warn!( - self.logger, - "Failed to deserialize cached ancestor block {} (offset {} from {}): {}. \ - This may indicate stale cache data from a previous version.", - block_ptr.hash_hex(), - offset, - ptr_for_log.hash_hex(), - e - ); - e - }) - }) - .transpose()?; - Ok(block.map(|block| { - BlockFinality::NonFinal(EthereumBlockWithCalls { + .await?; + + let Some((json_value, block_ptr)) = cached else { + return Ok(None); + }; + + match json::from_value::(json_value.clone()) { + Ok(block) => Ok(Some(BlockFinality::NonFinal(EthereumBlockWithCalls { ethereum_block: block, calls: None, - }) - })) + }))), + Err(e) => { + warn!( + self.logger, + "Failed to deserialize cached ancestor block {} (offset {} from {}): {}. \ + This may indicate stale cache data from a previous version. \ + Falling back to Firehose/RPC.", + block_ptr.hash_hex(), + offset, + ptr_for_log.hash_hex(), + e + ); + + match self.chain_client.as_ref() { + ChainClient::Firehose(endpoints) => { + let block = self + .fetch_block_with_firehose(endpoints, &block_ptr) + .await?; + let ethereum_block: EthereumBlockWithCalls = (&block).try_into()?; + Ok(Some(BlockFinality::NonFinal(ethereum_block))) + } + ChainClient::Rpc(adapters) => { + match self + .fetch_light_block_with_rpc(adapters, &block_ptr) + .await? + { + Some(light_block) => { + let ethereum_block = EthereumBlock { + block: light_block, + transaction_receipts: vec![], + }; + Ok(Some(BlockFinality::NonFinal(EthereumBlockWithCalls { + ethereum_block, + calls: None, + }))) + } + None => Ok(None), + } + } + } + } + } } async fn parent_ptr(&self, block: &BlockPtr) -> Result, Error> { @@ -1093,52 +1123,70 @@ impl TriggersAdapterTrait for TriggersAdapter { } // If not in store, fetch from Firehose - let endpoint = endpoints.endpoint().await?; - let logger = self.logger.clone(); - let retry_log_message = - format!("get_block_by_ptr for block {} with firehose", block); - let block = block.clone(); - - retry(retry_log_message, &logger) - .limit(ENV_VARS.request_retries) - .timeout_secs(ENV_VARS.json_rpc_timeout.as_secs()) - .run(move || { - let endpoint = endpoint.cheap_clone(); - let logger = logger.cheap_clone(); - let block = block.clone(); - async move { - endpoint - .get_block_by_ptr::(&block, &logger) - .await - .context(format!( - "Failed to fetch block by ptr {} from firehose", - block - )) - } - }) + self.fetch_block_with_firehose(endpoints, block) .await? .parent_ptr() } - ChainClient::Rpc(adapters) => { - let blocks = adapters - .cheapest_with(&self.capabilities) - .await? - .load_blocks( - self.logger.cheap_clone(), - self.chain_store.cheap_clone(), - HashSet::from_iter(Some(block.hash.as_b256())), - ) - .await?; - assert_eq!(blocks.len(), 1); - - blocks[0].parent_ptr() - } + ChainClient::Rpc(adapters) => self + .fetch_light_block_with_rpc(adapters, block) + .await? + .expect("block must exist for parent_ptr") + .parent_ptr(), }; Ok(block) } } +impl TriggersAdapter { + async fn fetch_block_with_firehose( + &self, + endpoints: &FirehoseEndpoints, + block_ptr: &BlockPtr, + ) -> Result { + let endpoint = endpoints.endpoint().await?; + let logger = self.logger.clone(); + let retry_log_message = format!("fetch_block_with_firehose {}", block_ptr); + let block_ptr = block_ptr.clone(); + + let block = retry(retry_log_message, &logger) + .limit(ENV_VARS.request_retries) + .timeout_secs(ENV_VARS.json_rpc_timeout.as_secs()) + .run(move || { + let endpoint = endpoint.cheap_clone(); + let logger = logger.cheap_clone(); + let block_ptr = block_ptr.clone(); + async move { + endpoint + .get_block_by_ptr::(&block_ptr, &logger) + .await + .context(format!("Failed to fetch block {} from firehose", block_ptr)) + } + }) + .await?; + + Ok(block) + } + + async fn fetch_light_block_with_rpc( + &self, + adapters: &EthereumNetworkAdapters, + block_ptr: &BlockPtr, + ) -> Result>, Error> { + let blocks = adapters + .cheapest_with(&self.capabilities) + .await? + .load_blocks( + self.logger.cheap_clone(), + self.chain_store.cheap_clone(), + HashSet::from_iter(Some(block_ptr.hash.as_b256())), + ) + .await?; + + Ok(blocks.into_iter().next()) + } +} + pub struct FirehoseMapper { adapter: Arc>, filter: Arc, diff --git a/chain/ethereum/src/codec.rs b/chain/ethereum/src/codec.rs index 2a402d4f5b2..f7d1af103bc 100644 --- a/chain/ethereum/src/codec.rs +++ b/chain/ethereum/src/codec.rs @@ -268,19 +268,21 @@ impl<'a> TryInto> for TransactionTraceAt<'a> { .trace .access_list .iter() - .map(|access_tuple| { - let address = Address::from_slice(&access_tuple.address); + .map(|access_tuple| -> Result<_, Error> { + let address = access_tuple + .address + .try_decode_proto("access tuple address")?; let storage_keys = access_tuple .storage_keys .iter() - .map(|key| B256::from_slice(key)) - .collect(); - AccessListItem { + .map(|key| key.try_decode_proto("storage key")) + .collect::, _>>()?; + Ok(AccessListItem { address, storage_keys, - } + }) }) - .collect::>() + .collect::, Error>>()? .into(); // Extract actual signature components from trace @@ -359,8 +361,8 @@ impl<'a> TryInto> for TransactionTraceAt<'a> { .trace .blob_hashes .iter() - .map(|hash| B256::from_slice(hash)) - .collect(); + .map(|hash| hash.try_decode_proto("blob hash")) + .collect::, _>>()?; let max_fee_per_blob_gas_u128 = self.trace.blob_gas_fee_cap.as_ref().map_or(0u128, |x| { @@ -401,10 +403,10 @@ impl<'a> TryInto> for TransactionTraceAt<'a> { .trace .set_code_authorizations .iter() - .map(|auth| { + .map(|auth| -> Result<_, Error> { let inner = alloy::eips::eip7702::Authorization { chain_id: U256::from_be_slice(&auth.chain_id), - address: Address::from_slice(&auth.address), + address: auth.address.try_decode_proto("authorization address")?, nonce: auth.nonce, }; @@ -412,11 +414,11 @@ impl<'a> TryInto> for TransactionTraceAt<'a> { let s = U256::from_be_slice(&auth.s); let y_parity = auth.v as u8; - alloy::eips::eip7702::SignedAuthorization::new_unchecked( + Ok(alloy::eips::eip7702::SignedAuthorization::new_unchecked( inner, y_parity, r, s, - ) + )) }) - .collect(); + .collect::, Error>>()?; let tx = TxEip7702 { // Firehose protobuf doesn't provide chain_id for transactions. diff --git a/graph/src/ipfs/server_address.rs b/graph/src/ipfs/server_address.rs index c7c8bc109f6..556997406ef 100644 --- a/graph/src/ipfs/server_address.rs +++ b/graph/src/ipfs/server_address.rs @@ -119,7 +119,7 @@ mod tests { assert_eq!( err.to_string(), - "'https://' is not a valid IPFS server address: invalid format", + "'https://' is not a valid IPFS server address: empty string", ); } From 71095309a613392d663165d2ed3c419ca704c3dd Mon Sep 17 00:00:00 2001 From: incrypto32 Date: Sun, 28 Dec 2025 12:27:27 +0400 Subject: [PATCH 3/8] chain/ethereum: Fix ABI conversion issues in alloy migration --- chain/ethereum/src/runtime/abi.rs | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/chain/ethereum/src/runtime/abi.rs b/chain/ethereum/src/runtime/abi.rs index 5641a501a6e..fadac985f42 100644 --- a/chain/ethereum/src/runtime/abi.rs +++ b/chain/ethereum/src/runtime/abi.rs @@ -508,10 +508,7 @@ impl<'a> ToAscObj for EthereumTransactionData<'a> hash: asc_new(heap, &self.hash(), gas).await?, index: asc_new(heap, &BigInt::from(self.index()), gas).await?, from: asc_new(heap, &self.from(), gas).await?, - to: match self.to() { - Some(to) => asc_new(heap, &to, gas).await?, - None => AscPtr::null(), - }, + to: asc_new_or_null(heap, &self.to(), gas).await?, value: asc_new(heap, &BigInt::from_unsigned_u256(&self.value()), gas).await?, gas_limit: asc_new(heap, &BigInt::from(self.gas_limit()), gas).await?, gas_price: asc_new(heap, &BigInt::from(self.gas_price()), gas).await?, @@ -530,10 +527,7 @@ impl<'a> ToAscObj for EthereumTransactionData<'a> hash: asc_new(heap, &self.hash(), gas).await?, index: asc_new(heap, &BigInt::from(self.index()), gas).await?, from: asc_new(heap, &self.from(), gas).await?, - to: match self.to() { - Some(to) => asc_new(heap, &to, gas).await?, - None => AscPtr::null(), - }, + to: asc_new_or_null(heap, &self.to(), gas).await?, value: asc_new(heap, &BigInt::from_unsigned_u256(&self.value()), gas).await?, gas_limit: asc_new(heap, &BigInt::from(self.gas_limit()), gas).await?, gas_price: asc_new(heap, &BigInt::from(self.gas_price()), gas).await?, @@ -675,8 +669,8 @@ impl ToAscObj for Log { transaction_hash: asc_new_or_null(heap, &self.transaction_hash, gas).await?, transaction_index: asc_new_or_null_u64(heap, &self.transaction_index, gas).await?, log_index: asc_new_or_null_u64(heap, &self.log_index, gas).await?, - transaction_log_index: AscPtr::null(), // TODO(alloy): figure out how to get transaction log index - log_type: AscPtr::null(), // TODO(alloy): figure out how to get log type + transaction_log_index: AscPtr::null(), // Non-standard field, not available in alloy + log_type: AscPtr::null(), // Non-standard field, not available in alloy removed: asc_new( heap, &AscWrapped { @@ -708,7 +702,8 @@ impl ToAscObj transaction_index: asc_new(heap, &BigInt::from(transaction_index), gas).await?, block_hash: asc_new_or_null(heap, &self.block_hash, gas).await?, block_number: asc_new_or_null_u64(heap, &self.block_number, gas).await?, - cumulative_gas_used: asc_new(heap, &BigInt::from(self.gas_used), gas).await?, + cumulative_gas_used: asc_new(heap, &BigInt::from(self.cumulative_gas_used()), gas) + .await?, gas_used: asc_new(heap, &BigInt::from(self.gas_used), gas).await?, contract_address: asc_new_or_null(heap, &self.contract_address, gas).await?, logs: asc_new(heap, &self.logs(), gas).await?, From ade80f1aaa9b5d89dec6372b91fd83a766406321 Mon Sep 17 00:00:00 2001 From: incrypto32 Date: Sun, 28 Dec 2025 13:04:21 +0400 Subject: [PATCH 4/8] store, chain/ethereum: Handle pre-Byzantium receipt status correctly --- chain/ethereum/src/runtime/abi.rs | 7 ++++++- store/postgres/src/transaction_receipt.rs | 15 ++++----------- 2 files changed, 10 insertions(+), 12 deletions(-) diff --git a/chain/ethereum/src/runtime/abi.rs b/chain/ethereum/src/runtime/abi.rs index fadac985f42..75717106ac6 100644 --- a/chain/ethereum/src/runtime/abi.rs +++ b/chain/ethereum/src/runtime/abi.rs @@ -6,6 +6,7 @@ use anyhow::anyhow; use async_trait::async_trait; use graph::abi; use graph::prelude::alloy; +use graph::prelude::alloy::consensus::TxReceipt; use graph::prelude::alloy::network::ReceiptResponse; use graph::prelude::alloy::rpc::types::{Log, TransactionReceipt}; use graph::prelude::alloy::serde::WithOtherFields; @@ -697,6 +698,10 @@ impl ToAscObj .ok_or(HostExportError::Unknown(anyhow!( "Transaction index is missing" )))?; + let status = match self.inner.status_or_post_state().as_eip658() { + Some(success) => asc_new(heap, &BigInt::from(success as u64), gas).await?, + None => AscPtr::null(), // Pre-EIP-658 (pre-Byzantium) receipt + }; Ok(AscEthereumTransactionReceipt { transaction_hash: asc_new(heap, &self.transaction_hash, gas).await?, transaction_index: asc_new(heap, &BigInt::from(transaction_index), gas).await?, @@ -707,7 +712,7 @@ impl ToAscObj gas_used: asc_new(heap, &BigInt::from(self.gas_used), gas).await?, contract_address: asc_new_or_null(heap, &self.contract_address, gas).await?, logs: asc_new(heap, &self.logs(), gas).await?, - status: asc_new(heap, &BigInt::from(self.status() as u64), gas).await?, + status, root: asc_new_or_null(heap, &self.state_root(), gas).await?, logs_bloom: asc_new(heap, self.inner.bloom().as_slice(), gas).await?, }) diff --git a/store/postgres/src/transaction_receipt.rs b/store/postgres/src/transaction_receipt.rs index 73b11c9c400..1177422f42b 100644 --- a/store/postgres/src/transaction_receipt.rs +++ b/store/postgres/src/transaction_receipt.rs @@ -45,18 +45,11 @@ impl TryFrom for LightTransactionReceipt { let block_number = block_number.map(u64::from_be_bytes); let gas_used = gas_used.map(u64::from_be_bytes).unwrap_or(0); - // Handle both old U64 format and new boolean format + // Status is non-zero for success, zero for failure. Works for any byte length. + // Defaults to true for pre-Byzantium receipts (no status field), consistent with alloy. let status = status - .map(|bytes| { - match bytes.len() { - 1 => bytes[0] != 0, // New format: single byte - 8 => { - u64::from_be_bytes(drain_vector::<8>(bytes.to_vec()).unwrap_or([0; 8])) != 0 - } // Old format: U64 - _ => false, // Fallback - } - }) - .unwrap_or(false); + .map(|bytes| bytes.iter().any(|&b| b != 0)) + .unwrap_or(true); Ok(LightTransactionReceipt { transaction_hash: transaction_hash.into(), From 1492eda573b725e8115e0b74608ffcbc12277c44 Mon Sep 17 00:00:00 2001 From: incrypto32 Date: Sat, 27 Dec 2025 13:04:23 +0400 Subject: [PATCH 5/8] graph: Normalize ABI JSON to handle undefined stateMutability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Some ABIs contain "undefined" as a stateMutability value, which is not part of the official Solidity ABI specification. The spec only defines four valid values: pure, view, nonpayable, and payable. Alloy's StateMutability enum strictly follows the spec and rejects "undefined" during deserialization, causing subgraph deployment failures with: unknown variant `undefined`, expected one of `pure`, `view`, `nonpayable`, `payable` This adds a normalize_abi_json() function that preprocesses ABI JSON before deserialization, replacing "undefined" with "nonpayable" (the default state mutability). This handles non-compliant ABIs gracefully while maintaining spec compliance. Also adds a unit test to verify the normalization works correctly. 🤖 Generated with Claude Code Co-Authored-By: Claude --- graph/src/data_source/common.rs | 60 +++++++++++++++++++++++++++++++-- 1 file changed, 58 insertions(+), 2 deletions(-) diff --git a/graph/src/data_source/common.rs b/graph/src/data_source/common.rs index 8090f9334b7..206de237f57 100644 --- a/graph/src/data_source/common.rs +++ b/graph/src/data_source/common.rs @@ -24,6 +24,26 @@ use slog::Logger; use std::collections::HashMap; use std::{str::FromStr, sync::Arc}; +fn normalize_abi_json(json_bytes: &[u8]) -> Result, anyhow::Error> { + let mut value: serde_json::Value = serde_json::from_slice(json_bytes)?; + + if let Some(array) = value.as_array_mut() { + for item in array { + if let Some(obj) = item.as_object_mut() { + if let Some(state_mutability) = obj.get_mut("stateMutability") { + if let Some(s) = state_mutability.as_str() { + if s == "undefined" { + *state_mutability = serde_json::Value::String("nonpayable".to_string()); + } + } + } + } + } + } + + Ok(serde_json::to_vec(&value)?) +} + #[derive(Clone, Debug, PartialEq)] pub struct MappingABI { pub name: String, @@ -364,11 +384,14 @@ impl UnresolvedMappingABI { self.name, self.file.link ) })?; - let contract = serde_json::from_slice(&*contract_bytes) + let normalized_bytes = normalize_abi_json(&contract_bytes) + .with_context(|| format!("failed to normalize ABI JSON for {}", self.name))?; + + let contract = serde_json::from_slice(&normalized_bytes) .with_context(|| format!("failed to load ABI {}", self.name))?; // Parse ABI JSON for on-demand struct field extraction - let abi_json = AbiJson::new(&contract_bytes) + let abi_json = AbiJson::new(&normalized_bytes) .with_context(|| format!("Failed to parse ABI JSON for {}", self.name))?; Ok(( @@ -2103,6 +2126,39 @@ mod tests { assert!(error_msg.contains("is not a struct")); } + #[test] + fn test_normalize_abi_json_with_undefined_state_mutability() { + let abi_with_undefined = r#"[ + { + "type": "function", + "name": "testFunction", + "inputs": [], + "outputs": [], + "stateMutability": "undefined" + }, + { + "type": "function", + "name": "normalFunction", + "inputs": [], + "outputs": [], + "stateMutability": "view" + } + ]"#; + + let normalized = normalize_abi_json(abi_with_undefined.as_bytes()).unwrap(); + let result: serde_json::Value = serde_json::from_slice(&normalized).unwrap(); + + if let Some(array) = result.as_array() { + assert_eq!(array[0]["stateMutability"], "nonpayable"); + assert_eq!(array[1]["stateMutability"], "view"); + } else { + panic!("Expected JSON array"); + } + + let json_abi: abi::JsonAbi = serde_json::from_slice(&normalized).unwrap(); + assert_eq!(json_abi.len(), 2); + } + // Helper function to create consistent test ABI fn create_test_mapping_abi() -> AbiJson { const ABI_JSON: &str = r#"[ From 928b0eb46a651a8566ccfcf679c28b56968f2839 Mon Sep 17 00:00:00 2001 From: incrypto32 Date: Sat, 27 Dec 2025 13:49:28 +0400 Subject: [PATCH 6/8] graph: Handle duplicate constructors in ABIs Extends normalize_abi_json() to also remove duplicate constructors from ABIs. Some non-compliant ABIs contain multiple constructor entries (e.g., DolomiteMargin ABI has two constructors, likely from incorrectly merged contract ABIs). Alloy's JsonAbi only allows one constructor and fails with 'duplicate field self.constructor' when encountering duplicates. ethabi's Contract type silently handled this by only storing one constructor (the last one encountered during deserialization). The fix keeps only the first constructor and removes any subsequent ones, matching the Solidity spec that a contract can only have one constructor. Fixes subgraph QmacPbft3reGGGL4VBzrZCKHeLpRgU9X2wUJjvPBVweyRV deployment. --- graph/src/data_source/common.rs | 57 ++++++++++++++++++++++++++++++++- 1 file changed, 56 insertions(+), 1 deletion(-) diff --git a/graph/src/data_source/common.rs b/graph/src/data_source/common.rs index 206de237f57..8f311f83738 100644 --- a/graph/src/data_source/common.rs +++ b/graph/src/data_source/common.rs @@ -28,7 +28,10 @@ fn normalize_abi_json(json_bytes: &[u8]) -> Result, anyhow::Error> { let mut value: serde_json::Value = serde_json::from_slice(json_bytes)?; if let Some(array) = value.as_array_mut() { - for item in array { + let mut found_constructor = false; + let mut indices_to_remove = Vec::new(); + + for (index, item) in array.iter_mut().enumerate() { if let Some(obj) = item.as_object_mut() { if let Some(state_mutability) = obj.get_mut("stateMutability") { if let Some(s) = state_mutability.as_str() { @@ -37,8 +40,22 @@ fn normalize_abi_json(json_bytes: &[u8]) -> Result, anyhow::Error> { } } } + + if let Some(item_type) = obj.get("type") { + if item_type == "constructor" { + if found_constructor { + indices_to_remove.push(index); + } else { + found_constructor = true; + } + } + } } } + + for index in indices_to_remove.iter().rev() { + array.remove(*index); + } } Ok(serde_json::to_vec(&value)?) @@ -2159,6 +2176,44 @@ mod tests { assert_eq!(json_abi.len(), 2); } + #[test] + fn test_normalize_abi_json_with_duplicate_constructors() { + let abi_with_duplicate_constructors = r#"[ + { + "type": "constructor", + "inputs": [{"name": "param1", "type": "address"}], + "stateMutability": "nonpayable" + }, + { + "type": "function", + "name": "someFunction", + "inputs": [], + "outputs": [], + "stateMutability": "view" + }, + { + "type": "constructor", + "inputs": [{"name": "param2", "type": "uint256"}], + "stateMutability": "nonpayable" + } + ]"#; + + let normalized = normalize_abi_json(abi_with_duplicate_constructors.as_bytes()).unwrap(); + let result: serde_json::Value = serde_json::from_slice(&normalized).unwrap(); + + if let Some(array) = result.as_array() { + assert_eq!(array.len(), 2); + assert_eq!(array[0]["type"], "constructor"); + assert_eq!(array[0]["inputs"][0]["name"], "param1"); + assert_eq!(array[1]["type"], "function"); + } else { + panic!("Expected JSON array"); + } + + let json_abi: abi::JsonAbi = serde_json::from_slice(&normalized).unwrap(); + assert_eq!(json_abi.len(), 2); + } + // Helper function to create consistent test ABI fn create_test_mapping_abi() -> AbiJson { const ABI_JSON: &str = r#"[ From 500df8246cb691f1ea5abd2cc0a97ebe0b6b9ee1 Mon Sep 17 00:00:00 2001 From: incrypto32 Date: Wed, 7 Jan 2026 14:08:28 +0400 Subject: [PATCH 7/8] extend normalizer to handle all alloy parsing incompatibilities --- graph/src/data_source/common.rs | 124 +++++++++++++++++++++++++++++--- 1 file changed, 116 insertions(+), 8 deletions(-) diff --git a/graph/src/data_source/common.rs b/graph/src/data_source/common.rs index 8f311f83738..6149742c3ee 100644 --- a/graph/src/data_source/common.rs +++ b/graph/src/data_source/common.rs @@ -24,11 +24,33 @@ use slog::Logger; use std::collections::HashMap; use std::{str::FromStr, sync::Arc}; +/// Normalizes ABI JSON to handle compatibility issues between the legacy `ethabi`/`rust-web3` +/// parser and the stricter `alloy` parser. +/// +/// Some deployed subgraph ABIs contain non-standard constructs that `ethabi` accepted but +/// `alloy` rejects. This function patches these issues to maintain backward compatibility: +/// +/// 1. **`stateMutability: "undefined"`** - Some ABIs use "undefined" which is not a valid +/// Solidity state mutability. We replace it with "nonpayable". +/// +/// 2. **Duplicate constructors** - Some ABIs contain multiple constructor definitions. +/// We keep only the first one. +/// +/// 3. **Duplicate fallback functions** - Similar to constructors, some ABIs have multiple +/// fallback definitions. We keep only the first one. +/// +/// 4. **`indexed` field in non-event params** - The `indexed` field is only valid for event +/// parameters, but some ABIs include it on function inputs/outputs. We strip it from +/// non-event items. +/// +/// These issues were identified by validating ABIs across deployed subgraphs in production +/// before the migration to alloy. fn normalize_abi_json(json_bytes: &[u8]) -> Result, anyhow::Error> { let mut value: serde_json::Value = serde_json::from_slice(json_bytes)?; if let Some(array) = value.as_array_mut() { let mut found_constructor = false; + let mut found_fallback = false; let mut indices_to_remove = Vec::new(); for (index, item) in array.iter_mut().enumerate() { @@ -41,14 +63,19 @@ fn normalize_abi_json(json_bytes: &[u8]) -> Result, anyhow::Error> { } } - if let Some(item_type) = obj.get("type") { - if item_type == "constructor" { - if found_constructor { - indices_to_remove.push(index); - } else { - found_constructor = true; - } - } + let item_type = obj.get("type").and_then(|t| t.as_str()); + + match item_type { + Some("constructor") if found_constructor => indices_to_remove.push(index), + Some("constructor") => found_constructor = true, + Some("fallback") if found_fallback => indices_to_remove.push(index), + Some("fallback") => found_fallback = true, + _ => {} + } + + if item_type != Some("event") { + strip_indexed_from_params(obj.get_mut("inputs")); + strip_indexed_from_params(obj.get_mut("outputs")); } } } @@ -61,6 +88,16 @@ fn normalize_abi_json(json_bytes: &[u8]) -> Result, anyhow::Error> { Ok(serde_json::to_vec(&value)?) } +fn strip_indexed_from_params(params: Option<&mut serde_json::Value>) { + if let Some(serde_json::Value::Array(arr)) = params { + for param in arr.iter_mut() { + if let Some(obj) = param.as_object_mut() { + obj.remove("indexed"); + } + } + } +} + #[derive(Clone, Debug, PartialEq)] pub struct MappingABI { pub name: String, @@ -401,6 +438,8 @@ impl UnresolvedMappingABI { self.name, self.file.link ) })?; + // Normalize the ABI to handle compatibility issues between ethabi and alloy parsers. + // See `normalize_abi_json` for details on the specific issues being addressed. let normalized_bytes = normalize_abi_json(&contract_bytes) .with_context(|| format!("failed to normalize ABI JSON for {}", self.name))?; @@ -2214,6 +2253,75 @@ mod tests { assert_eq!(json_abi.len(), 2); } + #[test] + fn test_normalize_abi_json_with_duplicate_fallbacks() { + let abi_with_duplicate_fallbacks = r#"[ + { + "type": "fallback", + "stateMutability": "payable" + }, + { + "type": "function", + "name": "someFunction", + "inputs": [], + "outputs": [], + "stateMutability": "view" + }, + { + "type": "fallback", + "stateMutability": "nonpayable" + } + ]"#; + + let normalized = normalize_abi_json(abi_with_duplicate_fallbacks.as_bytes()).unwrap(); + let result: serde_json::Value = serde_json::from_slice(&normalized).unwrap(); + + if let Some(array) = result.as_array() { + assert_eq!(array.len(), 2); + assert_eq!(array[0]["type"], "fallback"); + assert_eq!(array[0]["stateMutability"], "payable"); + assert_eq!(array[1]["type"], "function"); + } else { + panic!("Expected JSON array"); + } + + let json_abi: abi::JsonAbi = serde_json::from_slice(&normalized).unwrap(); + assert_eq!(json_abi.len(), 2); + } + + #[test] + fn test_normalize_abi_json_strips_indexed_from_non_events() { + let abi_with_indexed_in_function = r#"[ + { + "type": "function", + "name": "testFunction", + "inputs": [{"name": "x", "type": "uint256", "indexed": true}], + "outputs": [{"name": "y", "type": "address", "indexed": false}], + "stateMutability": "view" + }, + { + "type": "event", + "name": "TestEvent", + "anonymous": false, + "inputs": [{"name": "from", "type": "address", "indexed": true}] + } + ]"#; + + let normalized = normalize_abi_json(abi_with_indexed_in_function.as_bytes()).unwrap(); + let result: serde_json::Value = serde_json::from_slice(&normalized).unwrap(); + + if let Some(array) = result.as_array() { + assert!(array[0]["inputs"][0].get("indexed").is_none()); + assert!(array[0]["outputs"][0].get("indexed").is_none()); + assert_eq!(array[1]["inputs"][0]["indexed"], true); + } else { + panic!("Expected JSON array"); + } + + let json_abi: abi::JsonAbi = serde_json::from_slice(&normalized).unwrap(); + assert_eq!(json_abi.len(), 2); + } + // Helper function to create consistent test ABI fn create_test_mapping_abi() -> AbiJson { const ABI_JSON: &str = r#"[ From ee48083fdb1832fe97971d18a919c30f9782d525 Mon Sep 17 00:00:00 2001 From: Krishnanand V P <44740264+incrypto32@users.noreply.github.com> Date: Wed, 14 Jan 2026 14:05:06 +0400 Subject: [PATCH 8/8] Add fallback for ancestor_block to handle cache truncation (#6262) * ethereum: Add RPC fallback for ancestor_block to handle cache truncation * ethereum: Extract walk_back_ancestor logic and add tests * ethereum: Fetch full blocks with receipts in ancestor_block RPC fallback When ancestor_block falls back to RPC due to cache deserialization failures, fetch full blocks with receipts instead of light blocks. Empty receipts caused missing block handler call triggers and log triggers. --- chain/ethereum/src/chain.rs | 296 ++++++++++++++++++++++++++++++------ 1 file changed, 253 insertions(+), 43 deletions(-) diff --git a/chain/ethereum/src/chain.rs b/chain/ethereum/src/chain.rs index 4624675efb8..9114e136950 100644 --- a/chain/ethereum/src/chain.rs +++ b/chain/ethereum/src/chain.rs @@ -355,6 +355,38 @@ impl std::fmt::Debug for Chain { } } +/// Walk back from a block pointer by following parent pointers. +/// This is the core logic used as a fallback when the cache doesn't have ancestor block. +/// +async fn walk_back_ancestor( + start_ptr: BlockPtr, + offset: BlockNumber, + root: Option, + mut parent_getter: F, +) -> Result, E> +where + F: FnMut(BlockPtr) -> Fut, + Fut: std::future::Future, E>>, +{ + let mut current_ptr = start_ptr; + + for _ in 0..offset { + match parent_getter(current_ptr.clone()).await? { + Some(parent) => { + if let Some(root_hash) = &root { + if parent.hash == *root_hash { + break; + } + } + current_ptr = parent; + } + None => return Ok(None), + } + } + + Ok(Some(current_ptr)) +} + impl Chain { /// Creates a new Ethereum [`Chain`]. pub fn new( @@ -1030,6 +1062,20 @@ impl TriggersAdapterTrait for TriggersAdapter { } } + // Find an ancestor block at the specified offset from the given block pointer. + // Primarily used for reorg detection to verify if the indexed position remains + // on the main chain. + // + // Parameters: + // - ptr: Starting block pointer from which to walk backwards (typically the chain head) + // - offset: Number of blocks to traverse backwards (0 returns ptr, 1 returns parent, etc.) + // - root: Optional block hash that serves as a boundary for traversal. This is ESSENTIAL + // for chains with skipped blocks (e.g., Filecoin EVM) where block numbers are not + // consecutive. When provided, traversal stops upon reaching the child of root, + // ensuring correct ancestor relationships even with gaps in block numbers. + // + // The function attempts to use the database cache first for performance, + // with RPC fallback implemented to handle cases where the cache is unavailable. async fn ancestor_block( &self, ptr: BlockPtr, @@ -1040,56 +1086,83 @@ impl TriggersAdapterTrait for TriggersAdapter { let cached = self .chain_store .cheap_clone() - .ancestor_block(ptr, offset, root) + .ancestor_block(ptr.clone(), offset, root.clone()) .await?; - let Some((json_value, block_ptr)) = cached else { - return Ok(None); - }; - - match json::from_value::(json_value.clone()) { - Ok(block) => Ok(Some(BlockFinality::NonFinal(EthereumBlockWithCalls { - ethereum_block: block, - calls: None, - }))), - Err(e) => { - warn!( + // First check if we have the ancestor in cache and can deserialize it + let block_ptr = match cached { + Some((json, ptr)) => { + // Try to deserialize the cached block + match json::from_value::(json.clone()) { + Ok(block) => { + // Successfully cached and deserialized + return Ok(Some(BlockFinality::NonFinal(EthereumBlockWithCalls { + ethereum_block: block, + calls: None, + }))); + } + Err(e) => { + // Cache hit but deserialization failed + warn!( + self.logger, + "Failed to deserialize cached ancestor block {} (offset {} from {}): {}. \ + This may indicate stale cache data from a previous version. \ + Falling back to Firehose/RPC.", + ptr.hash_hex(), + offset, + ptr_for_log.hash_hex(), + e + ); + ptr + } + } + } + None => { + // Cache miss - fall back to walking the chain via parent_ptr() calls. + // This provides resilience when the block cache is empty (e.g., after truncation). + debug!( self.logger, - "Failed to deserialize cached ancestor block {} (offset {} from {}): {}. \ - This may indicate stale cache data from a previous version. \ - Falling back to Firehose/RPC.", - block_ptr.hash_hex(), - offset, + "ancestor_block cache miss for {} at offset {}, walking back via parent_ptr", ptr_for_log.hash_hex(), - e + offset ); - match self.chain_client.as_ref() { - ChainClient::Firehose(endpoints) => { - let block = self - .fetch_block_with_firehose(endpoints, &block_ptr) - .await?; - let ethereum_block: EthereumBlockWithCalls = (&block).try_into()?; - Ok(Some(BlockFinality::NonFinal(ethereum_block))) - } - ChainClient::Rpc(adapters) => { - match self - .fetch_light_block_with_rpc(adapters, &block_ptr) - .await? - { - Some(light_block) => { - let ethereum_block = EthereumBlock { - block: light_block, - transaction_receipts: vec![], - }; - Ok(Some(BlockFinality::NonFinal(EthereumBlockWithCalls { - ethereum_block, - calls: None, - }))) - } - None => Ok(None), - } + match walk_back_ancestor( + ptr.clone(), + offset, + root.clone(), + |block_ptr| async move { self.parent_ptr(&block_ptr).await }, + ) + .await? + { + Some(ptr) => ptr, + None => return Ok(None), + } + } + }; + + // Fetch the actual block data for the identified block pointer. + // This path is taken for both cache misses and deserialization failures. + match self.chain_client.as_ref() { + ChainClient::Firehose(endpoints) => { + let block = self + .fetch_block_with_firehose(endpoints, &block_ptr) + .await?; + let ethereum_block: EthereumBlockWithCalls = (&block).try_into()?; + Ok(Some(BlockFinality::NonFinal(ethereum_block))) + } + ChainClient::Rpc(adapters) => { + match self + .fetch_full_block_with_rpc(adapters, &block_ptr) + .await? + { + Some(ethereum_block) => { + Ok(Some(BlockFinality::NonFinal(EthereumBlockWithCalls { + ethereum_block, + calls: None, + }))) } + None => Ok(None), } } } @@ -1185,6 +1258,29 @@ impl TriggersAdapter { Ok(blocks.into_iter().next()) } + + async fn fetch_full_block_with_rpc( + &self, + adapters: &EthereumNetworkAdapters, + block_ptr: &BlockPtr, + ) -> Result, Error> { + let adapter = adapters.cheapest_with(&self.capabilities).await?; + + let block = adapter + .block_by_hash(&self.logger, block_ptr.hash.as_b256()) + .await?; + + match block { + Some(block) => { + let ethereum_block = adapter + .load_full_block(&self.logger, block) + .await + .map_err(|e| anyhow!("Failed to load full block: {}", e))?; + Ok(Some(ethereum_block)) + } + None => Ok(None), + } + } } pub struct FirehoseMapper { @@ -1461,4 +1557,118 @@ mod tests { assert!(missing.contains(&2)); assert!(missing.contains(&3)); } + + #[tokio::test] + async fn test_walk_back_ancestor() { + use std::collections::HashMap; + + let block_100_hash = BlockHash("block100".as_bytes().to_vec().into_boxed_slice()); + let block_101_hash = BlockHash("block101".as_bytes().to_vec().into_boxed_slice()); + let block_102_hash = BlockHash("block102".as_bytes().to_vec().into_boxed_slice()); + let block_103_hash = BlockHash("block103".as_bytes().to_vec().into_boxed_slice()); + let block_104_hash = BlockHash("block104".as_bytes().to_vec().into_boxed_slice()); + let block_105_hash = BlockHash("block105".as_bytes().to_vec().into_boxed_slice()); + + let block_105 = BlockPtr::new(block_105_hash.clone(), 105); + let block_104 = BlockPtr::new(block_104_hash.clone(), 104); + let block_103 = BlockPtr::new(block_103_hash.clone(), 103); + let block_102 = BlockPtr::new(block_102_hash.clone(), 102); + let block_101 = BlockPtr::new(block_101_hash.clone(), 101); + let block_100 = BlockPtr::new(block_100_hash.clone(), 100); + + let mut parent_map = HashMap::new(); + parent_map.insert(block_105_hash.clone(), block_104.clone()); + parent_map.insert(block_104_hash.clone(), block_103.clone()); + parent_map.insert(block_103_hash.clone(), block_102.clone()); + parent_map.insert(block_102_hash.clone(), block_101.clone()); + parent_map.insert(block_101_hash.clone(), block_100.clone()); + + let result = super::walk_back_ancestor(block_105.clone(), 2, None, |block_ptr| { + let parent = parent_map.get(&block_ptr.hash).cloned(); + async move { Ok::<_, std::convert::Infallible>(parent) } + }) + .await + .unwrap(); + assert_eq!(result, Some(block_103.clone())); + + let result = super::walk_back_ancestor( + block_105.clone(), + 10, + Some(block_102_hash.clone()), + |block_ptr| { + let parent = parent_map.get(&block_ptr.hash).cloned(); + async move { Ok::<_, std::convert::Infallible>(parent) } + }, + ) + .await + .unwrap(); + assert_eq!( + result, + Some(block_103.clone()), + "Should stop at child of root" + ); + } + + #[tokio::test] + async fn test_walk_back_ancestor_skipped_blocks_with_root() { + use std::collections::HashMap; + + let block_100_hash = BlockHash("block100".as_bytes().to_vec().into_boxed_slice()); + let block_101_hash = BlockHash("block101".as_bytes().to_vec().into_boxed_slice()); + let block_102_hash = BlockHash("block102".as_bytes().to_vec().into_boxed_slice()); + let block_110_hash = BlockHash("block110".as_bytes().to_vec().into_boxed_slice()); + let block_111_hash = BlockHash("block111".as_bytes().to_vec().into_boxed_slice()); + let block_112_hash = BlockHash("block112".as_bytes().to_vec().into_boxed_slice()); + let block_120_hash = BlockHash("block120".as_bytes().to_vec().into_boxed_slice()); + + let block_120 = BlockPtr::new(block_120_hash.clone(), 120); + let block_112 = BlockPtr::new(block_112_hash.clone(), 112); + let block_111 = BlockPtr::new(block_111_hash.clone(), 111); + let block_110 = BlockPtr::new(block_110_hash.clone(), 110); + let block_102 = BlockPtr::new(block_102_hash.clone(), 102); + let block_101 = BlockPtr::new(block_101_hash.clone(), 101); + let block_100 = BlockPtr::new(block_100_hash.clone(), 100); + + let mut parent_map = HashMap::new(); + parent_map.insert(block_120_hash.clone(), block_112.clone()); + parent_map.insert(block_112_hash.clone(), block_111.clone()); + parent_map.insert(block_111_hash.clone(), block_110.clone()); + parent_map.insert(block_110_hash.clone(), block_102.clone()); + parent_map.insert(block_102_hash.clone(), block_101.clone()); + parent_map.insert(block_101_hash.clone(), block_100.clone()); + + let result = super::walk_back_ancestor( + block_120.clone(), + 10, + Some(block_110_hash.clone()), + |block_ptr| { + let parent = parent_map.get(&block_ptr.hash).cloned(); + async move { Ok::<_, std::convert::Infallible>(parent) } + }, + ) + .await + .unwrap(); + assert_eq!( + result, + Some(block_111.clone()), + "root=110: should stop at 111 (child of root)" + ); + + let result = super::walk_back_ancestor( + block_120.clone(), + 10, + Some(block_101_hash.clone()), + |block_ptr| { + let parent = parent_map.get(&block_ptr.hash).cloned(); + async move { Ok::<_, std::convert::Infallible>(parent) } + }, + ) + .await + .unwrap(); + assert_eq!( + result, + Some(block_102.clone()), + "root=101: should stop at 102 (child of root, across skip)" + ); + } }