graph: enable ethereum host fns for subgraph datasource

Author: incrypto32

chain/ethereum/src/data_source.rs

@@ -6,6 +6,7 @@ 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::common::{MappingABI, UnresolvedMappingABI};
 use graph::data_source::CausalityRegion;
 use graph::env::ENV_VARS;
 use graph::futures03::future::try_join;
@@ -33,7 +34,7 @@ use graph::{
     derive::CheapClone,
     prelude::{
         async_trait,
-        ethabi::{Address, Contract, Event, Function, LogParam, ParamType, RawLog},
+        ethabi::{Address, Event, Function, LogParam, ParamType, RawLog},
         serde_json, warn,
         web3::types::{Log, Transaction, H256},
         BlockNumber, CheapClone, EthereumCall, LightEthereumBlock, LightEthereumBlockExt,
@@ -1436,82 +1437,6 @@ impl UnresolvedMapping {
     }
 }
 
-#[derive(Clone, Debug, Hash, Eq, PartialEq, Deserialize)]
-pub struct UnresolvedMappingABI {
-    pub name: String,
-    pub file: Link,
-}
-
-impl UnresolvedMappingABI {
-    pub async fn resolve(
-        self,
-        resolver: &Arc<dyn LinkResolver>,
-        logger: &Logger,
-    ) -> Result<MappingABI, anyhow::Error> {
-        let contract_bytes = resolver.cat(logger, &self.file).await.with_context(|| {
-            format!(
-                "failed to resolve ABI {} from {}",
-                self.name, self.file.link
-            )
-        })?;
-        let contract = Contract::load(&*contract_bytes)?;
-        Ok(MappingABI {
-            name: self.name,
-            contract,
-        })
-    }
-}
-
-#[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,

chain/ethereum/src/lib.rs

@@ -19,7 +19,7 @@ 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,
+    BlockHandlerFilter, DataSource, DataSourceTemplate, Mapping, TemplateSource,
 };
 
 pub mod chain;

chain/ethereum/src/runtime/runtime_adapter.rs

@@ -1,10 +1,9 @@
 use std::{sync::Arc, time::Instant};
 
 use crate::adapter::EthereumRpcError;
-use crate::data_source::MappingABI;
 use crate::{
     capabilities::NodeCapabilities, network::EthereumNetworkAdapters, Chain, ContractCall,
-    ContractCallError, DataSource, EthereumAdapter, EthereumAdapterTrait, ENV_VARS,
+    ContractCallError, EthereumAdapter, EthereumAdapterTrait, ENV_VARS,
 };
 use anyhow::{anyhow, Context, Error};
 use blockchain::HostFn;
@@ -13,6 +12,8 @@ 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::data_source;
+use graph::data_source::common::MappingABI;
 use graph::futures03::compat::Future01CompatExt;
 use graph::prelude::web3::types::H160;
 use graph::runtime::gas::Gas;
@@ -80,58 +81,93 @@ pub fn eth_call_gas(chain_identifier: &ChainIdentifier) -> Option<u32> {
 }
 
 impl blockchain::RuntimeAdapter<Chain> for RuntimeAdapter {
-    fn host_fns(&self, ds: &DataSource) -> Result<Vec<HostFn>, 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()?;
-        let eth_call_gas = eth_call_gas(&self.chain_identifier);
-
-        let ethereum_call = HostFn {
-            name: "ethereum.call",
-            func: Arc::new(move |ctx, wasm_ptr| {
-                // Ethereum calls should prioritise call-only adapters if one is available.
-                let eth_adapter = eth_adapters.call_or_cheapest(Some(&NodeCapabilities {
-                    archive,
-                    traces: false,
-                }))?;
-                ethereum_call(
-                    &eth_adapter,
-                    call_cache.cheap_clone(),
-                    ctx,
-                    wasm_ptr,
-                    &abis,
-                    eth_call_gas,
-                )
-                .map(|ptr| ptr.wasm_ptr())
-            }),
-        };
-
-        let eth_adapters = self.eth_adapters.cheap_clone();
-        let ethereum_get_balance = HostFn {
-            name: "ethereum.getBalance",
-            func: Arc::new(move |ctx, wasm_ptr| {
-                let eth_adapter = eth_adapters.unverified_cheapest_with(&NodeCapabilities {
-                    archive,
-                    traces: false,
-                })?;
-                eth_get_balance(&eth_adapter, ctx, wasm_ptr).map(|ptr| ptr.wasm_ptr())
-            }),
-        };
+    fn host_fns(&self, ds: &data_source::DataSource<Chain>) -> Result<Vec<HostFn>, Error> {
+        fn create_host_fns(
+            abis: Arc<Vec<Arc<MappingABI>>>, // Use Arc to ensure `'static` lifetimes.
+            archive: bool,
+            call_cache: Arc<dyn EthereumCallCache>,
+            eth_adapters: Arc<EthereumNetworkAdapters>,
+            eth_call_gas: Option<u32>,
+        ) -> Vec<HostFn> {
+            vec![
+                HostFn {
+                    name: "ethereum.call",
+                    func: Arc::new({
+                        let eth_adapters = eth_adapters.clone();
+                        let call_cache = call_cache.clone();
+                        let abis = abis.clone();
+                        move |ctx, wasm_ptr| {
+                            let eth_adapter =
+                                eth_adapters.call_or_cheapest(Some(&NodeCapabilities {
+                                    archive,
+                                    traces: false,
+                                }))?;
+                            ethereum_call(
+                                &eth_adapter,
+                                call_cache.clone(),
+                                ctx,
+                                wasm_ptr,
+                                &abis,
+                                eth_call_gas,
+                            )
+                            .map(|ptr| ptr.wasm_ptr())
+                        }
+                    }),
+                },
+                HostFn {
+                    name: "ethereum.getBalance",
+                    func: Arc::new({
+                        let eth_adapters = eth_adapters.clone();
+                        move |ctx, wasm_ptr| {
+                            let eth_adapter =
+                                eth_adapters.unverified_cheapest_with(&NodeCapabilities {
+                                    archive,
+                                    traces: false,
+                                })?;
+                            eth_get_balance(&eth_adapter, ctx, wasm_ptr).map(|ptr| ptr.wasm_ptr())
+                        }
+                    }),
+                },
+                HostFn {
+                    name: "ethereum.hasCode",
+                    func: Arc::new({
+                        let eth_adapters = eth_adapters.clone();
+                        move |ctx, wasm_ptr| {
+                            let eth_adapter =
+                                eth_adapters.unverified_cheapest_with(&NodeCapabilities {
+                                    archive,
+                                    traces: false,
+                                })?;
+                            eth_has_code(&eth_adapter, ctx, wasm_ptr).map(|ptr| ptr.wasm_ptr())
+                        }
+                    }),
+                },
+            ]
+        }
 
-        let eth_adapters = self.eth_adapters.cheap_clone();
-        let ethereum_get_code = HostFn {
-            name: "ethereum.hasCode",
-            func: Arc::new(move |ctx, wasm_ptr| {
-                let eth_adapter = eth_adapters.unverified_cheapest_with(&NodeCapabilities {
-                    archive,
-                    traces: false,
-                })?;
-                eth_has_code(&eth_adapter, ctx, wasm_ptr).map(|ptr| ptr.wasm_ptr())
-            }),
+        let host_fns = match ds {
+            data_source::DataSource::Onchain(onchain_ds) => {
+                let abis = Arc::new(onchain_ds.mapping.abis.clone());
+                let archive = onchain_ds.mapping.requires_archive()?;
+                let call_cache = self.call_cache.cheap_clone();
+                let eth_adapters = self.eth_adapters.cheap_clone();
+                let eth_call_gas = eth_call_gas(&self.chain_identifier);
+
+                create_host_fns(abis, archive, call_cache, eth_adapters, eth_call_gas)
+            }
+            data_source::DataSource::Subgraph(subgraph_ds) => {
+                let abis = Arc::new(subgraph_ds.mapping.abis.clone());
+                let archive = subgraph_ds.mapping.requires_archive()?;
+                let call_cache = self.call_cache.cheap_clone();
+                let eth_adapters = self.eth_adapters.cheap_clone();
+                let eth_call_gas = eth_call_gas(&self.chain_identifier);
+
+                create_host_fns(abis, archive, call_cache, eth_adapters, eth_call_gas)
+            }
+            data_source::DataSource::Offchain(_) => vec![],
         };
 
-        Ok(vec![ethereum_call, ethereum_get_balance, ethereum_get_code])
+        Ok(host_fns)
     }
 }
 

graph/src/blockchain/mock.rs

@@ -6,6 +6,7 @@ use crate::{
         subgraph::InstanceDSTemplateInfo,
     },
     data::subgraph::UnifiedMappingApiVersion,
+    data_source,
     prelude::{
         transaction_receipt::LightTransactionReceipt, BlockHash, ChainStore,
         DataSourceTemplateInfo, StoreError,
@@ -352,7 +353,7 @@ impl<C: Blockchain> TriggerFilter<C> for MockTriggerFilter {
 pub struct MockRuntimeAdapter;
 
 impl<C: Blockchain> RuntimeAdapter<C> for MockRuntimeAdapter {
-    fn host_fns(&self, _ds: &C::DataSource) -> Result<Vec<HostFn>, Error> {
+    fn host_fns(&self, _ds: &data_source::DataSource<C>) -> Result<Vec<HostFn>, Error> {
         todo!()
     }
 }

graph/src/blockchain/mod.rs

@@ -458,7 +458,6 @@ where
     }
 }
 
-// TODO(krishna): Proper ordering for triggers
 impl<C: Blockchain> Ord for Trigger<C>
 where
     C::TriggerData: Ord,
@@ -468,7 +467,7 @@ where
             (Trigger::Chain(data1), Trigger::Chain(data2)) => data1.cmp(data2),
             (Trigger::Subgraph(_), Trigger::Chain(_)) => std::cmp::Ordering::Greater,
             (Trigger::Chain(_), Trigger::Subgraph(_)) => std::cmp::Ordering::Less,
-            (Trigger::Subgraph(_), Trigger::Subgraph(_)) => std::cmp::Ordering::Equal,
+            (Trigger::Subgraph(t1), Trigger::Subgraph(t2)) => t1.entity.vid.cmp(&t2.entity.vid),
         }
     }
 }
@@ -545,7 +544,7 @@ pub struct HostFn {
 }
 
 pub trait RuntimeAdapter<C: Blockchain>: Send + Sync {
-    fn host_fns(&self, ds: &C::DataSource) -> Result<Vec<HostFn>, Error>;
+    fn host_fns(&self, ds: &data_source::DataSource<C>) -> Result<Vec<HostFn>, Error>;
 }
 
 pub trait NodeCapabilities<C: Blockchain> {

graph/src/blockchain/noop_runtime_adapter.rs

@@ -1,5 +1,7 @@
 use std::marker::PhantomData;
 
+use crate::data_source;
+
 use super::{Blockchain, HostFn, RuntimeAdapter};
 
 /// A [`RuntimeAdapter`] that does not expose any host functions.
@@ -16,7 +18,7 @@ impl<C> RuntimeAdapter<C> for NoopRuntimeAdapter<C>
 where
     C: Blockchain,
 {
-    fn host_fns(&self, _ds: &C::DataSource) -> anyhow::Result<Vec<HostFn>> {
+    fn host_fns(&self, _ds: &data_source::DataSource<C>) -> anyhow::Result<Vec<HostFn>> {
         Ok(vec![])
     }
 }