packages/cultcache-ts/CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to `@gamecult/cultcache-ts` are recorded here, following
 
 ### Breaking
 
+- `SingleFileMessagePackBackingStore.pushAll` writes the given entries onto the store as it is under the lock, as `push` does: a record under the same key is replaced and no other is removed. It used to replace the whole file with the entries, so `CultMesh.flush()` from a cache that held less than the store (a node created with `pullOnStart: false`, or one whose image was stale) wiped the records it never read. A caller that relied on `pushAll` to reset a store removes the file first. A soft flush and the dangling-link case are unchanged. `CultCache.storedSnapshot()` is the image as the stores hold it (a global loaded from a legacy key under that key) and is what `CultMesh.flush` persists, so a flush no longer leaves such a global under two keys; `snapshot()` is unchanged.
 - A single-file store marked `cultcache.store.v3` (it holds an element id, written by the C# runtime) is read, and a rewrite keeps the mark and the ids; this runtime does not decide when a document carries ids. A runtime from before this change refuses a v3 store by name, so rolling back to an older release after a store has been written as v3 is one-way: there is no carry-back, so restore from a copy taken before upgrading.
 - `indexes` are unique, as C# `[CultIndex]`: a write or a load that leaves two records of a type holding one non-blank
   value under one alias is refused naming both keys and the alias, and writes nothing. Index text is the C# canonical

packages/cultcache-ts/README.md

@@ -174,7 +174,7 @@ const settings = cache.getRequiredGlobal(settingsDocument);
 - `updateGlobal(...)`
 - `delete(...)`
 - `deleteGlobal(...)`
-- `snapshot()`
+- `snapshot()` (the logical image) and `storedSnapshot()` (the image as the stores hold it, for a whole flush)
 - `getEnvelope(...)`
 - `getRequiredEnvelope(...)`
 - `getGlobalEnvelope(...)`

packages/cultcache-ts/src/cult-cache.ts

@@ -636,6 +636,13 @@ export class CultCache {
     return [...this.#hydrated.entries.values()].map((entry) => this.#toEnvelope(entry));
   }
 
+  // The image as the stores hold it, for persisting it whole (a flush): a global loaded from a legacy key is handed out under
+  // that key until its first write retires it, so a flush cannot leave the store holding it under two keys. `snapshot` is the
+  // logical image and always names a global by CultCache.GLOBAL_KEY.
+  storedSnapshot(): CultCacheEnvelope[] {
+    return [...this.#hydrated.entries.values()].map((entry) => this.#hydrated.legacyGlobals.get(entry.type) ?? this.#toEnvelope(entry));
+  }
+
   #registerNow(definition: AnyCultCacheDocumentDefinition): void {
     // Every registration refusal is a SchemaConflictError naming the new definition's schema id and both schema names.
     const refuse = (holder: RegisteredDefinition, claimed: CultCacheSchemaCatalogEntry, what: string) =>

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

@@ -82,10 +82,12 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore {
 
   async pushAll(entries: CultCacheEnvelope[], options: PushAllOptions = {}): Promise<void> {
     await this.#enqueue(async () => {
-      // 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 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. A soft flush asks that reader too: nothing at the path is written, a store is left alone.
+      // A flush writes `entries` onto the store as it is under the lock, the way a push does: a record under the same key
+      // is replaced and no other is removed, so a cache whose image is stale (or empty, never pulled) cannot drop what
+      // another writer added. The store 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. A soft flush asks that reader too: nothing
+      // at the path is written, a store is left alone.
+      let held: CultCacheEnvelope[] = [];
       try {
         const disk = this.#readDisk();
         if (options.soft && !disk.absent) {
@@ -93,10 +95,11 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore {
         }
 
         this.#format = disk.format;
+        held = disk.envelopes;
       } catch (error) {
-        // Only a dangling link reaches here with ENOENT: it holds no store, so a whole-store flush has no header to keep and
-        // replaces the link (R3 pins this). A soft flush does not replace what it cannot read, and every merging writer
-        // has already refused the link when it read the current entries.
+        // Only a dangling link reaches here with ENOENT: it holds no store, so a flush has nothing to merge with and no
+        // header to keep, and replaces the link (R3 pins this). A soft flush does not replace what it cannot read, and every
+        // merging writer has already refused the link when it read the current entries.
         if (options.soft || (error as NodeJS.ErrnoException).code !== "ENOENT") {
           throw error;
         }
@@ -104,9 +107,10 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore {
         this.#format = STORE_FORMAT_VERSION;
       }
 
-      requireOneTypePerKey(this.filePath, entries, []);
       // A key is held once: of same-type entries sharing a key in one batch, the last is kept.
-      await this.#writeAll([...new Map(entries.map((entry) => [entry.key, entry])).values()], []);
+      const supplied = [...new Map(entries.map((entry) => [entry.key, entry])).values()];
+      requireOneTypePerKey(this.filePath, entries, held);
+      await this.#writeAll(supplied, held);
     });
   }
 

packages/cultcache-ts/test/cult-cache.test.ts

@@ -1019,6 +1019,7 @@ test("legacyKeyGlobalLoadsWithoutWritingAndFirstWriteLeavesOneGlobal", async ()
   assert.deepEqual(reloaded.getRequiredGlobal(settingsDocument), { t: "new" });
 
   // Deleting an adopted global removes the legacy record.
+  await rm(storePath);
   await new SingleFileMessagePackBackingStore(storePath).pushAll([legacy]);
   const deleting = build();
   await deleting.pullAllBackingStores();
@@ -1026,10 +1027,30 @@ test("legacyKeyGlobalLoadsWithoutWritingAndFirstWriteLeavesOneGlobal", async ()
   assert.deepEqual(await new SingleFileMessagePackBackingStore(storePath).pullAll(), []);
 
   // Two globals of one type on disk, under any keys, are refused.
+  await rm(storePath);
   await new SingleFileMessagePackBackingStore(storePath).pushAll([legacy, { ...legacy, key: CultCache.GLOBAL_KEY }]);
   await assert.rejects(build().pullAllBackingStores(), /has multiple persisted entries/u);
 });
 
+test("a flush of a cache that loaded a legacy-key global leaves the store reopenable", async () => {
+  const settingsDocument = defineDocumentType({ type: "settings", schema: z.object({ t: z.string() }), global: true });
+  const storePath = join(await mkdtemp(join(tmpdir(), "cultcache-legacy-flush-")), "generic.cc");
+  const store = new SingleFileMessagePackBackingStore(storePath);
+  const build = () => CultCache.builder().withRegistry(defineDocumentRegistry(settingsDocument)).withGenericStore(store).build();
+  const seed = CultCache.builder().withRegistry(defineDocumentRegistry(settingsDocument)).build();
+  await seed.putGlobal(settingsDocument, { t: "old" });
+  await store.push({ ...seed.getRequiredEnvelope(settingsDocument, CultCache.GLOBAL_KEY), key: "legacy" });
+
+  const cache = build();
+  await cache.pullAllBackingStores();
+  assert.equal(cache.snapshot()[0]!.key, CultCache.GLOBAL_KEY, "the logical image names the global by the global key");
+  await store.pushAll(cache.storedSnapshot());
+  assert.deepEqual((await store.pullAll()).map((entry) => entry.key), ["legacy"], "the flush wrote the record under the key the store holds it");
+  const reopened = build();
+  await reopened.pullAllBackingStores();
+  assert.deepEqual(reopened.getRequiredGlobal(settingsDocument), { t: "old" });
+});
+
 test("concurrentPutAndAttachCannotLandRecordInNonHomeStore", async () => {
   const settingsDocument = defineDocumentType({ type: "settings", schema: z.object({ t: z.string() }) });
   const tempDir = await mkdtemp(join(tmpdir(), "cultcache-put-attach-race-"));
@@ -2261,7 +2282,7 @@ test("a stamp this runtime reads, followed by bytes that are not UTF-8, gets the
   }
 });
 
-test("pushAll leaves a store it reads alone when soft, and replaces one it reads otherwise", async () => {
+test("pushAll leaves a store it reads alone when soft, and writes onto one it reads otherwise", 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 existing = join(await mkdtemp(join(tmpdir(), "cultcache-pushall-soft-")), "store.msgpack");
@@ -2273,7 +2294,7 @@ test("pushAll leaves a store it reads alone when soft, and replaces one it reads
   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"]);
+  assert.deepEqual((await store.pullAll()).map((entry) => entry.key), ["probe", "second"]);
 });
 
 test("SingleFileMessagePackBackingStore reads only a missing store as empty", async () => {
@@ -2575,7 +2596,7 @@ test("SingleFileMessagePackBackingStore pushAll refuses a store whose header it
 
 // A file is replaced by a rewrite exactly when this runtime's own reader opens it: one verdict per file, asked by pullAll,
 // pushAll, soft pushAll, push and delete alike; a soft flush leaves the files that read as they are. The bytes and every runtime's verdict: tests/vectors/document-variants-c2a/readability.
-test("SingleFileMessagePackBackingStore replaces a file exactly when it reads", async () => {
+test("SingleFileMessagePackBackingStore rewrites a file exactly when it reads", async () => {
   const root = join(c2aVectors, "readability");
   const rows = (await readFile(join(root, "manifest.txt"), "utf8"))
     .split(/\r?\n/u)
@@ -2612,6 +2633,7 @@ test("SingleFileMessagePackBackingStore replaces a file exactly when it reads",
                 ? store.push(envelopes[0]!)
                 : store.delete(envelopes[0]!);
       if (reads) {
+        const heldBefore = operation === "pushAll" ? (await new SingleFileMessagePackBackingStore(file).pullAll()).map((entry) => entry.key) : [];
         await run();
         if (operation === "pullAll") {
           continue;
@@ -2628,7 +2650,8 @@ test("SingleFileMessagePackBackingStore replaces a file exactly when it reads",
         assert.equal((decode(await readFile(file)) as unknown[])[0], vector.includes("v3") ? "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}`);
+          assert.deepEqual(envelopes.map((entry) => entry.key).filter((key) => !keys.includes(key)), [], `${vector} ${operation}: every written key is held`);
+          assert.deepEqual(heldBefore.filter((key) => !keys.includes(key)), [], `${vector} ${operation}: nothing else is removed`);
         } else if (operation === "push") {
           assert.ok(keys.includes(envelopes[0]!.key), `${vector} ${operation}`);
         } else {

packages/cultcache-ts/test/store-file.test.ts

@@ -670,7 +670,7 @@ test("delete removes only the record at that type and key", async () => {
   }
 });
 
-test("pushAll replaces the whole store, and a soft pushAll leaves an existing store alone", async () => {
+test("pushAll writes onto the store as it is: it replaces by key and removes nothing, and a soft pushAll leaves an existing store alone", async () => {
   const directory = await scratch();
   try {
     const path = join(directory, "store.cc");
@@ -681,7 +681,14 @@ test("pushAll replaces the whole store, and a soft pushAll leaves an existing st
     await store.pushAll([entryOf("b", "t", 2)], { soft: true });
     assert.deepEqual([...(await readFile(path))], [...before], "soft leaves an existing store byte for byte");
     await store.pushAll([entryOf("b", "t", 2)]);
-    assert.deepEqual(await snapshotOf(store), ["t/b:2"]);
+    assert.deepEqual(await snapshotOf(store), ["t/a:1", "t/b:2"], "a record the batch does not name stays");
+    await store.pushAll([entryOf("a", "t", 7)]);
+    assert.deepEqual(await snapshotOf(store), ["t/a:7", "t/b:2"], "a record the batch names is replaced");
+    await store.pushAll([]);
+    assert.deepEqual(await snapshotOf(store), ["t/a:7", "t/b:2"], "an empty flush from a cache that holds nothing wipes nothing");
+    const before2 = [...(await readFile(path))];
+    await assert.rejects(() => store.pushAll([entryOf("a", "u", 1)]), SchemaConflictError);
+    assert.deepEqual([...(await readFile(path))], before2, "a batch that retypes a held key is refused and the file is unchanged");
   } finally {
     await rm(directory, { recursive: true, force: true });
   }

packages/cultmesh-ts/src/index.ts

@@ -3933,7 +3933,7 @@ export class CultMeshNode {
   }
 
   public async flush(soft = false): Promise<void> {
-    await this.store.pushAll?.(this.cache.snapshot(), { soft });
+    await this.store.pushAll?.(this.cache.storedSnapshot(), { soft });
   }
 }
 

packages/cultmesh-ts/test/cultmesh.test.ts

@@ -6,7 +6,7 @@ import test from "node:test";
 import assert from "node:assert/strict";
 import { decodeWireValue, encodeWireValue } from "@gamecult/cultnet-ts/wire";
 import { z } from "zod";
-import { SchemaConflictError, defineDocumentType, documentPayloadBytes, documentPayloadFromBytes } from "@gamecult/cultcache-ts";
+import { SchemaConflictError, SingleFileMessagePackBackingStore, defineDocumentType, documentPayloadBytes, documentPayloadFromBytes } from "@gamecult/cultcache-ts";
 import {
   CultNetDocumentRegistry,
   CultNetPeer,
@@ -109,6 +109,34 @@ test("CultMesh TS opens a durable local node and registers document bindings", a
   assert.ok(reopened.documents.getBySchemaId("cultmesh.note.v0"));
 });
 
+test("CultMesh TS flush from a node that never pulled leaves the records the store holds", async () => {
+  const filePath = join(await mkdtemp(join(tmpdir(), "cultmesh-ts-flush-")), "node.ccmp");
+  const first = await CultMesh.startNode(filePath, { documents: [noteDocument] });
+  await first.put(noteDocument, "note:held", { noteId: "note:held", body: "already in the store" });
+  await first.flush();
+
+  const unpulled = await CultMesh.startNode(filePath, { documents: [noteDocument], pullOnStart: false });
+  await unpulled.flush();
+
+  const reopened = await CultMesh.startNode(filePath, { documents: [noteDocument] });
+  assert.equal(reopened.getRequired(noteDocument, "note:held").body, "already in the store");
+});
+
+test("CultMesh TS flush of a node that loaded a legacy-key global leaves the store reopenable", async () => {
+  const settings = defineDocumentType({ type: "cultmesh.settings", schema: z.object({ theme: z.string() }), global: true });
+  const filePath = join(await mkdtemp(join(tmpdir(), "cultmesh-ts-legacy-flush-")), "node.ccmp");
+  const seed = await CultMesh.startNode(join(await mkdtemp(join(tmpdir(), "cultmesh-ts-seed-")), "seed.ccmp"), { documents: [settings] });
+  await seed.cache.putGlobal(settings, { theme: "iron" });
+  const legacy = { ...seed.cache.getRequiredEnvelope(settings, "__global__"), key: "legacy" };
+  await new SingleFileMessagePackBackingStore(filePath).push(legacy);
+
+  const node = await CultMesh.startNode(filePath, { documents: [settings] });
+  await node.flush();
+
+  const reopened = await CultMesh.startNode(filePath, { documents: [settings] });
+  assert.deepEqual(reopened.cache.getRequiredGlobal(settings), { theme: "iron" });
+});
+
 test("CultMesh TS document handles hide local cache plumbing behind typed reactive reads", async () => {
   const filePath = join(await mkdtemp(join(tmpdir(), "cultmesh-ts-doc-")), "node.ccmp");
   const node = await CultMesh.startNode(filePath, {