fix(retention): bound store locks by cancel and deadlines

crates/tracedecay-application/src/git_intelligence.rs

@@ -1762,6 +1762,14 @@ mod tests {
                 "user.email=fixture@example.com",
                 "-c",
                 "commit.gpgsign=false",
+                // `git commit` spawns a detached `git maintenance run --auto`
+                // that holds `.git/objects/maintenance.lock` after the commit
+                // returns; the byte-identical snapshot must not see it appear
+                // or vanish between its two walks.
+                "-c",
+                "maintenance.auto=false",
+                "-c",
+                "gc.auto=0",
             ])
             .args(args)
             .current_dir(self.path())

crates/tracedecay-cli/tests/core_cli_suite/cli_boundary.rs

@@ -17,6 +17,10 @@ fn shipped_binary_stops_quietly_when_a_pipeline_reader_exits() {
     let output = Command::new("sh")
         .args(["-c", r#""$TRACEDECAY_BIN" tool | head -n 4"#])
         .env("TRACEDECAY_BIN", env!("CARGO_BIN_EXE_tracedecay"))
+        // A hotpath-enabled binary binds its metrics port on start; when a
+        // sibling test's daemon already holds it, the bind failure lands on
+        // stderr and breaks the quiet-pipeline assertion below.
+        .env("HOTPATH_METRICS_SERVER_OFF", "true")
         .output()
         .expect("tracedecay tool pipeline should run");
 

crates/tracedecay-code-index-retention/src/code_index_generations.rs

@@ -899,8 +899,11 @@ fn plan_code_generation_retention_with_verification_cancellable(
         let Some(file_name) = generation_file_name(&path) else {
             continue;
         };
-        let (format_revision, manifest, raw_state_digest, size_bytes) =
-            read_generation_metadata(&path, verification, is_cancelled)?;
+        let Some((format_revision, manifest, raw_state_digest, size_bytes)) =
+            read_generation_metadata(&path, verification, is_cancelled)?
+        else {
+            continue;
+        };
         let expected_file = format!(
             "generation-{}.json",
             sha256_hex_suffix(&raw_state_digest).unwrap_or(&raw_state_digest)
@@ -1212,13 +1215,18 @@ fn sweep_unreferenced_generation_segments(
                     })?
                     .to_owned()
             };
-            if read_generation_format_revision(&path, is_cancelled)?
-                != SEALED_GENERATION_FORMAT_REVISION_V1
-            {
+            let Some(revision) = read_generation_format_revision(&path, is_cancelled)? else {
+                continue;
+            };
+            if revision != SEALED_GENERATION_FORMAT_REVISION_V1 {
                 continue;
             }
             let mut reader = CancellableGenerationManifestReaderV1 {
-                file: File::open(&path).map_err(storage)?,
+                file: match File::open(&path) {
+                    Ok(file) => file,
+                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
+                    Err(error) => return Err(storage(error)),
+                },
                 hasher: Sha256::new(),
                 is_cancelled,
                 cancelled: false,
@@ -1281,7 +1289,11 @@ fn sweep_unreferenced_generation_segments(
         if live_segments.contains(&format!("sha256:{digest}")) {
             continue;
         }
-        let metadata = path.symlink_metadata().map_err(storage)?;
+        let metadata = match path.symlink_metadata() {
+            Ok(metadata) => metadata,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
+            Err(error) => return Err(storage(error)),
+        };
         if !metadata.file_type().is_file() {
             return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
                 "generation segment '{}' is not a regular file",

crates/tracedecay-code-index-retention/src/code_index_generations/generation_scan.rs

@@ -15,8 +15,14 @@ const MAX_FORMAT_REVISION_PREFIX_BYTES: usize = 4 * 1024;
 pub(super) fn read_generation_format_revision(
     path: &Path,
     is_cancelled: &dyn Fn() -> bool,
-) -> Result<u32, CodeGenerationRetentionErrorV1> {
-    let mut file = File::open(path).map_err(storage)?;
+) -> Result<Option<u32>, CodeGenerationRetentionErrorV1> {
+    let mut file = match File::open(path) {
+        Ok(file) => file,
+        // The directory entry was removed between listing and open. That is
+        // concurrent publication, not a broken store.
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => return Err(storage(error)),
+    };
     let mut prefix = vec![0_u8; MAX_FORMAT_REVISION_PREFIX_BYTES];
     let bytes_read = file.read(&mut prefix).map_err(storage)?;
     crate::hotpath_observe::retention_inspected(bytes_read as u64);
@@ -25,22 +31,30 @@ pub(super) fn read_generation_format_revision(
         return Err(CodeGenerationRetentionErrorV1::Cancelled);
     }
     prefix.truncate(bytes_read);
-    parse_json_u32_field(&prefix, b"format_revision").ok_or_else(|| {
-        CodeGenerationRetentionErrorV1::UnsafeState(format!(
-            "generation file '{}' has no readable format revision in its bounded prefix",
-            path.display()
-        ))
-    })
+    parse_json_u32_field(&prefix, b"format_revision")
+        .ok_or_else(|| {
+            CodeGenerationRetentionErrorV1::UnsafeState(format!(
+                "generation file '{}' has no readable format revision in its bounded prefix",
+                path.display()
+            ))
+        })
+        .map(Some)
 }
 
 #[hotpath::measure(label = "usecases.retention.read_metadata")]
 pub(super) fn read_generation_metadata(
     path: &Path,
     verification: GenerationDigestVerificationV1,
     is_cancelled: &dyn Fn() -> bool,
-) -> Result<(u32, SealedGenerationManifestMetadataV1, String, u64), CodeGenerationRetentionErrorV1>
-{
-    let mut file = File::open(path).map_err(storage)?;
+) -> Result<
+    Option<(u32, SealedGenerationManifestMetadataV1, String, u64)>,
+    CodeGenerationRetentionErrorV1,
+> {
+    let mut file = match File::open(path) {
+        Ok(file) => file,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => return Err(storage(error)),
+    };
     let size_bytes = file.metadata().map_err(storage)?.len();
     let mut hasher = Sha256::new();
     let mut prefix = Vec::with_capacity(MAX_GENERATION_METADATA_PREFIX_BYTES);
@@ -91,7 +105,7 @@ pub(super) fn read_generation_metadata(
         }
         GenerationDigestVerificationV1::MetadataOnly => named_state_digest(path)?,
     };
-    Ok((format_revision, manifest, state_digest, size_bytes))
+    Ok(Some((format_revision, manifest, state_digest, size_bytes)))
 }
 
 fn named_state_digest(path: &Path) -> Result<String, CodeGenerationRetentionErrorV1> {

crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs

@@ -2,14 +2,14 @@
 //!
 //! Quarantined generations are journaled, then hard-linked into the replay pool before the receipt is durable.
 
+#[cfg(test)]
+use std::cell::Cell;
 use std::collections::BTreeSet;
 use std::fs::File;
 use std::io::Read;
 #[cfg(unix)]
 use std::os::unix::fs::MetadataExt;
 use std::path::{Path, PathBuf};
-#[cfg(test)]
-use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Instant;
 
 use sha2::{Digest, Sha256};
@@ -264,23 +264,30 @@ pub(super) fn acquire_graph_replay_pool_lock_checked(
     GraphReplayPoolLockV1::acquire_exclusive(pool_root, deadline, is_cancelled)
 }
 
+// Per-thread, not process-wide: an acquire runs on its caller's thread, and
+// the test harness runs the other acquire tests in parallel on their own
+// threads. Shared statics let any concurrent acquire land between a test's
+// reset and its read, which is what turned the exact `(1, 0)` proof into an
+// occasional `(5, 3)`.
 #[cfg(test)]
-static GRAPH_REPLAY_POOL_ACQUIRE_TRIES: AtomicUsize = AtomicUsize::new(0);
-#[cfg(test)]
-static GRAPH_REPLAY_POOL_ACQUIRE_WAITS: AtomicUsize = AtomicUsize::new(0);
+thread_local! {
+    static GRAPH_REPLAY_POOL_ACQUIRE_TRIES: Cell<usize> = const { Cell::new(0) };
+    static GRAPH_REPLAY_POOL_ACQUIRE_WAITS: Cell<usize> = const { Cell::new(0) };
+}
 
 #[cfg(test)]
 pub(super) fn reset_graph_replay_pool_acquire_observation() {
-    GRAPH_REPLAY_POOL_ACQUIRE_TRIES.store(0, Ordering::SeqCst);
-    GRAPH_REPLAY_POOL_ACQUIRE_WAITS.store(0, Ordering::SeqCst);
+    GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(|tries| tries.set(0));
+    GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(|waits| waits.set(0));
 }
 
-/// `(non_blocking_tries, wait_for_exclusive_calls)` since the last reset.
+/// `(non_blocking_tries, wait_for_exclusive_calls)` on this thread since the
+/// last reset.
 #[cfg(test)]
 pub(super) fn graph_replay_pool_acquire_observation() -> (usize, usize) {
     (
-        GRAPH_REPLAY_POOL_ACQUIRE_TRIES.load(Ordering::SeqCst),
-        GRAPH_REPLAY_POOL_ACQUIRE_WAITS.load(Ordering::SeqCst),
+        GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(Cell::get),
+        GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(Cell::get),
     )
 }
 
@@ -307,7 +314,7 @@ impl GraphReplayPoolLockV1 {
             // the budget is gone. Windows lock-conflict is `Ok(None)` via
             // `is_lock_contended`, not Storage.
             #[cfg(test)]
-            GRAPH_REPLAY_POOL_ACQUIRE_TRIES.fetch_add(1, Ordering::SeqCst);
+            GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(|tries| tries.set(tries.get() + 1));
             match try_acquire_code_generation_store_lock(pool_root)? {
                 Some(guard) => {
                     crate::hotpath_observe::retention_replay_pool_acquired();
@@ -327,7 +334,7 @@ impl GraphReplayPoolLockV1 {
 
     fn wait_for_exclusive(deadline: Instant) {
         #[cfg(test)]
-        GRAPH_REPLAY_POOL_ACQUIRE_WAITS.fetch_add(1, Ordering::SeqCst);
+        GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(|waits| waits.set(waits.get() + 1));
         crate::hotpath_observe::retention_replay_pool_acquire_wait();
         let remaining = deadline.saturating_duration_since(Instant::now());
         if remaining.is_zero() {

crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs

@@ -1,7 +1,11 @@
 use std::fs::{File, OpenOptions};
 use std::path::{Path, PathBuf};
+use std::time::Instant;
 
-use super::{CodeGenerationRetentionErrorV1, SCOPE_RETENTION_LOCK_FILE, STORE_LOCK_FILE, storage};
+use super::{
+    CodeGenerationRetentionErrorV1, GRAPH_REPLAY_POOL_ACQUIRE_BUDGET,
+    GRAPH_REPLAY_POOL_ACQUIRE_POLL, SCOPE_RETENTION_LOCK_FILE, STORE_LOCK_FILE, storage,
+};
 
 pub struct CodeGenerationStoreLockV1 {
     file: File,
@@ -36,7 +40,33 @@ impl Drop for CodeGenerationStoreLockV1 {
 pub fn acquire_code_generation_store_lock(
     store_root: &Path,
 ) -> Result<CodeGenerationStoreLockV1, CodeGenerationRetentionErrorV1> {
-    lock_file(store_root, STORE_LOCK_FILE, true)
+    acquire_code_generation_store_lock_checked(
+        store_root,
+        Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET,
+        &|| false,
+    )
+}
+
+/// Exclusive generation-store lock that stops at `deadline` or cancellation.
+///
+/// A free lock is taken even when the deadline has already elapsed, so a
+/// caller that only needs one uncontended critical section is not refused.
+/// A held lock returns [`CodeGenerationRetentionErrorV1::GenerationStoreBusy`]
+/// or [`CodeGenerationRetentionErrorV1::Cancelled`] instead of blocking in
+/// `File::lock`, which cannot observe either signal.
+pub fn acquire_code_generation_store_lock_checked(
+    store_root: &Path,
+    deadline: Instant,
+    is_cancelled: &dyn Fn() -> bool,
+) -> Result<CodeGenerationStoreLockV1, CodeGenerationRetentionErrorV1> {
+    lock_file_checked(
+        store_root,
+        STORE_LOCK_FILE,
+        true,
+        deadline,
+        is_cancelled,
+        CodeGenerationRetentionErrorV1::GenerationStoreBusy,
+    )
 }
 
 /// Try to hold the generation store as a reader for one bounded read of
@@ -81,24 +111,63 @@ pub fn try_acquire_code_generation_store_lock(
 pub(super) fn acquire_scope_retention_lock(
     store_root: &Path,
 ) -> Result<CodeGenerationStoreLockV1, CodeGenerationRetentionErrorV1> {
-    lock_file(store_root, SCOPE_RETENTION_LOCK_FILE, false)
+    acquire_scope_retention_lock_checked(
+        store_root,
+        Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET,
+        &|| false,
+    )
+}
+
+pub(super) fn acquire_scope_retention_lock_checked(
+    store_root: &Path,
+    deadline: Instant,
+    is_cancelled: &dyn Fn() -> bool,
+) -> Result<CodeGenerationStoreLockV1, CodeGenerationRetentionErrorV1> {
+    lock_file_checked(
+        store_root,
+        SCOPE_RETENTION_LOCK_FILE,
+        false,
+        deadline,
+        is_cancelled,
+        CodeGenerationRetentionErrorV1::GenerationStoreBusy,
+    )
 }
 
 #[hotpath::measure(label = "code_index_retention.lock")]
-fn lock_file(
+fn lock_file_checked(
     store_root: &Path,
     lock_file: &str,
     generation_store: bool,
+    deadline: Instant,
+    is_cancelled: &dyn Fn() -> bool,
+    busy: CodeGenerationRetentionErrorV1,
 ) -> Result<CodeGenerationStoreLockV1, CodeGenerationRetentionErrorV1> {
     let store_root = canonical_store_root(store_root)?;
-    let lock = open_lock_file(&store_root.join(lock_file))?;
-    lock.lock().map_err(storage)?;
-    Ok(CodeGenerationStoreLockV1 {
-        file: lock,
-        store_root,
-        generation_store,
-        shared: false,
-    })
+    let deadline = deadline.min(Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET);
+    loop {
+        if is_cancelled() {
+            return Err(CodeGenerationRetentionErrorV1::Cancelled);
+        }
+        let lock = open_lock_file(&store_root.join(lock_file))?;
+        match lock.try_lock().map_err(std::io::Error::from) {
+            Ok(()) => {
+                return Ok(CodeGenerationStoreLockV1 {
+                    file: lock,
+                    store_root,
+                    generation_store,
+                    shared: false,
+                });
+            }
+            Err(error) if tracedecay_private_fs::is_lock_contended(&error) => {
+                if Instant::now() >= deadline {
+                    return Err(busy);
+                }
+                let remaining = deadline.saturating_duration_since(Instant::now());
+                std::thread::park_timeout(remaining.min(GRAPH_REPLAY_POOL_ACQUIRE_POLL));
+            }
+            Err(error) => return Err(storage(error)),
+        }
+    }
 }
 
 fn canonical_store_root(store_root: &Path) -> Result<PathBuf, CodeGenerationRetentionErrorV1> {