Sequencer better DA fork handling & minor improvements

crates/sequencer/src/runner.rs

@@ -1,8 +1,8 @@
+use std::cmp;
 use std::collections::HashSet;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
-use std::vec;
 
 use alloy_eips::eip2718::Encodable2718;
 use alloy_primitives::{Address, Bytes, TxHash};
@@ -254,7 +254,7 @@ where
         >,
         pub_key: &[u8],
         prestate: ProverStorage<SnapshotManager>,
-        da_block_header: <<Da as DaService>::Spec as DaSpec>::BlockHeader,
+        da_block_header: &<<Da as DaService>::Spec as DaSpec>::BlockHeader,
         soft_confirmation_info: HookSoftConfirmationInfo,
         l2_block_mode: L2BlockMode,
     ) -> anyhow::Result<(Vec<RlpEvmTransaction>, Vec<TxHash>)> {
@@ -273,7 +273,7 @@ where
             match self.stf.begin_soft_confirmation(
                 pub_key,
                 &mut working_set_to_discard,
-                &da_block_header,
+                da_block_header,
                 &soft_confirmation_info,
             ) {
                 Ok(_) => {
@@ -415,12 +415,12 @@ where
 
     async fn produce_l2_block(
         &mut self,
-        da_block: <Da as DaService>::FilteredBlock,
+        da_block_header: &<<Da as DaService>::Spec as DaSpec>::BlockHeader,
         l1_fee_rate: u128,
         l2_block_mode: L2BlockMode,
     ) -> anyhow::Result<(u64, u64, StateDiff)> {
         let start = Instant::now();
-        let da_height = da_block.header().height();
+        let da_height = da_block_header.height();
         let (l2_height, l1_height) = match self
             .ledger_db
             .get_head_soft_confirmation()
@@ -451,9 +451,9 @@ where
 
         let soft_confirmation_info = HookSoftConfirmationInfo {
             l2_height,
-            da_slot_height: da_block.header().height(),
-            da_slot_hash: da_block.header().hash().into(),
-            da_slot_txs_commitment: da_block.header().txs_commitment().into(),
+            da_slot_height: da_block_header.height(),
+            da_slot_hash: da_block_header.hash().into(),
+            da_slot_txs_commitment: da_block_header.txs_commitment().into(),
             pre_state_root: self.state_root.clone().as_ref().to_vec(),
             deposit_data: deposit_data.clone(),
             current_spec: active_fork_spec,
@@ -468,7 +468,7 @@ where
             .map_err(Into::<anyhow::Error>::into)?;
         debug!(
             "Applying soft confirmation on DA block: {}",
-            hex::encode(da_block.header().hash().into())
+            hex::encode(da_block_header.hash().into())
         );
 
         let evm_txs = self.get_best_transactions()?;
@@ -481,16 +481,15 @@ where
                 evm_txs,
                 &pub_key,
                 prestate.clone(),
-                da_block.header().clone(),
+                da_block_header,
                 soft_confirmation_info.clone(),
                 l2_block_mode,
             )
             .await?;
 
         let prestate = self
             .storage_manager
-            .create_storage_on_l2_height(l2_height)
-            .map_err(Into::<anyhow::Error>::into)?;
+            .create_storage_on_l2_height(l2_height)?;
 
         let checkpoint =
             StateCheckpoint::with_witness(prestate.clone(), Default::default(), Default::default());
@@ -500,7 +499,7 @@ where
         match self.stf.begin_soft_confirmation(
             &pub_key,
             &mut working_set,
-            da_block.header(),
+            da_block_header,
             &soft_confirmation_info,
         ) {
             Ok(_) => {
@@ -526,16 +525,15 @@ where
                             &txs_new,
                             &mut working_set,
                         )
-                        // TODO: handle this error
                         .expect("dry_run_transactions should have already checked this");
                 }
 
                 // create the unsigned batch with the txs then sign th sc
                 let unsigned_batch = UnsignedSoftConfirmation::new(
                     l2_height,
-                    da_block.header().height(),
-                    da_block.header().hash().into(),
-                    da_block.header().txs_commitment().into(),
+                    da_block_header.height(),
+                    da_block_header.hash().into(),
+                    da_block_header.txs_commitment().into(),
                     &txs,
                     &txs_new,
                     deposit_data.clone(),
@@ -602,11 +600,11 @@ where
 
                 // connect L1 and L2 height
                 self.ledger_db.extend_l2_range_of_l1_slot(
-                    SlotNumber(da_block.header().height()),
+                    SlotNumber(da_block_header.height()),
                     SoftConfirmationNumber(l2_height),
                 )?;
 
-                let l1_height = da_block.header().height();
+                let l1_height = da_block_header.height();
                 info!(
                     "New block #{}, DA #{}, Tx count: #{}",
                     l2_height, l1_height, evm_txs_count,
@@ -642,7 +640,7 @@ where
 
                 Ok((
                     l2_height,
-                    da_block.header().height(),
+                    da_block_header.height(),
                     soft_confirmation_result.state_diff,
                 ))
             }
@@ -675,17 +673,15 @@ where
         }
 
         let (mut last_finalized_block, mut l1_fee_rate) =
-            match get_da_block_data(self.da_service.clone()).await {
-                Ok(l1_data) => l1_data,
-                Err(e) => {
-                    error!("{}", e);
-                    return Err(e);
-                }
-            };
-        let mut last_finalized_height = last_finalized_block.header().height();
+            get_da_block_data(self.da_service.clone()).await?;
+        let last_finalized_height = last_finalized_block.header().height();
 
         let mut last_used_l1_height = match self.ledger_db.get_head_soft_confirmation() {
-            Ok(Some((_, sb))) => sb.da_slot_height,
+            Ok(Some((_, sb))) => {
+                ensure_no_da_forks(self.da_service.clone(), sb.da_slot_height, sb.da_slot_hash)
+                    .await;
+                sb.da_slot_height
+            }
             Ok(None) => last_finalized_height, // starting for the first time
             Err(e) => {
                 return Err(anyhow!("previous L1 height: {}", e));
@@ -718,6 +714,8 @@ where
                 da_height_update_tx,
                 self.config.da_update_interval_ms,
                 cancellation_token,
+                last_finalized_height,
+                last_finalized_block.hash(),
             )
         });
 
@@ -736,16 +734,17 @@ where
             tokio::select! {
                 // Receive updates from DA layer worker.
                 l1_data = da_height_update_rx.recv() => {
+                    let l1_data = l1_data.expect("Da block channel closed");
                     // Stop receiving updates from DA layer until we have caught up.
                     if missed_da_blocks_count > 0 {
                         continue;
                     }
-                    if let Some(l1_data) = l1_data {
-                        (last_finalized_block, l1_fee_rate) = l1_data;
-                        last_finalized_height = last_finalized_block.header().height();
 
-                        missed_da_blocks_count = self.da_blocks_missed(last_finalized_height, last_used_l1_height);
-                    }
+                    (last_finalized_block, l1_fee_rate) = l1_data;
+                    let last_finalized_height = last_finalized_block.header().height();
+
+                    missed_da_blocks_count = self.da_blocks_missed(last_finalized_height, last_used_l1_height);
+
                     SEQUENCER_METRICS.current_l1_block.set(last_finalized_height as f64);
                 },
                 // If sequencer is in test mode, it will build a block every time it receives a message
@@ -761,7 +760,7 @@ where
                         missed_da_blocks_count = 0;
                     }
 
-                    match self.produce_l2_block(last_finalized_block.clone(), l1_fee_rate, L2BlockMode::NotEmpty).await {
+                    match self.produce_l2_block(last_finalized_block.header(), l1_fee_rate, L2BlockMode::NotEmpty).await {
                         Ok((l2_height, l1_block_number, state_diff)) => {
                             last_used_l1_height = l1_block_number;
 
@@ -781,8 +780,6 @@ where
                     // last_finalized_block. If there are missed DA blocks, we start producing
                     // empty blocks at ~2 second rate, 1 L2 block per respective missed DA block
                     // until we know we caught up with L1.
-                    let da_block = last_finalized_block.clone();
-
                     if missed_da_blocks_count > 0 {
                         if let Err(e) = self.process_missed_da_blocks(missed_da_blocks_count, last_used_l1_height, l1_fee_rate).await {
                             error!("Sequencer error: {}", e);
@@ -793,7 +790,7 @@ where
                     }
 
 
-                    match self.produce_l2_block(da_block, l1_fee_rate, L2BlockMode::NotEmpty).await {
+                    match self.produce_l2_block(last_finalized_block.header(), l1_fee_rate, L2BlockMode::NotEmpty).await {
                         Ok((l2_height, l1_block_number, state_diff)) => {
                             last_used_l1_height = l1_block_number;
 
@@ -982,7 +979,7 @@ where
     pub async fn restore_mempool(&self) -> Result<(), anyhow::Error> {
         let mempool_txs = self.ledger_db.get_mempool_txs()?;
         for (_, tx) in mempool_txs {
-            let recovered = recover_raw_transaction(Bytes::from(tx.as_slice().to_vec()))?;
+            let recovered = recover_raw_transaction(Bytes::from(tx))?;
             let pooled_tx = EthPooledTransaction::from_pooled(recovered);
 
             let _ = self.mempool.add_external_transaction(pooled_tx).await?;
@@ -1051,7 +1048,7 @@ where
             .await?;
 
             debug!("Created an empty L2 for L1={}", needed_da_block_height);
-            self.produce_l2_block(da_block, l1_fee_rate, L2BlockMode::Empty)
+            self.produce_l2_block(da_block.header(), l1_fee_rate, L2BlockMode::Empty)
                 .await?;
         }
 
@@ -1084,6 +1081,8 @@ async fn da_block_monitor<Da>(
     sender: mpsc::Sender<L1Data<Da>>,
     loop_interval: u64,
     cancellation_token: CancellationToken,
+    mut last_da_height: u64,
+    mut last_da_hash: [u8; 32],
 ) where
     Da: DaService,
 {
@@ -1094,15 +1093,40 @@ async fn da_block_monitor<Da>(
                 return;
             }
             l1_data = get_da_block_data(da_service.clone()) => {
-                let l1_data = match l1_data {
+                let (l1_block, l1_fee_rate) = match l1_data {
                     Ok(l1_data) => l1_data,
                     Err(e) => {
                         error!("Could not fetch L1 data, {}", e);
                         continue;
                     }
                 };
 
-                let _ = sender.send(l1_data).await;
+                let header = l1_block.header();
+                match header.height().cmp(&last_da_height) {
+                    cmp::Ordering::Less => panic!("DA blockchain moved in reverse"),
+                    cmp::Ordering::Equal => assert_eq!(
+                        last_da_hash,
+                        header.hash().into(),
+                        "Detected fork on DA block {}",
+                        last_da_height,
+                    ),
+                    cmp::Ordering::Greater => {
+                        if header.height() == last_da_height + 1 {
+                            assert_eq!(
+                                last_da_hash,
+                                header.prev_hash().into(),
+                                "Detected fork on DA block {}",
+                                last_da_height,
+                            );
+                        } else {
+                            ensure_no_da_forks(da_service.clone(), last_da_height, last_da_hash).await;
+                        }
+                        last_da_height = header.height();
+                        last_da_hash = header.hash().into();
+                    }
+                }
+
+                let _ = sender.send((l1_block, l1_fee_rate)).await;
 
                 sleep(Duration::from_millis(loop_interval)).await;
             },
@@ -1114,31 +1138,43 @@ async fn get_da_block_data<Da>(da_service: Arc<Da>) -> anyhow::Result<L1Data<Da>
 where
     Da: DaService,
 {
-    let last_finalized_height = match da_service.get_last_finalized_block_header().await {
-        Ok(header) => header.height(),
-        Err(e) => {
-            return Err(anyhow!("Finalized L1 height: {}", e));
-        }
-    };
+    let last_finalized_height = da_service
+        .get_last_finalized_block_header()
+        .await
+        .map_err(|e| anyhow!("Failed to fetch finalized block header: {e}"))?
+        .height();
 
-    let last_finalized_block = match da_service.get_block_at(last_finalized_height).await {
-        Ok(block) => block,
-        Err(e) => {
-            return Err(anyhow!("Finalized L1 block: {}", e));
-        }
-    };
+    let last_finalized_block = da_service
+        .get_block_at(last_finalized_height)
+        .await
+        .map_err(|e| anyhow!("Failed to fetch finalized l1 block: {e}"))?;
 
     debug!(
         "Sequencer: last finalized L1 height: {:?}",
         last_finalized_height
     );
 
-    let l1_fee_rate = match da_service.get_fee_rate().await {
-        Ok(fee_rate) => fee_rate,
-        Err(e) => {
-            return Err(anyhow!("L1 fee rate: {}", e));
-        }
-    };
+    let l1_fee_rate = da_service
+        .get_fee_rate()
+        .await
+        .map_err(|e| anyhow!("Failed to fetch l1 fee rate: {e}"))?;
 
     Ok((last_finalized_block, l1_fee_rate))
 }
+
+async fn ensure_no_da_forks<Da: DaService>(
+    da_service: Arc<Da>,
+    da_block_height: u64,
+    expected_da_block_hash: [u8; 32],
+) {
+    let da_block = da_service
+        .get_block_at(da_block_height)
+        .await
+        .expect("Should not continue if we cannot ensure there are no DA forks");
+    assert_eq!(
+        da_block.header().hash(),
+        expected_da_block_hash.into(),
+        "Detected fork on DA block {}",
+        da_block_height,
+    );
+}