Merge CultCache R1: typed read failures, no empty-file stores, no echoed values, typed write outcome
Only "nothing at the path" reads as an empty store; a zero-byte file is Undecodable in every runtime; a dangling link is an I/O error. Errors name keys and schema ids only, never stored values, and echo a header only in the cultcache.store.v<digits> shape. Single-file writes report NotReplaced / ReplacedNotDurable / Rejected; an unlock failure after a successful write no longer replaces the write's result. Soul passes 1-3; batch 4 closes pass 3. Merged tree 9d9e76d4 verified on Yggdrasil: Rust (cultcache, cultnet, cultmesh), cultcache-ts 44 including the four-runtime interop test, cultnet-ts 198, cultcache-py 47, GameCult.Caching.Tests 346. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
commit
daedfdd658
9 changed files with 1486 additions and 165 deletions
|
|
@ -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<digits>; 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:
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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::<CultCacheStoreUnreadable>().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<digits>`.
|
||||
|
||||
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::<CultCacheStoreWriteFailed>().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**
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -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<CultCacheEnvelope[]> {
|
||||
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<void> {
|
||||
|
|
@ -124,16 +117,8 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore {
|
|||
|
||||
async pushAll(entries: CultCacheEnvelope[], options: PushAllOptions = {}): Promise<void> {
|
||||
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<Uint8Array | undefined> {
|
||||
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<void> {
|
||||
let delayMs = 10;
|
||||
let lastError: unknown;
|
||||
|
|
|
|||
|
|
@ -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<digits>`,
|
||||
* 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.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<digits>, 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);
|
||||
|
|
|
|||
|
|
@ -167,6 +167,61 @@ namespace GameCult.Caching.Tests
|
|||
Throws.TypeOf<NotSupportedException>().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<byte>());
|
||||
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<PreCut2FixtureItem>(), 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<digits>; 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<NotSupportedException>(() => 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()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue