diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md new file mode 100644 index 00000000..4fa680b9 --- /dev/null +++ b/packages/cultcache-rs/CHANGELOG.md @@ -0,0 +1,12 @@ +# Changelog + +All notable changes to `cultcache-rs` are recorded here, following +[Keep a Changelog](https://keepachangelog.com/en/1.1.0/). Earlier releases are recorded +only in git history. + +## [0.3.1] + +### Fixed +- `push_all` reads the store's header under the lock and refuses a format this runtime does + not read, with `CultCacheStoreUnreadable` `UnsupportedFormat`, leaving the file as it was. + A file with no readable stamp is still replaced, so a whole write heals it. diff --git a/packages/cultcache-rs/Cargo.toml b/packages/cultcache-rs/Cargo.toml index 9610aeb6..7954370c 100644 --- a/packages/cultcache-rs/Cargo.toml +++ b/packages/cultcache-rs/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cultcache-rs" -version = "0.3.0" +version = "0.3.1" edition = "2024" license = "MIT" publish = false diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index ecb69145..2818c638 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -1261,28 +1261,40 @@ impl SingleFileMessagePackBackingStore { .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 == STORE_FORMAT_V1 => decode_store_snapshot(&bytes) - .map_err(|error| unreadable(CultCacheStoreUnreadableKind::Undecodable, error)), - Some(header) if header.starts_with("cultcache.store.") => Err(unreadable( + self.refuse_unread_format(&bytes)?; + if store_header(&bytes).as_deref() == Some(STORE_FORMAT_V1) { + decode_store_snapshot(&bytes) + .map_err(|error| self.unreadable(CultCacheStoreUnreadableKind::Undecodable, error)) + } else { + decode_grown(&bytes) + .context("failed to decode legacy CultCache envelope array") + .map_err(|error| self.unreadable(CultCacheStoreUnreadableKind::Undecodable, error)) + } + } + + // The stamp rule, owned once for the reader and the whole-store writer: a + // cultcache.store.* header other than v1 is a format this runtime does not read. Only + // the header is judged, so a newer store cut short is still a newer store. + fn refuse_unread_format(&self, bytes: &[u8]) -> Result<()> { + match store_header(bytes) { + Some(header) if header != STORE_FORMAT_V1 && header.starts_with("cultcache.store.") => Err(self.unreadable( CultCacheStoreUnreadableKind::UnsupportedFormat, anyhow!("its header is {}; this runtime reads {STORE_FORMAT_V1:?} only", describe_header(&header)), )), - _ => decode_grown(&bytes) - .context("failed to decode legacy CultCache envelope array") - .map_err(|error| unreadable(CultCacheStoreUnreadableKind::Undecodable, error)), + _ => Ok(()), } } + fn unreadable(&self, kind: CultCacheStoreUnreadableKind, detail: anyhow::Error) -> 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, + }) + } + /// Reads one filesystem snapshot without creating or opening the sibling /// CultCache lock file. /// @@ -1571,6 +1583,11 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result<()> { self.with_exclusive_lock(|| { + // A file it cannot read is written as before (a dangling link is replaced, R3; a + // directory is a typed write failure); only a stamp it can read is judged. + if let Ok(bytes) = fs::read(&self.path) { + self.refuse_unread_format(&bytes)?; + } let mut entries = entries.to_vec(); entries.sort_by_key(entry_id); self.write_all_unlocked(&entries) @@ -5437,6 +5454,41 @@ mod tests { Ok(()) } + #[test] + fn push_all_refuses_a_store_in_a_format_this_runtime_does_not_read_and_leaves_it() -> Result<()> { + let vector = fs::read( + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors/document-variants-c2a/v3-base.msgpack"), + )?; + let entry = CultCacheEnvelope { + key: "probe".into(), + r#type: "probe".into(), + payload: vec![0x80], + stored_at: "2026-10-22T00:00:00Z".into(), + schema_id: None, + }; + // A newer store whole, and cut short after its header: the header alone is judged. + for bytes in [&vector[..], &vector[..vector.len() / 2]] { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.msgpack"); + fs::write(&path, bytes)?; + let error = SingleFileMessagePackBackingStore::new(&path) + .push_all(std::slice::from_ref(&entry), PushAllOptions::default()) + .unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::UnsupportedFormat), "{error:#}"); + assert!(format!("{error:#}").contains("cultcache.store.v3"), "{error:#}"); + assert_eq!(fs::read(&path)?, bytes); + } + // A file with no readable stamp is not a newer store: a whole write still heals it. + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.msgpack"); + fs::write(&path, [0x93u8, 0xa0])?; + let mut store = SingleFileMessagePackBackingStore::new(&path); + store.push_all(std::slice::from_ref(&entry), PushAllOptions::default())?; + assert_eq!(store.pull_all()?.len(), 1); + Ok(()) + } + + #[test] fn a_store_that_cannot_be_read_is_an_io_error_and_never_unreadable() -> Result<()> { let temp = tempfile::tempdir()?; diff --git a/packages/cultcache-ts/CHANGELOG.md b/packages/cultcache-ts/CHANGELOG.md index 5c7a1561..6304d347 100644 --- a/packages/cultcache-ts/CHANGELOG.md +++ b/packages/cultcache-ts/CHANGELOG.md @@ -23,6 +23,13 @@ All notable changes to `@gamecult/cultcache-ts` are recorded here, following - Opening a legacy store that needs healing rewrites it under the lock, from a read taken under the lock. +### Fixed +- A full `pushAll` reads the store's header under the lock first and refuses a store format + this runtime does not read (the same error a read gives), leaving the file as it was. A + file with no readable stamp is still replaced, so a whole write heals it. Soft `pushAll` + is unchanged. The stamp is decoded by the decoder a read uses, so a stamp followed by bytes + that are not UTF-8 is refused by a whole write as a read refuses it. + ### Dependencies - `koffi` 3.3.0, loaded only on the first write. 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 f10b146b..f0641e66 100644 --- a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts +++ b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts @@ -7,7 +7,7 @@ import { z } from "zod"; import { documentPayloadBytes, documentPayloadFromBytes } from "./document"; import { checkPayloadDepth } from "./payload-depth"; import { commitStoreFile, withStoreWriteLock } from "./store-file"; -import { STORE_FORMAT_VERSION, isV1Snapshot, requireV1RecordSlots } from "./store-format"; +import { STORE_FORMAT_VERSION, isV1Snapshot, refuseUnreadStoreBytes, requireV1RecordSlots } from "./store-format"; import type { CacheBackingStore, CultCacheEnvelope, @@ -135,8 +135,14 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { async pushAll(entries: CultCacheEnvelope[], options: PushAllOptions = {}): Promise { await this.#enqueue(async () => { - if (options.soft && (await readStore(this.filePath)) !== undefined) { - return; + // A whole write of a file it cannot read proceeds as before (a dangling link is + // replaced, R3); only a stamp it can read is judged. + const data = options.soft ? await readStore(this.filePath) : await readStore(this.filePath).catch(() => undefined); + if (data !== undefined) { + if (options.soft) { + return; + } + refuseUnreadStoreBytes(data); } await this.#writeAll(entries); diff --git a/packages/cultcache-ts/src/store-format.ts b/packages/cultcache-ts/src/store-format.ts index 4f579505..197dc659 100644 --- a/packages/cultcache-ts/src/store-format.ts +++ b/packages/cultcache-ts/src/store-format.ts @@ -1,3 +1,5 @@ +import { decode } from "@msgpack/msgpack"; + export const STORE_FORMAT_VERSION = "cultcache.store.v1"; const STORE_FORMAT_PREFIX = "cultcache.store."; @@ -12,13 +14,54 @@ export function isV1Snapshot(decoded: unknown): decoded is unknown[] { return false; } - if (decoded[0] !== STORE_FORMAT_VERSION) { + refuseUnreadFormat(decoded[0]); + return true; +} + +/** + * Refuses a store whose header names a format this runtime does not read, judging the header + * alone: the leading array's first element, when it is a complete string. Bytes with no such + * header (a legacy array, an empty, truncated or garbled file) are not refused here. + */ +export function refuseUnreadStoreBytes(bytes: Uint8Array): void { + const header = peekHeader(bytes); + if (header !== undefined && header.startsWith(STORE_FORMAT_PREFIX)) { + refuseUnreadFormat(header); + } +} + +function refuseUnreadFormat(header: string): void { + if (header !== STORE_FORMAT_VERSION) { throw new Error( - `CultCache store format ${describeHeader(decoded[0])} is not one this runtime reads; it reads "${STORE_FORMAT_VERSION}" only.`, + `CultCache store format ${describeHeader(header)} is not one this runtime reads; it reads "${STORE_FORMAT_VERSION}" only.`, ); } +} - return true; +// The first element of a leading non-empty MessagePack array when it is a complete string; +// anything else is undefined. The string is decoded by the decoder a read uses, so a stamp +// gets one verdict from a read and from a whole write. +function peekHeader(bytes: Uint8Array): string | undefined { + const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength); + const lead = bytes[0]; + let offset: number; + if (lead === undefined) return undefined; + if (lead >= 0x91 && lead <= 0x9f) offset = 1; + else if (lead === 0xdc && bytes.length >= 3 && view.getUint16(1) > 0) offset = 3; + else if (lead === 0xdd && bytes.length >= 5 && view.getUint32(1) > 0) offset = 5; + else return undefined; + const start = offset; + const tag = bytes[offset]; + let length: number; + if (tag === undefined) return undefined; + if (tag >= 0xa0 && tag <= 0xbf) { length = tag & 0x1f; offset += 1; } + else if (tag === 0xd9 && bytes.length >= offset + 2) { length = bytes[offset + 1]!; offset += 2; } + else if (tag === 0xda && bytes.length >= offset + 3) { length = view.getUint16(offset + 1); offset += 3; } + else if (tag === 0xdb && bytes.length >= offset + 5) { length = view.getUint32(offset + 1); offset += 5; } + else return undefined; + if (bytes.length < offset + length) return undefined; + const header = decode(bytes.subarray(start, offset + length)); + return typeof header === "string" ? header : undefined; } /** diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index 86d4a090..d767c599 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/packages/cultcache-ts/test/cult-cache.test.ts @@ -15,6 +15,7 @@ import { CultCache } from "../src/cult-cache"; import { inspectCultCacheBytes } from "../src/cult-cache-inspector"; import { defineDocumentRegistry, defineDocumentType, documentPayloadBytes, documentPayloadFromBytes } from "../src/document"; import { SingleFileMessagePackBackingStore } from "../src/single-file-messagepack-backing-store"; +import { refuseUnreadStoreBytes } from "../src/store-format"; import type { CacheBackingStore, CultCacheEnvelope, CultCacheSchema } from "../src/types"; const execFileAsync = promisify(execFile); @@ -1481,6 +1482,118 @@ test("SingleFileMessagePackBackingStore refusals say what was found without blam ); }); +test("pushAll refuses a store in a format this runtime does not read and leaves it byte-identical", async () => { + const vector = await readFile(resolve(cultLibRoot, "tests", "vectors", "document-variants-c2a", "v3-base.msgpack")); + const envelope = { key: "probe", type: "probe", payload: documentPayloadFromBytes(new Uint8Array([0x80])), storedAt: "2026-10-22T00:00:00Z" }; + for (const bytes of [vector, vector.subarray(0, Math.floor(vector.length / 2))]) { + const file = join(await mkdtemp(join(tmpdir(), "cultcache-pushall-v3-")), "store.msgpack"); + await writeFile(file, bytes); + await assert.rejects(() => new SingleFileMessagePackBackingStore(file).pushAll([envelope]), /cultcache\.store\.v3/u); + assert.equal(Buffer.compare(await readFile(file), bytes), 0); + } + const healed = join(await mkdtemp(join(tmpdir(), "cultcache-pushall-heal-")), "store.msgpack"); + await writeFile(healed, new Uint8Array([0x93, 0xa0])); + const store = new SingleFileMessagePackBackingStore(healed); + await store.pushAll([envelope]); + assert.equal((await store.pullAll()).length, 1); +}); + +function headerRefused(bytes: Uint8Array): boolean { + try { + refuseUnreadStoreBytes(bytes); + return false; + } catch (error) { + assert.match((error as Error).message, /not one this runtime reads/u); + return true; + } +} + +function storeStart(arrayLead: number[], text: string, strForm: "fix" | "8" | "16" | "32"): Uint8Array { + const body = Buffer.from(text, "utf8"); + const n = body.length; + const tag = + strForm === "fix" ? [0xa0 | n] + : strForm === "8" ? [0xd9, n] + : strForm === "16" ? [0xda, n >> 8, n & 0xff] + : [0xdb, 0, 0, n >> 8, n & 0xff]; + return new Uint8Array([...arrayLead, ...tag, ...body]); +} + +test("a store's header is judged from its first bytes, in every MessagePack array and string form", () => { + const leads = [[0x91], [0x9f], [0xdc, 0x00, 0x01], [0xdc, 0x01, 0x00], [0xdd, 0x00, 0x00, 0x00, 0x01], [0xdd, 0x00, 0x01, 0x00, 0x00]]; + for (const lead of leads) { + for (const form of ["fix", "8", "16", "32"] as const) { + const v3 = storeStart(lead, "cultcache.store.v3", form); + assert.equal(headerRefused(v3), true, `${lead} ${form} v3`); + assert.equal(headerRefused(v3.subarray(0, v3.length - 1)), false, `${lead} ${form} v3 one byte short`); + assert.equal(headerRefused(storeStart(lead, "cultcache.store.v1", form)), false, `${lead} ${form} v1`); + assert.equal(headerRefused(storeStart(lead, "other.store.v3", form)), false, `${lead} ${form} other`); + assert.equal(headerRefused(storeStart(lead, "", form)), false, `${lead} ${form} empty`); + } + } + // The longest fixstr, and a string long enough to need str16. + assert.equal(headerRefused(storeStart([0x91], `cultcache.store.v3${"x".repeat(13)}`, "fix")), true); + assert.equal(headerRefused(storeStart([0x91], `cultcache.store.v3${"x".repeat(300)}`, "16")), true); + assert.equal(headerRefused(storeStart([0x91], `cultcache.store.v3${"x".repeat(300)}`, "32")), true); + // A view into a larger buffer is read from its own start. + const padded = new Uint8Array([9, 9, ...storeStart([0x91], "cultcache.store.v3", "fix")]); + assert.equal(headerRefused(padded.subarray(2)), true); +}); + +test("a stamp followed by bytes that are not UTF-8 gets the verdict a read gives", async () => { + const stamp = [...Buffer.from("cultcache.store.v3", "utf8")]; + const forms = { fixstr: [0x91, 0xb3, ...stamp, 0xff], str8: [0x91, 0xd9, 19, ...stamp, 0xff] }; + const envelope = { key: "probe", type: "probe", payload: documentPayloadFromBytes(new Uint8Array([0x80])), storedAt: "2026-10-22T00:00:00Z" }; + for (const [name, bytes] of Object.entries(forms)) { + const file = join(await mkdtemp(join(tmpdir(), "cultcache-stamp-")), "store.msgpack"); + await writeFile(file, new Uint8Array(bytes)); + const store = new SingleFileMessagePackBackingStore(file); + await assert.rejects(() => store.pullAll(), /not one this runtime reads/u, `${name} read`); + await assert.rejects(() => store.pushAll([envelope]), /not one this runtime reads/u, `${name} pushAll`); + assert.deepEqual([...(await readFile(file))], bytes, `${name} left as it was`); + } +}); + +test("bytes with no readable store stamp are not refused", () => { + const stamp = [...Buffer.from("cultcache.store.v3", "utf8")]; + const cases: Record = { + empty: [], + "empty array": [0x90], + "empty array16": [0xdc, 0x00, 0x00, 0xb2, ...stamp], + "empty array32": [0xdd, 0x00, 0x00, 0x00, 0x00, 0xb2, ...stamp], + "array16 cut before its count": [0xdc, 0x00], + "array32 cut before its count": [0xdd, 0x00, 0x00, 0x00], + "array with no element": [0x91], + "first element is not a string": [0x92, 0x01, 0xb2, ...stamp], + "a bare string": [0xb2, ...stamp], + "a map": [0x81, 0xb2, ...stamp], + "a map whose bytes read as an array32 count": [0x81, 0x00, 0x00, 0x00, 0x01, 0xb2, ...stamp], + "a small integer before a stamp-looking body": [0x91, 0x12, ...stamp], + "a uint32 before a stamp-looking body": [0x91, 0xce, 0x00, 0x00, 0x00, 0x12, ...stamp], + "str8 cut before its length": [0x91, 0xd9], + "str16 cut before its length": [0x91, 0xda, 0x00], + "str32 cut before its length": [0x91, 0xdb, 0x00, 0x00, 0x00], + }; + for (const [name, bytes] of Object.entries(cases)) { + assert.equal(headerRefused(new Uint8Array(bytes)), false, name); + } +}); + +test("pushAll leaves a store in a newer format alone when soft, and replaces a store it reads", async () => { + const vector = await readFile(resolve(cultLibRoot, "tests", "vectors", "document-variants-c2a", "v3-base.msgpack")); + const envelope = { key: "probe", type: "probe", payload: documentPayloadFromBytes(new Uint8Array([0x80])), storedAt: "2026-10-22T00:00:00Z" }; + const newer = join(await mkdtemp(join(tmpdir(), "cultcache-pushall-soft-")), "store.msgpack"); + await writeFile(newer, vector); + await new SingleFileMessagePackBackingStore(newer).pushAll([envelope], { soft: true }); + assert.equal(Buffer.compare(await readFile(newer), vector), 0); + + const own = join(await mkdtemp(join(tmpdir(), "cultcache-pushall-own-")), "store.msgpack"); + const store = new SingleFileMessagePackBackingStore(own); + await store.pushAll([envelope]); + await store.pushAll([{ ...envelope, key: "second" }]); + assert.deepEqual((await store.pullAll()).map((entry) => entry.key), ["second"]); +}); + 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(), []); diff --git a/tests/vectors/document-variants-c2a/v3-base.msgpack b/tests/vectors/document-variants-c2a/v3-base.msgpack new file mode 100644 index 00000000..9fc9b64f --- /dev/null +++ b/tests/vectors/document-variants-c2a/v3-base.msgpack @@ -0,0 +1 @@ +“²cultcache.store.v3‘—¬vectors.item¬vectors.item¯vectors.item.v1¬vectors.itemÙL{"schemaName":"vectors.item","schemaVersion":"vectors.item.v1","members":[]}‘¬vectors.item’”¥alpha¬vectors.item¹2026-09-29T00:00:00+00:00Ä’¥alpha”€beta¬vectors.item¹2026-09-29T00:00:01+00:00Ä’€beta \ No newline at end of file