fix(en): make EN use main node's fee input (#3489)

## What ❔

Alternative to #3487

Before this PR EN based its gas pricing solely on main node's fee params
which are based on immediate market conditions and not representative of
the open batch's fee input. This is a problem on environments with
long-running batches. This PR makes EN fetch main node's batch fee input
along with fee params and report fetched batch fee input from its fee
input provider.

## 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. -->

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

core/node/fee_model/src/l1_gas_price/main_node_fetcher.rs

@@ -3,8 +3,9 @@ use std::{
     time::Duration,
 };
 
+use async_trait::async_trait;
 use tokio::sync::watch::Receiver;
-use zksync_types::fee_model::FeeParams;
+use zksync_types::fee_model::{BatchFeeInput, FeeParams};
 use zksync_web3_decl::{
     client::{DynClient, L2},
     error::ClientRpcContext,
@@ -15,37 +16,50 @@ use crate::BatchFeeModelInputProvider;
 
 const SLEEP_INTERVAL: Duration = Duration::from_secs(5);
 
-/// This structure maintains the known L1 gas price by periodically querying
+/// This structure maintains the known fee params/input by periodically querying
 /// the main node.
+///
 /// It is required since the main node doesn't only observe the current L1 gas price,
 /// but also applies adjustments to it in order to smooth out the spikes.
 /// The same algorithm cannot be consistently replicated on the external node side,
 /// since it relies on the configuration, which may change.
 #[derive(Debug)]
 pub struct MainNodeFeeParamsFetcher {
     client: Box<DynClient<L2>>,
-    main_node_fee_params: RwLock<FeeParams>,
+    main_node_fee_state: RwLock<(FeeParams, BatchFeeInput)>,
 }
 
 impl MainNodeFeeParamsFetcher {
     pub fn new(client: Box<DynClient<L2>>) -> Self {
+        let fee_params = FeeParams::sensible_v1_default();
+        let fee_input = fee_params.scale(1.0, 1.0);
         Self {
             client: client.for_component("fee_params_fetcher"),
-            main_node_fee_params: RwLock::new(FeeParams::sensible_v1_default()),
+            main_node_fee_state: RwLock::new((fee_params, fee_input)),
         }
     }
 
     pub async fn run(self: Arc<Self>, mut stop_receiver: Receiver<bool>) -> anyhow::Result<()> {
         while !*stop_receiver.borrow_and_update() {
-            let fetch_result = self
-                .client
-                .get_fee_params()
-                .rpc_context("get_fee_params")
-                .await;
-            let main_node_fee_params = match fetch_result {
-                Ok(price) => price,
+            // We query fee params and fee input together to minimize the potential for them to be
+            // out of sync. They can still be fetched out of sync in rare circumstances but nothing
+            // in the system *directly* relies on `BatchFeeModelInputProvider::get_fee_model_params`
+            // except for `zks_getFeeParams`. Which is likely fine because EN is essentially
+            // mimicking how it observed the call to main node.
+            let (params_result, input_result) = tokio::join!(
+                self.client.get_fee_params().rpc_context("get_fee_params"),
+                self.client
+                    .get_batch_fee_input()
+                    .rpc_context("get_batch_fee_input")
+            );
+            let fee_state_result =
+                params_result.and_then(|params| input_result.map(|input| (params, input)));
+            let main_node_fee_state = match fee_state_result {
+                Ok((fee_params, fee_input)) => {
+                    (fee_params, BatchFeeInput::PubdataIndependent(fee_input))
+                }
                 Err(err) => {
-                    tracing::warn!("Unable to get the gas price: {}", err);
+                    tracing::warn!("Unable to get main node's fee params/input: {}", err);
                     // A delay to avoid spamming the main node with requests.
                     if tokio::time::timeout(SLEEP_INTERVAL, stop_receiver.changed())
                         .await
@@ -56,7 +70,7 @@ impl MainNodeFeeParamsFetcher {
                     continue;
                 }
             };
-            *self.main_node_fee_params.write().unwrap() = main_node_fee_params;
+            *self.main_node_fee_state.write().unwrap() = main_node_fee_state;
 
             if tokio::time::timeout(SLEEP_INTERVAL, stop_receiver.changed())
                 .await
@@ -71,8 +85,18 @@ impl MainNodeFeeParamsFetcher {
     }
 }
 
+#[async_trait]
 impl BatchFeeModelInputProvider for MainNodeFeeParamsFetcher {
+    async fn get_batch_fee_input_scaled(
+        &self,
+        // EN's scale factors are ignored as we have already fetched scaled fee input from main node
+        _l1_gas_price_scale_factor: f64,
+        _l1_pubdata_price_scale_factor: f64,
+    ) -> anyhow::Result<BatchFeeInput> {
+        Ok(self.main_node_fee_state.read().unwrap().1)
+    }
+
     fn get_fee_model_params(&self) -> FeeParams {
-        *self.main_node_fee_params.read().unwrap()
+        self.main_node_fee_state.read().unwrap().0
     }
 }