fix: use spawn_blocking instead of runtime tasks for compaction (#527)

<!-- greptile_comment -->

## Greptile Summary

Updated On: 2025-09-04 05:20:29 UTC

This PR refactors RocksDB compaction operations in the ledger truncation
process to use `tokio::task::spawn_blocking` instead of async tasks. The
changes address a critical performance issue where CPU-intensive RocksDB
compaction operations were blocking the Tokio async runtime threads.

**Key Changes:**

1. **In `ledger_truncator.rs`**: Replaces the coordinated async approach
using `JoinSet` with three separate `spawn_blocking` calls. Previously,
the code used `JoinSet` to run concurrent async tasks for compaction and
awaited their completion. Now it spawns blocking tasks that run
independently in dedicated thread pools.

2. **In `store/api.rs`**: Adds performance measurement instrumentation
to the `compact_slot_range_cf` method, wrapping compaction operations
with timing measurements and logging duration for operational
visibility.

The changes integrate with the existing RocksDB column family system
defined in the ledger's database layer. The compaction operations target
specific column families like `Blocktime`, `TransactionStatus`,
`Transaction`, `TransactionMemos`, and `AddressSignatures`, which are
part of the structured database schema for managing Solana transaction
data.

This refactoring follows a common async Rust pattern of moving blocking
operations to dedicated thread pools to keep the main async runtime
responsive for I/O-bound operations.

## Confidence score: 2/5

- This PR introduces a significant behavioral change that could cause
race conditions or incomplete operations
- Score reflects the removal of synchronization without adding proper
coordination mechanisms
- Pay close attention to `ledger_truncator.rs` as it removes critical
awaiting of compaction completion

<!-- /greptile_comment -->

---------

Co-authored-by: Gabriele Picco <piccogabriele@gmail.com>

Author: taco-paco

magicblock-ledger/src/database/rocksdb_options.rs

@@ -9,20 +9,24 @@ pub fn get_rocksdb_options(access_type: &AccessType) -> Options {
     options.create_if_missing(true);
     options.create_missing_column_families(true);
 
-    // Per the docs, a good value for this is the number of cores on the machine
-    options.increase_parallelism(num_cpus::get() as i32);
-
     // Background thread prioritization: give flushes more threads, limit compaction threads (low-priority)
     let mut env = rocksdb::Env::new().unwrap();
     let cpus_env = num_cpus::get() as i32;
-    // Low-priority threads are used for compaction. Keep them small to favor foreground writes.
-    let low_pri = (cpus_env / 4).clamp(1, 2);
-    env.set_background_threads(low_pri);
+
+    // Bottom-priority are used for bottommost compactions. Keep it minimal - 1
+    let bottom_pri = 1;
+    env.set_bottom_priority_background_threads(bottom_pri);
     // High-priority threads are used for flush. Keep a few to avoid memtable flush backlog.
     let high_pri = cpus_env.clamp(2, 4);
     env.set_high_priority_background_threads(high_pri);
     options.set_env(&env);
 
+    // For every job RocksDB picks a thread from available pools
+    // Here we select ceiling so RocksDB doesn't use all of HIGH + LOW + BOTTOM threads
+    // By default num of threads in LOW is 1
+    let max_jobs = std::cmp::max(high_pri + 1, bottom_pri);
+    options.set_max_background_jobs(max_jobs);
+
     // Bound WAL size
     options.set_max_total_wal_size(4 * 1024 * 1024 * 1024);
 
@@ -44,13 +48,6 @@ pub fn get_rocksdb_options(access_type: &AccessType) -> Options {
     options.set_enable_pipelined_write(true);
     options.set_enable_write_thread_adaptive_yield(true);
 
-    // Background jobs: enough to keep up, not to starve CPU
-    // Cap at 8 or number of CPUs, whichever is smaller but at least 4
-    let cpus = num_cpus::get() as i32;
-    let max_jobs = cpus.clamp(4, 8);
-    options.set_max_background_jobs(max_jobs);
-    options.set_max_subcompactions(2);
-
     // Use direct IO for compaction/flush to avoid page cache contention
     options.set_use_direct_reads(true);
     options.set_use_direct_io_for_flush_and_compaction(true);
@@ -61,10 +58,6 @@ pub fn get_rocksdb_options(access_type: &AccessType) -> Options {
     // Start with a conservative 128 MiB/s, adjustable via config later if needed
     options.set_ratelimiter(128 * 1024 * 1024, 100 * 1000, 10);
 
-    // Prevent large compactions from monopolizing resources
-    options.set_soft_pending_compaction_bytes_limit(8 * 1024 * 1024 * 1024); // 8 GiB
-    options.set_hard_pending_compaction_bytes_limit(32 * 1024 * 1024 * 1024); // 32 GiB
-
     // Dynamic level bytes is a good default to balance levels
     options.set_level_compaction_dynamic_level_bytes(true);
     options.set_report_bg_io_stats(true);

magicblock-ledger/src/ledger_truncator.rs

@@ -2,9 +2,8 @@ use std::{cmp::min, sync::Arc, time::Duration};
 
 use log::{error, info, warn};
 use magicblock_core::traits::FinalityProvider;
-use solana_measure::measure::Measure;
 use tokio::{
-    task::{JoinError, JoinHandle, JoinSet},
+    task::{spawn_blocking, JoinError, JoinHandle},
     time::interval,
 };
 use tokio_util::sync::CancellationToken;
@@ -281,49 +280,33 @@ impl<T: FinalityProvider> LedgerTrunctationWorker<T> {
         // 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 measure = Measure::start("Manual compaction");
-        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)),
-                );
-            }
-        });
+        let ledger_copy = ledger.clone();
+        let handler = spawn_blocking(move || {
+            ledger_copy.compact_slot_range_cf::<Blocktime>(
+                Some(from_slot),
+                Some(to_slot + 1),
+            );
+            ledger_copy.compact_slot_range_cf::<Blockhash>(
+                Some(from_slot),
+                Some(to_slot + 1),
+            );
+            ledger_copy.compact_slot_range_cf::<PerfSamples>(
+                Some(from_slot),
+                Some(to_slot + 1),
+            );
+            ledger_copy.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);
-            }
+            ledger_copy.compact_slot_range_cf::<TransactionStatus>(None, None);
+            ledger_copy.compact_slot_range_cf::<Transaction>(None, None);
+            ledger_copy.compact_slot_range_cf::<TransactionMemos>(None, None);
+            ledger_copy.compact_slot_range_cf::<AddressSignatures>(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;
-        measure.stop();
-        info!("Manual compaction took: {measure}");
+        if let Err(err) = handler.await {
+            error!("compaction aborted {}", err);
+        }
     }
 }
 

magicblock-ledger/src/store/api.rs

@@ -1276,7 +1276,11 @@ impl Ledger {
         from: Option<C::Index>,
         to: Option<C::Index>,
     ) {
+        let mut measure = Measure::start("compaction");
         self.db.column::<C>().compact_range(from, to);
+        measure.stop();
+
+        info!("Compaction of column {} took: {}", C::NAME, measure);
     }
 
     /// Flushes all columns