fix(db): avoid long flush stall on restart

  enable_auto_compaction() was lowering level0_stop_writes_trigger from the
  bulk-load value (512) to the RocksDB default (36) on every update() call.
  At DB open the bulk-load triggers are applied by DB::open, so on any restart
  L0 can legitimately hold more than 36 files. When the first post-restart
  update() called enable_auto_compaction(), the trigger tightening instantly
  put the DB into pre-flush stall territory, and the end-of-batch
  db.flush() that follows parked inside WaitUntilFlushWouldNotStallWrites
  waiting for background compaction to bring L0 below 36.

  On production testnet this reliably cost 77 minutes of indexer freeze per
  restart (verified by 'Manual flush start' → 'Manual flush finished' in the
  RocksDB LOG). The actual memtable flush took 62 ms once unblocked; the rest
  was wait.

  Split enable_auto_compaction() into the minimal flag-flip and a new
  apply_steady_state_triggers() that holds the L0 trigger / pending-bytes-
  limit reset. Invoke the latter exactly once per DB lifetime, inside the
  F-sentinel gate in start_auto_compactions(), immediately after
  full_compaction() has drained L0. On DBs where F is already set (steady-
  state restart), triggers stay at bulk-load values — the comment in
  DB::open already argues that configuration is fine for steady-state reads
  given the prefix bloom filters.

Author: EddieHouston

src/new_index/db.rs

@@ -205,22 +205,38 @@ impl DB {
     }
 
     pub fn enable_auto_compaction(&self) {
-       // Reset L0 triggers and pending-compaction stall thresholds to RocksDB
-       // defaults, so that steady-state operation compacts promptly and avoids
-       // unbounded compaction backlogs that cause read latency spikes.
-       // RocksDB defaults (stable since v5.x through v10.4.2). Hardcoded because
-       // set_options() doesn't return previous values and the Rust bindings lack getters.
+        // Flip the enabled flag only. Trigger/stall-threshold tightening is handled
+        // once per DB lifetime in apply_steady_state_triggers(), gated on a prior
+        // full_compaction() draining L0 first. See the comment there for why the
+        // two are split.
+        let opts = [("disable_auto_compactions", "false")];
+        self.db.set_options(&opts).unwrap();
+    }
 
+    /// One-time transition from bulk-load compaction triggers (set at DB open in
+    /// [`DB::open`]) to RocksDB defaults, for lower read amplification in
+    /// steady-state operation.
+    ///
+    /// Must be called only after a compaction has drained L0 and any level-size
+    /// imbalance — otherwise the tightened `level0_stop_writes_trigger` parks
+    /// foreground flushes and sync writes in `WaitUntilFlushWouldNotStallWrites`
+    /// until background compaction catches up. On a mature DB with an hour-long
+    /// bottommost compaction in flight, that wait can exceed 70 minutes.
+    ///
+    /// Gated by the `F` sentinel key in `start_auto_compactions`, so it runs at
+    /// most once per DB lifetime, immediately after `full_compaction()`.
+    pub fn apply_steady_state_triggers(&self) {
+        // RocksDB defaults (stable since v5.x through v10.4.2). Hardcoded because
+        // set_options() doesn't return previous values and the Rust bindings lack getters.
         let soft_limit = (64u64 << 30).to_string(); // 64 GiB
         let hard_limit = (256u64 << 30).to_string(); // 256 GiB
 
         let opts = [
-            ("disable_auto_compactions", "false"),
             ("level0_file_num_compaction_trigger", "4"),
             ("level0_slowdown_writes_trigger", "20"),
             ("level0_stop_writes_trigger", "36"),
-            ("soft_pending_compaction_bytes_limit", &soft_limit),
-            ("hard_pending_compaction_bytes_limit", &hard_limit),
+            ("soft_pending_compaction_bytes_limit", soft_limit.as_str()),
+            ("hard_pending_compaction_bytes_limit", hard_limit.as_str()),
         ];
         self.db.set_options(&opts).unwrap();
     }

src/new_index/schema.rs

@@ -286,6 +286,12 @@ impl Indexer {
         let key = b"F".to_vec();
         if db.get(&key).is_none() {
             db.full_compaction();
+            // Tighten L0 triggers and pending-compaction stall thresholds now that
+            // full_compaction() has drained L0. Running this before full_compaction
+            // (or on every restart, as the code previously did) parks the end-of-batch
+            // flush in WaitUntilFlushWouldNotStallWrites for as long as it takes
+            // background compaction to bring L0 below the new stop_writes_trigger.
+            db.apply_steady_state_triggers();
             db.put_sync(&key, b"");
             assert!(db.get(&key).is_some());
         }