Merge pull request #996 from lambdaclass/gas-oracle

feat: Custom Base Token Gas Price Oracle

core/bin/zksync_server/src/main.rs

@@ -5,6 +5,7 @@ use clap::Parser;
 use zksync_config::{
     configs::{
         api::{HealthCheckConfig, MerkleTreeApiConfig, Web3JsonRpcConfig},
+        base_token_fetcher::BaseTokenFetcherConfig,
         chain::{
             CircuitBreakerConfig, MempoolConfig, NetworkConfig, OperationsManagerConfig,
             StateKeeperConfig,
@@ -273,5 +274,6 @@ fn load_env_config() -> anyhow::Result<TempConfigStore> {
         object_store_config: ObjectStoreConfig::from_env().ok(),
         observability: ObservabilityConfig::from_env().ok(),
         snapshot_creator: SnapshotsCreatorConfig::from_env().ok(),
+        base_token_fetcher: BaseTokenFetcherConfig::from_env().ok(),
     })
 }

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

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

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

@@ -1,3 +1,4 @@
+use super::BaseTokenFetcherConfig;
 use crate::{
     configs::{
         chain::{CircuitBreakerConfig, MempoolConfig, OperationsManagerConfig, StateKeeperConfig},
@@ -32,4 +33,5 @@ pub struct GeneralConfig {
     pub eth: Option<ETHConfig>,
     pub snapshot_creator: Option<SnapshotsCreatorConfig>,
     pub observability: Option<ObservabilityConfig>,
+    pub base_token_fetcher: Option<BaseTokenFetcherConfig>,
 }

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

@@ -1,6 +1,7 @@
 // Public re-exports
 pub use self::{
     api::ApiConfig,
+    base_token_fetcher::BaseTokenFetcherConfig,
     contract_verifier::ContractVerifierConfig,
     contracts::ContractsConfig,
     database::{DBConfig, PostgresConfig},
@@ -22,6 +23,7 @@ pub use self::{
 };
 
 pub mod api;
+pub mod base_token_fetcher;
 pub mod chain;
 pub mod contract_verifier;
 pub mod contracts;

core/lib/dal/src/blocks_dal.rs

@@ -2309,7 +2309,7 @@ impl BlocksDal<'_, '_> {
                 )
             WHERE
                 l1_batch_number IS NULL
-                AND fee_account_address = '\x0000000000000000000000000000000000000000'::bytea
+                AND fee_account_address = 'x0000000000000000000000000000000000000000'::bytea
             "#
         )
         .execute(self.storage.conn())
@@ -2349,7 +2349,7 @@ impl BlocksDal<'_, '_> {
             WHERE
                 l1_batches.number = miniblocks.l1_batch_number
                 AND miniblocks.number BETWEEN $1 AND $2
-                AND miniblocks.fee_account_address = '\x0000000000000000000000000000000000000000'::bytea
+                AND miniblocks.fee_account_address = 'x0000000000000000000000000000000000000000'::bytea
             "#,
             i64::from(numbers.start().0),
             i64::from(numbers.end().0)

core/lib/env_config/src/base_token_fetcher.rs

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

core/lib/env_config/src/lib.rs

@@ -2,6 +2,7 @@ use anyhow::Context as _;
 use serde::de::DeserializeOwned;
 
 mod api;
+pub mod base_token_fetcher;
 mod chain;
 mod contract_verifier;
 mod contracts;

core/lib/protobuf_config/src/base_token_fetcher.rs

@@ -0,0 +1,26 @@
+use anyhow::{Context as _, Ok};
+use zksync_config::configs;
+use zksync_protobuf::{repr::ProtoRepr, required};
+
+use crate::{parse_h160, proto::base_token_fetcher as proto};
+
+impl ProtoRepr for proto::BaseTokenFetcher {
+    type Type = configs::BaseTokenFetcherConfig;
+    fn read(&self) -> anyhow::Result<Self::Type> {
+        Ok(Self::Type {
+            poll_interval: *required(&self.poll_interval).context("poll_interval")?,
+            host: required(&self.host).context("host")?.clone(),
+            token_address: required(&self.token_address)
+                .and_then(|a| parse_h160(a))
+                .context("fee_account_addr")?,
+        })
+    }
+
+    fn build(this: &Self::Type) -> Self {
+        Self {
+            poll_interval: Some(this.poll_interval),
+            host: Some(this.host.clone()),
+            token_address: Some(this.token_address.to_string()),
+        }
+    }
+}

core/lib/protobuf_config/src/general.rs

@@ -37,6 +37,8 @@ impl ProtoRepr for proto::GeneralConfig {
             snapshot_creator: read_optional_repr(&self.snapshot_creator)
                 .context("snapshot_creator")?,
             observability: read_optional_repr(&self.observability).context("observability")?,
+            base_token_fetcher: read_optional_repr(&self.base_token_fetcher)
+                .context("base_token_fetcher")?,
         })
     }
 
@@ -68,6 +70,7 @@ impl ProtoRepr for proto::GeneralConfig {
             eth: this.eth.as_ref().map(ProtoRepr::build),
             snapshot_creator: this.snapshot_creator.as_ref().map(ProtoRepr::build),
             observability: this.observability.as_ref().map(ProtoRepr::build),
+            base_token_fetcher: this.base_token_fetcher.as_ref().map(ProtoRepr::build),
         }
     }
 }

core/lib/protobuf_config/src/lib.rs

@@ -5,6 +5,7 @@
 //! * protobuf json format
 
 mod api;
+mod base_token_fetcher;
 mod chain;
 mod circuit_breaker;
 mod contract_verifier;

core/lib/protobuf_config/src/proto/base_token_fetcher.proto

@@ -0,0 +1,9 @@
+syntax = "proto3";
+
+package zksync.config.base_token_fetcher;
+
+message BaseTokenFetcher {
+    optional uint64 poll_interval = 1;
+    optional string host = 2;
+    optional string token_address = 3;
+}

core/lib/protobuf_config/src/proto/general.proto

@@ -13,6 +13,8 @@ import "zksync/config/house_keeper.proto";
 import "zksync/config/observability.proto";
 import "zksync/config/snapshots_creator.proto";
 import "zksync/config/utils.proto";
+import "zksync/config/utils.proto";
+import "zksync/config/base_token_fetcher.proto";
 
 message GeneralConfig {
   optional config.database.Postgres postgres = 1;
@@ -35,6 +37,5 @@ message GeneralConfig {
   optional config.prover.ProverGateway prover_gateway = 30;
   optional config.snapshot_creator.SnapshotsCreator snapshot_creator = 31;
   optional config.observability.Observability observability = 32;
-
+  optional config.base_token_fetcher.BaseTokenFetcher base_token_fetcher = 33;
 }
-

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

@@ -0,0 +1,145 @@
+use std::{cmp::min, sync::Arc};
+
+use anyhow::Context;
+use async_trait::async_trait;
+use hex::ToHex;
+use metrics::atomics::AtomicU64;
+use tokio::sync::Mutex;
+use zksync_config::configs::base_token_fetcher::BaseTokenFetcherConfig;
+
+/// Trait used to query the stack's native token conversion rate. Used to properly
+/// determine gas prices, as they partially depend on L1 gas prices, denominated in `eth`.
+#[async_trait]
+pub trait ConversionRateFetcher: 'static + std::fmt::Debug + Send + Sync {
+    fn conversion_rate(&self) -> anyhow::Result<u64>;
+    async fn update(&self) -> anyhow::Result<()>;
+}
+
+#[derive(Debug)]
+pub(crate) struct NoOpConversionRateFetcher;
+
+impl NoOpConversionRateFetcher {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+#[async_trait]
+impl ConversionRateFetcher for NoOpConversionRateFetcher {
+    fn conversion_rate(&self) -> anyhow::Result<u64> {
+        Ok(1)
+    }
+
+    async fn update(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
+}
+
+/// Struct in charge of periodically querying and caching the native token's conversion rate
+/// to `eth`.
+#[derive(Debug)]
+pub(crate) struct BaseTokenFetcher {
+    pub config: BaseTokenFetcherConfig,
+    pub latest_to_eth_conversion_rate: AtomicU64,
+    http_client: reqwest::Client,
+    error_reporter: Arc<Mutex<ErrorReporter>>,
+}
+
+impl BaseTokenFetcher {
+    pub(crate) async fn new(config: BaseTokenFetcherConfig) -> anyhow::Result<Self> {
+        let http_client = reqwest::Client::new();
+
+        let conversion_rate = http_client
+            .get(format!(
+                "{}/conversion_rate/0x{}",
+                config.host,
+                config.token_address.encode_hex::<String>()
+            ))
+            .send()
+            .await?
+            .json::<u64>()
+            .await
+            .context("Unable to parse the response of the native token conversion rate server")?;
+
+        let error_reporter = Arc::new(Mutex::new(ErrorReporter::new()));
+
+        Ok(Self {
+            config,
+            latest_to_eth_conversion_rate: AtomicU64::new(conversion_rate),
+            http_client,
+            error_reporter,
+        })
+    }
+}
+
+#[async_trait]
+impl ConversionRateFetcher for BaseTokenFetcher {
+    fn conversion_rate(&self) -> anyhow::Result<u64> {
+        anyhow::Ok(
+            self.latest_to_eth_conversion_rate
+                .load(std::sync::atomic::Ordering::Relaxed),
+        )
+    }
+
+    async fn update(&self) -> anyhow::Result<()> {
+        match self
+            .http_client
+            .get(format!(
+                "{}/conversion_rate/0x{}",
+                &self.config.host,
+                &self.config.token_address.encode_hex::<String>()
+            ))
+            .send()
+            .await
+        {
+            Ok(response) => {
+                let conversion_rate = response.json::<u64>().await.context(
+                    "Unable to parse the response of the native token conversion rate server",
+                )?;
+                self.latest_to_eth_conversion_rate
+                    .store(conversion_rate, std::sync::atomic::Ordering::Relaxed);
+                self.error_reporter.lock().await.reset();
+            }
+            Err(err) => self
+                .error_reporter
+                .lock()
+                .await
+                .process(anyhow::anyhow!(err)),
+        }
+
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+struct ErrorReporter {
+    current_try: u8,
+    alert_spawned: bool,
+}
+
+impl ErrorReporter {
+    const MAX_CONSECUTIVE_NETWORK_ERRORS: u8 = 10;
+
+    fn new() -> Self {
+        Self {
+            current_try: 0,
+            alert_spawned: false,
+        }
+    }
+
+    fn reset(&mut self) {
+        self.current_try = 0;
+        self.alert_spawned = false;
+    }
+
+    fn process(&mut self, err: anyhow::Error) {
+        self.current_try = min(self.current_try + 1, Self::MAX_CONSECUTIVE_NETWORK_ERRORS);
+
+        tracing::error!("Failed to fetch native token conversion rate from the server: {err}");
+
+        if self.current_try >= Self::MAX_CONSECUTIVE_NETWORK_ERRORS && !self.alert_spawned {
+            vlog::capture_message(&err.to_string(), vlog::AlertLevel::Warning);
+            self.alert_spawned = true;
+        }
+    }
+}

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

@@ -0,0 +1,35 @@
+use axum::{extract, extract::Json, routing::get, Router};
+use tokio::sync::watch;
+use zksync_config::configs::base_token_fetcher::BaseTokenFetcherConfig;
+
+pub(crate) async fn run_server(
+    mut stop_receiver: watch::Receiver<bool>,
+    server_configs: &BaseTokenFetcherConfig,
+) -> anyhow::Result<()> {
+    let app = Router::new().route("/conversion_rate/:token_address", get(get_conversion_rate));
+
+    let bind_address = if server_configs.host.starts_with("http://") {
+        &server_configs.host[7..] // If it starts with "http://", strip the prefix
+    } else {
+        &server_configs.host // Otherwise, return the original string
+    };
+
+    axum::Server::bind(&bind_address.parse().expect("Unable to parse socket address"))
+        .serve(app.into_make_service())
+        .with_graceful_shutdown(async move {
+            if stop_receiver.changed().await.is_err() {
+                tracing::warn!("Stop signal sender for conversion rate API server was dropped without sending a signal");
+            }
+            tracing::info!("Stop signal received, conversion rate server is shutting down");
+        })
+        .await
+        .expect("Conversion rate server failed");
+    tracing::info!("Conversion rate server shut down");
+    Ok(())
+}
+
+// basic handler that responds with a static string
+async fn get_conversion_rate(extract::Path(_token_address): extract::Path<String>) -> Json<u64> {
+    tracing::info!("Received request for conversion rate");
+    Json(42)
+}