Declarative calls

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<Self>;
 
+    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<dyn RuntimeAdapterTrait<Self>> {
-        Arc::new(NoopRuntimeAdapter::default())
+    fn runtime(&self) -> (Arc<dyn RuntimeAdapterTrait<Self>>, Self::DecoderHook) {
+        (Arc::new(NoopRuntimeAdapter::default()), NoopDecoderHook)
     }
 
     fn chain_client(&self) -> Arc<ChainClient<Self>> {

chain/arweave/src/data_source.rs

@@ -157,7 +157,7 @@ impl blockchain::DataSource<Chain> for DataSource {
         todo!()
     }
 
-    fn validate(&self) -> Vec<Error> {
+    fn validate(&self, _: &semver::Version) -> Vec<Error> {
         let mut errors = Vec::new();
 
         if self.kind != ARWEAVE_KIND {

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<Self>;
 
+    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<dyn RuntimeAdapterTrait<Self>> {
-        Arc::new(NoopRuntimeAdapter::default())
+    fn runtime(&self) -> (Arc<dyn RuntimeAdapterTrait<Self>>, Self::DecoderHook) {
+        (Arc::new(NoopRuntimeAdapter::default()), NoopDecoderHook)
     }
 
     fn chain_client(&self) -> Arc<ChainClient<Self>> {

chain/cosmos/src/data_source.rs

@@ -193,7 +193,7 @@ impl blockchain::DataSource<Chain> for DataSource {
         Err(anyhow!(DYNAMIC_DATA_SOURCE_ERROR))
     }
 
-    fn validate(&self) -> Vec<Error> {
+    fn validate(&self, _: &semver::Version) -> Vec<Error> {
         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());
 

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,15 +48,15 @@ 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")]
     Timeout,
 }
 
 #[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<ContractCallError> 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<dyn Future<Item = Option<H256>, 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<dyn EthereumCallCache>,
+    ) -> Result<(Option<Vec<Token>>, 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<dyn EthereumCallCache>,
-    ) -> Box<dyn Future<Item = Vec<Token>, Error = EthereumContractCallError> + Send>;
+    ) -> Result<Vec<(Option<Vec<Token>>, call::Source)>, ContractCallError>;
 
     fn get_balance(
         &self,
         logger: &Logger,
         address: H160,
         block_ptr: BlockPtr,
-    ) -> Box<dyn Future<Item = U256, Error = EthereumGetBalanceError> + Send>;
+    ) -> Box<dyn Future<Item = U256, Error = GetBalanceError> + Send>;
 }
 
 #[cfg(test)]

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<dyn EthereumCallCache>,
+    buffer: Arc<Mutex<HashMap<call::Request, call::Retval>>>,
+    block: Arc<Mutex<Option<BlockPtr>>>,
+}
+
+impl BufferedCallCache {
+    pub fn new(call_cache: Arc<dyn EthereumCallCache>) -> 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<call::Response> {
+        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<Option<call::Response>, 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<call::Response>, Vec<call::Request>), 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<Vec<CachedEthereumCall>, 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(())
+    }
+}