Rust, Python, TS: a write through a live link at the store path is refused (store-link-rs-py-ts) (#150)
This commit is contained in:
parent
f1e72adf87
commit
3a8de542de
12 changed files with 288 additions and 54 deletions
|
|
@ -13,6 +13,7 @@
|
|||
|
||||
### Breaking
|
||||
|
||||
- A write at a store path that is a link, live or dangling, is refused with the new `StoreLinkedError` (an `OSError`, exported from `cultcache_py`; `.path` names the path) before the lock and the replace, for the single-file store and for `JsonLinesBackingStore` (`push`, `push_all`, `delete`), and nothing is written. Before, a write at a live link replaced the link with a regular file, and at a dangling link both stores raised `FileNotFoundError`. A caller catching `FileNotFoundError` there now catches `StoreLinkedError` (`OSError`); configure the link's target as the store path. On Windows a junction or any reparse point counts as a link. Reads through a live link work.
|
||||
- A record whose payload carries a slot past its reader's declared slots, at any depth (the document, a nested record, a list element, a union arm, a unit arm), is read, held read-only, and every write or removal of its key (`put`, `put_global`, `update`, `put_envelope`, `put_envelopes`, `delete`, `delete_global`) is refused with `SchemaConflictError` before any store is touched (`cultlib-gaps:ruling:extra-trailing-slots`). Reading dropped those slots silently and the next write shed them. `SchemaConflictError.unread_slots` is an `UnreadSlots(type_name, slot_count, declared_count, occurrences, depth0_only)` and the message names the type and both counts, never a slot value. A depth-0 mark is consented, and the record writable, only when the record is stored under an id the document lists in `compatible_schema_ids` and that id's catalog entry has slots past the document's, covering the mark's slot count (`cultlib-gaps:ruling:slot-mark-consent`); a nested mark, or a mark under the document's own id, keeps the record read-only whatever is declared. Missing trailing slots keep their declared defaults. An untouched marked record keeps its stored bytes through every other write. Added `CultCache.is_writable(document, key)` (advisory; the write still decides: it also answers False for a foreign-carried key, for a key held as a type the document does not own, and for a global document at a non-global key; `delete` asks the same judges, so removing a key held as another type is refused with `SchemaConflictError` unless the document owns that record's schema id (its own or one it lists compatible), in which case the removal removes that record, its file row and its lookups, as a write of the key would replace it; it returned silently and removed nothing), `DocumentDefinition.decode_payload_marked`, and `UnreadSlots`. A document with a custom `decode` (no slot codec) never marks. A store holding a variant record is still unreadable here. Implies a minor bump while the version is 0.x (0.3.0 to 0.4.0): code that wrote back a record from a newer writer now gets `SchemaConflictError`.
|
||||
- A payload is one MessagePack value: `DocumentDefinition.decode_payload` refuses bytes after it, by count, with `ValueError` `<type> payload has N bytes after its first MessagePack value (refused on read).` (never the bytes, never the depth refusal; a payload that ends early stays the decoder's), as the C# reference does (`tests/vectors/union-parity/bytes-after-outcomes.msgpack`). It raised `msgpack.ExtraData` (a `ValueError`, so only a caller naming `msgpack.ExtraData` is affected); this is a minor-version bump while pre-1.0. The refused set is unchanged. Cache loads, `get` and every definition built on the MessagePack decoder go through it; a user `payload_decoder` that raises `ExtraData` is refused the same way. Depth is judged first, so a value nested past 256 with a tail is the depth refusal, as in C#.
|
||||
- A cache holds one record per key whatever its type (`variants:ruling:cache-refuses-retype`). A `put`, `put_envelope`, `put_envelopes` or `put_global` of a second type under a key the cache holds, or under a key a store carries foreign, is refused with `SchemaConflictError` naming the key, even with no store, and changes nothing; retype a key by deleting it, then putting the new type. A write may replace another registered type's record only when the written type owns its schema id (declares it or lists it compatible) and both types share one home store. A load that reads one key from two stores, whatever the types, is refused with `CultCacheError` naming the key and both types, and admits nothing. A store that holds a record no registered type claims now loads and is carried: the cache does not serve it, the store keeps its bytes through every write, `delete` or `put` at its key is refused with `SchemaConflictError`, and registering a type that declares its schema id claims it at the next load. It raised `CultCacheError` "Unknown persisted document type". Two global types share the key `__global__`, so a second global type in one cache is now refused, and a store that already holds two global types at `__global__` (a JSONL store written by caches that never pulled could produce one) no longer loads: the pull raises `CultCacheError` and admits nothing. C# refuses a second persisted global on pull the same way (`cultcache-store-composition.md`, "Globals never invented"); repair such a file by deleting one of the two records with the runtime that wrote it.
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ from .documents import (
|
|||
read_payload_tree,
|
||||
)
|
||||
from .leaves import SlotCodec
|
||||
from .stores import JsonLinesBackingStore, SchemaConflictError, SingleFileMessagePackBackingStore, StoreUnreadableError
|
||||
from .stores import JsonLinesBackingStore, SchemaConflictError, SingleFileMessagePackBackingStore, StoreLinkedError, StoreUnreadableError
|
||||
|
||||
__all__ = [
|
||||
"BackingStore",
|
||||
|
|
@ -38,6 +38,7 @@ __all__ = [
|
|||
"SchemaConflictError",
|
||||
"SingleFileMessagePackBackingStore",
|
||||
"SlotCodec",
|
||||
"StoreLinkedError",
|
||||
"StoreUnreadableError",
|
||||
"UnionArm",
|
||||
"UnreadSlots",
|
||||
|
|
|
|||
|
|
@ -101,22 +101,24 @@ class JsonLinesBackingStore:
|
|||
return envelopes
|
||||
|
||||
def push(self, envelope: CultCacheEnvelope) -> None:
|
||||
with self._lock:
|
||||
existing = {(item.type, item.key): item for item in self.pull_all()}
|
||||
existing[(envelope.type, envelope.key)] = envelope
|
||||
self._replace_all(list(existing.values()))
|
||||
self.push_all([envelope])
|
||||
|
||||
def push_all(self, envelopes: list[CultCacheEnvelope]) -> None:
|
||||
with self._lock:
|
||||
existing = {(item.type, item.key): item for item in self.pull_all()}
|
||||
def merge(existing: dict[tuple[str, str], CultCacheEnvelope]) -> None:
|
||||
for envelope in envelopes:
|
||||
existing[(envelope.type, envelope.key)] = envelope
|
||||
self._replace_all(list(existing.values()))
|
||||
self._rewrite(merge)
|
||||
|
||||
def delete(self, type: str, key: str) -> None:
|
||||
self._rewrite(lambda existing: existing.pop((type, key), None))
|
||||
|
||||
def _rewrite(self, change: Callable[[dict[tuple[str, str], CultCacheEnvelope]], object]) -> None:
|
||||
"""The one entry of every write: refuses a link at the path, then reads the file, applies the change and replaces it."""
|
||||
with self._lock:
|
||||
existing = [item for item in self.pull_all() if not (item.type == type and item.key == key)]
|
||||
self._replace_all(existing)
|
||||
_refuse_linked(self.path)
|
||||
existing = {(item.type, item.key): item for item in self.pull_all()}
|
||||
change(existing)
|
||||
self._replace_all(list(existing.values()))
|
||||
|
||||
def _replace_all(self, envelopes: list[CultCacheEnvelope]) -> None:
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
|
@ -363,7 +365,8 @@ class ReplaceKernel:
|
|||
|
||||
def _replace_store(staged: str | os.PathLike[str], destination: str | os.PathLike[str]) -> None:
|
||||
"""Replaces `destination` with `staged`: POSIX-semantics rename first on Windows (so a reader holding the store open
|
||||
never sees it absent), os.replace elsewhere."""
|
||||
keeps it and rarely sees it absent; a stalled rename can still leave it absent for 5-47 ms on NTFS,
|
||||
variants:ruling:posix-rename-absent-window-document), os.replace elsewhere."""
|
||||
if sys.platform == "win32":
|
||||
replace_store_windows(os.fspath(staged), os.fspath(destination), WindowsReplaceKernel())
|
||||
else:
|
||||
|
|
@ -440,11 +443,33 @@ class WindowsReplaceKernel(ReplaceKernel):
|
|||
return 0 if self._k.MoveFileExW(staged, destination, flags) else self._ctypes.get_last_error()
|
||||
|
||||
|
||||
class StoreLinkedError(OSError):
|
||||
"""A write was asked of a store path that is a link: a replace there would swap the link for a regular file."""
|
||||
|
||||
def __init__(self, path: Path) -> None:
|
||||
super().__init__(f"CultCache store {path} is a link, and a write at it would replace the link; configure the link's target as the store path.")
|
||||
self.path = path
|
||||
|
||||
|
||||
def _refuse_linked(path: Path) -> None:
|
||||
"""Refuses a link at the store path: a file or directory link, and on Windows a junction or any reparse point (islink misses
|
||||
junctions before Python 3.12). Absent is not a link."""
|
||||
linked = os.path.islink(path)
|
||||
if not linked and sys.platform == "win32":
|
||||
try:
|
||||
linked = bool(os.lstat(path).st_file_attributes & 0x400) # FILE_ATTRIBUTE_REPARSE_POINT
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
if linked:
|
||||
raise StoreLinkedError(path)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def store_write_lock(path: Path, timeout: float = LOCK_TIMEOUT_SECONDS) -> Iterator[None]:
|
||||
"""Holds the exclusive store lock on `<path>.lock`, the lock the C#, Rust and TypeScript runtimes take: flock on Unix,
|
||||
LockFileEx over the whole file on Windows, polled every 10 ms with no handle held between polls. The lock file is
|
||||
created if absent and never written or deleted."""
|
||||
_refuse_linked(path)
|
||||
lock_path = path.with_name(path.name + ".lock")
|
||||
deadline = time.monotonic() + timeout
|
||||
while True:
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ from cultcache_py import (
|
|||
CultCache,
|
||||
JsonLinesBackingStore,
|
||||
SingleFileMessagePackBackingStore,
|
||||
StoreLinkedError,
|
||||
StoreUnreadableError,
|
||||
define_database_entry_type,
|
||||
define_document_type,
|
||||
|
|
@ -1431,10 +1432,10 @@ class CultCacheTests(unittest.TestCase):
|
|||
self.assertNotIn("SECRET", str(caught.exception))
|
||||
self.assertIn("not a valid store", str(caught.exception))
|
||||
|
||||
# What a write does through a symbolic link today. R3 decides whether a write resolves
|
||||
# the link or refuses it; this pins the current behaviour so that change is deliberate.
|
||||
# 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.
|
||||
@unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
|
||||
def test_r3_decides_resolve_or_refuse_a_push_through_a_live_link_replaces_the_link(self) -> None:
|
||||
def test_a_write_at_a_live_link_is_refused_and_nothing_changes_while_a_read_through_it_works(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
volume = Path(tmp) / "volume"
|
||||
volume.mkdir()
|
||||
|
|
@ -1442,12 +1443,65 @@ class CultCacheTests(unittest.TestCase):
|
|||
envelope = CultCacheEnvelope(key="a", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z")
|
||||
SingleFileMessagePackBackingStore(target).push(envelope)
|
||||
before = target.read_bytes()
|
||||
link = Path(tmp) / "store.cc"
|
||||
volume_before = sorted(p.name for p in volume.iterdir())
|
||||
link = Path(tmp) / "link.cc"
|
||||
link.symlink_to(target)
|
||||
SingleFileMessagePackBackingStore(link).push(replace(envelope, key="b"))
|
||||
self.assertFalse(link.is_symlink())
|
||||
self.assertTrue(link.is_file())
|
||||
store = SingleFileMessagePackBackingStore(link)
|
||||
for write in (
|
||||
lambda: store.push(replace(envelope, key="b")),
|
||||
lambda: store.push_all([replace(envelope, key="b")]),
|
||||
lambda: store.delete("t", "a"),
|
||||
):
|
||||
with self.assertRaises(StoreLinkedError) as caught:
|
||||
write()
|
||||
self.assertIn(str(link), str(caught.exception))
|
||||
self.assertTrue(link.is_symlink())
|
||||
self.assertEqual(target.read_bytes(), before)
|
||||
self.assertEqual(sorted(p.name for p in Path(tmp).iterdir()), ["link.cc", "volume"])
|
||||
self.assertEqual(sorted(p.name for p in volume.iterdir()), volume_before)
|
||||
self.assertEqual([item.key for item in store.pull_all()], ["a"])
|
||||
|
||||
@unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
|
||||
def test_a_json_lines_write_at_a_live_link_is_refused_and_the_link_survives(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
volume = Path(tmp) / "volume"
|
||||
volume.mkdir()
|
||||
target = volume / "store.jsonl"
|
||||
envelope = CultCacheEnvelope(key="a", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z")
|
||||
JsonLinesBackingStore(target).push(envelope)
|
||||
before = target.read_bytes()
|
||||
link = Path(tmp) / "link.jsonl"
|
||||
link.symlink_to(target)
|
||||
store = JsonLinesBackingStore(link)
|
||||
for write in (
|
||||
lambda: store.push(replace(envelope, key="b")),
|
||||
lambda: store.push_all([replace(envelope, key="b")]),
|
||||
lambda: store.delete("t", "a"),
|
||||
):
|
||||
with self.assertRaises(StoreLinkedError):
|
||||
write()
|
||||
self.assertTrue(link.is_symlink())
|
||||
self.assertEqual(target.read_bytes(), before)
|
||||
self.assertEqual(sorted(p.name for p in Path(tmp).iterdir()), ["link.jsonl", "volume"])
|
||||
self.assertEqual(sorted(p.name for p in volume.iterdir()), ["store.jsonl"])
|
||||
|
||||
def test_store_linked_error_is_an_os_error_carrying_the_path(self) -> None:
|
||||
error = StoreLinkedError(Path("some/store.cc"))
|
||||
self.assertIsInstance(error, OSError)
|
||||
self.assertEqual(error.path, Path("some/store.cc"))
|
||||
|
||||
@unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
|
||||
def test_a_write_at_a_live_directory_link_is_refused_as_a_link(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
real = Path(tmp) / "real"
|
||||
real.mkdir()
|
||||
link = Path(tmp) / "dirlink"
|
||||
link.symlink_to(real, target_is_directory=True)
|
||||
with self.assertRaises(StoreLinkedError):
|
||||
SingleFileMessagePackBackingStore(link).push(
|
||||
CultCacheEnvelope(key="a", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z"))
|
||||
self.assertTrue(link.is_symlink())
|
||||
self.assertEqual(list(real.iterdir()), [])
|
||||
|
||||
@unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
|
||||
def test_a_dangling_symlink_at_a_store_path_is_an_os_error_and_nothing_is_written(self) -> None:
|
||||
|
|
@ -1460,7 +1514,7 @@ class CultCacheTests(unittest.TestCase):
|
|||
store = store_type(path)
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
store.pull_all()
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
with self.assertRaises(StoreLinkedError):
|
||||
store.push(CultCacheEnvelope(key="k", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z"))
|
||||
self.assertTrue(path.is_symlink())
|
||||
self.assertEqual(list(volume.iterdir()), [])
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ only in git history.
|
|||
|
||||
Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0).
|
||||
|
||||
- A write at a store path that is a symbolic link, live or dangling, is refused with the new public `StoreLinkedError` (`error.downcast_ref::<StoreLinkedError>()`) before the lock and the replace, and writes nothing: `push`, `push_all`, `delete`, the exchanges, and a soft `push_all`, which used to skip a link and now refuses. Before, a write at a live link replaced the link with a regular file and a dangling link failed with `io::ErrorKind::NotFound`. The redb stores (`RedbMessagePackBackingStore`, `OwnedRedbMessagePackBackingStore`) are exempt: they are ruled migrated and deleted (`paged-store-redb-migrate-and-delete`) and still open a live link's target. Configure the link's target as the store path. Reads through a live link work.
|
||||
- A record whose payload carries unread slots is read-only. A payload with a slot past its type's declared slots, at any depth (record, nested record, list element, union arm, unit arm), still decodes, and every write or removal of its key is now refused before any store is touched: `put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch` (the whole batch) and `delete`, so `update` too. The refusal is `SchemaConflictError` with the new `unread_slots`; its message names the type and both slot counts and never a slot value. Holders of such a record used to overwrite it and drop the extra slots silently. The payload nesting refusal still runs first. A held payload this build cannot read at all (an arm it lacks from a newer writer, or a corrupt record) is read-only too: every write or removal is refused with an error naming the type and key (a newer writer's record is read with that build; a corrupt one is restored from a backup or removed at the store itself, as no cache call removes it), and `is_writable` answers false. `load_soa` is a write of every held key it omits or changes, so it refuses to drop or replace a read-only record. `is_writable` also answers false for a key a store carries as a foreign record and for a type other than the one holding the key, as the writes do. Each write over a held key decodes the held payload once to ask (about 40% of a `put` of a 200,000-element record, measured in release); keys nothing holds pay nothing.
|
||||
- cultmesh-rs `pull_rudp_catalog_snapshot` applies a snapshot as one cache commit (`put_prepared_batch`), as C# does: a snapshot document for a key the node holds read-only, or any other refusal, applies nothing. It used to apply document by document, so a refusal ended the pull with the earlier documents kept. A snapshot whose types route to different backing stores is now refused whole, as a batch cannot span stores. C# `CultCache.Commit` refuses a cross-store commit the same way (`The batch spans A and B; a commit lands in one home store`), so the two agree; a test pins it.
|
||||
- `SchemaConflictError` gains the public field `unread_slots: Option<UnreadSlots>`; code that builds it with a struct literal must set it (`None`).
|
||||
|
|
|
|||
|
|
@ -384,9 +384,10 @@ an empty store. Any other failure to reach it, such as a symlink loop, a file
|
|||
where a parent directory should be, or a missing permission, is an I/O error.
|
||||
A store file that exists and holds zero bytes is `Undecodable`: no CultCache
|
||||
writer leaves one. A symbolic link whose target is gone is not a missing store:
|
||||
reading it is an I/O error, so a read-then-write (`push`, `push_all`, `delete`, the exchanges)
|
||||
fails before writing. Writes do not yet follow links: a write through a live link
|
||||
replaces the link with a regular file and leaves its target alone. A refusal names no
|
||||
reading it is an I/O error. A write at a store path that is a link, live or dangling, is
|
||||
refused with `StoreLinkedError` (`error.downcast_ref::<StoreLinkedError>()`) before the
|
||||
lock and the replace, so no write swaps a link for a regular file; configure the link's
|
||||
target as the store path. Reads through a live link work. A refusal names no
|
||||
value read from the file; it names a record only by its key and schema id, and
|
||||
a header only when it has the shape `cultcache.store.v<digits>`.
|
||||
|
||||
|
|
|
|||
|
|
@ -1176,6 +1176,35 @@ impl std::fmt::Display for CultCacheStoreUnreadable {
|
|||
|
||||
impl std::error::Error for CultCacheStoreUnreadable {}
|
||||
|
||||
/// A write was asked of a store path that is a link (a file or directory symbolic link, or a Windows
|
||||
/// junction): a replace there would swap the link for a regular file and detach the store from its
|
||||
/// target. Nothing was locked or written. Configure the link's target as the store path. Reads through
|
||||
/// a live link are unaffected. Find it with `error.downcast_ref::<StoreLinkedError>()`.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct StoreLinkedError {
|
||||
pub path: PathBuf,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for StoreLinkedError {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(
|
||||
formatter,
|
||||
"CultCache store {} is a link, and a write at it would replace the link; configure the link's target as the store path",
|
||||
self.path.display()
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for StoreLinkedError {}
|
||||
|
||||
/// Refuses a write at a link. A path that cannot be inspected is left to the open that follows.
|
||||
fn refuse_linked(path: &Path) -> Result<()> {
|
||||
match fs::symlink_metadata(path) {
|
||||
Ok(metadata) if metadata.file_type().is_symlink() => Err(StoreLinkedError { path: path.to_path_buf() }.into()),
|
||||
_ => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// A write to a single-file store that failed, and what it left on disk. It sits in the
|
||||
/// error chain of every such failure, above the cause, so a consumer tells the outcomes
|
||||
/// apart with `error.downcast_ref::<CultCacheStoreWriteFailed>()`.
|
||||
|
|
@ -1719,6 +1748,7 @@ impl SingleFileMessagePackBackingStore {
|
|||
}
|
||||
|
||||
fn with_exclusive_lock<T>(&self, action: impl FnOnce() -> Result<T>) -> Result<T> {
|
||||
refuse_linked(&self.path)?;
|
||||
let lock = self.open_lock_file()?;
|
||||
fs2::FileExt::lock_exclusive(&lock)
|
||||
.with_context(|| format!("failed to lock {}", self.lock_path().display()))?;
|
||||
|
|
@ -1830,6 +1860,7 @@ impl SingleFileMessagePackBackingStore {
|
|||
}
|
||||
|
||||
fn try_with_exclusive_lock<T>(&self, action: impl FnOnce() -> Result<T>) -> Result<Option<T>> {
|
||||
refuse_linked(&self.path)?;
|
||||
let lock = self.open_lock_file()?;
|
||||
match fs2::FileExt::try_lock_exclusive(&lock) {
|
||||
Ok(()) => {}
|
||||
|
|
@ -5374,12 +5405,6 @@ mod tests {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
fn io_not_found(error: &anyhow::Error) -> bool {
|
||||
error
|
||||
.chain()
|
||||
.any(|cause| cause.downcast_ref::<std::io::Error>().is_some_and(|io| io.kind() == std::io::ErrorKind::NotFound))
|
||||
}
|
||||
|
||||
fn test_envelope(r#type: &str, key: &str, payload: &[u8]) -> CultCacheEnvelope {
|
||||
CultCacheEnvelope {
|
||||
key: key.to_string(),
|
||||
|
|
@ -7314,7 +7339,14 @@ mod tests {
|
|||
let temp = tempfile::tempdir()?;
|
||||
let path = temp.path().join("store.cc");
|
||||
std::os::unix::fs::symlink(&path, &path)?;
|
||||
assert_io_failure_on_every_read(&SingleFileMessagePackBackingStore::new(&path));
|
||||
let [pull, snapshot, exchange] = every_read(&SingleFileMessagePackBackingStore::new(&path));
|
||||
for error in [pull.unwrap_err(), snapshot.unwrap_err()] {
|
||||
let io = io_error_in(&error).expect("the failure to reach the store is an io::Error");
|
||||
assert_ne!(io.kind(), std::io::ErrorKind::NotFound, "{error:#}");
|
||||
assert_eq!(unreadable_kind(&error), None, "{error:#}");
|
||||
}
|
||||
// The exchange is a write: a link at the path is refused before it reads.
|
||||
assert!(exchange.unwrap_err().downcast_ref::<StoreLinkedError>().is_some());
|
||||
assert!(fs::symlink_metadata(&path)?.file_type().is_symlink());
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -7753,12 +7785,13 @@ mod tests {
|
|||
let path = temp.path().join("store.cc");
|
||||
std::os::unix::fs::symlink(&target, &path)?;
|
||||
let store = SingleFileMessagePackBackingStore::new(&path);
|
||||
for result in every_read(&store) {
|
||||
let error = result.expect_err("a dangling link must not read as an empty store");
|
||||
let [pull, snapshot, exchange] = every_read(&store);
|
||||
for error in [pull.unwrap_err(), snapshot.unwrap_err()] {
|
||||
assert!(io_error_in(&error).is_some(), "{error:#}");
|
||||
assert_eq!(unreadable_kind(&error), None, "{error:#}");
|
||||
}
|
||||
// Every merging writer refuses on its own read, so none can replace the link.
|
||||
assert!(exchange.unwrap_err().downcast_ref::<StoreLinkedError>().is_some());
|
||||
// Every writer refuses the link before its lock, so none can replace it.
|
||||
let mut writer = SingleFileMessagePackBackingStore::new(&path);
|
||||
let delta = snapshot_envelope("delta", b"4");
|
||||
let writes = [
|
||||
|
|
@ -7770,7 +7803,7 @@ mod tests {
|
|||
];
|
||||
for (index, result) in writes.into_iter().enumerate() {
|
||||
let error = result.expect_err("a merging write through a dangling link must refuse");
|
||||
assert!(io_error_in(&error).is_some(), "write {index}: {error:#}");
|
||||
assert!(error.downcast_ref::<StoreLinkedError>().is_some(), "write {index}: {error:#}");
|
||||
assert_eq!(write_failure_kind(&error), None, "write {index}: {error:#}");
|
||||
}
|
||||
assert!(fs::symlink_metadata(&path)?.file_type().is_symlink());
|
||||
|
|
@ -7909,27 +7942,65 @@ mod tests {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
// What a write does through a symbolic link today. R3 decides whether a write
|
||||
// resolves the link or refuses it; until then these pin the current behaviour so the
|
||||
// change is deliberate. Reads refuse a dangling link (see the test above).
|
||||
// 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 r3_decides_resolve_or_refuse_a_push_through_a_live_link_replaces_the_link() -> Result<()> {
|
||||
fn a_write_at_a_live_link_is_refused_and_nothing_changes_while_a_read_through_it_works() -> 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 link = temp.path().join("store.cc");
|
||||
let volume_before = dir_names(&volume)?;
|
||||
let link = temp.path().join("link.cc");
|
||||
std::os::unix::fs::symlink(&target, &link)?;
|
||||
|
||||
SingleFileMessagePackBackingStore::new(&link).push(&snapshot_envelope("b", b"2"))?;
|
||||
assert!(fs::symlink_metadata(&link)?.file_type().is_file(), "the link was replaced by a regular file");
|
||||
assert_eq!(fs::read(&target)?, before, "the store the link pointed to is unchanged");
|
||||
let mut store = SingleFileMessagePackBackingStore::new(&link);
|
||||
let errors = [
|
||||
store.push(&snapshot_envelope("b", b"2")).unwrap_err(),
|
||||
store.push_all(&[snapshot_envelope("b", b"2")], PushAllOptions::default()).unwrap_err(),
|
||||
store.push_all(&[snapshot_envelope("b", b"2")], PushAllOptions { soft: true }).unwrap_err(),
|
||||
store.delete(&snapshot_envelope("a", b"1")).unwrap_err(),
|
||||
store.compare_exchange_snapshot(&[], &[snapshot_envelope("b", b"2")]).unwrap_err(),
|
||||
store.try_compare_exchange_snapshot(&[], &[snapshot_envelope("b", b"2")]).err().expect("refused"),
|
||||
];
|
||||
for error in &errors {
|
||||
let linked = error.downcast_ref::<StoreLinkedError>().unwrap_or_else(|| panic!("{error:#}"));
|
||||
assert_eq!(linked.path, link);
|
||||
assert!(error.to_string().contains(&link.display().to_string()), "{error:#}");
|
||||
}
|
||||
assert!(fs::symlink_metadata(&link)?.file_type().is_symlink(), "the link is left as it was");
|
||||
assert_eq!(fs::read(&target)?, before);
|
||||
assert_eq!(dir_names(&volume)?, volume_before, "no lock or staging file beside the target");
|
||||
assert_eq!(dir_names(temp.path())?, ["link.cc", "volume"], "no lock or staging file beside the link");
|
||||
assert_eq!(store.pull_all()?, vec![snapshot_envelope("a", b"1")]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn a_write_at_a_live_directory_link_is_refused_as_a_link() -> Result<()> {
|
||||
let temp = tempfile::tempdir()?;
|
||||
let real = temp.path().join("real");
|
||||
fs::create_dir(&real)?;
|
||||
let link = temp.path().join("dirlink");
|
||||
std::os::unix::fs::symlink(&real, &link)?;
|
||||
let error = SingleFileMessagePackBackingStore::new(&link).push(&snapshot_envelope("a", b"1")).unwrap_err();
|
||||
assert!(error.downcast_ref::<StoreLinkedError>().is_some(), "{error:#}");
|
||||
assert!(fs::symlink_metadata(&link)?.file_type().is_symlink());
|
||||
assert_eq!(fs::read_dir(&real)?.count(), 0);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn dir_names(dir: &Path) -> Result<Vec<String>> {
|
||||
let mut names = fs::read_dir(dir)?.map(|entry| Ok(entry?.file_name().to_string_lossy().into_owned())).collect::<Result<Vec<_>>>()?;
|
||||
names.sort();
|
||||
Ok(names)
|
||||
}
|
||||
|
||||
// The two cases the vectors cannot hold: nothing at the path is written, and a dangling link is refused and left as it is.
|
||||
#[test]
|
||||
fn a_soft_push_all_writes_where_nothing_is_and_refuses_a_dangling_link() -> Result<()> {
|
||||
|
|
@ -7951,7 +8022,7 @@ mod tests {
|
|||
let error = SingleFileMessagePackBackingStore::new(&link)
|
||||
.push_all(&[snapshot_envelope("a", b"1")], PushAllOptions { soft: true })
|
||||
.unwrap_err();
|
||||
assert!(io_not_found(&error), "{error:#}");
|
||||
assert!(error.downcast_ref::<StoreLinkedError>().is_some(), "{error:#}");
|
||||
assert!(fs::symlink_metadata(&link)?.file_type().is_symlink(), "the link was replaced");
|
||||
assert_eq!(fs::read_dir(&volume)?.count(), 0, "nothing reached the link's volume");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ All notable changes to `@gamecult/cultcache-ts` are recorded here, following
|
|||
|
||||
### Breaking
|
||||
|
||||
- `SingleFileMessagePackBackingStore.pushAll` through a dangling symbolic link at the store path is refused with `ENOENT`, as `push` and `pullAll` are, and writes nothing; it used to replace the link with a new store. The reader owns the verdict on what is at the path.
|
||||
- A write at a single-file store path that is a symbolic link, live or dangling, is refused with the new `StoreLinkedError` (exported from the package entry; `error.path` names the path) before the lock and the replace, and writes nothing: `push`, `pushAll`, `delete` and a soft flush alike. Before, a write at a live link replaced the link with a regular file, a dangling link failed with `ENOENT` (or, for `pushAll`, was replaced), and a soft flush at a link skipped. A caller catching `ENOENT` around a write at a dangling link now catches `StoreLinkedError`; configure the link's target as the store path. Reads through a live link work. An `lstat` failure other than `ENOENT` (`ENOTDIR`, `EACCES`, `ELOOP`) is rethrown as it is.
|
||||
- `SingleFileMessagePackBackingStore.pushAll` writes the given entries onto the store as it is under the lock, as `push` does: a record under the same key is replaced and no other is removed. It used to replace the whole file with the entries, so `CultMesh.flush()` from a cache that held less than the store (a node created with `pullOnStart: false`, or one whose image was stale) wiped the records it never read. A caller that relied on `pushAll` to reset a store removes the file first. A soft flush is unchanged. `CultCache.storedSnapshot()` is the image as the stores hold it (a global loaded from a legacy key under that key) and is what `CultMesh.flush` persists, so a flush no longer leaves such a global under two keys; `snapshot()` is unchanged.
|
||||
- A single-file store marked `cultcache.store.v3` (it holds an element id, written by the C# runtime) is read, and a rewrite keeps the mark and the ids; this runtime does not decide when a document carries ids. A runtime from before this change refuses a v3 store by name, so rolling back to an older release after a store has been written as v3 is one-way: there is no carry-back, so restore from a copy taken before upgrading.
|
||||
- `indexes` are unique, as C# `[CultIndex]`: a write or a load that leaves two records of a type holding one non-blank
|
||||
|
|
|
|||
|
|
@ -1,2 +1,3 @@
|
|||
export * from "./browser";
|
||||
export { SingleFileMessagePackBackingStore, readSingleFileStore, type SingleFileStoreRead } from "./single-file-messagepack-backing-store";
|
||||
export { StoreLinkedError } from "./store-file";
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@
|
|||
// lazily inside `loadKernel`, so a consumer that never writes a store never loads it.
|
||||
|
||||
import { closeSync, openSync, constants as fsConstants } from "node:fs";
|
||||
import { mkdir, open, rename, rm } from "node:fs/promises";
|
||||
import { lstat, mkdir, open, rename, rm } from "node:fs/promises";
|
||||
import type { FileHandle } from "node:fs/promises";
|
||||
import { dirname } from "node:path";
|
||||
|
||||
|
|
@ -25,6 +25,30 @@ export class StoreReplacedNotDurableError extends Error {
|
|||
}
|
||||
}
|
||||
|
||||
/** Thrown when a write is asked of a store path that is a link: a replace there would swap the link for a regular file. */
|
||||
export class StoreLinkedError extends Error {
|
||||
constructor(readonly path: string) {
|
||||
super(`CultCache store ${path} is a link, and a write at it would replace the link; configure the link's target as the store path.`);
|
||||
this.name = "StoreLinkedError";
|
||||
}
|
||||
}
|
||||
|
||||
/** Refuses a link at the store path (file or directory link; libuv reports Windows junctions as links). Absent is not a link. */
|
||||
export async function refuseLinkedStore(storePath: string): Promise<void> {
|
||||
let linked = false;
|
||||
try {
|
||||
linked = (await lstat(storePath)).isSymbolicLink();
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
if (linked) {
|
||||
throw new StoreLinkedError(storePath);
|
||||
}
|
||||
}
|
||||
|
||||
/** The file operations of one commit. Tests record or fail them; production uses `nodeStoreFileOps`. */
|
||||
export interface StoreFileOps {
|
||||
createExclusive(path: string): Promise<FileHandle>;
|
||||
|
|
@ -177,6 +201,7 @@ export async function withStoreWriteLock<T>(
|
|||
): Promise<T> {
|
||||
const lockPath = `${storePath}.lock`;
|
||||
const timeoutMs = options.timeoutMs ?? LOCK_TIMEOUT_MS;
|
||||
await refuseLinkedStore(storePath);
|
||||
await mkdir(dirname(lockPath), { recursive: true });
|
||||
|
||||
const started = Date.now();
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ import { CultCache } from "../src/cult-cache";
|
|||
import { inspectCultCacheBytes } from "../src/cult-cache-inspector";
|
||||
import { defineDocumentRegistry, defineDocumentType, documentPayloadBytes, documentPayloadFromBytes } from "../src/document";
|
||||
import { SingleFileMessagePackBackingStore, readSingleFileStore, readStoreBytes } from "../src/single-file-messagepack-backing-store";
|
||||
import { StoreLinkedError, withStoreWriteLock } from "../src/store-file";
|
||||
import { SchemaConflictError, StoreUnreadableError, refuseUnreadStoreBytes } from "../src/store-format";
|
||||
import type { AnyCultCacheDocumentDefinition, CacheBackingStore, CultCacheEnvelope, CultCacheSchema } from "../src/types";
|
||||
|
||||
|
|
@ -2466,8 +2467,8 @@ test("SingleFileMessagePackBackingStore refuses a dangling symlink at its path a
|
|||
() => store.pullAll(),
|
||||
(error: NodeJS.ErrnoException) => error.code === "ENOENT" && error.message.includes("symbolic link") && error.message.includes(file),
|
||||
);
|
||||
await assert.rejects(() => store.push(envelope), (error: NodeJS.ErrnoException) => error.code === "ENOENT");
|
||||
await assert.rejects(() => store.pushAll([envelope], { soft: true }), (error: NodeJS.ErrnoException) => error.code === "ENOENT");
|
||||
await assert.rejects(() => store.push(envelope), (error: unknown) => error instanceof StoreLinkedError);
|
||||
await assert.rejects(() => store.pushAll([envelope], { soft: true }), (error: unknown) => error instanceof StoreLinkedError);
|
||||
assert.ok((await lstat(file)).isSymbolicLink());
|
||||
assert.deepEqual(await readdir(volume), []);
|
||||
});
|
||||
|
|
@ -2508,9 +2509,9 @@ test("SingleFileMessagePackBackingStore record refusal names only a string key",
|
|||
}
|
||||
});
|
||||
|
||||
// What a write does through a symbolic link today. R3 decides whether a write resolves the link
|
||||
// or refuses it; these pin the current behaviour so that change is deliberate.
|
||||
test("R3 decides resolve-or-refuse: a push through a live link replaces the link", { skip: process.platform === "win32" }, async () => {
|
||||
// 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.
|
||||
test("a write at a live link is refused and nothing changes, while a read through it still works", { skip: process.platform === "win32" }, async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "cultcache-live-link-"));
|
||||
const volume = join(dir, "volume");
|
||||
await mkdir(volume);
|
||||
|
|
@ -2518,11 +2519,31 @@ test("R3 decides resolve-or-refuse: a push through a live link replaces the link
|
|||
const envelope = { key: "a", type: "t", payload: documentPayloadFromBytes(Uint8Array.of(0x90)), storedAt: "2026-09-30T00:00:00Z" };
|
||||
await new SingleFileMessagePackBackingStore(target).push(envelope);
|
||||
const before = await readFile(target);
|
||||
const link = join(dir, "store.cc");
|
||||
const volumeBefore = (await readdir(volume)).sort();
|
||||
const link = join(dir, "link.cc");
|
||||
await symlink(target, link);
|
||||
await new SingleFileMessagePackBackingStore(link).push({ ...envelope, key: "b" });
|
||||
assert.ok((await lstat(link)).isFile());
|
||||
const store = new SingleFileMessagePackBackingStore(link);
|
||||
const refused = (error: unknown) => error instanceof StoreLinkedError && error.path === link && error.message.includes(link);
|
||||
await assert.rejects(() => store.push({ ...envelope, key: "b" }), refused);
|
||||
await assert.rejects(() => store.pushAll([{ ...envelope, key: "b" }]), refused);
|
||||
await assert.rejects(() => store.delete(envelope), refused);
|
||||
assert.ok((await lstat(link)).isSymbolicLink());
|
||||
assert.deepEqual(await readFile(target), before);
|
||||
assert.deepEqual((await readdir(dir)).sort(), ["link.cc", "volume"]);
|
||||
assert.deepEqual((await readdir(volume)).sort(), volumeBefore);
|
||||
assert.deepEqual((await store.pullAll()).map((e) => e.key), ["a"]);
|
||||
});
|
||||
|
||||
test("a write at a live directory link is refused as a link", { skip: process.platform === "win32" }, async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "cultcache-dir-link-"));
|
||||
const realDir = join(dir, "real");
|
||||
await mkdir(realDir);
|
||||
const link = join(dir, "dirlink");
|
||||
await symlink(realDir, link);
|
||||
const envelope = { key: "a", type: "t", payload: documentPayloadFromBytes(Uint8Array.of(0x90)), storedAt: "2026-09-30T00:00:00Z" };
|
||||
await assert.rejects(() => new SingleFileMessagePackBackingStore(link).push(envelope), (error: unknown) => error instanceof StoreLinkedError);
|
||||
assert.ok((await lstat(link)).isSymbolicLink());
|
||||
assert.deepEqual(await readdir(realDir), []);
|
||||
});
|
||||
|
||||
test("a pushAll through a dangling link is refused like push, and the link and its volume are left as they were", { skip: process.platform === "win32" }, async () => {
|
||||
|
|
@ -2532,7 +2553,7 @@ test("a pushAll through a dangling link is refused like push, and the link and i
|
|||
const link = join(dir, "store.cc");
|
||||
await symlink(join(volume, "store.cc"), link);
|
||||
const envelope = { key: "a", type: "t", payload: documentPayloadFromBytes(Uint8Array.of(0x90)), storedAt: "2026-09-30T00:00:00Z" };
|
||||
await assert.rejects(() => new SingleFileMessagePackBackingStore(link).pushAll([envelope]), (error: NodeJS.ErrnoException) => error.code === "ENOENT");
|
||||
await assert.rejects(() => new SingleFileMessagePackBackingStore(link).pushAll([envelope]), (error: unknown) => error instanceof StoreLinkedError);
|
||||
assert.ok((await lstat(link)).isSymbolicLink());
|
||||
assert.deepEqual(await readdir(volume), []);
|
||||
});
|
||||
|
|
@ -2827,7 +2848,6 @@ test("every registration refusal is a SchemaConflictError naming the id and both
|
|||
// Cross-runtime store lock: each runtime's writer excludes the others on `<file>.lock`.
|
||||
|
||||
import { spawn } from "node:child_process";
|
||||
import { withStoreWriteLock } from "../src/store-file";
|
||||
|
||||
type LockHolder = { exited: Promise<number | null>; stderr: () => string };
|
||||
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import { promisify } from "node:util";
|
|||
|
||||
import { decode, encode } from "@msgpack/msgpack";
|
||||
|
||||
import { StoreLinkedError as PackageStoreLinkedError } from "../src/index";
|
||||
import { documentPayloadBytes, documentPayloadFromBytes } from "../src/document";
|
||||
import { SingleFileMessagePackBackingStore } from "../src/single-file-messagepack-backing-store";
|
||||
import { SchemaConflictError, StoreUnreadableError } from "../src/store-format";
|
||||
|
|
@ -23,6 +24,7 @@ import {
|
|||
replaceStoreFile,
|
||||
replaceStoreFileWindows,
|
||||
windowsKernel,
|
||||
refuseLinkedStore,
|
||||
withStoreWriteLock,
|
||||
} from "../src/store-file";
|
||||
import type { ReplaceKernel, StoreFileOps } from "../src/store-file";
|
||||
|
|
@ -818,3 +820,35 @@ test("opening writes nothing: a missing store leaves no lock file, a legacy stor
|
|||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("StoreLinkedError is importable from the package entry and is what a write at a link throws", { skip: process.platform === "win32" }, async () => {
|
||||
const directory = await scratch();
|
||||
try {
|
||||
const target = join(directory, "target.cc");
|
||||
await writeFile(target, "x");
|
||||
const link = join(directory, "link.cc");
|
||||
await fsp.symlink(target, link);
|
||||
await assert.rejects(
|
||||
() => refuseLinkedStore(link),
|
||||
(error: unknown) => error instanceof PackageStoreLinkedError && error.path === link && error.name === "StoreLinkedError",
|
||||
);
|
||||
await assert.rejects(() => new SingleFileMessagePackBackingStore(link).pushAll([]), (error: unknown) => error instanceof PackageStoreLinkedError);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("refuseLinkedStore lets a path that is absent pass and rethrows any other lstat failure", { skip: process.platform === "win32" }, async () => {
|
||||
const directory = await scratch();
|
||||
try {
|
||||
await refuseLinkedStore(join(directory, "absent.cc"));
|
||||
const file = join(directory, "file");
|
||||
await writeFile(file, "x");
|
||||
await assert.rejects(
|
||||
() => refuseLinkedStore(join(file, "child.cc")),
|
||||
(error: unknown) => (error as NodeJS.ErrnoException).code === "ENOTDIR",
|
||||
);
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in a new issue