diff --git a/chain/arweave/src/chain.rs b/chain/arweave/src/chain.rs index 65019d0724b..1612d1cd9b5 100644 --- a/chain/arweave/src/chain.rs +++ b/chain/arweave/src/chain.rs @@ -3,7 +3,7 @@ use graph::blockchain::client::ChainClient; use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor; use graph::blockchain::{ BasicBlockchainBuilder, Block, BlockIngestor, BlockchainBuilder, BlockchainKind, - EmptyNodeCapabilities, NoopRuntimeAdapter, + EmptyNodeCapabilities, NoopDecoderHook, NoopRuntimeAdapter, }; use graph::cheap_clone::CheapClone; use graph::components::store::DeploymentCursorTracker; @@ -88,6 +88,8 @@ impl Blockchain for Chain { type NodeCapabilities = EmptyNodeCapabilities; + type DecoderHook = NoopDecoderHook; + fn triggers_adapter( &self, _loc: &DeploymentLocator, @@ -161,8 +163,8 @@ impl Blockchain for Chain { .map_err(Into::into) } - fn runtime_adapter(&self) -> Arc> { - Arc::new(NoopRuntimeAdapter::default()) + fn runtime(&self) -> (Arc>, Self::DecoderHook) { + (Arc::new(NoopRuntimeAdapter::default()), NoopDecoderHook) } fn chain_client(&self) -> Arc> { diff --git a/chain/arweave/src/data_source.rs b/chain/arweave/src/data_source.rs index d106829279f..f94d6260785 100644 --- a/chain/arweave/src/data_source.rs +++ b/chain/arweave/src/data_source.rs @@ -157,7 +157,7 @@ impl blockchain::DataSource for DataSource { todo!() } - fn validate(&self) -> Vec { + fn validate(&self, _: &semver::Version) -> Vec { let mut errors = Vec::new(); if self.kind != ARWEAVE_KIND { diff --git a/chain/cosmos/src/chain.rs b/chain/cosmos/src/chain.rs index 6a493a144d4..6d5ee27269e 100644 --- a/chain/cosmos/src/chain.rs +++ b/chain/cosmos/src/chain.rs @@ -1,5 +1,5 @@ use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor; -use graph::blockchain::BlockIngestor; +use graph::blockchain::{BlockIngestor, NoopDecoderHook}; use graph::env::EnvVars; use graph::prelude::MetricsRegistry; use graph::substreams::Clock; @@ -82,6 +82,8 @@ impl Blockchain for Chain { type NodeCapabilities = EmptyNodeCapabilities; + type DecoderHook = NoopDecoderHook; + fn is_refetch_block_required(&self) -> bool { false } @@ -155,8 +157,8 @@ impl Blockchain for Chain { .map_err(Into::into) } - fn runtime_adapter(&self) -> Arc> { - Arc::new(NoopRuntimeAdapter::default()) + fn runtime(&self) -> (Arc>, Self::DecoderHook) { + (Arc::new(NoopRuntimeAdapter::default()), NoopDecoderHook) } fn chain_client(&self) -> Arc> { diff --git a/chain/cosmos/src/data_source.rs b/chain/cosmos/src/data_source.rs index 9d8b2f26672..d7f4f951274 100644 --- a/chain/cosmos/src/data_source.rs +++ b/chain/cosmos/src/data_source.rs @@ -193,7 +193,7 @@ impl blockchain::DataSource for DataSource { Err(anyhow!(DYNAMIC_DATA_SOURCE_ERROR)) } - fn validate(&self) -> Vec { + fn validate(&self, _: &semver::Version) -> Vec { let mut errors = Vec::new(); if self.kind != COSMOS_KIND { @@ -555,7 +555,7 @@ fn duplicate_url_type(message: &str) -> Error { mod tests { use super::*; - use graph::blockchain::DataSource as _; + use graph::{blockchain::DataSource as _, data::subgraph::LATEST_VERSION}; #[test] fn test_event_handlers_origin_validation() { @@ -597,7 +597,7 @@ mod tests { ]; for (data_source, errors) in &cases { - let validation_errors = data_source.validate(); + let validation_errors = data_source.validate(&LATEST_VERSION); assert_eq!(errors.len(), validation_errors.len()); @@ -645,7 +645,7 @@ mod tests { ]; for (data_source, errors) in &cases { - let validation_errors = data_source.validate(); + let validation_errors = data_source.validate(&LATEST_VERSION); assert_eq!(errors.len(), validation_errors.len()); diff --git a/chain/ethereum/src/adapter.rs b/chain/ethereum/src/adapter.rs index 37efbf8553d..7bb40469e6f 100644 --- a/chain/ethereum/src/adapter.rs +++ b/chain/ethereum/src/adapter.rs @@ -2,6 +2,8 @@ use anyhow::Error; use ethabi::{Error as ABIError, Function, ParamType, Token}; use futures::Future; use graph::blockchain::ChainIdentifier; +use graph::components::subgraph::MappingError; +use graph::data::store::ethereum::call; use graph::firehose::CallToFilter; use graph::firehose::CombinedFilter; use graph::firehose::LogFilter; @@ -36,7 +38,8 @@ pub type EventSignature = H256; pub type FunctionSelector = [u8; 4]; #[derive(Clone, Debug)] -pub struct EthereumContractCall { +pub struct ContractCall { + pub contract_name: String, pub address: Address, pub block_ptr: BlockPtr, pub function: Function, @@ -45,7 +48,7 @@ pub struct EthereumContractCall { } #[derive(Error, Debug)] -pub enum EthereumGetBalanceError { +pub enum GetBalanceError { #[error("call error: {0}")] Web3Error(web3::Error), #[error("ethereum node took too long to perform call")] @@ -53,7 +56,7 @@ pub enum EthereumGetBalanceError { } #[derive(Error, Debug)] -pub enum EthereumContractCallError { +pub enum ContractCallError { #[error("ABI error: {0}")] ABIError(#[from] ABIError), /// `Token` is not of expected `ParamType` @@ -63,10 +66,28 @@ pub enum EthereumContractCallError { EncodingError(ethabi::Error), #[error("call error: {0}")] Web3Error(web3::Error), - #[error("call reverted: {0}")] - Revert(String), #[error("ethereum node took too long to perform call")] Timeout, + #[error("internal error: {0}")] + Internal(String), +} + +impl From for MappingError { + fn from(e: ContractCallError) -> Self { + match e { + // Any error reported by the Ethereum node could be due to the block no longer being on + // the main chain. This is very unespecific but we don't want to risk failing a + // subgraph due to a transient error such as a reorg. + ContractCallError::Web3Error(e) => MappingError::PossibleReorg(anyhow::anyhow!( + "Ethereum node returned an error for an eth_call: {e}" + )), + // Also retry on timeouts. + ContractCallError::Timeout => MappingError::PossibleReorg(anyhow::anyhow!( + "Ethereum node did not respond in time to eth_call" + )), + e => MappingError::Unknown(anyhow::anyhow!("Error when making an eth_call: {e}")), + } + } } #[derive(Copy, Clone, Debug, PartialEq, Eq, Ord, PartialOrd, Hash)] @@ -964,20 +985,32 @@ pub trait EthereumAdapter: Send + Sync + 'static { block_number: BlockNumber, ) -> Box, Error = Error> + Send>; - /// Call the function of a smart contract. - fn contract_call( + /// Call the function of a smart contract. A return of `None` indicates + /// that the call reverted. The returned `CallSource` indicates where + /// the result came from for accounting purposes + async fn contract_call( + &self, + logger: &Logger, + call: &ContractCall, + cache: Arc, + ) -> Result<(Option>, call::Source), ContractCallError>; + + /// Make multiple contract calls in a single batch. The returned `Vec` + /// has results in the same order as the calls in `calls` on input. The + /// calls must all be for the same block + async fn contract_calls( &self, logger: &Logger, - call: EthereumContractCall, + calls: &[&ContractCall], cache: Arc, - ) -> Box, Error = EthereumContractCallError> + Send>; + ) -> Result>, call::Source)>, ContractCallError>; fn get_balance( &self, logger: &Logger, address: H160, block_ptr: BlockPtr, - ) -> Box + Send>; + ) -> Box + Send>; } #[cfg(test)] diff --git a/chain/ethereum/src/buffered_call_cache.rs b/chain/ethereum/src/buffered_call_cache.rs new file mode 100644 index 00000000000..8a51bd9a0a4 --- /dev/null +++ b/chain/ethereum/src/buffered_call_cache.rs @@ -0,0 +1,145 @@ +use std::{ + collections::HashMap, + sync::{Arc, Mutex}, +}; + +use graph::{ + cheap_clone::CheapClone, + components::store::EthereumCallCache, + data::store::ethereum::call, + prelude::{BlockPtr, CachedEthereumCall}, + slog::{error, Logger}, +}; + +/// A wrapper around an Ethereum call cache that buffers call results in +/// memory for the duration of a block. If `get_call` or `set_call` are +/// called with a different block pointer than the one used in the previous +/// call, the buffer is cleared. +pub struct BufferedCallCache { + call_cache: Arc, + buffer: Arc>>, + block: Arc>>, +} + +impl BufferedCallCache { + pub fn new(call_cache: Arc) -> Self { + Self { + call_cache, + buffer: Arc::new(Mutex::new(HashMap::new())), + block: Arc::new(Mutex::new(None)), + } + } + + fn check_block(&self, block: &BlockPtr) { + let mut self_block = self.block.lock().unwrap(); + if self_block.as_ref() != Some(block) { + *self_block = Some(block.clone()); + self.buffer.lock().unwrap().clear(); + } + } + + fn get(&self, call: &call::Request) -> Option { + let buffer = self.buffer.lock().unwrap(); + buffer.get(call).map(|retval| { + call.cheap_clone() + .response(retval.clone(), call::Source::Memory) + }) + } +} + +impl EthereumCallCache for BufferedCallCache { + fn get_call( + &self, + call: &call::Request, + block: BlockPtr, + ) -> Result, graph::prelude::Error> { + self.check_block(&block); + + if let Some(value) = self.get(call) { + return Ok(Some(value)); + } + + let result = self.call_cache.get_call(&call, block)?; + + let mut buffer = self.buffer.lock().unwrap(); + if let Some(call::Response { + retval, + req: _, + source: _, + }) = &result + { + buffer.insert(call.cheap_clone(), retval.clone()); + } + Ok(result) + } + + fn get_calls( + &self, + reqs: &[call::Request], + block: BlockPtr, + ) -> Result<(Vec, Vec), graph::prelude::Error> { + self.check_block(&block); + + let mut missing = Vec::new(); + let mut resps = Vec::new(); + + for call in reqs { + match self.get(call) { + Some(resp) => resps.push(resp), + None => missing.push(call.cheap_clone()), + } + } + + let (stored, calls) = self.call_cache.get_calls(&missing, block)?; + + { + let mut buffer = self.buffer.lock().unwrap(); + for resp in &stored { + buffer.insert(resp.req.cheap_clone(), resp.retval.clone()); + } + } + + resps.extend(stored); + Ok((resps, calls)) + } + + fn get_calls_in_block( + &self, + block: BlockPtr, + ) -> Result, graph::prelude::Error> { + self.call_cache.get_calls_in_block(block) + } + + fn set_call( + &self, + logger: &Logger, + call: call::Request, + block: BlockPtr, + return_value: call::Retval, + ) -> Result<(), graph::prelude::Error> { + self.check_block(&block); + + // Enter the call into the in-memory cache immediately so that + // handlers will find it, but add it to the underlying cache in the + // background so we do not have to wait for that as it will be a + // cache backed by the database + { + let mut buffer = self.buffer.lock().unwrap(); + buffer.insert(call.cheap_clone(), return_value.clone()); + } + + let cache = self.call_cache.cheap_clone(); + let logger = logger.cheap_clone(); + let _ = graph::spawn_blocking_allow_panic(move || { + cache + .set_call(&logger, call.cheap_clone(), block, return_value) + .map_err(|e| { + error!(logger, "BufferedCallCache: call cache set error"; + "contract_address" => format!("{:?}", call.address), + "error" => e.to_string()) + }) + }); + + Ok(()) + } +} diff --git a/chain/ethereum/src/chain.rs b/chain/ethereum/src/chain.rs index 0754e7e2146..375b6b5f0ec 100644 --- a/chain/ethereum/src/chain.rs +++ b/chain/ethereum/src/chain.rs @@ -2,7 +2,9 @@ use anyhow::{anyhow, bail, Result}; use anyhow::{Context, Error}; use graph::blockchain::client::ChainClient; use graph::blockchain::firehose_block_ingestor::{FirehoseBlockIngestor, Transforms}; -use graph::blockchain::{BlockIngestor, BlockTime, BlockchainKind, TriggersAdapterSelector}; +use graph::blockchain::{ + BlockIngestor, BlockTime, BlockchainKind, ChainIdentifier, TriggersAdapterSelector, +}; use graph::components::store::DeploymentCursorTracker; use graph::data::subgraph::UnifiedMappingApiVersion; use graph::firehose::{FirehoseEndpoint, ForkStep}; @@ -42,8 +44,7 @@ use crate::data_source::DataSourceTemplate; use crate::data_source::UnresolvedDataSourceTemplate; use crate::ingestor::PollingBlockIngestor; use crate::network::EthereumNetworkAdapters; -use crate::EthereumAdapter; -use crate::NodeCapabilities; +use crate::runtime::runtime_adapter::eth_call_gas; use crate::{ adapter::EthereumAdapter as _, codec, @@ -54,6 +55,8 @@ use crate::{ }, SubgraphEthRpcMetrics, TriggerFilter, ENV_VARS, }; +use crate::{BufferedCallCache, NodeCapabilities}; +use crate::{EthereumAdapter, RuntimeAdapter}; use graph::blockchain::block_stream::{ BlockStream, BlockStreamBuilder, BlockStreamError, BlockStreamMapper, FirehoseCursor, }; @@ -251,10 +254,40 @@ impl TriggersAdapterSelector for EthereumAdapterSelector { } } +/// We need this so that the runner tests can use a `NoopRuntimeAdapter` +/// instead of the `RuntimeAdapter` from this crate to avoid needing +/// ethereum adapters +pub trait RuntimeAdapterBuilder: Send + Sync + 'static { + fn build( + &self, + eth_adapters: Arc, + call_cache: Arc, + chain_identifier: Arc, + ) -> Arc>; +} + +pub struct EthereumRuntimeAdapterBuilder {} + +impl RuntimeAdapterBuilder for EthereumRuntimeAdapterBuilder { + fn build( + &self, + eth_adapters: Arc, + call_cache: Arc, + chain_identifier: Arc, + ) -> Arc> { + Arc::new(RuntimeAdapter { + eth_adapters, + call_cache, + chain_identifier, + }) + } +} + pub struct Chain { logger_factory: LoggerFactory, name: String, node_id: NodeId, + chain_identifier: Arc, registry: Arc, client: Arc>, chain_store: Arc, @@ -266,7 +299,8 @@ pub struct Chain { block_stream_builder: Arc>, block_refetcher: Arc>, adapter_selector: Arc>, - runtime_adapter: Arc>, + runtime_adapter_builder: Arc, + eth_adapters: Arc, } impl std::fmt::Debug for Chain { @@ -289,15 +323,18 @@ impl Chain { block_stream_builder: Arc>, block_refetcher: Arc>, adapter_selector: Arc>, - runtime_adapter: Arc>, + runtime_adapter_builder: Arc, + eth_adapters: Arc, reorg_threshold: BlockNumber, polling_ingestor_interval: Duration, is_ingestible: bool, ) -> Self { + let chain_identifier = Arc::new(chain_store.chain_identifier().clone()); Chain { logger_factory, name, node_id, + chain_identifier, registry, client, chain_store, @@ -306,7 +343,8 @@ impl Chain { block_stream_builder, block_refetcher, adapter_selector, - runtime_adapter, + runtime_adapter_builder, + eth_adapters, reorg_threshold, is_ingestible, polling_ingestor_interval, @@ -354,6 +392,8 @@ impl Blockchain for Chain { type NodeCapabilities = crate::capabilities::NodeCapabilities; + type DecoderHook = crate::data_source::DecoderHook; + fn triggers_adapter( &self, loc: &DeploymentLocator, @@ -443,8 +483,23 @@ impl Blockchain for Chain { self.block_refetcher.get_block(self, logger, cursor).await } - fn runtime_adapter(&self) -> Arc> { - self.runtime_adapter.clone() + fn runtime(&self) -> (Arc>, Self::DecoderHook) { + let call_cache = Arc::new(BufferedCallCache::new(self.call_cache.cheap_clone())); + + let builder = self.runtime_adapter_builder.build( + self.eth_adapters.cheap_clone(), + call_cache.cheap_clone(), + self.chain_identifier.cheap_clone(), + ); + let eth_call_gas = eth_call_gas(&self.chain_identifier); + + let decoder_hook = crate::data_source::DecoderHook::new( + self.eth_adapters.cheap_clone(), + call_cache, + eth_call_gas, + ); + + (builder, decoder_hook) } fn chain_client(&self) -> Arc> { diff --git a/chain/ethereum/src/data_source.rs b/chain/ethereum/src/data_source.rs index 20aaf5e9651..7ebe1d7263f 100644 --- a/chain/ethereum/src/data_source.rs +++ b/chain/ethereum/src/data_source.rs @@ -1,19 +1,28 @@ use anyhow::{anyhow, Error}; use anyhow::{ensure, Context}; -use graph::blockchain::TriggerWithHandler; -use graph::components::store::StoredDynamicDataSource; -use graph::components::subgraph::InstanceDSTemplateInfo; +use graph::blockchain::{BlockPtr, TriggerWithHandler}; +use graph::components::metrics::subgraph::SubgraphInstanceMetrics; +use graph::components::store::{EthereumCallCache, StoredDynamicDataSource}; +use graph::components::subgraph::{HostMetrics, InstanceDSTemplateInfo, MappingError}; +use graph::components::trigger_processor::RunnableTriggers; +use graph::data::value::Word; use graph::data_source::CausalityRegion; +use graph::env::ENV_VARS; use graph::prelude::ethabi::ethereum_types::H160; -use graph::prelude::ethabi::StateMutability; +use graph::prelude::ethabi::{StateMutability, Token}; use graph::prelude::futures03::future::try_join; use graph::prelude::futures03::stream::FuturesOrdered; +use graph::prelude::regex::Regex; use graph::prelude::{Link, SubgraphManifestValidationError}; -use graph::slog::{o, trace}; +use graph::slog::{debug, error, o, trace}; +use itertools::Itertools; +use lazy_static::lazy_static; +use serde::de; use std::collections::HashSet; use std::num::NonZeroU32; use std::str::FromStr; use std::sync::Arc; +use std::time::{Duration, Instant}; use tiny_keccak::{keccak256, Keccak}; use graph::{ @@ -30,10 +39,14 @@ use graph::{ use graph::data::subgraph::{ calls_host_fn, DataSourceContext, Source, MIN_SPEC_VERSION, SPEC_VERSION_0_0_8, + SPEC_VERSION_1_2_0, }; +use crate::adapter::EthereumAdapter as _; use crate::chain::Chain; +use crate::network::EthereumNetworkAdapters; use crate::trigger::{EthereumBlockTriggerType, EthereumTrigger, MappingTrigger}; +use crate::{ContractCall, NodeCapabilities}; // The recommended kind is `ethereum`, `ethereum/contract` is accepted for backwards compatibility. const ETHEREUM_KINDS: &[&str] = &["ethereum/contract", "ethereum"]; @@ -268,7 +281,7 @@ impl blockchain::DataSource for DataSource { }) } - fn validate(&self) -> Vec { + fn validate(&self, spec_version: &semver::Version) -> Vec { let mut errors = vec![]; if !ETHEREUM_KINDS.contains(&self.kind.as_str()) { @@ -347,6 +360,33 @@ impl blockchain::DataSource for DataSource { } } + if spec_version < &SPEC_VERSION_1_2_0 { + for handler in &self.mapping.event_handlers { + if !handler.calls.decls.is_empty() { + errors.push(anyhow!( + "handler {}: declaring eth calls on handlers is only supported for specVersion >= 1.2.0", handler.event + )); + break; + } + } + } + + for handler in &self.mapping.event_handlers { + for call in handler.calls.decls.as_ref() { + match self.mapping.find_abi(&call.expr.abi) { + // TODO: Handle overloaded functions by passing a signature + Ok(abi) => match abi.function(&call.expr.abi, &call.expr.func, None) { + Ok(_) => {} + Err(e) => { + errors.push(e); + } + }, + Err(e) => { + errors.push(e); + } + } + } + } errors } @@ -741,6 +781,7 @@ impl DataSource { "transaction" => format!("{}", &transaction.hash), }); let handler = event_handler.handler.clone(); + let calls = DeclaredCall::new(&self.mapping, &event_handler, &log, ¶ms)?; Ok(Some(TriggerWithHandler::::new_with_logging_extras( MappingTrigger::Log { block: block.cheap_clone(), @@ -748,6 +789,7 @@ impl DataSource { log, params, receipt: receipt.map(|r| r.cheap_clone()), + calls, }, handler, block.block_ptr(), @@ -870,6 +912,289 @@ impl DataSource { } } +#[derive(Clone, Debug, PartialEq)] +pub struct DeclaredCall { + /// The user-supplied label from the manifest + label: String, + contract_name: String, + address: Address, + function: Function, + args: Vec, +} + +impl DeclaredCall { + fn new( + mapping: &Mapping, + handler: &MappingEventHandler, + log: &Log, + params: &[LogParam], + ) -> Result, anyhow::Error> { + let mut calls = Vec::new(); + for decl in handler.calls.decls.iter() { + let contract_name = decl.expr.abi.to_string(); + let function_name = decl.expr.func.as_str(); + // Obtain the path to the contract ABI + let abi = mapping.find_abi(&contract_name)?; + // TODO: Handle overloaded functions + let function = { + // Behavior for apiVersion < 0.0.4: look up function by name; for overloaded + // functions this always picks the same overloaded variant, which is incorrect + // and may lead to encoding/decoding errors + abi.contract.function(function_name).with_context(|| { + format!( + "Unknown function \"{}::{}\" called from WASM runtime", + contract_name, function_name + ) + })? + }; + + let address = decl.address(log, params)?; + let args = decl.args(log, params)?; + + let call = DeclaredCall { + label: decl.label.clone(), + contract_name, + address, + function: function.clone(), + args, + }; + calls.push(call); + } + + Ok(calls) + } + + fn as_eth_call(self, block_ptr: BlockPtr, gas: Option) -> (ContractCall, String) { + ( + ContractCall { + contract_name: self.contract_name, + address: self.address, + block_ptr, + function: self.function, + args: self.args, + gas, + }, + self.label, + ) + } +} + +pub struct DecoderHook { + eth_adapters: Arc, + call_cache: Arc, + eth_call_gas: Option, +} + +impl DecoderHook { + pub fn new( + eth_adapters: Arc, + call_cache: Arc, + eth_call_gas: Option, + ) -> Self { + Self { + eth_adapters, + call_cache, + eth_call_gas, + } + } +} + +impl DecoderHook { + /// Perform a batch of eth_calls, observing the execution time of each + /// call. Returns a list of the call labels for which we received a + /// `None` response, indicating a revert + async fn eth_calls( + &self, + logger: &Logger, + block_ptr: &BlockPtr, + calls_and_metrics: Vec<(Arc, DeclaredCall)>, + ) -> Result, MappingError> { + // This check is not just to speed things up, but is also needed to + // make sure the runner tests don't fail; they don't have declared + // eth calls, but without this check we try to get an eth adapter + // even when there are no calls, which fails in the runner test + // setup + if calls_and_metrics.is_empty() { + return Ok(vec![]); + } + + let start = Instant::now(); + + let (metrics, calls): (Vec<_>, Vec<_>) = calls_and_metrics.into_iter().unzip(); + + let (calls, labels): (Vec<_>, Vec<_>) = calls + .into_iter() + .map(|call| call.as_eth_call(block_ptr.clone(), self.eth_call_gas)) + .unzip(); + + let eth_adapter = self.eth_adapters.call_or_cheapest(Some(&NodeCapabilities { + archive: true, + traces: false, + }))?; + + let call_refs = calls.iter().collect::>(); + let results = eth_adapter + .contract_calls(logger, &call_refs, self.call_cache.cheap_clone()) + .await + .map_err(|e| { + // An error happened, everybody gets charged + let elapsed = start.elapsed().as_secs_f64() / call_refs.len() as f64; + for (metrics, call) in metrics.iter().zip(call_refs) { + metrics.observe_eth_call_execution_time( + elapsed, + &call.contract_name, + &call.function.name, + ); + } + MappingError::from(e) + })?; + + // We don't have time measurements for each call (though that would be nice) + // Use the average time of all calls that we want to observe as the time for + // each call + let to_observe = results.iter().map(|(_, source)| source.observe()).count() as f64; + let elapsed = start.elapsed().as_secs_f64() / to_observe; + + results + .iter() + .zip(metrics) + .zip(calls) + .for_each(|(((_, source), metrics), call)| { + if source.observe() { + metrics.observe_eth_call_execution_time( + elapsed, + &call.contract_name, + &call.function.name, + ); + } + }); + + let labels = results + .iter() + .zip(labels) + .filter_map(|((res, _), label)| if res.is_none() { Some(label) } else { None }) + .map(|s| s.to_string()) + .collect(); + Ok(labels) + } +} + +#[async_trait] +impl blockchain::DecoderHook for DecoderHook { + async fn after_decode<'a>( + &self, + logger: &Logger, + block_ptr: &BlockPtr, + runnables: Vec>, + metrics: &Arc, + ) -> Result>, MappingError> { + /// Log information about failed eth calls. 'Failure' here simply + /// means that the call was reverted; outright errors lead to a real + /// error. For reverted calls, `self.eth_calls` returns the label + /// from the manifest for that call. + /// + /// One reason why declared calls can fail is if they are attached + /// to the wrong handler, or if arguments are specified incorrectly. + /// Calls that revert every once in a while might be ok and what the + /// user intended, but we want to clearly log so that users can spot + /// mistakes in their manifest, which will lead to unnecessary eth + /// calls + fn log_results( + logger: &Logger, + failures: &[String], + calls_count: usize, + trigger_count: usize, + elapsed: Duration, + ) { + let fail_count = failures.len(); + + if fail_count > 0 { + let mut counts: Vec<_> = failures.iter().counts().into_iter().collect(); + counts.sort_by_key(|(label, _)| *label); + let counts = counts + .into_iter() + .map(|(label, count)| { + let times = if count == 1 { "time" } else { "times" }; + format!("{label} ({count} {times})") + }) + .join(", "); + error!(logger, "Declared calls failed"; + "triggers" => trigger_count, + "calls_count" => calls_count, + "fail_count" => fail_count, + "calls_ms" => elapsed.as_millis(), + "failures" => format!("[{}]", counts)); + } else { + debug!(logger, "Declared calls"; + "triggers" => trigger_count, + "calls_count" => calls_count, + "calls_ms" => elapsed.as_millis()); + } + } + + if ENV_VARS.mappings.disable_declared_calls { + return Ok(runnables); + } + + let _section = metrics.stopwatch.start_section("declared_ethereum_call"); + + let start = Instant::now(); + let calls: Vec<_> = runnables + .iter() + .map(|r| &r.hosted_triggers) + .flatten() + .filter_map(|trigger| { + trigger + .mapping_trigger + .trigger + .as_onchain() + .map(|t| (trigger.host.host_metrics(), t)) + }) + .filter_map(|(metrics, trigger)| match trigger { + MappingTrigger::Log { calls, .. } => Some( + calls + .clone() + .into_iter() + .map(move |call| (metrics.cheap_clone(), call)), + ), + MappingTrigger::Block { .. } | MappingTrigger::Call { .. } => None, + }) + .flatten() + .collect(); + + // Deduplicate calls. Unfortunately, we can't get `DeclaredCall` to + // implement `Hash` or `Ord` easily, so we can only deduplicate by + // comparing the whole call not with a `HashSet` or `BTreeSet`. + // Since that can be inefficient, we don't deduplicate if we have an + // enormous amount of calls; in that case though, things will likely + // blow up because of the amount of I/O that many calls cause. + // Cutting off at 1000 is fairly arbitrary + let calls = if calls.len() < 1000 { + let mut uniq_calls = Vec::new(); + for (metrics, call) in calls { + if !uniq_calls.iter().any(|(_, c)| c == &call) { + uniq_calls.push((metrics, call)); + } + } + uniq_calls + } else { + calls + }; + + let calls_count = calls.len(); + let results = self.eth_calls(logger, block_ptr, calls).await?; + log_results( + logger, + &results, + calls_count, + runnables.len(), + start.elapsed(), + ); + + Ok(runnables) + } +} + #[derive(Clone, Debug, Eq, PartialEq, Deserialize)] pub struct UnresolvedDataSource { pub kind: String, @@ -1096,12 +1421,6 @@ pub struct UnresolvedMappingABI { pub file: Link, } -#[derive(Clone, Debug, PartialEq)] -pub struct MappingABI { - pub name: String, - pub contract: Contract, -} - impl UnresolvedMappingABI { pub async fn resolve( self, @@ -1122,6 +1441,56 @@ impl UnresolvedMappingABI { } } +#[derive(Clone, Debug, PartialEq)] +pub struct MappingABI { + pub name: String, + pub contract: Contract, +} + +impl MappingABI { + pub fn function( + &self, + contract_name: &str, + name: &str, + signature: Option<&str>, + ) -> Result<&Function, Error> { + let contract = &self.contract; + let function = match signature { + // Behavior for apiVersion < 0.0.4: look up function by name; for overloaded + // functions this always picks the same overloaded variant, which is incorrect + // and may lead to encoding/decoding errors + None => contract.function(name).with_context(|| { + format!( + "Unknown function \"{}::{}\" called from WASM runtime", + contract_name, name + ) + })?, + + // Behavior for apiVersion >= 0.0.04: look up function by signature of + // the form `functionName(uint256,string) returns (bytes32,string)`; this + // correctly picks the correct variant of an overloaded function + Some(ref signature) => contract + .functions_by_name(name) + .with_context(|| { + format!( + "Unknown function \"{}::{}\" called from WASM runtime", + contract_name, name + ) + })? + .iter() + .find(|f| signature == &f.signature()) + .with_context(|| { + format!( + "Unknown function \"{}::{}\" with signature `{}` \ + called from WASM runtime", + contract_name, name, signature, + ) + })?, + }; + Ok(function) + } +} + #[derive(Clone, Debug, Hash, Eq, PartialEq, Deserialize)] pub struct MappingBlockHandler { pub handler: String, @@ -1166,6 +1535,8 @@ pub struct MappingEventHandler { pub handler: String, #[serde(default)] pub receipt: bool, + #[serde(default)] + pub calls: CallDecls, } impl MappingEventHandler { @@ -1193,3 +1564,216 @@ fn string_to_h256(s: &str) -> H256 { pub struct TemplateSource { pub abi: String, } + +/// Internal representation of declared calls. In the manifest that's +/// written as part of an event handler as +/// ```yaml +/// calls: +/// - myCall1: Contract[address].function(arg1, arg2, ...) +/// - .. +/// ``` +/// +/// The `address` and `arg` fields can be either `event.address` or +/// `event.params.`. Each entry under `calls` gets turned into a +/// `CallDcl` +#[derive(Clone, Debug, Default, Hash, Eq, PartialEq)] +pub struct CallDecls { + pub decls: Arc>, + readonly: (), +} + +impl CheapClone for CallDecls {} + +/// A single call declaration, like `myCall1: +/// Contract[address].function(arg1, arg2, ...)` +#[derive(Clone, Debug, Hash, Eq, PartialEq)] +pub struct CallDecl { + /// A user-defined label + pub label: String, + /// The call expression + pub expr: CallExpr, + readonly: (), +} +impl CallDecl { + fn address(&self, log: &Log, params: &[LogParam]) -> Result { + let address = match &self.expr.address { + CallArg::Address => log.address, + CallArg::Param(name) => { + let value = params + .iter() + .find(|param| ¶m.name == name.as_str()) + .ok_or_else(|| anyhow!("unknown param {name}"))? + .value + .clone(); + value + .into_address() + .ok_or_else(|| anyhow!("param {name} is not an address"))? + } + }; + Ok(address) + } + + fn args(&self, log: &Log, params: &[LogParam]) -> Result, Error> { + self.expr + .args + .iter() + .map(|arg| match arg { + CallArg::Address => Ok(Token::Address(log.address)), + CallArg::Param(name) => { + let value = params + .iter() + .find(|param| ¶m.name == name.as_str()) + .ok_or_else(|| anyhow!("unknown param {name}"))? + .value + .clone(); + Ok(value) + } + }) + .collect() + } +} + +impl<'de> de::Deserialize<'de> for CallDecls { + fn deserialize(deserializer: D) -> Result + where + D: de::Deserializer<'de>, + { + let decls: std::collections::HashMap = + de::Deserialize::deserialize(deserializer)?; + let decls = decls + .into_iter() + .map(|(name, expr)| { + expr.parse::().map(|expr| CallDecl { + label: name, + expr, + readonly: (), + }) + }) + .collect::>() + .map(|decls| Arc::new(decls)) + .map_err(de::Error::custom)?; + Ok(CallDecls { + decls, + readonly: (), + }) + } +} + +#[derive(Clone, Debug, Hash, Eq, PartialEq)] +pub struct CallExpr { + pub abi: Word, + pub address: CallArg, + pub func: Word, + pub args: Vec, + readonly: (), +} + +/// Parse expressions of the form `Contract[address].function(arg1, arg2, +/// ...)` where the `address` and the args are either `event.address` or +/// `event.params.`. +/// +/// The parser is pretty awful as it generates error messages that aren't +/// very helpful. We should replace all this with a real parser, most likely +/// `combine` which is what `graphql_parser` uses +impl FromStr for CallExpr { + type Err = anyhow::Error; + + fn from_str(s: &str) -> Result { + lazy_static! { + static ref RE: Regex = Regex::new( + r"(?x) + (?P[a-zA-Z0-9_]+)\[ + (?P
[^]]+)\] + \. + (?P[a-zA-Z0-9_]+)\( + (?P[^)]*) + \)" + ) + .unwrap(); + } + let x = RE + .captures(s) + .ok_or_else(|| anyhow!("invalid call expression `{s}`"))?; + let abi = Word::from(x.name("abi").unwrap().as_str()); + let address = x.name("address").unwrap().as_str().parse()?; + let func = Word::from(x.name("func").unwrap().as_str()); + let args: Vec = x + .name("args") + .unwrap() + .as_str() + .split(',') + .filter(|s| !s.is_empty()) + .map(|s| s.trim().parse::()) + .collect::>()?; + Ok(CallExpr { + abi, + address, + func, + args, + readonly: (), + }) + } +} + +#[derive(Clone, Debug, Hash, Eq, PartialEq)] +pub enum CallArg { + Address, + Param(Word), +} + +impl FromStr for CallArg { + type Err = anyhow::Error; + + fn from_str(s: &str) -> Result { + fn invalid(s: &str) -> Result { + Err(anyhow!("invalid call argument `{}`", s)) + } + + let mut parts = s.split("."); + match parts.next() { + Some("event") => { /* ok */ } + Some(_) => return Err(anyhow!("call arguments must start with `event`")), + None => return Err(anyhow!("empty call argument")), + } + match parts.next() { + Some("address") => Ok(CallArg::Address), + Some("params") => match parts.next() { + Some(s) => Ok(CallArg::Param(Word::from(s))), + None => invalid(s), + }, + Some(s) => invalid(s), + None => invalid(s), + } + } +} + +#[test] +fn test_call_expr() { + let expr: CallExpr = "ERC20[event.address].balanceOf(event.params.token)" + .parse() + .unwrap(); + assert_eq!(expr.abi, "ERC20"); + assert_eq!(expr.address, CallArg::Address); + assert_eq!(expr.func, "balanceOf"); + assert_eq!(expr.args, vec![CallArg::Param("token".into())]); + + let expr: CallExpr = "Pool[event.params.pool].fees(event.params.token0, event.params.token1)" + .parse() + .unwrap(); + assert_eq!(expr.abi, "Pool"); + assert_eq!(expr.address, CallArg::Param("pool".into())); + assert_eq!(expr.func, "fees"); + assert_eq!( + expr.args, + vec![ + CallArg::Param("token0".into()), + CallArg::Param("token1".into()) + ] + ); + + let expr: CallExpr = "Pool[event.address].growth()".parse().unwrap(); + assert_eq!(expr.abi, "Pool"); + assert_eq!(expr.address, CallArg::Address); + assert_eq!(expr.func, "growth"); + assert_eq!(expr.args, vec![]); +} diff --git a/chain/ethereum/src/ethereum_adapter.rs b/chain/ethereum/src/ethereum_adapter.rs index 26b4215734f..63a3de66a7a 100644 --- a/chain/ethereum/src/ethereum_adapter.rs +++ b/chain/ethereum/src/ethereum_adapter.rs @@ -1,14 +1,16 @@ -use futures::future; use futures::prelude::*; use futures03::{future::BoxFuture, stream::FuturesUnordered}; use graph::blockchain::client::ChainClient; use graph::blockchain::BlockHash; use graph::blockchain::ChainIdentifier; use graph::components::transaction_receipt::LightTransactionReceipt; +use graph::data::store::ethereum::call; +use graph::data::store::scalar; use graph::data::subgraph::UnifiedMappingApiVersion; use graph::data::subgraph::API_VERSION_0_0_7; use graph::prelude::ethabi::ParamType; use graph::prelude::ethabi::Token; +use graph::prelude::futures03::future::try_join_all; use graph::prelude::tokio::try_join; use graph::prelude::web3::types::U256; use graph::slog::o; @@ -44,7 +46,7 @@ use std::pin::Pin; use std::sync::Arc; use std::time::Instant; -use crate::adapter::EthereumGetBalanceError; +use crate::adapter::GetBalanceError; use crate::adapter::ProviderStatus; use crate::chain::BlockFinality; use crate::trigger::LogRef; @@ -52,9 +54,9 @@ use crate::Chain; use crate::NodeCapabilities; use crate::{ adapter::{ - EthGetLogsFilter, EthereumAdapter as EthereumAdapterTrait, EthereumBlockFilter, - EthereumCallFilter, EthereumContractCall, EthereumContractCallError, EthereumLogFilter, - ProviderEthRpcMetrics, SubgraphEthRpcMetrics, + ContractCall, ContractCallError, EthGetLogsFilter, EthereumAdapter as EthereumAdapterTrait, + EthereumBlockFilter, EthereumCallFilter, EthereumLogFilter, ProviderEthRpcMetrics, + SubgraphEthRpcMetrics, }, transport::Transport, trigger::{EthereumBlockTriggerType, EthereumTrigger}, @@ -417,7 +419,7 @@ impl EthereumAdapter { logger: &Logger, address: Address, block_ptr: BlockPtr, - ) -> impl Future + Send { + ) -> impl Future + Send { let web3 = self.web3.clone(); let logger = Logger::new(&logger, o!("provider" => self.provider.clone())); @@ -444,23 +446,27 @@ impl EthereumAdapter { web3.eth().balance(address, Some(block_id)).boxed().await; match result { Ok(balance) => Ok(balance), - Err(err) => Err(EthereumGetBalanceError::Web3Error(err)), + Err(err) => Err(GetBalanceError::Web3Error(err)), } } }) - .map_err(|e| e.into_inner().unwrap_or(EthereumGetBalanceError::Timeout)) + .map_err(|e| e.into_inner().unwrap_or(GetBalanceError::Timeout)) .boxed() .compat() } - fn call( + async fn call( &self, logger: Logger, - contract_address: Address, - call_data: Bytes, + call_data: call::Request, block_ptr: BlockPtr, gas: Option, - ) -> impl Future + Send { + ) -> Result { + fn reverted(logger: &Logger, reason: &str) -> Result { + info!(logger, "Contract call reverted"; "reason" => reason); + Ok(call::Retval::Null) + } + let web3 = self.web3.clone(); let logger = Logger::new(&logger, o!("provider" => self.provider.clone())); @@ -473,20 +479,17 @@ impl EthereumAdapter { }; let retry_log_message = format!("eth_call RPC call for block {}", block_ptr); retry(retry_log_message, &logger) - .when(|result| match result { - Ok(_) | Err(EthereumContractCallError::Revert(_)) => false, - Err(_) => true, - }) .limit(ENV_VARS.request_retries) .timeout_secs(ENV_VARS.json_rpc_timeout.as_secs()) .run(move || { let call_data = call_data.clone(); let web3 = web3.cheap_clone(); + let logger = logger.cheap_clone(); async move { let req = CallRequest { - to: Some(contract_address), + to: Some(call_data.address), gas: gas.map(|val| web3::types::U256::from(val)), - data: Some(call_data.clone()), + data: Some(Bytes::from(call_data.encoded_call.to_vec())), from: None, gas_price: None, value: None, @@ -557,14 +560,14 @@ impl EthereumAdapter { match result { // A successful response. - Ok(bytes) => Ok(bytes), + Ok(bytes) => Ok(call::Retval::Value(scalar::Bytes::from(bytes))), // Check for Geth revert. Err(web3::Error::Rpc(rpc_error)) if geth_execution_errors .any(|e| rpc_error.message.to_lowercase().contains(e)) => { - Err(EthereumContractCallError::Revert(rpc_error.message)) + reverted(&logger, &rpc_error.message) } // Check for Parity revert. @@ -592,26 +595,56 @@ impl EthereumAdapter { }) .unwrap_or("no reason".to_owned()) }; - Err(EthereumContractCallError::Revert(reason)) + reverted(&logger, &reason) } // The VM execution error was not identified as a revert. - _ => Err(EthereumContractCallError::Web3Error(web3::Error::Rpc( + _ => Err(ContractCallError::Web3Error(web3::Error::Rpc( rpc_error.clone(), ))), } } // The error was not identified as a revert. - Err(err) => Err(EthereumContractCallError::Web3Error(err)), + Err(err) => Err(ContractCallError::Web3Error(err)), } } }) - .map_err(|e| e.into_inner().unwrap_or(EthereumContractCallError::Timeout)) + .map_err(|e| e.into_inner().unwrap_or(ContractCallError::Timeout)) .boxed() - .compat() + .await } + async fn call_and_cache( + &self, + logger: &Logger, + call: &ContractCall, + req: call::Request, + cache: Arc, + ) -> Result { + let result = self + .call( + logger.clone(), + req.cheap_clone(), + call.block_ptr.clone(), + call.gas, + ) + .await?; + let _ = cache + .set_call( + &logger, + req.cheap_clone(), + call.block_ptr.cheap_clone(), + result.clone(), + ) + .map_err(|e| { + error!(logger, "EthereumAdapter: call cache set error"; + "contract_address" => format!("{:?}", req.address), + "error" => e.to_string()) + }); + + Ok(req.response(result, call::Source::Rpc)) + } /// Request blocks by hash through JSON-RPC. fn load_blocks_rpc( &self, @@ -1295,7 +1328,7 @@ impl EthereumAdapterTrait for EthereumAdapter { logger: &Logger, address: H160, block_ptr: BlockPtr, - ) -> Box + Send> { + ) -> Box + Send> { debug!( logger, "eth_getBalance"; "address" => format!("{}", address), @@ -1304,102 +1337,156 @@ impl EthereumAdapterTrait for EthereumAdapter { Box::new(self.balance(logger, address, block_ptr)) } - fn contract_call( + async fn contract_call( &self, logger: &Logger, - call: EthereumContractCall, + inp_call: &ContractCall, cache: Arc, - ) -> Box, Error = EthereumContractCallError> + Send> { - // Emit custom error for type mismatches. - for (token, kind) in call - .args - .iter() - .zip(call.function.inputs.iter().map(|p| &p.kind)) - { - if !token.type_check(kind) { - return Box::new(future::err(EthereumContractCallError::TypeError( - token.clone(), - kind.clone(), - ))); - } - } - - // Encode the call parameters according to the ABI - let call_data = match call.function.encode_input(&call.args) { - Ok(data) => data, - Err(e) => return Box::new(future::err(EthereumContractCallError::EncodingError(e))), - }; - - debug!(logger, "eth_call"; - "fn" => &call.function.name, - "address" => hex::encode(call.address), - "data" => hex::encode(&call_data), - "block_hash" => call.block_ptr.hash_hex(), - "block_number" => call.block_ptr.block_number() - ); + ) -> Result<(Option>, call::Source), ContractCallError> { + let mut result = self.contract_calls(logger, &[inp_call], cache).await?; + // unwrap: self.contract_calls returns as many results as there were calls + Ok(result.pop().unwrap()) + } - // Check if we have it cached, if not do the call and cache. - Box::new( - match cache - .get_call(call.address, &call_data, call.block_ptr.clone()) - .map_err(|e| error!(logger, "call cache get error"; "error" => e.to_string())) - .ok() - .flatten() + async fn contract_calls( + &self, + logger: &Logger, + calls: &[&ContractCall], + cache: Arc, + ) -> Result>, call::Source)>, ContractCallError> { + fn as_req( + logger: &Logger, + call: &ContractCall, + index: u32, + ) -> Result { + // Emit custom error for type mismatches. + for (token, kind) in call + .args + .iter() + .zip(call.function.inputs.iter().map(|p| &p.kind)) { - Some(result) => { - Box::new(future::ok(result)) as Box + Send> - } - None => { - let cache = cache.clone(); - let call = call.clone(); - let logger = logger.clone(); - Box::new( - self.call( - logger.clone(), - call.address, - Bytes(call_data.clone()), - call.block_ptr.clone(), - call.gas, - ) - .map(move |result| { - // Don't block handler execution on writing to the cache. - let for_cache = result.0.clone(); - if !result.0.is_empty() { - let _ = graph::spawn_blocking_allow_panic(move || { - cache - .set_call( - call.address, - &call_data, - call.block_ptr, - &for_cache, - ) - .map_err(|e| { - error!(logger, "call cache set error"; - "error" => e.to_string()) - }) - }); - } - result.0 - }), - ) + if !token.type_check(kind) { + return Err(ContractCallError::TypeError(token.clone(), kind.clone())); } } - // Decode the return values according to the ABI - .and_then(move |output| { - if output.is_empty() { + + // Encode the call parameters according to the ABI + let req = { + let encoded_call = call + .function + .encode_input(&call.args) + .map_err(ContractCallError::EncodingError)?; + call::Request::new(call.address, encoded_call, index) + }; + + trace!(logger, "eth_call"; + "fn" => &call.function.name, + "address" => hex::encode(call.address), + "data" => hex::encode(req.encoded_call.as_ref()), + "block_hash" => call.block_ptr.hash_hex(), + "block_number" => call.block_ptr.block_number() + ); + Ok(req) + } + + fn decode( + logger: &Logger, + resp: call::Response, + call: &ContractCall, + ) -> (Option>, call::Source) { + let call::Response { + retval, + source, + req: _, + } = resp; + use call::Retval::*; + match retval { + Value(output) => match call.function.decode_output(&output) { + Ok(tokens) => (Some(tokens), source), + Err(e) => { + // Decode failures are reverts. The reasoning is that if Solidity fails to + // decode an argument, that's a revert, so the same goes for the output. + let reason = format!("failed to decode output: {}", e); + info!(logger, "Contract call reverted"; "reason" => reason); + (None, call::Source::Rpc) + } + }, + Null => { // We got a `0x` response. For old Geth, this can mean a revert. It can also be // that the contract actually returned an empty response. A view call is meant // to return something, so we treat empty responses the same as reverts. - Err(EthereumContractCallError::Revert("empty response".into())) - } else { - // Decode failures are reverts. The reasoning is that if Solidity fails to - // decode an argument, that's a revert, so the same goes for the output. - call.function.decode_output(&output).map_err(|e| { - EthereumContractCallError::Revert(format!("failed to decode output: {}", e)) - }) + info!(logger, "Contract call reverted"; "reason" => "empty response"); + (None, call::Source::Rpc) } - }), - ) + } + } + + fn log_call_error(logger: &Logger, e: &ContractCallError, call: &ContractCall) { + match e { + ContractCallError::Web3Error(e) => error!(logger, + "Ethereum node returned an error when calling function \"{}\" of contract \"{}\": {}", + call.function.name, call.contract_name, e), + ContractCallError::Timeout => error!(logger, + "Ethereum node did not respond when calling function \"{}\" of contract \"{}\"", + call.function.name, call.contract_name), + _ => error!(logger, + "Failed to call function \"{}\" of contract \"{}\": {}", + call.function.name, call.contract_name, e), + } + } + + if calls.is_empty() { + return Ok(Vec::new()); + } + + let block_ptr = calls.first().unwrap().block_ptr.clone(); + if calls.iter().any(|call| call.block_ptr != block_ptr) { + return Err(ContractCallError::Internal( + "all calls must have the same block pointer".to_string(), + )); + } + + let reqs: Vec<_> = calls + .iter() + .enumerate() + .map(|(index, call)| as_req(logger, call, index as u32)) + .collect::>()?; + + let (mut resps, missing) = cache + .get_calls(&reqs, block_ptr) + .map_err(|e| error!(logger, "call cache get error"; "error" => e.to_string())) + .unwrap_or_else(|_| (Vec::new(), reqs)); + + let futs = missing.into_iter().map(|req| { + let cache = cache.clone(); + async move { + let call = calls[req.index as usize]; + match self.call_and_cache(logger, call, req, cache.clone()).await { + Ok(resp) => Ok(resp), + Err(e) => { + log_call_error(logger, &e, call); + Err(e) + } + } + } + }); + resps.extend(try_join_all(futs).await?); + + // If we make it here, we have a response for every call. + debug_assert_eq!(resps.len(), calls.len()); + + // Bring the responses into the same order as the calls + resps.sort_by_key(|resp| resp.req.index); + + let decoded: Vec<_> = resps + .into_iter() + .map(|res| { + let call = &calls[res.req.index as usize]; + decode(logger, res, call) + }) + .collect(); + + Ok(decoded) } /// Load Ethereum blocks in bulk, returning results as they come back as a Stream. diff --git a/chain/ethereum/src/lib.rs b/chain/ethereum/src/lib.rs index cc9c24e79a4..934830ecde5 100644 --- a/chain/ethereum/src/lib.rs +++ b/chain/ethereum/src/lib.rs @@ -1,4 +1,5 @@ mod adapter; +mod buffered_call_cache; mod capabilities; pub mod codec; mod data_source; @@ -14,6 +15,8 @@ pub use self::runtime::RuntimeAdapter; pub use self::transport::Transport; pub use env::ENV_VARS; +pub use buffered_call_cache::BufferedCallCache; + // ETHDEP: These concrete types should probably not be exposed. pub use data_source::{ BlockHandlerFilter, DataSource, DataSourceTemplate, Mapping, MappingABI, TemplateSource, @@ -25,7 +28,7 @@ pub mod network; pub mod trigger; pub use crate::adapter::{ - EthereumAdapter as EthereumAdapterTrait, EthereumContractCall, EthereumContractCallError, + ContractCall, ContractCallError, EthereumAdapter as EthereumAdapterTrait, ProviderEthRpcMetrics, SubgraphEthRpcMetrics, TriggerFilter, }; pub use crate::chain::Chain; diff --git a/chain/ethereum/src/runtime/runtime_adapter.rs b/chain/ethereum/src/runtime/runtime_adapter.rs index 868079402e7..2be0dc30042 100644 --- a/chain/ethereum/src/runtime/runtime_adapter.rs +++ b/chain/ethereum/src/runtime/runtime_adapter.rs @@ -1,21 +1,22 @@ use std::{sync::Arc, time::Instant}; -use crate::adapter::EthereumGetBalanceError; +use crate::adapter::GetBalanceError; use crate::data_source::MappingABI; use crate::{ - capabilities::NodeCapabilities, network::EthereumNetworkAdapters, Chain, DataSource, - EthereumAdapter, EthereumAdapterTrait, EthereumContractCall, EthereumContractCallError, - ENV_VARS, + capabilities::NodeCapabilities, network::EthereumNetworkAdapters, Chain, ContractCall, + ContractCallError, DataSource, EthereumAdapter, EthereumAdapterTrait, ENV_VARS, }; use anyhow::{anyhow, Context, Error}; use blockchain::HostFn; use graph::blockchain::ChainIdentifier; use graph::components::subgraph::HostMetrics; +use graph::data::store::ethereum::call; use graph::data::store::scalar::BigInt; use graph::data::subgraph::API_VERSION_0_0_9; use graph::prelude::web3::types::H160; use graph::runtime::gas::Gas; use graph::runtime::{AscIndexId, IndexForAscTypeId}; +use graph::slog::debug; use graph::{ blockchain::{self, BlockPtr, HostFnCtx}, cheap_clone::CheapClone, @@ -25,9 +26,10 @@ use graph::{ }, runtime::{asc_get, asc_new, AscPtr, HostExportError}, semver::Version, - slog::{info, trace, Logger}, + slog::Logger, }; use graph_runtime_wasm::asc_abi::class::{AscBigInt, AscEnumArray, EthereumValueKind}; +use itertools::Itertools; use super::abi::{AscUnresolvedContractCall, AscUnresolvedContractCall_0_0_4}; @@ -60,23 +62,26 @@ pub struct RuntimeAdapter { pub chain_identifier: Arc, } +pub fn eth_call_gas(chain_identifier: &ChainIdentifier) -> Option { + // Check if the current network version is in the eth_call_no_gas list + let should_skip_gas = ENV_VARS + .eth_call_no_gas + .contains(&chain_identifier.net_version); + + if should_skip_gas { + None + } else { + Some(ETH_CALL_GAS) + } +} + impl blockchain::RuntimeAdapter for RuntimeAdapter { fn host_fns(&self, ds: &DataSource) -> Result, Error> { let abis = ds.mapping.abis.clone(); let call_cache = self.call_cache.cheap_clone(); let eth_adapters = self.eth_adapters.cheap_clone(); let archive = ds.mapping.requires_archive()?; - - // Check if the current network version is in the eth_call_no_gas list - let should_skip_gas = ENV_VARS - .eth_call_no_gas - .contains(&self.chain_identifier.net_version); - - let eth_call_gas = if should_skip_gas { - None - } else { - Some(ETH_CALL_GAS) - }; + let eth_call_gas = eth_call_gas(&self.chain_identifier); let ethereum_call = HostFn { name: "ethereum.call", @@ -182,9 +187,9 @@ fn eth_get_balance( Ok(asc_new(ctx.heap, &bigint, &ctx.gas)?) } // Retry on any kind of error - Err(EthereumGetBalanceError::Web3Error(e)) => Err(HostExportError::PossibleReorg(e.into())), - Err(EthereumGetBalanceError::Timeout) => Err(HostExportError::PossibleReorg( - EthereumGetBalanceError::Timeout.into(), + Err(GetBalanceError::Web3Error(e)) => Err(HostExportError::PossibleReorg(e.into())), + Err(GetBalanceError::Timeout) => Err(HostExportError::PossibleReorg( + GetBalanceError::Timeout.into(), )), } } @@ -200,10 +205,23 @@ fn eth_call( eth_call_gas: Option, metrics: Arc, ) -> Result>, HostExportError> { + // Helpers to log the result of the call at the end + fn tokens_as_string(tokens: &[Token]) -> String { + tokens.iter().map(|arg| arg.to_string()).join(", ") + } + + fn result_as_string(result: &Result>, HostExportError>) -> String { + match result { + Ok(Some(tokens)) => format!("({})", tokens_as_string(&tokens)), + Ok(None) => "none".to_string(), + Err(_) => "error".to_string(), + } + } + let start_time = Instant::now(); // Obtain the path to the contract ABI - let contract = abis + let abi = abis .iter() .find(|abi| abi.name == unresolved_call.contract_name) .with_context(|| { @@ -213,51 +231,18 @@ fn eth_call( unresolved_call.contract_name ) }) - .map_err(HostExportError::Deterministic)? - .contract - .clone(); - - let function = match unresolved_call.function_signature { - // Behavior for apiVersion < 0.0.4: look up function by name; for overloaded - // functions this always picks the same overloaded variant, which is incorrect - // and may lead to encoding/decoding errors - None => contract - .function(unresolved_call.function_name.as_str()) - .with_context(|| { - format!( - "Unknown function \"{}::{}\" called from WASM runtime", - unresolved_call.contract_name, unresolved_call.function_name - ) - }) - .map_err(HostExportError::Deterministic)?, - - // Behavior for apiVersion >= 0.0.04: look up function by signature of - // the form `functionName(uint256,string) returns (bytes32,string)`; this - // correctly picks the correct variant of an overloaded function - Some(ref function_signature) => contract - .functions_by_name(unresolved_call.function_name.as_str()) - .with_context(|| { - format!( - "Unknown function \"{}::{}\" called from WASM runtime", - unresolved_call.contract_name, unresolved_call.function_name - ) - }) - .map_err(HostExportError::Deterministic)? - .iter() - .find(|f| function_signature == &f.signature()) - .with_context(|| { - format!( - "Unknown function \"{}::{}\" with signature `{}` \ - called from WASM runtime", - unresolved_call.contract_name, - unresolved_call.function_name, - function_signature, - ) - }) - .map_err(HostExportError::Deterministic)?, - }; - - let call = EthereumContractCall { + .map_err(HostExportError::Deterministic)?; + + let function = abi + .function( + &unresolved_call.contract_name, + &unresolved_call.function_name, + unresolved_call.function_signature.as_deref(), + ) + .map_err(HostExportError::Deterministic)?; + + let call = ContractCall { + contract_name: unresolved_call.contract_name.clone(), address: unresolved_call.contract_address, block_ptr: block_ptr.cheap_clone(), function: function.clone(), @@ -268,19 +253,18 @@ fn eth_call( // Run Ethereum call in tokio runtime let logger1 = logger.clone(); let call_cache = call_cache.clone(); - let result = match graph::block_on( - eth_adapter.contract_call(&logger1, call, call_cache).compat() - ) { - Ok(tokens) => Ok(Some(tokens)), - Err(EthereumContractCallError::Revert(reason)) => { - info!(logger, "Contract call reverted"; "reason" => reason); - Ok(None) - } + let (result, source) = + match graph::block_on(eth_adapter.contract_call(&logger1, &call, call_cache)) { + Ok((result, source)) => (Ok(result), source), + Err(e) => (Err(e), call::Source::Rpc), + }; + let result = match result { + Ok(res) => Ok(res), // Any error reported by the Ethereum node could be due to the block no longer being on // the main chain. This is very unespecific but we don't want to risk failing a // subgraph due to a transient error such as a reorg. - Err(EthereumContractCallError::Web3Error(e)) => Err(HostExportError::PossibleReorg(anyhow::anyhow!( + Err(ContractCallError::Web3Error(e)) => Err(HostExportError::PossibleReorg(anyhow::anyhow!( "Ethereum node returned an error when calling function \"{}\" of contract \"{}\": {}", unresolved_call.function_name, unresolved_call.contract_name, @@ -288,7 +272,7 @@ fn eth_call( ))), // Also retry on timeouts. - Err(EthereumContractCallError::Timeout) => Err(HostExportError::PossibleReorg(anyhow::anyhow!( + Err(ContractCallError::Timeout) => Err(HostExportError::PossibleReorg(anyhow::anyhow!( "Ethereum node did not respond when calling function \"{}\" of contract \"{}\"", unresolved_call.function_name, unresolved_call.contract_name, @@ -304,18 +288,24 @@ fn eth_call( let elapsed = start_time.elapsed(); - metrics.observe_eth_call_execution_time( - elapsed.as_secs_f64(), - &unresolved_call.contract_name, - &unresolved_call.function_name, - ); + if source.observe() { + metrics.observe_eth_call_execution_time( + elapsed.as_secs_f64(), + &unresolved_call.contract_name, + &unresolved_call.function_name, + ); + } - trace!(logger, "Contract call finished"; - "address" => &unresolved_call.contract_address.to_string(), + debug!(logger, "Contract call finished"; + "address" => format!("0x{:x}", &unresolved_call.contract_address), "contract" => &unresolved_call.contract_name, - "function" => &unresolved_call.function_name, - "function_signature" => &unresolved_call.function_signature, - "time" => format!("{}ms", elapsed.as_millis())); + "signature" => &unresolved_call.function_signature, + "args" => format!("[{}]", tokens_as_string(&unresolved_call.function_args)), + "time_ms" => format!("{}ms", elapsed.as_millis()), + "result" => result_as_string(&result), + "block_hash" => block_ptr.hash_hex(), + "block_number" => block_ptr.block_number(), + "source" => source.to_string()); result } diff --git a/chain/ethereum/src/trigger.rs b/chain/ethereum/src/trigger.rs index a57f4c8410f..128ed8d3e98 100644 --- a/chain/ethereum/src/trigger.rs +++ b/chain/ethereum/src/trigger.rs @@ -28,6 +28,7 @@ use graph_runtime_wasm::module::ToAscPtr; use std::ops::Deref; use std::{cmp::Ordering, sync::Arc}; +use crate::data_source::DeclaredCall; use crate::runtime::abi::AscEthereumBlock; use crate::runtime::abi::AscEthereumBlock_0_0_6; use crate::runtime::abi::AscEthereumCall; @@ -48,6 +49,7 @@ pub enum MappingTrigger { log: Arc, params: Vec, receipt: Option>, + calls: Vec, }, Call { block: Arc, @@ -102,6 +104,7 @@ impl std::fmt::Debug for MappingTrigger { log, params, receipt: _, + calls: _, } => MappingTriggerWithoutBlock::Log { _transaction: transaction.cheap_clone(), _log: log.cheap_clone(), @@ -139,6 +142,7 @@ impl ToAscPtr for MappingTrigger { log, params, receipt, + calls: _, } => { let api_version = heap.api_version(); let ethereum_event_data = EthereumEventData { diff --git a/chain/near/src/chain.rs b/chain/near/src/chain.rs index 135350a1deb..db2474a25cf 100644 --- a/chain/near/src/chain.rs +++ b/chain/near/src/chain.rs @@ -3,7 +3,8 @@ use graph::blockchain::client::ChainClient; use graph::blockchain::firehose_block_ingestor::FirehoseBlockIngestor; use graph::blockchain::substreams_block_stream::SubstreamsBlockStream; use graph::blockchain::{ - BasicBlockchainBuilder, BlockIngestor, BlockchainBuilder, BlockchainKind, NoopRuntimeAdapter, + BasicBlockchainBuilder, BlockIngestor, BlockchainBuilder, BlockchainKind, NoopDecoderHook, + NoopRuntimeAdapter, }; use graph::cheap_clone::CheapClone; use graph::components::store::DeploymentCursorTracker; @@ -209,6 +210,8 @@ impl Blockchain for Chain { type NodeCapabilities = EmptyNodeCapabilities; + type DecoderHook = NoopDecoderHook; + fn triggers_adapter( &self, _loc: &DeploymentLocator, @@ -283,8 +286,8 @@ impl Blockchain for Chain { .await } - fn runtime_adapter(&self) -> Arc> { - Arc::new(NoopRuntimeAdapter::default()) + fn runtime(&self) -> (Arc>, Self::DecoderHook) { + (Arc::new(NoopRuntimeAdapter::default()), NoopDecoderHook) } fn chain_client(&self) -> Arc> { @@ -565,6 +568,7 @@ mod test { use graph::{ blockchain::{block_stream::BlockWithTriggers, DataSource as _, TriggersAdapter as _}, + data::subgraph::LATEST_VERSION, prelude::{tokio, Link}, semver::Version, slog::{self, o, Logger}, @@ -587,7 +591,7 @@ mod test { #[test] fn validate_empty() { let ds = new_data_source(None, None); - let errs = ds.validate(); + let errs = ds.validate(LATEST_VERSION); assert_eq!(errs.len(), 1, "{:?}", ds); assert_eq!(errs[0].to_string(), "subgraph source address is required"); } @@ -595,7 +599,7 @@ mod test { #[test] fn validate_empty_account_none_partial() { let ds = new_data_source(None, Some(PartialAccounts::default())); - let errs = ds.validate(); + let errs = ds.validate(LATEST_VERSION); assert_eq!(errs.len(), 1, "{:?}", ds); assert_eq!(errs[0].to_string(), "subgraph source address is required"); } @@ -609,7 +613,7 @@ mod test { suffixes: vec!["x.near".to_string()], }), ); - let errs = ds.validate(); + let errs = ds.validate(LATEST_VERSION); assert_eq!(errs.len(), 0, "{:?}", ds); } @@ -623,7 +627,7 @@ mod test { }), ); let errs: Vec = ds - .validate() + .validate(LATEST_VERSION) .into_iter() .map(|err| err.to_string()) .collect(); @@ -644,7 +648,7 @@ mod test { #[test] fn validate_empty_partials() { let ds = new_data_source(Some("x.near".to_string()), None); - let errs = ds.validate(); + let errs = ds.validate(LATEST_VERSION); assert_eq!(errs.len(), 0, "{:?}", ds); } diff --git a/chain/near/src/data_source.rs b/chain/near/src/data_source.rs index 7b001098993..ea54c31d157 100644 --- a/chain/near/src/data_source.rs +++ b/chain/near/src/data_source.rs @@ -226,7 +226,7 @@ impl blockchain::DataSource for DataSource { todo!() } - fn validate(&self) -> Vec { + fn validate(&self, _: &semver::Version) -> Vec { let mut errors = Vec::new(); if self.kind != NEAR_KIND { diff --git a/chain/starknet/src/chain.rs b/chain/starknet/src/chain.rs index 060a502d80d..9f5097a8c6b 100644 --- a/chain/starknet/src/chain.rs +++ b/chain/starknet/src/chain.rs @@ -10,7 +10,7 @@ use graph::{ firehose_block_ingestor::FirehoseBlockIngestor, firehose_block_stream::FirehoseBlockStream, BasicBlockchainBuilder, Block, BlockIngestor, BlockPtr, Blockchain, BlockchainBuilder, - BlockchainKind, EmptyNodeCapabilities, IngestorError, NoopRuntimeAdapter, + BlockchainKind, EmptyNodeCapabilities, IngestorError, NoopDecoderHook, NoopRuntimeAdapter, RuntimeAdapter as RuntimeAdapterTrait, }, cheap_clone::CheapClone, @@ -94,6 +94,8 @@ impl Blockchain for Chain { type NodeCapabilities = EmptyNodeCapabilities; + type DecoderHook = NoopDecoderHook; + fn triggers_adapter( &self, _log: &DeploymentLocator, @@ -153,8 +155,8 @@ impl Blockchain for Chain { .await } - fn runtime_adapter(&self) -> Arc> { - Arc::new(NoopRuntimeAdapter::default()) + fn runtime(&self) -> (Arc>, Self::DecoderHook) { + (Arc::new(NoopRuntimeAdapter::default()), NoopDecoderHook) } fn chain_client(&self) -> Arc> { @@ -430,7 +432,7 @@ impl TriggersAdapterTrait for TriggersAdapter { mod tests { use std::sync::Arc; - use graph::blockchain::DataSource as _; + use graph::{blockchain::DataSource as _, data::subgraph::LATEST_VERSION}; use crate::{ data_source::{ @@ -443,7 +445,7 @@ mod tests { fn validate_no_handler() { let ds = new_data_source(None); - let errs = ds.validate(); + let errs = ds.validate(LATEST_VERSION); assert_eq!(errs.len(), 1, "{:?}", ds); assert_eq!( errs[0].to_string(), @@ -458,7 +460,7 @@ mod tests { handler: "asdf".into(), }); - let errs = ds.validate(); + let errs = ds.validate(LATEST_VERSION); assert_eq!(errs.len(), 1, "{:?}", ds); assert_eq!( errs[0].to_string(), @@ -474,7 +476,7 @@ mod tests { event_selector: [2u8; 32].into(), }); - let errs = ds.validate(); + let errs = ds.validate(LATEST_VERSION); assert_eq!(errs.len(), 1, "{:?}", ds); assert_eq!(errs[0].to_string(), "subgraph source address is required"); } diff --git a/chain/starknet/src/data_source.rs b/chain/starknet/src/data_source.rs index ef8ccd3ad9b..8f168dc47c5 100644 --- a/chain/starknet/src/data_source.rs +++ b/chain/starknet/src/data_source.rs @@ -208,7 +208,7 @@ impl blockchain::DataSource for DataSource { todo!() } - fn validate(&self) -> Vec { + fn validate(&self, _: &semver::Version) -> Vec { let mut errors = Vec::new(); if self.kind != STARKNET_KIND { diff --git a/chain/substreams/src/chain.rs b/chain/substreams/src/chain.rs index 1dad1764707..a871d813e08 100644 --- a/chain/substreams/src/chain.rs +++ b/chain/substreams/src/chain.rs @@ -3,7 +3,8 @@ use crate::{data_source::*, EntityChanges, TriggerData, TriggerFilter, TriggersA use anyhow::Error; use graph::blockchain::client::ChainClient; use graph::blockchain::{ - BasicBlockchainBuilder, BlockIngestor, BlockTime, EmptyNodeCapabilities, NoopRuntimeAdapter, + BasicBlockchainBuilder, BlockIngestor, BlockTime, EmptyNodeCapabilities, NoopDecoderHook, + NoopRuntimeAdapter, }; use graph::components::store::DeploymentCursorTracker; use graph::env::EnvVars; @@ -120,6 +121,8 @@ impl Blockchain for Chain { type NodeCapabilities = EmptyNodeCapabilities; + type DecoderHook = NoopDecoderHook; + fn triggers_adapter( &self, _log: &DeploymentLocator, @@ -178,8 +181,8 @@ impl Blockchain for Chain { number, }) } - fn runtime_adapter(&self) -> Arc> { - Arc::new(NoopRuntimeAdapter::default()) + fn runtime(&self) -> (Arc>, Self::DecoderHook) { + (Arc::new(NoopRuntimeAdapter::default()), NoopDecoderHook) } fn chain_client(&self) -> Arc> { diff --git a/chain/substreams/src/data_source.rs b/chain/substreams/src/data_source.rs index ba7666fc9d3..4d3230cae6b 100644 --- a/chain/substreams/src/data_source.rs +++ b/chain/substreams/src/data_source.rs @@ -106,7 +106,7 @@ impl blockchain::DataSource for DataSource { unimplemented!("{}", DYNAMIC_DATA_SOURCE_ERROR) } - fn validate(&self) -> Vec { + fn validate(&self, _: &semver::Version) -> Vec { let mut errs = vec![]; if &self.kind != SUBSTREAMS_KIND { @@ -327,11 +327,14 @@ mod test { use graph::{ blockchain::{DataSource as _, UnresolvedDataSource as _}, components::link_resolver::LinkResolver, + data::subgraph::LATEST_VERSION, prelude::{async_trait, serde_yaml, JsonValueStream, Link}, slog::{o, Discard, Logger}, - substreams::module::{Kind, KindMap, KindStore}, substreams::{ - module::input::{Input, Params}, + module::{ + input::{Input, Params}, + Kind, KindMap, KindStore, + }, Module, Modules, Package, }, }; @@ -459,15 +462,19 @@ mod test { #[test] fn data_source_validation() { let mut ds = gen_data_source(); - assert_eq!(true, ds.validate().is_empty()); + assert_eq!(true, ds.validate(LATEST_VERSION).is_empty()); ds.network = None; - assert_eq!(true, ds.validate().is_empty()); + assert_eq!(true, ds.validate(LATEST_VERSION).is_empty()); ds.kind = "asdasd".into(); ds.name = "".into(); ds.mapping.kind = "asdasd".into(); - let errs: Vec = ds.validate().into_iter().map(|e| e.to_string()).collect(); + let errs: Vec = ds + .validate(LATEST_VERSION) + .into_iter() + .map(|e| e.to_string()) + .collect(); assert_eq!( errs, vec![ diff --git a/chain/substreams/src/trigger.rs b/chain/substreams/src/trigger.rs index a3cdf5cd5cb..2360e8a711f 100644 --- a/chain/substreams/src/trigger.rs +++ b/chain/substreams/src/trigger.rs @@ -8,8 +8,8 @@ use graph::{ components::{ store::{DeploymentLocator, SubgraphFork}, subgraph::{MappingError, ProofOfIndexingEvent, SharedProofOfIndexing}, + trigger_processor::HostedTrigger, }, - data_source, prelude::{ anyhow, async_trait, BlockHash, BlockNumber, BlockState, CheapClone, RuntimeHostBuilder, }, @@ -197,9 +197,8 @@ where async fn process_trigger<'a>( &'a self, logger: &Logger, - _: Box + Send + 'a>, + _: Vec>, block: &Arc, - _trigger: &data_source::TriggerData, mut state: BlockState, proof_of_indexing: &SharedProofOfIndexing, causality_region: &str, diff --git a/core/src/subgraph/context/instance/mod.rs b/core/src/subgraph/context/instance/mod.rs index 49e5cfd3c86..df330881d8d 100644 --- a/core/src/subgraph/context/instance/mod.rs +++ b/core/src/subgraph/context/instance/mod.rs @@ -13,7 +13,7 @@ use graph::{ use hosts::{OffchainHosts, OnchainHosts}; use std::collections::HashMap; -pub(super) struct SubgraphInstance> { +pub(crate) struct SubgraphInstance> { subgraph_id: DeploymentHash, network: String, host_builder: T, diff --git a/core/src/subgraph/context/mod.rs b/core/src/subgraph/context/mod.rs index 6c87e095d1e..42c6ef2b192 100644 --- a/core/src/subgraph/context/mod.rs +++ b/core/src/subgraph/context/mod.rs @@ -15,7 +15,7 @@ use graph::{ data_source::{ causality_region::CausalityRegionSeq, offchain::{self, Base64}, - CausalityRegion, DataSource, DataSourceTemplate, TriggerData, + CausalityRegion, DataSource, DataSourceTemplate, }, ipfs_client::CidFile, prelude::{ @@ -31,6 +31,8 @@ use std::{collections::HashMap, time::Instant}; use self::instance::SubgraphInstance; +use super::Decoder; + #[derive(Clone, Debug)] pub struct SubgraphKeepAlive { alive_map: Arc>>, @@ -69,11 +71,12 @@ where T: RuntimeHostBuilder, C: Blockchain, { - instance: SubgraphInstance, + pub(crate) instance: SubgraphInstance, pub instances: SubgraphKeepAlive, pub offchain_monitor: OffchainMonitor, pub filter: Option, - trigger_processor: Box>, + pub(crate) trigger_processor: Box>, + pub(crate) decoder: Box>, } impl> IndexingContext { @@ -85,6 +88,7 @@ impl> IndexingContext { instances: SubgraphKeepAlive, offchain_monitor: OffchainMonitor, trigger_processor: Box>, + decoder: Box>, ) -> Self { let instance = SubgraphInstance::new( manifest, @@ -99,36 +103,10 @@ impl> IndexingContext { offchain_monitor, filter: None, trigger_processor, + decoder, } } - pub async fn process_trigger( - &self, - logger: &Logger, - block: &Arc, - trigger: &TriggerData, - state: BlockState, - proof_of_indexing: &SharedProofOfIndexing, - causality_region: &str, - debug_fork: &Option>, - subgraph_metrics: &Arc, - instrument: bool, - ) -> Result { - self.process_trigger_in_hosts( - logger, - self.instance.hosts_for_trigger(trigger), - block, - trigger, - state, - proof_of_indexing, - causality_region, - debug_fork, - subgraph_metrics, - instrument, - ) - .await - } - pub async fn process_block( &self, logger: &Logger, @@ -189,34 +167,6 @@ impl> IndexingContext { Ok(state) } - pub async fn process_trigger_in_hosts( - &self, - logger: &Logger, - hosts: Box + Send + '_>, - block: &Arc, - trigger: &TriggerData, - state: BlockState, - proof_of_indexing: &SharedProofOfIndexing, - causality_region: &str, - debug_fork: &Option>, - subgraph_metrics: &Arc, - instrument: bool, - ) -> Result { - self.trigger_processor - .process_trigger( - logger, - hosts, - block, - trigger, - state, - proof_of_indexing, - causality_region, - debug_fork, - subgraph_metrics, - instrument, - ) - .await - } /// Removes data sources hosts with a creation block greater or equal to `reverted_block`, so /// that they are no longer candidates for `process_trigger`. diff --git a/core/src/subgraph/instance_manager.rs b/core/src/subgraph/instance_manager.rs index d0176238013..223b855d132 100644 --- a/core/src/subgraph/instance_manager.rs +++ b/core/src/subgraph/instance_manager.rs @@ -2,6 +2,7 @@ use crate::polling_monitor::{ArweaveService, IpfsService}; use crate::subgraph::context::{IndexingContext, SubgraphKeepAlive}; use crate::subgraph::inputs::IndexingInputs; use crate::subgraph::loader::load_dynamic_data_sources; +use crate::subgraph::Decoder; use std::collections::BTreeSet; use crate::subgraph::runner::SubgraphRunner; @@ -389,8 +390,9 @@ impl SubgraphInstanceManager { let deployment_head = store.block_ptr().map(|ptr| ptr.number).unwrap_or(0) as f64; block_stream_metrics.deployment_head.set(deployment_head); + let (runtime_adapter, decoder_hook) = chain.runtime(); let host_builder = graph_runtime_wasm::RuntimeHostBuilder::new( - chain.runtime_adapter(), + runtime_adapter, self.link_resolver.cheap_clone(), subgraph_store.ens_lookup(), ); @@ -408,6 +410,8 @@ impl SubgraphInstanceManager { let instrument = self.subgraph_store.instrument(&deployment)?; + let decoder = Box::new(Decoder::new(decoder_hook)); + let inputs = IndexingInputs { deployment: deployment.clone(), features, @@ -438,6 +442,7 @@ impl SubgraphInstanceManager { self.instances.cheap_clone(), offchain_monitor, tp, + decoder, ); for data_source in data_sources { ctx.add_dynamic_data_source(&logger, data_source)?; diff --git a/core/src/subgraph/runner.rs b/core/src/subgraph/runner.rs index 53e08154a8b..489dd473861 100644 --- a/core/src/subgraph/runner.rs +++ b/core/src/subgraph/runner.rs @@ -299,47 +299,90 @@ where // Causality region for onchain triggers. let causality_region = PoICausalityRegion::from_network(&self.inputs.network); - let mut block_state = { - let _section = self - .metrics - .stream - .stopwatch - .start_section(PROCESS_TRIGGERS_SECTION_NAME); - - // Process events one after the other, passing in entity operations - // collected previously to every new event being processed - match self - .process_triggers( - &proof_of_indexing, - &block, - triggers.into_iter().map(TriggerData::Onchain), - &causality_region, - ) - .await - { - // Triggers processed with no errors or with only deterministic errors. - Ok(block_state) => block_state, - - // Some form of unknown or non-deterministic error ocurred. - Err(MappingError::Unknown(e)) => return Err(BlockProcessingError::Unknown(e)), - Err(MappingError::PossibleReorg(e)) => { - info!(logger, - "Possible reorg detected, retrying"; - "error" => format!("{:#}", e), - ); + let mut block_state = BlockState::new( + self.inputs.store.clone(), + std::mem::take(&mut self.state.entity_lfu_cache), + ); - // In case of a possible reorg, we want this function to do nothing and restart the - // block stream so it has a chance to detect the reorg. - // - // The state is unchanged at this point, except for having cleared the entity cache. - // Losing the cache is a bit annoying but not an issue for correctness. - // - // See also b21fa73b-6453-4340-99fb-1a78ec62efb1. - return Ok(Action::Restart); + let _section = self + .metrics + .stream + .stopwatch + .start_section(PROCESS_TRIGGERS_SECTION_NAME); + + // Match and decode all triggers in the block + let hosts_filter = |trigger: &TriggerData| self.ctx.instance.hosts_for_trigger(trigger); + let match_res = self + .ctx + .decoder + .match_and_decode_many( + &logger, + &block, + triggers.into_iter().map(TriggerData::Onchain), + hosts_filter, + &self.metrics.subgraph, + ) + .await; + + // Process events one after the other, passing in entity operations + // collected previously to every new event being processed + let mut res = Ok(block_state); + match match_res { + Ok(runnables) => { + for runnable in runnables { + let process_res = self + .ctx + .trigger_processor + .process_trigger( + &self.logger, + runnable.hosted_triggers, + &block, + res.unwrap(), + &proof_of_indexing, + &causality_region, + &self.inputs.debug_fork, + &self.metrics.subgraph, + self.inputs.instrument, + ) + .await + .map_err(|e| e.add_trigger_context(&runnable.trigger)); + match process_res { + Ok(state) => res = Ok(state), + Err(e) => { + res = Err(e); + break; + } + } } } + Err(e) => { + res = Err(e); + } }; + match res { + // Triggers processed with no errors or with only deterministic errors. + Ok(state) => block_state = state, + + // Some form of unknown or non-deterministic error ocurred. + Err(MappingError::Unknown(e)) => return Err(BlockProcessingError::Unknown(e)), + Err(MappingError::PossibleReorg(e)) => { + info!(logger, + "Possible reorg detected, retrying"; + "error" => format!("{:#}", e), + ); + + // In case of a possible reorg, we want this function to do nothing and restart the + // block stream so it has a chance to detect the reorg. + // + // The state is unchanged at this point, except for having cleared the entity cache. + // Losing the cache is a bit annoying but not an issue for correctness. + // + // See also b21fa73b-6453-4340-99fb-1a78ec62efb1. + return Ok(Action::Restart); + } + } + // Check if there are any datasources that have expired in this block. ie: the end_block // of that data source is equal to the block number of the current block. let has_expired_data_sources = self.inputs.end_blocks.contains(&block_ptr.number); @@ -428,34 +471,63 @@ where // Process the triggers in each host in the same order the // corresponding data sources have been created. - for trigger in triggers { - block_state = self - .ctx - .process_trigger_in_hosts( - &logger, - Box::new(runtime_hosts.iter().map(|host| host.as_ref())), - &block, - &TriggerData::Onchain(trigger), - block_state, - &proof_of_indexing, - &causality_region, - &self.inputs.debug_fork, - &self.metrics.subgraph, - self.inputs.instrument, - ) - .await - .map_err(|e| { - // This treats a `PossibleReorg` as an ordinary error which will fail the subgraph. - // This can cause an unnecessary subgraph failure, to fix it we need to figure out a - // way to revert the effect of `create_dynamic_data_sources` so we may return a - // clean context as in b21fa73b-6453-4340-99fb-1a78ec62efb1. - match e { - MappingError::PossibleReorg(e) | MappingError::Unknown(e) => { - BlockProcessingError::Unknown(e) + let match_res: Result, _> = self + .ctx + .decoder + .match_and_decode_many( + &logger, + &block, + triggers.into_iter().map(TriggerData::Onchain), + |_| Box::new(runtime_hosts.iter().map(Arc::as_ref)), + &self.metrics.subgraph, + ) + .await; + + let mut res = Ok(block_state); + match match_res { + Ok(runnables) => { + for runnable in runnables { + let process_res = self + .ctx + .trigger_processor + .process_trigger( + &self.logger, + runnable.hosted_triggers, + &block, + res.unwrap(), + &proof_of_indexing, + &causality_region, + &self.inputs.debug_fork, + &self.metrics.subgraph, + self.inputs.instrument, + ) + .await + .map_err(|e| e.add_trigger_context(&runnable.trigger)); + match process_res { + Ok(state) => res = Ok(state), + Err(e) => { + res = Err(e); + break; } } - })?; + } + } + Err(e) => { + res = Err(e); + } } + + block_state = res.map_err(|e| { + // This treats a `PossibleReorg` as an ordinary error which will fail the subgraph. + // This can cause an unnecessary subgraph failure, to fix it we need to figure out a + // way to revert the effect of `create_dynamic_data_sources` so we may return a + // clean context as in b21fa73b-6453-4340-99fb-1a78ec62efb1. + match e { + MappingError::PossibleReorg(e) | MappingError::Unknown(e) => { + BlockProcessingError::Unknown(e) + } + } + })?; } } @@ -622,43 +694,6 @@ where } } - async fn process_triggers( - &mut self, - proof_of_indexing: &SharedProofOfIndexing, - block: &Arc, - triggers: impl Iterator>, - causality_region: &str, - ) -> Result { - let mut block_state = BlockState::new( - self.inputs.store.clone(), - std::mem::take(&mut self.state.entity_lfu_cache), - ); - - for trigger in triggers { - block_state = self - .ctx - .process_trigger( - &self.logger, - block, - &trigger, - block_state, - proof_of_indexing, - causality_region, - &self.inputs.debug_fork, - &self.metrics.subgraph, - self.inputs.instrument, - ) - .await - .map_err(move |mut e| { - let error_context = trigger.error_context(); - if !error_context.is_empty() { - e = e.context(error_context); - } - e.context("failed to process trigger".to_string()) - })?; - } - Ok(block_state) - } async fn process_wasm_block( &mut self, proof_of_indexing: &SharedProofOfIndexing, @@ -1024,28 +1059,47 @@ where let proof_of_indexing = None; let causality_region = ""; - block_state = self - .ctx - .process_trigger( + let trigger = TriggerData::Offchain(trigger); + let process_res = { + let hosts = self.ctx.instance.hosts_for_trigger(&trigger); + let triggers_res = self.ctx.decoder.match_and_decode( &self.logger, block, - &TriggerData::Offchain(trigger), - block_state, - &proof_of_indexing, - causality_region, - &self.inputs.debug_fork, + trigger, + hosts, &self.metrics.subgraph, - self.inputs.instrument, - ) - .await - .map_err(move |err| { + ); + match triggers_res { + Ok(runnable) => { + self.ctx + .trigger_processor + .process_trigger( + &self.logger, + runnable.hosted_triggers, + block, + block_state, + &proof_of_indexing, + causality_region, + &self.inputs.debug_fork, + &self.metrics.subgraph, + self.inputs.instrument, + ) + .await + } + Err(e) => Err(e), + } + }; + match process_res { + Ok(state) => block_state = state, + Err(err) => { let err = match err { // Ignoring `PossibleReorg` isn't so bad since the subgraph will retry // non-deterministic errors. MappingError::PossibleReorg(e) | MappingError::Unknown(e) => e, }; - err.context("failed to process trigger".to_string()) - })?; + return Err(err.context("failed to process trigger".to_string())); + } + } anyhow::ensure!( !block_state.has_created_on_chain_data_sources(), diff --git a/core/src/subgraph/trigger_processor.rs b/core/src/subgraph/trigger_processor.rs index f0dc70712d6..0057e9e1354 100644 --- a/core/src/subgraph/trigger_processor.rs +++ b/core/src/subgraph/trigger_processor.rs @@ -1,14 +1,16 @@ use async_trait::async_trait; -use graph::blockchain::{Block, Blockchain}; +use graph::blockchain::{Block, Blockchain, DecoderHook as _}; use graph::cheap_clone::CheapClone; use graph::components::store::SubgraphFork; use graph::components::subgraph::{MappingError, SharedProofOfIndexing}; -use graph::data_source::{MappingTrigger, TriggerData, TriggerWithHandler}; +use graph::components::trigger_processor::{HostedTrigger, RunnableTriggers}; +use graph::data_source::TriggerData; use graph::prelude::tokio::time::Instant; use graph::prelude::{ BlockState, RuntimeHost, RuntimeHostBuilder, SubgraphInstanceMetrics, TriggerProcessor, }; use graph::slog::Logger; +use std::marker::PhantomData; use std::sync::Arc; pub struct SubgraphTriggerProcessor {} @@ -22,9 +24,8 @@ where async fn process_trigger<'a>( &'a self, logger: &Logger, - hosts: Box + Send + 'a>, + triggers: Vec>, block: &Arc, - trigger: &TriggerData, mut state: BlockState, proof_of_indexing: &SharedProofOfIndexing, causality_region: &str, @@ -34,25 +35,7 @@ where ) -> Result { let error_count = state.deterministic_errors.len(); - let mut host_mapping: Vec<(&T::Host, TriggerWithHandler>)> = vec![]; - - { - let _section = subgraph_metrics.stopwatch.start_section("match_and_decode"); - - for host in hosts { - let mapping_trigger = match host.match_and_decode(trigger, block, logger)? { - // Trigger matches and was decoded as a mapping trigger. - Some(mapping_trigger) => mapping_trigger, - - // Trigger does not match, do not process it. - None => continue, - }; - - host_mapping.push((host, mapping_trigger)); - } - } - - if host_mapping.is_empty() { + if triggers.is_empty() { return Ok(state); } @@ -62,7 +45,11 @@ where .start_handler(causality_region); } - for (host, mapping_trigger) in host_mapping { + for HostedTrigger { + host, + mapping_trigger, + } in triggers + { let start = Instant::now(); state = host .process_mapping_trigger( @@ -101,3 +88,102 @@ where Ok(state) } } + +/// A helper for taking triggers as `TriggerData` (usually from the block +/// stream) and turning them into `HostedTrigger`s that are ready to run. +/// +/// The output triggers will be run in the order in which they are returned. +pub struct Decoder +where + C: Blockchain, + T: RuntimeHostBuilder, +{ + hook: C::DecoderHook, + _builder: PhantomData, +} + +impl Decoder +where + C: Blockchain, + T: RuntimeHostBuilder, +{ + pub fn new(hook: C::DecoderHook) -> Self { + Decoder { + hook, + _builder: PhantomData, + } + } +} + +impl> Decoder { + fn match_and_decode_inner<'a>( + &'a self, + logger: &Logger, + block: &Arc, + trigger: &TriggerData, + hosts: Box + Send + 'a>, + subgraph_metrics: &Arc, + ) -> Result>, MappingError> { + let mut host_mapping = vec![]; + + { + let _section = subgraph_metrics.stopwatch.start_section("match_and_decode"); + + for host in hosts { + let mapping_trigger = match host.match_and_decode(trigger, block, logger)? { + // Trigger matches and was decoded as a mapping trigger. + Some(mapping_trigger) => mapping_trigger, + + // Trigger does not match, do not process it. + None => continue, + }; + + host_mapping.push(HostedTrigger { + host, + mapping_trigger, + }); + } + } + Ok(host_mapping) + } + + pub(crate) fn match_and_decode<'a>( + &'a self, + logger: &Logger, + block: &Arc, + trigger: TriggerData, + hosts: Box + Send + 'a>, + subgraph_metrics: &Arc, + ) -> Result, MappingError> { + self.match_and_decode_inner(logger, block, &trigger, hosts, subgraph_metrics) + .map_err(|e| e.add_trigger_context(&trigger)) + .map(|hosted_triggers| RunnableTriggers { + trigger, + hosted_triggers, + }) + } + + pub(crate) async fn match_and_decode_many<'a, F>( + &'a self, + logger: &Logger, + block: &Arc, + triggers: impl Iterator>, + hosts_filter: F, + metrics: &Arc, + ) -> Result>, MappingError> + where + F: Fn(&TriggerData) -> Box + Send + 'a>, + { + let mut runnables = vec![]; + for trigger in triggers { + let hosts = hosts_filter(&trigger); + match self.match_and_decode(logger, block, trigger, hosts, metrics) { + Ok(runnable_triggers) => runnables.push(runnable_triggers), + Err(e) => return Err(e), + } + } + self.hook + .after_decode(logger, &block.ptr(), runnables, metrics) + .await + } +} diff --git a/docs/subgraph-manifest.md b/docs/subgraph-manifest.md index e595690e9f3..4f766829cfc 100644 --- a/docs/subgraph-manifest.md +++ b/docs/subgraph-manifest.md @@ -74,6 +74,7 @@ The `mapping` field may be one of the following supported mapping manifests: | **event** | *String* | An identifier for an event that will be handled in the mapping script. For Ethereum contracts, this must be the full event signature to distinguish from events that may share the same name. No alias types can be used. For example, uint will not work, uint256 must be used.| | **handler** | *String* | The name of an exported function in the mapping script that should handle the specified event. | | **topic0** | optional *String* | A `0x` prefixed hex string. If provided, events whose topic0 is equal to this value will be processed by the given handler. When topic0 is provided, _only_ the topic0 value will be matched, and not the hash of the event signature. This is useful for processing anonymous events in Solidity, which can have their topic0 set to anything. By default, topic0 is equal to the hash of the event signature. | +| **calls** | optional [*CallDecl*](#153-declaring-calls) | A list of predeclared `eth_calls` that will be made before running the handler | #### 1.5.2.3 CallHandler @@ -95,6 +96,30 @@ The `mapping` field may be one of the following supported mapping manifests: | --- | --- | --- | | **kind** | *String* | The selected block handler filter. Only option for now: `call`: This will only run the handler if the block contains at least one call to the data source contract. | +### 1.5.3 Declaring calls + +_Available from spec version 1.2.0_ + +Declared calls are performed in parallel before the handler is run and can +greatly speed up syncing. Mappings access the call results simply by using +`ethereum.call` from the mappings. The **calls** are a map of key value pairs: + +| Field | Type | Description | +| --- | --- | --- | +| **label** | *String* | A label for the call for error messages etc. | +| **call** | *String* | See below | + +Each call is of the form `[
].()`: + +| Field | Type | Description | +| --- | --- | --- | +| **ABI** | *String* | The name of an ABI from the `abis` section | +| **address** | *Expr* | The address of a contract that follows the `ABI` | +| **function** | *String* | The name of a view function in the contract | +| **args** | *[Expr]* | The arguments to pass to the function | + +The `Expr` can be either `event.address` or `event.params.`. + ## 1.6 Path A path has one field `path`, which either refers to a path of a file on the local dev machine or an [IPLD link](https://github.com/ipld/specs/). diff --git a/graph/src/blockchain/mock.rs b/graph/src/blockchain/mock.rs index 1319dd5f054..6bde388b99a 100644 --- a/graph/src/blockchain/mock.rs +++ b/graph/src/blockchain/mock.rs @@ -16,7 +16,7 @@ use super::{ block_stream::{self, BlockStream, FirehoseCursor}, client::ChainClient, BlockIngestor, BlockTime, EmptyNodeCapabilities, HostFn, IngestorError, MappingTriggerTrait, - TriggerWithHandler, + NoopDecoderHook, TriggerWithHandler, }; use super::{ @@ -140,7 +140,7 @@ impl DataSource for MockDataSource { todo!() } - fn validate(&self) -> Vec { + fn validate(&self, _: &semver::Version) -> Vec { todo!() } } @@ -325,6 +325,8 @@ impl Blockchain for MockBlockchain { type NodeCapabilities = EmptyNodeCapabilities; + type DecoderHook = NoopDecoderHook; + fn triggers_adapter( &self, _loc: &crate::components::store::DeploymentLocator, @@ -369,7 +371,7 @@ impl Blockchain for MockBlockchain { todo!() } - fn runtime_adapter(&self) -> std::sync::Arc> { + fn runtime(&self) -> (std::sync::Arc>, Self::DecoderHook) { todo!() } diff --git a/graph/src/blockchain/mod.rs b/graph/src/blockchain/mod.rs index c4ec226ad2f..b4ff1026590 100644 --- a/graph/src/blockchain/mod.rs +++ b/graph/src/blockchain/mod.rs @@ -18,9 +18,10 @@ mod types; use crate::{ cheap_clone::CheapClone, components::{ + metrics::subgraph::SubgraphInstanceMetrics, store::{DeploymentCursorTracker, DeploymentLocator, StoredDynamicDataSource}, - subgraph::HostMetrics, - subgraph::InstanceDSTemplateInfo, + subgraph::{HostMetrics, InstanceDSTemplateInfo, MappingError}, + trigger_processor::RunnableTriggers, }, data::subgraph::{UnifiedMappingApiVersion, MIN_SPEC_VERSION}, data_source::{self, DataSourceTemplateInfo}, @@ -168,6 +169,11 @@ pub trait Blockchain: Debug + Sized + Send + Sync + Unpin + 'static { type NodeCapabilities: NodeCapabilities + std::fmt::Display; + /// A callback that is called after the triggers have been decoded and + /// gets an opportunity to post-process triggers before they are run on + /// hosts + type DecoderHook: DecoderHook + Sync + Send; + fn triggers_adapter( &self, log: &DeploymentLocator, @@ -200,7 +206,7 @@ pub trait Blockchain: Debug + Sized + Send + Sync + Unpin + 'static { fn is_refetch_block_required(&self) -> bool; - fn runtime_adapter(&self) -> Arc>; + fn runtime(&self) -> (Arc>, Self::DecoderHook); fn chain_client(&self) -> Arc>; @@ -300,7 +306,7 @@ pub trait DataSource: 'static + Sized + Send + Sync + Clone { fn as_stored_dynamic_data_source(&self) -> StoredDynamicDataSource; /// Used as part of manifest validation. If there are no errors, return an empty vector. - fn validate(&self) -> Vec; + fn validate(&self, spec_version: &semver::Version) -> Vec; fn has_expired(&self, block: BlockNumber) -> bool { self.end_block() @@ -369,6 +375,35 @@ pub trait MappingTriggerTrait { fn error_context(&self) -> String; } +/// A callback that is called after the triggers have been decoded. +#[async_trait] +pub trait DecoderHook { + async fn after_decode<'a>( + &self, + logger: &Logger, + block_ptr: &BlockPtr, + triggers: Vec>, + metrics: &Arc, + ) -> Result>, MappingError>; +} + +/// A decoder hook that does nothing and just returns the triggers that were +/// passed in +pub struct NoopDecoderHook; + +#[async_trait] +impl DecoderHook for NoopDecoderHook { + async fn after_decode<'a>( + &self, + _: &Logger, + _: &BlockPtr, + triggers: Vec>, + _: &Arc, + ) -> Result>, MappingError> { + Ok(triggers) + } +} + pub struct HostFnCtx<'a> { pub logger: Logger, pub block_ptr: BlockPtr, diff --git a/graph/src/components/store/traits.rs b/graph/src/components/store/traits.rs index 6834524916a..b95a6e9d0ea 100644 --- a/graph/src/components/store/traits.rs +++ b/graph/src/components/store/traits.rs @@ -6,13 +6,14 @@ use web3::types::{Address, H256}; use super::*; use crate::blockchain::block_stream::FirehoseCursor; -use crate::blockchain::BlockTime; +use crate::blockchain::{BlockTime, ChainIdentifier}; use crate::components::metrics::stopwatch::StopwatchMetrics; use crate::components::server::index_node::VersionInfo; use crate::components::subgraph::SubgraphVersionSwitchingMode; use crate::components::transaction_receipt; use crate::components::versions::ApiVersion; use crate::data::query::Trace; +use crate::data::store::ethereum::call; use crate::data::store::QueryObject; use crate::data::subgraph::{status, DeploymentFeatures}; use crate::data::{query::QueryTarget, subgraph::schema::*}; @@ -523,17 +524,29 @@ pub trait ChainStore: Send + Sync + 'static { /// Clears call cache of the chain for the given `from` and `to` block number. async fn clear_call_cache(&self, from: BlockNumber, to: BlockNumber) -> Result<(), Error>; + + /// Return the chain identifier for this store. + fn chain_identifier(&self) -> &ChainIdentifier; } pub trait EthereumCallCache: Send + Sync + 'static { - /// Returns the return value of the provided Ethereum call, if present in - /// the cache. + /// Returns the return value of the provided Ethereum call, if present + /// in the cache. A return of `None` indicates that we know nothing + /// about the call. fn get_call( &self, - contract_address: ethabi::Address, - encoded_call: &[u8], + call: &call::Request, + block: BlockPtr, + ) -> Result, Error>; + + /// Get the return values of many Ethereum calls. For the ones found in + /// the cache, return a `Response`; the ones that were not found are + /// returned as the original `Request` + fn get_calls( + &self, + reqs: &[call::Request], block: BlockPtr, - ) -> Result>, Error>; + ) -> Result<(Vec, Vec), Error>; /// Returns all cached calls for a given `block`. This method does *not* /// update the last access time of the returned cached calls. @@ -542,10 +555,10 @@ pub trait EthereumCallCache: Send + Sync + 'static { /// Stores the provided Ethereum call in the cache. fn set_call( &self, - contract_address: ethabi::Address, - encoded_call: &[u8], + logger: &Logger, + call: call::Request, block: BlockPtr, - return_value: &[u8], + return_value: call::Retval, ) -> Result<(), Error>; } diff --git a/graph/src/components/subgraph/host.rs b/graph/src/components/subgraph/host.rs index f133e0ac086..e7b9c4b8a74 100644 --- a/graph/src/components/subgraph/host.rs +++ b/graph/src/components/subgraph/host.rs @@ -48,6 +48,15 @@ impl MappingError { Unknown(e) => Unknown(e.context(s)), } } + + pub fn add_trigger_context(mut self, trigger: &TriggerData) -> MappingError { + let error_context = trigger.error_context(); + if !error_context.is_empty() { + self = self.context(error_context) + } + self = self.context("failed to process trigger".to_string()); + self + } } /// Common trait for runtime host implementations. @@ -96,6 +105,9 @@ pub trait RuntimeHost: Send + Sync + 'static { /// Convenience function to avoid leaking internal representation of /// mutable number. Calling this on OnChain Datasources is a noop. fn set_done_at(&self, block: Option); + + /// Return a metrics object for this host. + fn host_metrics(&self) -> Arc; } pub struct HostMetrics { diff --git a/graph/src/components/trigger_processor.rs b/graph/src/components/trigger_processor.rs index 03b7c84065d..f21fe5b7894 100644 --- a/graph/src/components/trigger_processor.rs +++ b/graph/src/components/trigger_processor.rs @@ -3,13 +3,37 @@ use std::sync::Arc; use async_trait::async_trait; use slog::Logger; -use crate::{blockchain::Blockchain, data_source::TriggerData, prelude::SubgraphInstanceMetrics}; +use crate::{ + blockchain::Blockchain, + data_source::{MappingTrigger, TriggerData, TriggerWithHandler}, + prelude::SubgraphInstanceMetrics, +}; use super::{ store::SubgraphFork, - subgraph::{BlockState, MappingError, RuntimeHostBuilder, SharedProofOfIndexing}, + subgraph::{BlockState, MappingError, RuntimeHost, RuntimeHostBuilder, SharedProofOfIndexing}, }; +/// A trigger that is almost ready to run: we have a host to run it on, and +/// transformed the `TriggerData` into a `MappingTrigger`. +pub struct HostedTrigger<'a, C> +where + C: Blockchain, +{ + pub host: &'a dyn RuntimeHost, + pub mapping_trigger: TriggerWithHandler>, +} + +/// The `TriggerData` and the `HostedTriggers` that were derived from it. We +/// need to hang on to the `TriggerData` solely for error reporting. +pub struct RunnableTriggers<'a, C> +where + C: Blockchain, +{ + pub trigger: TriggerData, + pub hosted_triggers: Vec>, +} + #[async_trait] pub trait TriggerProcessor: Sync + Send where @@ -19,9 +43,8 @@ where async fn process_trigger<'a>( &'a self, logger: &Logger, - hosts: Box + Send + 'a>, + triggers: Vec>, block: &Arc, - trigger: &TriggerData, mut state: BlockState, proof_of_indexing: &SharedProofOfIndexing, causality_region: &str, @@ -30,3 +53,44 @@ where instrument: bool, ) -> Result; } + +/// A trait for taking triggers as `TriggerData` (usually from the block +/// stream) and turning them into `HostedTrigger`s that are ready to run. +/// +/// The output triggers will be run in the order in which they are returned. +pub trait Decoder: Sync + Send +where + C: Blockchain, + T: RuntimeHostBuilder, +{ + fn match_and_decode<'a>( + &'a self, + logger: &Logger, + block: &Arc, + trigger: TriggerData, + hosts: Box + Send + 'a>, + subgraph_metrics: &Arc, + ) -> Result, MappingError>; + + fn match_and_decode_many<'a, F>( + &'a self, + logger: &Logger, + block: &Arc, + triggers: Box>>, + hosts_filter: F, + subgraph_metrics: &Arc, + ) -> Result>, MappingError> + where + F: Fn(&TriggerData) -> Box + Send + 'a>, + { + let mut runnables = vec![]; + for trigger in triggers { + let hosts = hosts_filter(&trigger); + match self.match_and_decode(logger, block, trigger, hosts, subgraph_metrics) { + Ok(runnable_triggers) => runnables.push(runnable_triggers), + Err(e) => return Err(e), + } + } + Ok(runnables) + } +} diff --git a/graph/src/data/store/ethereum.rs b/graph/src/data/store/ethereum.rs index 7bbf2f17542..6a277583de4 100644 --- a/graph/src/data/store/ethereum.rs +++ b/graph/src/data/store/ethereum.rs @@ -37,3 +37,120 @@ impl From for Value { Value::BigInt(BigInt::from(n)) } } + +/// Helper structs for dealing with ethereum calls +pub mod call { + use std::sync::Arc; + + use crate::data::store::scalar::Bytes; + + use super::CheapClone; + + /// The return value of an ethereum call. `Null` indicates that we made + /// the call but didn't get a value back (including when we get the + /// error 'call reverted') + #[derive(Debug, Clone, PartialEq)] + pub enum Retval { + Null, + Value(Bytes), + } + + impl Retval { + pub fn unwrap(self) -> Bytes { + use Retval::*; + match self { + Value(val) => val, + Null => panic!("called `call::Retval::unwrap()` on a `Null` value"), + } + } + } + + /// Indication of where the result of an ethereum call comes from. We + /// unfortunately need that so we can avoid double-counting declared calls + /// as they are accessed as normal eth calls and we'd count them twice + /// without this. + #[derive(Debug, Clone, Copy, PartialEq)] + pub enum Source { + Memory, + Store, + Rpc, + } + + impl Source { + /// Return `true` if calls from this source should be observed, + /// i.e., counted as actual calls + pub fn observe(&self) -> bool { + matches!(self, Source::Rpc | Source::Store) + } + } + + impl std::fmt::Display for Source { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + match self { + Source::Memory => write!(f, "memory"), + Source::Store => write!(f, "store"), + Source::Rpc => write!(f, "rpc"), + } + } + } + + /// The address and encoded name and parms for an `eth_call`, the raw + /// ingredients to make an `eth_call` request. Because we cache this, it + /// gets cloned a lot and needs to remain cheap to clone. + /// + /// For equality and hashing, we only consider the address and the + /// encoded call as the index is set by the caller and has no influence + /// on the call's return value + #[derive(Debug, Clone)] + pub struct Request { + pub address: ethabi::Address, + pub encoded_call: Arc, + /// The index is set by the caller and is used to identify the + /// request in related data structures that the caller might have + pub index: u32, + } + + impl CheapClone for Request {} + + impl Request { + pub fn new(address: ethabi::Address, encoded_call: Vec, index: u32) -> Self { + Request { + address, + encoded_call: Arc::new(Bytes::from(encoded_call)), + index, + } + } + + /// Create a response struct for this request + pub fn response(self, retval: Retval, source: Source) -> Response { + Response { + req: self, + retval, + source, + } + } + } + + impl PartialEq for Request { + fn eq(&self, other: &Self) -> bool { + self.address == other.address + && self.encoded_call.as_ref() == other.encoded_call.as_ref() + } + } + + impl Eq for Request {} + + impl std::hash::Hash for Request { + fn hash(&self, state: &mut H) { + self.address.hash(state); + self.encoded_call.as_ref().hash(state); + } + } + + #[derive(Debug, PartialEq)] + pub struct Response { + pub req: Request, + pub retval: Retval, + pub source: Source, + } +} diff --git a/graph/src/data/store/scalar.rs b/graph/src/data/store/scalar.rs index 78aa2b1b800..55b6ede03eb 100644 --- a/graph/src/data/store/scalar.rs +++ b/graph/src/data/store/scalar.rs @@ -589,7 +589,7 @@ impl GasSizeOf for BigInt { } /// A byte array that's serialized as a hex string prefixed by `0x`. -#[derive(Clone, PartialEq, Eq, PartialOrd, Ord)] +#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct Bytes(Box<[u8]>); impl Deref for Bytes { @@ -678,6 +678,12 @@ impl From> for Bytes { } } +impl AsRef<[u8]> for Bytes { + fn as_ref(&self) -> &[u8] { + self.0.as_ref() + } +} + impl ToSql for Bytes { fn to_sql<'b>( &'b self, diff --git a/graph/src/data/subgraph/api_version.rs b/graph/src/data/subgraph/api_version.rs index 691fac3ed48..95271b83db7 100644 --- a/graph/src/data/subgraph/api_version.rs +++ b/graph/src/data/subgraph/api_version.rs @@ -51,8 +51,11 @@ pub const SPEC_VERSION_1_0_0: Version = Version::new(1, 0, 0); // Enables `id: Int8` pub const SPEC_VERSION_1_1_0: Version = Version::new(1, 1, 0); +// Enables eth call declarations +pub const SPEC_VERSION_1_2_0: Version = Version::new(1, 2, 0); + // The latest spec version available -pub const LATEST_VERSION: &Version = &SPEC_VERSION_1_1_0; +pub const LATEST_VERSION: &Version = &SPEC_VERSION_1_2_0; pub const MIN_SPEC_VERSION: Version = Version::new(0, 0, 2); diff --git a/graph/src/data/subgraph/mod.rs b/graph/src/data/subgraph/mod.rs index c69cef1fe26..7c377a8e6e2 100644 --- a/graph/src/data/subgraph/mod.rs +++ b/graph/src/data/subgraph/mod.rs @@ -691,7 +691,7 @@ impl UnvalidatedSubgraphManifest { } for ds in &self.0.data_sources { - errors.extend(ds.validate().into_iter().map(|e| { + errors.extend(ds.validate(&self.0.spec_version).into_iter().map(|e| { SubgraphManifestValidationError::DataSourceValidation(ds.name().to_owned(), e) })); } diff --git a/graph/src/data_source/mod.rs b/graph/src/data_source/mod.rs index 2c982146c16..27e49927d02 100644 --- a/graph/src/data_source/mod.rs +++ b/graph/src/data_source/mod.rs @@ -236,9 +236,9 @@ impl DataSource { } } - pub fn validate(&self) -> Vec { + pub fn validate(&self, spec_version: &semver::Version) -> Vec { match self { - Self::Onchain(ds) => ds.validate(), + Self::Onchain(ds) => ds.validate(spec_version), Self::Offchain(_) => vec![], } } @@ -492,6 +492,13 @@ impl MappingTrigger { Self::Offchain(_) => None, // TODO: Add error context for offchain triggers } } + + pub fn as_onchain(&self) -> Option<&C::MappingTrigger> { + match self { + Self::Onchain(trigger) => Some(trigger), + Self::Offchain(_) => None, + } + } } macro_rules! clone_data_source { diff --git a/graph/src/env/mappings.rs b/graph/src/env/mappings.rs index c2fb81c0799..41499056b5b 100644 --- a/graph/src/env/mappings.rs +++ b/graph/src/env/mappings.rs @@ -57,6 +57,11 @@ pub struct EnvVarsMapping { /// Set by the flag `GRAPH_ALLOW_NON_DETERMINISTIC_IPFS`. Off by /// default. pub allow_non_deterministic_ipfs: bool, + + /// Set by the flag `GRAPH_DISABLE_DECLARED_CALLS`. Disables performing + /// eth calls before running triggers; instead eth calls happen when + /// mappings call `ethereum.call`. Off by default. + pub disable_declared_calls: bool, } // This does not print any values avoid accidentally leaking any sensitive env vars @@ -83,6 +88,7 @@ impl From for EnvVarsMapping { max_ipfs_file_bytes: x.max_ipfs_file_bytes.0, ipfs_request_limit: x.ipfs_request_limit, allow_non_deterministic_ipfs: x.allow_non_deterministic_ipfs.0, + disable_declared_calls: x.disable_declared_calls.0, } } } @@ -115,4 +121,6 @@ pub struct InnerMappingHandlers { ipfs_request_limit: u16, #[envconfig(from = "GRAPH_ALLOW_NON_DETERMINISTIC_IPFS", default = "false")] allow_non_deterministic_ipfs: EnvVarBoolean, + #[envconfig(from = "GRAPH_DISABLE_DECLARED_CALLS", default = "false")] + disable_declared_calls: EnvVarBoolean, } diff --git a/node/src/main.rs b/node/src/main.rs index 8fbac6f3a84..912ea9e86db 100644 --- a/node/src/main.rs +++ b/node/src/main.rs @@ -1,6 +1,9 @@ use clap::Parser as _; -use ethereum::chain::{EthereumAdapterSelector, EthereumBlockRefetcher, EthereumStreamBuilder}; -use ethereum::{BlockIngestor, EthereumNetworks, RuntimeAdapter}; +use ethereum::chain::{ + EthereumAdapterSelector, EthereumBlockRefetcher, EthereumRuntimeAdapterBuilder, + EthereumStreamBuilder, +}; +use ethereum::{BlockIngestor, EthereumNetworks}; use git_testament::{git_testament, render_testament}; use graph::blockchain::client::ChainClient; use graph_chain_ethereum::codec::HeaderOnlyBlock; @@ -867,11 +870,7 @@ fn ethereum_networks_as_chains( chain_store.clone(), ); - let runtime_adapter = Arc::new(RuntimeAdapter { - eth_adapters: Arc::new(eth_adapters.clone()), - call_cache: chain_store.cheap_clone(), - chain_identifier: Arc::new(chain_store.chain_identifier.clone()), - }); + let call_cache = chain_store.cheap_clone(); let chain_config = config.chains.chains.get(network_name).unwrap(); let chain = ethereum::Chain::new( @@ -880,13 +879,14 @@ fn ethereum_networks_as_chains( node_id.clone(), registry.clone(), chain_store.cheap_clone(), - chain_store, + call_cache, client, chain_head_update_listener.clone(), Arc::new(EthereumStreamBuilder {}), Arc::new(EthereumBlockRefetcher {}), Arc::new(adapter_selector), - runtime_adapter, + Arc::new(EthereumRuntimeAdapterBuilder {}), + Arc::new(eth_adapters.clone()), ENV_VARS.reorg_threshold, chain_config.polling_interval, is_ingestible, diff --git a/node/src/manager/commands/run.rs b/node/src/manager/commands/run.rs index 960b01d62b3..025c74b4a13 100644 --- a/node/src/manager/commands/run.rs +++ b/node/src/manager/commands/run.rs @@ -10,8 +10,11 @@ use crate::config::Config; use crate::manager::PanicSubscriptionManager; use crate::store_builder::StoreBuilder; use crate::MetricsContext; -use ethereum::chain::{EthereumAdapterSelector, EthereumBlockRefetcher, EthereumStreamBuilder}; -use ethereum::{ProviderEthRpcMetrics, RuntimeAdapter as EthereumRuntimeAdapter}; +use ethereum::chain::{ + EthereumAdapterSelector, EthereumBlockRefetcher, EthereumRuntimeAdapterBuilder, + EthereumStreamBuilder, +}; +use ethereum::ProviderEthRpcMetrics; use graph::anyhow::{bail, format_err}; use graph::blockchain::client::ChainClient; use graph::blockchain::{BlockchainKind, BlockchainMap}; @@ -125,7 +128,6 @@ pub async fn run( }; let eth_adapters2 = eth_adapters.clone(); - let (_, ethereum_idents) = connect_ethereum_networks(&logger, eth_networks).await?; // let (near_networks, near_idents) = connect_firehose_networks::( // &logger, @@ -147,6 +149,8 @@ pub async fn run( let client = Arc::new(ChainClient::new(firehose_endpoints, eth_adapters)); + let call_cache = Arc::new(ethereum::BufferedCallCache::new(chain_store.cheap_clone())); + let chain_config = config.chains.chains.get(&network_name).unwrap(); let chain = ethereum::Chain::new( logger_factory.clone(), @@ -154,7 +158,7 @@ pub async fn run( node_id.clone(), metrics_registry.clone(), chain_store.cheap_clone(), - chain_store.cheap_clone(), + call_cache.cheap_clone(), client.clone(), chain_head_update_listener, Arc::new(EthereumStreamBuilder {}), @@ -165,11 +169,8 @@ pub async fn run( metrics_registry.clone(), chain_store.cheap_clone(), )), - Arc::new(EthereumRuntimeAdapter { - call_cache: chain_store.cheap_clone(), - eth_adapters: Arc::new(eth_adapters2), - chain_identifier: Arc::new(chain_store.chain_identifier.clone()), - }), + Arc::new(EthereumRuntimeAdapterBuilder {}), + Arc::new(eth_adapters2), graph::env::ENV_VARS.reorg_threshold, chain_config.polling_interval, // We assume the tested chain is always ingestible for now diff --git a/runtime/wasm/src/host.rs b/runtime/wasm/src/host.rs index 661e3a65cec..5029025b645 100644 --- a/runtime/wasm/src/host.rs +++ b/runtime/wasm/src/host.rs @@ -373,6 +373,10 @@ impl RuntimeHostTrait for RuntimeHost { DataSource::Offchain(ds) => ds.set_done_at(block), } } + + fn host_metrics(&self) -> Arc { + self.metrics.cheap_clone() + } } impl PartialEq for RuntimeHost { diff --git a/store/postgres/src/chain_store.rs b/store/postgres/src/chain_store.rs index f00c8fb9a18..504431fcb5f 100644 --- a/store/postgres/src/chain_store.rs +++ b/store/postgres/src/chain_store.rs @@ -3,6 +3,7 @@ use diesel::prelude::*; use diesel::r2d2::{ConnectionManager, PooledConnection}; use diesel::sql_types::Text; use diesel::{insert_into, update}; +use graph::data::store::ethereum::call; use graph::env::ENV_VARS; use graph::parking_lot::RwLock; use graph::prelude::MetricsRegistry; @@ -22,9 +23,9 @@ use graph::blockchain::{Block, BlockHash, ChainIdentifier}; use graph::cheap_clone::CheapClone; use graph::prelude::web3::types::H256; use graph::prelude::{ - async_trait, ethabi, serde_json as json, transaction_receipt::LightTransactionReceipt, - BlockNumber, BlockPtr, CachedEthereumCall, CancelableError, ChainStore as ChainStoreTrait, - Error, EthereumCallCache, StoreError, + async_trait, serde_json as json, transaction_receipt::LightTransactionReceipt, BlockNumber, + BlockPtr, CachedEthereumCall, CancelableError, ChainStore as ChainStoreTrait, Error, + EthereumCallCache, StoreError, }; use graph::{constraint_violation, ensure}; @@ -87,6 +88,7 @@ mod data { }; use graph::blockchain::{Block, BlockHash}; use graph::constraint_violation; + use graph::data::store::scalar::Bytes; use graph::prelude::ethabi::ethereum_types::H160; use graph::prelude::transaction_receipt::LightTransactionReceipt; use graph::prelude::web3::types::H256; @@ -1117,7 +1119,7 @@ mod data { &self, conn: &mut PgConnection, id: &[u8], - ) -> Result, bool)>, Error> { + ) -> Result, Error> { match self { Storage::Shared => { use public::eth_call_cache as cache; @@ -1154,10 +1156,62 @@ mod data { CallMetaTable::ACCESSED_AT )), )) - .first(conn) + .first::<(Vec, bool)>(conn) .optional() .map_err(Error::from), } + .map(|row| row.map(|(return_value, expired)| (Bytes::from(return_value), expired))) + } + + pub(super) fn get_calls_and_access( + &self, + conn: &mut PgConnection, + ids: &[&[u8]], + ) -> Result, Bytes, bool)>, Error> { + let rows = match self { + Storage::Shared => { + use public::eth_call_cache as cache; + use public::eth_call_meta as meta; + + cache::table + .inner_join(meta::table) + .filter(cache::id.eq_any(ids)) + .select(( + cache::id, + cache::return_value, + sql::("CURRENT_DATE > eth_call_meta.accessed_at"), + )) + .load(conn) + .map_err(Error::from) + } + Storage::Private(Schema { + call_cache, + call_meta, + .. + }) => call_cache + .table() + .inner_join( + call_meta.table().on(call_meta + .contract_address() + .eq(call_cache.contract_address())), + ) + .filter(call_cache.id().eq_any(ids)) + .select(( + call_cache.id(), + call_cache.return_value(), + sql::(&format!( + "CURRENT_DATE > {}.{}", + CallMetaTable::TABLE_NAME, + CallMetaTable::ACCESSED_AT + )), + )) + .load::<(Vec, Vec, bool)>(conn) + .map_err(Error::from), + }?; + Ok(rows + .into_iter() + .map(|(id, return_value, expired)| (id, Bytes::from(return_value), expired)) + .collect()) } pub(super) fn get_calls_in_block( @@ -2195,6 +2249,10 @@ impl ChainStoreTrait for ChainStore { }) .await } + + fn chain_identifier(&self) -> &ChainIdentifier { + &self.chain_identifier + } } mod recent_blocks_cache { @@ -2389,29 +2447,71 @@ fn try_parse_timestamp(ts: Option) -> Result, StoreError> { impl EthereumCallCache for ChainStore { fn get_call( &self, - contract_address: ethabi::Address, - encoded_call: &[u8], + req: &call::Request, block: BlockPtr, - ) -> Result>, Error> { - let id = contract_call_id(&contract_address, encoded_call, &block); + ) -> Result, Error> { + let id = contract_call_id(req, &block); let conn = &mut *self.get_conn()?; - if let Some(call_output) = conn.transaction::<_, Error, _>(|conn| { + let return_value = conn.transaction::<_, Error, _>(|conn| { if let Some((return_value, update_accessed_at)) = self.storage.get_call_and_access(conn, id.as_ref())? { if update_accessed_at { self.storage - .update_accessed_at(conn, contract_address.as_ref())?; + .update_accessed_at(conn, req.address.as_ref())?; } Ok(Some(return_value)) } else { Ok(None) } - })? { - Ok(Some(call_output)) - } else { - Ok(None) + })?; + Ok(return_value.map(|return_value| { + req.cheap_clone() + .response(call::Retval::Value(return_value), call::Source::Store) + })) + } + + fn get_calls( + &self, + reqs: &[call::Request], + block: BlockPtr, + ) -> Result<(Vec, Vec), Error> { + if reqs.is_empty() { + return Ok((Vec::new(), Vec::new())); } + + let ids: Vec<_> = reqs + .into_iter() + .map(|req| contract_call_id(req, &block)) + .collect(); + let id_refs: Vec<_> = ids.iter().map(|id| id.as_slice()).collect(); + + let conn = &mut *self.get_conn()?; + let rows = conn + .transaction::<_, Error, _>(|conn| self.storage.get_calls_and_access(conn, &id_refs))?; + + let mut found: Vec = Vec::new(); + let mut resps = Vec::new(); + for (id, retval, _) in rows { + let idx = ids.iter().position(|i| i.as_ref() == id).ok_or_else(|| { + constraint_violation!( + "get_calls returned a call id that was not requested: {}", + hex::encode(id) + ) + })?; + found.push(idx); + let resp = reqs[idx] + .cheap_clone() + .response(call::Retval::Value(retval), call::Source::Store); + resps.push(resp); + } + let calls = reqs + .into_iter() + .enumerate() + .filter(|(idx, _)| !found.contains(&idx)) + .map(|(_, call)| call.cheap_clone()) + .collect(); + Ok((resps, calls)) } fn get_calls_in_block(&self, block: BlockPtr) -> Result, Error> { @@ -2421,20 +2521,27 @@ impl EthereumCallCache for ChainStore { fn set_call( &self, - contract_address: ethabi::Address, - encoded_call: &[u8], + _: &Logger, + call: call::Request, block: BlockPtr, - return_value: &[u8], + return_value: call::Retval, ) -> Result<(), Error> { - let id = contract_call_id(&contract_address, encoded_call, &block); + let call::Retval::Value(return_value) = return_value else { + // We do not want to cache unsuccessful calls as some RPC nodes + // have weird behavior near the chain head. The details are lost + // to time, but we had issues with some RPC clients in the past + // where calls first failed and later succeeded + return Ok(()); + }; + let id = contract_call_id(&call, &block); let conn = &mut *self.get_conn()?; conn.transaction(|conn| { self.storage.set_call( conn, id.as_ref(), - contract_address.as_ref(), + call.address.as_ref(), block.number, - return_value, + &return_value, ) }) } @@ -2443,14 +2550,10 @@ impl EthereumCallCache for ChainStore { /// The id is the hashed encoded_call + contract_address + block hash to uniquely identify the call. /// 256 bits of output, and therefore 128 bits of security against collisions, are needed since this /// could be targeted by a birthday attack. -fn contract_call_id( - contract_address: ðabi::Address, - encoded_call: &[u8], - block: &BlockPtr, -) -> [u8; 32] { +fn contract_call_id(call: &call::Request, block: &BlockPtr) -> [u8; 32] { let mut hash = blake3::Hasher::new(); - hash.update(encoded_call); - hash.update(contract_address.as_ref()); + hash.update(&call.encoded_call); + hash.update(call.address.as_ref()); hash.update(block.hash_slice()); *hash.finalize().as_bytes() } diff --git a/store/test-store/tests/chain/ethereum/manifest.rs b/store/test-store/tests/chain/ethereum/manifest.rs index 2aca9d5f759..4a8a2cca70c 100644 --- a/store/test-store/tests/chain/ethereum/manifest.rs +++ b/store/test-store/tests/chain/ethereum/manifest.rs @@ -10,8 +10,8 @@ use graph::data::store::scalar::Bytes; use graph::data::store::Value; use graph::data::subgraph::schema::SubgraphError; use graph::data::subgraph::{ - Prune, SPEC_VERSION_0_0_4, SPEC_VERSION_0_0_7, SPEC_VERSION_0_0_8, SPEC_VERSION_0_0_9, - SPEC_VERSION_1_0_0, + Prune, LATEST_VERSION, SPEC_VERSION_0_0_4, SPEC_VERSION_0_0_7, SPEC_VERSION_0_0_8, + SPEC_VERSION_0_0_9, SPEC_VERSION_1_0_0, SPEC_VERSION_1_2_0, }; use graph::data_source::offchain::OffchainDataSourceKind; use graph::data_source::DataSourceTemplate; @@ -592,7 +592,7 @@ specVersion: 0.0.8 .collect::>(); let data_source = onchain_data_sources.get(0).unwrap(); - let validation_errors = data_source.validate(); + let validation_errors = data_source.validate(&LATEST_VERSION); let filter = data_source.mapping.block_handlers[0].filter.clone(); assert_eq!(0, validation_errors.len()); @@ -688,7 +688,7 @@ specVersion: 0.0.8 .collect::>(); let data_source = onchain_data_sources.get(0).unwrap(); - let validation_errors = data_source.validate(); + let validation_errors = data_source.validate(LATEST_VERSION); let filters = data_source .mapping .block_handlers @@ -753,7 +753,7 @@ specVersion: 0.0.8 .collect::>(); let data_source = onchain_data_sources.get(0).unwrap(); - let validation_errors = data_source.validate(); + let validation_errors = data_source.validate(LATEST_VERSION); let filters = data_source .mapping .block_handlers @@ -1302,3 +1302,72 @@ schema: assert!(manifest.features.contains(&SubgraphFeature::NonFatalErrors)) }); } + +#[test] +fn parses_eth_call_decls() { + const YAML: &str = " +specVersion: 1.2.0 +schema: + file: + /: /ipfs/Qmschema +features: + - ipfsOnEthereumContracts +dataSources: + - kind: ethereum/contract + name: Factory + network: mainnet + source: + abi: Factory + startBlock: 9562480 + mapping: + kind: ethereum/events + apiVersion: 0.0.4 + language: wasm/assemblyscript + entities: + - TestEntity + file: + /: /ipfs/Qmmapping + abis: + - name: Factory + file: + /: /ipfs/Qmabi + eventHandlers: + - event: Created(address) + handler: handleGet + calls: + fake1: Factory[event.address].get(event.params.address) + fake2: Factory[event.params.address].get(event.params.address) +"; + + test_store::run_test_sequentially(|store| async move { + let store = store.subgraph_store(); + let unvalidated: UnvalidatedSubgraphManifest = { + let mut resolver = TextResolver::default(); + let id = DeploymentHash::new("Qmmanifest").unwrap(); + resolver.add(id.as_str(), &YAML); + resolver.add("/ipfs/Qmabi", &ABI); + resolver.add("/ipfs/Qmschema", &GQL_SCHEMA); + resolver.add("/ipfs/Qmmapping", &MAPPING_WITH_IPFS_FUNC_WASM); + + let resolver: Arc = Arc::new(resolver); + + let raw = serde_yaml::from_str(YAML).unwrap(); + UnvalidatedSubgraphManifest::resolve( + id, + raw, + &resolver, + &LOGGER, + SPEC_VERSION_1_2_0.clone(), + ) + .await + .expect("Parsing simple manifest works") + }; + + let manifest = unvalidated.validate(store.clone(), true).await.unwrap(); + let ds = &manifest.data_sources[0].as_onchain().unwrap(); + // For more detailed tests of parsing CallDecls see the data_soure + // module in chain/ethereum + let decls = &ds.mapping.event_handlers[0].calls.decls; + assert_eq!(2, decls.len()); + }); +} diff --git a/store/test-store/tests/postgres/chain_head.rs b/store/test-store/tests/postgres/chain_head.rs index ef9e00b3f59..230d3ceaa71 100644 --- a/store/test-store/tests/postgres/chain_head.rs +++ b/store/test-store/tests/postgres/chain_head.rs @@ -2,6 +2,8 @@ //! the chain head pointer gets updated in various situations use graph::blockchain::{BlockHash, BlockPtr}; +use graph::data::store::ethereum::call; +use graph::data::store::scalar::Bytes; use graph::env::ENV_VARS; use graph::prelude::futures03::executor; use std::future::Future; @@ -373,40 +375,67 @@ fn eth_call_cache() { let chain = vec![&*GENESIS_BLOCK, &*BLOCK_ONE, &*BLOCK_TWO]; run_test(chain, |store, _| { + let logger = LOGGER.cheap_clone(); + fn ccr(value: &[u8]) -> call::Retval { + call::Retval::Value(Bytes::from(value)) + } + let address = H160([1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]); let call: [u8; 6] = [1, 2, 3, 4, 5, 6]; let return_value: [u8; 3] = [7, 8, 9]; + let call = call::Request::new(address, call.to_vec(), 0); store - .set_call(address, &call, BLOCK_ONE.block_ptr(), &return_value) + .set_call( + &logger, + call.cheap_clone(), + BLOCK_ONE.block_ptr(), + ccr(&return_value), + ) .unwrap(); - let ret = store - .get_call(address, &call, GENESIS_BLOCK.block_ptr()) - .unwrap(); + let ret = store.get_call(&call, GENESIS_BLOCK.block_ptr()).unwrap(); assert!(ret.is_none()); let ret = store - .get_call(address, &call, BLOCK_ONE.block_ptr()) + .get_call(&call, BLOCK_ONE.block_ptr()) + .unwrap() .unwrap() + .retval .unwrap(); assert_eq!(&return_value, ret.as_slice()); - let ret = store - .get_call(address, &call, BLOCK_TWO.block_ptr()) - .unwrap(); + let ret = store.get_call(&call, BLOCK_TWO.block_ptr()).unwrap(); assert!(ret.is_none()); let new_return_value: [u8; 3] = [10, 11, 12]; store - .set_call(address, &call, BLOCK_TWO.block_ptr(), &new_return_value) + .set_call( + &logger, + call.cheap_clone(), + BLOCK_TWO.block_ptr(), + ccr(&new_return_value), + ) .unwrap(); let ret = store - .get_call(address, &call, BLOCK_TWO.block_ptr()) + .get_call(&call, BLOCK_TWO.block_ptr()) + .unwrap() .unwrap() + .retval .unwrap(); assert_eq!(&new_return_value, ret.as_slice()); + store + .set_call( + &logger, + call.cheap_clone(), + BLOCK_THREE.block_ptr(), + call::Retval::Null, + ) + .unwrap(); + let ret = store.get_call(&call, BLOCK_THREE.block_ptr()).unwrap(); + assert_eq!(None, ret); + Ok(()) }) } diff --git a/tests/src/fixture/ethereum.rs b/tests/src/fixture/ethereum.rs index b59d10c9e31..faa100be7b2 100644 --- a/tests/src/fixture/ethereum.rs +++ b/tests/src/fixture/ethereum.rs @@ -3,8 +3,8 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use super::{ - test_ptr, CommonChainConfig, MutexBlockStreamBuilder, NoopAdapterSelector, NoopRuntimeAdapter, - StaticBlockRefetcher, StaticStreamBuilder, Stores, TestChain, + test_ptr, CommonChainConfig, MutexBlockStreamBuilder, NoopAdapterSelector, + NoopRuntimeAdapterBuilder, StaticBlockRefetcher, StaticStreamBuilder, Stores, TestChain, }; use graph::blockchain::client::ChainClient; use graph::blockchain::{BlockPtr, TriggersAdapterSelector}; @@ -13,6 +13,7 @@ use graph::prelude::ethabi::ethereum_types::H256; use graph::prelude::web3::types::{Address, Log, Transaction, H160}; use graph::prelude::{ethabi, tiny_keccak, LightEthereumBlock, ENV_VARS}; use graph::{blockchain::block_stream::BlockWithTriggers, prelude::ethabi::ethereum_types::U64}; +use graph_chain_ethereum::network::EthereumNetworkAdapters; use graph_chain_ethereum::trigger::LogRef; use graph_chain_ethereum::Chain; use graph_chain_ethereum::{ @@ -44,6 +45,8 @@ pub async fn chain( let static_block_stream = Arc::new(StaticStreamBuilder { chain: blocks }); let block_stream_builder = Arc::new(MutexBlockStreamBuilder(Mutex::new(static_block_stream))); + let eth_adapters = Arc::new(EthereumNetworkAdapters::default()); + let chain = Chain::new( logger_factory, stores.network_name.clone(), @@ -56,7 +59,8 @@ pub async fn chain( block_stream_builder.clone(), Arc::new(StaticBlockRefetcher { x: PhantomData }), triggers_adapter, - Arc::new(NoopRuntimeAdapter { x: PhantomData }), + Arc::new(NoopRuntimeAdapterBuilder {}), + eth_adapters, ENV_VARS.reorg_threshold, ENV_VARS.ingestor_polling_interval, // We assume the tested chain is always ingestible for now diff --git a/tests/src/fixture/mod.rs b/tests/src/fixture/mod.rs index 6dbdaf97688..19b8de69fe0 100644 --- a/tests/src/fixture/mod.rs +++ b/tests/src/fixture/mod.rs @@ -20,7 +20,7 @@ use graph::blockchain::{ use graph::cheap_clone::CheapClone; use graph::components::link_resolver::{ArweaveClient, ArweaveResolver, FileSizeLimit}; use graph::components::metrics::MetricsRegistry; -use graph::components::store::{BlockStore, DeploymentLocator}; +use graph::components::store::{BlockStore, DeploymentLocator, EthereumCallCache}; use graph::components::subgraph::Settings; use graph::data::graphql::load_manager::LoadManager; use graph::data::query::{Query, QueryTarget}; @@ -38,6 +38,8 @@ use graph::prelude::{ SubgraphStore as _, SubgraphVersionSwitchingMode, TriggerProcessor, }; use graph::schema::InputSchema; +use graph_chain_ethereum::chain::RuntimeAdapterBuilder; +use graph_chain_ethereum::network::EthereumNetworkAdapters; use graph_chain_ethereum::Chain; use graph_core::polling_monitor::{arweave_service, ipfs_service}; use graph_core::{ @@ -224,7 +226,9 @@ impl TestContext { RuntimeHostBuilder, > { let (logger, deployment, raw) = self.get_runner_context().await; - let tp: Box> = Box::new(SubgraphTriggerProcessor {}); + let tp: Box> = Box::new( + graph_chain_substreams::TriggerProcessor::new(deployment.clone()), + ); self.instance_manager .build_subgraph_runner( @@ -867,6 +871,19 @@ impl RuntimeAdapter for NoopRuntimeAdapter { } } +struct NoopRuntimeAdapterBuilder {} + +impl RuntimeAdapterBuilder for NoopRuntimeAdapterBuilder { + fn build( + &self, + _: Arc, + _: Arc, + _: Arc, + ) -> Arc + 'static> { + Arc::new(NoopRuntimeAdapter { x: PhantomData }) + } +} + pub struct NoopAdapterSelector { pub x: PhantomData, pub triggers_in_block_sleep: Duration,