TypeScript: a flush asks the same full reader as an open, and refuses with StoreUnreadableError

pushAll, push and delete read the file through the one store reader that pullAll uses, so a store holding a variant or a bad
body is not replaced by a header-only check. A store with fewer or more than three slots is refused. 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-ts/src/index.ts

@@ -2,6 +2,7 @@ export { CultCache, CultCacheBuilder } from "./cult-cache";
 export { inspectCultCacheBytes } from "./cult-cache-inspector";
 export { defineDocumentRegistry, defineDocumentType } from "./document";
 export { SingleFileMessagePackBackingStore } from "./single-file-messagepack-backing-store";
+export { StoreUnreadableError } from "./store-format";
 export type {
   AnyCultCacheDocumentDefinition,
   CacheBackingStore,

packages/cultcache-ts/src/single-file-messagepack-backing-store.ts

@@ -4,7 +4,7 @@ import { dirname, resolve } from "node:path";
 import { decode, encode } from "@msgpack/msgpack";
 import { z } from "zod";
 
-import { STORE_FORMAT_VERSION, type StoreFormat, isStoreSnapshot, requireV1RecordSlots, storeHeader } from "./store-format";
+import { STORE_FORMAT_VERSION, type StoreFormat, StoreUnreadableError, isStoreSnapshot, requireStoreSlots, requireV1RecordSlots } from "./store-format";
 import type {
   CacheBackingStore,
   CultCacheEnvelope,
@@ -46,65 +46,13 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore {
   async pullAll(): Promise<CultCacheEnvelope[]> {
     // The disk decides the header: a file that is gone, empty or legacy is not marked.
     this.#format = STORE_FORMAT_VERSION;
-    try {
-      const data = await readFile(this.filePath);
-      if (data.length === 0) {
-        return [];
-      }
-
-      const decoded = decode(data);
-      const snapshot = decodeSnapshot(decoded);
-      if (snapshot) {
-        this.#format = snapshot.format;
-        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);
-        }
-
-        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;
+    const disk = await this.#readDisk();
+    this.#format = disk.format;
+    if (disk.repairedLegacyPayload) {
+      await this.#writeAll(disk.envelopes);
     }
+
+    return disk.envelopes;
   }
 
   async push(entry: CultCacheEnvelope): Promise<void> {
@@ -143,33 +91,36 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore {
       }
 
       // A flush of the whole store writes the header the file on disk carries, read now: a file marked for element ids
-      // stays marked, and one that is not (or is gone, empty or legacy) is written unmarked. A file `pullAll` would refuse
-      // (not exactly one store, not a legacy envelope array) is refused and left as it is.
-      await this.#readDiskFormat();
+      // stays marked, and one that is not (or is gone, empty or legacy) is written unmarked. The file is read by the same
+      // reader as `pullAll`, so one it would refuse (not exactly one store, a variant, a body it cannot decode) is refused
+      // and left as it is.
+      this.#format = (await this.#readDisk()).format;
       await this.#writeAll(entries);
     });
   }
 
-  async #readDiskFormat(): Promise<void> {
+  // The one reader of the store file, asked by open, push, delete and flush alike. A file that is gone or zero bytes is an
+  // empty unmarked store; anything else must decode completely or the read throws StoreUnreadableError.
+  async #readDisk(): Promise<DiskStore> {
+    let data: Uint8Array;
     try {
-      const data = await readFile(this.filePath);
-      let header: StoreFormat | undefined;
-      if (data.length > 0) {
-        const decoded = decode(data);
-        header = storeHeader(decoded);
-        if (header === undefined) {
-          // Not a store header: only a legacy envelope array is replaced, and only one `pullAll` would read.
-          decodeLegacyEnvelopeArray(decoded);
-        }
-      }
-
-      this.#format = header ?? STORE_FORMAT_VERSION;
+      data = await readFile(this.filePath);
     } catch (error) {
-      if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
-        throw error;
+      if ((error as NodeJS.ErrnoException).code === "ENOENT") {
+        return { format: STORE_FORMAT_VERSION, envelopes: [], repairedLegacyPayload: false };
       }
 
-      this.#format = STORE_FORMAT_VERSION;
+      throw error;
+    }
+
+    if (data.length === 0) {
+      return { format: STORE_FORMAT_VERSION, envelopes: [], repairedLegacyPayload: false };
+    }
+
+    try {
+      return decodeStoreFile(data);
+    } catch (error) {
+      throw new StoreUnreadableError(this.filePath, error);
     }
   }
 
@@ -317,11 +268,61 @@ function encodeCatalogMember(member: CultCacheSchemaCatalogMember): unknown[] {
   ];
 }
 
+type DiskStore = {
+  format: StoreFormat;
+  envelopes: CultCacheEnvelope[];
+  // A legacy file whose payloads were not bytes: reading it repairs them, and the store rewrites the file.
+  repairedLegacyPayload: boolean;
+};
+
+function decodeStoreFile(data: Uint8Array): DiskStore {
+  const decoded = decode(data);
+  const snapshot = decodeSnapshot(decoded);
+  if (snapshot) {
+    const envelopes = snapshot.records.map((record): CultCacheEnvelope => {
+      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,
+      };
+    });
+    return { format: snapshot.format, envelopes, repairedLegacyPayload: false };
+  }
+
+  const legacy = decodeLegacyEnvelopeArray(decoded);
+  if (!legacy) {
+    throw new Error("CultCache file is not a recognized CultCache MessagePack store.");
+  }
+
+  let repairedLegacyPayload = false;
+  const normalized = legacy.map((entry) => {
+    const payload = normalizePayload(entry.payload);
+    if (payload !== entry.payload) {
+      repairedLegacyPayload = true;
+    }
+
+    return {
+      ...entry,
+      payload,
+    };
+  });
+  return { format: STORE_FORMAT_VERSION, envelopes: envelopeArraySchema.parse(normalized) as CultCacheEnvelope[], repairedLegacyPayload };
+}
+
 function decodeSnapshot(decoded: unknown): DecodedSnapshot | undefined {
   if (!isStoreSnapshot(decoded)) {
     return undefined;
   }
 
+  requireStoreSlots(decoded);
   const catalogRaw = decoded[1];
   const recordsRaw = decoded[2];
   if (!Array.isArray(catalogRaw) || !Array.isArray(recordsRaw)) {

packages/cultcache-ts/src/store-format.ts

@@ -11,6 +11,7 @@ export type StoreFormat = typeof STORE_FORMAT_VERSION | typeof STORE_FORMAT_ELEM
 
 const STORE_FORMAT_PREFIX = "cultcache.store.";
 const PERSISTED_RECORD_SLOTS = 4;
+const STORE_SLOTS = 3;
 
 /**
  * True for a v1 or v3 snapshot, false for anything that is not a CultCache store header (the legacy
@@ -32,25 +33,18 @@ export function isStoreSnapshot(decoded: unknown): decoded is [StoreFormat, ...u
 }
 
 /**
- * The header of a decoded store file about to be rewritten: `undefined` for a legacy envelope array (an array whose first
- * slot is not a string, empty included), the header otherwise. A value that is not an array, or whose first slot is a string
- * but not a header this runtime reads, is refused, so a writer never overwrites a file it cannot see. Bytes after the array
- * are refused by `decode` itself.
+ * 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, flush and every rewrite refuse a file with this
+ * error, and a refused file is left as it was. `cause` is what the reader choked on.
  */
-export function storeHeader(decoded: unknown): StoreFormat | undefined {
-  if (!Array.isArray(decoded)) {
-    throw new Error("CultCache store is not a MessagePack array; it is not rewritten.");
-  }
+export class StoreUnreadableError extends Error {
+  readonly filePath: string;
 
-  if (typeof decoded[0] !== "string") {
-    return undefined;
+  constructor(filePath: string, cause: unknown) {
+    super(`CultCache store ${filePath} is not readable: ${cause instanceof Error ? cause.message : String(cause)}`, { cause });
+    this.name = "StoreUnreadableError";
+    this.filePath = filePath;
   }
-
-  if (!isStoreSnapshot(decoded)) {
-    throw new Error(`CultCache store's first slot is the string "${decoded[0]}", not a store header; it is not rewritten.`);
-  }
-
-  return decoded[0];
 }
 
 /** Refuses a persisted record with more slots than v1 defines, naming its key and schema id. */
@@ -62,3 +56,10 @@ export function requireV1RecordSlots(record: unknown[]): void {
     );
   }
 }
+
+/** Refuses a store array with more or fewer slots than the header, the schema catalog and the records; a header alone is not a store. */
+export function requireStoreSlots(decoded: unknown[]): void {
+  if (decoded.length !== STORE_SLOTS) {
+    throw new Error(`CultCache store has ${decoded.length} top-level slots; this runtime reads ${STORE_SLOTS} (header, schema catalog, records).`);
+  }
+}

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, rm, writeFile } from "node:fs/promises";
+import { access, mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises";
 import { homedir, tmpdir } from "node:os";
 import { dirname, join, resolve, delimiter } from "node:path";
 import { test } from "node:test";
@@ -15,6 +15,7 @@ import { CultCache } from "../src/cult-cache";
 import { inspectCultCacheBytes } from "../src/cult-cache-inspector";
 import { defineDocumentRegistry, defineDocumentType } from "../src/document";
 import { SingleFileMessagePackBackingStore } from "../src/single-file-messagepack-backing-store";
+import { StoreUnreadableError } from "../src/store-format";
 import type { CacheBackingStore, CultCacheEnvelope, CultCacheSchema } from "../src/types";
 
 const execFileAsync = promisify(execFile);
@@ -1577,7 +1578,11 @@ test("SingleFileMessagePackBackingStore pushAll refuses a store whose header it
   ]) {
     const file = await copyVector(dir, vector, "store.msgpack");
     const before = await readFile(file);
-    await assert.rejects(() => new SingleFileMessagePackBackingStore(file).pushAll(envelopes), /is not readable/u, vector);
+    await assert.rejects(
+      () => new SingleFileMessagePackBackingStore(file).pushAll(envelopes),
+      (error) => error instanceof StoreUnreadableError && /is not readable/u.test(error.message),
+      vector,
+    );
     assert.deepEqual(await readFile(file), before, `${vector} was rewritten`);
   }
 });
@@ -1590,7 +1595,9 @@ test("SingleFileMessagePackBackingStore replaces a file exactly when it reads",
     .split(/\r?\n/u)
     .filter((line) => line.length > 0 && !line.startsWith("#"))
     .map((line) => line.split(/\s+/u));
-  assert.equal(rows.length, 15);
+  // Every vector in the folder has a manifest row.
+  const listed = rows.map(([vector]) => vector!).filter((vector) => !vector.startsWith("..")).sort();
+  assert.deepEqual((await readdir(root)).filter((name) => name.endsWith(".bin")).sort(), listed);
   const dir = await mkdtemp(join(tmpdir(), "cultcache-readability-"));
   const seed = await copyVector(dir, join(c2aVectors, "v3-base.msgpack"), "seed.msgpack");
   const envelopes = await new SingleFileMessagePackBackingStore(seed).pullAll();
@@ -1611,8 +1618,27 @@ test("SingleFileMessagePackBackingStore replaces a file exactly when it reads",
               : store.delete(envelopes[0]!);
       if (reads) {
         await run();
+        if (operation === "pullAll") {
+          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.
+        assert.equal((decode(await readFile(file)) as unknown[])[0], vector.endsWith("v3-base.msgpack") ? "cultcache.store.v3" : "cultcache.store.v1", `${vector} ${operation}`);
+        const keys = (await new SingleFileMessagePackBackingStore(file).pullAll()).map((entry) => entry.key);
+        if (operation === "pushAll") {
+          assert.deepEqual(keys, envelopes.map((entry) => entry.key).sort(), `${vector} ${operation}`);
+        } else if (operation === "push") {
+          assert.ok(keys.includes(envelopes[0]!.key), `${vector} ${operation}`);
+        } else {
+          assert.ok(!keys.includes(envelopes[0]!.key), `${vector} ${operation}`);
+        }
       } else {
-        await assert.rejects(run, Error, `${vector} ${operation}`);
+        await assert.rejects(
+          run,
+          (error) => error instanceof StoreUnreadableError && error.cause !== undefined,
+          `${vector} ${operation}`,
+        );
         assert.ok(bytes.equals(await readFile(file)), `${vector} ${operation} rewrote a file it cannot read`);
       }
     }