Python: StoreUnreadableError for a store file it cannot read; a store is exactly three slots

The refusal is a ValueError subclass with the cause chained. A store with more than three slots is refused instead of read.
The readability test walks every vector in the folder, asserts the error type and cause, the header a replaced file carries
and what it holds.

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

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

@@ -9,7 +9,7 @@ from .documents import (
     define_document_registry,
     define_document_type,
 )
-from .stores import JsonLinesBackingStore, SingleFileMessagePackBackingStore
+from .stores import JsonLinesBackingStore, SingleFileMessagePackBackingStore, StoreUnreadableError
 
 __all__ = [
     "BackingStore",
@@ -22,6 +22,7 @@ __all__ = [
     "DocumentDefinition",
     "JsonLinesBackingStore",
     "SingleFileMessagePackBackingStore",
+    "StoreUnreadableError",
     "database_entry_field",
     "define_database_entry_type",
     "define_document_registry",

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

@@ -23,6 +23,16 @@ _STORE_FORMAT_PREFIX = "cultcache.store."
 _PERSISTED_RECORD_SLOTS = 4
 
 
+class StoreUnreadableError(ValueError):
+    """A store file this runtime cannot read: not exactly one complete store (truncated, bytes after it, a missing or extra
+    slot), a header or record it does not know, or a body it cannot decode. Open, push, delete and push_all refuse a file
+    with this error, and a refused file is left as it was. The cause is `__cause__`."""
+
+    def __init__(self, path: Path, cause: BaseException) -> None:
+        super().__init__(f"CultCache store {path} is not readable: {cause}")
+        self.path = path
+
+
 class JsonLinesBackingStore:
     def __init__(self, path: str | os.PathLike[str]) -> None:
         self.path = Path(path)
@@ -102,12 +112,15 @@ class SingleFileMessagePackBackingStore:
             data = self.path.read_bytes()
             if not data:
                 return []
-            decoded = msgpack.unpackb(data, raw=False)
-            snapshot = _decode_snapshot(decoded)
-            if snapshot is not None:
-                self._format, envelopes = snapshot
-                return envelopes
-            return _decode_legacy_envelopes(decoded, msgpack)
+            try:
+                decoded = msgpack.unpackb(data, raw=False)
+                snapshot = _decode_snapshot(decoded)
+                if snapshot is not None:
+                    self._format, envelopes = snapshot
+                    return envelopes
+                return _decode_legacy_envelopes(decoded, msgpack)
+            except Exception as exc:
+                raise StoreUnreadableError(self.path, exc) from exc
 
     def push(self, envelope: CultCacheEnvelope) -> None:
         with self._lock:
@@ -174,7 +187,9 @@ def _decode_snapshot(decoded: Any) -> tuple[str, list[CultCacheEnvelope]] | None
         )
     if decoded[0] not in _READABLE_STORE_FORMATS:
         return None
-    if len(decoded) < 3 or not isinstance(decoded[1], list) or not isinstance(decoded[2], list):
+    if len(decoded) != 3:
+        raise ValueError(f"CultCache store has {len(decoded)} top-level slots; this runtime reads 3 (header, schema catalog, records)")
+    if not isinstance(decoded[1], list) or not isinstance(decoded[2], list):
         raise ValueError("CultCache v1 snapshot must contain a schema catalog and record array")
 
     catalog_by_schema_id: dict[str, CultCacheSchemaCatalogEntry] = {}

packages/cultcache-py/tests/test_cultcache.py

@@ -11,6 +11,7 @@ from cultcache_py import (
     CultCache,
     JsonLinesBackingStore,
     SingleFileMessagePackBackingStore,
+    StoreUnreadableError,
     define_database_entry_type,
     define_document_type,
 )
@@ -899,7 +900,7 @@ class CultCacheTests(unittest.TestCase):
                 store_path = Path(tmp) / "store.msgpack"
                 store_path.write_bytes(vector.read_bytes())
                 before = store_path.read_bytes()
-                with self.assertRaisesRegex(ValueError, "is not readable"):
+                with self.assertRaisesRegex(StoreUnreadableError, "is not readable"):
                     SingleFileMessagePackBackingStore(store_path).push_all(envelopes)
                 self.assertEqual(store_path.read_bytes(), before, f"{vector.name} was rewritten")
 
@@ -907,13 +908,19 @@ class CultCacheTests(unittest.TestCase):
     # pull_all, push_all, push and delete alike. The bytes and every runtime's verdict:
     # tests/vectors/document-variants-c2a/readability.
     def test_single_file_replaces_a_file_exactly_when_it_reads(self) -> None:
+        import msgpack  # type: ignore
+
         root = self._C2A_VECTORS / "readability"
         rows = [
             line.split()
             for line in (root / "manifest.txt").read_text(encoding="utf-8").splitlines()
             if line and not line.startswith("#")
         ]
-        self.assertEqual(len(rows), 15)
+        # Every vector in the folder has a manifest row.
+        self.assertEqual(
+            sorted(path.name for path in root.glob("*.bin")),
+            sorted(vector for vector, *_ in rows if not vector.startswith("..")),
+        )
         envelopes = SingleFileMessagePackBackingStore(self._C2A_VECTORS / "v3-base.msgpack").pull_all()
         for vector, _, _, _, python in rows:
             content = (root / vector).read_bytes()
@@ -930,9 +937,23 @@ class CultCacheTests(unittest.TestCase):
                     }[operation]
                     if python == "reads":
                         run()
+                        if operation == "pull_all":
+                            continue
+                        # The replaced file is a store: it carries the header its old content decides (a v3 store keeps its
+                        # marker, all else is v1), and it holds what the operation wrote.
+                        expected = "cultcache.store.v3" if vector.endswith("v3-base.msgpack") else "cultcache.store.v1"
+                        self.assertEqual(msgpack.unpackb(store_path.read_bytes(), raw=False)[0], expected, f"{vector} {operation}")
+                        keys = [envelope.key for envelope in SingleFileMessagePackBackingStore(store_path).pull_all()]
+                        if operation == "push_all":
+                            self.assertEqual(sorted(set(keys) & {e.key for e in envelopes}), sorted(e.key for e in envelopes), f"{vector} {operation}")
+                        elif operation == "push":
+                            self.assertIn(envelopes[0].key, keys, f"{vector} {operation}")
+                        else:
+                            self.assertNotIn(envelopes[0].key, keys, f"{vector} {operation}")
                     else:
-                        with self.assertRaises(Exception, msg=f"{vector} {operation}"):
+                        with self.assertRaises(StoreUnreadableError, msg=f"{vector} {operation}") as refused:
                             run()
+                        self.assertIsNotNone(refused.exception.__cause__, f"{vector} {operation}")
                         self.assertEqual(store_path.read_bytes(), content, f"{vector} {operation} rewrote a file it cannot read")