Late-link follow-up: Python probes the writer's own lock, Rust shares one locked tail, dangling-link cases, doc comment restored

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

packages/cultcache-py/CHANGELOG.md

@@ -8,6 +8,7 @@
 
 ### Changed
 
+- `store_write_lock` looks for a link at the store path again once the lock is held, so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, instead of being replaced. `JsonLinesBackingStore` takes no cross-process lock (only its thread lock, under which it already refuses a link), so it has no lock wait to re-check across. Pinned by a test that puts the link while a foreign process holds the lock.
 - `CultCache.is_writable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store.
 - `JsonLinesBackingStore` writes LF line endings on Windows. It wrote CRLF there, through `write_text` newline translation, before the commit_store_file change (#154, adbbb780).
 - `msgpack` is a hard dependency, as `pyproject.toml` declared it. The optional-import fallbacks and the `cultcache-py[msgpack]` extra they named (which never existed) are gone: a missing msgpack is an `ImportError` at import.

packages/cultcache-py/src/cultcache_py/stores.py

@@ -495,6 +495,8 @@ def store_write_lock(path: Path, timeout: float = LOCK_TIMEOUT_SECONDS) -> Itera
             raise TimeoutError(f"CultCache timed out after {timeout} s waiting for the store lock {lock_path}.")
         time.sleep(_LOCK_POLL_SECONDS)
     try:
+        # A link may have been put at the path while the lock was awaited; the look above only saw the path before.
+        _refuse_linked(path)
         yield
     finally:
         release()

packages/cultcache-py/tests/test_store_lock.py

@@ -5,12 +5,14 @@ import shutil
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import unittest
 from pathlib import Path
 from unittest import mock
 
 from cultcache_py import SingleFileMessagePackBackingStore, stores
+from cultcache_py.stores import StoreLinkedError, store_write_lock
 from cultcache_py.backing_store import CultCacheEnvelope
 
 REPO = Path(__file__).resolve().parents[3]
@@ -73,6 +75,48 @@ class StoreLockTests(unittest.TestCase):
         holder = start_holder(["node", "-e", TS_HOLDER, str(TS_STORE_FILE), str(self.path), str(HOLD_SECONDS)])
         self.assert_write_waits_for(holder)
 
+    def test_a_link_put_while_a_writer_waits_on_the_lock_is_refused_once_it_holds_it(self) -> None:
+        target = self.dir / "volume" / "store.cc"
+        target.parent.mkdir()
+        SingleFileMessagePackBackingStore(target).push(envelope("a"))
+        before = target.read_bytes()
+        writes = {
+            "push": lambda store: store.push(envelope("b")),
+            "push_all": lambda store: store.push_all([envelope("b")]),
+            "delete": lambda store: store.delete("t", "a"),
+        }
+        missing = self.dir / "volume" / "missing.cc"
+        for (name, write), link_to in [(item, to) for item in writes.items() for to in (target, missing)]:
+            with self.subTest(name, dangling=link_to == missing):
+                if os.path.lexists(self.path):
+                    os.unlink(self.path)
+                holder = start_holder([sys.executable, "-c", PY_HOLDER, str(self.path), str(HOLD_SECONDS)])
+                self.addCleanup(holder.wait)
+                self.addCleanup(holder.stdout.close)  # type: ignore[union-attr]
+                outcome: list[BaseException | None] = []
+
+                def run() -> None:
+                    try:
+                        write(SingleFileMessagePackBackingStore(self.path, lock_timeout=10))
+                        outcome.append(None)
+                    except BaseException as exc:  # noqa: BLE001 - the thread hands the outcome to the test
+                        outcome.append(exc)
+
+                writer = threading.Thread(target=run)
+                writer.start()
+                time.sleep(0.3)
+                os.symlink(link_to, self.path)  # after the writer's pre-lock look, while it waits
+                writer.join()
+                holder.wait()
+                self.assertEqual(1, len(outcome))
+                self.assertIsInstance(outcome[0], StoreLinkedError)
+                self.assertTrue(os.path.islink(self.path), "the link was replaced")
+                self.assertEqual(before, target.read_bytes())
+                self.assertFalse(missing.exists(), "the dangling target was created")
+                os.unlink(self.path)  # a link is refused before the lock, so remove it to probe the writer's own lock file
+                with store_write_lock(self.path, timeout=0.2):  # the lock is free
+                    pass
+
     def test_the_lock_is_released_after_each_write(self) -> None:
         store = SingleFileMessagePackBackingStore(self.path, lock_timeout=0.2)
         for key in "abc":

packages/cultcache-rs/CHANGELOG.md

@@ -8,6 +8,7 @@ only in git history.
 
 ### Changed
 
+- The single-file store re-checks for a link once the store lock is held (every write and the try-exchange), so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, as C# does, instead of being replaced. Pinned by a test that puts the link while a second holder owns the lock.
 - `CultCache::is_writable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store.
 
 ### Breaking

packages/cultcache-rs/src/lib.rs

@@ -1752,7 +1752,7 @@ impl SingleFileMessagePackBackingStore {
         let lock = self.open_lock_file()?;
         fs2::FileExt::lock_exclusive(&lock)
             .with_context(|| format!("failed to lock {}", self.lock_path().display()))?;
-        release_lock(lock, action())
+        self.run_locked(lock, action)
     }
 
     /// Replaces the named entries only if each still matches its expectation.
@@ -1870,7 +1870,13 @@ impl SingleFileMessagePackBackingStore {
                     .with_context(|| format!("failed to lock {}", self.lock_path().display()));
             }
         }
-        release_lock(lock, action()).map(Some)
+        self.run_locked(lock, action).map(Some)
+    }
+
+    /// The one tail of both exclusive entries: with the lock held, look for a link again (one may have been put at the
+    /// path while the lock was awaited; the look before the lock saw the path earlier), run the action, release.
+    fn run_locked<T>(&self, lock: File, action: impl FnOnce() -> Result<T>) -> Result<T> {
+        release_lock(lock, refuse_linked(&self.path).and_then(|()| action()))
     }
 
     fn open_lock_file(&self) -> Result<File> {
@@ -8199,6 +8205,46 @@ mod tests {
         Ok(())
     }
 
+    // A link put at the store path while a writer waits on the lock is refused once the lock is taken, as C# does, and the
+    // lock is free afterwards.
+    #[cfg(unix)]
+    #[test]
+    fn a_link_put_while_a_writer_waits_on_the_lock_is_refused_once_it_holds_it() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let volume = temp.path().join("volume");
+        fs::create_dir(&volume)?;
+        let target = volume.join("store.cc");
+        SingleFileMessagePackBackingStore::new(&target).push(&snapshot_envelope("a", b"1"))?;
+        let before = fs::read(&target)?;
+        let dangling = volume.join("missing.cc");
+        let path = temp.path().join("late.cc");
+        let writers: [(&str, fn(&mut SingleFileMessagePackBackingStore) -> Result<()>); 4] = [
+            ("push", |s| s.push(&snapshot_envelope("b", b"2"))),
+            ("push_all", |s| s.push_all(&[snapshot_envelope("b", b"2")], PushAllOptions::default())),
+            ("delete", |s| s.delete(&snapshot_envelope("a", b"1"))),
+            ("compare_exchange_snapshot", |s| s.compare_exchange_snapshot(&[], &[snapshot_envelope("b", b"2")]).map(|_| ())),
+        ];
+        for (name, write, link_to) in writers.into_iter().flat_map(|(n, w)| [(n, w, &target), (n, w, &dangling)]) {
+            let _ = fs::remove_file(&path);
+            let store = SingleFileMessagePackBackingStore::new(&path);
+            let holder = store.open_lock_file()?;
+            fs2::FileExt::lock_exclusive(&holder)?;
+            let mut waiting = SingleFileMessagePackBackingStore::new(&path);
+            let handle = std::thread::spawn(move || write(&mut waiting));
+            std::thread::sleep(std::time::Duration::from_millis(300));
+            std::os::unix::fs::symlink(link_to, &path)?;
+            fs2::FileExt::unlock(&holder)?;
+            let error = handle.join().expect("writer thread").expect_err("the late link must be refused");
+            assert!(error.downcast_ref::<StoreLinkedError>().is_some(), "{name}: {error:#}");
+            assert!(fs::symlink_metadata(&path)?.file_type().is_symlink(), "{name}: the link was replaced");
+            assert_eq!(fs::read(&target)?, before, "{name}: the target changed");
+            assert!(!dangling.exists(), "{name}: the dangling target was created");
+            let probe = store.open_lock_file()?;
+            fs2::FileExt::try_lock_exclusive(&probe).map_err(|e| anyhow!("{name}: lock still held: {e}"))?;
+        }
+        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)]

packages/cultcache-ts/CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to `@gamecult/cultcache-ts` are recorded here, following
 
 ### Changed
 
+- `withStoreWriteLock` looks for a link at the store path again once the lock is held, so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, instead of being replaced. Pinned by a test that puts the link while another holder owns the lock.
 - `CultCache.isWritable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store.
 
 ### Breaking

packages/cultcache-ts/src/store-file.ts

@@ -218,6 +218,8 @@ export async function withStoreWriteLock<T>(
   }
 
   try {
+    // A link may have been put at the path while the lock was awaited; the look above only saw the path before.
+    await refuseLinkedStore(storePath);
     return await operation();
   } finally {
     release();

packages/cultcache-ts/test/store-file.test.ts

@@ -417,6 +417,43 @@ test("a second writer in this process waits for the store lock and runs after re
   }
 });
 
+test("a link put while a writer waits on the store lock is refused once it holds the lock", { skip: process.platform === "win32" }, async () => {
+  const directory = await scratch();
+  try {
+    const target = join(directory, "target.cc");
+    await new SingleFileMessagePackBackingStore(target).push(noteEnvelope("a"));
+    const before = await readFile(target);
+    const writes: Array<[string, (store: SingleFileMessagePackBackingStore) => Promise<void>]> = [
+      ["push", (store) => store.push(noteEnvelope("b"))],
+      ["pushAll", (store) => store.pushAll([noteEnvelope("b")])],
+      ["delete", (store) => store.delete(noteEnvelope("a"))],
+    ];
+    const missing = join(directory, "missing.cc");
+    for (const [name, write, linkTo] of writes.flatMap(([n, w]) => [[n, w, target], [`${n} (dangling)`, w, missing]] as const)) {
+      const link = join(directory, `${name.replace(/\W+/g, "-")}.cc`);
+      let writer: Promise<void> | undefined;
+      await withStoreWriteLock(link, async () => {
+        writer = write(new SingleFileMessagePackBackingStore(link));
+        writer.catch(() => undefined);
+        await sleepFor(150);
+        await fsp.symlink(linkTo, link); // after the writer's pre-lock look, while it waits
+      });
+      await assert.rejects(writer!, (error: unknown) => error instanceof PackageStoreLinkedError, name);
+      assert.ok((await fsp.lstat(link)).isSymbolicLink(), `${name}: the link was replaced`);
+      assert.deepEqual(await readFile(target), before, `${name}: the target changed`);
+      await assert.rejects(fsp.lstat(missing), { code: "ENOENT" }, `${name}: the dangling target was created`);
+      await rm(link); // the link is refused before the lock, so remove it to probe the lock
+      assert.equal(await withStoreWriteLock(link, async () => "free", { timeoutMs: 200 }), "free", `${name}: the lock is held`);
+    }
+  } finally {
+    await rm(directory, { recursive: true, force: true });
+  }
+});
+
+function noteEnvelope(key: string): CultCacheEnvelope {
+  return { key, type: "note", payload: documentPayloadFromBytes(Uint8Array.of(0x91, 1)), storedAt: "2026-10-20T00:00:00Z" };
+}
+
 test("the store lock gives up after its wait bound and names the lock path", async () => {
   const directory = await scratch();
   try {