fix(l2): remove one-time checkpoint if already exists (#5083)

**Motivation**

In a previous PR, DB checkpoints were introduced to ensure old state
availability in the current path-based fashion. Every time a batch is
sealed, a checkpoint whose state is the state of the latest block of the
sealed batch is created to be used in the next batch.

The checkpoint is needed in two different steps of the batch commitment:
for batch preparation (this is essentially building the batch) and for
witness generation. Both steps need a non-modified checkpoint, but they
both need to modify the checkpoint to be able to re-execute the batch.

As batch preparation occurs before witness generation, we opted to
create a one-time checkpoint out of the main checkpoint that can be
modified during batch preparation if needed (sometimes the batch was
already available in the DB, and there's no need to re-execute
anything); then, witness generation modifies the original checkpoint as
needed because it is no longer needed.

Once the one-time checkpoint fulfills its purpose, it is removed.
Currently, if batch preparation fails, the one-time checkpoint is not
removed, and after retrying batch preparation, there's another attempt
at creating the one-time checkpoint, which ends in an error because the
directory already exists. We need to either avoid creating the one-time
checkpoint again or to remove the existing one.

**Description**

Remove the existing one-time checkpoint if it already exists.

---------

Co-authored-by: avilagaston9 <[email protected]>
Co-authored-by: Gianbelinche <[email protected]>

Author: 3 people

crates/l2/sequencer/l1_committer.rs

@@ -46,6 +46,7 @@ use ethrex_storage::EngineType;
 use ethrex_storage::Store;
 use ethrex_storage_rollup::StoreRollup;
 use ethrex_vm::{BlockExecutionResult, Evm};
+use rand::Rng;
 use serde::Serialize;
 use std::{
     collections::{BTreeMap, HashMap},
@@ -270,16 +271,50 @@ impl L1Committer {
                     )?;
                 let first_block_to_commit = last_block + 1;
 
+                // We need to guarantee that the checkpoint path is new
+                // to avoid causing a lock error under rocksdb feature.
+                let rand_suffix: u32 = rand::thread_rng().r#gen();
+                let one_time_checkpoint_path = self
+                    .checkpoints_dir
+                    .join(format!("temp_checkpoint_{batch_to_commit}_{rand_suffix}"));
+
+                // For re-execution we need to use a checkpoint to the previous state
+                // (i.e. checkpoint of the state to the latest block from the previous
+                // batch, or the state of the genesis if this is the first batch).
+                // We already have this initial checkpoint as part of the L1Committer
+                // struct, but we need to create a one-time copy of it because
+                // we still need to use the current checkpoint store later for witness
+                // generation.
+
+                let (one_time_checkpoint_store, one_time_checkpoint_blockchain) = self
+                    .create_checkpoint(&self.current_checkpoint_store, &one_time_checkpoint_path)
+                    .await?;
+
                 // Try to prepare batch
+                let result = self
+                    .prepare_batch_from_block(
+                        *last_block,
+                        batch_to_commit,
+                        one_time_checkpoint_store,
+                        one_time_checkpoint_blockchain,
+                    )
+                    .await;
+
+                if one_time_checkpoint_path.exists() {
+                    let _ = remove_dir_all(&one_time_checkpoint_path).inspect_err(|e| {
+                        error!(
+                            "Failed to remove one-time checkpoint directory at path {one_time_checkpoint_path:?}. Should be removed manually. Error: {}", e.to_string()
+                        )
+                    });
+                }
+
                 let (
                     blobs_bundle,
                     new_state_root,
                     message_hashes,
                     privileged_transactions_hash,
                     last_block_of_batch,
-                ) = self
-                    .prepare_batch_from_block(*last_block, batch_to_commit)
-                    .await?;
+                ) = result?;
 
                 if *last_block == last_block_of_batch {
                     debug!("No new blocks to commit, skipping");
@@ -370,6 +405,8 @@ impl L1Committer {
         &mut self,
         mut last_added_block_number: BlockNumber,
         batch_number: u64,
+        one_time_checkpoint_store: Store,
+        one_time_checkpoint_blockchain: Arc<Blockchain>,
     ) -> Result<(BlobsBundle, H256, Vec<H256>, H256, BlockNumber), CommitterError> {
         let first_block_of_batch = last_added_block_number + 1;
         let mut blobs_bundle = BlobsBundle::default();
@@ -391,21 +428,6 @@ impl L1Committer {
 
         info!("Preparing state diff from block {first_block_of_batch}, {batch_number}");
 
-        let one_time_checkpoint_path = self
-            .checkpoints_dir
-            .join(format!("temp_checkpoint_batch_{batch_number}"));
-
-        // For re-execution we need to use a checkpoint to the previous state
-        // (i.e. checkpoint of the state to the latest block from the previous
-        // batch, or the state of the genesis if this is the first batch).
-        // We already have this initial checkpoint as part of the L1Committer
-        // struct, but we need to create a one-time copy of it because
-        // we still need to use the current checkpoint store later for witness
-        // generation.
-        let (one_time_checkpoint_store, one_time_checkpoint_blockchain) = self
-            .create_checkpoint(&self.current_checkpoint_store, &one_time_checkpoint_path)
-            .await?;
-
         loop {
             let block_to_commit_number = last_added_block_number + 1;
 
@@ -660,12 +682,6 @@ impl L1Committer {
         let privileged_transactions_hash =
             compute_privileged_transactions_hash(privileged_transactions_hashes)?;
 
-        remove_dir_all(&one_time_checkpoint_path).map_err(|e| {
-            CommitterError::FailedToCreateCheckpoint(format!(
-                "Failed to remove one-time checkpoint directory {one_time_checkpoint_path:?}: {e}"
-            ))
-        })?;
-
         Ok((
             blobs_bundle,
             new_state_root,
@@ -686,11 +702,30 @@ impl L1Committer {
         )
         .await?;
 
-        let batch_witness = self
-            .current_checkpoint_blockchain
+        let rand_suffix: u32 = rand::thread_rng().r#gen();
+        let one_time_checkpoint_path = self.checkpoints_dir.join(format!(
+            "temp_checkpoint_witness_{}_{rand_suffix}",
+            batch.number
+        ));
+        // We need to create a one-time checkpoint copy because if witness generation fails the checkpoint would be modified
+        let (_, one_time_checkpoint_blockchain) = self
+            .create_checkpoint(&self.current_checkpoint_store, &one_time_checkpoint_path)
+            .await?;
+
+        let result = one_time_checkpoint_blockchain
             .generate_witness_for_blocks_with_fee_configs(&blocks, Some(&fee_configs))
             .await
-            .map_err(CommitterError::FailedToGenerateBatchWitness)?;
+            .map_err(CommitterError::FailedToGenerateBatchWitness);
+
+        if one_time_checkpoint_path.exists() {
+            let _ = remove_dir_all(&one_time_checkpoint_path).inspect_err(|e| {
+                error!(
+                    "Failed to remove one-time checkpoint directory at path {one_time_checkpoint_path:?}. Should be removed manually. Error: {}", e.to_string()
+                )
+            });
+        }
+
+        let batch_witness = result?;
 
         // We still need to differentiate the validium case because for validium
         // we are generating the BlobsBundle with BlobsBundle::default which
@@ -755,7 +790,7 @@ impl L1Committer {
         // We need to skip checkpoint creation if the directory already exists.
         // Sometimes the commit_next_batch task is retried after a failure, and in
         // that case we would try to create a checkpoint again at the same path,
-        // causing an lock error under rocksdb feature.
+        // causing a lock error under rocksdb feature.
         if new_checkpoint_path.exists() {
             debug!("Checkpoint at path {new_checkpoint_path:?} already exists, skipping creation");
             return Ok(());
@@ -778,7 +813,7 @@ impl L1Committer {
     /// 1. Creates a checkpoint of the provided store at the specified path.
     /// 2. Initializes a new store and blockchain for the checkpoint.
     /// 3. Regenerates the head state in the checkpoint store.
-    /// 4. Validates that the checkpoint store's head block number and latest block match those of the original store.
+    /// 4. TODO: Validates that the checkpoint contains the needed state root.
     async fn create_checkpoint(
         &self,
         checkpointee: &Store,
@@ -806,54 +841,8 @@ impl L1Committer {
             self.blockchain.options.clone(),
         ));
 
-        let checkpoint_head_block_number = checkpoint_store.get_latest_block_number().await?;
-
-        let db_head_block_number = checkpointee.get_latest_block_number().await?;
-
-        if checkpoint_head_block_number != db_head_block_number {
-            return Err(CommitterError::FailedToCreateCheckpoint(
-                "checkpoint store head block number does not match main store head block number before regeneration".to_string(),
-            ));
-        }
-
         regenerate_head_state(&checkpoint_store, &checkpoint_blockchain).await?;
 
-        let checkpoint_latest_block_number = checkpoint_store.get_latest_block_number().await?;
-
-        let db_latest_block_number = checkpointee.get_latest_block_number().await?;
-
-        let checkpoint_latest_block = checkpoint_store
-            .get_block_by_number(checkpoint_latest_block_number)
-            .await?
-            .ok_or(CommitterError::FailedToCreateCheckpoint(
-                "latest block not found in checkpoint store".to_string(),
-            ))?;
-
-        let db_latest_block = checkpointee
-            .get_block_by_number(db_latest_block_number)
-            .await?
-            .ok_or(CommitterError::FailedToCreateCheckpoint(
-                "latest block not found in main store".to_string(),
-            ))?;
-
-        if !checkpoint_store.has_state_root(checkpoint_latest_block.header.state_root)? {
-            return Err(CommitterError::FailedToCreateCheckpoint(
-                "checkpoint store state is not regenerated properly".to_string(),
-            ));
-        }
-
-        if checkpoint_latest_block_number != db_head_block_number {
-            return Err(CommitterError::FailedToCreateCheckpoint(
-                "checkpoint store latest block number does not match main store head block number after regeneration".to_string(),
-            ));
-        }
-
-        if checkpoint_latest_block.hash() != db_latest_block.hash() {
-            return Err(CommitterError::FailedToCreateCheckpoint(
-                "checkpoint store latest block hash does not match main store latest block hash after regeneration".to_string(),
-            ));
-        }
-
         Ok((checkpoint_store, checkpoint_blockchain))
     }