Pin the late link check: Python leak fails by name, Rust try entry reaches the check and release

The Python holder's HELD read is bounded, so a lock leaked on refusal fails the
subtest naming the leak instead of hanging CI. A test-only hook runs a closure
between try_with_exclusive_lock's pre-lock look and its try-lock, so a test puts
the link there and checks the refusal, the untouched target and the freed lock.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/cultcache-py/tests/test_store_lock.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import select
 import shutil
 import subprocess
 import sys
@@ -18,6 +19,7 @@ from cultcache_py.backing_store import CultCacheEnvelope
 REPO = Path(__file__).resolve().parents[3]
 TS_STORE_FILE = REPO / "packages" / "cultcache-ts" / "dist" / "store-file.js"
 HOLD_SECONDS = 1.5
+HELD_WAIT_SECONDS = 3
 
 # A holder that takes the lock the way the Rust runtime does (an exclusive file lock), not through this package.
 PY_HOLDER = """
@@ -43,7 +45,14 @@ def envelope(key: str) -> CultCacheEnvelope:
 
 def start_holder(argv: list[str]) -> subprocess.Popen[str]:
     holder = subprocess.Popen(argv, stdout=subprocess.PIPE, text=True)
-    assert holder.stdout is not None and holder.stdout.readline().strip() == "HELD"
+    assert holder.stdout is not None
+    ready, _, _ = select.select([holder.stdout], [], [], HELD_WAIT_SECONDS)  # a holder blocked on a leaked lock never prints
+    if not ready:
+        holder.kill()
+        holder.wait()
+        holder.stdout.close()
+        raise AssertionError(f"the holder never took {argv[-2]}.lock within {HELD_WAIT_SECONDS} s: an earlier writer leaked the lock")
+    assert holder.stdout.readline().strip() == "HELD"
     holder.held_at = time.monotonic()  # type: ignore[attr-defined]
     return holder
 

packages/cultcache-rs/src/lib.rs

@@ -1861,6 +1861,8 @@ impl SingleFileMessagePackBackingStore {
 
     fn try_with_exclusive_lock<T>(&self, action: impl FnOnce() -> Result<T>) -> Result<Option<T>> {
         refuse_linked(&self.path)?;
+        #[cfg(test)]
+        run_after_pre_lock_look();
         let lock = self.open_lock_file()?;
         match fs2::FileExt::try_lock_exclusive(&lock) {
             Ok(()) => {}
@@ -4773,6 +4775,20 @@ fn injected_write_fault(step: WriteStep) -> std::io::Result<()> {
     Ok(())
 }
 
+// The one gap a test cannot hold open from outside: a try-lock never waits, so what happens between its pre-lock look and
+// the lock is only reachable by running the test's closure there.
+#[cfg(test)]
+thread_local! {
+    static AFTER_PRE_LOCK_LOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const { std::cell::RefCell::new(None) };
+}
+
+#[cfg(test)]
+fn run_after_pre_lock_look() {
+    if let Some(hook) = AFTER_PRE_LOCK_LOOK.with(|hook| hook.borrow_mut().take()) {
+        hook();
+    }
+}
+
 #[cfg(not(test))]
 #[inline(always)]
 fn injected_write_fault(_step: WriteStep) -> std::io::Result<()> {
@@ -8245,6 +8261,38 @@ mod tests {
         Ok(())
     }
 
+    // The try entry has no wait, so the late window is the gap between its look and its lock: a link put there is refused
+    // once the lock is held, nothing is written through or over it, and the lock is released.
+    #[cfg(unix)]
+    #[test]
+    fn a_link_put_between_the_try_entrys_look_and_its_lock_is_refused_and_the_lock_is_released() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let target = temp.path().join("target.cc");
+        SingleFileMessagePackBackingStore::new(&target).push(&snapshot_envelope("a", b"1"))?;
+        let before = fs::read(&target)?;
+        let dangling = temp.path().join("missing.cc");
+        let path = temp.path().join("late.cc");
+        for link_to in [target.clone(), dangling.clone()] {
+            let store = SingleFileMessagePackBackingStore::new(&path);
+            let (link, at) = (link_to.clone(), path.clone());
+            AFTER_PRE_LOCK_LOOK.with(|hook| {
+                *hook.borrow_mut() = Some(Box::new(move || std::os::unix::fs::symlink(&link, &at).expect("put the link")));
+            });
+            let error = store
+                .try_compare_exchange_snapshot(&[], &[snapshot_envelope("b", b"2")])
+                .expect_err("the late link must be refused");
+            assert!(error.downcast_ref::<StoreLinkedError>().is_some(), "{error:#}");
+            assert!(fs::symlink_metadata(&path)?.file_type().is_symlink(), "the link was replaced");
+            assert_eq!(fs::read(&target)?, before, "the target changed");
+            assert!(!dangling.exists(), "the dangling target was created");
+            let probe = store.open_lock_file()?;
+            fs2::FileExt::try_lock_exclusive(&probe).map_err(|e| anyhow!("lock still held: {e}"))?;
+            drop(probe);
+            fs::remove_file(&path)?;
+        }
+        Ok(())
+    }
+
     // A write at a store path that is a link is refused: a replace there would swap the link for a regular file and
     // detach the store from its target. Reads through a live link keep working.
     #[cfg(unix)]