diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index 0980f743..704a3851 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -5,7 +5,7 @@ import json import os import threading from pathlib import Path -from typing import Any +from typing import Any, Callable, TypeVar from .backing_store import ( CultCacheEnvelope, @@ -16,6 +16,7 @@ from .backing_store import ( STORE_FORMAT_VERSION = "cultcache.store.v1" _STORE_FORMAT_PREFIX = "cultcache.store." _PERSISTED_RECORD_SLOTS = 4 +_Contents = TypeVar("_Contents") class JsonLinesBackingStore: @@ -25,10 +26,11 @@ class JsonLinesBackingStore: def pull_all(self) -> list[CultCacheEnvelope]: with self._lock: - if not self.path.exists(): + text = _read_store(self.path, lambda path: path.read_text(encoding="utf-8")) + if text is None: return [] envelopes: list[CultCacheEnvelope] = [] - for line_number, line in enumerate(self.path.read_text(encoding="utf-8").splitlines(), start=1): + for line_number, line in enumerate(text.splitlines(), start=1): if not line.strip(): continue raw = json.loads(line) @@ -88,10 +90,9 @@ class SingleFileMessagePackBackingStore: def pull_all(self) -> list[CultCacheEnvelope]: msgpack = self._msgpack() with self._lock: - if not self.path.exists(): - return [] - data = self.path.read_bytes() - if not data: + # An empty file is not a store (no writer leaves one), so it is decoded and refused. + data = _read_store(self.path, Path.read_bytes) + if data is None: return [] decoded = msgpack.unpackb(data, raw=False) snapshot = _decode_v1_snapshot(decoded) @@ -154,13 +155,44 @@ def _encode_v1_snapshot(envelopes: list[CultCacheEnvelope]) -> list[Any]: return [STORE_FORMAT_VERSION, catalog, records] +def _read_store(path: Path, read: Callable[[Path], _Contents]) -> _Contents | None: + """Reads a store file, or returns None when nothing is at its path. + + Only nothing is absent. A dangling symbolic link is something: reading it as + empty would let the writer replace the link and move the store off its volume. + Any other failure to reach the file stays an OSError. + """ + try: + return read(path) + except FileNotFoundError: + if path.is_symlink(): + raise + return None + + +def _describe_identity(value: Any) -> str: + """A record key or schema id as a refusal may show it: a string is an identity and is + named; anything else is described by its type, since the value is the store's.""" + if isinstance(value, str): + return repr(value) + return f"<{type(value).__name__}>" + + +def _describe_header(header: str) -> str: + """Echoes a header only in the shape cultcache.store.v; the bytes are the store's.""" + version = header[len(_STORE_FORMAT_PREFIX) + 1:] if header.startswith(_STORE_FORMAT_PREFIX + "v") else "" + if version.isascii() and version.isdigit(): + return repr(header) + return f"an unrecognised {_STORE_FORMAT_PREFIX}* header of {len(header.encode('utf-8'))} bytes" + + def _decode_v1_snapshot(decoded: Any) -> list[CultCacheEnvelope] | None: if not isinstance(decoded, list) or not decoded: return None if isinstance(decoded[0], str) and decoded[0].startswith(_STORE_FORMAT_PREFIX) and decoded[0] != STORE_FORMAT_VERSION: raise ValueError( - f"CultCache store format {decoded[0]!r} is not readable; this runtime reads {STORE_FORMAT_VERSION!r} only. " - "The store needs a runtime that resolves document variants." + f"CultCache store format {_describe_header(decoded[0])} is not one this runtime reads; " + f"it reads {STORE_FORMAT_VERSION!r} only" ) if decoded[0] != STORE_FORMAT_VERSION: return None @@ -180,8 +212,9 @@ def _decode_v1_snapshot(decoded: Any) -> list[CultCacheEnvelope] | None: raise ValueError("CultCache persisted records must be MessagePack arrays") if len(raw_record) > _PERSISTED_RECORD_SLOTS: raise ValueError( - f"CultCache record {raw_record[0]!r} (schema {raw_record[1]!r}) has {len(raw_record)} slots; " - f"this runtime reads {_PERSISTED_RECORD_SLOTS}. The store needs a runtime that resolves document variants." + f"CultCache record {_describe_identity(raw_record[0])} (schema {_describe_identity(raw_record[1])}) " + f"has {len(raw_record)} slots, more than " + f"the {_PERSISTED_RECORD_SLOTS} of a {STORE_FORMAT_VERSION} record, so this is not a valid store" ) key, schema_id, stored_at, payload = raw_record if not isinstance(key, str) or not key: diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 6001c905..6e85f833 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import tempfile import threading import unittest @@ -841,6 +842,129 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(envelopes[0].payload, b"\x92\xa5alpha\x01") self.assertEqual(envelopes[1].payload, b"\x92\xa4beta\x02") + def test_single_file_refusals_say_what_was_found_without_blaming_variants(self) -> None: + with self.assertRaises(ValueError) as unsupported: + self._pull_vector("unknown-header.msgpack") + self.assertIn("not one this runtime reads", str(unsupported.exception)) + self.assertNotIn("variant", str(unsupported.exception)) + with self.assertRaises(ValueError) as invalid: + self._pull_vector("extra-slot-full-payload.msgpack") + self.assertIn("not a valid store", str(invalid.exception)) + self.assertNotIn("variant", str(invalid.exception)) + + def test_single_file_only_a_missing_store_reads_as_empty(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + self.assertEqual(SingleFileMessagePackBackingStore(Path(tmp) / "store.cc").pull_all(), []) + + # An existing empty file is not a store: no CultCache writer leaves one. + empty = Path(tmp) / "empty.cc" + empty.write_bytes(b"") + with self.assertRaises(ValueError): + SingleFileMessagePackBackingStore(empty).pull_all() + self.assertEqual(empty.read_bytes(), b"") + + # An empty array is the legacy envelope array with no envelopes: no header, no format claimed. + legacy_empty = Path(tmp) / "legacy-empty.cc" + legacy_empty.write_bytes(b"\x90") + self.assertEqual(SingleFileMessagePackBackingStore(legacy_empty).pull_all(), []) + + def test_single_file_refusals_never_echo_a_value_from_the_store(self) -> None: + import msgpack # type: ignore + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "store.cc" + cases = [ + (["cultcache.store.v12", [], []], True), + (["cultcache.store.SECRET-HEADER", [], []], False), + (["cultcache.store.v12SECRET", [], []], False), + (["cultcache.store.v", [], []], False), + (["cultcache.store.v1", ["SECRET-CATALOG-TEXT"], []], None), + ] + for value, echoed in cases: + path.write_bytes(msgpack.packb(value, use_bin_type=True)) + with self.assertRaises(ValueError) as caught: + SingleFileMessagePackBackingStore(path).pull_all() + message = str(caught.exception) + if echoed: + self.assertIn(repr(value[0]), message) + else: + self.assertNotIn("SECRET", message) + if echoed is False: + self.assertNotIn(repr(value[0]), message) + self.assertIn(f"of {len(value[0])} bytes", message) + + def test_single_file_record_refusal_names_only_a_string_key(self) -> None: + import msgpack # type: ignore + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "store.cc" + for key, schema in ((["SECRET-IN-LIST"], "s"), (b"SECRET-BYTES", "s"), ("k", {"SECRET": 1})): + path.write_bytes(msgpack.packb( + ["cultcache.store.v1", [], [[key, schema, "t", b"\x90", "extra"]]], use_bin_type=True)) + with self.assertRaises(ValueError) as caught: + SingleFileMessagePackBackingStore(path).pull_all() + 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. + @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: + with tempfile.TemporaryDirectory() as tmp: + volume = Path(tmp) / "volume" + volume.mkdir() + target = volume / "store.cc" + 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" + link.symlink_to(target) + SingleFileMessagePackBackingStore(link).push(replace(envelope, key="b")) + self.assertFalse(link.is_symlink()) + self.assertTrue(link.is_file()) + self.assertEqual(target.read_bytes(), before) + + @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: + for store_type in (SingleFileMessagePackBackingStore, JsonLinesBackingStore): + with tempfile.TemporaryDirectory() as tmp: + volume = Path(tmp) / "volume" + volume.mkdir() + path = Path(tmp) / "store.cc" + path.symlink_to(volume / "store.cc") + store = store_type(path) + with self.assertRaises(FileNotFoundError): + store.pull_all() + with self.assertRaises(FileNotFoundError): + 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()), []) + + @unittest.skipIf(os.name != "posix", "symlink loops and file-as-parent errors are POSIX errno cases") + def test_json_lines_store_that_cannot_be_reached_is_an_os_error(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + self.assertEqual(JsonLinesBackingStore(Path(tmp) / "missing.jsonl").pull_all(), []) + loop = Path(tmp) / "loop.jsonl" + loop.symlink_to(loop) + parent = Path(tmp) / "body" + parent.write_bytes(b"not a directory") + for path in (loop, parent / "store.jsonl"): + with self.assertRaises(OSError) as caught: + JsonLinesBackingStore(path).pull_all() + self.assertNotIsInstance(caught.exception, FileNotFoundError) + + @unittest.skipIf(os.name != "posix", "symlink loops and file-as-parent errors are POSIX errno cases") + def test_single_file_store_that_cannot_be_reached_is_an_os_error(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + loop = Path(tmp) / "loop.cc" + loop.symlink_to(loop) + parent = Path(tmp) / "body" + parent.write_bytes(b"not a directory") + for path in (loop, parent / "store.cc"): + with self.assertRaises(OSError) as caught: + SingleFileMessagePackBackingStore(path).pull_all() + self.assertNotIsInstance(caught.exception, FileNotFoundError) + if __name__ == "__main__": unittest.main() diff --git a/packages/cultcache-rs/README.md b/packages/cultcache-rs/README.md index 1ad01eb0..6841791f 100644 --- a/packages/cultcache-rs/README.md +++ b/packages/cultcache-rs/README.md @@ -293,6 +293,61 @@ writes. That protects ordinary multi-process access to the same file. It is still not a multi-master replication protocol, and it is not a substitute for a coordinator when higher-level write ordering matters. +A write that fails part way, for example on a full disk, removes its staging +file and leaves the store as it was. A read that fails tells you why through +the error chain: + +```rust +use cultcache_rs::{CacheBackingStore, CultCacheStoreUnreadable, CultCacheStoreUnreadableKind, SingleFileMessagePackBackingStore}; + +let store = SingleFileMessagePackBackingStore::new("cache.cc"); +if let Err(error) = store.pull_all() { + match error.downcast_ref::().map(|unreadable| unreadable.kind) { + // The bytes were read and are not a store: truncated, garbled, or invalid. + Some(CultCacheStoreUnreadableKind::Undecodable) => {} + // The header names a format this runtime does not read, such as a newer one. + Some(CultCacheStoreUnreadableKind::UnsupportedFormat) => {} + // The bytes could not be read at all: a std::io::Error is in the chain. + _ => {} + } +} +``` + +An I/O failure never carries `CultCacheStoreUnreadable`, and the chain under +`CultCacheStoreUnreadable` never holds a `std::io::Error`. + +Only a store file that does not exist (`std::io::ErrorKind::NotFound`) reads as +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`, `delete`, the exchanges) +fails before writing. Writes do not yet follow links: a write through a live link, +and a `push_all` through a dangling one, replace the link with a regular file and +leave its target alone. 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`. + +A write that fails says what it left on disk: + +```rust +use cultcache_rs::{CultCacheStoreWriteFailed, CultCacheStoreWriteFailedKind}; + +fn outcome(error: &anyhow::Error) { + match error.downcast_ref::().map(|failed| failed.kind) { + // The entries cannot be stored; retrying them fails the same way. + Some(CultCacheStoreWriteFailedKind::Rejected) => {} + // Staging, fsync or rename failed: the store file holds what it held before, + // and a retry may succeed once the fault (a full disk, say) clears. + Some(CultCacheStoreWriteFailedKind::NotReplaced) => {} + // The rename landed and the directory sync failed: the new snapshot is in + // place but may not survive a crash. + Some(CultCacheStoreWriteFailedKind::ReplacedNotDurable) => {} + _ => {} + } +} +``` + ## Near-Term Ergonomic Improvements 1. **Derive macro** diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index c9a7d282..71bf8fca 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -280,8 +280,8 @@ struct PersistedRecord( ); impl<'de> serde::Deserialize<'de> for PersistedRecord { - /// Reads the four v1 slots and refuses a fifth, naming the record. A record with more - /// slots was written by a runtime that resolves document variants. + /// Reads the four v1 slots and refuses a fifth, naming the record. No runtime reads a + /// v1 store holding such a record, so the store is not a valid one. fn deserialize>(deserializer: D) -> std::result::Result { use serde::de::{Error, IgnoredAny, SeqAccess, Visitor}; @@ -301,7 +301,7 @@ impl<'de> serde::Deserialize<'de> for PersistedRecord { seq.next_element()?.ok_or_else(|| A::Error::invalid_length(3, &self))?; if seq.next_element::()?.is_some() { return Err(A::Error::custom(format!( - "CultCache record {key:?} (schema {schema_id:?}) has more than 4 slots; this runtime reads 4. The store needs a runtime that resolves document variants." + "{RECORD_REFUSAL}{key:?} (schema {schema_id:?}) has more than the 4 slots of a {STORE_FORMAT_V1} record, so this is not a valid store" ))); } Ok(PersistedRecord(key, schema_id, stored_at, payload.into_vec())) @@ -312,34 +312,159 @@ impl<'de> serde::Deserialize<'de> for PersistedRecord { } } -/// The store header string, or `None` when the bytes are not a snapshot-shaped array -/// (the legacy envelope array starts with a map). +const STORE_FORMAT_V1: &str = "cultcache.store.v1"; + +/// Opens every refusal CultCache raises about one persisted record. The decoder's own +/// messages can quote the bytes; these name only the record's key and schema id. +const RECORD_REFUSAL: &str = "CultCache record "; + +/// A store header as a refusal may show it: echoed when it has the shape +/// `cultcache.store.v`, and otherwise described only by its length, since the +/// bytes are the store's and an error text must not carry them. +fn describe_header(header: &str) -> String { + let known_shape = header + .strip_prefix("cultcache.store.v") + .is_some_and(|version| !version.is_empty() && version.bytes().all(|byte| byte.is_ascii_digit())); + if known_shape { + format!("{header:?}") + } else { + format!("an unrecognised cultcache.store.* header of {} bytes", header.len()) + } +} + +/// What a failed decode tells the operator: CultCache's own context and record +/// refusals, and for a decoder failure only its kind, never a value from the bytes. +fn describe_undecodable(error: &anyhow::Error) -> String { + let mut parts = Vec::new(); + for cause in error.chain() { + if let Some(decode) = cause.downcast_ref::() { + parts.push(describe_decode_error(decode)); + break; + } + parts.push(cause.to_string()); + } + parts.join(": ") +} + +/// A decoder failure as an error that keeps only its kind. Every decode of stored bytes +/// goes through this or [`describe_undecodable`], so no error text quotes a stored value. +fn opaque_decode_error(error: rmp_serde::decode::Error) -> anyhow::Error { + anyhow!(describe_decode_error(&error)) +} + +fn describe_decode_error(error: &rmp_serde::decode::Error) -> String { + use rmp_serde::decode::Error; + match error { + Error::InvalidMarkerRead(_) | Error::InvalidDataRead(_) => "the bytes end before the store does", + Error::TypeMismatch(_) => "a value has a MessagePack type the store format does not allow", + Error::OutOfRange => "a number is out of range for its slot", + Error::LengthMismatch(_) => "an array has a length the store format does not allow", + Error::Utf8Error(_) => "a string is not valid UTF-8", + Error::DepthLimitExceeded => "values are nested deeper than the decoder allows", + Error::Syntax(message) | Error::Uncategorized(message) if message.starts_with(RECORD_REFUSAL) => { + return message.clone(); + } + Error::Syntax(_) | Error::Uncategorized(_) => "a value does not fit the store format", + } + .to_string() +} + +/// The store header string, or `None` when the bytes do not open with an array whose +/// first slot is a string (the legacy envelope array starts with a map). Only the +/// leading bytes are read, so a store cut short after its header still names its format. fn store_header(bytes: &[u8]) -> Option { - struct Header(String); - impl<'de> serde::Deserialize<'de> for Header { - fn deserialize>(deserializer: D) -> std::result::Result { - struct HeaderVisitor; - impl<'de> serde::de::Visitor<'de> for HeaderVisitor { - type Value = Header; + let mut offset = 0usize; + if read_array_header(bytes, &mut offset)? == 0 { + return None; + } + read_string(bytes, &mut offset) +} - fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { - formatter.write_str("an array whose first slot is the store header") - } +/// A single-file store whose bytes were read and not accepted. It sits in the error +/// chain of every such read, so a consumer tells it apart with +/// `error.downcast_ref::()`. A failure to read the bytes +/// at all stays a [`std::io::Error`] in the chain and never carries this type; the +/// chain under this type never holds an `io::Error`, so the two never overlap. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct CultCacheStoreUnreadable { + pub path: PathBuf, + pub kind: CultCacheStoreUnreadableKind, +} - fn visit_seq>(self, mut seq: A) -> std::result::Result { - let header: String = seq - .next_element()? - .ok_or_else(|| ::invalid_length(0, &self))?; - while seq.next_element::()?.is_some() {} - Ok(Header(header)) - } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum CultCacheStoreUnreadableKind { + /// The bytes are not a store this runtime's format accepts: truncated, garbled, + /// or a record the format does not allow. + Undecodable, + /// The header names a store format this runtime does not read, such as one + /// written by a newer runtime. The rest of the file was not judged. + UnsupportedFormat, +} + +impl std::fmt::Display for CultCacheStoreUnreadable { + fn fmt(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + match self.kind { + CultCacheStoreUnreadableKind::Undecodable => { + write!(formatter, "CultCache store {} is not decodable", self.path.display()) } - deserializer.deserialize_seq(HeaderVisitor) + CultCacheStoreUnreadableKind::UnsupportedFormat => write!( + formatter, + "CultCache store {} is in a format this runtime does not read", + self.path.display() + ), } } - rmp_serde::from_slice::
(bytes).ok().map(|header| header.0) } +impl std::error::Error for CultCacheStoreUnreadable {} + +/// 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::()`. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct CultCacheStoreWriteFailed { + pub path: PathBuf, + pub kind: CultCacheStoreWriteFailedKind, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum CultCacheStoreWriteFailedKind { + /// The entries cannot be stored as a snapshot, for example two document types + /// claim one schema id. Nothing was written, and retrying the same entries fails + /// the same way: this is a refusal, not an I/O failure. + Rejected, + /// The write failed before the store file was replaced (creating, writing or + /// syncing the staging file, or renaming it): the file holds what it held before. + NotReplaced, + /// The store file was replaced, then syncing its directory failed: the new + /// snapshot is in place but may not survive a crash. + ReplacedNotDurable, +} + +impl std::fmt::Display for CultCacheStoreWriteFailed { + fn fmt(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + match self.kind { + CultCacheStoreWriteFailedKind::Rejected => { + write!(formatter, "CultCache store {} refused a snapshot it cannot store", self.path.display()) + } + CultCacheStoreWriteFailedKind::NotReplaced => { + write!(formatter, "CultCache store {} was not replaced", self.path.display()) + } + CultCacheStoreWriteFailedKind::ReplacedNotDurable => write!( + formatter, + "CultCache store {} was replaced, but the replacement is not yet durable", + self.path.display() + ), + } + } +} + +impl std::error::Error for CultCacheStoreWriteFailed {} + #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct PushAllOptions { pub soft: bool, @@ -416,10 +541,12 @@ impl SingleFileMessagePackBackingStore { /// either file. Writers using the sibling lock cannot replace the snapshot /// until `action` returns. /// - /// Releasing the lock is RAII cleanup. A release failure therefore cannot - /// rewrite the completed action's result; consumers that need to observe - /// that cleanup failure can use - /// [`Self::with_read_only_shared_snapshot_and_unlock_diagnostic`]. + /// Releasing the lock is RAII cleanup, under the same policy as every lock in + /// this crate (see `release_lock`): a release failure cannot rewrite the + /// completed action's result. This path alone also offers an observer for that + /// failure, [`Self::with_read_only_shared_snapshot_and_unlock_diagnostic`], + /// because a provider crossing may need to report it; the locked read and write + /// paths discard it. pub fn with_read_only_shared_snapshot( &self, action: impl FnOnce(Vec) -> Result, @@ -711,19 +838,44 @@ impl SingleFileMessagePackBackingStore { } fn read_all_unlocked(&self) -> Result> { - if !self.path.exists() { - return Ok(Vec::new()); - } - let bytes = fs::read(&self.path) - .with_context(|| format!("failed to read {}", self.path.display()))?; - if bytes.is_empty() { - return Ok(Vec::new()); - } + // Only a store that is not there is empty. A file that is there and empty is + // not a store (no writer leaves one), and any other failure to reach the file, + // such as a symlink loop, a file where a directory should be, or no permission, + // stays an io::Error: reading it as empty would let a writer replace a store it + // never saw. + // A dangling symbolic link is something at the path, not nothing: reading it as + // empty would let the writer rename a file over the link and move the store off + // the volume the link pointed to. + let bytes = match fs::read(&self.path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound && is_absent(&self.path)? => { + return Ok(Vec::new()); + } + Err(error) => { + return Err(anyhow::Error::new(error) + .context(format!("failed to read {}", self.path.display()))); + } + }; + let unreadable = |kind, detail: anyhow::Error| { + // The detail is rebuilt as text that names no value read from the bytes. The + // decoder reports a short read as an io::Error, and dropping it keeps an + // io::Error in the chain meaning only that the file itself could not be read. + anyhow!(describe_undecodable(&detail)).context(CultCacheStoreUnreadable { + path: self.path.clone(), + kind, + }) + }; match store_header(&bytes) { - Some(header) if header.starts_with("cultcache.store.") => decode_store_snapshot(&bytes), - _ => rmp_serde::from_slice(&bytes).map_err(anyhow::Error::from), + Some(header) if header == STORE_FORMAT_V1 => decode_store_snapshot(&bytes) + .map_err(|error| unreadable(CultCacheStoreUnreadableKind::Undecodable, error)), + Some(header) if header.starts_with("cultcache.store.") => Err(unreadable( + CultCacheStoreUnreadableKind::UnsupportedFormat, + anyhow!("its header is {}; this runtime reads {STORE_FORMAT_V1:?} only", describe_header(&header)), + )), + _ => rmp_serde::from_slice(&bytes) + .context("failed to decode legacy CultCache envelope array") + .map_err(|error| unreadable(CultCacheStoreUnreadableKind::Undecodable, error)), } - .map_err(|error| anyhow!("failed to decode MessagePack {}: {error:#}", self.path.display())) } /// Reads one filesystem snapshot without creating or opening the sibling @@ -739,29 +891,39 @@ impl SingleFileMessagePackBackingStore { } fn write_all_unlocked(&self, entries: &[CultCacheEnvelope]) -> Result<()> { + let failed = |kind| CultCacheStoreWriteFailed { + path: self.path.clone(), + kind, + }; + let bytes = encode_store_snapshot(entries) + .and_then(|snapshot| rmp_serde::to_vec(&snapshot).context("failed to encode MessagePack")) + .map_err(|error| error.context(failed(CultCacheStoreWriteFailedKind::Rejected)))?; + self.replace_unlocked(&bytes) + .map_err(|error| error.context(failed(CultCacheStoreWriteFailedKind::NotReplaced)))?; + sync_parent_directory(&self.path) + .map_err(|error| error.context(failed(CultCacheStoreWriteFailedKind::ReplacedNotDurable))) + } + + /// Everything up to and including the rename over the store file. On failure the + /// store file is as it was. + fn replace_unlocked(&self, bytes: &[u8]) -> Result<()> { if let Some(parent) = self.path.parent() { fs::create_dir_all(parent) .with_context(|| format!("failed to create {}", parent.display()))?; } remove_abandoned_staging_files(&self.path)?; - let bytes = rmp_serde::to_vec(&encode_store_snapshot(entries)?) - .context("failed to encode MessagePack")?; let tmp_path = temporary_path_for(&self.path); - let mut staged = OpenOptions::new() + let staged = OpenOptions::new() .create_new(true) .write(true) .open(&tmp_path) .with_context(|| format!("failed to create {}", tmp_path.display()))?; - use std::io::Write; - staged - .write_all(&bytes) - .with_context(|| format!("failed to write {}", tmp_path.display()))?; - staged - .sync_all() - .with_context(|| format!("failed to sync {}", tmp_path.display()))?; - drop(staged); - replace_file_atomically(&tmp_path, &self.path)?; - sync_parent_directory(&self.path)?; + // From here the staging file is this write's to remove: a write that fails + // for lack of space must not keep holding the space it failed on. + if let Err(error) = stage_and_replace(staged, &tmp_path, bytes, &self.path) { + let _ = fs::remove_file(&tmp_path); + return Err(error); + } Ok(()) } @@ -769,20 +931,14 @@ impl SingleFileMessagePackBackingStore { let lock = self.open_lock_file()?; fs2::FileExt::lock_shared(&lock) .with_context(|| format!("failed to lock {}", self.lock_path().display()))?; - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result + release_lock(lock, action()) } fn with_exclusive_lock(&self, action: impl FnOnce() -> Result) -> Result { let lock = self.open_lock_file()?; fs2::FileExt::lock_exclusive(&lock) .with_context(|| format!("failed to lock {}", self.lock_path().display()))?; - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result + release_lock(lock, action()) } /// Replaces the named entries only if each still matches its expectation. @@ -899,10 +1055,7 @@ impl SingleFileMessagePackBackingStore { .with_context(|| format!("failed to lock {}", self.lock_path().display())); } } - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result.map(Some) + release_lock(lock, action()).map(Some) } fn open_lock_file(&self) -> Result { @@ -987,7 +1140,7 @@ fn unique_batch_ids( impl CacheBackingStore for SingleFileMessagePackBackingStore { fn pull_all(&self) -> Result> { - if !self.path.exists() && !self.lock_path().exists() { + if is_absent(&self.path)? && is_absent(&self.lock_path())? { return Ok(Vec::new()); } self.with_shared_lock(|| self.read_all_unlocked()) @@ -1092,10 +1245,7 @@ impl RedbMessagePackBackingStore { .with_context(|| format!("failed to open {}", lock_path.display()))?; fs2::FileExt::lock_exclusive(&lock) .with_context(|| format!("failed to lock {}", lock_path.display()))?; - let result = self.open_database().and_then(|database| action(&database)); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", lock_path.display()))?; - result + release_lock(lock, self.open_database().and_then(|database| action(&database))) } fn lock_path(&self) -> PathBuf { @@ -1875,7 +2025,9 @@ fn read_redb_entry( table .get(key.as_slice())? .map(|value| { - rmp_serde::from_slice(value.value()).context("failed to decode redb CultCache envelope") + rmp_serde::from_slice(value.value()) + .map_err(opaque_decode_error) + .context("failed to decode redb CultCache envelope") }) .transpose() } @@ -1897,7 +2049,9 @@ fn read_all_redb( .iter()? .map(|row| { let (_, value) = row?; - rmp_serde::from_slice(value.value()).context("failed to decode redb CultCache envelope") + rmp_serde::from_slice(value.value()) + .map_err(opaque_decode_error) + .context("failed to decode redb CultCache envelope") }) .collect::>>()?; entries.sort_by_key(entry_id); @@ -2128,7 +2282,7 @@ impl CultCache { let Some(entry) = self.entries.get(&entry_id_parts(T::TYPE, key)) else { return Ok(None); }; - let payload = rmp_serde::from_slice(&entry.payload).with_context(|| { + let payload = rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2161,7 +2315,7 @@ impl CultCache { if entry.r#type != T::TYPE { continue; } - values.push(rmp_serde::from_slice(&entry.payload).with_context(|| { + values.push(rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2182,7 +2336,7 @@ impl CultCache { } values.push(( entry.key.clone(), - rmp_serde::from_slice(&entry.payload).with_context(|| { + rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2264,7 +2418,7 @@ impl CultCache { key: String, payload: Vec, ) -> Result<(CultCacheEnvelope, T)> { - let parsed: T = rmp_serde::from_slice(&payload).with_context(|| { + let parsed: T = rmp_serde::from_slice(&payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to validate CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2400,7 +2554,7 @@ impl CultCache { )); } - rmp_serde::from_slice(&entry.payload).with_context(|| { + rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to validate CultCache envelope {:?} at key {:?} as {}", T::TYPE, @@ -2574,20 +2728,14 @@ fn encode_store_snapshot(entries: &[CultCacheEnvelope]) -> Result Result> { - if let Some(header) = store_header(bytes) { - if header != "cultcache.store.v1" { - return Err(anyhow!( - "CultCache store format {header:?} is not readable; this runtime reads \"cultcache.store.v1\" only. The store needs a runtime that resolves document variants." - )); - } - } let snapshot: PersistedStoreSnapshot = rmp_serde::from_slice(bytes).context("failed to decode CultCache v1 snapshot")?; @@ -2727,6 +2875,47 @@ fn temporary_path_for(path: &Path) -> PathBuf { path.with_file_name(file_name) } +/// Writes and syncs the staged snapshot, closes it, and renames it over `destination`. +fn stage_and_replace(mut staged: File, staged_path: &Path, bytes: &[u8], destination: &Path) -> Result<()> { + use std::io::Write; + staged + .write_all(bytes) + .with_context(|| format!("failed to write {}", staged_path.display()))?; + injected_write_fault(WriteStep::StagingSync) + .and_then(|()| staged.sync_all()) + .with_context(|| format!("failed to sync {}", staged_path.display()))?; + drop(staged); + replace_file_atomically(staged_path, destination) +} + +/// Releases a lock taken around `result`'s action and returns that result. +/// +/// The policy for every lock in this crate: an unlock failure never replaces the action's +/// result, so a write that landed is reported as landed. The failure is discarded here +/// because these paths have no observer; the read-only snapshot, whose API offers one, +/// hands it to [`SingleFileMessagePackBackingStore::with_read_only_shared_snapshot_and_unlock_diagnostic`]. +/// +/// Dropping `lock` closes this descriptor, which releases the lock only when no other +/// descriptor shares its open file description. A child forked without exec holds such a +/// duplicate, and the lock then stays held until the child closes it or exits. +fn release_lock(lock: File, result: Result) -> Result { + let _unlock_failure_released_by_close = + injected_write_fault(WriteStep::Unlock).and_then(|()| fs2::FileExt::unlock(&lock)); + drop(lock); + result +} + +/// True only when nothing is at `path`; a symbolic link, dangling or not, is something. +/// A path that cannot be inspected (a file where a directory should be, no permission) +/// is an error, never absent. +fn is_absent(path: &Path) -> Result { + match fs::symlink_metadata(path) { + Ok(_) => Ok(false), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(true), + Err(error) => Err(anyhow::Error::new(error).context(format!("failed to inspect {}", path.display()))), + } +} + fn remove_abandoned_staging_files(destination: &Path) -> Result<()> { let Some(parent) = destination.parent() else { return Ok(()); @@ -2870,11 +3059,41 @@ fn sync_parent_directory(path: &Path) -> Result<()> { let Some(parent) = path.parent() else { return Ok(()); }; - File::open(parent) + injected_write_fault(WriteStep::DirectorySync) + .and_then(|()| File::open(parent)) .and_then(|directory| directory.sync_all()) .with_context(|| format!("failed to sync {}", parent.display())) } +/// The steps no test can make fail for real: fsync and unlock do not fail on a healthy +/// filesystem. Tests arm one step on their own thread; builds outside tests never fail here. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[allow(dead_code)] +enum WriteStep { + StagingSync, + DirectorySync, + Unlock, +} + +#[cfg(test)] +thread_local! { + static INJECTED_WRITE_FAULT: std::cell::Cell> = const { std::cell::Cell::new(None) }; +} + +#[cfg(test)] +fn injected_write_fault(step: WriteStep) -> std::io::Result<()> { + if INJECTED_WRITE_FAULT.with(|fault| fault.get()) == Some(step) { + return Err(std::io::Error::other("injected write fault")); + } + Ok(()) +} + +#[cfg(not(test))] +#[inline(always)] +fn injected_write_fault(_step: WriteStep) -> std::io::Result<()> { + Ok(()) +} + #[cfg(windows)] fn sync_parent_directory(_path: &Path) -> Result<()> { // MoveFileExW with MOVEFILE_WRITE_THROUGH owns publication durability. @@ -2993,6 +3212,8 @@ mod tests { store.push(&released)?; let writer_path = store_path.clone(); + let gate_path = store_path.clone(); + let gate_released = released.clone(); let (started_tx, started_rx) = mpsc::channel(); let (finished_tx, finished_rx) = mpsc::channel(); store.with_read_only_shared_snapshot(|captured| { @@ -3007,15 +3228,21 @@ mod tests { writer.push(&engaged).unwrap(); finished_tx.send(()).unwrap(); }); - started_rx.recv_timeout(Duration::from_secs(1))?; + started_rx.recv_timeout(Duration::from_secs(30))?; assert!( finished_rx .recv_timeout(Duration::from_millis(100)) .is_err() ); + // The writer is still waiting on the gate: the file on disk is untouched. + assert_eq!( + SingleFileMessagePackBackingStore::new(&gate_path).pull_all_read_only_snapshot()?, + vec![gate_released.clone()] + ); Ok(()) })?; - finished_rx.recv_timeout(Duration::from_secs(1))?; + // The writer fsyncs twice once the gate opens; under IO pressure that is slow. + finished_rx.recv_timeout(Duration::from_secs(60))?; assert_eq!(store.pull_all()?[0].payload, b"engaged"); Ok(()) } @@ -4698,4 +4925,657 @@ mod tests { assert_eq!(envelopes[1].payload, b"\x92\xa4beta\x02"); Ok(()) } + + // Store read and write faults: what a consumer can tell apart, and what a failed + // write leaves on disk. + + fn unreadable_kind(error: &anyhow::Error) -> Option { + error + .downcast_ref::() + .map(|unreadable| unreadable.kind) + } + + fn io_error_in(error: &anyhow::Error) -> Option<&std::io::Error> { + error.chain().find_map(|cause| cause.downcast_ref::()) + } + + /// A store of three records written by this runtime, returned with its bytes. + fn written_store(path: &Path) -> Result<(SingleFileMessagePackBackingStore, Vec)> { + let store = SingleFileMessagePackBackingStore::new(path); + let records = vec![ + snapshot_envelope("alpha", &[1; 64]), + snapshot_envelope("beta", &[2; 64]), + snapshot_envelope("gamma", &[3; 64]), + ]; + assert!(store.compare_exchange_snapshot(&[], &records)?); + let bytes = fs::read(path)?; + Ok((store, bytes)) + } + + #[test] + fn a_truncated_store_is_undecodable_on_every_read_and_is_left_as_it_was() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, bytes) = written_store(&path)?; + let truncated = &bytes[..bytes.len() / 2]; + fs::write(&path, truncated)?; + + let failures = [ + store.pull_all().unwrap_err(), + store.pull_all_read_only_snapshot().unwrap_err(), + store + .compare_exchange_snapshot(&[], &[snapshot_envelope("delta", b"4")]) + .unwrap_err(), + ]; + for error in &failures { + assert_eq!(unreadable_kind(error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); + assert_eq!(error.downcast_ref::().unwrap().path, path); + assert!(error.to_string().contains(&path.display().to_string()), "{error}"); + assert!(io_error_in(error).is_none(), "{error:#}"); + } + assert_eq!(fs::read(&path)?, truncated); + Ok(()) + } + + #[test] + fn bytes_that_are_not_a_store_are_undecodable() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + // The last opens like a store, with a string where the header goes, but only a + // `cultcache.store.*` header claims a format this runtime might not read. + for bytes in [ + &b"not a store at all"[..], + &[0x92, 0xa3, b'a', b'b'][..], + &[0xc1][..], + &[0x92, 0xa2, b'a', b'b', 0x90][..], + &[0x93, 0xb0, b'c', b'u', b'l', b't', b'c', b'a', b'c', b'h', b'e', b'.', b's', b't', b'o', b'r', b'e', b'X', 0x90, 0x90][..], + ] { + fs::write(&path, bytes)?; + let error = store.pull_all().unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{bytes:?}: {error:#}"); + } + Ok(()) + } + + #[test] + fn a_v1_store_holding_a_record_v1_does_not_allow_is_undecodable() { + let error = pull_vector("extra-slot-full-payload.msgpack").unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); + } + + #[test] + fn a_store_in_a_format_this_runtime_does_not_read_is_unsupported_even_when_cut_short() -> Result<()> { + for name in ["unknown-header.msgpack", "variant-v2.msgpack", "../document-variants-c1/variant-store.msgpack"] { + let error = pull_vector(name).unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::UnsupportedFormat), "{name}: {error:#}"); + } + + // The header is all this runtime judges: a newer store cut short is still a + // newer store, not a damaged one this runtime could replace. + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.msgpack"); + let vector = fs::read( + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors/document-variants-c0/unknown-header.msgpack"), + )?; + let truncated = &vector[..vector.len() / 2]; + fs::write(&path, truncated)?; + let store = SingleFileMessagePackBackingStore::new(&path); + for error in [store.pull_all().unwrap_err(), store.pull_all_read_only_snapshot().unwrap_err()] { + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::UnsupportedFormat), "{error:#}"); + assert!(format!("{error:#}").contains("cultcache.store.v9"), "{error:#}"); + assert!(io_error_in(&error).is_none(), "{error:#}"); + } + assert_eq!(fs::read(&path)?, truncated); + Ok(()) + } + + #[test] + fn a_store_that_cannot_be_read_is_an_io_error_and_never_unreadable() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + fs::create_dir(&path)?; + let store = SingleFileMessagePackBackingStore::new(&path); + for error in [store.pull_all().unwrap_err(), store.pull_all_read_only_snapshot().unwrap_err()] { + assert!(io_error_in(&error).is_some(), "{error:#}"); + assert_eq!(unreadable_kind(&error), None, "{error:#}"); + } + Ok(()) + } + + /// Runs one write in a child process whose file-size limit is far below the + /// snapshot, with SIGXFSZ ignored so the write fails with EFBIG instead of + /// killing the child: a real mid-write failure after the staging file exists, + /// reached without permissions (the verification container runs as root). + #[cfg(unix)] + #[test] + fn a_write_that_fails_part_way_leaves_no_staging_file_and_keeps_the_store() -> Result<()> { + const CHILD_STORE: &str = "CULTCACHE_RS_FILE_SIZE_LIMITED_STORE"; + if let Ok(path) = std::env::var(CHILD_STORE) { + let mut store = SingleFileMessagePackBackingStore::new(path); + let error = store + .push(&snapshot_envelope("large", &vec![7; 1 << 20])) + .unwrap_err(); + let io = io_error_in(&error).expect("a size-limited write fails with an io::Error"); + assert_eq!(io.kind(), std::io::ErrorKind::FileTooLarge, "{error:#}"); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::NotReplaced), "{error:#}"); + assert_eq!(unreadable_kind(&error), None, "{error:#}"); + return Ok(()); + } + + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, before) = written_store(&path)?; + let child = std::process::Command::new("sh") + .arg("-c") + .arg("trap '' XFSZ; ulimit -f 64; exec \"$0\" --exact --test-threads=1 --quiet tests::a_write_that_fails_part_way_leaves_no_staging_file_and_keeps_the_store") + .arg(std::env::current_exe()?) + .env(CHILD_STORE, &path) + .output()?; + assert!( + child.status.success(), + "the size-limited child did not fail its write as expected: {} +{}{}", + child.status, + String::from_utf8_lossy(&child.stdout), + String::from_utf8_lossy(&child.stderr) + ); + + let staging: Vec<_> = fs::read_dir(temp.path())? + .map(|entry| entry.map(|entry| entry.file_name().to_string_lossy().into_owned())) + .collect::>>()? + .into_iter() + .filter(|name| name.starts_with("store.cc.") && name.ends_with(".tmp")) + .collect(); + assert_eq!(staging, Vec::::new()); + assert_eq!(fs::read(&path)?, before); + assert_eq!(store.pull_all()?.len(), 3); + Ok(()) + } + + /// Every read of a single-file store: the locked pull, the lock-free provider + /// snapshot, and the exchange a writer reads through. + fn every_read(store: &SingleFileMessagePackBackingStore) -> [Result>; 3] { + [ + store.pull_all(), + store.pull_all_read_only_snapshot(), + store + .compare_exchange_snapshot(&[], &[snapshot_envelope("delta", b"4")]) + .map(|_| Vec::new()), + ] + } + + fn assert_io_failure_on_every_read(store: &SingleFileMessagePackBackingStore) { + for result in every_read(store) { + let error = result.expect_err("a store that cannot be reached must not read as empty"); + 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:#}"); + } + } + + #[test] + fn only_a_missing_store_reads_as_empty() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + assert!(store.pull_all()?.is_empty()); + assert!(store.pull_all_read_only_snapshot()?.is_empty()); + assert!(store.compare_exchange_snapshot(&[], &[snapshot_envelope("delta", b"4")])?); + + // A store whose lock file is gone is still a store, not an absent one. + fs::remove_file(store.lock_path())?; + assert_eq!(store.pull_all()?, vec![snapshot_envelope("delta", b"4")]); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn a_store_behind_a_symlink_loop_is_an_io_error_on_every_read() -> Result<()> { + 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)); + assert!(fs::symlink_metadata(&path)?.file_type().is_symlink()); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn a_store_under_a_file_instead_of_a_directory_is_an_io_error_on_every_read() -> Result<()> { + let temp = tempfile::tempdir()?; + let parent = temp.path().join("body"); + fs::write(&parent, b"not a directory")?; + assert_io_failure_on_every_read(&SingleFileMessagePackBackingStore::new(parent.join("store.cc"))); + assert_eq!(fs::read(&parent)?, b"not a directory"); + Ok(()) + } + + #[test] + fn an_existing_empty_file_is_undecodable_and_is_left_as_it_was() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + fs::write(&path, b"")?; + let store = SingleFileMessagePackBackingStore::new(&path); + for result in every_read(&store) { + let error = result.expect_err("no CultCache writer leaves an empty file, so it is not a store"); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); + assert!(io_error_in(&error).is_none(), "{error:#}"); + } + assert_eq!(fs::read(&path)?, b""); + + // An empty array is the legacy envelope array with no envelopes: it has no + // header, so it claims no format, and it is an empty store. + fs::write(&path, [0x90])?; + assert!(store.pull_all()?.is_empty()); + assert!(store.pull_all_read_only_snapshot()?.is_empty()); + Ok(()) + } + + #[test] + fn a_newer_store_is_named_once_its_header_string_is_complete() -> Result<()> { + let vector = fs::read( + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors/document-variants-c0/unknown-header.msgpack"), + )?; + // One byte of array header, one of string header, then the 18-byte header string. + let header = "cultcache.store.v9"; + let header_end = 2 + header.len(); + assert_eq!(header_end, 20); + assert_eq!(&vector[2..header_end], header.as_bytes()); + + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + for length in 1..=vector.len() { + fs::write(&path, &vector[..length])?; + let error = store.pull_all().unwrap_err(); + let expected = if length < header_end { + CultCacheStoreUnreadableKind::Undecodable + } else { + CultCacheStoreUnreadableKind::UnsupportedFormat + }; + assert_eq!(unreadable_kind(&error), Some(expected), "cut to {length}: {error:#}"); + } + Ok(()) + } + + #[test] + fn refusals_say_what_was_found_without_blaming_variants() { + let unsupported = format!("{:#}", pull_vector("unknown-header.msgpack").unwrap_err()); + assert!(unsupported.contains("\"cultcache.store.v9\""), "{unsupported}"); + assert!(unsupported.contains("does not read"), "{unsupported}"); + assert!(!unsupported.contains("variant"), "{unsupported}"); + + // The C# reference refuses the same bytes: a v1 store with a longer record is + // not a store any runtime reads, so no upgrade is suggested. + let invalid = format!("{:#}", pull_vector("extra-slot-full-payload.msgpack").unwrap_err()); + assert!(invalid.contains("not a valid store"), "{invalid}"); + assert!(!invalid.contains("variant"), "{invalid}"); + assert!(!invalid.contains("runtime that"), "{invalid}"); + } + + /// The store path is a directory, so the staging file is written and synced and + /// then cannot be renamed over it: a failure after the write, reached without + /// permissions. + #[test] + fn a_write_that_fails_at_the_rename_leaves_no_staging_file() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + fs::create_dir(&path)?; + fs::write(path.join("kept"), b"kept")?; + let mut store = SingleFileMessagePackBackingStore::new(&path); + + let error = store + .push_all(&[snapshot_envelope("alpha", b"1")], PushAllOptions::default()) + .unwrap_err(); + assert!(format!("{error:#}").contains("failed to atomically replace"), "{error:#}"); + + let staging: Vec<_> = fs::read_dir(temp.path())? + .map(|entry| entry.map(|entry| entry.file_name().to_string_lossy().into_owned())) + .collect::>>()? + .into_iter() + .filter(|name| name.starts_with("store.cc.") && name.ends_with(".tmp")) + .collect(); + assert_eq!(staging, Vec::::new()); + assert_eq!(fs::read(path.join("kept"))?, b"kept"); + Ok(()) + } + + /// MessagePack bytes of a fixstr, for store fixtures built by hand. + fn fixstr(text: &str) -> Vec { + assert!(text.len() < 32); + let mut bytes = vec![0xa0 | text.len() as u8]; + bytes.extend_from_slice(text.as_bytes()); + bytes + } + + #[test] + fn a_refusal_never_echoes_a_value_from_the_store() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + + // A v1 store whose catalog holds a string where an entry goes, and a legacy + // envelope array holding a string where an envelope goes: the decoder's own + // message would quote both strings. + let mut v1 = vec![0x93]; + v1.extend(fixstr(STORE_FORMAT_V1)); + v1.push(0x91); + v1.extend(fixstr("SECRET-CATALOG-TEXT")); + v1.push(0x90); + let mut legacy = vec![0x91]; + legacy.extend(fixstr("SECRET-LEGACY-TEXT")); + for bytes in [v1, legacy] { + fs::write(&path, &bytes)?; + for result in every_read(&store) { + let error = result.unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); + assert!(!format!("{error:#}").contains("SECRET"), "{error:#}"); + assert!(!format!("{error:?}").contains("SECRET"), "{error:?}"); + } + } + Ok(()) + } + + #[test] + fn a_header_is_echoed_only_in_the_known_shape() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + for (header, echoed) in [ + ("cultcache.store.v12", true), + ("cultcache.store.SECRET-HEADER", false), + ("cultcache.store.v12SECRET", false), + ("cultcache.store.v", false), + ] { + let mut bytes = vec![0x93]; + bytes.extend(fixstr(header)); + bytes.extend([0x90, 0x90]); + fs::write(&path, &bytes)?; + let error = store.pull_all().unwrap_err(); + let text = format!("{error:#}"); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::UnsupportedFormat), "{text}"); + if echoed { + assert!(text.contains(&format!("\"{header}\"")), "{text}"); + } else { + assert!(!text.contains(&format!("\"{header}\"")), "{text}"); + assert!(!text.contains("SECRET"), "{text}"); + assert!(text.contains(&format!("of {} bytes", header.len())), "{text}"); + } + } + Ok(()) + } + + /// Nothing at the store path is an empty store; a symbolic link whose target is + /// gone is something, and a writer must not rename a file over it. + #[cfg(unix)] + #[test] + fn a_dangling_symlink_at_the_store_path_is_an_error_and_nothing_is_written() -> Result<()> { + let temp = tempfile::tempdir()?; + let volume = temp.path().join("volume"); + fs::create_dir(&volume)?; + let target = volume.join("store.cc"); + 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"); + assert!(io_error_in(&error).is_some(), "{error:#}"); + assert_eq!(unreadable_kind(&error), None, "{error:#}"); + } + assert!(fs::symlink_metadata(&path)?.file_type().is_symlink()); + assert!(!target.exists()); + assert_eq!(fs::read_dir(&volume)?.count(), 0); + Ok(()) + } + + fn write_failure_kind(error: &anyhow::Error) -> Option { + error + .downcast_ref::() + .map(|failed| failed.kind) + } + + /// Runs `write` with `step` made to fail on this thread. + fn with_write_fault(step: WriteStep, write: impl FnOnce() -> T) -> T { + INJECTED_WRITE_FAULT.with(|fault| fault.set(Some(step))); + let result = write(); + INJECTED_WRITE_FAULT.with(|fault| fault.set(None)); + result + } + + #[test] + fn a_write_that_fails_before_the_rename_is_not_replaced_and_the_store_is_intact() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, before) = written_store(&path)?; + let error = with_write_fault(WriteStep::StagingSync, || { + store.compare_exchange_snapshot(&store.pull_all().unwrap(), &[snapshot_envelope("delta", b"4")]) + }) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::NotReplaced), "{error:#}"); + assert!(io_error_in(&error).is_some(), "{error:#}"); + assert!(error.to_string().contains(&format!("{} was not replaced", path.display())), "{error}"); + assert_eq!(fs::read(&path)?, before); + assert_eq!(fs::read_dir(temp.path())?.count(), 2, "only the store and its lock remain"); + + // The rename fails when the store path is a directory. + let blocked = temp.path().join("blocked.cc"); + fs::create_dir(&blocked)?; + let mut blocked_store = SingleFileMessagePackBackingStore::new(&blocked); + let error = blocked_store + .push_all(&[snapshot_envelope("alpha", b"1")], PushAllOptions::default()) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::NotReplaced), "{error:#}"); + assert!(fs::metadata(&blocked)?.is_dir()); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn a_write_whose_directory_sync_fails_is_replaced_but_not_durable() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, _) = written_store(&path)?; + let replacement = vec![snapshot_envelope("delta", b"4")]; + let error = with_write_fault(WriteStep::DirectorySync, || { + store.compare_exchange_snapshot(&store.pull_all().unwrap(), &replacement) + }) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::ReplacedNotDurable), "{error:#}"); + assert!(error.to_string().contains(&format!("{} was replaced, but the replacement is not yet durable", path.display())), "{error}"); + assert!(io_error_in(&error).is_some(), "{error:#}"); + assert_eq!(store.pull_all()?, replacement); + Ok(()) + } + + #[test] + fn a_write_failure_names_no_value_from_the_snapshot() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + let error = with_write_fault(WriteStep::StagingSync, || { + store.compare_exchange_snapshot(&[], &[snapshot_envelope("SECRET-KEY", b"SECRET-PAYLOAD")]) + }) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::NotReplaced), "{error:#}"); + assert!(!format!("{error:#}").contains("SECRET"), "{error:#}"); + Ok(()) + } + + #[test] + fn a_write_that_landed_is_reported_as_landed_when_the_unlock_after_it_fails() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (mut store, _) = written_store(&path)?; + let replacement = vec![snapshot_envelope("delta", b"4")]; + let exchanged = with_write_fault(WriteStep::Unlock, || { + store.compare_exchange_snapshot(&store.pull_all().unwrap(), &replacement) + })?; + assert!(exchanged); + assert_eq!(store.pull_all()?, replacement); + + let pushed = snapshot_envelope("epsilon", b"5"); + with_write_fault(WriteStep::Unlock, || store.push(&pushed))?; + assert!(with_write_fault(WriteStep::Unlock, || store.pull_all())?.contains(&pushed)); + + // The lock was released: another writer takes it at once. + assert!(matches!( + store.try_compare_exchange_snapshot(&store.pull_all()?, &replacement)?, + TryCompareExchangeSnapshotOutcome::Exchanged + )); + + // The keyed store releases its lock the same way. + let mut keyed = RedbMessagePackBackingStore::new(temp.path().join("keyed.redb"))?; + with_write_fault(WriteStep::Unlock, || keyed.push(&pushed))?; + assert_eq!(keyed.pull_all()?, vec![pushed]); + 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). + #[cfg(unix)] + #[test] + fn r3_decides_resolve_or_refuse_a_push_through_a_live_link_replaces_the_link() -> 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"); + 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"); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn r3_decides_resolve_or_refuse_a_push_all_through_a_dangling_link_replaces_the_link() -> Result<()> { + let temp = tempfile::tempdir()?; + let volume = temp.path().join("volume"); + fs::create_dir(&volume)?; + let link = temp.path().join("store.cc"); + std::os::unix::fs::symlink(volume.join("store.cc"), &link)?; + + SingleFileMessagePackBackingStore::new(&link).push_all(&[snapshot_envelope("a", b"1")], PushAllOptions::default())?; + assert!(fs::symlink_metadata(&link)?.file_type().is_file(), "the link was replaced by a regular file"); + assert_eq!(fs::read_dir(&volume)?.count(), 0, "nothing reached the link's volume"); + Ok(()) + } + + #[test] + fn a_keyed_store_refusal_never_echoes_a_stored_value() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("keyed.redb"); + let envelope = snapshot_envelope("k", b"1"); + RedbMessagePackBackingStore::new(&path)?.push(&envelope)?; + { + let database = redb::Database::create(&path)?; + let write = database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + let keys: Vec> = table + .iter()? + .map(|row| row.map(|(key, _)| key.value().to_vec())) + .collect::>()?; + let mut value = vec![0xab]; + value.extend_from_slice(b"SECRET-REDB"); + for key in keys { + table.insert(key.as_slice(), value.as_slice())?; + } + } + write.commit()?; + } + let store = RedbMessagePackBackingStore::new(&path)?; + for error in [store.pull_all().unwrap_err(), store.compare_and_swap_entry(&envelope, snapshot_envelope("k", b"2")).unwrap_err()] { + assert!(format!("{error:#}").contains("failed to decode redb CultCache envelope"), "{error:#}"); + assert!(!format!("{error:#}").contains("SECRET"), "{error:#}"); + } + Ok(()) + } + + #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] + enum Choice { + Only, + } + + #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] + struct SkippedChoice { + #[serde(default, skip_serializing_if = "Option::is_none")] + choice: Option, + text: String, + } + + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] + #[cultcache(type = "skipped-choice")] + struct ChoiceNote { + #[cultcache(key = 0)] + value: SkippedChoice, + } + + #[test] + fn a_document_decode_refusal_names_the_entry_and_never_a_stored_value() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + // A compact SkippedChoice with no choice is [text], so decoding it reads the text + // where the enum goes, and serde's own message would quote it as an unknown variant. + let secret = ChoiceNote { + value: SkippedChoice { + choice: None, + text: "SECRET-CHOICE".into(), + }, + }; + let mut payload = vec![0x91, 0x91, 0xad]; + payload.extend_from_slice(b"SECRET-CHOICE"); + let malformed = CultCacheEnvelope { + key: "k".into(), + r#type: ::TYPE.into(), + payload, + stored_at: "2026-09-30T00:00:00Z".into(), + schema_id: None, + }; + SingleFileMessagePackBackingStore::new(&path).push(&malformed)?; + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + + let refusals = [ + ("get", cache.get::("k").unwrap_err()), + ("get_all", cache.get_all::().unwrap_err()), + ("get_all_with_keys", cache.get_all_with_keys::().unwrap_err()), + ("put_envelope", cache.put_envelope::(malformed.clone()).unwrap_err()), + ("prepare_entry", cache.prepare_entry("compact", &secret).unwrap_err()), + ]; + for (path, error) in refusals { + let text = format!("{error:#}"); + assert!(!text.contains("SECRET"), "{path}: {text}"); + assert!(text.contains("CultCache"), "{path}: {text}"); + } + Ok(()) + } + + #[test] + fn a_snapshot_the_store_cannot_hold_is_rejected_before_any_write() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, before) = written_store(&path)?; + let mut first = snapshot_envelope("x", b"1"); + first.schema_id = Some("shared".into()); + let mut second = snapshot_envelope("y", b"2"); + second.r#type = "other".into(); + second.schema_id = Some("shared".into()); + let error = store + .compare_exchange_snapshot(&store.pull_all()?, &[first, second]) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::Rejected), "{error:#}"); + assert!(error.to_string().contains("refused a snapshot it cannot store"), "{error}"); + assert!(io_error_in(&error).is_none(), "{error:#}"); + assert_eq!(fs::read(&path)?, before); + Ok(()) + } } diff --git a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts index b7d9684c..b7df1454 100644 --- a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts +++ b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts @@ -1,4 +1,4 @@ -import { mkdir, readFile, rename, rm, writeFile } from "node:fs/promises"; +import { lstat, mkdir, readFile, rename, rm, writeFile } from "node:fs/promises"; import { dirname, resolve } from "node:path"; import { decode, encode } from "@msgpack/msgpack"; @@ -41,64 +41,57 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { } async pullAll(): Promise { - try { - const data = await readFile(this.filePath); - if (data.length === 0) { - return []; - } + // An empty file is not a store, since no CultCache writer leaves one, so it is + // decoded and refused. Only a read that finds nothing at the path is an empty store. + const data = await readStore(this.filePath); + if (data === undefined) { + return []; + } - const decoded = decode(data); - const snapshot = decodeSnapshot(decoded); - if (snapshot) { - return snapshot.records.map((record) => { - const catalogEntry = resolveCatalogEntryForRecord(record, snapshot.catalogBySchemaId); - if (!catalogEntry) { - throw new Error(`CultCache persisted record "${record.key}" references missing schema id "${record.schemaId}".`); - } - - return { - key: record.key, - type: catalogEntry.schemaName, - schemaId: record.schemaId, - catalogEntry, - payload: record.payload, - storedAt: record.storedAt, - }; - }); - } - - const legacy = decodeLegacyEnvelopeArray(decoded); - if (legacy) { - let repairedLegacyPayload = false; - const normalized = legacy.map((entry) => { - const payload = normalizePayload(entry.payload); - if (payload !== entry.payload) { - repairedLegacyPayload = true; - } - - return { - ...entry, - payload, - }; - }); - const parsed = envelopeArraySchema.parse(normalized) as CultCacheEnvelope[]; - - if (repairedLegacyPayload) { - await this.#writeAll(parsed); + const decoded = decode(data); + const snapshot = decodeSnapshot(decoded); + if (snapshot) { + return snapshot.records.map((record) => { + const catalogEntry = resolveCatalogEntryForRecord(record, snapshot.catalogBySchemaId); + if (!catalogEntry) { + throw new Error(`CultCache persisted record "${record.key}" references missing schema id "${record.schemaId}".`); } - return parsed; - } - - throw new Error(`CultCache file ${this.filePath} is not a recognized CultCache MessagePack store.`); - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code === "ENOENT") { - return []; - } - - throw error; + return { + key: record.key, + type: catalogEntry.schemaName, + schemaId: record.schemaId, + catalogEntry, + payload: record.payload, + storedAt: record.storedAt, + }; + }); } + + const legacy = decodeLegacyEnvelopeArray(decoded); + if (legacy) { + let repairedLegacyPayload = false; + const normalized = legacy.map((entry) => { + const payload = normalizePayload(entry.payload); + if (payload !== entry.payload) { + repairedLegacyPayload = true; + } + + return { + ...entry, + payload, + }; + }); + const parsed = envelopeArraySchema.parse(normalized) as CultCacheEnvelope[]; + + if (repairedLegacyPayload) { + await this.#writeAll(parsed); + } + + return parsed; + } + + throw new Error(`CultCache file ${this.filePath} is not a recognized CultCache MessagePack store.`); } async push(entry: CultCacheEnvelope): Promise { @@ -124,16 +117,8 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { async pushAll(entries: CultCacheEnvelope[], options: PushAllOptions = {}): Promise { await this.#enqueue(async () => { - if (options.soft) { - try { - await readFile(this.filePath); - return; - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code !== "ENOENT") { - throw error; - } - } + if (options.soft && (await readStore(this.filePath)) !== undefined) { + return; } await this.#writeAll(entries); @@ -173,6 +158,33 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { } } +/** + * Reads a store file, or returns undefined when nothing is at its path. A dangling symbolic + * link is something: reading it as empty would let the writer rename a file over the link + * and move the store off its volume. Any other failure to reach the file is thrown. + */ +async function readStore(path: string): Promise { + try { + return await readFile(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + throw error; + } + } + + try { + await lstat(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return undefined; + } + + throw error; + } + + throw Object.assign(new Error(`CultCache store ${path} is a symbolic link whose target does not exist.`), { code: "ENOENT" }); +} + async function renameWithRetry(source: string, destination: string): Promise { let delayMs = 10; let lastError: unknown; diff --git a/packages/cultcache-ts/src/store-format.ts b/packages/cultcache-ts/src/store-format.ts index 02de60e7..4f579505 100644 --- a/packages/cultcache-ts/src/store-format.ts +++ b/packages/cultcache-ts/src/store-format.ts @@ -14,20 +14,48 @@ export function isV1Snapshot(decoded: unknown): decoded is unknown[] { if (decoded[0] !== STORE_FORMAT_VERSION) { throw new Error( - `CultCache store format "${decoded[0]}" is not readable; this runtime reads "${STORE_FORMAT_VERSION}" only. ` + - "The store needs a runtime that resolves document variants.", + `CultCache store format ${describeHeader(decoded[0])} is not one this runtime reads; it reads "${STORE_FORMAT_VERSION}" only.`, ); } return true; } +/** + * A store header as a refusal may show it: echoed in the shape `cultcache.store.v`, + * and otherwise described only by its length, since the bytes are the store's. + */ +function describeHeader(header: string): string { + return /^cultcache\.store\.v[0-9]+$/u.test(header) + ? `"${header}"` + : `an unrecognised ${STORE_FORMAT_PREFIX}* header of ${new TextEncoder().encode(header).length} bytes`; +} + +/** + * A record key or schema id as a refusal may show it: a string is an identity and is named; + * anything else is described by its type, since the value is the store's. + */ +function describeIdentity(value: unknown): string { + if (typeof value === "string") { + return `"${value}"`; + } + + const kind = Array.isArray(value) + ? "array" + : value instanceof Uint8Array + ? "bytes" + : value === null + ? "nil" + : typeof value; + return `<${kind}>`; +} + /** Refuses a persisted record with more slots than v1 defines, naming its key and schema id. */ export function requireV1RecordSlots(record: unknown[]): void { if (record.length > PERSISTED_RECORD_SLOTS) { throw new Error( - `CultCache record "${String(record[0])}" (schema "${String(record[1])}") has ${record.length} slots; ` + - `this runtime reads ${PERSISTED_RECORD_SLOTS}. The store needs a runtime that resolves document variants.`, + `CultCache record ${describeIdentity(record[0])} (schema ${describeIdentity(record[1])}) has ${record.length} slots, more than ` + + `the ${PERSISTED_RECORD_SLOTS} of a ${STORE_FORMAT_VERSION} record, so this is not a valid store.`, ); } } diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index 8753af47..b7c51cc1 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/packages/cultcache-ts/test/cult-cache.test.ts @@ -2,7 +2,7 @@ import assert from "node:assert/strict"; import { exec, execFile } from "node:child_process"; import { EventEmitter } from "node:events"; import { existsSync } from "node:fs"; -import { access, mkdtemp, readFile, writeFile } from "node:fs/promises"; +import { access, lstat, mkdir, mkdtemp, readdir, readFile, symlink, writeFile } from "node:fs/promises"; import { homedir, tmpdir } from "node:os"; import { dirname, join, resolve, delimiter } from "node:path"; import { test } from "node:test"; @@ -1450,6 +1450,129 @@ test("SingleFileMessagePackBackingStore refuses the C#-written variant store by ); }); +test("SingleFileMessagePackBackingStore refusals say what was found without blaming variants", async () => { + await assert.rejects( + () => pullVector("unknown-header.msgpack"), + (error: Error) => error.message.includes("not one this runtime reads") && !error.message.includes("variant"), + ); + await assert.rejects( + () => pullVector("extra-slot-full-payload.msgpack"), + (error: Error) => error.message.includes("not a valid store") && !error.message.includes("variant"), + ); +}); + +test("SingleFileMessagePackBackingStore reads only a missing store as empty", async () => { + const dir = await mkdtemp(join(tmpdir(), "cultcache-absent-")); + assert.deepEqual(await new SingleFileMessagePackBackingStore(join(dir, "store.cc")).pullAll(), []); + + // An existing empty file is not a store: no CultCache writer leaves one. + const empty = join(dir, "empty.cc"); + await writeFile(empty, new Uint8Array()); + await assert.rejects(() => new SingleFileMessagePackBackingStore(empty).pullAll()); + assert.equal((await readFile(empty)).length, 0); + + // An empty array is the legacy envelope array with no envelopes: no header, no format claimed. + const legacyEmpty = join(dir, "legacy-empty.cc"); + await writeFile(legacyEmpty, Uint8Array.of(0x90)); + assert.deepEqual(await new SingleFileMessagePackBackingStore(legacyEmpty).pullAll(), []); +}); + +test("SingleFileMessagePackBackingStore reports a store it cannot reach as an I/O error", { skip: process.platform === "win32" }, async () => { + const dir = await mkdtemp(join(tmpdir(), "cultcache-unreachable-")); + const loop = join(dir, "loop.cc"); + await symlink(loop, loop); + const parent = join(dir, "body"); + await writeFile(parent, "not a directory"); + for (const [path, code] of [[loop, "ELOOP"], [join(parent, "store.cc"), "ENOTDIR"]] as const) { + await assert.rejects( + () => new SingleFileMessagePackBackingStore(path).pullAll(), + (error: NodeJS.ErrnoException) => error.code === code, + ); + } +}); + +test("SingleFileMessagePackBackingStore refusals never echo a value from the store", async () => { + const dir = await mkdtemp(join(tmpdir(), "cultcache-canary-")); + const file = join(dir, "store.cc"); + const cases: Array<[unknown[], boolean | undefined]> = [ + [["cultcache.store.v12", [], []], true], + [["cultcache.store.SECRET-HEADER", [], []], false], + [["cultcache.store.v12SECRET", [], []], false], + [["cultcache.store.v", [], []], false], + [["cultcache.store.v1", ["SECRET-CATALOG-TEXT"], []], undefined], + ]; + for (const [value, echoed] of cases) { + await writeFile(file, encode(value)); + const header = value[0] as string; + await assert.rejects( + () => new SingleFileMessagePackBackingStore(file).pullAll(), + (error: Error) => { + if (echoed) { + return error.message.includes(`"${header}"`); + } + return !error.message.includes("SECRET") + && (echoed === undefined || (!error.message.includes(`"${header}"`) && error.message.includes(`of ${header.length} bytes`))); + }, + ); + } +}); + +test("SingleFileMessagePackBackingStore refuses a dangling symlink at its path and writes nothing", { skip: process.platform === "win32" }, async () => { + const dir = await mkdtemp(join(tmpdir(), "cultcache-dangling-")); + const volume = join(dir, "volume"); + await mkdir(volume); + const file = join(dir, "store.cc"); + await symlink(join(volume, "store.cc"), file); + const store = new SingleFileMessagePackBackingStore(file); + const envelope = { key: "k", type: "t", payload: Uint8Array.of(0x90), storedAt: "2026-09-30T00:00:00Z" }; + await assert.rejects(() => store.pullAll(), (error: NodeJS.ErrnoException) => error.code === "ENOENT"); + 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"); + assert.ok((await lstat(file)).isSymbolicLink()); + assert.deepEqual(await readdir(volume), []); +}); + +test("SingleFileMessagePackBackingStore record refusal names only a string key", async () => { + const dir = await mkdtemp(join(tmpdir(), "cultcache-record-key-")); + const file = join(dir, "store.cc"); + const secret = new TextEncoder().encode("SECRET-BYTES"); + for (const [key, schema] of [[["SECRET-IN-ARRAY"], "s"], [secret, "s"], ["k", { SECRET: 1 }]] as const) { + await writeFile(file, encode(["cultcache.store.v1", [], [[key, schema, "t", Uint8Array.of(0x90), "extra"]]])); + await assert.rejects( + () => new SingleFileMessagePackBackingStore(file).pullAll(), + (error: Error) => !error.message.includes("SECRET") && !error.message.includes("83,69,67") && error.message.includes("not a valid store"), + ); + } +}); + +// 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 () => { + const dir = await mkdtemp(join(tmpdir(), "cultcache-live-link-")); + const volume = join(dir, "volume"); + await mkdir(volume); + const target = join(volume, "store.cc"); + const envelope = { key: "a", type: "t", payload: 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"); + await symlink(target, link); + await new SingleFileMessagePackBackingStore(link).push({ ...envelope, key: "b" }); + assert.ok((await lstat(link)).isFile()); + assert.deepEqual(await readFile(target), before); +}); + +test("R3 decides resolve-or-refuse: a pushAll through a dangling link replaces the link", { skip: process.platform === "win32" }, async () => { + const dir = await mkdtemp(join(tmpdir(), "cultcache-dangling-push-all-")); + const volume = join(dir, "volume"); + await mkdir(volume); + const link = join(dir, "store.cc"); + await symlink(join(volume, "store.cc"), link); + await new SingleFileMessagePackBackingStore(link).pushAll([{ key: "a", type: "t", payload: Uint8Array.of(0x90), storedAt: "2026-09-30T00:00:00Z" }]); + assert.ok((await lstat(link)).isFile()); + assert.deepEqual(await readdir(volume), []); +}); + test("CultCache inspector refuses the same vectors by name", async () => { const inspect = async (name: string) => inspectCultCacheBytes(name, await readFile(join(variantVectors, name))); await assert.rejects(() => inspect("unknown-header.msgpack"), /cultcache\.store\.v9/u); diff --git a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs index f591d764..51193806 100644 --- a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs +++ b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs @@ -224,7 +224,7 @@ public static class CultDocumentMessagePackSerialization !string.Equals(version, CultPersistedStoreSnapshot.FormatV2, StringComparison.Ordinal)) { throw new NotSupportedException( - $"Store format {version} is not readable; this runtime reads {CultPersistedStoreSnapshot.FormatV1} and {CultPersistedStoreSnapshot.FormatV2}."); + $"Store format {DescribeHeader(version)} is not readable; this runtime reads {CultPersistedStoreSnapshot.FormatV1} and {CultPersistedStoreSnapshot.FormatV2}."); } // Only a v2 store may hold a variant: a v1 header over a variant record is a store no reader could trust. @@ -239,6 +239,17 @@ public static class CultDocumentMessagePackSerialization } } + // A store header as a refusal may show it: echoed in the shape cultcache.store.v, and otherwise described + // only by its length, since the bytes are the store's. + private static string DescribeHeader(string header) + { + const string prefix = "cultcache.store.v"; + var version = header.StartsWith(prefix, StringComparison.Ordinal) ? header.Substring(prefix.Length) : ""; + return version.Length > 0 && version.All(character => character >= '0' && character <= '9') + ? header + : $"an unrecognised cultcache.store.* header of {System.Text.Encoding.UTF8.GetByteCount(header)} bytes"; + } + private static void WritePersistedRecord(ref MessagePackWriter writer, CultPersistedRecord record) { writer.WriteArrayHeader(record.Variant == null ? PersistedRecordFieldCount : VariantRecordFieldCount); diff --git a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs index 181eca56..e9d8797d 100644 --- a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs +++ b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs @@ -167,6 +167,61 @@ namespace GameCult.Caching.Tests Throws.TypeOf().With.Message.Contains("format version")); } + // Every runtime reads a store file the same way: a missing file is an empty store, an existing empty file is not + // a store (no CultCache writer leaves one), and an empty array carries no header, so it claims no format and is an + // empty store too. + [Test] + public async Task SingleFileRefusesAnEmptyFileAndOpensAMissingFileOrAnEmptyArrayAsEmpty() + { + var root = Path.Combine(Path.GetTempPath(), $"cultlib-empty-{Guid.NewGuid():N}"); + Directory.CreateDirectory(root); + try + { + var empty = Path.Combine(root, "empty.msgpack"); + File.WriteAllBytes(empty, Array.Empty()); + using (var cache = new CultCache()) + { + Assert.Catch(() => cache.AddBackingStore(new SingleFileMessagePackBackingStore(empty))); + Assert.That(cache.BackingStores, Is.Empty); + } + Assert.That(File.ReadAllBytes(empty), Is.Empty); + + var emptyArray = Path.Combine(root, "empty-array.msgpack"); + File.WriteAllBytes(emptyArray, new byte[] { 0x90 }); + foreach (var path in new[] { Path.Combine(root, "missing.msgpack"), emptyArray }) + { + using var cache = new CultCache(); + cache.AddBackingStore(new SingleFileMessagePackBackingStore(path)); + await cache.PullAllBackingStoresAsync(); + Assert.That(cache.GetAll(), Is.Empty, path); + } + } + finally + { + if (Directory.Exists(root)) Directory.Delete(root, recursive: true); + } + } + + // A refusal echoes a header only in the shape cultcache.store.v; any other is described by its length. + [TestCase("cultcache.store.v12", true)] + [TestCase("cultcache.store.SECRET-HEADER", false)] + [TestCase("cultcache.store.v12SECRET", false)] + [TestCase("cultcache.store.v", false)] + public void SingleFileRefusalEchoesAHeaderOnlyInTheKnownShape(string header, bool echoed) + { + var snapshot = new CultPersistedStoreSnapshot { FormatVersion = header }; + var error = Assert.Throws(() => CultDocumentMessagePackSerialization.RequireSingleFileFormat(snapshot))!; + if (echoed) + { + Assert.That(error.Message, Does.Contain("format " + header + " is not")); + } + else + { + Assert.That(error.Message, Does.Not.Contain("SECRET").And.Not.Contain("format " + header + " is not")); + Assert.That(error.Message, Does.Contain($"of {header.Length} bytes")); + } + } + [Test] public void V1StoreWrittenAtTheBaseCommitStillDecodesByteForByte() {