fix: raise LMDB max_readers to avoid MDB_READERS_FULL (#783)

crates/fff-core/src/dbs/db_healthcheck.rs

@@ -14,7 +14,7 @@ pub struct DbHealth {
 }
 
 pub trait DbHealthChecker {
-    fn get_env(&self) -> &heed::Env;
+    fn get_env(&self) -> &heed::Env<heed::WithoutTls>;
     fn is_healthy(&self) -> bool;
     /// Entries per database, each group has a static string label
     fn count_entries(&self) -> Result<Vec<(&'static str, u64)>>;

crates/fff-core/src/dbs/env_pool.rs

@@ -1,4 +1,4 @@
-use heed::{Env, EnvOpenOptions};
+use heed::{Env, EnvOpenOptions, WithoutTls};
 use std::collections::HashMap;
 use std::fs;
 use std::ops::Deref;
@@ -19,7 +19,7 @@ pub(crate) struct EnvSpec {
 }
 
 pub(crate) struct PooledEnv {
-    env: Env,
+    env: Env<WithoutTls>,
     key: PathBuf,
     /// lmdb's env spec label
     label: &'static str,
@@ -47,8 +47,8 @@ impl Drop for PooledEnv {
 pub(crate) struct SharedEnv(Arc<PooledEnv>);
 
 impl Deref for SharedEnv {
-    type Target = Env;
-    fn deref(&self) -> &Env {
+    type Target = Env<WithoutTls>;
+    fn deref(&self) -> &Env<WithoutTls> {
         &self.0.env
     }
 }
@@ -92,8 +92,11 @@ impl SharedEnv {
 
                 erase_if_oversized(&path, spec);
                 let result = unsafe {
-                    let mut opts = EnvOpenOptions::new();
+                    // MDB_NOTLS: reader slots are tied to txn objects (freed on
+                    // commit/abort) instead of pinned per thread for its lifetime (#783).
+                    let mut opts = EnvOpenOptions::new().read_txn_without_tls();
                     opts.map_size(spec.map_size);
+                    opts.max_readers(max_readers());
                     if spec.max_dbs > 0 {
                         opts.max_dbs(spec.max_dbs);
                     }
@@ -219,6 +222,23 @@ const MAX_TRANSIENT_RETRIES: u32 = 8;
 
 // Concurrent mdb_env_open calls on the same path can race on macOS
 // this is for some reason fixable by simple retry of the open
+// heed's default reader table is 126 slots. In TLS mode each thread pins a slot
+// for its lifetime, so long-lived embedders (Neovim, node agents) that share one
+// lock file across many processes/threads exhaust it (#783). Reader slots are
+// tiny (~64B), so raise the ceiling; `FFF_LMDB_MAX_READERS` lets hosts tune it.
+const DEFAULT_MAX_READERS: u32 = 1024;
+
+fn max_readers() -> u32 {
+    parse_max_readers(std::env::var("FFF_LMDB_MAX_READERS").ok())
+}
+
+// Never drop below heed's default 126; ignore missing/garbage/too-small values.
+fn parse_max_readers(raw: Option<String>) -> u32 {
+    raw.and_then(|v| v.trim().parse::<u32>().ok())
+        .filter(|&n| n >= 126)
+        .unwrap_or(DEFAULT_MAX_READERS)
+}
+
 fn is_transient_env_open_error(err: &heed::Error) -> bool {
     match err {
         heed::Error::Io(io) => matches!(
@@ -248,3 +268,17 @@ fn erase_if_oversized(db_path: &Path, spec: &EnvSpec) {
     let _ = fs::remove_file(&data);
     let _ = fs::remove_file(db_path.join("lock.mdb"));
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{DEFAULT_MAX_READERS, parse_max_readers};
+
+    #[test]
+    fn max_readers_parsing() {
+        assert_eq!(parse_max_readers(None), DEFAULT_MAX_READERS);
+        assert_eq!(parse_max_readers(Some("nan".into())), DEFAULT_MAX_READERS);
+        assert_eq!(parse_max_readers(Some("64".into())), DEFAULT_MAX_READERS); // below 126 floor
+        assert_eq!(parse_max_readers(Some(" 512 ".into())), 512);
+        assert_eq!(parse_max_readers(Some("126".into())), 126);
+    }
+}

crates/fff-core/src/dbs/frecency.rs

@@ -43,7 +43,7 @@ const AI_MODIFICATION_THRESHOLDS: [(i64, u64); 5] = [
 ];
 
 impl DbHealthChecker for FrecencyTracker {
-    fn get_env(&self) -> &heed::Env {
+    fn get_env(&self) -> &heed::Env<heed::WithoutTls> {
         &self.env
     }
 

crates/fff-core/src/dbs/lmdb.rs

@@ -1,4 +1,4 @@
-use heed::{Database, Env};
+use heed::{Database, Env, WithoutTls};
 use std::path::Path;
 use std::sync::Arc;
 use std::sync::RwLock;
@@ -124,7 +124,7 @@ pub(crate) trait LmdbStore: Sized + Send + Sync + 'static {
     fn health(&self) -> &DbHealth;
 
     /// Borrow the raw heed env.
-    fn env(&self) -> &Env {
+    fn env(&self) -> &Env<WithoutTls> {
         self.shared_env()
     }
 

crates/fff-core/src/dbs/query_tracker.rs

@@ -39,7 +39,7 @@ pub struct QueryTracker {
 }
 
 impl DbHealthChecker for QueryTracker {
-    fn get_env(&self) -> &Env {
+    fn get_env(&self) -> &Env<heed::WithoutTls> {
         &self.env
     }
 
@@ -198,7 +198,7 @@ impl QueryTracker {
     /// offset=0 returns most recent, offset=1 returns 2nd most recent, etc.
     fn read_history_at_offset(
         db: &Database<Bytes, SerdeBincode<VecDeque<HistoryEntry>>>,
-        env: &Env,
+        env: &Env<heed::WithoutTls>,
         project_key: &[u8; 32],
         offset: usize,
     ) -> Result<Option<String>, Error> {

crates/fff-core/tests/lmdb_readers_full_repro.rs

@@ -0,0 +1,163 @@
+// Repro for #783: fff opens LMDB envs with only map_size set, leaving heed's
+// default max_readers (126) and default TLS mode. Long-lived threads each pin a
+// reader slot for the thread's lifetime, so >126 live reader threads exhaust the
+// table with MDB_READERS_FULL.
+
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Barrier, mpsc};
+use std::time::Duration;
+
+use heed::EnvOpenOptions;
+
+// This binary links heed directly without the fff lib, so nothing pulls in
+// advapi32 for lmdb's security-descriptor calls in mdb_env_setup_locks.
+#[cfg(windows)]
+#[link(name = "advapi32")]
+unsafe extern "C" {}
+
+fn temp_env_dir(name: &str) -> std::path::PathBuf {
+    let dir = std::env::temp_dir().join(format!("fff-readers-{name}-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).unwrap();
+    dir
+}
+
+// Regression for #783: with only map_size set (pre-fix), heed's default 126
+// reader slots are exhausted once >126 live threads each hold a read txn. fff now
+// raises max_readers, so this many live readers must all get a slot.
+const FFF_MAX_READERS: u32 = 1024;
+
+#[test]
+fn raised_max_readers_admits_more_than_126_live_readers() {
+    let dir = temp_env_dir("raised");
+    let env = unsafe {
+        EnvOpenOptions::new()
+            .map_size(10 * 1024 * 1024)
+            .max_readers(FFF_MAX_READERS)
+            .open(&dir)
+    }
+    .unwrap();
+
+    const THREADS: usize = 200;
+    let stop = Arc::new(AtomicBool::new(false));
+    let ready = Arc::new(Barrier::new(THREADS + 1));
+    let readers_full = Arc::new(AtomicBool::new(false));
+    let (tx, rx) = mpsc::channel::<bool>(); // true = read txn acquired
+
+    let mut handles = Vec::new();
+    for _ in 0..THREADS {
+        let env = env.clone();
+        let stop = stop.clone();
+        let ready = ready.clone();
+        let readers_full = readers_full.clone();
+        let tx = tx.clone();
+        handles.push(std::thread::spawn(move || {
+            match env.read_txn() {
+                Ok(txn) => {
+                    tx.send(true).ok();
+                    ready.wait();
+                    while !stop.load(Ordering::Relaxed) {
+                        std::thread::park_timeout(Duration::from_millis(5));
+                    }
+                    drop(txn); // hold the slot for the whole test
+                }
+                Err(e) => {
+                    if e.to_string().contains("MDB_READERS_FULL") {
+                        readers_full.store(true, Ordering::Relaxed);
+                    }
+                    tx.send(false).ok();
+                    ready.wait();
+                }
+            }
+        }));
+    }
+    drop(tx);
+
+    // Collect exactly one result per thread; parked threads keep their tx clone
+    // alive, so we must not wait for the channel to close.
+    let acquired = AtomicUsize::new(0);
+    for _ in 0..THREADS {
+        if rx.recv().unwrap() {
+            acquired.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    ready.wait();
+
+    let acquired = acquired.load(Ordering::Relaxed);
+    stop.store(true, Ordering::Relaxed);
+    for h in handles {
+        h.join().unwrap();
+    }
+    let _ = std::fs::remove_dir_all(&dir);
+
+    // With max_readers raised, all 200 live reader threads must get a slot and
+    // none may see MDB_READERS_FULL. On the pre-fix default of 126 this plateaus
+    // at 126 and the rest fail.
+    assert!(
+        !readers_full.load(Ordering::Relaxed),
+        "MDB_READERS_FULL hit: only {acquired}/{THREADS} live reader threads got a slot"
+    );
+    assert_eq!(
+        acquired, THREADS,
+        "all {THREADS} live reader threads should get a slot; got {acquired}"
+    );
+}
+
+// Structural fix for #783: with MDB_NOTLS a reader slot is tied to the txn
+// object and freed on drop, not pinned per thread. 200 long-lived threads each
+// open+drop a txn against the *default* 126-slot table; in TLS mode this
+// plateaus at 126, in NOTLS mode every thread must succeed.
+#[test]
+fn notls_releases_slots_of_live_threads() {
+    let dir = temp_env_dir("notls");
+    let env = unsafe {
+        EnvOpenOptions::new()
+            .read_txn_without_tls()
+            .map_size(10 * 1024 * 1024)
+            .open(&dir)
+    }
+    .unwrap();
+
+    const THREADS: usize = 200;
+    let stop = Arc::new(AtomicBool::new(false));
+    let ready = Arc::new(Barrier::new(THREADS + 1));
+    // Serialize txns so the test measures slot *release*, not concurrency.
+    let txn_gate = Arc::new(std::sync::Mutex::new(()));
+    let acquired = Arc::new(AtomicUsize::new(0));
+
+    let mut handles = Vec::new();
+    for _ in 0..THREADS {
+        let env = env.clone();
+        let stop = stop.clone();
+        let ready = ready.clone();
+        let txn_gate = txn_gate.clone();
+        let acquired = acquired.clone();
+        handles.push(std::thread::spawn(move || {
+            {
+                let _gate = txn_gate.lock().unwrap();
+                if let Ok(txn) = env.read_txn() {
+                    acquired.fetch_add(1, Ordering::Relaxed);
+                    drop(txn); // NOTLS: slot returns to the pool here
+                }
+            }
+            // Stay alive: in TLS mode this thread would keep its slot pinned.
+            ready.wait();
+            while !stop.load(Ordering::Relaxed) {
+                std::thread::park_timeout(Duration::from_millis(5));
+            }
+        }));
+    }
+
+    ready.wait();
+    let got = acquired.load(Ordering::Relaxed);
+    stop.store(true, Ordering::Relaxed);
+    for h in handles {
+        h.join().unwrap();
+    }
+    let _ = std::fs::remove_dir_all(&dir);
+
+    assert_eq!(
+        got, THREADS,
+        "NOTLS must free slots on txn drop; only {got}/{THREADS} live threads got one"
+    );
+}