fix(rocksdb): Apply native backpressure during bulk indexing

- Retain the bulk L0 compaction trigger at 32, but slow and stop writes at
  48/64 files and at 8/32 GiB of pending compaction debt. This lets RocksDB
  regulate ingestion when compaction cannot keep up.
- Expose the target SST size and bulk-load pending-compaction limits as
  tuning knobs, with defaults of 1024 MB and 8/32 GiB.
- Cap max_subcompactions at four to limit compaction concurrency and memory
  pressure under high db_parallelism settings.

Spenttxouts processing remains fully parallel, with no application-level
polling or chunking.

Co-authored-by: Philippe McLean <philippe.mclean@gmail.com>

Author: shesek

src/config.rs

@@ -77,6 +77,23 @@ pub struct Config {
     /// per SST file of unbounded memory.
     pub db_cache_index_filter_blocks: bool,
 
+    /// RocksDB target_file_size_base in MB (per CF). Smaller values produce smaller
+    /// L1+ SST files at the cost of more files and more compactions overall. Default
+    /// 1024 (1 GiB).
+    pub db_target_file_size_mb: usize,
+
+    /// Used during initial sync.
+    /// RocksDB soft_pending_compaction_bytes_limit in GiB (per CF). When the estimated
+    /// compaction backlog exceeds this, RocksDB rate-limits writes. 0 disables the limit.
+    /// A finite value (e.g. 8) provides automatic backpressure when compaction falls behind.
+    pub db_soft_pending_compaction_gb: u64,
+
+    /// Used during initial sync.
+    /// RocksDB hard_pending_compaction_bytes_limit in GiB (per CF). When the estimated
+    /// compaction backlog exceeds this, RocksDB stops writes entirely until compaction
+    /// catches up. 0 disables. Should be 3-4x the soft limit (e.g. 32 with soft=8).
+    pub db_hard_pending_compaction_gb: u64,
+
     #[cfg(feature = "liquid")]
     pub parent_network: BNetwork,
     #[cfg(feature = "liquid")]
@@ -274,6 +291,24 @@ impl Config {
                 Arg::with_name("cache_index_filter_blocks")
                     .long("cache-index-filter-blocks")
                     .help("Store index/filter blocks in the block cache instead of on the heap. Bounds memory but allows eviction under cache pressure.")
+             ).arg(
+                Arg::with_name("db_target_file_size_mb")
+                    .long("db-target-file-size-mb")
+                    .help("RocksDB target_file_size_base in MB per CF. Smaller values produce more, smaller L1+ SST files. Default 1024.")
+                    .takes_value(true)
+                    .default_value("1024")
+             ).arg(
+                Arg::with_name("db_soft_pending_compaction_gb")
+                    .long("db-soft-pending-compaction-gb")
+                    .help("RocksDB soft_pending_compaction_bytes_limit in GiB per CF during initial sync. RocksDB rate-limits writes above this. 0 disables.")
+                    .takes_value(true)
+                    .default_value("8")
+             ).arg(
+                Arg::with_name("db_hard_pending_compaction_gb")
+                    .long("db-hard-pending-compaction-gb")
+                    .help("RocksDB hard_pending_compaction_bytes_limit in GiB per CF during initial sync. RocksDB stops writes above this. 0 disables.")
+                    .takes_value(true)
+                    .default_value("32")
              ).arg(
                 Arg::with_name("zmq_addr")
                     .long("zmq-addr")
@@ -529,6 +564,17 @@ impl Config {
             db_write_buffer_size_mb: value_t_or_exit!(m, "db_write_buffer_size_mb", usize),
             initial_sync_batch_size: value_t_or_exit!(m, "initial_sync_batch_size", usize),
             db_cache_index_filter_blocks: m.is_present("cache_index_filter_blocks"),
+            db_target_file_size_mb: value_t_or_exit!(m, "db_target_file_size_mb", usize),
+            db_soft_pending_compaction_gb: value_t_or_exit!(
+                m,
+                "db_soft_pending_compaction_gb",
+                u64
+            ),
+            db_hard_pending_compaction_gb: value_t_or_exit!(
+                m,
+                "db_hard_pending_compaction_gb",
+                u64
+            ),
             zmq_addr,
 
             #[cfg(feature = "liquid")]

src/new_index/db.rs

@@ -114,6 +114,7 @@ impl DB {
                 "sentinel 'F' present in {} — using steady-state L0 triggers",
                 cf_name
             );
+            db.apply_steady_state_triggers();
         }
         db
     }
@@ -159,29 +160,23 @@ impl DB {
         //
         // With bloom filters at 10 bits/key and a 128 MB write buffer, each L0
         // file has ~3.9 M keys, so its filter block is ~4.9 MB. At the slowdown
-        // threshold (96 files) that is ~470 MB of pinned filter blocks per CF,
-        // ~1.41 GB across 3 data CFs — within a 2 GB cache. At the stop threshold
-        // (128 files) it is ~628 MB per CF / ~1.88 GB total, still within bounds.
+        // threshold (48 files) that is ~235 MB of pinned filter blocks per CF,
+        // ~705 MB across 3 data CFs — within a 2 GB cache. At the stop threshold
+        // (64 files) it is ~314 MB per CF / ~940 MB total, still within bounds.
         // Previously trigger=64 with 256 MB buffers caused pinned metadata to
         // overflow the 2 GB cache at L0=128 (~3.7 GB), spilling to uncontrolled
-        // heap and triggering OOM. Trigger=32 + slowdown=96 keeps the peak safe
-        // while allowing enough L0 accumulation for good bulk-load throughput.
+        // heap and triggering OOM.
         //
-        // Set slowdown/stop triggers well above the compaction trigger so writes
-        // are never stalled while background compaction catches up.
-        // Disable the pending-compaction-bytes stall so the large backlog that
-        // builds up during the bulk load does not block writes.
-        const L0_BULK_TRIGGER: u32 = 32;
-        let trigger = L0_BULK_TRIGGER.to_string();
-        let slowdown = (L0_BULK_TRIGGER * 3).to_string();
-        let stop = (L0_BULK_TRIGGER * 4).to_string();
+        // Set slowdown/stop triggers above the compaction trigger so RocksDB
+        // throttles writes while background compaction catches up.
+        const L0_BULK_TRIGGER: &str = "32";
+        const L0_BULK_SLOWDOWN: &str = "48";
+        const L0_BULK_STOP: &str = "64";
 
         let opts = [
-            ("level0_file_num_compaction_trigger", trigger.as_str()),
-            ("level0_slowdown_writes_trigger", slowdown.as_str()),
-            ("level0_stop_writes_trigger", stop.as_str()),
-            ("soft_pending_compaction_bytes_limit", "0"),
-            ("hard_pending_compaction_bytes_limit", "0"),
+            ("level0_file_num_compaction_trigger", L0_BULK_TRIGGER),
+            ("level0_slowdown_writes_trigger", L0_BULK_SLOWDOWN),
+            ("level0_stop_writes_trigger", L0_BULK_STOP),
         ];
         self.db.set_options_cf(self.cf(), &opts).unwrap();
     }
@@ -497,9 +492,10 @@ pub fn open_rocksdb(path: &Path, config: &Config) -> rocksdb::DB {
 
     // Parallelize sub-ranges within a single compaction job (including the one-time
     // full_compaction at the end of initial sync). Without this, compact_range() is
-    // single-threaded regardless of increase_parallelism(). Setting it equal to the
-    // parallelism level keeps all background threads busy during the final compaction.
-    db_opts.set_max_subcompactions(parallelism as u32);
+    // single-threaded regardless of increase_parallelism(). Cap it at four because
+    // subcompactions can multiply the configured background-job concurrency:
+    // https://github.com/facebook/rocksdb/wiki/Subcompaction#options
+    db_opts.set_max_subcompactions(parallelism.min(4) as u32);
 
     // Create a single shared LRU cache for all CFs. The total size is
     // --db-block-cache-mb (not multiplied by 3). RocksDB's LRU cache is
@@ -563,7 +559,17 @@ fn data_cf_options(config: &Config, shared_cache: &rocksdb::Cache) -> rocksdb::O
     cf_opts.set_compaction_style(rocksdb::DBCompactionStyle::Level);
     cf_opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
     cf_opts.set_bottommost_compression_type(rocksdb::DBCompressionType::Zstd);
-    cf_opts.set_target_file_size_base(1_073_741_824);
+    cf_opts.set_target_file_size_base((config.db_target_file_size_mb as u64) * 1024 * 1024);
+    // 0 disables the limit. Set finite values via
+    // --db-{soft,hard}-pending-compaction-gb to engage RocksDB's automatic write
+    // throttling when the compaction backlog grows past the threshold. Steady-state
+    // defaults are restored after full compaction.
+    cf_opts.set_soft_pending_compaction_bytes_limit(
+        (config.db_soft_pending_compaction_gb as usize) << 30,
+    );
+    cf_opts.set_hard_pending_compaction_bytes_limit(
+        (config.db_hard_pending_compaction_gb as usize) << 30,
+    );
     // L0 compaction triggers are left at RocksDB defaults (4/20/36) here.
     // After open, apply_bulk_load_triggers() widens them for initial sync
     // when the full-compaction sentinel 'F' is absent.

src/new_index/schema.rs

@@ -2219,6 +2219,8 @@ pub mod bench {
                 index_unspendables: false,
                 network: crate::chain::Network::Regtest,
                 block_batch_size: 250,
+                #[cfg(not(feature = "liquid"))]
+                use_spenttxouts: false,
             };
             let height = 702861;
             let hash = block.block_hash();

tests/common.rs

@@ -124,6 +124,9 @@ impl TestRunner {
             db_write_buffer_size_mb: 256,
             initial_sync_batch_size: 250,
             db_cache_index_filter_blocks: false,
+            db_target_file_size_mb: 1024,
+            db_soft_pending_compaction_gb: 0,
+            db_hard_pending_compaction_gb: 0,
             //#[cfg(feature = "electrum-discovery")]
             //electrum_public_hosts: Option<crate::electrum::ServerHosts>,
             //#[cfg(feature = "electrum-discovery")]