feat(l2): batch reversion (#3136)

**Motivation**

As outlined in #3124, sometimes a committed batch can't be verified or
the operator wants to prevent it from going though.

**Description**

This PR implements a `revertBatch` function that allows reverting back
to any batch, as long as no verified batches are being discarded.

There's also a l2 CLI subcommand, revert-batch that lets you revert a
batch and remove it from the local database.

Usage on local network:
```
PRIVATE_KEY=key cargo run --features l2,rollup_storage_libmdbx -- l2 revert-batch \
  <batch to revert to> <OnChainProposer address> \
  --datadir dev_ethrex_l2 --network test_data/genesis-l2.json
```

Closes #3124

Author: iovoid

cmd/ethrex/Cargo.toml

@@ -17,6 +17,7 @@ ethrex-rlp.workspace = true
 ethrex-storage-rollup = { workspace = true, optional = true }
 ethrex-l2 = { workspace = true, optional = true }
 ethrex-l2-common = { workspace = true, optional = true }
+ethrex-sdk = { workspace = true, optional = true }
 
 bytes.workspace = true
 hex.workspace = true
@@ -71,6 +72,7 @@ blst = ["ethrex-vm/blst"]
 l2 = [
   "dep:ethrex-l2",
   "dep:ethrex-l2-common",
+  "dep:ethrex-sdk",
   "ethrex-vm/l2",
   "ethrex-blockchain/l2",
   "ethrex-rpc/l2",

cmd/ethrex/l2/command.rs

@@ -6,7 +6,8 @@ use crate::{
         init_metrics, init_network, init_rollup_store, init_rpc_api, init_store,
     },
     l2::options::Options,
-    utils::{NodeConfigFile, set_datadir, store_node_config_file},
+    networks::Network,
+    utils::{NodeConfigFile, parse_private_key, set_datadir, store_node_config_file},
 };
 use clap::Subcommand;
 use ethrex_common::{
@@ -15,6 +16,8 @@ use ethrex_common::{
 };
 use ethrex_l2::SequencerConfig;
 use ethrex_l2_common::state_diff::StateDiff;
+use ethrex_l2_sdk::call_contract;
+use ethrex_l2_sdk::calldata::Value;
 use ethrex_p2p::network::peer_table;
 use ethrex_rpc::{
     EthClient,
@@ -27,6 +30,7 @@ use eyre::OptionExt;
 use itertools::Itertools;
 use keccak_hash::keccak;
 use reqwest::Url;
+use secp256k1::SecretKey;
 use std::{
     fs::{create_dir_all, read_dir},
     future::IntoFuture,
@@ -79,6 +83,39 @@ pub enum Command {
         #[arg(short = 'c', long, help = "Address of the L2 proposer coinbase")]
         coinbase: Address,
     },
+    #[command(about = "Reverts unverified batches.")]
+    RevertBatch {
+        #[arg(help = "ID of the batch to revert to")]
+        batch: u64,
+        #[arg(help = "The address of the OnChainProposer contract")]
+        contract_address: Address,
+        #[arg(long, value_parser = parse_private_key, env = "PRIVATE_KEY", help = "The private key of the owner. Assumed to have sequencing permission.")]
+        private_key: Option<SecretKey>,
+        #[arg(
+            long,
+            default_value = "http://localhost:8545",
+            env = "RPC_URL",
+            help = "URL of the L1 RPC"
+        )]
+        rpc_url: Url,
+        #[arg(
+            long = "network",
+            default_value_t = Network::default(),
+            value_name = "GENESIS_FILE_PATH",
+            help = "Receives a `Genesis` struct in json format. This is the only argument which is required. You can look at some example genesis files at `test_data/genesis*`.",
+            env = "ETHREX_NETWORK",
+            value_parser = clap::value_parser!(Network),
+        )]
+        network: Network,
+        #[arg(
+            long = "datadir",
+            value_name = "DATABASE_DIRECTORY",
+            default_value = DEFAULT_L2_DATADIR,
+            help = "Receives the name of the directory where the Database is located.",
+            env = "ETHREX_DATADIR"
+        )]
+        datadir: String,
+    },
 }
 
 impl Command {
@@ -429,6 +466,63 @@ impl Command {
                     }
                 }
             }
+            Command::RevertBatch {
+                batch,
+                contract_address,
+                rpc_url,
+                private_key,
+                datadir,
+                network,
+            } => {
+                let data_dir = set_datadir(&datadir);
+                let rollup_store_dir = data_dir.clone() + "/rollup_store";
+
+                let client = EthClient::new(rpc_url.as_str())?;
+                if let Some(private_key) = private_key {
+                    info!("Pausing OnChainProposer...");
+                    call_contract(&client, &private_key, contract_address, "pause()", vec![])
+                        .await?;
+                    info!("Doing revert on OnChainProposer...");
+                    call_contract(
+                        &client,
+                        &private_key,
+                        contract_address,
+                        "revertBatch(uint256)",
+                        vec![Value::Uint(batch.into())],
+                    )
+                    .await?;
+                } else {
+                    info!("Private key not given, not updating contract.");
+                }
+                info!("Updating store...");
+                let rollup_store = init_rollup_store(&rollup_store_dir).await;
+                let last_kept_block = rollup_store
+                    .get_block_numbers_by_batch(batch)
+                    .await?
+                    .and_then(|kept_blocks| kept_blocks.iter().max().cloned())
+                    .unwrap_or(0);
+
+                let genesis = network.get_genesis();
+                let store = init_store(&data_dir, genesis).await;
+
+                rollup_store.revert_to_batch(batch).await?;
+                store.update_latest_block_number(last_kept_block).await?;
+
+                let mut block_to_delete = last_kept_block + 1;
+                while store
+                    .get_canonical_block_hash(block_to_delete)
+                    .await?
+                    .is_some()
+                {
+                    store.remove_block(block_to_delete).await?;
+                    block_to_delete += 1;
+                }
+                if let Some(private_key) = private_key {
+                    info!("Unpausing OnChainProposer...");
+                    call_contract(&client, &private_key, contract_address, "unpause()", vec![])
+                        .await?;
+                }
+            }
         }
         Ok(())
     }

crates/l2/contracts/src/l1/OnChainProposer.sol

@@ -4,6 +4,7 @@ pragma solidity =0.8.29;
 import "@openzeppelin/contracts-upgradeable/proxy/utils/UUPSUpgradeable.sol";
 import "@openzeppelin/contracts-upgradeable/proxy/utils/Initializable.sol";
 import "@openzeppelin/contracts-upgradeable/access/Ownable2StepUpgradeable.sol";
+import "@openzeppelin/contracts-upgradeable/utils/PausableUpgradeable.sol";
 import "./interfaces/IOnChainProposer.sol";
 import {CommonBridge} from "./CommonBridge.sol";
 import {ICommonBridge} from "./interfaces/ICommonBridge.sol";
@@ -17,7 +18,8 @@ contract OnChainProposer is
     IOnChainProposer,
     Initializable,
     UUPSUpgradeable,
-    Ownable2StepUpgradeable
+    Ownable2StepUpgradeable,
+    PausableUpgradeable
 {
     /// @notice Committed batches data.
     /// @dev This struct holds the information about the committed batches.
@@ -219,7 +221,7 @@ contract OnChainProposer is
         bytes32 withdrawalsLogsMerkleRoot,
         bytes32 processedDepositLogsRollingHash,
         bytes32 lastBlockHash
-    ) external override onlySequencer {
+    ) external override onlySequencer whenNotPaused {
         // TODO: Refactor validation
         require(
             batchNumber == lastCommittedBatch + 1,
@@ -289,7 +291,7 @@ contract OnChainProposer is
         //tdx
         bytes calldata tdxPublicValues,
         bytes memory tdxSignature
-    ) external override onlySequencer {
+    ) external override onlySequencer whenNotPaused {
         // TODO: Refactor validation
         // TODO: imageid, programvkey and riscvvkey should be constants
         // TODO: organize each zkvm proof arguments in their own structs
@@ -355,7 +357,7 @@ contract OnChainProposer is
         bytes calldata alignedPublicInputs,
         bytes32 alignedProgramVKey,
         bytes32[] calldata alignedMerkleProof
-    ) external override onlySequencer {
+    ) external override onlySequencer whenNotPaused {
         require(
             ALIGNEDPROOFAGGREGATOR != DEV_MODE,
             "OnChainProposer: ALIGNEDPROOFAGGREGATOR is not set"
@@ -452,9 +454,36 @@ contract OnChainProposer is
         );
     }
 
+    /// @inheritdoc IOnChainProposer
+    function revertBatch(
+        uint256 batchNumber
+    ) external override onlySequencer whenPaused {
+        require(batchNumber >= lastVerifiedBatch, "OnChainProposer: can't revert verified batch");
+        require(batchNumber < lastCommittedBatch, "OnChainProposer: no batches are being reverted");
+
+        // Remove old batches
+        for (uint256 i = batchNumber; i < lastCommittedBatch; i++) {
+            delete batchCommitments[i + 1];
+        }
+
+        lastCommittedBatch = batchNumber;
+
+        emit BatchReverted(batchCommitments[lastCommittedBatch].newStateRoot);
+    }
+
     /// @notice Allow owner to upgrade the contract.
     /// @param newImplementation the address of the new implementation
     function _authorizeUpgrade(
         address newImplementation
     ) internal virtual override onlyOwner {}
+
+    /// @inheritdoc IOnChainProposer
+    function pause() external override onlyOwner {
+        _pause();
+    }
+
+    /// @inheritdoc IOnChainProposer
+    function unpause() external override onlyOwner {
+        _unpause();
+    }
 }

crates/l2/contracts/src/l1/interfaces/IOnChainProposer.sol

@@ -23,6 +23,10 @@ interface IOnChainProposer {
     /// @dev Event emitted when a batch is verified.
     event BatchVerified(uint256 indexed lastVerifiedBatch);
 
+    /// @notice A batch has been reverted.
+    /// @dev Event emitted when a batch is reverted.
+    event BatchReverted(bytes32 indexed newStateRoot);
+
     /// @notice Set the bridge address for the first time.
     /// @dev This method is separated from initialize because both the CommonBridge
     /// and the OnChainProposer need to know the address of the other. This solves
@@ -88,4 +92,13 @@ interface IOnChainProposer {
         bytes32 alignedProgramVKey,
         bytes32[] calldata alignedMerkleProof
     ) external;
+
+    /// @notice Allows unverified batches to be reverted
+    function revertBatch(uint256 batchNumber) external;
+
+    /// @notice Allows the owner to pause the contract
+    function pause() external;
+
+    /// @notice Allows the owner to unpause the contract
+    function unpause() external;
 }

crates/l2/sdk/src/sdk.rs

@@ -557,3 +557,22 @@ pub async fn initialize_contract(
 
     Ok(initialize_tx_hash)
 }
+
+pub async fn call_contract(
+    client: &EthClient,
+    private_key: &SecretKey,
+    to: Address,
+    signature: &str,
+    parameters: Vec<Value>,
+) -> Result<H256, EthClientError> {
+    let calldata = encode_calldata(signature, &parameters)?.into();
+    let from = get_address_from_secret_key(private_key)?;
+    let tx = client
+        .build_eip1559_transaction(to, from, calldata, Default::default())
+        .await?;
+
+    let tx_hash = client.send_eip1559_transaction(&tx, private_key).await?;
+
+    wait_for_transaction_receipt(tx_hash, client, 100).await?;
+    Ok(tx_hash)
+}

crates/l2/storage/src/api.rs

@@ -99,4 +99,6 @@ pub trait StoreEngineRollup: Debug + Send + Sync + RefUnwindSafe {
     async fn get_lastest_sent_batch_proof(&self) -> Result<u64, StoreError>;
 
     async fn set_lastest_sent_batch_proof(&self, batch_number: u64) -> Result<(), StoreError>;
+
+    async fn revert_to_batch(&self, batch_number: u64) -> Result<(), StoreError>;
 }

crates/l2/storage/src/store.rs

@@ -288,4 +288,9 @@ impl Store {
     pub async fn set_lastest_sent_batch_proof(&self, batch_number: u64) -> Result<(), StoreError> {
         self.engine.set_lastest_sent_batch_proof(batch_number).await
     }
+
+    /// Reverts to a previous batch, discarding operations in them
+    pub async fn revert_to_batch(&self, batch_number: u64) -> Result<(), StoreError> {
+        self.engine.revert_to_batch(batch_number).await
+    }
 }

crates/l2/storage/src/store_db/in_memory.rs

@@ -199,6 +199,25 @@ impl StoreEngineRollup for Store {
         self.inner()?.lastest_sent_batch_proof = batch_number;
         Ok(())
     }
+
+    async fn revert_to_batch(&self, batch_number: u64) -> Result<(), StoreError> {
+        let mut store = self.inner()?;
+        store
+            .batches_by_block
+            .retain(|_, batch| *batch <= batch_number);
+        store
+            .withdrawal_hashes_by_batch
+            .retain(|batch, _| *batch <= batch_number);
+        store
+            .block_numbers_by_batch
+            .retain(|batch, _| *batch <= batch_number);
+        store
+            .deposit_logs_hashes
+            .retain(|batch, _| *batch <= batch_number);
+        store.state_roots.retain(|batch, _| *batch <= batch_number);
+        store.blobs.retain(|batch, _| *batch <= batch_number);
+        Ok(())
+    }
 }
 
 impl Debug for Store {

crates/l2/storage/src/store_db/libmdbx.rs

@@ -11,8 +11,8 @@ use ethrex_common::{
 use ethrex_rlp::encode::RLPEncode;
 use ethrex_storage::error::StoreError;
 use libmdbx::{
-    DatabaseOptions, Mode, PageSize, ReadWriteOptions,
-    orm::{Database, Table},
+    DatabaseOptions, Mode, PageSize, RW, ReadWriteOptions,
+    orm::{Database, Table, Transaction},
     table, table_info,
 };
 
@@ -271,6 +271,38 @@ impl StoreEngineRollup for Store {
     async fn set_lastest_sent_batch_proof(&self, batch_number: u64) -> Result<(), StoreError> {
         self.write::<LastSentBatchProof>(0, batch_number).await
     }
+
+    async fn revert_to_batch(&self, batch_number: u64) -> Result<(), StoreError> {
+        let Some(kept_blocks) = self.get_block_numbers_by_batch(batch_number).await? else {
+            return Ok(());
+        };
+        let last_kept_block = *kept_blocks.iter().max().unwrap_or(&0);
+        let txn = self
+            .db
+            .begin_readwrite()
+            .map_err(StoreError::LibmdbxError)?;
+        delete_starting_at::<BatchesByBlockNumber>(&txn, last_kept_block + 1)?;
+        delete_starting_at::<WithdrawalHashesByBatch>(&txn, batch_number + 1)?;
+        delete_starting_at::<BlockNumbersByBatch>(&txn, batch_number + 1)?;
+        delete_starting_at::<DepositLogsHash>(&txn, batch_number + 1)?;
+        delete_starting_at::<StateRoots>(&txn, batch_number + 1)?;
+        delete_starting_at::<BlobsBundles>(&txn, batch_number + 1)?;
+        txn.commit().map_err(StoreError::LibmdbxError)?;
+        Ok(())
+    }
+}
+
+/// Deletes keys above key, assuming they are contiguous
+fn delete_starting_at<T: Table<Key = u64>>(
+    txn: &Transaction<RW>,
+    mut key: u64,
+) -> Result<(), StoreError> {
+    while let Some(val) = txn.get::<T>(key).map_err(StoreError::LibmdbxError)? {
+        txn.delete::<T>(key, Some(val))
+            .map_err(StoreError::LibmdbxError)?;
+        key += 1;
+    }
+    Ok(())
 }
 
 table!(