feat: Custom Base Token Gas Price Oracle

core/bin/zksync_server/src/main.rs

@@ -11,6 +11,7 @@ use zksync_config::{
         },
         fri_prover_group::FriProverGroupConfig,
         house_keeper::HouseKeeperConfig,
+        native_erc20_fetcher::NativeErc20FetcherConfig,
         FriProofCompressorConfig, FriProverConfig, FriWitnessGeneratorConfig, PrometheusConfig,
         ProofDataHandlerConfig, WitnessGeneratorConfig,
     },
@@ -119,6 +120,7 @@ async fn main() -> anyhow::Result<()> {
         eth_watch_config: ETHWatchConfig::from_env().ok(),
         gas_adjuster_config: GasAdjusterConfig::from_env().ok(),
         object_store_config: ObjectStoreConfig::from_env().ok(),
+        native_erc20_fetcher_config: NativeErc20FetcherConfig::from_env().ok(),
     };
 
     let postgres_config = configs.postgres_config.clone().context("PostgresConfig")?;

core/lib/config/src/configs/mod.rs

@@ -36,6 +36,7 @@ pub mod fri_prover_group;
 pub mod fri_witness_generator;
 pub mod fri_witness_vector_generator;
 pub mod house_keeper;
+pub mod native_erc20_fetcher;
 pub mod object_store;
 pub mod proof_data_handler;
 pub mod snapshots_creator;

core/lib/config/src/configs/native_erc20_fetcher.rs

@@ -0,0 +1,7 @@
+use serde::Deserialize;
+
+#[derive(Debug, Clone, Deserialize, PartialEq)]
+pub struct NativeErc20FetcherConfig {
+    pub poll_interval: u64,
+    pub host: String,
+}

core/lib/env_config/src/lib.rs

@@ -17,6 +17,7 @@ mod fri_prover_group;
 mod fri_witness_generator;
 mod fri_witness_vector_generator;
 mod house_keeper;
+pub mod native_erc20_fetcher;
 pub mod object_store;
 mod proof_data_handler;
 mod snapshots_creator;

core/lib/env_config/src/native_erc20_fetcher.rs

@@ -0,0 +1,9 @@
+use zksync_config::configs::native_erc20_fetcher::NativeErc20FetcherConfig;
+
+use crate::{envy_load, FromEnv};
+
+impl FromEnv for NativeErc20FetcherConfig {
+    fn from_env() -> anyhow::Result<Self> {
+        envy_load("native_erc20_fetcher", "NATIVE_ERC20_FETCHER_")
+    }
+}

core/lib/zksync_core/src/consensus/testonly.rs

@@ -375,6 +375,7 @@ impl StateKeeperRunner {
                     Box::new(io),
                     Box::new(MockBatchExecutorBuilder),
                     Box::new(NoopSealer),
+                    Box::new(None),
                 )
                 .run(),
             );

core/lib/zksync_core/src/l1_gas_price/gas_adjuster/mod.rs

@@ -12,7 +12,7 @@ use zksync_system_constants::L1_GAS_PER_PUBDATA_BYTE;
 
 use self::metrics::METRICS;
 use super::{L1GasPriceProvider, L1TxParamsProvider};
-use crate::state_keeper::metrics::KEEPER_METRICS;
+use crate::{native_erc20_fetcher::Erc20Fetcher, state_keeper::metrics::KEEPER_METRICS};
 
 mod metrics;
 #[cfg(test)]
@@ -25,10 +25,15 @@ pub struct GasAdjuster<E> {
     pub(super) statistics: GasStatistics,
     pub(super) config: GasAdjusterConfig,
     eth_client: E,
+    erc20_fetcher_dyn: Option<Arc<dyn Erc20Fetcher>>,
 }
 
 impl<E: EthInterface> GasAdjuster<E> {
-    pub async fn new(eth_client: E, config: GasAdjusterConfig) -> Result<Self, Error> {
+    pub async fn new(
+        eth_client: E,
+        config: GasAdjusterConfig,
+        erc20_fetcher_dyn: Option<Arc<dyn Erc20Fetcher>>,
+    ) -> Result<Self, Error> {
         // Subtracting 1 from the "latest" block number to prevent errors in case
         // the info about the latest block is not yet present on the node.
         // This sometimes happens on Infura.
@@ -44,6 +49,7 @@ impl<E: EthInterface> GasAdjuster<E> {
             statistics: GasStatistics::new(config.max_base_fee_samples, current_block, &history),
             eth_client,
             config,
+            erc20_fetcher_dyn,
         })
     }
 
@@ -125,7 +131,12 @@ impl<E: EthInterface> L1GasPriceProvider for GasAdjuster<E> {
             (self.config.internal_l1_pricing_multiplier * effective_gas_price as f64) as u64;
 
         // Bound the price if it's too high.
-        self.bound_gas_price(calculated_price)
+        let conversion_rate = match self.erc20_fetcher_dyn.as_ref() {
+            Some(fetcher) => fetcher.conversion_rate().unwrap_or(1),
+            None => 1,
+        };
+
+        self.bound_gas_price(calculated_price) * conversion_rate
     }
 
     fn estimate_effective_pubdata_price(&self) -> u64 {

core/lib/zksync_core/src/l1_gas_price/singleton.rs

@@ -8,7 +8,7 @@ use tokio::{
 use zksync_config::GasAdjusterConfig;
 use zksync_eth_client::clients::QueryClient;
 
-use crate::l1_gas_price::GasAdjuster;
+use crate::{l1_gas_price::GasAdjuster, native_erc20_fetcher::Erc20Fetcher};
 
 /// Special struct for creating a singleton of `GasAdjuster`.
 /// This is needed only for running the server.
@@ -17,6 +17,7 @@ pub struct GasAdjusterSingleton {
     web3_url: String,
     gas_adjuster_config: GasAdjusterConfig,
     singleton: OnceCell<Result<Arc<GasAdjuster<QueryClient>>, Error>>,
+    erc20_fetcher_dyn: Option<Arc<dyn Erc20Fetcher>>,
 }
 
 #[derive(thiserror::Error, Debug, Clone)]
@@ -30,11 +31,16 @@ impl From<anyhow::Error> for Error {
 }
 
 impl GasAdjusterSingleton {
-    pub fn new(web3_url: String, gas_adjuster_config: GasAdjusterConfig) -> Self {
+    pub fn new(
+        web3_url: String,
+        gas_adjuster_config: GasAdjusterConfig,
+        erc20_fetcher_dyn: Option<Arc<dyn Erc20Fetcher>>,
+    ) -> Self {
         Self {
             web3_url,
             gas_adjuster_config,
             singleton: OnceCell::new(),
+            erc20_fetcher_dyn,
         }
     }
 
@@ -44,9 +50,13 @@ impl GasAdjusterSingleton {
             .get_or_init(|| async {
                 let query_client =
                     QueryClient::new(&self.web3_url).context("QueryClient::new()")?;
-                let adjuster = GasAdjuster::new(query_client.clone(), self.gas_adjuster_config)
-                    .await
-                    .context("GasAdjuster::new()")?;
+                let adjuster = GasAdjuster::new(
+                    query_client.clone(),
+                    self.gas_adjuster_config,
+                    self.erc20_fetcher_dyn.clone(),
+                )
+                .await
+                .context("GasAdjuster::new()")?;
                 Ok(Arc::new(adjuster))
             })
             .await;

core/lib/zksync_core/src/lib.rs

@@ -5,6 +5,7 @@ use std::{net::Ipv4Addr, str::FromStr, sync::Arc, time::Instant};
 use anyhow::Context as _;
 use fee_model::MainNodeFeeInputProvider;
 use futures::channel::oneshot;
+use native_erc20_fetcher::Erc20Fetcher;
 use prometheus_exporter::PrometheusExporterConfig;
 use temp_config_store::TempConfigStore;
 use tokio::{sync::watch, task::JoinHandle};
@@ -69,6 +70,7 @@ use crate::{
     l1_gas_price::{GasAdjusterSingleton, L1GasPriceProvider},
     metadata_calculator::{MetadataCalculator, MetadataCalculatorConfig},
     metrics::{InitStage, APP_METRICS},
+    native_erc20_fetcher::NativeErc20FetcherSingleton,
     state_keeper::{
         create_state_keeper, MempoolFetcher, MempoolGuard, MiniblockSealer, SequencerSealer,
     },
@@ -88,6 +90,7 @@ pub mod house_keeper;
 pub mod l1_gas_price;
 pub mod metadata_calculator;
 mod metrics;
+pub mod native_erc20_fetcher;
 pub mod proof_data_handler;
 pub mod reorg_detector;
 pub mod state_keeper;
@@ -230,6 +233,8 @@ pub enum Component {
     Housekeeper,
     /// Component for exposing APIs to prover for providing proof generation data and accepting proofs.
     ProofDataHandler,
+    /// Native ERC20 fetcher
+    NativeERC20Fetcher,
 }
 
 #[derive(Debug)]
@@ -264,6 +269,7 @@ impl FromStr for Components {
             "eth_tx_aggregator" => Ok(Components(vec![Component::EthTxAggregator])),
             "eth_tx_manager" => Ok(Components(vec![Component::EthTxManager])),
             "proof_data_handler" => Ok(Components(vec![Component::ProofDataHandler])),
+            "native_erc20_fetcher" => Ok(Components(vec![Component::NativeERC20Fetcher])),
             other => Err(format!("{} is not a valid component name", other)),
         }
     }
@@ -320,14 +326,44 @@ pub async fn initialize_components(
         panic!("Circuit breaker triggered: {}", err);
     });
 
-    let query_client = QueryClient::new(&eth_client_config.web3_url).unwrap();
-    let gas_adjuster_config = configs.gas_adjuster_config.context("gas_adjuster_config")?;
-    let mut gas_adjuster =
-        GasAdjusterSingleton::new(eth_client_config.web3_url.clone(), gas_adjuster_config);
+    // spawn the native ERC20 fetcher if it is enabled
+    let mut fetcher_component = if components.contains(&Component::NativeERC20Fetcher) {
+        let fetcher = NativeErc20FetcherSingleton::new(
+            configs
+                .native_erc20_fetcher_config
+                .clone()
+                .context("native_erc20_fetcher_config")?,
+        );
 
+        Some(fetcher)
+    } else {
+        None
+    };
     let (stop_sender, stop_receiver) = watch::channel(false);
     let (cb_sender, cb_receiver) = oneshot::channel();
 
+    let native_erc20_fetcher = if let Some(ref mut fetcher_singleton) = fetcher_component {
+        let fetcher = fetcher_singleton
+            .get_or_init()
+            .await
+            .context("fetcher.get_or_init()")?;
+        Some(fetcher)
+    } else {
+        None
+    };
+
+    let erc20_fetcher_dyn: Option<Arc<dyn Erc20Fetcher>> = native_erc20_fetcher
+        .as_ref()
+        .map(|fetcher| fetcher.clone() as Arc<dyn Erc20Fetcher>);
+
+    let query_client = QueryClient::new(&eth_client_config.web3_url).unwrap();
+    let gas_adjuster_config = configs.gas_adjuster_config.context("gas_adjuster_config")?;
+    let mut gas_adjuster = GasAdjusterSingleton::new(
+        eth_client_config.web3_url.clone(),
+        gas_adjuster_config,
+        erc20_fetcher_dyn,
+    );
+
     // Prometheus exporter and circuit breaker checker should run for every component configuration.
     let prom_config = configs
         .prometheus_config
@@ -482,6 +518,7 @@ pub async fn initialize_components(
             .get_or_init()
             .await
             .context("gas_adjuster.get_or_init()")?;
+
         add_state_keeper_to_task_futures(
             &mut task_futures,
             &postgres_config,
@@ -671,6 +708,15 @@ pub async fn initialize_components(
     if let Some(task) = gas_adjuster.run_if_initialized(stop_receiver.clone()) {
         task_futures.push(task);
     }
+
+    // check if the native ERC20 fetcher is enabled and run it if it is
+    fetcher_component
+        .and_then(|c| c.run_if_initialized(stop_receiver.clone()))
+        .into_iter()
+        .for_each(|handle| {
+            task_futures.push(handle);
+        });
+
     Ok((task_futures, stop_sender, cb_receiver, health_check_handle))
 }
 

core/lib/zksync_core/src/native_erc20_fetcher/mod.rs

@@ -0,0 +1,107 @@
+use std::{sync::Arc, time::Duration};
+
+use async_trait::async_trait;
+use metrics::atomics::AtomicU64;
+use tokio::{
+    sync::{watch, OnceCell},
+    task::JoinHandle,
+};
+use zksync_config::configs::native_erc20_fetcher::NativeErc20FetcherConfig;
+
+// TODO: this error type is also defined by the gasAdjuster module,
+//       we should consider moving it to a common place
+#[derive(thiserror::Error, Debug, Clone)]
+#[error(transparent)]
+pub struct Error(Arc<anyhow::Error>);
+
+impl From<anyhow::Error> for Error {
+    fn from(err: anyhow::Error) -> Self {
+        Self(Arc::new(err))
+    }
+}
+
+pub trait Erc20Fetcher: 'static + std::fmt::Debug + Send + Sync {
+    fn conversion_rate(&self) -> anyhow::Result<u64>;
+}
+
+pub(crate) struct NativeErc20FetcherSingleton {
+    native_erc20_fetcher_config: NativeErc20FetcherConfig,
+    singleton: OnceCell<Result<Arc<NativeErc20Fetcher>, Error>>,
+}
+
+impl NativeErc20FetcherSingleton {
+    pub fn new(native_erc20_fetcher_config: NativeErc20FetcherConfig) -> Self {
+        Self {
+            native_erc20_fetcher_config,
+            singleton: OnceCell::new(),
+        }
+    }
+
+    pub async fn get_or_init(&mut self) -> Result<Arc<NativeErc20Fetcher>, Error> {
+        let adjuster = self
+            .singleton
+            .get_or_init(|| async {
+                let fetcher = NativeErc20Fetcher::new(self.native_erc20_fetcher_config.clone());
+                Ok(Arc::new(fetcher))
+            })
+            .await;
+        adjuster.clone()
+    }
+
+    pub fn run_if_initialized(
+        self,
+        stop_signal: watch::Receiver<bool>,
+    ) -> Option<JoinHandle<anyhow::Result<()>>> {
+        let fetcher = self.singleton.get()?.clone();
+        Some(tokio::spawn(async move { fetcher?.run(stop_signal).await }))
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct NativeErc20Fetcher {
+    // TODO: we probably need to add a http client here
+    // to avoid creating a new one for each request
+    pub config: NativeErc20FetcherConfig,
+    pub latest_to_eth_conversion_rate: AtomicU64,
+}
+
+impl NativeErc20Fetcher {
+    pub(crate) fn new(config: NativeErc20FetcherConfig) -> Self {
+        Self {
+            config,
+            latest_to_eth_conversion_rate: AtomicU64::new(0),
+        }
+    }
+
+    pub(crate) async fn run(&self, stop_receiver: watch::Receiver<bool>) -> anyhow::Result<()> {
+        loop {
+            if *stop_receiver.borrow() {
+                tracing::info!("Stop signal received, eth_tx_manager is shutting down");
+                break;
+            }
+
+            let conversion_rate = reqwest::get(format!("{}/conversion_rate", &self.config.host))
+                .await?
+                .json::<u64>()
+                .await
+                .unwrap();
+
+            self.latest_to_eth_conversion_rate
+                .store(conversion_rate, std::sync::atomic::Ordering::Relaxed);
+
+            tokio::time::sleep(Duration::from_secs(self.config.poll_interval)).await;
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl Erc20Fetcher for NativeErc20Fetcher {
+    fn conversion_rate(&self) -> anyhow::Result<u64> {
+        anyhow::Ok(
+            self.latest_to_eth_conversion_rate
+                .load(std::sync::atomic::Ordering::Relaxed),
+        )
+    }
+}

core/lib/zksync_core/src/state_keeper/tests/tester.rs

@@ -201,6 +201,7 @@ impl TestScenario {
             Box::new(io),
             Box::new(batch_executor_base),
             Box::new(sealer),
+            Box::new(None),
         );
         let sk_thread = tokio::spawn(sk.run());
 

core/lib/zksync_core/src/sync_layer/tests.rs

@@ -85,6 +85,7 @@ impl StateKeeperHandles {
             Box::new(io),
             Box::new(batch_executor_base),
             Box::new(NoopSealer),
+            Box::new(None),
         );
         Self {
             stop_sender,