FjallSnapshot: watchdog that drains L0 when fjall's write halt wedges

ARCHITECTURE.md

@@ -468,6 +468,8 @@ Three invariants bind every implementation:
 
 The two LSM backends are interchangeable in contract and share the value-record codec (`snapshot_record.rs`); fjall keeps the crate pure-Rust, RocksDB trades a C++ build dependency for the battle-tested engine and its operational tooling (`ldb`, `sst_dump`). Both keep the cursor in the same atomic batch as the data it names, so under NO_SYNC a crash can lose the un-synced tail but never desynchronize cursor from data — on reopen the recovered cursor is consistent and the watch re-folds the tail. The rest of this section describes the **append-log backend** (the default), whose on-disk format is below.
 
+**fjall write-halt watchdog.** fjall halts a writer while L0 holds 30 runs, but enqueues compaction only from flushes, and a halted writer makes no flushes: if the queued compactions finish without getting L0 under 30, `apply` parks forever (measured on fjall 3.1.4 at 190M entries and 3.1.5 at 282M entries with 4 workers and 64 MiB memtables, L0 at 33 to 35 tables, no compaction running; `check_write_halt` is unchanged through 3.1.13). `FjallSnapshot` runs a watchdog thread that samples once a second and, after 5 consecutive samples with L0 at 30 or more tables, no running compaction, no queued flush and a sequence number that has not moved (a halted writer commits nothing; L0 *tables* run far past 30 on a healthy fold, so the table count alone would fire needless full rewrites), forces `major_compact` on the data partition, which needs no queue message. The thread stops and is joined when the snapshot drops. `FjallStats` (`FjallSnapshot::stats`, or `FjallReader::stats` from a scraper thread, since `apply` holds `&mut self` while parked) exposes `l0_tables`, `outstanding_flushes`, `active_compactions` and `halt_rescues`; a non-zero `halt_rescues` means the wedge was hit and recovered. `FjallConfig::worker_threads` and `memtable_bytes` override the tuned defaults (`examples/halt_bench.rs` is the bulk-apply harness).
+
 ### File Format
 
 ```

examples/halt_bench.rs

@@ -0,0 +1,127 @@
+//! Bulk-apply bench for fjall's write-halt wedge (J11).
+//!
+//! Applies `ENTRIES` puts through `FjallSnapshot::apply` in 1024-entry batches
+//! as fast as possible, printing one line per `REPORT` entries: entries,
+//! interval writes/s, L0 tables, active compactions, watchdog rescues. Exits 2
+//! with `WEDGED` when the interval rate stays under `WEDGE_RATE` writes/s for
+//! `WEDGE_SECS` seconds.
+//!
+//! Env: ENTRIES, KEYS=rand|seq, DIR, REPORT, WEDGE_RATE, WEDGE_SECS,
+//! WORKERS and MEMTABLE_MB (0 = shipped tuning; 4 and 64 is the config that
+//! wedged raw fjall 3.1.5 at 257M entries).
+//!
+//!     cargo run --release --example halt_bench --features fjall
+
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
+use std::time::{Duration, Instant};
+
+use slipstream::{
+    FjallConfig, FjallSnapshot, KvEntry, KvUpdate, SnapshotStore, VersionToken, WatchCursor,
+};
+
+const BATCH: usize = 1024;
+
+fn env<T: std::str::FromStr>(k: &str, d: T) -> T {
+    std::env::var(k)
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(d)
+}
+
+fn mix(mut x: u64) -> u64 {
+    x ^= x >> 33;
+    x = x.wrapping_mul(0xff51_afd7_ed55_8ccd);
+    x ^= x >> 33;
+    x = x.wrapping_mul(0xc4ce_b9fe_1a85_ec53);
+    x ^ (x >> 33)
+}
+
+fn key(i: u64, rand: bool) -> String {
+    let k = if rand { mix(i) } else { i };
+    format!("route.svc-{:08}.{:08}", k / 1000, k % 1000)
+}
+
+fn main() {
+    let entries: u64 = env("ENTRIES", 300_000_000);
+    let rand = env::<String>("KEYS", "rand".into()) == "rand";
+    let report: u64 = env("REPORT", 5_000_000);
+    let wedge_rate: f64 = env("WEDGE_RATE", 2000.0);
+    let wedge_secs: u64 = env("WEDGE_SECS", 120);
+    let dir: PathBuf = env::<String>("DIR", "/tmp/halt_bench".into()).into();
+    let _ = std::fs::remove_dir_all(&dir);
+    let mut value = vec![0u8; 200];
+    let mut s = 0x5EEDu64;
+    for b in &mut value {
+        s = mix(s);
+        *b = s as u8;
+    }
+
+    let config = FjallConfig {
+        worker_threads: env("WORKERS", 0),
+        memtable_bytes: env::<u64>("MEMTABLE_MB", 0) << 20,
+        ..FjallConfig::default()
+    };
+    let (_, mut snap) = FjallSnapshot::open(&dir, config).unwrap();
+    let rd = snap.reader();
+    let applied = Arc::new(AtomicU64::new(0));
+
+    let a2 = applied.clone();
+    let t0 = Instant::now();
+    std::thread::spawn(move || {
+        let (mut last_n, mut last_t, mut next) = (0u64, Instant::now(), report);
+        let mut slow_since = None::<Instant>;
+        loop {
+            std::thread::sleep(Duration::from_secs(1));
+            let n = a2.load(Relaxed);
+            let now = Instant::now();
+            let rate = (n - last_n) as f64 / now.duration_since(last_t).as_secs_f64();
+            let st = rd.stats();
+            let line = format!(
+                "entries={n} writes/s={rate:.0} l0={} active_compactions={} halt_rescues={} t={}s",
+                st.l0_tables,
+                st.active_compactions,
+                st.halt_rescues,
+                t0.elapsed().as_secs()
+            );
+            if n >= next {
+                println!("{line}");
+                next += report;
+            }
+            if n >= entries {
+                println!("DONE {line} wedged=no");
+                std::process::exit(0);
+            }
+            if rate < wedge_rate {
+                let since = *slow_since.get_or_insert(now);
+                if now.duration_since(since).as_secs() >= wedge_secs {
+                    println!("WEDGED {line} wedged=yes");
+                    std::process::exit(2);
+                }
+            } else {
+                slow_since = None;
+            }
+            last_n = n;
+            last_t = now;
+        }
+    });
+
+    let mut batch = Vec::with_capacity(BATCH);
+    let mut i = 0u64;
+    while i < entries {
+        batch.clear();
+        let end = (i + BATCH as u64).min(entries);
+        for j in i..end {
+            batch.push(KvUpdate::Put(KvEntry {
+                key: key(j, rand),
+                value: value.clone(),
+                version: VersionToken::from_u64(j + 1),
+            }));
+        }
+        snap.apply(&batch, &WatchCursor::from_u64(end)).unwrap();
+        applied.store(end, Relaxed);
+        i = end;
+    }
+    std::thread::sleep(Duration::from_secs(2));
+}

src/lib.rs

@@ -55,7 +55,7 @@ pub use nats::{NatsConnection, NatsConnectionConfig, nats_connect};
 pub use repair::{ExpiryRepair, RestoreSource, RestoredFold};
 pub use snapshot::{AppendLogSnapshot, SnapshotStore};
 #[cfg(feature = "fjall")]
-pub use snapshot_fjall::{FjallConfig, FjallReader, FjallSnapshot};
+pub use snapshot_fjall::{FjallConfig, FjallReader, FjallSnapshot, FjallStats};
 #[cfg(feature = "rocksdb")]
 pub use snapshot_rocksdb::{RocksDbConfig, RocksDbReader, RocksDbSnapshot};
 pub use stores::{

src/snapshot_fjall.rs

@@ -58,6 +58,9 @@
 //! cold-read folds favor RocksDB.
 
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::mpsc;
 use std::time::{Duration, Instant};
 
 use fjall::config::{BlockSizePolicy, PinningPolicy};
@@ -99,6 +102,18 @@ const META_MEMTABLE_BYTES: u64 = 8 << 20;
 /// instead of 4 KiB on a cache-miss point read.
 const DATA_BLOCK_SIZE: u32 = 16 * 1024;
 
+/// L0 table count at which fjall's write halt can engage (it counts L0 *runs*,
+/// which are never more than tables, so tables is the conservative trigger).
+const L0_HALT_TABLES: usize = 30;
+
+/// How often the watchdog samples the data partition.
+const WATCHDOG_INTERVAL: Duration = Duration::from_secs(1);
+
+/// Consecutive samples of "no commits, L0 at the halt line, compactor idle" before the
+/// watchdog acts. Compaction picks come in bursts; 5 s idle behind 30 L0 tables
+/// is the wedge, not a gap between picks.
+const WATCHDOG_STRIKES: u32 = 5;
+
 /// Durability and read-cache configuration for [`FjallSnapshot`].
 ///
 /// Defaults to NO_SYNC (`sync: false`) — same cache philosophy as the append
@@ -119,6 +134,14 @@ pub struct FjallConfig {
     /// the cache to the hot set so hydrations stay cache-resident. `0` falls back to
     /// fjall's 32 MiB default. Set this to roughly the resident working-set size.
     pub cache_size_bytes: u64,
+
+    /// fjall flush/compaction worker threads. `0` (the default) uses
+    /// `min(cores, 16)`; on a small host that is `cores`, so hydration shares
+    /// few workers with serving.
+    pub worker_threads: usize,
+
+    /// Data-partition memtable size in bytes. `0` (the default) uses 256 MiB.
+    pub memtable_bytes: u64,
 }
 
 impl Default for FjallConfig {
@@ -128,10 +151,32 @@ impl Default for FjallConfig {
             // 1 GiB: holds index/data blocks for a ~1e6-service working set
             // resident, matching the routing registries' default resident cap.
             cache_size_bytes: 1024 * 1024 * 1024,
+            worker_threads: 0,
+            memtable_bytes: 0,
         }
     }
 }
 
+/// Point-in-time LSM health of a [`FjallSnapshot`], for a consumer's gauges.
+///
+/// `l0_tables >= 30` with `active_compactions == 0` is the shape of fjall's
+/// write-halt wedge: the writer is parked in `apply` and nothing drains L0.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct FjallStats {
+    /// L0 tables in the data partition. fjall throttles writes at 20 runs and halts at 30.
+    pub l0_tables: usize,
+    /// All tables in the data partition.
+    pub tables: usize,
+    /// Memtable flushes queued or running.
+    pub outstanding_flushes: usize,
+    /// Compactions running now.
+    pub active_compactions: usize,
+    /// Times the write-halt watchdog forced a major compaction to drain a wedged
+    /// L0. Non-zero means fjall's write-halt livelock was hit and recovered; a
+    /// rate of it means this node ingests faster than it compacts.
+    pub halt_rescues: u64,
+}
+
 /// On-disk durable fold backed by fjall. See the module docs in `src/snapshot_fjall.rs`.
 pub struct FjallSnapshot {
     // fjall 3 renamed its types: the database root is `Database` (was `Keyspace`)
@@ -144,6 +189,57 @@ pub struct FjallSnapshot {
     // fjall's `Config.path` is `pub(crate)`, so the export path keeps its own
     // copy of the DB directory for the artifact copy.
     path: PathBuf,
+    rescues: Arc<AtomicU64>,
+    /// Dropping the sender stops the watchdog; `Drop` joins it.
+    watchdog: Option<(mpsc::Sender<()>, std::thread::JoinHandle<()>)>,
+}
+
+impl Drop for FjallSnapshot {
+    fn drop(&mut self) {
+        if let Some((stop, handle)) = self.watchdog.take() {
+            drop(stop);
+            let _ = handle.join();
+        }
+    }
+}
+
+/// Un-wedge fjall's write halt.
+///
+/// fjall halts a writer while L0 holds `>= 30` runs, but enqueues compaction only
+/// from flushes (`try_send`, dropped when the queue is full) — and a halted
+/// writer makes no flushes. If the queued compactions finish without bringing L0
+/// under 30 the writer is parked forever (measured: raw 4-worker fjall 3.1.5 at
+/// 257M entries, L0 at 34 tables, no compaction running). A major compaction
+/// needs no queue message and drains L0, so run one when the condition persists.
+fn watchdog(db: Database, data: Keyspace, rescues: Arc<AtomicU64>, stop: mpsc::Receiver<()>) {
+    let mut strikes = 0u32;
+    let mut last_seqno = db.seqno();
+    // `recv_timeout` returns `Disconnected` the moment the snapshot drops.
+    while let Err(mpsc::RecvTimeoutError::Timeout) = stop.recv_timeout(WATCHDOG_INTERVAL) {
+        // A halted writer commits nothing, so a moving sequence number means L0
+        // is high (tables, unlike runs, can be far past 30 on a healthy fold)
+        // but writes are flowing: not the wedge.
+        let seqno = db.seqno();
+        let stalled = seqno == last_seqno;
+        last_seqno = seqno;
+        let wedged = stalled
+            && data.l0_table_count() >= L0_HALT_TABLES
+            && db.active_compactions() == 0
+            && db.outstanding_flushes() == 0;
+        strikes = if wedged { strikes + 1 } else { 0 };
+        if strikes < WATCHDOG_STRIKES {
+            continue;
+        }
+        strikes = 0;
+        tracing::warn!(
+            l0_tables = data.l0_table_count(),
+            "fjall L0 at the write-halt line with an idle compactor; forcing a major compaction"
+        );
+        rescues.fetch_add(1, Ordering::Relaxed);
+        if let Err(e) = data.major_compact() {
+            tracing::error!(error = %e, "fjall write-halt rescue compaction failed");
+        }
+    }
 }
 
 impl FjallSnapshot {
@@ -153,10 +249,13 @@ impl FjallSnapshot {
     /// persisted resume cursor — [`WatchCursor::none`] when fresh — and the store.
     pub fn open(path: &Path, config: FjallConfig) -> Result<(WatchCursor, Self), SnapshotError> {
         std::fs::create_dir_all(path)?;
-        let workers = std::thread::available_parallelism()
-            .map(std::num::NonZero::get)
-            .unwrap_or(4)
-            .min(MAX_WORKER_THREADS);
+        let workers = match config.worker_threads {
+            0 => std::thread::available_parallelism()
+                .map(std::num::NonZero::get)
+                .unwrap_or(4)
+                .min(MAX_WORKER_THREADS),
+            n => n,
+        };
         let mut builder = Database::builder(path).worker_threads(workers);
         // Size the LSM block cache to the working set (default 1 GiB). fjall's own
         // default is 32 MiB, far too small for the fold — see
@@ -168,7 +267,10 @@ impl FjallSnapshot {
         let data = db
             .keyspace(DATA_PARTITION, || {
                 KeyspaceCreateOptions::default()
-                    .max_memtable_size(DATA_MEMTABLE_BYTES)
+                    .max_memtable_size(match config.memtable_bytes {
+                        0 => DATA_MEMTABLE_BYTES,
+                        n => n,
+                    })
                     .data_block_size_policy(BlockSizePolicy::all(DATA_BLOCK_SIZE))
                     // Last-level filters are kept (no `expect_point_read_hits`):
                     // they are the only in-memory rejection for absent-key
@@ -198,6 +300,15 @@ impl FjallSnapshot {
             None => WatchCursor::none(),
         };
 
+        let rescues = Arc::new(AtomicU64::new(0));
+        let (stop_tx, stop_rx) = mpsc::channel();
+        let handle = std::thread::Builder::new()
+            .name("fjall-halt-watchdog".into())
+            .spawn({
+                let (db, data, rescues) = (db.clone(), data.clone(), rescues.clone());
+                move || watchdog(db, data, rescues, stop_rx)
+            })?;
+
         Ok((
             cursor.clone(),
             Self {
@@ -207,6 +318,8 @@ impl FjallSnapshot {
                 config,
                 cursor,
                 path: path.to_path_buf(),
+                rescues,
+                watchdog: Some((stop_tx, handle)),
             },
         ))
     }
@@ -222,10 +335,19 @@ impl FjallSnapshot {
     /// cache miss — without the serving path ever touching the writer.
     pub fn reader(&self) -> FjallReader {
         FjallReader {
+            db: self.db.clone(),
             data: self.data.clone(),
+            rescues: self.rescues.clone(),
         }
     }
 
+    /// Current LSM health. Cheap. A scraper should poll this on a
+    /// [`reader`](Self::reader) handle: `apply` takes `&mut self`, so the
+    /// snapshot itself is unreachable while a halted write is parked inside it.
+    pub fn stats(&self) -> FjallStats {
+        self.reader().stats()
+    }
+
     /// Force a major compaction of the data partition, blocking until done.
     ///
     /// fjall's background compaction is write-driven: after a bulk hydration
@@ -284,6 +406,7 @@ impl FjallSnapshot {
                 FjallConfig {
                     sync: true,
                     cache_size_bytes: 0,
+                    ..FjallConfig::default()
                 },
             )?;
             if staged_cursor != manifest.cursor {
@@ -399,10 +522,23 @@ fn copy_db_dir(src: &Path, dst: &Path) -> Result<(), SnapshotError> {
 /// are safe to run concurrently with it.
 #[derive(Clone)]
 pub struct FjallReader {
+    db: Database,
     data: Keyspace,
+    rescues: Arc<AtomicU64>,
 }
 
 impl FjallReader {
+    /// Current LSM health of the data partition; see [`FjallStats`].
+    pub fn stats(&self) -> FjallStats {
+        FjallStats {
+            l0_tables: self.data.l0_table_count(),
+            tables: self.data.table_count(),
+            outstanding_flushes: self.db.outstanding_flushes(),
+            active_compactions: self.db.active_compactions(),
+            halt_rescues: self.rescues.load(Ordering::Relaxed),
+        }
+    }
+
     /// Live entry for `key`, or `None` if absent/deleted.
     pub fn get(&self, key: &str) -> Result<Option<KvEntry>, SnapshotError> {
         match self.data.get(key.as_bytes()).map_err(map_fjall)? {
@@ -573,6 +709,7 @@ impl SnapshotStore for FjallSnapshot {
                 FjallConfig {
                     sync: true,
                     cache_size_bytes: 0,
+                    ..FjallConfig::default()
                 },
             )?;
             if staged_cursor != self.cursor {