runtime_adapter.rs

use std::{sync::Arc, time::Instant};

use async_trait::async_trait;

use crate::adapter::EthereumRpcError;
use crate::{
    capabilities::NodeCapabilities, network::EthereumNetworkAdapters, Chain, ContractCallError,
    EthereumAdapter, EthereumAdapterTrait, ENV_VARS,
};
use anyhow::{anyhow, Context, Error};
use blockchain::HostFn;
use graph::abi;
use graph::abi::DynSolValueExt;
use graph::blockchain::{ChainIdentifier, RawEthCall};
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_4, API_VERSION_0_0_9};
use graph::data_source;
use graph::data_source::common::{ContractCall, MappingABI};
use graph::runtime::gas::Gas;
use graph::runtime::{AscIndexId, IndexForAscTypeId};
use graph::slog::{debug, o, Discard};
use graph::{
    blockchain::{self, BlockPtr, HostFnCtx},
    cheap_clone::CheapClone,
    futures03::FutureExt,
    prelude::{alloy::primitives::Address, EthereumCallCache},
    runtime::{asc_get, asc_new, AscPtr, HostExportError},
    slog::Logger,
};
use graph_runtime_wasm::asc_abi::class::{AscBigInt, AscEnumArray, AscWrapped, EthereumValueKind};
use itertools::Itertools;

use super::abi::{AscUnresolvedContractCall, AscUnresolvedContractCall_0_0_4};

/// Gas limit for `eth_call`. The value of 50_000_000 is a protocol-wide parameter so this
/// should be changed only for debugging purposes and never on an indexer in the network. This
/// value was chosen because it is the Geth default
/// https://github.com/ethereum/go-ethereum/blob/e4b687cf462870538743b3218906940ae590e7fd/eth/ethconfig/config.go#L91.
/// It is not safe to set something higher because Geth will silently override the gas limit
/// with the default. This means that we do not support indexing against a Geth node with
/// `RPCGasCap` set below 50 million.
// See also f0af4ab0-6b7c-4b68-9141-5b79346a5f61.
const ETH_CALL_GAS: u32 = 50_000_000;

// When making an ethereum call, the maximum ethereum gas is ETH_CALL_GAS which is 50 million. One
// unit of Ethereum gas is at least 100ns according to these benchmarks [1], so 1000 of our gas. In
// the worst case an Ethereum call could therefore consume 50 billion of our gas. However the
// averarge call a subgraph makes is much cheaper or even cached in the call cache. So this cost is
// set to 5 billion gas as a compromise. This allows for 2000 calls per handler with the current
// limits.
//
// [1] - https://www.sciencedirect.com/science/article/abs/pii/S0166531620300900
pub const ETHEREUM_CALL: Gas = Gas::new(5_000_000_000);

// TODO: Determine the appropriate gas cost for `ETH_GET_BALANCE`, initially aligned with `ETHEREUM_CALL`.
pub const ETH_GET_BALANCE: Gas = Gas::new(5_000_000_000);

// TODO: Determine the appropriate gas cost for `ETH_HAS_CODE`, initially aligned with `ETHEREUM_CALL`.
pub const ETH_HAS_CODE: Gas = Gas::new(5_000_000_000);

pub struct RuntimeAdapter {
    pub eth_adapters: Arc<EthereumNetworkAdapters>,
    pub call_cache: Arc<dyn EthereumCallCache>,
    pub chain_identifier: Arc<ChainIdentifier>,
}

pub fn eth_call_gas(chain_identifier: &ChainIdentifier) -> Option<u32> {
    // 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<Chain> for RuntimeAdapter {
    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_adapters = eth_adapters.cheap_clone();
                            let call_cache = call_cache.cheap_clone();
                            let abis = abis.cheap_clone();
                            async move {
                                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,
                                )
                                .await
                                .map(|ptr| ptr.wasm_ptr())
                            }
                            .boxed()
                        }
                    }),
                },
                HostFn {
                    name: "ethereum.getBalance",
                    func: Arc::new({
                        let eth_adapters = eth_adapters.clone();
                        move |ctx, wasm_ptr| {
                            let eth_adapters = eth_adapters.cheap_clone();
                            async move {
                                let eth_adapter =
                                    eth_adapters.unverified_cheapest_with(&NodeCapabilities {
                                        archive,
                                        traces: false,
                                    })?;
                                eth_get_balance(&eth_adapter, ctx, wasm_ptr)
                                    .await
                                    .map(|ptr| ptr.wasm_ptr())
                            }
                            .boxed()
                        }
                    }),
                },
                HostFn {
                    name: "ethereum.hasCode",
                    func: Arc::new({
                        move |ctx, wasm_ptr| {
                            let eth_adapters = eth_adapters.cheap_clone();
                            async move {
                                let eth_adapter =
                                    eth_adapters.unverified_cheapest_with(&NodeCapabilities {
                                        archive,
                                        traces: false,
                                    })?;
                                eth_has_code(&eth_adapter, ctx, wasm_ptr)
                                    .await
                                    .map(|ptr| ptr.wasm_ptr())
                            }
                            .boxed()
                        }
                    }),
                },
            ]
        }

        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![],
            data_source::DataSource::Amp(_) => vec![],
        };

        Ok(host_fns)
    }

    fn raw_eth_call(&self) -> Option<Arc<dyn RawEthCall>> {
        Some(Arc::new(EthereumRawEthCall {
            eth_adapters: self.eth_adapters.cheap_clone(),
            call_cache: self.call_cache.cheap_clone(),
            eth_call_gas: eth_call_gas(&self.chain_identifier),
        }))
    }
}

/// Implementation of RawEthCall for Ethereum chains.
/// Used by Rust ABI subgraphs for making raw eth_call without ABI encoding.
pub struct EthereumRawEthCall {
    eth_adapters: Arc<EthereumNetworkAdapters>,
    call_cache: Arc<dyn EthereumCallCache>,
    eth_call_gas: Option<u32>,
}

#[async_trait]
impl RawEthCall for EthereumRawEthCall {
    async fn call(
        &self,
        address: [u8; 20],
        calldata: &[u8],
        block_ptr: &BlockPtr,
        gas: Option<u32>,
    ) -> Result<Option<Vec<u8>>, HostExportError> {
        // Get an adapter suitable for calls (non-archive is fine)
        let eth_adapter = self
            .eth_adapters
            .call_or_cheapest(Some(&NodeCapabilities {
                archive: false,
                traces: false,
            }))
            .map_err(HostExportError::Unknown)?;

        // Create a raw call request
        let req = call::Request::new(Address::from(address), calldata.to_vec(), 0);

        // Check cache first
        let (cached, _missing) = self
            .call_cache
            .get_calls(&[req.cheap_clone()], block_ptr.cheap_clone())
            .await
            .unwrap_or_else(|_| (Vec::new(), vec![req.cheap_clone()]));

        if let Some(resp) = cached.into_iter().next() {
            return match resp.retval {
                call::Retval::Value(bytes) => Ok(Some(bytes.to_vec())),
                call::Retval::Null => Ok(None),
            };
        }

        // Make the actual call
        let result = eth_adapter
            .raw_call(
                req.cheap_clone(),
                block_ptr.cheap_clone(),
                gas.or(self.eth_call_gas),
            )
            .await;

        match result {
            Ok(retval) => {
                // Cache the result
                let cache = self.call_cache.cheap_clone();
                let _ = cache
                    .set_call(
                        &Logger::root(Discard, o!()),
                        req,
                        block_ptr.cheap_clone(),
                        retval.clone(),
                    )
                    .await;

                match retval {
                    call::Retval::Value(bytes) => Ok(Some(bytes.to_vec())),
                    call::Retval::Null => Ok(None),
                }
            }
            Err(ContractCallError::AlloyError(e)) => Err(HostExportError::PossibleReorg(
                anyhow::anyhow!("eth_call RPC error: {}", e),
            )),
            Err(ContractCallError::Timeout) => Err(HostExportError::PossibleReorg(
                anyhow::anyhow!("eth_call timed out"),
            )),
            Err(e) => Err(HostExportError::Unknown(anyhow::anyhow!(
                "eth_call failed: {}",
                e
            ))),
        }
    }
}

/// function ethereum.call(call: SmartContractCall): Array<Token> | null
async fn ethereum_call(
    eth_adapter: &EthereumAdapter,
    call_cache: Arc<dyn EthereumCallCache>,
    ctx: HostFnCtx<'_>,
    wasm_ptr: u32,
    abis: &[Arc<MappingABI>],
    eth_call_gas: Option<u32>,
) -> Result<AscEnumArray<EthereumValueKind>, HostExportError> {
    ctx.gas
        .consume_host_fn_with_metrics(ETHEREUM_CALL, "ethereum_call")?;

    // For apiVersion >= 0.0.4 the call passed from the mapping includes the
    // function signature; subgraphs using an apiVersion < 0.0.4 don't pass
    // the signature along with the call.
    let call: UnresolvedContractCall = if ctx.heap.api_version() >= &API_VERSION_0_0_4 {
        asc_get::<_, AscUnresolvedContractCall_0_0_4, _>(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)?
    } else {
        asc_get::<_, AscUnresolvedContractCall, _>(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)?
    };

    let result = eth_call(
        eth_adapter,
        call_cache,
        &ctx.logger,
        &ctx.block_ptr,
        call,
        abis,
        eth_call_gas,
        ctx.metrics.cheap_clone(),
    )
    .await?;
    match result {
        Some(tokens) => Ok(asc_new(ctx.heap, tokens.as_slice(), &ctx.gas).await?),
        None => Ok(AscPtr::null()),
    }
}

async fn eth_get_balance(
    eth_adapter: &EthereumAdapter,
    ctx: HostFnCtx<'_>,
    wasm_ptr: u32,
) -> Result<AscPtr<AscBigInt>, HostExportError> {
    ctx.gas
        .consume_host_fn_with_metrics(ETH_GET_BALANCE, "eth_get_balance")?;

    if ctx.heap.api_version() < &API_VERSION_0_0_9 {
        return Err(HostExportError::Deterministic(anyhow!(
            "ethereum.getBalance call is not supported before API version 0.0.9"
        )));
    }

    let logger = &ctx.logger;
    let block_ptr = &ctx.block_ptr;

    let address: Address = asc_get(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)?;

    let result = eth_adapter
        .get_balance(logger, address, block_ptr.clone())
        .await;

    match result {
        Ok(v) => {
            let bigint = BigInt::from_unsigned_u256(&v);
            Ok(asc_new(ctx.heap, &bigint, &ctx.gas).await?)
        }
        // Retry on any kind of error
        Err(EthereumRpcError::AlloyError(e)) => Err(HostExportError::PossibleReorg(e.into())),
        Err(EthereumRpcError::Timeout) => Err(HostExportError::PossibleReorg(
            EthereumRpcError::Timeout.into(),
        )),
    }
}

async fn eth_has_code(
    eth_adapter: &EthereumAdapter,
    ctx: HostFnCtx<'_>,
    wasm_ptr: u32,
) -> Result<AscPtr<AscWrapped<bool>>, HostExportError> {
    ctx.gas
        .consume_host_fn_with_metrics(ETH_HAS_CODE, "eth_has_code")?;

    if ctx.heap.api_version() < &API_VERSION_0_0_9 {
        return Err(HostExportError::Deterministic(anyhow!(
            "ethereum.hasCode call is not supported before API version 0.0.9"
        )));
    }

    let logger = &ctx.logger;
    let block_ptr = &ctx.block_ptr;

    let address: Address = asc_get(ctx.heap, wasm_ptr.into(), &ctx.gas, 0)?;

    let result = eth_adapter
        .get_code(logger, address, block_ptr.clone())
        .await
        .map(|v| !v.0.is_empty());

    match result {
        Ok(v) => Ok(asc_new(ctx.heap, &AscWrapped { inner: v }, &ctx.gas).await?),
        // Retry on any kind of error
        Err(EthereumRpcError::AlloyError(e)) => Err(HostExportError::PossibleReorg(e.into())),
        Err(EthereumRpcError::Timeout) => Err(HostExportError::PossibleReorg(
            EthereumRpcError::Timeout.into(),
        )),
    }
}

/// Returns `Ok(None)` if the call was reverted.
async fn eth_call(
    eth_adapter: &EthereumAdapter,
    call_cache: Arc<dyn EthereumCallCache>,
    logger: &Logger,
    block_ptr: &BlockPtr,
    unresolved_call: UnresolvedContractCall,
    abis: &[Arc<MappingABI>],
    eth_call_gas: Option<u32>,
    metrics: Arc<HostMetrics>,
) -> Result<Option<Vec<abi::DynSolValue>>, HostExportError> {
    let start_time = Instant::now();

    // Obtain the path to the contract ABI
    let abi = abis
        .iter()
        .find(|abi| abi.name == unresolved_call.contract_name)
        .with_context(|| {
            format!(
                "Could not find ABI for contract \"{}\", try adding it to the 'abis' section \
                     of the subgraph manifest",
                unresolved_call.contract_name
            )
        })
        .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(),
        args: unresolved_call.function_args.clone(),
        gas: eth_call_gas,
    };

    // Run Ethereum call in tokio runtime
    let logger1 = logger.clone();
    let call_cache = call_cache.clone();
    let (result, source) = match eth_adapter.contract_call(&logger1, &call, call_cache).await {
        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(ContractCallError::AlloyError(e)) => Err(HostExportError::PossibleReorg(anyhow::anyhow!(
                "Ethereum node returned an error when calling function \"{}\" of contract \"{}\": {}",
                unresolved_call.function_name,
                unresolved_call.contract_name,
                e
            ))),

            // Also retry on timeouts.
            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,
            ))),

            Err(e) => Err(HostExportError::Unknown(anyhow::anyhow!(
                "Failed to call function \"{}\" of contract \"{}\": {}",
                unresolved_call.function_name,
                unresolved_call.contract_name,
                e
            ))),
        };

    let elapsed = start_time.elapsed();

    if source.observe() {
        metrics.observe_eth_call_execution_time(
            elapsed.as_secs_f64(),
            &unresolved_call.contract_name,
            &unresolved_call.function_name,
        );
    }

    let args_as_string = format!("[{}]", values_to_string(&unresolved_call.function_args));

    let result_as_string = match &result {
        Ok(Some(values)) => format!("({})", values_to_string(values)),
        Ok(None) => "none".to_owned(),
        Err(_err) => "error".to_owned(),
    };

    debug!(
        logger, "Contract call finished";
        "address" => format!("0x{:x}", &unresolved_call.contract_address),
        "contract" => &unresolved_call.contract_name,
        "signature" => &unresolved_call.function_signature,
        "args" => args_as_string,
        "time_ms" => format!("{}ms", elapsed.as_millis()),
        "result" => result_as_string,
        "block_hash" => block_ptr.hash_hex(),
        "block_number" => block_ptr.block_number(),
        "source" => source.to_string(),
    );

    result
}

#[derive(Clone, Debug)]
pub struct UnresolvedContractCall {
    pub contract_name: String,
    pub contract_address: Address,
    pub function_name: String,
    pub function_signature: Option<String>,
    pub function_args: Vec<abi::DynSolValue>,
}

impl AscIndexId for AscUnresolvedContractCall {
    const INDEX_ASC_TYPE_ID: IndexForAscTypeId = IndexForAscTypeId::SmartContractCall;
}

#[inline]
fn values_to_string(values: &[abi::DynSolValue]) -> String {
    values
        .iter()
        .map(|x| x.to_string())
        .collect_vec()
        .join(", ")
}