feat: improve ledger metrics gathering + updated ledger size management

magicblock-api/src/errors.rs

@@ -26,6 +26,9 @@ pub enum ApiError {
     #[error("Ledger error: {0}")]
     LedgerError(#[from] magicblock_ledger::errors::LedgerError),
 
+    #[error("LedgerSizeManager error: {0}")]
+    LedgerSizeManagerError(#[from] magicblock_ledger::ledger_size_manager::errors::LedgerSizeManagerError),
+
     #[error("Failed to load programs into bank: {0}")]
     FailedToLoadProgramsIntoBank(String),
 

magicblock-api/src/magic_validator.rs

@@ -41,7 +41,13 @@ use magicblock_config::{EphemeralConfig, LifecycleMode, ProgramConfig};
 use magicblock_geyser_plugin::rpc::GeyserRpcService;
 use magicblock_ledger::{
     blockstore_processor::process_ledger,
-    ledger_truncator::{LedgerTruncator, DEFAULT_TRUNCATION_TIME_INTERVAL},
+    ledger_size_manager::{
+        config::{
+            ExistingLedgerState, LedgerSizeManagerConfig, ResizePercentage,
+            CHECK_LEDGER_SIZE_INTERVAL_MS,
+        },
+        TruncatingLedgerSizeManager,
+    },
     Ledger,
 };
 use magicblock_metrics::MetricsService;
@@ -127,7 +133,7 @@ pub struct MagicValidator {
     token: CancellationToken,
     bank: Arc<Bank>,
     ledger: Arc<Ledger>,
-    ledger_truncator: LedgerTruncator<Bank>,
+    ledger_size_manager: TruncatingLedgerSizeManager,
     slot_ticker: Option<tokio::task::JoinHandle<()>>,
     pubsub_handle: RwLock<Option<thread::JoinHandle<()>>>,
     pubsub_close_handle: PubsubServiceCloseHandle,
@@ -193,6 +199,13 @@ impl MagicValidator {
             config.validator_config.ledger.reset,
         )?;
 
+        let existing_ledger_state = (!config.validator_config.ledger.reset)
+            .then_some(ExistingLedgerState {
+                size: ledger.storage_size()?,
+                slot: ledger.last_slot(),
+                mod_id: ledger.last_mod_id(),
+            });
+
         let exit = Arc::<AtomicBool>::default();
         // SAFETY:
         // this code will never panic as the ledger_path always appends the
@@ -212,13 +225,6 @@ impl MagicValidator {
             ledger.get_max_blockhash().map(|(slot, _)| slot)?,
         )?;
 
-        let ledger_truncator = LedgerTruncator::new(
-            ledger.clone(),
-            bank.clone(),
-            DEFAULT_TRUNCATION_TIME_INTERVAL,
-            config.validator_config.ledger.size,
-        );
-
         fund_validator_identity(&bank, &validator_pubkey);
         fund_magic_context(&bank);
         let faucet_keypair = funded_faucet(
@@ -241,6 +247,18 @@ impl MagicValidator {
                 Some(TransactionNotifier::new(geyser_manager)),
             );
 
+        let ledger_size_manager: TruncatingLedgerSizeManager =
+            TruncatingLedgerSizeManager::new_from_ledger(
+                ledger.clone(),
+                bank.clone(),
+                existing_ledger_state,
+                LedgerSizeManagerConfig {
+                    max_size: config.validator_config.ledger.size,
+                    size_check_interval_ms: CHECK_LEDGER_SIZE_INTERVAL_MS,
+                    resize_percentage: ResizePercentage::Large,
+                },
+            );
+
         let metrics_config = &config.validator_config.metrics;
         let metrics = if metrics_config.enabled {
             let metrics_service =
@@ -254,7 +272,6 @@ impl MagicValidator {
                 Duration::from_secs(
                     metrics_config.system_metrics_tick_interval_secs,
                 ),
-                &ledger,
                 &bank,
                 token.clone(),
             );
@@ -370,7 +387,7 @@ impl MagicValidator {
             token,
             bank,
             ledger,
-            ledger_truncator,
+            ledger_size_manager,
             accounts_manager,
             transaction_listener,
             transaction_status_sender,
@@ -658,7 +675,7 @@ impl MagicValidator {
         self.start_remote_account_updates_worker();
         self.start_remote_account_cloner_worker().await?;
 
-        self.ledger_truncator.start();
+        self.ledger_size_manager.try_start()?;
 
         self.rpc_service.start().map_err(|err| {
             ApiError::FailedToStartJsonRpcService(format!("{:?}", err))
@@ -759,7 +776,7 @@ impl MagicValidator {
         self.rpc_service.close();
         PubsubService::close(&self.pubsub_close_handle);
         self.token.cancel();
-        self.ledger_truncator.stop();
+        self.ledger_size_manager.stop();
 
         // wait a bit for services to stop
         thread::sleep(Duration::from_secs(1));

magicblock-api/src/tickers.rs

@@ -113,46 +113,9 @@ pub fn init_commit_accounts_ticker(
 
 pub fn init_system_metrics_ticker(
     tick_duration: Duration,
-    ledger: &Arc<Ledger>,
     bank: &Arc<Bank>,
     token: CancellationToken,
 ) -> tokio::task::JoinHandle<()> {
-    fn try_set_ledger_counts(ledger: &Ledger) {
-        macro_rules! try_set_ledger_count {
-            ($name:ident) => {
-                paste::paste! {
-                    match ledger.[< count_ $name >]() {
-                        Ok(count) => {
-                            metrics::[< set_ledger_ $name _count >](count);
-                        }
-                        Err(err) => warn!(
-                            "Failed to get ledger {} count: {:?}",
-                            stringify!($name),
-                            err
-                        ),
-                    }
-                }
-            };
-        }
-        try_set_ledger_count!(block_times);
-        try_set_ledger_count!(blockhashes);
-        try_set_ledger_count!(slot_signatures);
-        try_set_ledger_count!(address_signatures);
-        try_set_ledger_count!(transaction_status);
-        try_set_ledger_count!(transaction_successful_status);
-        try_set_ledger_count!(transaction_failed_status);
-        try_set_ledger_count!(transactions);
-        try_set_ledger_count!(transaction_memos);
-        try_set_ledger_count!(perf_samples);
-        try_set_ledger_count!(account_mod_data);
-    }
-
-    fn try_set_ledger_storage_size(ledger: &Ledger) {
-        match ledger.storage_size() {
-            Ok(byte_size) => metrics::set_ledger_size(byte_size),
-            Err(err) => warn!("Failed to get ledger storage size: {:?}", err),
-        }
-    }
     fn set_accounts_storage_size(bank: &Bank) {
         let byte_size = bank.accounts_db_storage_size();
         metrics::set_accounts_size(byte_size);
@@ -161,15 +124,12 @@ pub fn init_system_metrics_ticker(
         metrics::set_accounts_count(bank.accounts_db.get_accounts_count());
     }
 
-    let ledger = ledger.clone();
     let bank = bank.clone();
     tokio::task::spawn(async move {
         loop {
             tokio::select! {
                 _ = tokio::time::sleep(tick_duration) => {
-                    try_set_ledger_storage_size(&ledger);
                     set_accounts_storage_size(&bank);
-                    try_set_ledger_counts(&ledger);
                     set_accounts_count(&bank);
                 },
                 _ = token.cancelled() => {

magicblock-ledger/Cargo.toml

@@ -8,6 +8,7 @@ license.workspace = true
 edition.workspace = true
 
 [dependencies]
+async-trait = { workspace = true }
 bincode = { workspace = true }
 log = { workspace = true }
 byteorder = { workspace = true }
@@ -20,6 +21,7 @@ serde = { workspace = true }
 magicblock-bank = { workspace = true }
 magicblock-accounts-db = { workspace = true }
 magicblock-core = { workspace = true }
+magicblock-metrics = { workspace = true }
 solana-account-decoder = { workspace = true }
 solana-measure = { workspace = true }
 solana-metrics = { workspace = true }

magicblock-ledger/src/database/cf_descriptors.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashSet, path::Path};
+use std::{
+    collections::HashSet,
+    path::Path,
+    sync::{atomic::AtomicU64, Arc},
+};
 
 use log::*;
 use rocksdb::{ColumnFamilyDescriptor, DBCompressionType, Options, DB};
@@ -9,7 +13,9 @@ use super::{
     options::{LedgerColumnOptions, LedgerOptions},
     rocksdb_options::should_disable_auto_compactions,
 };
-use crate::database::{columns, options::AccessType};
+use crate::database::{
+    columns, compaction_filter::PurgedSlotFilterFactory, options::AccessType,
+};
 
 /// Create the column family (CF) descriptors necessary to open the database.
 ///
@@ -23,19 +29,20 @@ use crate::database::{columns, options::AccessType};
 pub fn cf_descriptors(
     path: &Path,
     options: &LedgerOptions,
+    oldest_slot: &Arc<AtomicU64>,
 ) -> Vec<ColumnFamilyDescriptor> {
     use columns::*;
 
     let mut cf_descriptors = vec![
-        new_cf_descriptor::<TransactionStatus>(options),
-        new_cf_descriptor::<AddressSignatures>(options),
-        new_cf_descriptor::<SlotSignatures>(options),
-        new_cf_descriptor::<Blocktime>(options),
-        new_cf_descriptor::<Blockhash>(options),
-        new_cf_descriptor::<Transaction>(options),
-        new_cf_descriptor::<TransactionMemos>(options),
-        new_cf_descriptor::<PerfSamples>(options),
-        new_cf_descriptor::<AccountModDatas>(options),
+        new_cf_descriptor::<TransactionStatus>(options, oldest_slot),
+        new_cf_descriptor::<AddressSignatures>(options, oldest_slot),
+        new_cf_descriptor::<SlotSignatures>(options, oldest_slot),
+        new_cf_descriptor::<Blocktime>(options, oldest_slot),
+        new_cf_descriptor::<Blockhash>(options, oldest_slot),
+        new_cf_descriptor::<Transaction>(options, oldest_slot),
+        new_cf_descriptor::<TransactionMemos>(options, oldest_slot),
+        new_cf_descriptor::<PerfSamples>(options, oldest_slot),
+        new_cf_descriptor::<AccountModDatas>(options, oldest_slot),
     ];
 
     // If the access type is Secondary, we don't need to open all of the
@@ -87,13 +94,18 @@ pub fn cf_descriptors(
 
 fn new_cf_descriptor<C: 'static + Column + ColumnName>(
     options: &LedgerOptions,
+    oldest_slot: &Arc<AtomicU64>,
 ) -> ColumnFamilyDescriptor {
-    ColumnFamilyDescriptor::new(C::NAME, get_cf_options::<C>(options))
+    ColumnFamilyDescriptor::new(
+        C::NAME,
+        get_cf_options::<C>(options, oldest_slot),
+    )
 }
 
 // FROM ledger/src/blockstore_db.rs :2010
 fn get_cf_options<C: 'static + Column + ColumnName>(
     options: &LedgerOptions,
+    oldest_slot: &Arc<AtomicU64>,
 ) -> Options {
     let mut cf_options = Options::default();
     // 256 * 8 = 2GB. 6 of these columns should take at most 12GB of RAM
@@ -111,7 +123,12 @@ fn get_cf_options<C: 'static + Column + ColumnName>(
     );
     cf_options.set_max_bytes_for_level_base(total_size_base);
     cf_options.set_target_file_size_base(file_size_base);
+    cf_options.set_compaction_filter_factory(
+        PurgedSlotFilterFactory::<C>::new(oldest_slot.clone()),
+    );
 
+    // TODO(edwin): check if needed
+    // cf_options.set_max_total_wal_size(4 * 1024 * 1024 * 1024);
     let disable_auto_compactions =
         should_disable_auto_compactions(&options.access_type);
     if disable_auto_compactions {

magicblock-ledger/src/database/columns.rs

@@ -129,6 +129,9 @@ pub trait Column {
     // first item in the key.
     fn as_index(slot: Slot) -> Self::Index;
     fn slot(index: Self::Index) -> Slot;
+    fn keep_all_on_compaction() -> bool {
+        false
+    }
 }
 
 pub trait ColumnName {
@@ -651,6 +654,9 @@ impl Column for AccountModDatas {
     fn as_index(slot: Slot) -> Self::Index {
         slot
     }
+    fn keep_all_on_compaction() -> bool {
+        true
+    }
 }
 
 impl TypedColumn for AccountModDatas {
@@ -665,8 +671,3 @@ impl TypedColumn for AccountModDatas {
 pub fn should_enable_compression<C: 'static + Column + ColumnName>() -> bool {
     C::NAME == TransactionStatus::NAME
 }
-
-// -----------------
-// Column Queries
-// -----------------
-pub(crate) const DIRTY_COUNT: i64 = -1;