feat(api): Consider State keeper fee model input in the API (#901)

## What ❔

Before, under big spikes in L1 gas price, it was possible that the state
keeper retained the inflated gas price, while the API got the smaller
one. In this PR we ensure that the API takes into account the fee params
of the latest sealed miniblock

## Why ❔

<!-- Why are these changes done? What goal do they contribute to? What
are the principles behind them? -->
<!-- Example: PR templates ensure PR reviewers, observers, and future
iterators are in context about the evolution of repos. -->

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zk fmt` and `zk lint`.
- [ ] Spellcheck has been run via `zk spellcheck`.

core/lib/types/src/fee_model.rs

@@ -30,6 +30,18 @@ impl BatchFeeInput {
             fair_l2_gas_price,
         })
     }
+
+    pub fn pubdata_independent(
+        l1_gas_price: u64,
+        fair_l2_gas_price: u64,
+        fair_pubdata_price: u64,
+    ) -> Self {
+        Self::PubdataIndependent(PubdataIndependentBatchFeeModelInput {
+            l1_gas_price,
+            fair_l2_gas_price,
+            fair_pubdata_price,
+        })
+    }
 }
 
 impl Default for BatchFeeInput {
@@ -101,6 +113,27 @@ impl BatchFeeInput {
             })
         }
     }
+
+    pub fn stricter(self, other: BatchFeeInput) -> Self {
+        match (self, other) {
+            (BatchFeeInput::L1Pegged(first), BatchFeeInput::L1Pegged(second)) => Self::l1_pegged(
+                first.l1_gas_price.max(second.l1_gas_price),
+                first.fair_l2_gas_price.max(second.fair_l2_gas_price),
+            ),
+            input @ (_, _) => {
+                let (first, second) = (
+                    input.0.into_pubdata_independent(),
+                    input.1.into_pubdata_independent(),
+                );
+
+                Self::pubdata_independent(
+                    first.l1_gas_price.max(second.l1_gas_price),
+                    first.fair_l2_gas_price.max(second.fair_l2_gas_price),
+                    first.fair_pubdata_price.max(second.fair_pubdata_price),
+                )
+            }
+        }
+    }
 }
 
 /// Pubdata is only published via calldata and so its price is pegged to the L1 gas price.

core/lib/zksync_core/src/api_server/tx_sender/mod.rs

@@ -284,7 +284,7 @@ impl TxSender {
         stage_latency.observe();
 
         let stage_latency = SANDBOX_METRICS.submit_tx[&SubmitTxStage::DryRun].start();
-        let shared_args = self.shared_args();
+        let shared_args = self.shared_args().await;
         let vm_permit = self.0.vm_concurrency_limiter.acquire().await;
         let vm_permit = vm_permit.ok_or(SubmitTxError::ServerShuttingDown)?;
         let mut connection = self
@@ -401,10 +401,10 @@ impl TxSender {
         }
     }
 
-    fn shared_args(&self) -> TxSharedArgs {
+    async fn shared_args(&self) -> TxSharedArgs {
         TxSharedArgs {
             operator_account: AccountTreeId::new(self.0.sender_config.fee_account_addr),
-            fee_input: self.0.batch_fee_input_provider.get_batch_fee_input(),
+            fee_input: self.0.batch_fee_input_provider.get_batch_fee_input().await,
             base_system_contracts: self.0.api_contracts.eth_call.clone(),
             caches: self.storage_caches(),
             validation_computational_gas_limit: self
@@ -423,7 +423,7 @@ impl TxSender {
             return Err(SubmitTxError::GasLimitIsTooBig);
         }
 
-        let fee_input = self.0.batch_fee_input_provider.get_batch_fee_input();
+        let fee_input = self.0.batch_fee_input_provider.get_batch_fee_input().await;
 
         // TODO (SMA-1715): do not subsidize the overhead for the transaction
 
@@ -684,10 +684,14 @@ impl TxSender {
 
         let fee_input = {
             // For now, both L1 gas price and pubdata price are scaled with the same coefficient
-            let fee_input = self.0.batch_fee_input_provider.get_batch_fee_input_scaled(
-                self.0.sender_config.gas_price_scale_factor,
-                self.0.sender_config.gas_price_scale_factor,
-            );
+            let fee_input = self
+                .0
+                .batch_fee_input_provider
+                .get_batch_fee_input_scaled(
+                    self.0.sender_config.gas_price_scale_factor,
+                    self.0.sender_config.gas_price_scale_factor,
+                )
+                .await;
             adjust_pubdata_price_for_tx(
                 fee_input,
                 tx.gas_per_pubdata_byte_limit(),
@@ -908,7 +912,7 @@ impl TxSender {
             .executor
             .execute_tx_eth_call(
                 vm_permit,
-                self.shared_args(),
+                self.shared_args().await,
                 self.0.replica_connection_pool.clone(),
                 tx,
                 block_args,
@@ -936,10 +940,13 @@ impl TxSender {
 
         let (base_fee, _) = derive_base_fee_and_gas_per_pubdata(
             // For now, both the L1 gas price and the L1 pubdata price are scaled with the same coefficient
-            self.0.batch_fee_input_provider.get_batch_fee_input_scaled(
-                self.0.sender_config.gas_price_scale_factor,
-                self.0.sender_config.gas_price_scale_factor,
-            ),
+            self.0
+                .batch_fee_input_provider
+                .get_batch_fee_input_scaled(
+                    self.0.sender_config.gas_price_scale_factor,
+                    self.0.sender_config.gas_price_scale_factor,
+                )
+                .await,
             protocol_version.into(),
         );
         base_fee

core/lib/zksync_core/src/api_server/web3/backend_jsonrpsee/namespaces/zks.rs

@@ -142,7 +142,7 @@ impl ZksNamespaceServer for ZksNamespace {
     }
 
     async fn get_l1_gas_price(&self) -> RpcResult<U64> {
-        Ok(self.get_l1_gas_price_impl())
+        Ok(self.get_l1_gas_price_impl().await)
     }
 
     async fn get_fee_params(&self) -> RpcResult<FeeParams> {

core/lib/zksync_core/src/api_server/web3/namespaces/debug.rs

@@ -38,7 +38,8 @@ impl DebugNamespace {
                 .get_batch_fee_input_scaled(
                     state.api_config.estimate_gas_scale_factor,
                     state.api_config.estimate_gas_scale_factor,
-                ),
+                )
+                .await,
             state,
             api_contracts,
         }

core/lib/zksync_core/src/api_server/web3/namespaces/zks.rs

@@ -542,7 +542,7 @@ impl ZksNamespace {
     }
 
     #[tracing::instrument(skip(self))]
-    pub fn get_l1_gas_price_impl(&self) -> U64 {
+    pub async fn get_l1_gas_price_impl(&self) -> U64 {
         const METHOD_NAME: &str = "get_l1_gas_price";
 
         let method_latency = API_METRICS.start_call(METHOD_NAME);
@@ -552,6 +552,7 @@ impl ZksNamespace {
             .0
             .batch_fee_input_provider
             .get_batch_fee_input()
+            .await
             .l1_gas_price();
 
         method_latency.observe();

core/lib/zksync_core/src/fee_model.rs

@@ -1,5 +1,6 @@
 use std::{fmt, sync::Arc};
 
+use zksync_dal::ConnectionPool;
 use zksync_types::{
     fee_model::{
         BatchFeeInput, FeeModelConfig, FeeModelConfigV2, FeeParams, FeeParamsV1, FeeParamsV2,
@@ -12,10 +13,11 @@ use zksync_utils::ceil_div_u256;
 use crate::l1_gas_price::L1GasPriceProvider;
 
 /// Trait responsible for providing fee info for a batch
+#[async_trait::async_trait]
 pub trait BatchFeeModelInputProvider: fmt::Debug + 'static + Send + Sync {
     /// Returns the batch fee with scaling applied. This may be used to account for the fact that the L1 gas and pubdata prices may fluctuate, esp.
     /// in API methods that should return values that are valid for some period of time after the estimation was done.
-    fn get_batch_fee_input_scaled(
+    async fn get_batch_fee_input_scaled(
         &self,
         l1_gas_price_scale_factor: f64,
         l1_pubdata_price_scale_factor: f64,
@@ -38,8 +40,8 @@ pub trait BatchFeeModelInputProvider: fmt::Debug + 'static + Send + Sync {
     }
 
     /// Returns the batch fee input as-is, i.e. without any scaling for the L1 gas and pubdata prices.
-    fn get_batch_fee_input(&self) -> BatchFeeInput {
-        self.get_batch_fee_input_scaled(1.0, 1.0)
+    async fn get_batch_fee_input(&self) -> BatchFeeInput {
+        self.get_batch_fee_input_scaled(1.0, 1.0).await
     }
 
     /// Returns the fee model parameters.
@@ -77,6 +79,59 @@ impl MainNodeFeeInputProvider {
     }
 }
 
+/// The fee model provider to be used in the API. It returns the maximal batch fee input between the projected main node one and
+/// the one from the last sealed miniblock.
+#[derive(Debug)]
+pub(crate) struct ApiFeeInputProvider {
+    inner: MainNodeFeeInputProvider,
+    connection_pool: ConnectionPool,
+}
+
+impl ApiFeeInputProvider {
+    pub fn new(
+        provider: Arc<dyn L1GasPriceProvider>,
+        config: FeeModelConfig,
+        connection_pool: ConnectionPool,
+    ) -> Self {
+        Self {
+            inner: MainNodeFeeInputProvider::new(provider, config),
+            connection_pool,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl BatchFeeModelInputProvider for ApiFeeInputProvider {
+    async fn get_batch_fee_input_scaled(
+        &self,
+        l1_gas_price_scale_factor: f64,
+        l1_pubdata_price_scale_factor: f64,
+    ) -> BatchFeeInput {
+        let inner_input = self
+            .inner
+            .get_batch_fee_input_scaled(l1_gas_price_scale_factor, l1_pubdata_price_scale_factor)
+            .await;
+        let last_miniblock_params = self
+            .connection_pool
+            .access_storage_tagged("api_fee_input_provider")
+            .await
+            .unwrap()
+            .blocks_dal()
+            .get_last_sealed_miniblock_header()
+            .await
+            .unwrap();
+
+        last_miniblock_params
+            .map(|header| inner_input.stricter(header.batch_fee_input))
+            .unwrap_or(inner_input)
+    }
+
+    /// Returns the fee model parameters.
+    fn get_fee_model_params(&self) -> FeeParams {
+        self.inner.get_fee_model_params()
+    }
+}
+
 /// Calculates the batch fee input based on the main node parameters.
 /// This function uses the `V1` fee model, i.e. where the pubdata price does not include the proving costs.
 fn compute_batch_fee_model_input_v1(

core/lib/zksync_core/src/lib.rs

@@ -3,7 +3,7 @@
 use std::{net::Ipv4Addr, str::FromStr, sync::Arc, time::Instant};
 
 use anyhow::Context as _;
-use fee_model::MainNodeFeeInputProvider;
+use fee_model::{ApiFeeInputProvider, MainNodeFeeInputProvider};
 use futures::channel::oneshot;
 use prometheus_exporter::PrometheusExporterConfig;
 use temp_config_store::TempConfigStore;
@@ -1006,16 +1006,17 @@ async fn build_tx_sender(
     storage_caches: PostgresStorageCaches,
 ) -> (TxSender, VmConcurrencyBarrier) {
     let sequencer_sealer = SequencerSealer::new(state_keeper_config.clone());
-    let tx_sender_builder = TxSenderBuilder::new(tx_sender_config.clone(), replica_pool)
+    let tx_sender_builder = TxSenderBuilder::new(tx_sender_config.clone(), replica_pool.clone())
         .with_main_connection_pool(master_pool)
         .with_sealer(Arc::new(sequencer_sealer));
 
     let max_concurrency = web3_json_config.vm_concurrency_limit();
     let (vm_concurrency_limiter, vm_barrier) = VmConcurrencyLimiter::new(max_concurrency);
 
-    let batch_fee_input_provider = MainNodeFeeInputProvider::new(
+    let batch_fee_input_provider = ApiFeeInputProvider::new(
         l1_gas_price_provider,
         FeeModelConfig::from_state_keeper_config(state_keeper_config),
+        replica_pool,
     );
 
     let tx_sender = tx_sender_builder

core/lib/zksync_core/src/state_keeper/io/mempool.rs

@@ -163,7 +163,8 @@ impl StateKeeperIO for MempoolIO {
             self.filter = l2_tx_filter(
                 self.batch_fee_input_provider.as_ref(),
                 protocol_version.into(),
-            );
+            )
+            .await;
             // We only need to get the root hash when we're certain that we have a new transaction.
             if !self.mempool.has_next(&self.filter) {
                 tokio::time::sleep(self.delay_interval).await;

core/lib/zksync_core/src/state_keeper/io/tests/mod.rs

@@ -106,7 +106,8 @@ async fn test_filter_with_no_pending_batch() {
     let want_filter = l2_tx_filter(
         &tester.create_batch_fee_input_provider().await,
         ProtocolVersionId::latest().into(),
-    );
+    )
+    .await;
 
     // Create a mempool without pending batch and ensure that filter is not initialized just yet.
     let (mut mempool, mut guard) = tester.create_test_mempool_io(connection_pool, 1).await;
@@ -150,7 +151,8 @@ async fn test_timestamps_are_distinct(
     let tx_filter = l2_tx_filter(
         &tester.create_batch_fee_input_provider().await,
         ProtocolVersionId::latest().into(),
-    );
+    )
+    .await;
     tester.insert_tx(&mut guard, tx_filter.fee_per_gas, tx_filter.gas_per_pubdata);
 
     let batch_params = mempool

core/lib/zksync_core/src/state_keeper/mempool_actor.rs

@@ -13,11 +13,11 @@ use crate::{api_server::execution_sandbox::BlockArgs, fee_model::BatchFeeModelIn
 /// Creates a mempool filter for L2 transactions based on the current L1 gas price.
 /// The filter is used to filter out transactions from the mempool that do not cover expenses
 /// to process them.
-pub fn l2_tx_filter(
+pub async fn l2_tx_filter(
     batch_fee_input_provider: &dyn BatchFeeModelInputProvider,
     vm_version: VmVersion,
 ) -> L2TxFilter {
-    let fee_input = batch_fee_input_provider.get_batch_fee_input();
+    let fee_input = batch_fee_input_provider.get_batch_fee_input().await;
 
     let (base_fee, gas_per_pubdata) = derive_base_fee_and_gas_per_pubdata(fee_input, vm_version);
     L2TxFilter {
@@ -87,7 +87,8 @@ impl<G: BatchFeeModelInputProvider> MempoolFetcher<G> {
             let l2_tx_filter = l2_tx_filter(
                 self.batch_fee_input_provider.as_ref(),
                 protocol_version.into(),
-            );
+            )
+            .await;
 
             let (transactions, nonces) = storage
                 .transactions_dal()