Prune ledger DB tables

bin/citrea/src/main.rs

@@ -15,6 +15,7 @@ use citrea_common::{from_toml_path, FromEnv, FullNodeConfig};
 use citrea_light_client_prover::da_block_handler::StartVariant;
 use citrea_stf::genesis_config::GenesisPaths;
 use citrea_stf::runtime::{CitreaRuntime, DefaultContext};
+use citrea_storage_ops::pruning::types::PruningNodeType;
 use clap::Parser;
 use metrics_exporter_prometheus::PrometheusBuilder;
 use metrics_util::MetricKindMask;
@@ -370,7 +371,9 @@ where
             // Spawn pruner if configs are set
             if let Some(pruner_service) = pruner_service {
                 task_manager.spawn(|cancellation_token| async move {
-                    pruner_service.run(cancellation_token).await
+                    pruner_service
+                        .run(PruningNodeType::FullNode, cancellation_token)
+                        .await
                 });
             }
 

bin/citrea/src/rollup/mod.rs

@@ -211,7 +211,7 @@ pub trait CitreaRollupBlueprint: RollupBlueprint {
     ) -> Result<(
         CitreaFullnode<Self::DaService, LedgerDB>,
         FullNodeL1BlockHandler<Self::Vm, Self::DaService, LedgerDB>,
-        Option<PrunerService<LedgerDB>>,
+        Option<PrunerService>,
     )> {
         let runner_config = rollup_config.runner.expect("Runner config is missing");
 

bin/citrea/tests/common/helpers.rs

@@ -17,6 +17,7 @@ use citrea_common::{
 use citrea_light_client_prover::da_block_handler::StartVariant;
 use citrea_primitives::TEST_PRIVATE_KEY;
 use citrea_stf::genesis_config::GenesisPaths;
+use citrea_storage_ops::pruning::types::PruningNodeType;
 use citrea_storage_ops::pruning::PruningConfig;
 use sov_db::ledger_db::SharedLedgerOps;
 use sov_db::rocks_db_config::RocksdbConfig;
@@ -336,8 +337,11 @@ pub async fn start_rollup(
 
         // Spawn pruner if configs are set
         if let Some(pruner) = pruner {
-            task_manager
-                .spawn(|cancellation_token| async move { pruner.run(cancellation_token).await });
+            task_manager.spawn(|cancellation_token| async move {
+                pruner
+                    .run(PruningNodeType::FullNode, cancellation_token)
+                    .await
+            });
         }
 
         task_manager.spawn(|cancellation_token| async move {

bin/cli/src/commands/prune.rs

@@ -1,23 +1,54 @@
 use std::path::PathBuf;
 use std::sync::Arc;
 
+use citrea_storage_ops::pruning::types::PruningNodeType;
 use citrea_storage_ops::pruning::{Pruner, PruningConfig};
+use clap::ValueEnum;
 use sov_db::ledger_db::{LedgerDB, SharedLedgerOps};
 use sov_db::native_db::NativeDB;
 use sov_db::rocks_db_config::RocksdbConfig;
+use sov_db::schema::tables::{
+    BATCH_PROVER_LEDGER_TABLES, FULL_NODE_LEDGER_TABLES, LIGHT_CLIENT_PROVER_LEDGER_TABLES,
+    SEQUENCER_LEDGER_TABLES,
+};
 use sov_db::state_db::StateDB;
 use sov_prover_storage_manager::SnapshotManager;
 use tracing::{debug, info};
 
-pub(crate) async fn prune(db_path: PathBuf, distance: u64) -> anyhow::Result<()> {
+#[derive(Copy, Clone, ValueEnum)]
+pub enum PruningNodeTypeArg {
+    Sequencer,
+    FullNode,
+    BatchProver,
+    LightClient,
+}
+
+impl From<PruningNodeTypeArg> for PruningNodeType {
+    fn from(value: PruningNodeTypeArg) -> Self {
+        match value {
+            PruningNodeTypeArg::Sequencer => PruningNodeType::Sequencer,
+            PruningNodeTypeArg::FullNode => PruningNodeType::FullNode,
+            PruningNodeTypeArg::BatchProver => PruningNodeType::BatchProver,
+            PruningNodeTypeArg::LightClient => PruningNodeType::LightClient,
+        }
+    }
+}
+
+pub(crate) async fn prune(
+    node_type: PruningNodeTypeArg,
+    db_path: PathBuf,
+    distance: u64,
+) -> anyhow::Result<()> {
     info!(
         "Pruning DB at {} with pruning distance of {}",
         db_path.display(),
         distance
     );
     let config = PruningConfig { distance };
 
-    let rocksdb_config = RocksdbConfig::new(&db_path, None, None);
+    let column_families = cfs_from_node_type(node_type);
+
+    let rocksdb_config = RocksdbConfig::new(&db_path, None, Some(column_families.to_vec()));
     let ledger_db = LedgerDB::with_config(&rocksdb_config)?;
     let native_db = NativeDB::<SnapshotManager>::setup_schema_db(&rocksdb_config)?;
     let state_db = StateDB::<SnapshotManager>::setup_schema_db(&rocksdb_config)?;
@@ -31,8 +62,26 @@ pub(crate) async fn prune(db_path: PathBuf, distance: u64) -> anyhow::Result<()>
         soft_confirmation_number, distance
     );
 
-    let pruner = Pruner::new(config, ledger_db, Arc::new(state_db), Arc::new(native_db));
-    pruner.prune(soft_confirmation_number).await;
+    let pruner = Pruner::new(
+        config,
+        ledger_db.inner(),
+        Arc::new(state_db),
+        Arc::new(native_db),
+    );
+    pruner
+        .prune(node_type.into(), soft_confirmation_number)
+        .await;
 
     Ok(())
 }
+
+fn cfs_from_node_type(node_type: PruningNodeTypeArg) -> Vec<String> {
+    let cfs = match node_type {
+        PruningNodeTypeArg::Sequencer => SEQUENCER_LEDGER_TABLES,
+        PruningNodeTypeArg::FullNode => FULL_NODE_LEDGER_TABLES,
+        PruningNodeTypeArg::BatchProver => BATCH_PROVER_LEDGER_TABLES,
+        PruningNodeTypeArg::LightClient => LIGHT_CLIENT_PROVER_LEDGER_TABLES,
+    };
+
+    cfs.iter().map(|x| x.to_string()).collect::<Vec<_>>()
+}

bin/cli/src/main.rs

@@ -1,6 +1,7 @@
 use std::path::PathBuf;
 
 use clap::{Parser, Subcommand, ValueEnum};
+use commands::PruningNodeTypeArg;
 use tracing_subscriber::fmt;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::util::SubscriberInitExt;
@@ -39,6 +40,8 @@ struct Cli {
 enum Commands {
     /// Prune old DB entries
     Prune {
+        #[arg(long)]
+        node_type: PruningNodeTypeArg,
         /// The path of the database to prune
         #[arg(long)]
         db_path: PathBuf,
@@ -78,8 +81,12 @@ async fn main() -> anyhow::Result<()> {
     // You can check for the existence of subcommands, and if found use their
     // matches just as you would the top level cmd
     match cli.command {
-        Commands::Prune { db_path, distance } => {
-            commands::prune(db_path.clone(), distance).await?;
+        Commands::Prune {
+            node_type,
+            db_path,
+            distance,
+        } => {
+            commands::prune(node_type, db_path.clone(), distance).await?;
         }
         Commands::Rollback {
             db_path: _db_path,

crates/fullnode/src/lib.rs

@@ -44,7 +44,7 @@ pub fn build_services<Da, DB, Vm>(
 ) -> Result<(
     CitreaFullnode<Da, DB>,
     L1BlockHandler<Vm, Da, DB>,
-    Option<PrunerService<DB>>,
+    Option<PrunerService>,
 )>
 where
     Da: DaService<Error = anyhow::Error>,
@@ -53,9 +53,9 @@ where
 {
     let last_pruned_block = ledger_db.get_last_pruned_l2_height()?.unwrap_or(0);
     let pruner = runner_config.pruning_config.as_ref().map(|pruning_config| {
-        let pruner = Pruner::<DB>::new(
+        let pruner = Pruner::new(
             pruning_config.clone(),
-            ledger_db.clone(),
+            ledger_db.inner(),
             storage_manager.get_state_db_handle(),
             storage_manager.get_native_db_handle(),
         );

crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/mod.rs

@@ -152,6 +152,11 @@ impl SharedLedgerOps for LedgerDB {
         self.db.path()
     }
 
+    /// Returns the inner DB instance
+    fn inner(&self) -> Arc<DB> {
+        self.db.clone()
+    }
+
     #[instrument(level = "trace", skip(self, schema_batch), err, ret)]
     fn put_soft_confirmation(
         &self,

crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/traits.rs

@@ -1,4 +1,5 @@
 use std::path::Path;
+use std::sync::Arc;
 
 use anyhow::Result;
 use serde::de::DeserializeOwned;
@@ -20,6 +21,9 @@ pub trait SharedLedgerOps {
     /// Return DB path
     fn path(&self) -> &Path;
 
+    /// Returns the inner DB instance
+    fn inner(&self) -> Arc<sov_schema_db::DB>;
+
     /// Put soft confirmation to db
     fn put_soft_confirmation(
         &self,

crates/sovereign-sdk/full-node/db/sov-db/src/schema/tables.rs

@@ -48,7 +48,6 @@ pub const STATE_TABLES: &[&str] = &[
 /// A list of all tables used by Sequencer LedgerDB
 pub const SEQUENCER_LEDGER_TABLES: &[&str] = &[
     ExecutedMigrations::table_name(),
-    SlotByHash::table_name(),
     SoftConfirmationByNumber::table_name(),
     SoftConfirmationByHash::table_name(),
     L2RangeByL1Height::table_name(),
@@ -58,10 +57,17 @@ pub const SEQUENCER_LEDGER_TABLES: &[&str] = &[
     LastSequencerCommitmentSent::table_name(),
     SoftConfirmationStatus::table_name(),
     CommitmentsByNumber::table_name(),
-    VerifiedBatchProofsBySlotNumber::table_name(),
-    ProverLastScannedSlot::table_name(),
     MempoolTxs::table_name(),
     LastPrunedBlock::table_name(),
+    // ########
+    // The following tables exist in the sequencer since they enable
+    // using the fullnode's backup as a sequencer database without having
+    // to remove these tables first as demonstrated by the
+    // `test_sequencer_crash_and_replace_full_node` test.
+    VerifiedBatchProofsBySlotNumber::table_name(),
+    ProverLastScannedSlot::table_name(),
+    SlotByHash::table_name(),
+    // ########
     #[cfg(test)]
     TestTableOld::table_name(),
     #[cfg(test)]
@@ -77,13 +83,10 @@ pub const FULL_NODE_LEDGER_TABLES: &[&str] = &[
     SoftConfirmationByHash::table_name(),
     L2RangeByL1Height::table_name(),
     L2GenesisStateRoot::table_name(),
-    LastStateDiff::table_name(),
-    PendingSequencerCommitmentL2Range::table_name(),
     LastSequencerCommitmentSent::table_name(),
     SoftConfirmationStatus::table_name(),
     ProverLastScannedSlot::table_name(),
     CommitmentsByNumber::table_name(),
-    MempoolTxs::table_name(),
     LastPrunedBlock::table_name(),
     VerifiedBatchProofsBySlotNumber::table_name(),
     #[cfg(test)]
@@ -119,9 +122,9 @@ pub const BATCH_PROVER_LEDGER_TABLES: &[&str] = &[
 pub const LIGHT_CLIENT_PROVER_LEDGER_TABLES: &[&str] = &[
     ExecutedMigrations::table_name(),
     SlotByHash::table_name(),
+    SoftConfirmationByNumber::table_name(),
     LightClientProofBySlotNumber::table_name(),
     ProverLastScannedSlot::table_name(),
-    SoftConfirmationByNumber::table_name(),
     #[cfg(test)]
     TestTableOld::table_name(),
     #[cfg(test)]

crates/storage-ops/Cargo.toml

@@ -27,4 +27,5 @@ tracing = { workspace = true }
 
 [dev-dependencies]
 sov-prover-storage-manager = { path = "../sovereign-sdk/full-node/sov-prover-storage-manager" }
+sov-rollup-interface = { path = "../../crates/sovereign-sdk/rollup-interface", features = ["testing"] }
 tempfile = { workspace = true }

crates/storage-ops/src/pruning/components/ledger_db/mod.rs

@@ -0,0 +1,64 @@
+use std::sync::Arc;
+
+use sov_schema_db::DB;
+use tracing::{debug, error};
+
+use self::slots::prune_slots;
+use self::soft_confirmations::prune_soft_confirmations;
+use crate::pruning::types::PruningNodeType;
+
+mod slots;
+mod soft_confirmations;
+
+macro_rules! log_result_or_error {
+    ($tables_group:literal, $call:expr) => {{
+        match $call {
+            Ok(result) => {
+                debug!("Deleted {} records from {} group", result, $tables_group);
+            }
+            Err(e) => {
+                error!(
+                    "Failed to prune ledger's {} table group: {:?}",
+                    $tables_group, e
+                );
+                return;
+            }
+        }
+    }};
+}
+
+/// Prune ledger
+pub(crate) fn prune_ledger(node_type: PruningNodeType, ledger_db: Arc<DB>, up_to_block: u64) {
+    debug!("Pruning Ledger, up to L2 block {}", up_to_block);
+
+    match node_type {
+        PruningNodeType::Sequencer => {
+            log_result_or_error!(
+                "soft_confirmations",
+                prune_soft_confirmations(node_type, &ledger_db, up_to_block)
+            );
+            log_result_or_error!("slots", prune_slots(node_type, &ledger_db, up_to_block));
+        }
+        PruningNodeType::FullNode => {
+            log_result_or_error!(
+                "soft_confirmations",
+                prune_soft_confirmations(node_type, &ledger_db, up_to_block)
+            );
+            log_result_or_error!("slots", prune_slots(node_type, &ledger_db, up_to_block));
+        }
+        PruningNodeType::BatchProver => {
+            log_result_or_error!(
+                "soft_confirmations",
+                prune_soft_confirmations(node_type, &ledger_db, up_to_block)
+            );
+            log_result_or_error!("slots", prune_slots(node_type, &ledger_db, up_to_block));
+        }
+        PruningNodeType::LightClient => {
+            log_result_or_error!(
+                "soft_confirmations",
+                prune_soft_confirmations(node_type, &ledger_db, up_to_block)
+            );
+            log_result_or_error!("slots", prune_slots(node_type, &ledger_db, up_to_block));
+        }
+    }
+}