Ledger purgatory (#332)

Implementation of service that will periodically purge database. It
preserves slots after latest snapshots + some interval after , that
depends on user configuration of `desired-size` field. Estimation
function for interval is
`EphemeralConfig::estimate_purge_slot_interval`. Using that interval,
there will be a periodic cleaning of slots prior to
`latest_snapshot_slot - interval ` once deserized size reached.

Since implementation may change in the future, I added
`FinalityProvider` that for now just provides `latest_snapshot_slot `,
put may in the future be changed to provide latest final slot.

<!-- greptile_comment -->

## Greptile Summary

Implemented a ledger purging service called "Ledger Purgatory" that
periodically cleans up old database slots while preserving recent
history based on snapshot slots and configurable size limits.

- Added `LedgerPurgatory` service in
`/magicblock-ledger/src/ledger_purgatory.rs` that safely purges slots
before `latest_snapshot_slot - slots_to_preserve`
- Introduced `FinalityProvider` trait and implementation that currently
uses latest snapshot slot but is extensible for future finality
mechanisms
- Added `desired_size` configuration in `LedgerConfig` to control
purging interval based on estimated transaction sizes
- Implemented chunked purging in `store/api.rs` to avoid overwhelming
RocksDB when cleaning up large ranges of slots
- Added comprehensive tests in `tests/ledger_purgatory.rs` covering
various purging scenarios and edge cases

The implementation appears sound but would benefit from additional
documentation around the purging strategy and configuration options.



<sub>💡 (2/5) Greptile learns from your feedback when you react with
👍/👎!</sub>

<!-- /greptile_comment -->

---------

Co-authored-by: Babur Makhmudov <bmuddha13@gmail.com>
Co-authored-by: Thorsten Lorenz <thlorenz@gmx.de>

Author: 3 people

Cargo.lock

@@ -11,6 +11,9 @@ use solana_account::{
 };
 use solana_pubkey::Pubkey;
 use storage::AccountsStorage;
+
+use crate::snapshot::SnapSlot;
+
 pub type AdbResult<T> = Result<T, AccountsDbError>;
 /// Stop the World Lock, used to halt all writes to adb while
 /// some critical operation is in action, e.g. snapshotting
@@ -257,6 +260,42 @@ impl AccountsDb {
         }
     }
 
+    /// Returns slot of latest snapshot or None
+    /// Parses path to extract slot
+    pub fn get_latest_snapshot_slot(&self) -> Option<u64> {
+        self.snapshot_engine
+            .with_snapshots(|snapshots| -> Option<u64> {
+                let latest_path = snapshots.back()?;
+                SnapSlot::try_from_path(latest_path)
+                    .map(|snap_slot: SnapSlot| snap_slot.slot())
+                    .or_else(|| {
+                        error!(
+                            "Failed to parse the path into SnapSlot: {}",
+                            latest_path.display()
+                        );
+                        None
+                    })
+            })
+    }
+
+    /// Return slot of oldest maintained snapshot or None
+    /// Parses path to extract slot
+    pub fn get_oldest_snapshot_slot(&self) -> Option<u64> {
+        self.snapshot_engine
+            .with_snapshots(|snapshots| -> Option<u64> {
+                let latest_path = snapshots.front()?;
+                SnapSlot::try_from_path(latest_path)
+                    .map(|snap_slot: SnapSlot| snap_slot.slot())
+                    .or_else(|| {
+                        error!(
+                            "Failed to parse the path into SnapSlot: {}",
+                            latest_path.display()
+                        );
+                        None
+                    })
+            })
+    }
+
     /// Checks whether AccountsDB has "freshness", not exceeding given slot
     /// Returns current slot if true, otherwise tries to rollback to the
     /// most recent snapshot, which is older than the provided slot

magicblock-accounts-db/src/lib.rs

@@ -70,6 +70,18 @@ impl SnapshotEngine {
         Ok(())
     }
 
+    /// Provides read-only access to the internal snapshots queue.
+    ///
+    /// Executes the given closure `f` with an immutable reference to the snapshots [`VecDeque`].
+    /// This guarantees thread-safe access while preventing modification of the underlying data.
+    pub(crate) fn with_snapshots<F, R>(&self, f: F) -> R
+    where
+        F: Fn(&VecDeque<PathBuf>) -> R,
+    {
+        let snapshots = self.snapshots.lock();
+        f(&snapshots)
+    }
+
     /// Try to rollback to snapshot which is the most recent one before given slot
     ///
     /// NOTE: In case of success, this deletes the primary
@@ -205,11 +217,11 @@ impl SnapshotEngine {
 }
 
 #[derive(Eq, PartialEq, PartialOrd, Ord)]
-struct SnapSlot(u64);
+pub(crate) struct SnapSlot(u64);
 
 impl SnapSlot {
     /// parse snapshot path to extract slot number
-    fn try_from_path(path: &Path) -> Option<Self> {
+    pub(crate) fn try_from_path(path: &Path) -> Option<Self> {
         path.file_name()
             .and_then(|s| s.to_str())
             .and_then(|s| s.split('-').nth(1))
@@ -221,6 +233,10 @@ impl SnapSlot {
         // enforce strict alphanumeric ordering by introducing extra padding
         ppath.join(format!("snapshot-{:0>12}", self.0))
     }
+
+    pub(crate) fn slot(&self) -> u64 {
+        self.0
+    }
 }
 
 /// Conventional byte to byte recursive directory copy,

magicblock-accounts-db/src/snapshot.rs

@@ -209,7 +209,6 @@ where
                 )
                 .map_err(Box::new)?;
         }
-
         // Done
         Ok(signatures)
     }

magicblock-accounts/src/external_accounts_manager.rs

@@ -39,7 +39,11 @@ use magicblock_bank::{
 };
 use magicblock_config::{EphemeralConfig, ProgramConfig};
 use magicblock_geyser_plugin::rpc::GeyserRpcService;
-use magicblock_ledger::{blockstore_processor::process_ledger, Ledger};
+use magicblock_ledger::{
+    blockstore_processor::process_ledger,
+    ledger_truncator::{LedgerTruncator, DEFAULT_TRUNCATION_TIME_INTERVAL},
+    Ledger,
+};
 use magicblock_metrics::MetricsService;
 use magicblock_perf_service::SamplePerformanceService;
 use magicblock_processor::execute_transaction::TRANSACTION_INDEX_LOCK;
@@ -114,6 +118,7 @@ pub struct MagicValidator {
     token: CancellationToken,
     bank: Arc<Bank>,
     ledger: Arc<Ledger>,
+    ledger_truncator: LedgerTruncator<Bank>,
     slot_ticker: Option<tokio::task::JoinHandle<()>>,
     pubsub_handle: RwLock<Option<thread::JoinHandle<()>>>,
     pubsub_close_handle: PubsubServiceCloseHandle,
@@ -192,6 +197,13 @@ 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(
@@ -344,6 +356,7 @@ impl MagicValidator {
             token,
             bank,
             ledger,
+            ledger_truncator,
             accounts_manager,
             transaction_listener,
             transaction_status_sender,
@@ -580,6 +593,8 @@ impl MagicValidator {
         self.start_remote_account_updates_worker();
         self.start_remote_account_cloner_worker().await?;
 
+        self.ledger_truncator.start();
+
         self.rpc_service.start().map_err(|err| {
             ApiError::FailedToStartJsonRpcService(format!("{:?}", err))
         })?;
@@ -677,11 +692,13 @@ impl MagicValidator {
         Ok(())
     }
 
-    pub fn stop(&self) {
+    pub fn stop(&mut self) {
         self.exit.store(true, Ordering::Relaxed);
         self.rpc_service.close();
         PubsubService::close(&self.pubsub_close_handle);
         self.token.cancel();
+        self.ledger_truncator.stop();
+
         // wait a bit for services to stop
         thread::sleep(Duration::from_secs(1));
 

magicblock-api/src/magic_validator.rs

@@ -17,6 +17,7 @@ rayon = { workspace = true, optional = true }
 serde = { workspace = true, features = ["rc"] }
 magicblock-accounts-db = { workspace = true }
 magicblock-program = { workspace = true }
+magicblock-core = { workspace = true }
 solana-accounts-db = { workspace = true }
 solana-address-lookup-table-program = { workspace = true }
 solana-bpf-loader-program = { workspace = true }

magicblock-bank/Cargo.toml

@@ -39,7 +39,7 @@ impl Bank {
         slot_hashes: &SlotHashes,
     ) -> Result<LoadedAddresses, AddressLoaderError> {
         let table_account = self
-            .adb
+            .accounts_db
             .get_account(&table.account_key)
             .map(AccountSharedData::from)
             .map_err(|_| AddressLoaderError::LookupTableAccountNotFound)?;

magicblock-bank/src/address_lookup_table.rs

@@ -17,6 +17,7 @@ use log::{debug, info, trace};
 use magicblock_accounts_db::{
     config::AccountsDbConfig, error::AccountsDbError, AccountsDb, StWLock,
 };
+use magicblock_core::traits::FinalityProvider;
 use solana_accounts_db::{
     accounts_update_notifier_interface::AccountsUpdateNotifierInterface,
     blockhash_queue::BlockhashQueue,
@@ -149,7 +150,7 @@ impl ForkGraph for SimpleForkGraph {
 //#[derive(Debug)]
 pub struct Bank {
     /// Shared reference to accounts database
-    pub adb: AccountsDb,
+    pub accounts_db: AccountsDb,
 
     /// Bank epoch
     epoch: Epoch,
@@ -300,14 +301,16 @@ impl TransactionProcessingCallback for Bank {
         account: &Pubkey,
         owners: &[Pubkey],
     ) -> Option<usize> {
-        self.adb.account_matches_owners(account, owners).ok()
+        self.accounts_db
+            .account_matches_owners(account, owners)
+            .ok()
     }
 
     fn get_account_shared_data(
         &self,
         pubkey: &Pubkey,
     ) -> Option<AccountSharedData> {
-        self.adb.get_account(pubkey).map(Into::into).ok()
+        self.accounts_db.get_account(pubkey).map(Into::into).ok()
     }
 
     // NOTE: must hold idempotent for the same set of arguments
@@ -499,7 +502,7 @@ impl Bank {
         feature_set.activate(&disable_rent_fees_collection::ID, 1);
 
         let mut bank = Self {
-            adb,
+            accounts_db: adb,
             epoch: Epoch::default(),
             epoch_schedule: EpochSchedule::default(),
             is_delta: AtomicBool::default(),
@@ -693,11 +696,11 @@ impl Bank {
     // Slot, Epoch
     // -----------------
     pub fn slot(&self) -> Slot {
-        self.adb.slot()
+        self.accounts_db.slot()
     }
 
     fn set_slot(&self, slot: Slot) {
-        self.adb.set_slot(slot);
+        self.accounts_db.set_slot(slot);
     }
 
     pub fn advance_slot(&self) -> Slot {
@@ -829,16 +832,16 @@ impl Bank {
     // Accounts
     // -----------------
     pub fn has_account(&self, pubkey: &Pubkey) -> bool {
-        self.adb.contains_account(pubkey)
+        self.accounts_db.contains_account(pubkey)
     }
 
     pub fn get_account(&self, pubkey: &Pubkey) -> Option<AccountSharedData> {
-        self.adb.get_account(pubkey).map(Into::into).ok()
+        self.accounts_db.get_account(pubkey).map(Into::into).ok()
     }
 
     /// fn store the single `account` with `pubkey`.
     pub fn store_account(&self, pubkey: Pubkey, account: AccountSharedData) {
-        self.adb.insert_account(&pubkey, &account);
+        self.accounts_db.insert_account(&pubkey, &account);
         if let Some(notifier) = &self.accounts_update_notifier {
             let slot = self.slot();
             notifier.notify_account_update(slot, &account, &None, &pubkey, 0);
@@ -850,13 +853,13 @@ impl Bank {
         &self,
         _sorted: bool,
     ) -> impl Iterator<Item = (Pubkey, AccountSharedData)> + '_ {
-        self.adb.iter_all()
+        self.accounts_db.iter_all()
     }
 
     pub fn store_accounts(&self, accounts: Vec<(Pubkey, AccountSharedData)>) {
         let slot = self.slot();
         for (pubkey, acc) in accounts {
-            self.adb.insert_account(&pubkey, &acc);
+            self.accounts_db.insert_account(&pubkey, &acc);
             if let Some(notifier) = &self.accounts_update_notifier {
                 notifier.notify_account_update(slot, &acc, &None, &pubkey, 0);
             }
@@ -977,7 +980,7 @@ impl Bank {
     where
         F: Fn(&AccountSharedData) -> bool + Send + Sync,
     {
-        self.adb
+        self.accounts_db
             .get_program_accounts(program_id, filter)
             .inspect_err(|err| {
                 log::error!("failed to load program accounts: {err}")
@@ -2303,7 +2306,7 @@ impl Bank {
     }
 
     pub fn accounts_db_storage_size(&self) -> u64 {
-        self.adb.storage_size()
+        self.accounts_db.storage_size()
     }
 
     // -----------------
@@ -2445,7 +2448,7 @@ impl Bank {
     }
 
     pub fn flush(&self) {
-        self.adb.flush(true);
+        self.accounts_db.flush(true);
     }
 }
 
@@ -2527,3 +2530,10 @@ fn max_number_of_accounts_to_collect(
         })
         .sum()
 }
+
+impl FinalityProvider for Bank {
+    fn get_latest_final_slot(&self) -> Slot {
+        // Oldest snapshot or genesis slot
+        self.accounts_db.get_oldest_snapshot_slot().unwrap_or(0)
+    }
+}

magicblock-bank/src/bank.rs

@@ -2,8 +2,11 @@ use serde::{Deserialize, Serialize};
 
 use crate::helpers::serde_defaults::bool_true;
 
+// Default desired ledger size 100 GiB
+pub const DEFAULT_LEDGER_SIZE_BYTES: u64 = 100 * 1024 * 1024 * 1024;
+
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
-#[serde(deny_unknown_fields)]
+#[serde(deny_unknown_fields, rename_all = "kebab-case")]
 pub struct LedgerConfig {
     /// If a previous ledger is found it is removed before starting the validator
     /// This can be disabled by setting [Self::reset] to `false`.
@@ -13,13 +16,21 @@ pub struct LedgerConfig {
     // If left empty it will be auto-generated to a temporary folder
     #[serde(default)]
     pub path: Option<String>,
+    // The size under which it's desired to keep ledger in bytes.
+    #[serde(default = "default_ledger_size")]
+    pub size: u64,
+}
+
+const fn default_ledger_size() -> u64 {
+    DEFAULT_LEDGER_SIZE_BYTES
 }
 
 impl Default for LedgerConfig {
     fn default() -> Self {
         Self {
             reset: bool_true(),
             path: Default::default(),
+            size: DEFAULT_LEDGER_SIZE_BYTES,
         }
     }
 }

magicblock-config/src/ledger.rs

@@ -233,6 +233,11 @@ impl EphemeralConfig {
         if let Ok(ledger_path) = env::var("LEDGER_PATH") {
             config.ledger.path = Some(ledger_path);
         }
+        if let Ok(ledger_path) = env::var("LEDGER_SIZE") {
+            config.ledger.size = ledger_path.parse().unwrap_or_else(|err| {
+                panic!("Failed to parse 'LEDGER_SIZE' as u64: {:?}", err)
+            });
+        }
 
         // -----------------
         // Metrics