feat: return ledger + accountsdb metrics (#624)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Performance**
* Compaction now preserves account-modification data during cleanup to
avoid losing important account-related entries.
* Faster metadata access via column-count caching, reducing overhead
during storage operations.

* **Monitoring**
* Continuous metrics for ledger storage size, accounts storage size, and
account counts.
* New duration histogram for column-count measurements and
renamed/updated execution-time histogram for improved observability.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: taco-paco

magicblock-api/src/tickers.rs

@@ -109,67 +109,70 @@ pub fn init_system_metrics_ticker(
     accountsdb: &Arc<AccountsDb>,
     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);
-    // }
-    // fn set_accounts_count(bank: &Bank) {
-    //     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() => {
-    //                 break;
-    //             }
-    //         }
-    //     }
-    // })
+    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(accounts_db: &AccountsDb) {
+        let byte_size = accounts_db.storage_size();
+        metrics::set_accounts_size(byte_size.try_into().unwrap_or(i64::MAX));
+    }
+    fn set_accounts_count(accounts_db: &AccountsDb) {
+        metrics::set_accounts_count(
+            accounts_db
+                .get_accounts_count()
+                .try_into()
+                .unwrap_or(i64::MAX),
+        );
+    }
 
-    tokio::task::spawn(async move {})
+    let ledger = ledger.clone();
+    let bank = accountsdb.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);
+                    metrics::observe_columns_count_duration(|| try_set_ledger_counts(&ledger));
+                    set_accounts_count(&bank);
+                },
+                _ = token.cancelled() => {
+                    break;
+                }
+            }
+        }
+    })
 }

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,11 @@ impl Column for AccountModDatas {
     fn as_index(slot: Slot) -> Self::Index {
         slot
     }
+
+    /// We don't clean AccountModData on compaction as it isn't slot based
+    fn keep_all_on_compaction() -> bool {
+        true
+    }
 }
 
 impl TypedColumn for AccountModDatas {

magicblock-ledger/src/database/compaction_filter.rs

@@ -7,7 +7,6 @@ use std::{
     },
 };
 
-use log::trace;
 use rocksdb::{
     compaction_filter::CompactionFilter,
     compaction_filter_factory::{
@@ -82,21 +81,17 @@ pub(crate) struct PurgedSlotFilter<C: Column + ColumnName> {
 impl<C: Column + ColumnName> CompactionFilter for PurgedSlotFilter<C> {
     fn filter(
         &mut self,
-        level: u32,
+        _level: u32,
         key: &[u8],
         _value: &[u8],
     ) -> CompactionDecision {
         use rocksdb::CompactionDecision::*;
-        trace!("CompactionFilter: triggered!");
+        if C::keep_all_on_compaction() {
+            return Keep;
+        }
 
         let slot_in_key = C::slot(C::index(key));
         if slot_in_key < self.oldest_slot {
-            trace!(
-                "CompactionFilter: removing key. level: {}, slot: {}",
-                level,
-                slot_in_key
-            );
-
             // It is safe to delete this key
             // since those slots were truncated anyway
             Remove

magicblock-ledger/src/database/ledger_column.rs

@@ -7,9 +7,9 @@ use std::{
 };
 
 use bincode::{deserialize, serialize};
-use log::{error, warn};
+use log::warn;
 use prost::Message;
-use rocksdb::{properties as RocksProperties, ColumnFamily};
+use rocksdb::{properties as RocksProperties, CStrLike, ColumnFamily};
 use serde::de::DeserializeOwned;
 
 use super::{
@@ -228,7 +228,7 @@ where
     /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L654-L689).
     pub fn get_int_property(
         &self,
-        name: &'static std::ffi::CStr,
+        name: impl CStrLike,
     ) -> Result<i64, LedgerError> {
         self.backend.get_int_property_cf(self.handle(), name)
     }
@@ -277,23 +277,45 @@ where
         self.backend.flush_cf(self.handle())
     }
 
+    #[inline(always)]
+    fn is_sequential_cf<T: ColumnName>() -> bool {
+        matches!(
+            T::NAME,
+            crate::database::columns::Blocktime::NAME
+                | crate::database::columns::Blockhash::NAME
+                | crate::database::columns::PerfSamples::NAME
+                | crate::database::columns::AccountModDatas::NAME
+        )
+    }
+
+    /// Initialize value when current one is `DIRTY_VALUE`
+    /// Value isn't set if current doesn't equal `DIRTY_VALUE`
+    fn init_column_count_cache(&self) -> LedgerResult<()> {
+        let count = if Self::is_sequential_cf::<C>() {
+            get_column_count_sequential_column(self)? as i64
+        } else {
+            get_column_count_complex_column(self)? as i64
+        };
+
+        // We can ignore error here since it means value already initialized
+        let _ = self.entry_counter.compare_exchange(
+            DIRTY_COUNT,
+            count,
+            Ordering::AcqRel,
+            Ordering::Relaxed,
+        );
+
+        Ok(())
+    }
+
     pub fn count_column_using_cache(&self) -> LedgerResult<i64> {
         let cached = self.entry_counter.load(Ordering::Relaxed);
         if cached != DIRTY_COUNT {
-            return Ok(cached);
+            Ok(cached)
+        } else {
+            self.init_column_count_cache()?;
+            Ok(self.entry_counter.load(Ordering::Acquire))
         }
-
-        self
-            .iter(IteratorMode::Start)
-            .map(Iterator::count)
-            .map(|val| if val > i64::MAX as usize {
-                // NOTE: this value is only used for metrics/diagnostics and
-                // aside from the fact that we will never encounter this case,
-                // it is good enough to return i64::MAX
-                error!("Column {} count is too large: {} for metrics, returning max.", C::NAME, val);
-                i64::MAX
-            } else { val as i64 })
-            .inspect(|updated| self.entry_counter.store(*updated, Ordering::Relaxed))
     }
 
     /// Increases entries counter if it's not [`DIRTY_COUNT`]
@@ -540,6 +562,38 @@ where
     }
 }
 
+/// When column key format is sequentially incremented key, like: `SlotColumn`
+/// We can simplify extraction of count by calculating `LastKey - FirstKey + 1`
+fn get_column_count_sequential_column<C: Column + ColumnName>(
+    ledger_column: &LedgerColumn<C>,
+) -> LedgerResult<u64> {
+    let last_key = ledger_column.iter(IteratorMode::End)?.next();
+    let start_key = ledger_column.iter(IteratorMode::Start)?.next();
+    let count = match (start_key, last_key) {
+        (Some((start_key, _)), Some((last_key, _))) => {
+            let last_slot = C::slot(last_key);
+            let start_slot = C::slot(start_key);
+
+            last_slot - start_slot + 1
+        }
+        // Empty ColumnFamily
+        _ => 0,
+    };
+
+    Ok(count)
+}
+
+/// For complex columns, like: `AddressSignatures`
+/// We get column count using rocksdb's "estimate-num-keys" proprty
+/// Due to properies of how we use DB this value shall be ~correct on start
+fn get_column_count_complex_column<C: Column + ColumnName>(
+    ledger_column: &LedgerColumn<C>,
+) -> LedgerResult<u64> {
+    const ESTIMATE_NUM_KEYS: &str = "rocksdb.estimate-num-keys";
+
+    Ok(ledger_column.get_int_property(ESTIMATE_NUM_KEYS)? as u64)
+}
+
 /// Increases entries counter if it's not [`DIRTY_COUNT`]
 /// Otherwise just skips it until it is set
 pub fn try_increase_entry_counter(entry_counter: &AtomicI64, by: u64) {

magicblock-ledger/src/database/rocks_db.rs

@@ -8,7 +8,7 @@ use std::{
 };
 
 use rocksdb::{
-    AsColumnFamilyRef, ColumnFamily, DBIterator, DBPinnableSlice,
+    AsColumnFamilyRef, CStrLike, ColumnFamily, DBIterator, DBPinnableSlice,
     DBRawIterator, FlushOptions, IteratorMode as RocksIteratorMode, LiveFile,
     Options, WriteBatch as RWriteBatch, DB,
 };
@@ -237,7 +237,7 @@ impl Rocks {
     pub fn get_int_property_cf(
         &self,
         cf: &ColumnFamily,
-        name: &'static std::ffi::CStr,
+        name: impl CStrLike,
     ) -> LedgerResult<i64> {
         match self.db.property_int_value_cf(cf, name) {
             Ok(Some(value)) => Ok(value.try_into().unwrap()),

magicblock-metrics/src/metrics/mod.rs

@@ -82,6 +82,20 @@ lazy_static::lazy_static! {
     static ref LEDGER_ACCOUNT_MOD_DATA_GAUGE: IntGauge = IntGauge::new(
         "ledger_account_mod_data_gauge", "Ledger Account Mod Data Gauge",
     ).unwrap();
+    pub static ref LEDGER_COLUMNS_COUNT_DURATION_SECONDS: Histogram = Histogram::with_opts(
+        HistogramOpts::new(
+            "ledger_columns_count_duration_seconds",
+            "Time taken to compute ledger columns counts"
+        )
+        .buckets(
+            MICROS_10_90.iter().chain(
+            MICROS_100_900.iter()).chain(
+            MILLIS_1_9.iter()).chain(
+            MILLIS_10_90.iter()).chain(
+            MILLIS_100_900.iter()).chain(
+            SECONDS_1_9.iter()).cloned().collect()
+        ),
+    ).unwrap();
 
     // -----------------
     // Accounts
@@ -193,14 +207,11 @@ lazy_static::lazy_static! {
 
     static ref COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM: HistogramVec = HistogramVec::new(
         HistogramOpts::new(
-            "committor_intent_execution_time_histogram",
+            "committor_intent_execution_time_histogram_v2",
             "Time in seconds spent on intent execution"
         )
         .buckets(
-            MILLIS_1_9.iter()
-            .chain(MILLIS_10_90.iter())
-            .chain(MILLIS_100_900.iter())
-            .chain(SECONDS_1_9.iter()).cloned().collect(),
+            vec![0.01, 0.1, 1.0, 3.0, 5.0, 10.0, 15.0, 20.0, 25.0]
         ),
         &["intent_kind", "outcome_kind"],
     ).unwrap();
@@ -234,6 +245,7 @@ pub(crate) fn register() {
         register!(LEDGER_TRANSACTION_MEMOS_GAUGE);
         register!(LEDGER_PERF_SAMPLES_GAUGE);
         register!(LEDGER_ACCOUNT_MOD_DATA_GAUGE);
+        register!(LEDGER_COLUMNS_COUNT_DURATION_SECONDS);
         register!(ACCOUNTS_SIZE_GAUGE);
         register!(ACCOUNTS_COUNT_GAUGE);
         register!(PENDING_ACCOUNT_CLONES_GAUGE);
@@ -317,6 +329,21 @@ pub fn set_ledger_account_mod_data_count(count: i64) {
     LEDGER_ACCOUNT_MOD_DATA_GAUGE.set(count);
 }
 
+pub fn observe_columns_count_duration<F, T>(f: F) -> T
+where
+    F: FnOnce() -> T,
+{
+    LEDGER_COLUMNS_COUNT_DURATION_SECONDS.observe_closure_duration(f)
+}
+
+pub fn set_accounts_size(value: i64) {
+    ACCOUNTS_SIZE_GAUGE.set(value)
+}
+
+pub fn set_accounts_count(value: i64) {
+    ACCOUNTS_COUNT_GAUGE.set(value)
+}
+
 pub fn inc_pending_clone_requests() {
     PENDING_ACCOUNT_CLONES_GAUGE.inc()
 }