chore: integration batch E (post-fold commits of #1804, #1818, #1821, #1825)

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,26 @@ 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(super) fn acquire_code_generation_store_lock_checked(
+    store_root: &Path,
+    deadline: Instant,
+    is_cancelled: &dyn Fn() -> bool,
+) -> Result<CodeGenerationStoreLockV1, CodeGenerationRetentionErrorV1> {
+    lock_file(store_root, STORE_LOCK_FILE, true, deadline, is_cancelled)
 }
 
 /// Try to hold the generation store as a reader for one bounded read of
@@ -81,24 +104,49 @@ 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)
+    lock_file(
+        store_root,
+        SCOPE_RETENTION_LOCK_FILE,
+        false,
+        Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET,
+        &|| false,
+    )
 }
 
 #[hotpath::measure(label = "code_index_retention.lock")]
 fn lock_file(
     store_root: &Path,
     lock_file: &str,
     generation_store: bool,
+    deadline: Instant,
+    is_cancelled: &dyn Fn() -> bool,
 ) -> 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(CodeGenerationRetentionErrorV1::GenerationStoreBusy);
+                }
+                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> {

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

@@ -1,6 +1,7 @@
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::{Duration, Instant};
 
+use super::super::locking::acquire_code_generation_store_lock_checked;
 use super::*;
 
 fn ensure_replay_pool(pool_root: &std::path::Path) {
@@ -306,6 +307,52 @@ fn checked_acquire_returns_busy_when_the_carried_deadline_has_elapsed() {
     drop(publisher);
 }
 
+#[test]
+fn store_lock_returns_cancelled_without_waiting_out_the_budget() {
+    let (_root, store) = isolated_pool();
+    let holder = hold_replay_pool(&store);
+    let started = Instant::now();
+    let error = match acquire_code_generation_store_lock_checked(
+        &store,
+        Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET,
+        &|| true,
+    ) {
+        Ok(_) => panic!("cancellation must win a held store lock"),
+        Err(error) => error,
+    };
+
+    assert!(matches!(error, CodeGenerationRetentionErrorV1::Cancelled));
+    assert!(
+        started.elapsed() < Duration::from_millis(20),
+        "cancelled store lock must not poll the budget, took {:?}",
+        started.elapsed()
+    );
+    drop(holder);
+}
+
+#[test]
+fn store_lock_returns_busy_when_the_carried_deadline_has_elapsed() {
+    let (_root, store) = isolated_pool();
+    let holder = hold_replay_pool(&store);
+    let started = Instant::now();
+    let error = match acquire_code_generation_store_lock_checked(&store, Instant::now(), &|| false)
+    {
+        Ok(_) => panic!("an elapsed deadline must defer a held store lock"),
+        Err(error) => error,
+    };
+
+    assert!(matches!(
+        error,
+        CodeGenerationRetentionErrorV1::GenerationStoreBusy
+    ));
+    assert!(
+        started.elapsed() < Duration::from_millis(20),
+        "an expired held store lock must not poll, took {:?}",
+        started.elapsed()
+    );
+    drop(holder);
+}
+
 #[test]
 fn checked_acquire_takes_a_free_pool_and_releases_without_leak() {
     let (_root, pool) = isolated_pool();

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

@@ -442,13 +442,15 @@ pub(super) fn plan_collectable_text_artifacts_cancellable(
                 } else {
                     verification
                 };
-                verify_unreferenced_completed_text_artifact(
+                if !verify_unreferenced_completed_text_artifact(
                     &path,
                     digest,
                     metadata.len(),
                     candidate_verification,
                     is_cancelled,
-                )?;
+                )? {
+                    continue;
+                }
                 Some(CodeTextArtifactRetentionCandidateV1 {
                     artifact_file: file_name,
                     kind: CodeTextArtifactRetentionKindV1::Completed,
@@ -554,13 +556,19 @@ pub(super) fn verify_completed_text_artifact(
     is_cancelled: &dyn Fn() -> bool,
 ) -> Result<(), CodeGenerationRetentionErrorV1> {
     let digest = sha256_file_component(&descriptor.artifact_digest, "text artifact")?;
-    verify_unreferenced_completed_text_artifact(
+    if !verify_unreferenced_completed_text_artifact(
         path,
         digest,
         descriptor.artifact_size_bytes,
         verification,
         is_cancelled,
-    )
+    )? {
+        return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
+            "code text artifact '{}' disappeared while its identity was being verified",
+            path.display()
+        )));
+    }
+    Ok(())
 }
 
 /// A content-addressed path is trusted only after the open file and its path
@@ -573,15 +581,23 @@ pub(super) fn verify_unreferenced_completed_text_artifact(
     expected_size_bytes: u64,
     verification: GenerationDigestVerificationV1,
     is_cancelled: &dyn Fn() -> bool,
-) -> Result<(), CodeGenerationRetentionErrorV1> {
-    let before = std::fs::symlink_metadata(path).map_err(storage)?;
+) -> Result<bool, CodeGenerationRetentionErrorV1> {
+    let before = match std::fs::symlink_metadata(path) {
+        Ok(metadata) => metadata,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
+        Err(error) => return Err(storage(error)),
+    };
     if !before.file_type().is_file() || before.len() != expected_size_bytes {
         return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
             "code text artifact '{}' has an invalid regular-file identity",
             path.display()
         )));
     }
-    let file = File::open(path).map_err(storage)?;
+    let file = match File::open(path) {
+        Ok(file) => file,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
+        Err(error) => return Err(storage(error)),
+    };
     if !path_still_names_open_file(path, &file, &before)? {
         return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
             "code text artifact '{}' changed while its identity was being verified",
@@ -602,7 +618,7 @@ pub(super) fn verify_unreferenced_completed_text_artifact(
             path.display()
         )));
     }
-    Ok(())
+    Ok(true)
 }
 
 /// `active_pointer` is the pointer the store carries *now*, which is not
@@ -861,13 +877,18 @@ pub(super) fn stage_collectable_text_artifacts_cancellable(
                     } else {
                         GenerationDigestVerificationV1::Full
                     };
-                    verify_unreferenced_completed_text_artifact(
+                    if !verify_unreferenced_completed_text_artifact(
                         &source,
                         digest,
                         candidate.size_bytes,
                         candidate_verification,
                         is_cancelled,
-                    )?;
+                    )? {
+                        return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
+                            "text-artifact candidate '{}' disappeared before quarantine",
+                            candidate.artifact_file
+                        )));
+                    }
                 }
                 if observe_cancel(is_cancelled) {
                     return Err(CodeGenerationRetentionErrorV1::Cancelled);