Release lease and SQLite descriptors when a project closes; recover the heartbeat after a malformed read

.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.

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"
+  ]
+}

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<Config>, 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::<Vec<_>>()
+        };
+        #[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();

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::<Vec<_>>()
+        };
+        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))

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<std::sync::mpsc::SyncSender<()>>,
+    shutdown: Arc<AtomicBool>,
+    worker: Option<std::thread::JoinHandle<()>>,
 }
 
 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::<Vec<_>>();
+            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();