From 32116878ba0ac68a7b90afa1a7e0fe170bdcd902 Mon Sep 17 00:00:00 2001 From: Stanislav Bezkorovainyi Date: Fri, 19 Jan 2024 14:48:51 +0100 Subject: [PATCH] feat(api): Consider State keeper fee model input in the API (#901) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 ❔ ## Checklist - [ ] 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 | 33 ++++++++++ .../src/api_server/tx_sender/mod.rs | 33 ++++++---- .../web3/backend_jsonrpsee/namespaces/zks.rs | 2 +- .../src/api_server/web3/namespaces/debug.rs | 3 +- .../src/api_server/web3/namespaces/zks.rs | 3 +- core/lib/zksync_core/src/fee_model.rs | 61 ++++++++++++++++++- core/lib/zksync_core/src/lib.rs | 7 ++- .../src/state_keeper/io/mempool.rs | 3 +- .../src/state_keeper/io/tests/mod.rs | 6 +- .../src/state_keeper/mempool_actor.rs | 7 ++- 10 files changed, 130 insertions(+), 28 deletions(-) diff --git a/core/lib/types/src/fee_model.rs b/core/lib/types/src/fee_model.rs index 8f8d43a4ab7e..bb183b327db6 100644 --- a/core/lib/types/src/fee_model.rs +++ b/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. diff --git a/core/lib/zksync_core/src/api_server/tx_sender/mod.rs b/core/lib/zksync_core/src/api_server/tx_sender/mod.rs index 35740950b3aa..653e75b5feb3 100644 --- a/core/lib/zksync_core/src/api_server/tx_sender/mod.rs +++ b/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 diff --git a/core/lib/zksync_core/src/api_server/web3/backend_jsonrpsee/namespaces/zks.rs b/core/lib/zksync_core/src/api_server/web3/backend_jsonrpsee/namespaces/zks.rs index 8ba611b14073..db0760a66c76 100644 --- a/core/lib/zksync_core/src/api_server/web3/backend_jsonrpsee/namespaces/zks.rs +++ b/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 { - Ok(self.get_l1_gas_price_impl()) + Ok(self.get_l1_gas_price_impl().await) } async fn get_fee_params(&self) -> RpcResult { diff --git a/core/lib/zksync_core/src/api_server/web3/namespaces/debug.rs b/core/lib/zksync_core/src/api_server/web3/namespaces/debug.rs index 693ebb8f3582..4014296f5568 100644 --- a/core/lib/zksync_core/src/api_server/web3/namespaces/debug.rs +++ b/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, } diff --git a/core/lib/zksync_core/src/api_server/web3/namespaces/zks.rs b/core/lib/zksync_core/src/api_server/web3/namespaces/zks.rs index 5b5aad6e1d43..8bb610bbb9a7 100644 --- a/core/lib/zksync_core/src/api_server/web3/namespaces/zks.rs +++ b/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(); diff --git a/core/lib/zksync_core/src/fee_model.rs b/core/lib/zksync_core/src/fee_model.rs index 9798beffd80f..8d43cdc99a7a 100644 --- a/core/lib/zksync_core/src/fee_model.rs +++ b/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, + 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( diff --git a/core/lib/zksync_core/src/lib.rs b/core/lib/zksync_core/src/lib.rs index a246b43216ae..8d7a9d630dc0 100644 --- a/core/lib/zksync_core/src/lib.rs +++ b/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 diff --git a/core/lib/zksync_core/src/state_keeper/io/mempool.rs b/core/lib/zksync_core/src/state_keeper/io/mempool.rs index 9ba6ff0ac9fd..4fa9a79638b0 100644 --- a/core/lib/zksync_core/src/state_keeper/io/mempool.rs +++ b/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; diff --git a/core/lib/zksync_core/src/state_keeper/io/tests/mod.rs b/core/lib/zksync_core/src/state_keeper/io/tests/mod.rs index b210307f2b95..69c68f3f5c1b 100644 --- a/core/lib/zksync_core/src/state_keeper/io/tests/mod.rs +++ b/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 diff --git a/core/lib/zksync_core/src/state_keeper/mempool_actor.rs b/core/lib/zksync_core/src/state_keeper/mempool_actor.rs index 673f0d5ea7aa..070783e51af8 100644 --- a/core/lib/zksync_core/src/state_keeper/mempool_actor.rs +++ b/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 MempoolFetcher { let l2_tx_filter = l2_tx_filter( self.batch_fee_input_provider.as_ref(), protocol_version.into(), - ); + ) + .await; let (transactions, nonces) = storage .transactions_dal()