From c2a43bdf905421bdbd66392a267e5c0b0577786f Mon Sep 17 00:00:00 2001 From: Dmitriy Kovalenko Date: Wed, 12 Aug 2026 23:22:17 -0700 Subject: [PATCH] fix(core): share one LMDB env per path within a process because LMDB usee posix lock one process can not hold many open databases --- crates/fff-core/src/dbs/env_pool.rs | 250 ++++++++++++++++++++ crates/fff-core/src/dbs/frecency.rs | 13 +- crates/fff-core/src/dbs/lmdb.rs | 130 +++------- crates/fff-core/src/dbs/mod.rs | 1 + crates/fff-core/src/dbs/query_tracker.rs | 7 +- crates/fff-core/src/error.rs | 16 ++ crates/fff-core/src/shared.rs | 19 +- crates/fff-core/tests/lmdb_env_pool.rs | 123 ++++++++++ packages/fff-bun/test/multi-session.test.ts | 230 ++++++++++++++++++ packages/pi-fff/src/aux-finders.ts | 11 +- packages/pi-fff/src/index.ts | 12 +- packages/pi-fff/test/aux-pool.test.ts | 22 +- 12 files changed, 708 insertions(+), 126 deletions(-) create mode 100644 crates/fff-core/src/dbs/env_pool.rs create mode 100644 crates/fff-core/tests/lmdb_env_pool.rs create mode 100644 packages/fff-bun/test/multi-session.test.ts diff --git a/crates/fff-core/src/dbs/env_pool.rs b/crates/fff-core/src/dbs/env_pool.rs new file mode 100644 index 000000000..0b8e1507d --- /dev/null +++ b/crates/fff-core/src/dbs/env_pool.rs @@ -0,0 +1,250 @@ +use heed::{Env, EnvOpenOptions}; +use std::collections::HashMap; +use std::fs; +use std::ops::Deref; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, LazyLock, Mutex, MutexGuard, PoisonError, Weak}; +use std::thread; +use std::time::Duration; + +use crate::error::{Error, Result}; +use crate::lmdb::DbHealth; + +pub(crate) struct EnvSpec { + pub label: &'static str, + pub map_size: usize, + pub max_dbs: u32, + pub size_cap_bytes: u64, +} + +pub(crate) struct PooledEnv { + env: Env, + key: PathBuf, + /// lmdb's env spec label + label: &'static str, + map_size: usize, + max_dbs: u32, + health: DbHealth, + gc_started: AtomicBool, + dbi_lock: Mutex<()>, +} + +impl Drop for PooledEnv { + fn drop(&mut self) { + let mut pool = POOL.lock().unwrap_or_else(PoisonError::into_inner); + // Only remove a dead entry: begin_exclusive_destroy may have removed ours. + if pool.get(&self.key).is_some_and(|w| w.strong_count() == 0) { + pool.remove(&self.key); + } + // heed closes the env right after this body; a concurrent reopen of the + // same path rides out that gap via env_closing_event in get_or_open. + } +} + +// Cloneable handle to a process-shared LMDB env, derefs to `heed::Env`. +#[derive(Clone)] +pub(crate) struct SharedEnv(Arc); + +impl Deref for SharedEnv { + type Target = Env; + fn deref(&self) -> &Env { + &self.0.env + } +} + +impl std::fmt::Debug for SharedEnv { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_tuple("SharedEnv").field(&self.0.env).finish() + } +} + +impl SharedEnv { + pub(crate) fn get_or_open(db_path: &Path, spec: &EnvSpec) -> Result { + fs::create_dir_all(db_path).map_err(Error::CreateDir)?; + let path = fs::canonicalize(db_path).map_err(|e| Error::EnvOpen { + db: spec.label, + source: heed::Error::Io(e), + })?; + + let mut close_waits = 0u32; + let mut transient_retries = 0u32; + + loop { + let mut open_failed = false; + + { + let mut pool = POOL.lock().unwrap_or_else(PoisonError::into_inner); + if let Some(existing) = pool.get(&path).and_then(Weak::upgrade) { + drop(pool); + if existing.label != spec.label + || existing.map_size != spec.map_size + || existing.max_dbs != spec.max_dbs + { + return Err(Error::EnvSpecMismatch { + path, + open_as: existing.label, + requested_as: spec.label, + }); + } + return Ok(Self(existing)); + } + + erase_if_oversized(&path, spec); + let result = unsafe { + let mut opts = EnvOpenOptions::new(); + opts.map_size(spec.map_size); + if spec.max_dbs > 0 { + opts.max_dbs(spec.max_dbs); + } + opts.open(&path) + }; + + match result { + Ok(env) => { + let entry = Arc::new(PooledEnv { + env, + key: path.clone(), + label: spec.label, + map_size: spec.map_size, + max_dbs: spec.max_dbs, + health: DbHealth::new(), + gc_started: AtomicBool::new(false), + dbi_lock: Mutex::new(()), + }); + pool.insert(path.clone(), Arc::downgrade(&entry)); + drop(pool); + let shared = Self(entry); + + match shared.clear_stale_readers() { + Ok(cleared_count) if cleared_count > 0 => { + tracing::info!( + cleared_count, + db = spec.label, + "reclaimed stale LMDB reader slots at open" + ); + } + Ok(_) => {} + Err(e) => { + tracing::debug!("clear_stale_readers at open failed: {e}") + } + } + + return Ok(shared); + } + Err(heed::Error::EnvAlreadyOpened) => open_failed = true, + // special handling cause we know this happens randomly + Err(e) + if is_transient_env_open_error(&e) + && transient_retries < MAX_TRANSIENT_RETRIES => + { + transient_retries += 1; + tracing::debug!( + path = %path.display(), + transient_retries, + error = ?e, + "transient LMDB env open error, retrying" + ); + } + Err(e) => { + return Err(Error::EnvOpen { + db: spec.label, + source: e, + }); + } + } + } + + if open_failed { + close_waits += 1; + if close_waits > MAX_CLOSE_WAITS { + return Err(Error::EnvOpen { + db: spec.label, + source: heed::Error::EnvAlreadyOpened, + }); + } + + match heed::env_closing_event(&path) { + Some(event) => { + event.wait_timeout(CLOSE_WAIT); + } + None => thread::sleep(Duration::from_millis(2)), + } + } else { + thread::sleep(TRANSIENT_RETRY_SLEEP); + } + } + } + + pub(crate) fn health(&self) -> &DbHealth { + &self.0.health + } + + // First caller wins: GC runs once per opened env, not once per tracker. + pub(crate) fn try_start_gc(&self) -> bool { + !self.0.gc_started.swap(true, Ordering::AcqRel) + } + + // LMDB forbids mdb_dbi_open from concurrent txns in the same process. + pub(crate) fn lock_dbi_open(&self) -> MutexGuard<'_, ()> { + self.0 + .dbi_lock + .lock() + .unwrap_or_else(PoisonError::into_inner) + } + + pub(crate) fn destroy(&self) -> Result> { + let mut pool = POOL.lock().unwrap_or_else(PoisonError::into_inner); + let holders = Arc::strong_count(&self.0); + + if holders > 1 { + return Err(Error::DbInUse { + db: self.0.label, + path: self.0.key.clone(), + holders: holders - 1, + }); + } + + pool.remove(&self.0.key); + Ok(heed::env_closing_event(&self.0.key)) + } +} + +static POOL: LazyLock>>> = LazyLock::new(Mutex::default); + +const CLOSE_WAIT: Duration = Duration::from_millis(100); +const MAX_CLOSE_WAITS: u32 = 100; +const TRANSIENT_RETRY_SLEEP: Duration = Duration::from_millis(50); +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 +fn is_transient_env_open_error(err: &heed::Error) -> bool { + match err { + heed::Error::Io(io) => matches!( + io.kind(), + std::io::ErrorKind::InvalidInput | std::io::ErrorKind::NotFound + ), + _ => false, + } +} + +fn erase_if_oversized(db_path: &Path, spec: &EnvSpec) { + let data = db_path.join("data.mdb"); + let Ok(meta) = fs::metadata(&data) else { + return; + }; + + if meta.len() <= spec.size_cap_bytes { + return; + } + + tracing::error!( + path = %db_path.display(), + size = meta.len(), + cap = spec.size_cap_bytes, + "LMDB db exceeds size cap, erasing" + ); + let _ = fs::remove_file(&data); + let _ = fs::remove_file(db_path.join("lock.mdb")); +} diff --git a/crates/fff-core/src/dbs/frecency.rs b/crates/fff-core/src/dbs/frecency.rs index 7413396a7..de8f9469e 100644 --- a/crates/fff-core/src/dbs/frecency.rs +++ b/crates/fff-core/src/dbs/frecency.rs @@ -1,10 +1,11 @@ use super::db_healthcheck::DbHealthChecker; -use super::lmdb::{DbHealth, LmdbStore, is_map_full}; +use super::env_pool::SharedEnv; use crate::error::{Error, Result}; use crate::file_picker::FFFMode; use crate::git::is_modified_status; +use crate::lmdb::{DbHealth, LmdbStore, is_map_full}; +use heed::Database; use heed::types::{Bytes, SerdeBincode}; -use heed::{Database, Env}; use std::time::{SystemTime, UNIX_EPOCH}; use std::{collections::VecDeque, path::Path}; @@ -19,7 +20,7 @@ const AI_MAX_HISTORY_DAYS: f64 = 7.0; // Only consider accesses within 7 days #[derive(Debug)] pub struct FrecencyTracker { - env: Env, + env: SharedEnv, db: Database>>, health: DbHealth, } @@ -77,7 +78,7 @@ impl LmdbStore for FrecencyTracker { // MAP_SIZE so we don't hit MDB_MAP_FULL before the open-time erase fires. const SIZE_CAP_BYTES: u64 = 12 * 1024 * 1024; - fn env(&self) -> &Env { + fn shared_env(&self) -> &SharedEnv { &self.env } @@ -85,7 +86,7 @@ impl LmdbStore for FrecencyTracker { &self.health } - fn purge_stale_data(env: &Env) -> Result<()> { + fn purge_stale_data(env: &SharedEnv) -> Result<()> { let (deleted, pruned) = Self::purge_stale_entries(env)?; if deleted > 0 || pruned > 0 { tracing::info!(deleted, pruned, "Frecency GC purged entries"); @@ -121,7 +122,7 @@ impl FrecencyTracker { /// Removes entries where all timestamps are older than MAX_HISTORY_DAYS, /// and prunes stale timestamps from entries that still have recent ones. /// Returns (deleted_count, pruned_count). - fn purge_stale_entries(env: &Env) -> Result<(usize, usize)> { + fn purge_stale_entries(env: &SharedEnv) -> Result<(usize, usize)> { let now = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap() diff --git a/crates/fff-core/src/dbs/lmdb.rs b/crates/fff-core/src/dbs/lmdb.rs index 9eba707cb..04361b370 100644 --- a/crates/fff-core/src/dbs/lmdb.rs +++ b/crates/fff-core/src/dbs/lmdb.rs @@ -1,12 +1,11 @@ -use heed::{Database, Env, EnvOpenOptions}; -use std::fs; +use heed::{Database, Env}; use std::path::Path; use std::sync::Arc; use std::sync::RwLock; use std::sync::atomic::{AtomicU8, Ordering}; use std::thread; -use std::time::Duration; +use super::env_pool::{EnvSpec, SharedEnv}; use crate::error::{Error, Result}; pub(crate) fn is_map_full(err: &heed::Error) -> bool { @@ -85,9 +84,13 @@ pub(crate) fn spawn_lmdb_gc(shared: Arc>>) { let Some(ref tracker) = *guard else { return; // destroyed before we started }; - let env = tracker.env(); + // Trackers attaching to an already-pooled env must not repeat the + // GC; the first opener's run flips the shared health flag. + if !tracker.shared_env().try_start_gc() { + return; + } - if let Err(e) = T::purge_stale_data(env) { + if let Err(e) = T::purge_stale_data(tracker.shared_env()) { tracing::debug!("purge_stale_data failed: {e}"); } @@ -105,18 +108,6 @@ pub(crate) fn spawn_lmdb_gc(shared: Arc>>) { } } -// Concurrent `mdb_env_open` calls on the same path can race on macOS -// this is for some reason fixabtly by simple retry of the open -fn is_transient_env_open_error(err: &heed::Error) -> bool { - match err { - heed::Error::Io(io) => matches!( - io.kind(), - std::io::ErrorKind::InvalidInput | std::io::ErrorKind::NotFound - ), - _ => false, - } -} - pub(crate) trait LmdbStore: Sized + Send + Sync + 'static { /// Short label used to defferintiate different instances of this trait const LABEL: &'static str; @@ -127,85 +118,51 @@ pub(crate) trait LmdbStore: Sized + Send + Sync + 'static { /// Hard cap on `data.mdb` size. const SIZE_CAP_BYTES: u64; - /// Borrow the env in the read lock - fn env(&self) -> &Env; + /// Borrow the pooled env handle shared by every tracker of this path. + fn shared_env(&self) -> &SharedEnv; /// Borrow the health flag from the tracker. fn health(&self) -> &DbHealth; + /// Borrow the raw heed env. + fn env(&self) -> &Env { + self.shared_env() + } + /// Override to purge stale rows, compact, etc. Default no-op. Runs on /// the GC thread while a read lock is held against the shared handle, /// so destroy / re-init naturally wait for it. - fn purge_stale_data(_env: &Env) -> Result<()> { + fn purge_stale_data(_env: &SharedEnv) -> Result<()> { Ok(()) } - /// Open the LMDB env. Returns env + a `DbHealth` starting in Pending; - /// the GC thread spawned by `spawn_gc` flips it to Healthy. Write - /// paths flip it to Degraded on MDB_MAP_FULL. + /// Open (or join) the process-shared LMDB env for `db_path`. The health + /// flag is per-env: the GC of the first opener flips it for everyone. #[tracing::instrument] - fn open_env(db_path: &Path) -> Result<(Env, DbHealth)> { - Self::erase_if_oversized(db_path); - fs::create_dir_all(db_path).map_err(Error::CreateDir)?; - let db = Self::LABEL; - - const MAX_ATTEMPTS: u32 = 8; - let mut attempt = 0u32; - let env = loop { - let result = unsafe { - let mut opts = EnvOpenOptions::new(); - opts.map_size(Self::MAP_SIZE); - if Self::MAX_DBS > 0 { - opts.max_dbs(Self::MAX_DBS); - } - opts.open(db_path) - }; - - match result { - Ok(env) => break env, - Err(e) if is_transient_env_open_error(&e) && attempt + 1 < MAX_ATTEMPTS => { - attempt += 1; - tracing::debug!( - path = %db_path.display(), - attempt, - error = ?e, - "transient LMDB env open error, retrying" - ); - - thread::sleep(Duration::from_millis(50)); - } - Err(e) => return Err(Error::EnvOpen { db, source: e }), - } - }; - - // Reclaim reader slots left behind by prior processes that died - // without cleanup. Must run before we start any read txns (which - // open_database_safe does) — otherwise we may hit MDB_READERS_FULL - // on a fresh env just because lock.mdb still has stale entries - // from a previous crash. - // - // This is the one LMDB maintenance call we run on the caller's - // thread. If the lock file is genuinely wedged this will block - // forever, but the alternative — never getting past init — is - // worse and the bg-thread trick doesn't solve it anyway. - match env.clear_stale_readers() { - Ok(cleared) if cleared > 0 => { - tracing::warn!(cleared, "reclaimed stale LMDB reader slots at open"); - } - Ok(_) => {} - Err(e) => tracing::debug!("clear_stale_readers at open failed: {e}"), - } - - Ok((env, DbHealth::new())) + fn open_env(db_path: &Path) -> Result<(SharedEnv, DbHealth)> { + let shared = SharedEnv::get_or_open( + db_path, + &EnvSpec { + label: Self::LABEL, + map_size: Self::MAP_SIZE, + max_dbs: Self::MAX_DBS, + size_cap_bytes: Self::SIZE_CAP_BYTES, + }, + )?; + let health = shared.health().clone(); + Ok((shared, health)) } /// Open or create a database without blocking on the LMDB writer mutex /// when the database already exists. - fn open_database_safe(env: &Env, name: Option<&str>) -> Result> + fn open_database_safe(env: &SharedEnv, name: Option<&str>) -> Result> where KC: 'static, DC: 'static, { let db = Self::LABEL; + // mdb_dbi_open must not run from concurrent txns in this process. + let _dbi_guard = env.lock_dbi_open(); + let rtxn = env .read_txn() .map_err(|source| Error::DbStartReadTxn { db, source })?; @@ -237,23 +194,4 @@ pub(crate) trait LmdbStore: Sized + Send + Sync + 'static { } } } - - fn erase_if_oversized(db_path: &Path) { - let data = db_path.join("data.mdb"); - let Ok(meta) = fs::metadata(&data) else { - return; - }; - if meta.len() <= Self::SIZE_CAP_BYTES { - return; - } - - tracing::error!( - path = %db_path.display(), - size = meta.len(), - cap = Self::SIZE_CAP_BYTES, - "LMDB db exceeds size cap, erasing" - ); - let _ = fs::remove_file(&data); - let _ = fs::remove_file(db_path.join("lock.mdb")); - } } diff --git a/crates/fff-core/src/dbs/mod.rs b/crates/fff-core/src/dbs/mod.rs index 63073cc82..07b4b173f 100644 --- a/crates/fff-core/src/dbs/mod.rs +++ b/crates/fff-core/src/dbs/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod env_pool; pub(crate) mod lmdb; pub mod db_healthcheck; diff --git a/crates/fff-core/src/dbs/query_tracker.rs b/crates/fff-core/src/dbs/query_tracker.rs index 40d01178f..e7ddf1994 100644 --- a/crates/fff-core/src/dbs/query_tracker.rs +++ b/crates/fff-core/src/dbs/query_tracker.rs @@ -1,6 +1,7 @@ use super::db_healthcheck::DbHealthChecker; -use super::lmdb::{DbHealth, LmdbStore, is_map_full}; +use super::env_pool::SharedEnv; use crate::error::Error; +use crate::lmdb::{DbHealth, LmdbStore, is_map_full}; use heed::types::{Bytes, SerdeBincode}; use heed::{Database, Env}; use serde::{Deserialize, Serialize}; @@ -27,7 +28,7 @@ struct HistoryEntry { #[derive(Debug)] pub struct QueryTracker { - env: Env, + env: SharedEnv, // Database for (project_path, query) -> QueryMatchEntry mappings query_file_db: Database>, // Database for project_path -> VecDeque mappings (file picker) @@ -92,7 +93,7 @@ impl LmdbStore for QueryTracker { const MAX_DBS: u32 = 16; const SIZE_CAP_BYTES: u64 = 8 * 1024 * 1024; - fn env(&self) -> &Env { + fn shared_env(&self) -> &SharedEnv { &self.env } diff --git a/crates/fff-core/src/error.rs b/crates/fff-core/src/error.rs index 17a3ebd45..0439559e5 100644 --- a/crates/fff-core/src/error.rs +++ b/crates/fff-core/src/error.rs @@ -34,6 +34,22 @@ pub enum Error { #[source] source: heed::Error, }, + #[error( + "LMDB env at {path} is already open as the '{open_as}' database with different options; requested by '{requested_as}'. Use a distinct path per database." + )] + EnvSpecMismatch { + path: std::path::PathBuf, + open_as: &'static str, + requested_as: &'static str, + }, + #[error( + "The {db} database at {path} is still used by {holders} other tracker(s) in this process" + )] + DbInUse { + db: &'static str, + path: std::path::PathBuf, + holders: usize, + }, #[error("Failed to create {db} database: {source}")] DbCreate { db: &'static str, diff --git a/crates/fff-core/src/shared.rs b/crates/fff-core/src/shared.rs index 06883ea8f..8ae735221 100644 --- a/crates/fff-core/src/shared.rs +++ b/crates/fff-core/src/shared.rs @@ -491,20 +491,31 @@ impl SharedDb { /// Drop the in-memory tracker and delete the on-disk database directory. /// - /// Acquires the write lock, ensuring all readers (including any active mmap - /// access) are finished before the LMDB environment is closed and the files - /// are removed. - /// /// Returns `Ok(Some(path))` with the deleted path, or `Ok(None)` if no tracker was initialized. pub fn destroy(&self) -> Result, Error> { let mut guard = self.write()?; let Some(tracker) = guard.take() else { return Ok(None); }; + + let closing_event = match tracker.shared_env().destroy() { + Ok(closing) => closing, + Err(e) => { + *guard = Some(tracker); + return Err(e); + } + }; + let db_path = tracker.env().path().to_path_buf(); // Drop closes the LMDB env and unmaps the files drop(tracker); drop(guard); + + // Deleting before mdb_env_close finishes would race the unmap. + if let Some(event) = closing_event { + event.wait_timeout(Duration::from_secs(5)); + } + std::fs::remove_dir_all(&db_path).map_err(|source| Error::RemoveDbDir { path: db_path.clone(), source, diff --git a/crates/fff-core/tests/lmdb_env_pool.rs b/crates/fff-core/tests/lmdb_env_pool.rs new file mode 100644 index 000000000..5c67e87ed --- /dev/null +++ b/crates/fff-core/tests/lmdb_env_pool.rs @@ -0,0 +1,123 @@ +//! One process must be able to hold many trackers over the same LMDB path +//! (issues #700/#760): they share a single pooled env instead of failing +//! with `EnvAlreadyOpened`. + +use std::path::{Path, PathBuf}; + +use fff_search::frecency::FrecencyTracker; +use fff_search::query_tracker::QueryTracker; +use fff_search::shared::SharedFrecency; + +fn unique_dir(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("fff-env-pool-{name}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + dir +} + +#[test] +fn same_path_trackers_share_one_env() { + let dir = unique_dir("share"); + let file = Path::new("/virtual/env-pool/shared.rs"); + + let a = FrecencyTracker::open(&dir).expect("first open"); + let b = FrecencyTracker::open(&dir).expect("second open in the same process (#700/#760)"); + + a.track_access(file).expect("write via a"); + assert_eq!(b.access_count(file).expect("read via b"), 1); + + drop(a); + b.track_access(file) + .expect("b must stay usable after a drops"); + assert_eq!(b.access_count(file).unwrap(), 2); + drop(b); + + let c = FrecencyTracker::open(&dir).expect("reopen after all handles dropped"); + assert_eq!( + c.access_count(file).unwrap(), + 2, + "data persisted across reopen" + ); + + drop(c); + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn concurrent_open_and_drop_never_collide() { + let dir = unique_dir("hammer"); + let file = Path::new("/virtual/env-pool/hammer.rs"); + + let mut handles = Vec::new(); + for t in 0..8 { + let dir = dir.clone(); + handles.push(std::thread::spawn(move || { + for i in 0..100 { + let tracker = FrecencyTracker::open(&dir) + .unwrap_or_else(|e| panic!("thread {t} iteration {i}: {e}")); + if i % 20 == 0 { + tracker.track_access(file).expect("track access"); + } + } + })); + } + for handle in handles { + handle.join().expect("no thread may panic"); + } + + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn different_store_on_same_path_is_rejected_with_clear_error() { + let dir = unique_dir("mismatch"); + + let _frecency = FrecencyTracker::open(&dir).expect("frecency open"); + let err = QueryTracker::open(&dir).expect_err("env options differ, must be rejected"); + + let msg = err.to_string(); + assert!( + msg.contains("frecency") && msg.contains("query"), + "error must name both stores so the user can fix their config, got: {msg}" + ); + + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] +fn destroy_refuses_while_shared_then_succeeds_when_sole() { + let dir = unique_dir("destroy"); + let file = Path::new("/virtual/env-pool/destroy.rs"); + + let shared = SharedFrecency::default(); + shared + .init(FrecencyTracker::open(&dir).expect("init open")) + .expect("init"); + let other = FrecencyTracker::open(&dir).expect("second handle over the same db"); + + shared + .destroy() + .expect_err("destroy must refuse while another tracker uses the env"); + + // Refusal must keep both the files and the shared handle intact. + assert!( + dir.join("data.mdb").exists(), + "db files survive a refused destroy" + ); + shared + .read() + .expect("read lock") + .as_ref() + .expect("tracker restored after refused destroy") + .track_access(file) + .expect("shared handle still works"); + + drop(other); + let removed = shared + .destroy() + .expect("sole-owner destroy succeeds") + .expect("a path was removed"); + assert!( + !removed.exists(), + "db dir deleted once nobody shares the env" + ); +} diff --git a/packages/fff-bun/test/multi-session.test.ts b/packages/fff-bun/test/multi-session.test.ts new file mode 100644 index 000000000..b6a4a0aa6 --- /dev/null +++ b/packages/fff-bun/test/multi-session.test.ts @@ -0,0 +1,230 @@ +import { afterEach, describe, expect, mock, test } from "bun:test"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { FileFinder } from "../src/index"; + +// Real-native tests for #700/#760. Lives here, not in pi-fff/test: that suite +// mocks @ff-labs/fff-bun process-globally and bun module mocks can't be undone. +mock.module("@earendil-works/pi-tui", () => ({ + Text: class Text { + text: string; + constructor(text: string) { + this.text = text; + } + setText(text: string) { + this.text = text; + } + }, +})); + +const schema = (type: string) => (options?: unknown) => ({ type, options }); +mock.module("@sinclair/typebox", () => ({ + Type: { + Array: (items: unknown, options?: unknown) => ({ type: "array", items, options }), + Boolean: schema("boolean"), + Number: schema("number"), + Object: (properties: unknown, options?: unknown) => ({ + type: "object", + properties, + options, + }), + Optional: (value: unknown) => ({ ...(value as object), optional: true }), + String: schema("string"), + Union: (items: unknown[], options?: unknown) => ({ type: "union", items, options }), + }, +})); + +const { default: fffExtension } = await import("../../pi-fff/src/index"); + +// Inject this package as the extension's SDK through the cache hook sdk.ts +// already uses for reloads: CI has no node_modules to resolve "@ff-labs/fff-bun" +// from pi-fff, and the finder stays the real native one either way. +(globalThis as Record).__fffSdkPromiseGlobal = Promise.resolve({ + FileFinder, +}); + +const cleanups: Array<() => void> = []; + +afterEach(() => { + while (cleanups.length) cleanups.pop()?.(); +}); + +function makeWorkspace(name: string): string { + const dir = mkdtempSync(join(tmpdir(), `pi-fff-native-${name}-`)); + cleanups.push(() => rmSync(dir, { recursive: true, force: true })); + writeFileSync(join(dir, "alpha.ts"), "export const alpha = 1;\n"); + writeFileSync(join(dir, "beta.ts"), "export const beta = 2;\n"); + mkdirSync(join(dir, "src")); + writeFileSync(join(dir, "src", "gamma.ts"), "export const gamma = 3;\n"); + return dir; +} + +function makeDbPaths() { + const root = mkdtempSync(join(tmpdir(), "pi-fff-native-dbs-")); + cleanups.push(() => rmSync(root, { recursive: true, force: true })); + return { frecencyDbPath: join(root, "frecency"), historyDbPath: join(root, "history") }; +} + +function createFinder(options: Parameters[0]) { + const result = FileFinder.create(options); + if (result.ok) { + const finder = result.value; + cleanups.push(() => { + if (!finder.isDestroyed) finder.destroy(); + }); + } + return result; +} + +type SearchOk = { ok: true; value: { items: Array<{ fileName: string }> } }; +type SearchResult = SearchOk | { ok: false; error: string }; + +function fileNames( + finder: { fileSearch: (q: string, o: { pageSize: number }) => SearchResult }, + query: string, +): string[] { + const search = finder.fileSearch(query, { pageSize: 10 }); + expect(search.ok).toBe(true); + return search.ok ? search.value.items.map((i) => i.fileName) : []; +} + +describe("fff-bun: many finders share one LMDB env per path (#700/#760)", () => { + test("a second finder on the same db paths works and searches", async () => { + const dbs = makeDbPaths(); + const main = createFinder({ basePath: makeWorkspace("main"), ...dbs }); + expect(main.ok).toBe(true); + if (!main.ok) return; + await main.value.waitForScan(15_000); + expect(fileNames(main.value, "alpha")).toContain("alpha.ts"); + + // The createAgentSession scenario: same process, same db paths. + const sub = createFinder({ basePath: makeWorkspace("subagent"), ...dbs }); + expect(sub.ok).toBe(true); + if (!sub.ok) return; + await sub.value.waitForScan(15_000); + expect(fileNames(sub.value, "gamma")).toContain("gamma.ts"); + }, 30_000); + + test("destroying one finder keeps the shared env alive for the other", async () => { + const dbs = makeDbPaths(); + const first = createFinder({ basePath: makeWorkspace("first"), ...dbs }); + const second = createFinder({ basePath: makeWorkspace("second"), ...dbs }); + expect(first.ok).toBe(true); + expect(second.ok).toBe(true); + if (!first.ok || !second.ok) return; + + first.value.destroy(); + await second.value.waitForScan(15_000); + expect(fileNames(second.value, "alpha")).toContain("alpha.ts"); + + // And once the survivor is gone too, the paths are reusable. + second.value.destroy(); + const third = createFinder({ basePath: makeWorkspace("third"), ...dbs }); + expect(third.ok).toBe(true); + }, 30_000); + + test("a db-less aux finder coexists with the main finder (#700)", async () => { + const main = createFinder({ basePath: makeWorkspace("main"), ...makeDbPaths() }); + expect(main.ok).toBe(true); + + const aux = createFinder({ basePath: makeWorkspace("aux") }); + expect(aux.ok).toBe(true); + if (!aux.ok) return; + await aux.value.waitForScan(15_000); + expect(fileNames(aux.value, "gamma")).toContain("gamma.ts"); + }, 30_000); +}); + +type EventHandler = (...args: unknown[]) => unknown; +type RegisteredTool = { + name: string; + execute: ( + toolCallId: string, + params: unknown, + signal?: AbortSignal, + ) => Promise; +}; + +function startSession( + cwd: string, + dbs: { frecencyDbPath: string; historyDbPath: string }, +) { + const events = new Map(); + const tools = new Map(); + const notifications: Array<{ message: string; level?: string }> = []; + + const flags: Record = { + "fff-frecency-db": dbs.frecencyDbPath, + "fff-history-db": dbs.historyDbPath, + }; + + const pi = { + getFlag: (name: string) => flags[name], + on: (event: string, handler: EventHandler) => events.set(event, handler), + registerCommand: () => undefined, + registerFlag: () => undefined, + registerTool: (tool: RegisteredTool) => tools.set(tool.name, tool), + appendEntry: () => undefined, + }; + + const ctx = { + cwd, + ui: { + notify: (message: string, level?: string) => notifications.push({ message, level }), + setStatus: () => undefined, + }, + }; + + fffExtension(pi as never); + cleanups.push(() => { + void events.get("session_shutdown")?.({}, undefined); + }); + + return { + start: async () => events.get("session_start")?.({ reason: "startup" }, ctx), + shutdown: async () => events.get("session_shutdown")?.({}, undefined), + find: async (pattern: string, params?: Record) => + JSON.stringify( + await tools.get("fffind")?.execute("test-call", { pattern, ...params }), + ), + errors: () => notifications.filter((n) => n.level === "error").map((n) => n.message), + }; +} + +describe("pi-fff: in-process double activation works (#760)", () => { + test("two sessions in one process both search against the same dbs", async () => { + const dbs = makeDbPaths(); + const first = startSession(makeWorkspace("session1"), dbs); + await first.start(); + expect(first.errors()).toEqual([]); + expect(await first.find("alpha")).toContain("alpha.ts"); + + // What createAgentSession does: activate the extension again in-process. + const second = startSession(makeWorkspace("session2"), dbs); + await second.start(); + expect(second.errors()).toEqual([]); + expect(await second.find("gamma")).toContain("gamma.ts"); + + // And the first session keeps working alongside it. + expect(await first.find("beta")).toContain("beta.ts"); + + await second.shutdown(); + await first.shutdown(); + }, 40_000); + + test("aux finder over an external root shares the session dbs (#700)", async () => { + const dbs = makeDbPaths(); + const session = startSession(makeWorkspace("aux-session"), dbs); + await session.start(); + expect(session.errors()).toEqual([]); + + // An absolute out-of-workspace path constraint routes to an aux finder, + // which now opens the same frecency/history LMDB paths as the main finder. + const external = makeWorkspace("aux-external"); + expect(await session.find("gamma", { path: external })).toContain("gamma.ts"); + expect(session.errors()).toEqual([]); + + await session.shutdown(); + }, 40_000); +}); diff --git a/packages/pi-fff/src/aux-finders.ts b/packages/pi-fff/src/aux-finders.ts index 48a8b20c6..73ba022d6 100644 --- a/packages/pi-fff/src/aux-finders.ts +++ b/packages/pi-fff/src/aux-finders.ts @@ -18,6 +18,8 @@ export interface AuxOpts { enableHomeDirScanning?: boolean; // Called before a newly spawned aux picker starts a scan that covers $HOME. onHomeDirScan?: (root: string) => void; + frecencyDbPath?: string; + historyDbPath?: string; } export class AuxFinderPool { @@ -100,17 +102,18 @@ export class AuxFinderPool { } const { FileFinder } = await loadSdk(); - // LMDB env can only be opened once per process; the main finder already - // owns the frecency/history DBs. Aux finders are transient and run without - // persistent scoring — see issue #700. const result = FileFinder.create({ basePath: root, + frecencyDbPath: this.opts.frecencyDbPath, + historyDbPath: this.opts.historyDbPath, aiMode: true, enableHomeDirScanning, enableFsRootScanning: this.opts.enableFsRootScanning, }); - if (!result.ok) + + if (!result.ok) { throw new Error(`Failed to create aux file finder for ${root}: ${result.error}`); + } await result.value.waitForScan(SCAN_TIMEOUT_MS); const entry: AuxPicker = { diff --git a/packages/pi-fff/src/index.ts b/packages/pi-fff/src/index.ts index a4261dfb3..616fcf9e6 100644 --- a/packages/pi-fff/src/index.ts +++ b/packages/pi-fff/src/index.ts @@ -318,7 +318,7 @@ export default function fffExtension(pi: ExtensionAPI) { const toolNames = resolveToolNames(currentMode); - // DB path resolution: flag > env > undefined (use fff-node defaults) + // DB path resolution: flag > env > undefined (no persistent DBs) const frecencyDbPath = (pi.getFlag("fff-frecency-db") as string | undefined) ?? process.env.FFF_FRECENCY_DB ?? @@ -329,11 +329,7 @@ export default function fffExtension(pi: ExtensionAPI) { undefined; // flag (boolean) > env ("1"/"true", or "0"/"false") > default. - function resolveBoolOpt( - flagName: string, - envName: string, - fallback = false, - ): boolean { + function resolveBoolOpt(flagName: string, envName: string, fallback = false): boolean { const flag = pi.getFlag(flagName); if (typeof flag === "boolean") return flag; if (typeof flag === "string") return flag === "true" || flag === "1"; @@ -384,10 +380,12 @@ export default function fffExtension(pi: ExtensionAPI) { ); } - let auxPool = new AuxFinderPool({ + const auxPool = new AuxFinderPool({ enableFsRootScanning, enableHomeDirScanning, onHomeDirScan: warnHomeDirScan, + frecencyDbPath, + historyDbPath, }); // in case cwd changes we need to figure this out diff --git a/packages/pi-fff/test/aux-pool.test.ts b/packages/pi-fff/test/aux-pool.test.ts index 6834356a6..da12f6b41 100644 --- a/packages/pi-fff/test/aux-pool.test.ts +++ b/packages/pi-fff/test/aux-pool.test.ts @@ -119,16 +119,26 @@ describe("AuxFinderPool covering reuse", () => { expect(createOptions[0].enableHomeDirScanning).toBe(false); }); - // Regression for #700: aux finders must not reopen the main frecency/history - // LMDB envs, or heed fails with "environment already open in this program". - test("aux finders are created without frecency/history db paths", async () => { - const pool = makePool(); + // #700 is fixed by the process-wide LMDB env pool: same-path opens share one + // env, so aux finders now reuse the session's frecency/history DBs. + test("aux finders receive the pool's frecency/history db paths", async () => { + const pool = makePool({ + frecencyDbPath: "/dbs/frecency", + historyDbPath: "/dbs/history", + }); await pool.acquire("/a/b/c"); await pool.acquire("/x/y"); expect(createOptions.length).toBe(2); for (const opts of createOptions) { - expect(opts.frecencyDbPath).toBeUndefined(); - expect(opts.historyDbPath).toBeUndefined(); + expect(opts.frecencyDbPath).toBe("/dbs/frecency"); + expect(opts.historyDbPath).toBe("/dbs/history"); } }); + + test("aux finders stay db-less when the session has no db paths", async () => { + const pool = makePool(); + await pool.acquire("/a/b/c"); + expect(createOptions[0].frecencyDbPath).toBeUndefined(); + expect(createOptions[0].historyDbPath).toBeUndefined(); + }); });