Merge hands/whole-store-writers-read-header: Rust push_all and TS pushAll read the header under the lock and refuse a newer store format (cultcache-rs 0.3.1; Soul s2 holds)

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

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.

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

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()?;

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.
 

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<void> {
     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);

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;
 }
 
 /**

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<string, number[]> = {
+    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(), []);

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<65>’”¥alpha¬vectors.item¹2026-09-29T00:00:00+00:00Ä’¥alpha
”¤beta¬vectors.item¹2026-09-29T00:00:01+00:00Ä’¤beta