diff --git a/.alfonso/release-notes/next/lease-release-descriptors.md b/.alfonso/release-notes/next/lease-release-descriptors.md new file mode 100644 index 000000000..32f161cf0 --- /dev/null +++ b/.alfonso/release-notes/next/lease-release-descriptors.md @@ -0,0 +1 @@ +AFT no longer keeps a closed project's lock files and database open, and a lease heartbeat recovers on its next beat after one unreadable read of its lock file. diff --git a/benchmarks/aft-search/slice-descriptors/land-lease-release-descriptors.json b/benchmarks/aft-search/slice-descriptors/land-lease-release-descriptors.json new file mode 100644 index 000000000..ebd574711 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-lease-release-descriptors.json @@ -0,0 +1,8 @@ +{ + "targeted_mechanism": "none", + "kind": "harness", + "slice_class": "non_ranking", + "fixtures": [ + "harness-goldens" + ] +} diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index ac0b5e22e..48e3b1436 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -8703,6 +8703,13 @@ impl AppContext { /// Refreshes the project snapshot on one worker. Batches arriving during a /// walk replace one pending request instead of creating waiting threads. fn refresh_project_applicability(&self, config: Arc, root_changed: bool) { + // Applicability only reads project files and LSP settings. A detached + // walk must not keep storage locks and pinned directories alive after + // the context's last storage-using owner has gone away. + let mut walk_config = (*config).clone(); + walk_config.cache_lease = None; + walk_config.storage_lease = None; + let config = Arc::new(walk_config); let progress = Arc::clone(&self.project_applicability_progress); let mut refresh = progress.refresh.lock(); let epoch = self @@ -21971,6 +21978,59 @@ mod health_warming_honesty_tests { mod configure_applicability_tests { use super::*; + #[test] + fn applicability_walk_does_not_retain_released_storage_leases() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("project"); + let storage = temp.path().join("storage"); + std::fs::create_dir(&root).unwrap(); + std::fs::create_dir(&storage).unwrap(); + let mut config = Config { + project_root: Some(root), + storage_dir: Some(storage.clone()), + ..Config::default() + }; + crate::root_cache::ensure_config_storage_lease(&mut config).unwrap(); + assert!(crate::root_cache::prepare_cache_root(&mut config).is_none()); + let cache_lease = Arc::downgrade(config.cache_lease.as_ref().unwrap()); + let storage_lease = Arc::downgrade(config.storage_lease.as_ref().unwrap()); + let ctx = AppContext::new( + Box::new(crate::parser::TreeSitterProvider::new()), + config.clone(), + ); + let _held_walk = ctx.hold_project_applicability_walks_for_test(); + ctx.refresh_project_applicability(Arc::new(config), false); + + #[cfg(target_os = "linux")] + let open_under_storage = || { + std::fs::read_dir("/proc/self/fd") + .unwrap() + .filter_map(Result::ok) + .filter_map(|entry| std::fs::read_link(entry.path()).ok()) + .filter(|target| target.starts_with(&storage)) + .collect::>() + }; + #[cfg(target_os = "linux")] + assert!(!open_under_storage().is_empty(), "live context owns leases"); + drop(ctx); + #[cfg(target_os = "linux")] + { + let targets = open_under_storage(); + assert!( + targets.is_empty(), + "released storage retained descriptors: {targets:?}" + ); + } + assert!( + cache_lease.upgrade().is_none(), + "cache lease released synchronously" + ); + assert!( + storage_lease.upgrade().is_none(), + "storage lease released synchronously" + ); + } + #[test] fn configure_applicability_content_batches_keep_cache_and_health_counters() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/aft/src/fs_lock.rs b/crates/aft/src/fs_lock.rs index b994361fe..4aed98329 100644 --- a/crates/aft/src/fs_lock.rs +++ b/crates/aft/src/fs_lock.rs @@ -1210,11 +1210,14 @@ impl HeartbeatPump { } self.index(&handle); } - // A partially visible rewrite, or a lease this process is - // replacing right now: read it afresh on the next attempt. - Err(HeartbeatError::Malformed(_)) => { + // A partial rewrite does not prove ownership was lost. Retire + // the descriptor so the next scheduled beat opens the live + // path, including when recovery replaces the inode atomically. + // Let the heartbeat loop schedule that retry: waiting here can + // turn a transient read into a terminal NotOwner timeout. + Err(error @ HeartbeatError::Malformed(_)) => { self.retire_handle(&handle); - self.index(&handle); + return Err(error); } Err(error) => return Err(error), } @@ -4334,6 +4337,71 @@ mod tests { ))); } + #[cfg(target_os = "linux")] + #[test] + fn released_lease_closes_all_descriptors_synchronously() { + let (dir, path) = test_lock_path(); + let open_under_root = || { + fs::read_dir("/proc/self/fd") + .expect("descriptor census") + .filter_map(Result::ok) + .filter_map(|entry| fs::read_link(entry.path()).ok()) + .filter(|target| target.starts_with(dir.path())) + .collect::>() + }; + let guard = acquire_with_config(&path, None, test_config()).expect("acquire lease"); + let handle = Arc::downgrade(&guard.heartbeat_file); + assert!(!open_under_root().is_empty(), "live lease keeps its handle"); + drop(guard); + assert!( + handle.upgrade().is_none(), + "heartbeat must release its handle" + ); + assert!(open_under_root().is_empty(), "no descriptors after release"); + } + + #[test] + fn malformed_beat_reopens_atomic_replacement_on_next_beat() { + let (_dir, path) = test_lock_path(); + let guard = acquire_with_config( + &path, + None, + LockConfig { + heartbeat_interval_ms: 60_000, + ..test_config() + }, + ) + .expect("acquire lease"); + let live = fs::read(&path).expect("read lease"); + fs::write(&path, vec![b'x'; live.len()]).expect("corrupt lease"); + + assert!(matches!( + guard.pump.heartbeat_once(), + Err(HeartbeatError::Malformed(_)) + )); + assert!(guard.heartbeat_file.lock().unwrap().is_none()); + assert!(guard.heartbeat_is_visible()); + + let mut restored = guard.metadata.clone(); + restored.heartbeat_at_ms = now_ms().saturating_sub(1_000_000); + let replacement = path.with_extension("replacement"); + fs::write(&replacement, lock_metadata_bytes(&restored).unwrap()).unwrap(); + rename_over(&replacement, &path).expect("atomic recovery outside the lease registry"); + let _ = io_ledger::take(); + guard + .pump + .heartbeat_once() + .expect("retry through live inode"); + let recovered = read_lock_metadata(&path).expect("independently read recovered lease"); + assert!(lock_identity_matches(&recovered, &guard.metadata)); + assert!(recovered.heartbeat_at_ms > restored.heartbeat_at_ms); + let recovery = io_ledger::take(); + assert_eq!((recovery.opens, recovery.closes), (1, 0)); + + guard.pump.heartbeat_once().expect("healthy next beat"); + assert_beat_opens_no_descriptor(&io_ledger::take()); + } + #[test] fn heartbeat_survives_transient_malformed_and_recovers() { // Regression: a single transient failure (e.g. a read that races a @@ -4371,9 +4439,8 @@ mod tests { let mut restored = owner.clone(); restored.heartbeat_at_ms = now_ms().saturating_sub(1_000_000); let sentinel = restored.heartbeat_at_ms; - // The atomic replacement, not a plain write: it releases the beat's - // kept descriptor, while a plain write would leave that descriptor - // still looking at the corrupted contents. The bytes keep the length. + // Use an atomic replacement to exercise reopening the lease, rather + // than only rewriting the inode that the heartbeat previously held. { let bytes = lock_metadata_bytes(&restored).expect("serialize restored metadata"); with_lease_handles_closed(&path, || replace_lock_bytes_unsynced(&path, &bytes)) diff --git a/crates/aft/src/subc/standing.rs b/crates/aft/src/subc/standing.rs index 55acb898e..8239e74fe 100644 --- a/crates/aft/src/subc/standing.rs +++ b/crates/aft/src/subc/standing.rs @@ -25,24 +25,50 @@ use crate::standing_roots::{StandingRootEntry, StandingRoots}; pub(super) const STANDING_MAINTENANCE_INTERVAL: std::time::Duration = super::DRAIN_TICK_PERIOD; pub(super) struct StandingWorker { - requests: std::sync::mpsc::SyncSender<()>, + requests: Option>, + shutdown: Arc, + worker: Option>, } impl StandingWorker { pub(super) fn start(mut work: impl FnMut() + Send + 'static) -> Self { let (requests, rx) = std::sync::mpsc::sync_channel(1); - std::thread::spawn(move || { + let shutdown = Arc::new(AtomicBool::new(false)); + let worker_shutdown = Arc::clone(&shutdown); + let worker = std::thread::spawn(move || { while rx.recv().is_ok() { + if worker_shutdown.load(Ordering::Acquire) { + break; + } work(); } }); - Self { requests } + Self { + requests: Some(requests), + shutdown, + worker: Some(worker), + } } pub(super) fn request(&self) { // At most one follow-up pass is queued while a reconciliation is busy. // Resolution remains live on each pass, with no stale identity cache. - let _ = self.requests.try_send(()); + if let Some(requests) = &self.requests { + let _ = requests.try_send(()); + } + } +} + +impl Drop for StandingWorker { + fn drop(&mut self) { + // Reconciliation can own a SQLite connection even after the transport + // has disconnected. Stop queued passes and join the current pass so its + // owning connection closes before the module loop releases storage. + self.shutdown.store(true, Ordering::Release); + self.requests.take(); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } } } @@ -1835,6 +1861,60 @@ mod tests { ); } + #[test] + fn standing_worker_drop_waits_for_owned_sqlite_connection() { + use std::time::Duration; + + let temp = tempfile::tempdir().unwrap(); + let connection = Arc::new(parking_lot::Mutex::new( + crate::db::open(&temp.path().join("aft.db")).unwrap(), + )); + let owner = Arc::downgrade(&connection); + let (entered_tx, entered_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let worker = StandingWorker::start(move || { + assert!(connection.lock().is_autocommit()); + entered_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + }); + worker.request(); + entered_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + let (dropping_tx, dropping_rx) = std::sync::mpsc::channel(); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let dropping = std::thread::spawn(move || { + dropping_tx.send(()).unwrap(); + drop(worker); + done_tx.send(()).unwrap(); + }); + dropping_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + let returned_early = done_rx.recv_timeout(Duration::from_millis(100)).is_ok(); + // Release the controlled work even if teardown incorrectly returned, + // so a failed assertion cannot strand its connection-owning thread. + release_tx.send(()).unwrap(); + dropping.join().unwrap(); + assert!( + !returned_early, + "worker drop returned while SQLite was still owned" + ); + assert!( + owner.upgrade().is_none(), + "connection owner survived worker drop" + ); + #[cfg(target_os = "linux")] + { + let targets = std::fs::read_dir("/proc/self/fd") + .unwrap() + .filter_map(Result::ok) + .filter_map(|entry| std::fs::read_link(entry.path()).ok()) + .filter(|target| target.starts_with(temp.path())) + .collect::>(); + assert!( + targets.is_empty(), + "SQLite descriptors survived teardown: {targets:?}" + ); + } + } + #[test] fn standing_tick_requests_do_no_frame_thread_reconciliation() { let frame_thread = std::thread::current().id();