From a0328066bc2edd36642295e53609f64b2ac7892b Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 05:02:01 +0000 Subject: [PATCH 1/2] Rust, Python, TypeScript single-file stores re-check for a link once the store lock is held (link-late-check-rs-py-ts) Co-Authored-By: Claude Sonnet 5.5 --- packages/cultcache-py/CHANGELOG.md | 1 + .../cultcache-py/src/cultcache_py/stores.py | 2 + .../cultcache-py/tests/test_store_lock.py | 41 ++++++++++++++++++ packages/cultcache-rs/CHANGELOG.md | 1 + packages/cultcache-rs/src/lib.rs | 43 ++++++++++++++++++- packages/cultcache-ts/CHANGELOG.md | 1 + packages/cultcache-ts/src/store-file.ts | 2 + packages/cultcache-ts/test/store-file.test.ts | 35 +++++++++++++++ 8 files changed, 124 insertions(+), 2 deletions(-) diff --git a/packages/cultcache-py/CHANGELOG.md b/packages/cultcache-py/CHANGELOG.md index e5a8b8e8..84d454bf 100644 --- a/packages/cultcache-py/CHANGELOG.md +++ b/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. diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index 36729d44..bc3e5429 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/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() diff --git a/packages/cultcache-py/tests/test_store_lock.py b/packages/cultcache-py/tests/test_store_lock.py index 5ee29aa2..0122f094 100644 --- a/packages/cultcache-py/tests/test_store_lock.py +++ b/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,45 @@ 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"), + } + for name, write in writes.items(): + with self.subTest(name): + 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(target, 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()) + with store_write_lock(target, 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": diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md index 3978e248..327f42aa 100644 --- a/packages/cultcache-rs/CHANGELOG.md +++ b/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 diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index edfc0c91..9bd88fb5 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -1752,7 +1752,8 @@ 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()) + // A link may have been put at the path while the lock was awaited; the refusal above only saw the path before. + release_lock(lock, refuse_linked(&self.path).and_then(|()| action())) } /// Replaces the named entries only if each still matches its expectation. @@ -1870,7 +1871,7 @@ impl SingleFileMessagePackBackingStore { .with_context(|| format!("failed to lock {}", self.lock_path().display())); } } - release_lock(lock, action()).map(Some) + release_lock(lock, refuse_linked(&self.path).and_then(|()| action())).map(Some) } fn open_lock_file(&self) -> Result { @@ -8201,6 +8202,44 @@ mod tests { // 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. + // 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 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) in writers { + 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(&target, &path)?; + fs2::FileExt::unlock(&holder)?; + let error = handle.join().expect("writer thread").expect_err("the late link must be refused"); + assert!(error.downcast_ref::().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"); + let probe = store.open_lock_file()?; + fs2::FileExt::try_lock_exclusive(&probe).map_err(|e| anyhow!("{name}: lock still held: {e}"))?; + } + Ok(()) + } + #[cfg(unix)] #[test] fn a_write_at_a_live_link_is_refused_and_nothing_changes_while_a_read_through_it_works() -> Result<()> { diff --git a/packages/cultcache-ts/CHANGELOG.md b/packages/cultcache-ts/CHANGELOG.md index 5583b45c..c41071a7 100644 --- a/packages/cultcache-ts/CHANGELOG.md +++ b/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 diff --git a/packages/cultcache-ts/src/store-file.ts b/packages/cultcache-ts/src/store-file.ts index e6988d23..0488960e 100644 --- a/packages/cultcache-ts/src/store-file.ts +++ b/packages/cultcache-ts/src/store-file.ts @@ -218,6 +218,8 @@ export async function withStoreWriteLock( } 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(); diff --git a/packages/cultcache-ts/test/store-file.test.ts b/packages/cultcache-ts/test/store-file.test.ts index 485df058..13e1956a 100644 --- a/packages/cultcache-ts/test/store-file.test.ts +++ b/packages/cultcache-ts/test/store-file.test.ts @@ -417,6 +417,41 @@ 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]> = [ + ["push", (store) => store.push(noteEnvelope("b"))], + ["pushAll", (store) => store.pushAll([noteEnvelope("b")])], + ["delete", (store) => store.delete(noteEnvelope("a"))], + ]; + for (const [name, write] of writes) { + const link = join(directory, `${name}.cc`); + let writer: Promise | undefined; + await withStoreWriteLock(link, async () => { + writer = write(new SingleFileMessagePackBackingStore(link)); + writer.catch(() => undefined); + await sleepFor(150); + await fsp.symlink(target, 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 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 { -- 2.47.3 From 5241b8aa98b78a0c76733373f18555f4a3734cae Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 05:25:53 +0000 Subject: [PATCH 2/2] 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 --- .../cultcache-py/tests/test_store_lock.py | 11 ++++++---- packages/cultcache-rs/src/lib.rs | 21 ++++++++++++------- packages/cultcache-ts/test/store-file.test.ts | 8 ++++--- 3 files changed, 26 insertions(+), 14 deletions(-) diff --git a/packages/cultcache-py/tests/test_store_lock.py b/packages/cultcache-py/tests/test_store_lock.py index 0122f094..50b95436 100644 --- a/packages/cultcache-py/tests/test_store_lock.py +++ b/packages/cultcache-py/tests/test_store_lock.py @@ -85,8 +85,9 @@ class StoreLockTests(unittest.TestCase): "push_all": lambda store: store.push_all([envelope("b")]), "delete": lambda store: store.delete("t", "a"), } - for name, write in writes.items(): - with self.subTest(name): + 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)]) @@ -104,14 +105,16 @@ class StoreLockTests(unittest.TestCase): writer = threading.Thread(target=run) writer.start() time.sleep(0.3) - os.symlink(target, self.path) # after the writer's pre-lock look, while it waits + 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()) - with store_write_lock(target, timeout=0.2): # the lock is free + 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: diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 9bd88fb5..9ba800e9 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -1752,8 +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()))?; - // A link may have been put at the path while the lock was awaited; the refusal above only saw the path before. - release_lock(lock, refuse_linked(&self.path).and_then(|()| action())) + self.run_locked(lock, action) } /// Replaces the named entries only if each still matches its expectation. @@ -1871,7 +1870,13 @@ impl SingleFileMessagePackBackingStore { .with_context(|| format!("failed to lock {}", self.lock_path().display())); } } - release_lock(lock, refuse_linked(&self.path).and_then(|()| 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(&self, lock: File, action: impl FnOnce() -> Result) -> Result { + release_lock(lock, refuse_linked(&self.path).and_then(|()| action())) } fn open_lock_file(&self) -> Result { @@ -8200,8 +8205,6 @@ mod tests { 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. // 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)] @@ -8213,6 +8216,7 @@ mod tests { 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"))), @@ -8220,7 +8224,7 @@ mod tests { ("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) in writers { + 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()?; @@ -8228,18 +8232,21 @@ mod tests { 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(&target, &path)?; + 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::().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)] #[test] fn a_write_at_a_live_link_is_refused_and_nothing_changes_while_a_read_through_it_works() -> Result<()> { diff --git a/packages/cultcache-ts/test/store-file.test.ts b/packages/cultcache-ts/test/store-file.test.ts index 13e1956a..7ddf7fe3 100644 --- a/packages/cultcache-ts/test/store-file.test.ts +++ b/packages/cultcache-ts/test/store-file.test.ts @@ -428,18 +428,20 @@ test("a link put while a writer waits on the store lock is refused once it holds ["pushAll", (store) => store.pushAll([noteEnvelope("b")])], ["delete", (store) => store.delete(noteEnvelope("a"))], ]; - for (const [name, write] of writes) { - const link = join(directory, `${name}.cc`); + 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 | undefined; await withStoreWriteLock(link, async () => { writer = write(new SingleFileMessagePackBackingStore(link)); writer.catch(() => undefined); await sleepFor(150); - await fsp.symlink(target, link); // after the writer's pre-lock look, while it waits + 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`); } -- 2.47.3