feat(eth-sender): add option to pause aggregator for gateway migration (

#2644)

Signed-off-by: tomg10 <[email protected]>

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

@@ -40,6 +40,8 @@ impl EthConfig {
                 l1_batch_min_age_before_execute_seconds: None,
                 max_acceptable_priority_fee_in_gwei: 100000000000,
                 pubdata_sending_mode: PubdataSendingMode::Calldata,
+                tx_aggregation_paused: false,
+                tx_aggregation_only_prove_and_execute: false,
             }),
             gas_adjuster: Some(GasAdjusterConfig {
                 default_priority_fee_per_gas: 1000000000,
@@ -119,6 +121,12 @@ pub struct SenderConfig {
 
     /// The mode in which we send pubdata: Calldata, Blobs or Custom (DA layers, Object Store, etc.)
     pub pubdata_sending_mode: PubdataSendingMode,
+    /// special mode specifically for gateway migration to allow all inflight txs to be processed
+    #[serde(default = "SenderConfig::default_tx_aggregation_paused")]
+    pub tx_aggregation_paused: bool,
+    /// special mode specifically for gateway migration to decrease number of non-executed batches
+    #[serde(default = "SenderConfig::default_tx_aggregation_only_prove_and_execute")]
+    pub tx_aggregation_only_prove_and_execute: bool,
 }
 
 impl SenderConfig {
@@ -153,6 +161,13 @@ impl SenderConfig {
             .ok()
             .map(|pk| pk.parse().unwrap())
     }
+
+    const fn default_tx_aggregation_paused() -> bool {
+        false
+    }
+    const fn default_tx_aggregation_only_prove_and_execute() -> bool {
+        false
+    }
 }
 
 #[derive(Debug, Deserialize, Copy, Clone, PartialEq, Default)]

core/lib/config/src/testonly.rs

@@ -407,6 +407,8 @@ impl Distribution<configs::eth_sender::SenderConfig> for EncodeDist {
             l1_batch_min_age_before_execute_seconds: self.sample(rng),
             max_acceptable_priority_fee_in_gwei: self.sample(rng),
             pubdata_sending_mode: PubdataSendingMode::Calldata,
+            tx_aggregation_paused: false,
+            tx_aggregation_only_prove_and_execute: false,
         }
     }
 }

core/lib/env_config/src/eth_sender.rs

@@ -70,6 +70,8 @@ mod tests {
                     l1_batch_min_age_before_execute_seconds: Some(1000),
                     max_acceptable_priority_fee_in_gwei: 100_000_000_000,
                     pubdata_sending_mode: PubdataSendingMode::Calldata,
+                    tx_aggregation_only_prove_and_execute: false,
+                    tx_aggregation_paused: false,
                 }),
                 gas_adjuster: Some(GasAdjusterConfig {
                     default_priority_fee_per_gas: 20000000000,

core/lib/protobuf_config/src/eth.rs

@@ -113,6 +113,8 @@ impl ProtoRepr for proto::Sender {
                 .and_then(|x| Ok(proto::PubdataSendingMode::try_from(*x)?))
                 .context("pubdata_sending_mode")?
                 .parse(),
+            tx_aggregation_only_prove_and_execute: self.tx_aggregation_paused.unwrap_or(false),
+            tx_aggregation_paused: self.tx_aggregation_only_prove_and_execute.unwrap_or(false),
         })
     }
 
@@ -143,6 +145,8 @@ impl ProtoRepr for proto::Sender {
             pubdata_sending_mode: Some(
                 proto::PubdataSendingMode::new(&this.pubdata_sending_mode).into(),
             ),
+            tx_aggregation_only_prove_and_execute: Some(this.tx_aggregation_only_prove_and_execute),
+            tx_aggregation_paused: Some(this.tx_aggregation_paused),
         }
     }
 }

core/lib/protobuf_config/src/proto/config/eth_sender.proto

@@ -46,6 +46,8 @@ message Sender {
   optional uint64 max_acceptable_priority_fee_in_gwei = 16; // required; gwei
   optional PubdataSendingMode pubdata_sending_mode = 18; // required
   reserved 19; reserved "proof_loading_mode";
+  optional bool tx_aggregation_paused = 20; // required
+  optional bool tx_aggregation_only_prove_and_execute = 21; // required
 }
 
 message GasAdjuster {

core/node/eth_sender/src/aggregated_operations.rs

@@ -53,4 +53,9 @@ impl AggregatedOperation {
             Self::Execute(op) => op.l1_batches[0].header.protocol_version.unwrap(),
         }
     }
+
+    pub fn is_prove_or_execute(&self) -> bool {
+        self.get_action_type() == AggregatedActionType::PublishProofOnchain
+            || self.get_action_type() == AggregatedActionType::Execute
+    }
 }

core/node/eth_sender/src/eth_tx_aggregator.rs

@@ -355,6 +355,25 @@ impl EthTxAggregator {
             )
             .await
         {
+            if self.config.tx_aggregation_paused {
+                tracing::info!(
+                    "Skipping sending operation of type {} for batches {}-{} \
+                as tx_aggregation_paused=true",
+                    agg_op.get_action_type(),
+                    agg_op.l1_batch_range().start(),
+                    agg_op.l1_batch_range().end()
+                );
+                return Ok(());
+            }
+            if self.config.tx_aggregation_only_prove_and_execute && !agg_op.is_prove_or_execute() {
+                tracing::info!(
+                    "Skipping sending commit operation for batches {}-{} \
+                as tx_aggregation_only_prove_and_execute=true",
+                    agg_op.l1_batch_range().start(),
+                    agg_op.l1_batch_range().end()
+                );
+                return Ok(());
+            }
             let tx = self
                 .save_eth_tx(storage, &agg_op, contracts_are_pre_shared_bridge, false)
                 .await?;