Compaction V1: Manual compaction (#383)

This implements manual compaction strategy. 

Since `delete` in rocksdb doesn't cleanup space, but rather adds more
space(tombstones), we trigger manual compaction on required ranges.

Ranges are estimated as follows:
1. take size of dbs, take num of slots in db
2. estimate average size per slot
3. calculate how much slots we need to delete to clean 10%
4. deduce range from available range to
truncate(LedgerTruncator::available_truncation_range) & num of slots
from 3.
5. delete & compact

see `LedgerTruncator::estimate_truncation_range` for implementation

<!-- greptile_comment -->

## Greptile Summary

Implements manual compaction strategy for RocksDB to manage tombstones
left by deletions, with size-based truncation and concurrent compaction
across column families.

- Added `compact_range_cf` methods to `Database`, `Rocks`, and
`LedgerColumn` structs to support manual compaction with optional range
parameters
- Implemented size-based truncation in `LedgerTruncator` that estimates
slot ranges to clean ~10% of ledger space
- Added concurrent compaction for different column families using
`JoinSet` with proper memtable flushing
- Renamed `truncate_slots` to `delete_slot_range` for clarity and
improved error handling
- Added comprehensive tests including block writing, finality boundaries
and 1GB database scenarios



<!-- /greptile_comment -->

Author: taco-paco

magicblock-api/src/magic_validator.rs

@@ -757,7 +757,9 @@ impl MagicValidator {
 
         // we have two memory mapped databases, flush them to disk before exitting
         self.bank.flush();
-        self.ledger.flush();
+        if let Err(err) = self.ledger.shutdown(false) {
+            error!("Failed to shutdown ledger: {:?}", err);
+        }
     }
 
     pub fn join(self) {

magicblock-ledger/src/database/db.rs

@@ -25,7 +25,7 @@ impl Database {
     pub fn open(
         path: &Path,
         options: LedgerOptions,
-    ) -> std::result::Result<Self, LedgerError> {
+    ) -> Result<Self, LedgerError> {
         let column_options = Arc::new(options.column_options.clone());
         let backend = Arc::new(Rocks::open(path, options)?);
 
@@ -36,16 +36,13 @@ impl Database {
         })
     }
 
-    pub fn destroy(path: &Path) -> std::result::Result<(), LedgerError> {
+    pub fn destroy(path: &Path) -> Result<(), LedgerError> {
         Rocks::destroy(path)?;
 
         Ok(())
     }
 
-    pub fn get<C>(
-        &self,
-        key: C::Index,
-    ) -> std::result::Result<Option<C::Type>, LedgerError>
+    pub fn get<C>(&self, key: C::Index) -> Result<Option<C::Type>, LedgerError>
     where
         C: TypedColumn + ColumnName,
     {
@@ -63,10 +60,7 @@ impl Database {
     pub fn iter<C>(
         &self,
         iterator_mode: IteratorMode<C::Index>,
-    ) -> std::result::Result<
-        impl Iterator<Item = (C::Index, Box<[u8]>)> + '_,
-        LedgerError,
-    >
+    ) -> Result<impl Iterator<Item = (C::Index, Box<[u8]>)> + '_, LedgerError>
     where
         C: Column + ColumnName,
     {
@@ -160,6 +154,21 @@ impl Database {
         )
     }
 
+    /// See [crate::database::rocks_db::Rocks::compact_range_cf] for documentation.
+    pub fn compact_range_cf<C>(
+        &self,
+        from: Option<C::Index>,
+        to: Option<C::Index>,
+    ) where
+        C: Column + ColumnName,
+    {
+        self.backend.compact_range_cf(
+            self.cf_handle::<C>(),
+            from.map(|index| C::key(index)),
+            to.map(|index| C::key(index)),
+        )
+    }
+
     pub fn is_primary_access(&self) -> bool {
         self.backend.is_primary_access()
     }

magicblock-ledger/src/database/ledger_column.rs

@@ -246,6 +246,20 @@ where
     ) {
         write_batch.delete_range_cf::<C>(self.handle(), from, to);
     }
+
+    /// See [crate::database::rocks_db::Rocks::compact_range_cf] for documentation.
+    pub fn compact_range(&self, from: Option<C::Index>, to: Option<C::Index>) {
+        self.backend.compact_range_cf(
+            self.handle(),
+            from.map(|index| C::key(index)),
+            to.map(|index| C::key(index)),
+        )
+    }
+
+    /// See [crate::database::rocks_db::Rocks::flush_cf] for documentation.
+    pub fn flush(&self) -> LedgerResult<()> {
+        self.backend.flush_cf(self.handle())
+    }
 }
 
 impl<C> LedgerColumn<C>

magicblock-ledger/src/database/rocks_db.rs

@@ -1,9 +1,9 @@
 use std::{fs, path::Path};
 
 use rocksdb::{
-    ColumnFamily, DBIterator, DBPinnableSlice, DBRawIterator,
-    IteratorMode as RocksIteratorMode, LiveFile, Options,
-    WriteBatch as RWriteBatch, DB,
+    AsColumnFamilyRef, ColumnFamily, DBIterator, DBPinnableSlice,
+    DBRawIterator, FlushOptions, IteratorMode as RocksIteratorMode, LiveFile,
+    Options, WriteBatch as RWriteBatch, DB,
 };
 
 use super::{
@@ -115,6 +115,33 @@ impl Rocks {
         Ok(())
     }
 
+    /// Compacts keys in range \[`from`, `to`\].
+    /// For leveled compaction style, all files containing keys in the given range
+    /// are compacted to the last level containing files.
+    /// https://github.com/facebook/rocksdb/wiki/Manual-Compaction#compactrange
+    pub fn compact_range_cf<S: AsRef<[u8]>, E: AsRef<[u8]>>(
+        &self,
+        cf: &ColumnFamily,
+        from_key: Option<S>,
+        to_key: Option<E>,
+    ) {
+        self.db.compact_range_cf(cf, from_key, to_key)
+    }
+
+    /// Flushes column family
+    pub fn flush_cf(&self, cf: &ColumnFamily) -> LedgerResult<()> {
+        Ok(self.db.flush_cf(cf)?)
+    }
+
+    /// Flushed column families
+    pub fn flush_cfs_opt(
+        &self,
+        cfs: &[&impl AsColumnFamilyRef],
+        options: &FlushOptions,
+    ) -> LedgerResult<()> {
+        Ok(self.db.flush_cfs_opt(cfs, options)?)
+    }
+
     pub fn iterator_cf<C>(
         &self,
         cf: &ColumnFamily,

magicblock-ledger/src/ledger_truncator.rs

@@ -1,17 +1,25 @@
-use std::{cmp::min, ops::ControlFlow, sync::Arc, time::Duration};
+use std::{cmp::min, sync::Arc, time::Duration};
 
-use log::{error, warn};
+use log::{error, info, warn};
 use magicblock_core::traits::FinalityProvider;
 use tokio::{
-    task::{JoinError, JoinHandle},
+    task::{JoinError, JoinHandle, JoinSet},
     time::interval,
 };
 use tokio_util::sync::CancellationToken;
 
-use crate::{errors::LedgerResult, Ledger};
+use crate::{
+    database::columns::{
+        AddressSignatures, Blockhash, Blocktime, PerfSamples, SlotSignatures,
+        Transaction, TransactionMemos, TransactionStatus,
+    },
+    errors::LedgerResult,
+    Ledger,
+};
 
 pub const DEFAULT_TRUNCATION_TIME_INTERVAL: Duration =
-    Duration::from_secs(10 * 60);
+    Duration::from_secs(2 * 60);
+const PERCENTAGE_TO_TRUNCATE: u8 = 10;
 
 struct LedgerTrunctationWorker<T> {
     finality_provider: Arc<T>,
@@ -46,34 +54,62 @@ impl<T: FinalityProvider> LedgerTrunctationWorker<T> {
                     return;
                 }
                 _ = interval.tick() => {
-                    const TRUNCATE_TO_PERCENTAGE: u64 = 90;
-
-                    match self.should_truncate() {
-                        Ok(true) => {
-                            if let Some((from_slot, to_slot)) = self.next_truncation_range() {
-                                let to_size = ( self.ledger_size / 100 ) * TRUNCATE_TO_PERCENTAGE;
-                                Self::truncate_to_size(&self.ledger, to_size, from_slot, to_slot);
-                            } else {
-                                warn!("Failed to get truncation range! Ledger size exceeded desired threshold");
-                            }
-                        },
-                        Ok(false) => (),
-                        Err(err) => error!("Failed to check truncation condition: {err}"),
+                    // Note: since we clean 10%, tomstones will take around 10% as well
+                    const FILLED_PERCENTAGE_LIMIT: u8 = 100 - PERCENTAGE_TO_TRUNCATE;
+
+                    let current_size = match self.ledger.storage_size() {
+                        Ok(value) => value,
+                        Err(err) => {
+                            error!("Failed to check truncation condition: {err}");
+                            continue;
+                        }
+                    };
+
+                    // Check if we should truncate
+                    if current_size < (self.ledger_size / 100) * FILLED_PERCENTAGE_LIMIT as u64 {
+                        continue;
+                    }
+
+                    info!("Ledger size: {current_size}");
+                    match self.estimate_truncation_range(current_size) {
+                        Ok(Some((from_slot, to_slot))) => Self::truncate_slot_range(&self.ledger, from_slot, to_slot).await,
+                        Ok(None) => warn!("Could not estimate truncation range"),
+                        Err(err) => error!("Failed to estimate truncation range: {:?}", err),
                     }
                 }
             }
         }
     }
 
-    fn should_truncate(&self) -> LedgerResult<bool> {
-        // Once size percentage reached, we start truncation
-        const FILLED_PERCENTAGE_LIMIT: u64 = 98;
-        Ok(self.ledger.storage_size()?
-            >= (self.ledger_size / 100) * FILLED_PERCENTAGE_LIMIT)
+    /// Returns range to truncate [from_slot, to_slot]
+    fn estimate_truncation_range(
+        &self,
+        current_ledger_size: u64,
+    ) -> LedgerResult<Option<(u64, u64)>> {
+        let (from_slot, to_slot) =
+            if let Some(val) = self.available_truncation_range() {
+                val
+            } else {
+                return Ok(None);
+            };
+
+        let num_slots = self.ledger.count_blockhashes()?;
+        if num_slots == 0 {
+            info!("No slot were written yet. Nothing to truncate!");
+            return Ok(None);
+        }
+
+        let slot_size = current_ledger_size / num_slots as u64;
+        let size_to_truncate =
+            (current_ledger_size / 100) * PERCENTAGE_TO_TRUNCATE as u64;
+        let num_slots_to_truncate = size_to_truncate / slot_size;
+
+        let to_slot = min(from_slot + num_slots_to_truncate, to_slot);
+        Ok(Some((from_slot, to_slot)))
     }
 
     /// Returns [from_slot, to_slot] range that's safe to truncate
-    fn next_truncation_range(&self) -> Option<(u64, u64)> {
+    fn available_truncation_range(&self) -> Option<(u64, u64)> {
         let lowest_cleanup_slot = self.ledger.get_lowest_cleanup_slot();
         let latest_final_slot = self.finality_provider.get_latest_final_slot();
 
@@ -83,10 +119,16 @@ impl<T: FinalityProvider> LedgerTrunctationWorker<T> {
                 // This could not happen because of Truncator
                 warn!("Slots after latest final slot have been truncated!");
             }
+
+            info!(
+                "Lowest cleanup slot ge than latest final slot. {}, {}",
+                lowest_cleanup_slot, latest_final_slot
+            );
             return None;
         }
-        // Nothing to clean
-        if latest_final_slot - 1 == lowest_cleanup_slot {
+        // Nothing to truncate
+        if latest_final_slot == lowest_cleanup_slot + 1 {
+            info!("Nothing to truncate");
             return None;
         }
 
@@ -103,9 +145,8 @@ impl<T: FinalityProvider> LedgerTrunctationWorker<T> {
 
     /// Utility function for splitting truncation into smaller chunks
     /// Cleans slots [from_slot; to_slot] inclusive range
-    pub fn truncate_to_size(
+    pub async fn truncate_slot_range(
         ledger: &Arc<Ledger>,
-        size: u64,
         from_slot: u64,
         to_slot: u64,
     ) {
@@ -116,9 +157,13 @@ impl<T: FinalityProvider> LedgerTrunctationWorker<T> {
             warn!("LedgerTruncator: Nani?");
             return;
         }
+
+        info!(
+            "LedgerTruncator: truncating slot range [{from_slot}; {to_slot}]"
+        );
         (from_slot..=to_slot)
             .step_by(SINGLE_TRUNCATION_LIMIT)
-            .try_for_each(|cur_from_slot| {
+            .for_each(|cur_from_slot| {
                 let num_slots_to_truncate = min(
                     to_slot - cur_from_slot + 1,
                     SINGLE_TRUNCATION_LIMIT as u64,
@@ -127,30 +172,76 @@ impl<T: FinalityProvider> LedgerTrunctationWorker<T> {
                     cur_from_slot + num_slots_to_truncate - 1;
 
                 if let Err(err) =
-                    ledger.truncate_slots(cur_from_slot, truncate_to_slot)
+                    ledger.delete_slot_range(cur_from_slot, truncate_to_slot)
                 {
                     warn!(
                         "Failed to truncate slots {}-{}: {}",
                         cur_from_slot, truncate_to_slot, err
                     );
-
-                    return ControlFlow::Continue(());
-                }
-
-                match ledger.storage_size() {
-                    Ok(current_size) => {
-                        if current_size <= size {
-                            ControlFlow::Break(())
-                        } else {
-                            ControlFlow::Continue(())
-                        }
-                    }
-                    Err(err) => {
-                        warn!("Failed to fetch Ledger size: {err}");
-                        ControlFlow::Continue(())
-                    }
                 }
             });
+        // Flush memtables with tombstones prior to compaction
+        if let Err(err) = ledger.flush() {
+            error!("Failed to flush ledger: {err}");
+        }
+
+        Self::compact_slot_range(ledger, from_slot, to_slot).await;
+    }
+
+    /// Synchronous utility function that triggers and awaits compaction on all the columns
+    pub async fn compact_slot_range(
+        ledger: &Arc<Ledger>,
+        from_slot: u64,
+        to_slot: u64,
+    ) {
+        if to_slot < from_slot {
+            warn!("LedgerTruncator: Nani2?");
+            return;
+        }
+
+        // Compaction can be run concurrently for different cf
+        // but it utilizes rocksdb threads, in order not to drain
+        // our tokio rt threads, we split the effort in just 3 tasks
+        let mut join_set = JoinSet::new();
+        join_set.spawn({
+            let ledger = ledger.clone();
+            async move {
+                ledger.compact_slot_range_cf::<Blocktime>(
+                    Some(from_slot),
+                    Some(to_slot + 1),
+                );
+                ledger.compact_slot_range_cf::<Blockhash>(
+                    Some(from_slot),
+                    Some(to_slot + 1),
+                );
+                ledger.compact_slot_range_cf::<PerfSamples>(
+                    Some(from_slot),
+                    Some(to_slot + 1),
+                );
+                ledger.compact_slot_range_cf::<SlotSignatures>(
+                    Some((from_slot, u32::MIN)),
+                    Some((to_slot + 1, u32::MAX)),
+                );
+            }
+        });
+
+        // Can not compact with specific range
+        join_set.spawn({
+            let ledger = ledger.clone();
+            async move {
+                ledger.compact_slot_range_cf::<TransactionStatus>(None, None);
+                ledger.compact_slot_range_cf::<Transaction>(None, None);
+            }
+        });
+        join_set.spawn({
+            let ledger = ledger.clone();
+            async move {
+                ledger.compact_slot_range_cf::<TransactionMemos>(None, None);
+                ledger.compact_slot_range_cf::<AddressSignatures>(None, None);
+            }
+        });
+
+        let _ = join_set.join_all().await;
     }
 }