diff --git a/docs/document-variants-cut.md b/docs/document-variants-cut.md index bd340c87..044646dc 100644 --- a/docs/document-variants-cut.md +++ b/docs/document-variants-cut.md @@ -546,3 +546,791 @@ dependency-level for most, read in detail for Aetheria and AetheriaEve). `Aetheria.cc`. Raw readers refuse through the empty payload plus C0/C1: `CultMesh` single-file reads, AetheriaEve's directory-page reader (the directory store refuses variants, Q6), and Mimir (writer only). + +## Schema identity across runtimes (Imagination, 2026-09-30) + +**RULINGS (operator, 2026-09-30).** +- **C1: agreed** ("Agreed on schema identity"). C# puts the declared version string on the wire as the schema id. + F-ID.1 follows Self's recommendation: the id is `.v`, not the bare schemaName. +- **F-ID.2: AetheriaEve is not a consumer.** Operator: "AetheriaEve is purely archaeological, I would recommend + archiving it purely as evidence of what happens when you try to build a game with hyperfocused agents and with + CultLib under construction. We might try again at some point, but it won't look anything like AetheriaEve." C1 + deletes the fallbacks that exist only for AetheriaEve's readers. It does not migrate those readers. +- **F-ID.3: stored C# hash ids.** Operator: "I have no idea". Self's default is to defer C2. C1 changes nothing on + disk. C# keeps reading its own `sha256` ids. F2 already makes other runtimes keep C# records byte-for-byte. C2 + gets mapped only when a cross-runtime store read of a C# store is actually needed. +- C1 is mapped (section "C1 cut: one schema id on the wire", below). **Operator, 2026-09-30: "Yes to both, defaults + are fine".** The per-binding wire-id override is deleted. A declared compatible id resolves on the wire, and the + owner beats a lister. All six defaults stand. Stonks is fixed in Stonks before C1 merges. Order: F2 Soul, then the + put-serve merge, then c2a merges main, then C1.0-C1.5. + +Probed at CultLib `hands/variants-c2a` 92d9e138. The probe ran on Yggdrasil in `ack1d-interop:2` from a scratch branch, which has since been deleted. Sources: `scratchpad/schema-identity-probe/`. Full output: `scratchpad/probe2.log` (store round trip, section 1), plus a wire rerun shown inline below. Every probe used one declared schema: the interop note, name `cultcache.interop-note`, version `cultcache.interop_note.v1`, with six members. + +### 1. How each runtime derives a schema id (probed) + +| Runtime | Id it stamps | Derived from | Carries name / version / content hash? | +|---|---|---|---| +| C# | `sha256:eafe3e…b249` | `sha256("name\|version\|slot:Member:ClrType:value/ref:target:one/many\|…")`. Recomputing it by hand gives the same id. | Store catalog: name, version, `contentHash = sha256(canonical JSON)`, canonical JSON, members. Wire: `schemaName`, `schemaVersion`, `schemaContentHash` are all filled. | +| TS | `cultcache.interop-note` | `definition.schemaId ?? schemaName ?? type`. There is no hash. With no declared version, the version defaults to `.v1`. | contentHash is whatever the author declares. The interop test declares the id string itself. | +| Python | `cultcache.interop-note` | `schema_id or type`. There is no hash. contentHash defaults to the id. | Same as TS. | +| Rust | `cultcache.interop-note` | `#[cultcache(type=…)]`. There is no hash. `schema = "CultCacheInteropNote"` never reaches the catalog. | The catalog entry it writes is fabricated: version `".v1"` (`cultcache.interop-note.v1`, not the declared `…_note.v1`), `members: []`, `contentHash = type`. | +| Kotlin (CultMesh only; it has no store) | `cultcache.interop_note.v1` | `codec.schemaVersion` | Nothing else. | + +On the C# fingerprint: +- **The id stays the same** when only the CLR class or namespace changes (V1). +- **The id changes** when a member is renamed (V2), a slot is appended (V3), `string[]` becomes `List` (V4), the version string changes (V5), or a slot is removed (V6). +- **V2 and V4 produce identical wire bytes.** MessagePack slot arrays carry neither member names nor CLR collection types. So the fingerprint witnesses the C# declaration, not the wire layout. + +The C# registry already refuses two CLR types for one `(name, version)` in a process. The probe crashed on exactly that when V6 and the base type shared an assembly. Inside C#, `(name, version)` is already the ownership key, and the hash adds nothing to ownership. + +### 2. Wire parity for identity does not hold (probed) + +No two runtimes agree, and C# is the only one that hashes. The CultNet interop harness is the parity witness in CI. It never exercises the C# default: `tests/GameCult.Networking.InteropPeer/Program.cs:77,201,1183-1196` binds every type with `ForDocument(schemaId: )`. + +What reads what today: + +- **Stores (`.cc`). C#, Rust, TS and Python all read each other's notes.** It works only because each reader, when the id is unknown, falls back to the embedded catalog's **schemaName**: C# `CultCache.cs:653` `BySchemaName`, TS `cult-cache.ts:827`, Rust `resolve_registered_type` `lib.rs:2754`. Round trip, C# → X → C#: + - Each writer stamps its own record with its own id. + - **A C# rewrite restamps every other runtime's record to `sha256:…`** and drops their catalog entries. + - **A Rust write destroys the C# catalog entry for the id it read.** The entry becomes name=`cultcache.interop-note`, version=`sha256:…eafe….v1`, contentHash=`sha256:…eafe…`, canonical JSON with `schemaName: "sha256:…"`, `members: []`. The C# record bytes survive, but their description is lost. That breaks F2 ("never relabels…") at the catalog level, and it is a defect whatever this fork decides. + - TS and Python kept the C# catalog entry intact. +- **Wire (CultNet raw record carrying a C# `sha256` id):** + +| Reader | `sha256:…` | `cultcache.interop-note` (name) | `cultcache.interop_note.v1` (version) | +|---|---|---|---| +| TS `CultNetDocumentRegistry.applyRawDocumentPut` (probed) | refused | ok | refused | +| Kotlin `CultCache.putRaw` (probed) | refused | ok (documentType) | ok | +| C# `CultNetDocumentRegistry.DeserializeRawDocument` (probed) | ok | ok | ok, **and `gamecult.unknown.v1` → `Note` too** | +| Python `cultnet_py` `resolve_document_for_raw_record` (read only) | only if declared | ok | ok, plus payload sniffing | +| Rust `pull_rudp_catalog_snapshot` (read only, per the Eyes sweep) | dropped silently | — | strips `.vN` and matches the type | + +C# accepts any id because `TryResolveDescriptorBySchemaAlias` and `TryResolveDescriptorByPayloadSchema` (`CultNetDocumentRegistry.cs:755-830`) read slot 0 of the payload as a schema version. Python does the same (`replication.py:160-196`). **Shape routing and aliasing are still live in C# and Python CultNet on c2a**, and so is C# request matching `CultNetSchemaAliasMatching` (`CultNetDatabase.cs:52-89`). F7 has not reached them. + +- **The C# wire id is readable only when the binding is overridden.** The probe showed `ForDocument(schemaId: "cultcache.interop-note")` emits that id with the same payload, name, version and contentHash. `CultMesh.CreateCultNetDocumentRegistry` (`src/GameCult.Mesh/CultMeshDocumentRegistries.cs:69`) never overrides it, so every C# CultMesh host publishes `sha256` ids. + +### 3. Live cross-runtime paths that depend on the fallback + +This section comes from the Eyes sweep. I spot-checked the Hermodr lines, the Odin rule and `CultMeshDocumentRegistries.cs`; the rest I only read. Every one of these paths is **TS reader → C# host (the Aetheria daemon in `F:\Projects\AetheriaEve`)**. + +| # | Consumer | Asks for | Works via | Under C1 | +|---|---|---|---|---| +| 1 | `EveElectron/src/cultmesh-provider-client.mjs:129-143`, started by `AetheriaEve/Aetheria.Rts.Web/Electron/main.ts:65` | `gamecult.eve.provider_advertisement.v1`, `…surface.v1`, `gamecult.fields.*.v1`, `…command_receipt.v1`, `gamecult.cultmesh.cdn.asset_blob.v1` | cultmesh-ts HELD `firstRecordAtKey` | Exact match. The exception is `cdn.asset_blob.v1`: **no writer anywhere declares it**, so that read needs a declaration. | +| 1b | same file, lines 94-108 (command put) | `gamecult.eve.command_invocation.v1` | C# inbound alias (g), probably | Exact match | +| 2 | `AetheriaEve/Aetheria.Rts.Web/Electron/aetheria-cultmesh.ts:138-147, 345-406` (only the verify-stage7* scripts use it) | `gamecult.aetheria.*.v1`, `gamecult.fields.*.v1` | first-record fallback | Exact match | +| 3 | `Hermodr/src/hermodr-daemon.cjs:1335-1363` | provider state, surfaces, CDN | Its own `|| candidates[0]` / `|| documents[0]`, plus C# request alias (f) | Exact match. Delete both fallbacks. | +| 4 | `AetheriaEve/scripts/aetheria-browser-provider-witness.ts:49-60` (a witness, not product code) | eve schemas | cultmesh-browser `recordMatchesSchema` name/version arms | Exact match | +| 5 | `aetheria-cultmesh.ts:408-411`, local `.cc` read | same ids | none: probably already broken (unverified) | still a store-path question | + +**Paths that already work without a fallback, because everyone hand-stamps `.vN`:** +- The Aetheria daemon and Gjallar hand-stamp `.v1` puts to Odin: `Program.cs:4574-4830` and `Gjallar/Program.cs:3040-3080`. +- Odin **refuses any id without `.vN`**: `Odin/crates/odin-daemon/src/main.rs:1310-1326`. +- Rust hosts bind `.vN` through `cultmesh_documents!`: Ghostlight, Epiphany, Muninn. +- Kotlin uses its schemaVersion. +- The Eve C# declaration itself reads `[CultDocument("gamecult.eve.provider_advertisement", SchemaId)]` with `const SchemaId = "gamecult.eve.provider_advertisement.v1"`. Authors already treat the version string as the id. + +Scope flag: every broken consumer lives in `AetheriaEve`, which the CultCache campaign ruled "taxidermy". The operator should confirm those paths are live consumers of this campaign before any cut is shaped for them. + +Stored data: +- `Aetheria/GameData/Aetheria.cc`, `run.cc` and `player.cc` hold 21 C# schemas under `sha256` ids. The Eyes sweep found only C# readers. +- The versions there are the bare string `"1"`: `[CultDocument("aetheria.faction", "1")]`. So a version string is not globally unique across the estate. +- TS stores (Bifrost, VoidBot, weksa, Stonks) are TS-only and hold readable ids. +- Odin's Rust store holds `.vN` ids. + +### 4. What the C# fingerprint protects + +A C#-to-C# store read. A record whose declaration drifted misses the exact id and falls to the persisted-catalog comparison: `CompatibleDrift`, or refused as incompatible (`CultCache.cs:900-936`). + +Across runtimes it protects nothing today: +- The other runtimes do not compute it. +- Their content hashes are placeholder strings. +- The C# wire receiver accepts any id through payload sniffing (probed). +- The hash reacts to renames and CLR collection spellings, which do not change wire bytes (probed V2, V4). + +### Fork F-ID: what names a schema across runtimes + +**(A) Every runtime computes the C# semantic fingerprint; readers name schemas by definition.** +- Every non-C# runtime has to emit C# type spellings: `System.String[]` versus `System.Collections.Generic.List`. +- Rust has to start declaring members and versions at all. +- String-named readers can no longer name a schema. Eve Electron and Hermodr read ids from provider advertisements. +- Odin refuses the ids outright (`.vN` rule). +- Every TS, Python and Rust store holds readable ids. Under no-relabel those records become permanently held unless each runtime lists its old ids. +- Gain: a layout witness that is still over-sensitive (V2 and V4 change the id with no wire change). +- Stored data: Aetheria is untouched; every non-C# store is orphaned. +- Not recommended. + +**(B) Readers match the declared `(schemaName, schemaVersion)` pair the envelope carries.** +- The C# wire already carries both. Store catalogs carry both. +- Two identity strings stay in play: the id for storage, the pair for lookup. That is a split owner. +- `sha256` ids still reach Odin, which refuses them. +- The version alone is ambiguous (Aetheria's `"1"`), so every reader API has to take a pair instead of the one string readers pass today. +- Rust writes a fabricated version, so pair matching against a Rust host fails until Rust declares versions. +- Layout protection is only as good as a contentHash comparison, and no runtime but C# computes a real one. +- Stored data: unchanged, and nothing is relabelled. Resolution becomes pair matching against the catalog, which is a formalised form of today's schemaName fallback. + +**(C) C# puts the readable id on the wire, and the id is the declared version string `.v`.** + +This is what Odin enforces, Rust and Kotlin use, and Eve and Gjallar already hand-stamp. +- C1, wire only: + - `ForDocument` and `CultMesh.CreateCultNetDocumentRegistry` default the binding id to `descriptor.SchemaVersion`. + - Registering a wire binding whose version lacks `.vN` is refused. Aetheria's `"1"` types are not on the wire today. + - Then delete the HELD first-record fallback (cultmesh-ts), Hermodr's two fallbacks, the name and version arms of cultmesh-browser `recordMatchesSchema`, C# `TryResolveDescriptorBySchemaAlias`, payload sniffing and `CultNetSchemaAliasMatching`, and the Python CultNet alias map and sniffing. That finishes F7 on the wire. +- C2, the store as well, as a later cut: C# stamps new writes with the version id and resolves an old `sha256` record by the type's own **computed** former fingerprint. That is a derivation, not an alias. Where a version string changes (Aetheria's `"1"` → `aetheria.faction.v1`), the old id goes into the existing `CompatibleSchemaIds`. +- The fingerprint is demoted to `schemaContentHash`, which the wire already carries. It stays a C#-to-C# drift witness and does not decide identity. +- Costs: + - C#: the default in two files, plus deleting the alias and sniffing paths. + - TS, Hermodr and Python: deletions only. + - Rust and Kotlin: none. + - Mixed-version C# peers: an old host sends `sha256` and a new receiver without sniffing refuses it, so C# peers upgrade together. + - Eve Electron `cdn.asset_blob.v1`: needs a declaring writer. +- Stored data: C1 changes nothing on disk. Under C2, Aetheria's catalogs are read unchanged and records are restamped only when C# rewrites records it understands, which F2 allows. TS, Rust and Odin stores are untouched. + +**Recommendation: C1 now; decide C2 separately.** C1 is the only option where one string names a schema on every runtime and at Odin, where every reader already asks for that string, and where the work is almost entirely deletion. It makes the F7 retirement possible on the wire, because readers no longer need a fallback to find their records. Aetheria's `.cc` files are untouched. B keeps the split and A breaks every readable-id store. Separately from F-ID, the Rust catalog-destruction defect (§2) and the live C# and Python CultNet aliasing should be admitted as findings or follow-ups whatever is ruled. + +**Sub-questions for the operator:** +- F-ID.1: Is the id the version string (`.vN`, as Odin, Rust, Kotlin and Eve do), or `schemaName`, which CultLib's own TS and Python tests use and which is also TS's default? +- F-ID.2: Are the AetheriaEve consumers live for this campaign? +- F-ID.3: Should C2 (store identity) be mapped now, or deferred until a non-C# reader of a C# store exists? + +### Probed versus read + +- **Probed:** + - All C# ids and variants, and the hand-recomputed fingerprint. + - The TS and Python default ids. + - Rust's id and catalog entry, from store bytes. + - Kotlin's id and its accept/refuse behaviour. + - Store round trips C#→Rust/Python/TS→C#, with bytes, including the C# restamp and the Rust catalog destruction. + - C# wire record bytes, default and overridden. + - C# CultNet accepting any id. + - TS CultNet exact-id behaviour. +- **Read only:** + - Python CultNet aliasing and sniffing, and Rust `pull_rudp_catalog_snapshot` dropping unknown records. + - C# request alias matching (f). + - Every consumer path in §3, from the Eyes sweep. I spot-checked Hermodr 1335-1363, Odin 1310-1326, `CultMeshDocumentRegistries.cs:69` and the Eve declaration. + - Path 1b/2b's dependence on (g). + - Path 5 being broken. + - The Aetheria `.cc` catalogs were decoded from the files, with no git. + +## C1 cut: one schema id on the wire + +Imagination, 2026-09-30. This maps the operator's C1 ruling. On the wire, a document's schema id is its declared +version string, `.v`. Every runtime stamps it and every runtime resolves it exactly. A wire binding whose +version lacks `.v` is refused. The AetheriaEve readers are archaeology, and C2 (store identity) is deferred. + +**Anchors.** Line numbers are against CultLib `hands/variants-c2a` at `976b57f1`. That commit is F2 in progress, the +Rust catalog lay-back. I read it from a `git archive` snapshot and did not touch the worktree. Where main +(`4a7fa1d6`, code as of `daedfdd6`) or `hands/cultmesh-put-serve-bound` (`fa34b685`) differs in a file C1 edits, it is +said at the spot. Each anchor also names its symbol, so it can be found again after c2a merges main (see Sequencing). + +### 0. Probe facts + +These are new and were run, not inferred. The sources are in `scratchpad/c1-probe/`. They ran on Yggdrasil in +`ack1d-interop:2` from the scratch ref `refs/scratch/c1probe`, a detached commit on `976b57f1` that is deleted after +this pass. Output is in `scratchpad/c1-probe3.log`. The run was commit `43677b97ac`. + +- **P-VER: a C# version change does not orphan stored records.** The probe used two builds of one type, + `probe.faction`, declared first with version `"1"` and then with `"probe.faction.v1"`, plus a sibling + `probe.ship "1"`. The same members. + - The id changes: `sha256:64eb…` becomes `sha256:0603…`. + - The new build reads the old store through the catalog's schema name (`Claimant`, `CultCache.cs:697`) and reports + `CompatibleDrift` with warnings `compatible_schema_id_drift` and `content_hash_drift`. The record decodes. + - **Opening and flushing without edits leaves the file byte-identical** (1529 bytes, old id). + - **Upserting restamps** that record and its catalog entry to the new id. The untouched sibling keeps its own. + - The old build then reads the rewritten store the same way. + - So a version-string change is a restamp on the next write, not an orphan. This holds only while one local type owns + the schema name. Two local versions of one name make the fallback ambiguous and it throws (`CultCache.cs:697-705`). + - C1 does not need this. It is recorded for C2. +- **P-WIRE: what C1 changes on the C# wire.** + - Binding both types by their version string: with `"1"` twice, c2a's `Register` refuses the second binding + (`CultSchemaConflictException … schema id "1"`). With `probe.faction.v1` and `"1"`, it accepts both. + - The default wire id is still sha256 in both builds. + - With an unbound registry, `DeserializeRawDocument` returned `Faction` for `"1"`, `probe.faction` (the name) and + `probe.faction.v7` (a version nobody declared). Only `some.other.v1` was refused. + - So today a C# receiver resolves a version it has never seen by stripping `.vN` and matching the name. That is the + path C1.0 deletes, and the negative check in §8 pins its death. + +Earlier facts this map rests on, from the §1-§4 probes above: C# hashes the version into the id (V5). The C# wire +id is sha256 unless the binding overrides it. C# CultNet accepts any id through sniffing. TS refuses a version id. +Kotlin stamps its version. + +### 1. What the consumer sweep settles + +This comes from three Eyes-class sweeps over F:\Projects. I read the rows marked (R); the rest were confirmed by the +sweeping agent reading the files. + +- **Aetheria's `"1"` versions never reach the wire.** Aetheria has no CultNet or CultMesh code. `Aetheria.Shared.csproj` + references only the caching assemblies. So C1's refusal never fires for its 24 declarations, and **nothing in + Aetheria changes, on disk or in source.** +- The other C# `"1"` declarations are in StreamPixels `apps/overlay-unity` (24). They are client-only and never bound: + frame ids map to types by hand (`OverlayFrameDocuments.cs:98-116`). +- Every C# type that is bound in any repo already declares `.vN`: AetheriaEve 61, AquaSynth 28, Mimir 37, Brokkr, + Eve, Gjallar, Ymir, Delvehold. **No repo outside CultLib passes `ForDocument(schemaId:)`.** +- **Live C# hosts with default (sha256) bindings:** + - AquaSynth, `AquaSynthCultNetDaemon.cs:431-449`, 19 bindings. Its peers are C# (`AquaDaemonClient.cs`). + - Mimir: `EveDashboard/Program.cs:877-886`, read by Eve Android, which ignores schema ids + (`MainActivity.kt:618-625`); and `CultMeshMedia/Program.cs:277`, C#. + - Delvehold: no server. + - All of these are C#-to-C# or id-blind. They change id together when they rebuild on C1, so no shim is needed. +- **The C# publishers to Odin bypass bindings.** AetheriaEve `Program.cs:4574-4831` and Gjallar `Program.cs:3045,3069` + build their ids by hand and serialize payloads directly. C1 does not change them. Odin's rule is at + `Odin/crates/odin-daemon/src/main.rs:1311-1326` (R). +- **Every Rust wire binding in every repo already binds `.vN`** through `cultmesh_documents!` or + `for_entry_with_schema_id` with `.vN` constants: Odin, Muninn, Ghostlight, Epiphany, Ratatoskr, Huginn. No consumer + uses bare `for_entry::()`. +- **The AetheriaEve readers are archaeology:** + - EveElectron's only code consumer is AetheriaEve (`Aetheria.Rts.Web/package.json:30`). + - Hermodr was parked on 2026-09-30: `gamecult-ops/inventory.md:631-634`, `runbooks/odin-yggdrasil.md:35-48`, and + `scripts/idunn/idunn-deployment-targets.ps1:36-46` (status "archived"). + - cultmesh-browser has two consumers. One is the AetheriaEve witness. The other is CultLib's own sample + `samples/eve-browser-network/browser/main.ts:8`. + - **So C1 edits neither Hermodr nor EveElectron.** Both are archived, so C1 deletes the CultLib fallbacks that exist + only for them and does not migrate them. +- **Non-C# consumers that change id under C1:** + - **Stonks is the only one that breaks.** `F:\Projects\Stonks\src\stonks-daemon.cjs:152-195` declares + `schemaVersion: "v1"` on six definitions, one of them `provider_advertisement` published to Odin at `:303`. Under + C1 its TS registration is refused. Stonks loads `../CultLib` from disk, so it breaks on the first restart after C1 + reaches that checkout. The fix is in Stonks: set each `schemaVersion` to its current `schemaId`. That is six lines, + and they land before C1 is merged to main. + - Bifrost `tools/agent-transport.mjs:62-66`: the id is hyphenated (`…update-request.v0`) and the version has + underscores (`…update_request.v0`). The wire id changes to the version. The snapshot and apply sides are the same + tool on one pinned CultLib (`f67f5122`), so they change together. Its store keeps the hyphenated id. + - Brokkr Python `surfaces/blender/brokkr_bridge/blender_target.py:630-637` declares eight type-only definitions whose + type already ends in `.v0`. The Python default version would make the wire id `….v0.v1`. The fix is in Brokkr: + declare `schema_version=`. No live reader was found, so this is hygiene, not a break. + - repixelizer: `verse_state.py:139-230` declares version ≠ id. Its wire id moves to the `.v0` version, which Odin + accepts. + - Everything else already has id == version: VoidBot, Vili, weksa, Heimdall, StreamPixels, Bifrost's other tools, + Hermodr. + - VoidBot's persona stores use a vendored cultcache-ts and are untouched. +- **Raw health puts are out of C1's reach.** Vili, Stonks, weksa, Bifrost and repixelizer send hand-built + `idunn.daemon_health` with no version, and Bifrost sends `idunn.signed_daemon_health`. These never register a + binding. No receiving binding was found in Idunn or Odin main. That is Soul's consumer check, not a C1 edit. + +**Q3, settled:** no declaration in any repo has to change to satisfy the refusal on a binding that actually exists. +Aetheria and the StreamPixels overlay are the `"1"` repos, and neither binds. Stonks's `"v1"` is TS; it is fixed in +Stonks and touches no store. Stonks's store keeps the ids it holds today, because TS stores under `schemaId`. So +**no declaration change touches disk.** Whether a future change of Aetheria's version string would orphan its records +is the P-VER probe in §0. It is recorded for C2 and not needed by C1. + +**Q4, settled:** C1 changes no store record in any runtime. The store id and the wire id become two derivations: +- C#: sha256 on disk, version on the wire. +- TS: `schemaId ?? schemaName ?? type` on disk, version on the wire. +- Python: `schema_id or type` on disk, version on the wire. +- Rust: the store id and apply-path stamping are unchanged (see C1.4). +- Kotlin: version for both, as today. + +Four places where one variable serves both today must be split. Each has its cut below: +- TS `replication.ts:363-368`, `schemaIdForEnvelope`, which serves the stored id. +- Python `cultmesh_py/node.py:767` and `interop_peer.py:1085`. +- The C# durable shard log, `CultNetShardMutationLogStorage.cs:75-88`. It persists wire-form entries, and so it is the + one C# store that holds wire ids. +- Rust `pull_rudp_catalog_snapshot`, `cultmesh-rs/src/lib.rs:742`. + +### 2. Authority map + +- **Owner.** The declaration's version string is the only wire identity: + - C# `CultDocumentAttribute.SchemaVersion` → `CultDocumentDescriptor.SchemaVersion`; + - TS `definition.schemaVersion`, defaulting to `${schemaName}.v1`, through `schemaIdentityOf(...).schemaVersion`; + - Python `schema_version`, defaulting to `f"{schema_name}.v1"`; + - Rust: the version given to `cultmesh_documents!` or `for_entry_with_schema_id`; + - Kotlin `codec.schemaVersion`. +- **One derivation per runtime:** + - C#: `CultNetDocumentRegistry.WireSchemaId(descriptor)` returns `descriptor.SchemaVersion`. It is the single place + `.vN` is checked, and it is also called eagerly from `Register`. + - TS: `wireSchemaIdOf(definition)` in cultnet-ts. + - Python: `wire_schema_id(definition)` in cultnet_py. + - Rust: `CultNetDocumentBinding::schema_id`, which is already the version. + - Kotlin: `codec.schemaVersion`. +- **Inputs:** the declaration only. Never a payload, a schema name, an inferred name, a record's position in a + response, or a stored record's id. +- **Outputs:** every schema id that crosses a process boundary: + - raw records, puts and deletes; + - snapshot and selection responses, headers, edges and cursors; + - database change messages and shard-log entries; + - snapshot, subscribe and selection filters, and the shard catalog; + - the document-payload schema catalog; + - CultMesh projection-source and state-binding records. +- **Resolution.** A wire id resolves to the binding that owns it (exact). Otherwise it resolves to the one binding that + *declares* it compatible (`CompatibleSchemaIds`, owner beats lister), which is the rule stores already use. **OQ-C1.2 + decides whether declared compatibility applies on the wire.** Anything else is refused. +- **Derived state after C1:** + - C# `descriptor.SchemaId` (sha256) is **store-only**. It keys CultCache, `_predictedDocuments`, + `_lastWriteSequence`, and in-process dictionaries. + - TS and Python `schemaId` are **store-only**. + - `CultNetDocumentBinding.SchemaId` is **derived** (the version). It is no longer an owner. + - `schemaName` is **metadata only**, never matched. + - Payload slot 0 or `schemaVersion` is **data only**, never read for routing. + - `schemaContentHash` is a **C#-to-C# drift witness**, carried and never matched. +- **Forbidden writers, all deleted:** + - C#: `CultNetSchemaAliasMatching`, `TryResolveDescriptorBySchemaAlias`, `TryResolveDescriptorByPayloadSchema`, + `PayloadMatchesSchema`, `TryReadSchemaVersion`, both `InferSchemaName`, `ResolvePayloadSchemaCandidates`, + `ExpandSchemaBindingAliases`, the `ForDocument(schemaId:)` override (OQ-C1.1), the CultMesh `aliased` and + `byPayload` branches, and the descriptor-name and version keys in the C# and TS mesh catalogs; + - TS: `firstRecordAtKey`, cultmesh-browser's name and version arms, `shard-catalog.ts` `inferSchemaName`, and + `inferCultMeshSchemaName` / `bySchemaNameVersion`; + - Python: the name and version keys in `schema_document_map`, `_infer_schema_version_from_payload`, and every + `_infer_schema_name`; + - Rust: `strip_schema_version_suffix` and `selection::schema_alias`; + - Kotlin: `codecs[schemaId]`, the documentType fallback. +- **Shared paths.** Put, delete, snapshot (v0, v1, raw), selection, subscription, database change, shard log (live and + replayed from disk), shard catalog, schema catalog, and CultMesh peer reads all take the id from the one derivation. + A served stored record is stamped by the derivation, never by its stored id. + +### 3. Deletes first (C1.0, subtraction) + +C1.0 deletes every inference path and makes each comparison exact against the existing wire id: the binding id, or +sha256 for an unbound type. C#-to-C# peers keep working. Only readers that relied on a fallback stop resolving, and +those are the AetheriaEve readers, which are archaeology. C1.0 adds no behaviour; C1.1 flips the id. + +**C#, `src/GameCult.Networking`:** +- `CultNetDatabase.cs:47-104`: delete the whole `CultNetSchemaAliasMatching` class (58 lines, public). + - Replace every caller with an exact `string.Equals` against the registry's wire id for the descriptor. + - Callers: `:143, :150` (`ClientAuthorityScope.Matches`); `:252, :260` (`Shard.Matches`); `:1757` (shard-catalog + filter). + - `Shard.Matches(string)` and `ClientAuthorityScope.Matches(string)` compare a configured id with the wire id. + - `EnsureClientAuthority(string)` at `:1640-1652` has no callers; delete it. +- `CultNetDocumentRegistry.cs`: + - Delete `:779-799` `TryResolveDescriptorBySchemaAlias`, `:813-830` `TryResolveDescriptorByPayloadSchema` (public), + `:832-842` `ResolvePayloadSchemaCandidates`, `:852-854` `PayloadMatchesSchema` (public), `:865-898` + `TryReadSchemaVersion`, `:900-910` `InferSchemaName`, and `:538-578` `ExpandSchemaBindingAliases` with its comment. + About 175 lines in all, doc comments included. + - `ResolveDescriptorForRawDocument` (`:755-765`) and `ResolveDescriptorForSchemaId` (`:767-777`) collapse into one + exact lookup. It uses `_bindingsBySchemaId`, or, for a registry with no bindings, the one descriptor whose wire id + equals the id. A miss throws `InvalidOperationException` naming the id. + - Call sites stay on the one method: `CultNetDatabase.cs:717, 947, 998, 1234, 1455`, + `CultNetDatabaseSubscriptionClient.cs:254`, and `CultNetDocumentRegistry.cs:691, 713`. +- `CultNetDocumentRegistry.cs:407, 507`: drop the `ExpandSchemaBindingAliases` calls. `CultNetDatabaseServer.cs:566-567` + and `CultNetDatabaseSubscriptionServer.cs:559-560` pass the selection unchanged. +- `CultNetSelection.cs:393, 453` and `CultNetSelectionEvaluator.cs:88, 422`: exact comparison against the wire id. + - The evaluator reads only descriptors, so C1.0 gives it the registry's `WireSchemaId` as a `Func`. + - After C1.1 the id is `descriptor.SchemaVersion`, and the parameter collapses to that (see C1.1). + - `CultNetSelection.cs:178` doc: "schema ids", not "or aliases". + +**C#, `src/GameCult.Mesh`:** +- `CultMesh.cs:2724-2774` `ReadDocumentFromSnapshotResponse`: delete the `aliased` and `byPayload` branches + (`:2744-2745, 2754-2757`). Pick `exact` or throw. Rewrite the comment block at `:2724-2733`. +- `CultMesh.cs:3441-3464` `IsSameCultDocumentSchema`: the id is local (a `CultNetDatabaseChange` sha256). Replace with + `schemaId == descriptor.SchemaId`, or `storedDescriptor.SchemaId == descriptor.SchemaId`. Delete the comment + claiming "every … decision goes through" the matcher. +- `CultMeshSnapshots.cs:982-985` `DecodeSnapshotDocuments`: keep only `canDeserializeWithBinding`, or + `record.SchemaId == wire id of TDocument`. +- `CultMeshDocumentRegistries.cs:64-71`: the `GroupBy(SchemaId)…Reverse()` loop is left over from aliasing. With c2a's + `Register` refusal, collapse it to one `Register` per distinct type. +- `CultMeshPrimitives.cs` ~`1806-1860` and ~`1996-2050`, the C# `CultMeshDocumentCatalog` and + `CultMeshCollectionCatalog`: drop the schema-version and schema-name keys. A lookup by schema answers by the handle's + type only. **A lookup that more than one handle answers is refused, naming their document ids.** This is the "typed + catalog lookup returns the last handle added" defect, and TS has the same fix below. + +**TS:** +- `cultmesh-ts/src/index.ts:6297-6328` `requestCultNetRawSnapshotDocument`: delete the HELD comment, the + `firstRecordAtKey` option, and `?? atKey[0]`. Delete the caller's `firstRecordAtKey:` at `:2691`. +- `cultmesh-ts/src/index.ts:6175-6191` `normalizeCultMeshDocumentSchema`: drop the `inferCultMeshSchemaName` arm and the + version-from-id arm, and delete `inferCultMeshSchemaName` at `:6228-6238`. +- `cultmesh-ts` catalogs: `CultMeshDocumentCatalog` `:1462-1519` and `CultMeshCollectionCatalog` `:1688-1731`. + - Delete `#bySchemaNameVersion`, `cultMeshSchemaNameVersionKey` (`:6220-6226`), and its arm in + `cultMeshRefuseSecondCatalogType` (`:6197-6218`). + - `tryDocument` and `tryCollection` answer by type or by wire id. When `#byType` holds several handles for one type + (`add` at `:1482` overwrites, so today the last one wins), the lookup is refused with the handles' document ids. + That needs `#byType: Map`. +- `cultmesh-browser/src/index.ts:770-774` `recordMatchesSchema`: exact `record.schemaId === requestedSchema`. Callers + are `:704, :721`. After C1.2, CultLib's `samples/eve-browser-network` asks by version. +- `cultnet-ts/src/shard-catalog.ts:103-130`: `schemaIdsMatch` becomes `===`. Delete `inferSchemaName`. + +**Python:** +- `cultnet_py/replication.py:21-30` `schema_document_map`: key by wire id and compatible ids only; drop `schema_name` and + `schema_version` keys that differ from the wire id. +- `replication.py:158-194`: `resolve_document_for_raw_record` becomes `dict` lookup or `KeyError`. Delete + `_infer_schema_version_from_payload` and `_infer_schema_name`. +- `cultnet_py/shard_catalog.py:143-154`: exact. +- `cultmesh_py/node.py:1311-1343`: `_schema_matches_request` and `_document_matches_schema_id` become exact; delete + `_infer_schema_name`. +- `cultmesh_py/server.py:687-721`: the map copy and request filter become exact; delete `_infer_schema_name`. +- Keep `_documents_alias` / `_resolve_document_alias` (`node.py:1188-1243`). They compare local definitions, not wire + ids (follow-up F-C1-PYALIAS). + +**Rust:** +- `cultmesh-rs/src/lib.rs:750-773`: delete `strip_schema_version_suffix`. `registered_document_type_for_schema` becomes + `binding_by_schema_id(..).map(|b| b.document_type.clone())` and nothing else. Delete the `documents.binding(schema_id)` + type-name arm. +- `cultnet-rs/src/selection.rs:709-851`: delete `schema_alias`, tests included. Its uses at `:1432, 1487, 1651, 2112` + become exact comparisons against the row's binding id. +- `cultnet-rs/src/snapshot_query.rs:160-181` `RawSnapshotRow`: stop defaulting `schema_name` to the id; it reaches + `schema_alias` from `serve_read_only_raw_snapshot` (`:129`). + +**Kotlin:** `cultmesh-kotlin/.../CultMesh.kt:154-155` `codecForSchema`: drop `?: codecs[schemaId]`. That also fixes +`deleteBySchema` at `:170-171`. + +**Not edited, archaeology:** Hermodr `hermodr-daemon.cjs:1335-1363`, EveElectron, and AetheriaEve's own readers. + +**Tests deleted or inverted in C1.0.** Each rule that dies keeps a test that proves it is dead. +- C#: + - `NetworkingTests.cs:3177` (a selection by version matches through the alias): invert to not-found before C1.1. + - `NetworkingTests.cs:4022` (foreign id by sniffing): invert to refused. + - `NetworkingTests.cs:4109, 4140, 4276, 4317, 4366` (shard and scope configured by version): configure by the wire id. + - `CultNetSelectionFixBatchTests.cs:121, 158` (name alias): invert. + - `CultMeshStreamingTests.cs:2638-2708, 3062, 3105` (payload fallback): invert to refused or skipped. +- TS: + - `cultmesh.test.ts:3061-3074, 3244-3249` (HELD): invert to not-found. + - `cultmesh.test.ts:1170-1196, 1241-1286`: catalog by version against a sha256 handle. + - `cultnet.test.ts:1064-1068` (shard by name). + - `cultmesh-browser.test.ts:139, 452, 530, 536, 692-697`. +- Python: + - `test_cultnet.py:1685-1726, 2059-2063`. + - `test_cultmesh.py:291-339, 343-379` (drop the name case). + - `test_cultmesh.py:4150-4302` (sniffing and deletes by version against a sha256 id). +- Rust: `tests/selection.rs:2134, 2151, 2179, 2193, 2216-2219`. + +### 4. C1.1: the C# wire id is the version + +This is a behaviour cut on `src/GameCult.Networking` and `src/GameCult.Mesh`, plus the shared selection fixture with +Rust. + +**The store id and the wire id stay separate.** `descriptor.SchemaId` (sha256) keeps keying CultCache and every +in-process dictionary. The wire id is `descriptor.SchemaVersion`. Once the override is deleted (OQ-C1.1), the wire id +derives from the descriptor alone, with no registry. That is what lets the selection evaluator, the mesh and the +database compute it without a binding. + +**How a C# reader maps `.vN` to its descriptor:** +- `_bindingsBySchemaId` is keyed by `binding.SchemaId`, which now equals `SchemaVersion`. +- An unbound registry looks the version up among its descriptors. +- C# already refuses two types for one `(name, version)` in a registry (`CultCache.cs` `RegisterDescriptor`, "One type + owns a schema"), and c2a's `Register` refuses two types for one binding id. So an exact version lookup has at most one + answer. +- Two types sharing version `"1"` cannot both be bound, because `"1"` is refused first. + +File changes: +- `CultNetDocumentRegistry.cs:76-135` `ForDocument` and `ForDocument(Type…)`: delete the `schemaId` parameter + (OQ-C1.1). The binding id is `descriptor.SchemaVersion`. +- `CultNetDocumentBinding`: the internal constructor at `:44` takes the id from the descriptor. +- `CultNetDocumentRegistry.cs:173` `Register`: before c2a's conflict check, refuse a binding whose id fails + `^.+\.v[0-9]+$`. Throw `CultSchemaConflictException` or `ArgumentException`, naming the type, the declared version, + and the rule. The refusal happens at registration, never at first send. +- `CultNetDocumentRegistry.cs:675-676` `WireSchemaId`: becomes `descriptor.SchemaVersion`, checked by the same rule, so + an unbound `"1"` type is refused when it is first emitted. Every emitter below calls it. +- `CultNetDocumentRegistry.cs:616-645` `ToRawRecord`: the synthesized binding at `:621-626` uses `WireSchemaId`, not + `descriptor.SchemaId`. +- `CultNetDocumentRegistry.cs:217-228` `CreateDocumentDeleteMessage(string schemaId…)`: add + `CreateDocumentDeleteMessage(messageId, CultRecordHandle)`, which derives the wire id. Keep the string overload, + documented as "the wire id". +- `CultNetDatabase.cs` emitters: publish the wire id wherever a `CultNetDatabaseChange.SchemaId` leaves the process. + - `CultNetDatabaseServer.cs:555/581` (removal change) sends `WireSchemaId(descriptor)`, not `change.SchemaId`. The + change record keeps the sha256 locally. + - `CultNetDatabase.cs:1554` `ToLogEntryMessage` `Delete.SchemaId` becomes the wire id. + - `CultNetDatabase.cs:1512` `ApplyCommittedDelete` resolves the incoming delete id through the one exact lookup, not + `GetRequiredBySchemaId`. + - Error text at `:1664, :1725` names the wire id. +- **Durable shard log** (`CultNetShardMutationLogStorage.cs:60-88`, `CultNetDatabase.cs:586-591, 1385-1409`): + - It persists store ids. `Append` rewrites each entry's `schemaId` (put document and delete) to + `descriptor.SchemaId`. + - `GetMutationLogMessages` restamps with `WireSchemaId` on the way out. + - `KeyOf` (`:1399-1409`, the restart rebuild of `_lastWriteSequence`) reads store ids and so agrees with the live + writes at `:1294, 1486, 1531`. That also fixes the rebuild disagreement that exists today when a binding overrides + the id. + - **Existing log files hold sha256 for default bindings, which is the store id, so nothing on disk changes.** Only + logs written through an override hold a foreign id, and only the interop peer used one. No repo outside CultLib uses + `CultNetFileShardMutationLogStore` (grep over F:\Projects). +- Request filters that send the sha256 today switch to `WireSchemaId`: + - `CultNetDatabaseSubscriptionClient.cs:70-71`; + - `CultMeshClient.cs:602, 843, 944`; + - `CultMesh.cs:2696` (`RequestPeerSnapshotAsync`) and `:2038`; + - `CultMeshSnapshots.cs:874` `ResolveDefaultSelection`. +- **Shard catalog and config:** `CultNetShardDescriptor.SchemaIds` and `CultNetClientAuthorityScope.SchemaIds` are wire + ids. They are documented so, and matched exactly (C1.0). +- CultMesh projection sources and binding records (`CultMesh.cs:1521, 1602, 1972, 2032, 2273, 2300, 2331, 2357, 2384, + 2413`; `CultMeshSnapshots.cs:530, 555`) cross the wire as `CultMeshStateBindingRecord [Key(3)]` + (`CultMeshPrimitives.cs:3334`). Stamp them with the version. + - The collection and feed ids built from them (`CultMesh.cs:2298, 2329, 2382, 2411, 3266`) move to the version too. + - The cultmesh-ts type of the same shape then agrees. +- **Selection order and cursor** (`CultNetSelectionEvaluator.cs:213-214, 318, 345, 586, 593`): tie-break and cursor on + the wire id, as Rust does on its binding id. + - This changes a wire-visible order: the parity vector R-AJ flips (`leaf_a.v1 < leaf_b.v1`, where the sha256 order + was `b1f2… < ff35…`). + - Cursors minted before the upgrade stop resuming. That is acceptable, because cursors are per session. +- **Shared fixture, landed in this cut with the Rust side:** + - `contracts/cultnet/interop/selection-vectors.fixture.json`: `schemaId` becomes the version. Delete `nameAlias` and + `hashAlias`. + - Regenerate `selection-vectors.cs-written.json` and `selection-vectors.rs-written.json`. + - `SelectionParityVectorTests.cs:112-117, 156-159` assert the fixture ids equal the declared versions. Delete the + alias cases at `:235, :238, :329`; keep the refusal of an unknown id at `:325`. + - Rust `cultnet-rs/tests/selection.rs:1561-1690, 1849` (vector writer): drop the alias rows; order by the version. +- **Left alone:** `global:{descriptor.SchemaId}` record keys (`CultCache.cs:2473`, `CultNetDatabase.cs:1120`). They are + a record key, not a schema id, and changing them changes store keys. See F-C1-GLOBALKEY. +- **Schema catalog:** `CultNetSchemaRegistry` holds JSON-Schema contracts keyed by `$id` URL, and no production C# host + answers a catalog request (`Server.cs:565` is an allowlist only). The only document-payload registrations are the + interop peer's, which C1.5 changes. Nothing to cut here. + +### 5. C1.2: TypeScript + +Runtime packages `cultcache-ts`, `cultnet-ts`, `cultmesh-ts` and `cultmesh-browser`. + +- `cultcache-ts/src/document.ts:82-91` `schemaIdentityOf` stays the store identity, unchanged. +- **cultnet-ts owns the wire:** `export function wireSchemaIdOf(definition)` returns + `schemaIdentityOf(definition).schemaVersion`. The store keeps a definition with no declared version valid; only a + wire registration refuses it. +- `cultnet-ts/src/replication.ts:52-80` `register`: + - refuse a version failing `/\.v\d+$/` with `SchemaConflictError` or `TypeError`, naming the type; + - key `#owners` by `wireSchemaIdOf`; + - `#listers` keep the declared compatible ids (OQ-C1.2). + - The conflict check at `:59-71` also compares wire ids. +- Outgoing ids: `schemaIdForBinding` (`:349-351`) splits in two. + - Outgoing messages (`:123, :178, :287` put echo) use `wireSchemaIdOf`. + - `applyRawDocumentPutMessage` (`:280-290`) keeps storing under the store id, `schemaIdentityOf(...).schemaId`. + That is the leak point: today the same function feeds both. +- **Serving** (`:196-260`, `:363-368` `schemaIdForEnvelope`, `:336-345` `#createRawDocumentRecord`): + - A served record carries `wireSchemaIdOf(binding)`, never `envelope.schemaId`. + - A record stored under a compatible id holds that version's bytes. It is decoded and re-encoded through the owning + definition before it is served under the owner's id. This matches C#, which serializes from the decoded document + (`CultNetDocumentRegistry.cs:640`). + - `answersTo` (`:354-361`) matches the requested id against the binding's wire id (and its compatible ids if + OQ-C1.2 is yes). + - Tests `cultnet.test.ts:1627-1664` (the wire id equals the store id), `:1703-1708` (served under the stored id) and + `:1713-1744` (a version id does not resolve) are inverted. +- `cultnet-ts/src/shard-catalog.ts`: exact after C1.0; shard ids are wire ids. +- **cultmesh-ts:** + - `cultMeshSchemaFromDefinition` (`index.ts:6168-6173`): `schemaId` for peer reads is `wireSchemaIdOf`. + - `syncDocumentFromPeerSnapshot` (`:3995-4019`) and `cultMeshDocumentFromPeerSnapshot` (`:2668-2710`) ask by wire id. + - `resolveCultMeshStoreDocumentRecord` / `cultMeshAcceptedSchemaIds` (`:6243-6256`) are store-local and stay on + `schemaIdentityOf`. + - The receipt filter at `:5976` compares `receiptBinding.definition.schemaId`, which is `undefined` when not + declared. Compare with `wireSchemaIdOf`. +- Put-serve overlap (not on c2a): `hands/cultmesh-put-serve-bound` adds `#rawPutEnvelope` and + `createRawSnapshotResponseForPut` to this file, on main's shape resolution (`preserveIncomingSchemaId`). After the + merge, the put-sizing response must be built by the same serving path, so it carries the wire id. +- `cultmesh-browser`: exact after C1.0. `samples/eve-browser-network` declares or asks by version. +- **Consumer action, not in CultLib:** Stonks `stonks-daemon.cjs:152-195` sets `schemaVersion` to the full id. It + lands in Stonks before C1 merges to main. + +### 6. C1.3: Python + +- `cultcache_py/documents.py:105-143` `catalog_entry` stays the store identity. Add `wire_schema_id(definition)` in + cultnet_py, returning `schema_version`. +- Registration in `cultnet_py` and `cultmesh_py` refuses a version failing `\.v\d+$`. +- Outgoing ids use `wire_schema_id`: + - `cultmesh_py/node.py:577` (put), `:641` (delete), `:824-827` (sync); + - `client.py:430` (subscribe); + - `node.py:767, 776` (snapshot, never the stored id; a compatible-id record is re-encoded as in TS); + - `server.py:617-633` (catalog). + - `server.py:416, 444` echo a record that has already been resolved exactly, so it carries a wire id. +- Applied records keep being stored under the local store id: `replication.py:42, 84`, `node.py:664-670`. +- Inverted tests: `test_cultmesh.py:4116-4148` (a version request is served under the sha256 id) and `:4215-4258` + (echo of a foreign id). +- `interop_peer.py:1085` serves the stored id; that moves in C1.5. + +### 7. C1.4: Rust mesh and Kotlin + +- **Rust.** The wire id is already the version on every consumer binding. + - `for_entry::()` (`cultnet-rs/src/replication.rs:44-47`) has no consumer. It binds `T::TYPE`, which fails the + refusal unless it ends in `.vN`. Refuse it at `register` like any other id without a version. Keep the constructor, + so a type whose TYPE is `x.v1` still works. + - `for_entry_with_schema_id(id, version)` (`:50-64`): refuse `id != version` or a missing `.vN`. That is one id, and + `cultmesh_documents!` already passes the same string twice. + - `register` (`replication.rs`) applies the rule. + - The apply path keeps stamping `binding.schema_id` (`:400`). That is what Odin, Muninn and the rest store today, so + nothing on disk changes. + - `pull_rudp_catalog_snapshot` (`cultmesh-rs/src/lib.rs:714-747`) stamps the resolved binding's id. It is exact after + C1.0, so no foreign id is written. + - Rust's mixed store ids (typed put stamps `T::TYPE` at `cultcache-rs/src/lib.rs:2362`; apply stamps the version) are + left as they are. That is C2's question, F-C1-RUSTSTOREID. +- **Kotlin.** + - `register` (`CultMesh.kt:79-93`) refuses a `schemaVersion` failing `\.v\d+$`. + - The interop codecs at `CultMesh.kt:3872-3880, 4317-4322, 4351-4354, 3779-3789` put the contract URL in + `schemaVersion`. They become `cultnet.interop_note.v0` and the capability `*.v0` versions. Kotlin's own suite + (`CultMesh.kt` self-checks) moves with them. + +### 7b. C1.5: the interop harness tests the default binding + +The parity witness hid the mismatch because every peer names the note by its JSON-Schema `$id` URL. C1.5 removes the +overrides so each runtime's default binding meets the others. + +- C# `tests/GameCult.Networking.InteropPeer/Program.cs`: + - `:77-80, :201-204`: `ForDocument(payloadSerializer:…, payloadDeserializer:…)` with no id. The + id is `cultnet.interop_note.v0`. + - `:1183-1196`: four capability bindings with no id; their versions are `*.v0` (`:1560-1572`). + - Delete the `*SchemaId` URL constants. + - `LoadSchemaRegistration` (`:1116-1137`): the catalog registration's `SchemaId` becomes `InteropSchemaVersion`. The + JSON `$id` stays inside `SchemaJson`. + - The catalog server at `:682-714` stops copying `SchemaId` into `SchemaVersion`. +- TS `packages/cultnet-ts/test/interop/cultnet-interop-shared.ts:10-19, 289-306` and `cultnet-interop-peer.ts:123, + 135, 896-930`: definitions declare `schemaVersion` and no `schemaId` override. The note id constant is at + `cultnet-interop.test.ts:43`. +- Python `cultnet_py/interop_peer.py:59-68, 869-884, 936-981, 1085, 1149`. +- Rust `cultnet-rs/examples/cultnet_interop_peer.rs:56-67, 1227-1261`: `for_entry_with_schema_id` with the version + only. +- Kotlin: C1.4. +- The Python cultmesh scripts inside `cultnet-interop.test.ts:3872-4060` declare `schema_version` only. +- **The added negative lane** is one scenario in `cultnet-interop.test.ts`. A C# host serves a note from a + **CultMesh-created registry** (`CultMesh.CreateCultNetDocumentRegistry`, the production path). A TS reader: + - asks by `cultnet.interop_note.v0` and gets it; + - asks by `cultnet.interop-note` (the name) and gets not-found; + - asks by the sha256 id and gets not-found. +- CI: `.github/workflows/cultnet-interop.yml:141-163` already runs every pair lane. The new scenario joins the + `TypeScript and C#` pattern. + +### 8. Verification + +Each rule gets a behavioural test that fails when the rule breaks, plus the ecosystem mutation tool on the cut's +diff: Stryker.NET, StrykerJS, mutmut, cargo-mutants. Kotlin has no tool, so its rules are defended by scenario. + +**Positive, across runtimes, with default bindings** (C1.5 lanes, run on Yggdrasil in `ack1d-interop:2`): +- A C# host with a CultMesh-created registry publishes a note, and TS, Rust, Python and Kotlin each read it. Each asks + only by `cultnet.interop_note.v0`. +- Then each non-C# host publishes, and C# reads. +- The wire bytes carry `schemaId = cultnet.interop_note.v0`. Assert the decoded `CultNetRawDocumentRecord`, not a log + line. + +**Negative checks:** +- **No fallback.** In each runtime, a record at the requested key under another id is not returned (TS, C#, browser), + and a payload whose slot 0 names a registered version but whose id is foreign is refused (C#, Python). + - Mutant: restore `?? atKey[0]`, or restore the `byPayload` or `aliased` branch. It must die. +- **Unknown id refused.** + - C# `DeserializeRawDocument` with `gamecult.unknown.v1` throws. Today it returns `Note` (probed). + - TS, Python, Rust and Kotlin `applyRawDocumentPut` with an unknown id throw. +- **No name match.** Asking by `schemaName` finds nothing, in every runtime and in the selection evaluator (C# and + Rust). Mutant: add `|| candidate == descriptor.SchemaName`. +- **Refusal.** + - Registering a C# binding for a `[CultDocument("x", "1")]` type throws at `Register` and names the type. + - An unbound `"1"` type in a registry with no bindings throws at first emission. + - TS, Python, Rust and Kotlin refuse `"v1"` and `"1"`, and accept `x.v0` and `x.v12`. + - Mutants that must die: `\.v\d*$` (accepts `x.v`), `.v` anywhere, and `\d+$` alone (accepts `"1"`). +- **The wire id is derived from the declaration, not a constant.** Two types with different versions each emit their + own. A mutant that returns `SchemaName + ".v1"` must die, so use a fixture whose version is not `.v1`: the + interop note's name is `cultnet.interop-note` and its version is `cultnet.interop_note.v0`. +- **The store id is untouched.** + - A C# store written before C1 is read after it. `dump.py` shows record ids unchanged after a C1 host reads and + serves. + - A TS store with `schemaId: "p1.shared"` keeps that id after a raw put arrives under `p1.shared.v1`. + - Mutant: store under the wire id. It must die. +- **The served id is not the stored id.** A TS or Python record stored under a compatible id is served under the + owner's wire id, with bytes that decode as the owner. Mutant: serve `envelope.schemaId`. +- **Durable log.** + - A C# database with `CultNetFileShardMutationLogStore` writes, restarts, and replays to a replica. The replayed + entries carry the version, and `LastWriteSequence` after the restart equals the value before it. + - A log file written before C1 (sha256 entries) replays under the version. +- **Selection.** The parity vectors agree C#↔Rust with version ids. The tie-break vector R-AJ orders by version. +- **Catalog lookup.** A catalog with two handles of one type refuses a by-schema lookup and names both document ids + (C# and TS). Mutant: return the first or last handle. + +**Builds.** These are the focused set, with no workspace-wide build: +- `dotnet test` on `GameCult.Networking.Tests` and `GameCult.Mesh.Tests`; +- `npm test` in `cultnet-ts`, `cultmesh-ts` and `cultmesh-browser`; +- `pytest` in `cultnet-py` and `cultmesh-py`; +- `cargo test -p cultnet-rs -p cultmesh-rs`; +- the Kotlin self-check; +- the interop lanes. +All of them run on Yggdrasil through `ygg-verify.sh`. + +**Negative greps**, which must return nothing in `src/` and `packages/*/src`: +- `CultNetSchemaAliasMatching|InferSchemaName|inferSchemaName|_infer_schema_name|infer_schema_name` +- `strip_schema_version_suffix|schema_alias|PayloadMatchesSchema|TryReadSchemaVersion|firstRecordAtKey` +- `bySchemaNameVersion|ExpandSchemaBindingAliases` +- in `CultNetDocumentRegistry.cs`: `schemaId:` as a `ForDocument` parameter. + +**Live consumer checks**, after release and before any consumer bumps: +- **Stonks** has its six versions fixed. `idunn` shows Odin holding Stonks' `provider_advertisement` after a Stonks + restart on C1. +- **AquaSynth** daemon and clients rebuilt together; the Dings host round-trips. +- **Mimir** EveDashboard is visible in Eve Android. +- **Bifrost** agent-transport snapshot and apply round-trip. +- **Odin** still ingests from Heimdall, VoidBot, weksa and StreamPixels. Those ids do not change. +- **Grep** Idunn and Odin for a receiving binding of `idunn.daemon_health` with no version. If one exists, it is + refused at startup under C1.4. It was not found in main. + +### 9. Sequencing + +1. **F2 finishes first.** C1 and F2 touch different layers: + - F2 is store-side: `CultCache.cs` `Claimant`, the single-file and directory stores, and the Rust catalog lay-back. + - C1 is wire-side. + - They overlap in three places: + - `cultnet-ts/src/replication.ts`. F2 and batch 10 built owner-beats-lister and the `#listers` refusal (F9, F10) + there, and C1.2 re-keys `#owners` by wire id. The lister rule is kept, with OQ-C1.2 deciding its wire reach. + - `cultnet-ts/test/cultnet.test.ts:1346-1394, 1627-1744`. C1.2 inverts tests F2's batches wrote. + - `cultnet_py/replication.py` `schema_document_map`. + - C1 does not relabel any stored record and adds no store write, so F2's rule ("never relabels … a write over it is + refused") is not in play. F2's test `976b57f1` ("each runtime's writes leave another runtime's records and catalog + entries unchanged") must stay green through C1. + - Hands on F2 should not be interrupted. C1 starts after F2's Soul pass holds. +2. **Then put-serve merges to main** (`hands/cultmesh-put-serve-bound`, based on `b04d03b6`). + - It edits `cultnet-ts/src/replication.ts` on main's resolution. Main still has `preserveIncomingSchemaId` and + payload-shape resolution, which c2a deleted in `0ee98e8c`. + - It also edits `cultmesh-ts/src/index.ts`, `cultmesh_py/server.py` and `node.py`, and `cultmesh-rs`. +3. **Then c2a merges main.** This is the conflict to resolve before C1, whatever C1 is: + - c2a's exact resolution plus put-serve's `createRawSnapshotResponseForPut`; + - main's R1 changes to `CultNetDocumentRegistry.cs` (`rowFilter`) against c2a's `Register` refusal. + - C1 re-anchors on that merge commit. Every anchor above names its symbol for that reason. +4. **C1 cuts, in order, on the variants branch** (or `hands/variants-c1id` cut from it): + - C1.0 C# subtraction; + - C1.0 TS, Python, Rust and Kotlin subtraction, one cut per runtime or together, since they are deletions only; + - C1.1 C# wire id with the shared selection fixture and Rust selection order; + - C1.2 TS; + - C1.3 Python; + - C1.4 Rust and Kotlin; + - C1.5 harness. + - The interop lane is red between C1.1 and C1.5 on the branch. Soul measures at C1.5. + - Stonks' fix lands before the variants branch merges to main. +5. **Release.** C1 is wire-breaking between C# peers of different releases, because an old host sends sha256 and a new + receiver refuses it. It rides the C0/C2a release that the 2026-09-30 ruling already gates. The release note names: + C# peers upgrade together, the refusal of versions without `.vN`, and Stonks. + +### 10. Ledger estimate + +| Cut | Deleted (src) | Added (src) | Tests | Notes | +|---|---|---|---|---| +| C1.0 C# | ~290: alias class 58; registry block ~175; mesh branches ~35; expand ~40 | ~40 exact lookups | ~15 tests inverted, ~6 deleted | public API removed: `CultNetSchemaAliasMatching`, `TryResolveDescriptorByPayloadSchema`, `PayloadMatchesSchema` | +| C1.0 TS/Py/Rust/Kt | ~230: selection `schema_alias` ~140 with tests; Python infer/sniff ~45; TS infer/HELD/catalog keys ~45 | ~20 | ~25 tests deleted or inverted | | +| C1.1 C# | ~30 (override parameter, sha256 fallbacks) | ~80 (refusal, log restamp, emitters) | ~20 changed; fixture and vectors regenerated | wire-visible order change | +| C1.2 TS | ~15 | ~50 (`wireSchemaIdOf`, refusal, serve re-encode) | ~10 | | +| C1.3 Py | ~10 | ~40 | ~8 | | +| C1.4 Rs/Kt | ~5 | ~25 | ~6 | | +| C1.5 harness | ~60 URL constants and overrides | ~40 negative lane | — | | +| **Net** | **~-640 / +295 src**, about **-345** | | | Nothing on disk changes. Two public C# methods and one public C# class are gone. | + +### 11. Operator questions + +**OQ-C1.1: delete the per-binding wire-id override?** +- The overrides are C# `ForDocument(schemaId:)`, Rust's free first argument of `for_entry_with_schema_id`, and the + implicit TS and Python `schemaId`-on-the-wire. +- A. Delete. The declared version is the one wire id, and it derives from the descriptor alone, so the selection + evaluator, the durable log and the mesh need no registry to compute it. No repo outside CultLib passes the override; + only CultLib's interop peer and two conflict tests do. +- B. Keep it, but refuse ids without `.vN`. A second declaration of identity survives beside the version, and every + descriptor-only site (evaluator order, cursor, log, projections) needs the registry again. That is today's + `WireSchemaId` split. +- **Recommendation: A.** It removes a public parameter, which the CultLib consumer rule asks to be put to you rather + than cut silently. What a reasonable consumer loses is the ability to put an id on the wire that its declaration + does not state. That is the thing C1 rules out. +- C1.1's shape depends on the answer. + +**OQ-C1.2: does declared compatibility (`CompatibleSchemaIds`) resolve on the wire?** +- A. Yes, with the same rule the stores use: owner beats lister, and two listers with no owner are refused. A v2 type + that declares `x.v1` compatible accepts a v1 peer's records, so mixed-version peers work when an author declares it. + It is declared, never inferred. c2a's TS and Python registries already do this. Served records always go out under + the owner's wire id, re-encoded (§5). +- B. No, wire resolution is the owner only. Compatible ids become store-only. It is simpler, and a version bump is a + hard break between peers. +- **Recommendation: A.** It is the rule the operator ruled for stores (owner beats lister), applied to one more layer. + It is not aliasing, because nothing is guessed. +- C1.0's Python map, C1.2's `#listers`, and the C# lookup depend on the answer. + +**Defaulted, not asked. Veto if wrong:** +- TS and Python keep their default version `${name}.v1` when none is declared. +- A served record stored under a compatible id is re-encoded through its owner, as C# does. +- `global:{sha256}` record keys stay. +- Rust's apply path keeps stamping the binding id. +- A catalog lookup by schema that more than one handle answers is refused. +- Stonks is fixed in Stonks. + +### 12. Follow-ups (no cut owns them) + +- **F-C1-GLOBALKEY:** `global:{descriptor.SchemaId}` keys (`CultCache.cs:2473`, `CultNetDatabase.cs:1120`) carry the + sha256 across the wire as a record key. This is for C2. +- **F-C1-RUSTSTOREID:** Rust stores hold `T::TYPE` for typed puts and the version for replicated puts. Its catalog + fabricates version `.v1` (`cultcache-rs/src/lib.rs:2694-2708`) even where `cultmesh_documents!` declares another + one. The derive cannot declare a version (`cultcache-rs-derive/src/lib.rs:24-33`), which is a gap beside the other + runtimes. This is for C2. +- **F-C1-PYALIAS:** Python cultmesh `_documents_alias` compares two local definitions by name and version. It is + redundant under one-type-per-schema, but local. +- **F-C1-HEALTHID:** Bifrost sends `idunn.signed_daemon_health` without `.v1` + (`persona-feedback-idunn-health.cjs:41`), where every other sender uses `.v1`. It is pre-existing and unrelated to + C1. +- **F-C1-AEVE-CONST:** the AetheriaEve daemon compares wire ids with version constants (`Program.cs:1591`) and so never + matched under sha256. It is archaeology; record only. diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index 0980f743..704a3851 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -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; 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: diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 6001c905..6e85f833 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -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() diff --git a/packages/cultcache-rs/README.md b/packages/cultcache-rs/README.md index 1ad01eb0..6841791f 100644 --- a/packages/cultcache-rs/README.md +++ b/packages/cultcache-rs/README.md @@ -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::().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`. + +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::().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** diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index c9a7d282..71bf8fca 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -280,8 +280,8 @@ struct PersistedRecord( ); impl<'de> serde::Deserialize<'de> for PersistedRecord { - /// Reads the four v1 slots and refuses a fifth, naming the record. A record with more - /// slots was written by a runtime that resolves document variants. + /// Reads the four v1 slots and refuses a fifth, naming the record. No runtime reads a + /// v1 store holding such a record, so the store is not a valid one. fn deserialize>(deserializer: D) -> std::result::Result { use serde::de::{Error, IgnoredAny, SeqAccess, Visitor}; @@ -301,7 +301,7 @@ impl<'de> serde::Deserialize<'de> for PersistedRecord { seq.next_element()?.ok_or_else(|| A::Error::invalid_length(3, &self))?; if seq.next_element::()?.is_some() { return Err(A::Error::custom(format!( - "CultCache record {key:?} (schema {schema_id:?}) has more than 4 slots; this runtime reads 4. The store needs a runtime that resolves document variants." + "{RECORD_REFUSAL}{key:?} (schema {schema_id:?}) has more than the 4 slots of a {STORE_FORMAT_V1} record, so this is not a valid store" ))); } Ok(PersistedRecord(key, schema_id, stored_at, payload.into_vec())) @@ -312,34 +312,159 @@ impl<'de> serde::Deserialize<'de> for PersistedRecord { } } -/// The store header string, or `None` when the bytes are not a snapshot-shaped array -/// (the legacy envelope array starts with a map). +const STORE_FORMAT_V1: &str = "cultcache.store.v1"; + +/// Opens every refusal CultCache raises about one persisted record. The decoder's own +/// messages can quote the bytes; these name only the record's key and schema id. +const RECORD_REFUSAL: &str = "CultCache record "; + +/// A store header as a refusal may show it: echoed when it has the shape +/// `cultcache.store.v`, and otherwise described only by its length, since the +/// bytes are the store's and an error text must not carry them. +fn describe_header(header: &str) -> String { + let known_shape = header + .strip_prefix("cultcache.store.v") + .is_some_and(|version| !version.is_empty() && version.bytes().all(|byte| byte.is_ascii_digit())); + if known_shape { + format!("{header:?}") + } else { + format!("an unrecognised cultcache.store.* header of {} bytes", header.len()) + } +} + +/// What a failed decode tells the operator: CultCache's own context and record +/// refusals, and for a decoder failure only its kind, never a value from the bytes. +fn describe_undecodable(error: &anyhow::Error) -> String { + let mut parts = Vec::new(); + for cause in error.chain() { + if let Some(decode) = cause.downcast_ref::() { + parts.push(describe_decode_error(decode)); + break; + } + parts.push(cause.to_string()); + } + parts.join(": ") +} + +/// A decoder failure as an error that keeps only its kind. Every decode of stored bytes +/// goes through this or [`describe_undecodable`], so no error text quotes a stored value. +fn opaque_decode_error(error: rmp_serde::decode::Error) -> anyhow::Error { + anyhow!(describe_decode_error(&error)) +} + +fn describe_decode_error(error: &rmp_serde::decode::Error) -> String { + use rmp_serde::decode::Error; + match error { + Error::InvalidMarkerRead(_) | Error::InvalidDataRead(_) => "the bytes end before the store does", + Error::TypeMismatch(_) => "a value has a MessagePack type the store format does not allow", + Error::OutOfRange => "a number is out of range for its slot", + Error::LengthMismatch(_) => "an array has a length the store format does not allow", + Error::Utf8Error(_) => "a string is not valid UTF-8", + Error::DepthLimitExceeded => "values are nested deeper than the decoder allows", + Error::Syntax(message) | Error::Uncategorized(message) if message.starts_with(RECORD_REFUSAL) => { + return message.clone(); + } + Error::Syntax(_) | Error::Uncategorized(_) => "a value does not fit the store format", + } + .to_string() +} + +/// The store header string, or `None` when the bytes do not open with an array whose +/// first slot is a string (the legacy envelope array starts with a map). Only the +/// leading bytes are read, so a store cut short after its header still names its format. fn store_header(bytes: &[u8]) -> Option { - struct Header(String); - impl<'de> serde::Deserialize<'de> for Header { - fn deserialize>(deserializer: D) -> std::result::Result { - struct HeaderVisitor; - impl<'de> serde::de::Visitor<'de> for HeaderVisitor { - type Value = Header; + let mut offset = 0usize; + if read_array_header(bytes, &mut offset)? == 0 { + return None; + } + read_string(bytes, &mut offset) +} - fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { - formatter.write_str("an array whose first slot is the store header") - } +/// A single-file store whose bytes were read and not accepted. It sits in the error +/// chain of every such read, so a consumer tells it apart with +/// `error.downcast_ref::()`. A failure to read the bytes +/// at all stays a [`std::io::Error`] in the chain and never carries this type; the +/// chain under this type never holds an `io::Error`, so the two never overlap. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct CultCacheStoreUnreadable { + pub path: PathBuf, + pub kind: CultCacheStoreUnreadableKind, +} - fn visit_seq>(self, mut seq: A) -> std::result::Result { - let header: String = seq - .next_element()? - .ok_or_else(|| ::invalid_length(0, &self))?; - while seq.next_element::()?.is_some() {} - Ok(Header(header)) - } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum CultCacheStoreUnreadableKind { + /// The bytes are not a store this runtime's format accepts: truncated, garbled, + /// or a record the format does not allow. + Undecodable, + /// The header names a store format this runtime does not read, such as one + /// written by a newer runtime. The rest of the file was not judged. + UnsupportedFormat, +} + +impl std::fmt::Display for CultCacheStoreUnreadable { + fn fmt(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + match self.kind { + CultCacheStoreUnreadableKind::Undecodable => { + write!(formatter, "CultCache store {} is not decodable", self.path.display()) } - deserializer.deserialize_seq(HeaderVisitor) + CultCacheStoreUnreadableKind::UnsupportedFormat => write!( + formatter, + "CultCache store {} is in a format this runtime does not read", + self.path.display() + ), } } - rmp_serde::from_slice::
(bytes).ok().map(|header| header.0) } +impl std::error::Error for CultCacheStoreUnreadable {} + +/// A write to a single-file store that failed, and what it left on disk. It sits in the +/// error chain of every such failure, above the cause, so a consumer tells the outcomes +/// apart with `error.downcast_ref::()`. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct CultCacheStoreWriteFailed { + pub path: PathBuf, + pub kind: CultCacheStoreWriteFailedKind, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum CultCacheStoreWriteFailedKind { + /// The entries cannot be stored as a snapshot, for example two document types + /// claim one schema id. Nothing was written, and retrying the same entries fails + /// the same way: this is a refusal, not an I/O failure. + Rejected, + /// The write failed before the store file was replaced (creating, writing or + /// syncing the staging file, or renaming it): the file holds what it held before. + NotReplaced, + /// The store file was replaced, then syncing its directory failed: the new + /// snapshot is in place but may not survive a crash. + ReplacedNotDurable, +} + +impl std::fmt::Display for CultCacheStoreWriteFailed { + fn fmt(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + match self.kind { + CultCacheStoreWriteFailedKind::Rejected => { + write!(formatter, "CultCache store {} refused a snapshot it cannot store", self.path.display()) + } + CultCacheStoreWriteFailedKind::NotReplaced => { + write!(formatter, "CultCache store {} was not replaced", self.path.display()) + } + CultCacheStoreWriteFailedKind::ReplacedNotDurable => write!( + formatter, + "CultCache store {} was replaced, but the replacement is not yet durable", + self.path.display() + ), + } + } +} + +impl std::error::Error for CultCacheStoreWriteFailed {} + #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct PushAllOptions { pub soft: bool, @@ -416,10 +541,12 @@ impl SingleFileMessagePackBackingStore { /// either file. Writers using the sibling lock cannot replace the snapshot /// until `action` returns. /// - /// Releasing the lock is RAII cleanup. A release failure therefore cannot - /// rewrite the completed action's result; consumers that need to observe - /// that cleanup failure can use - /// [`Self::with_read_only_shared_snapshot_and_unlock_diagnostic`]. + /// Releasing the lock is RAII cleanup, under the same policy as every lock in + /// this crate (see `release_lock`): a release failure cannot rewrite the + /// completed action's result. This path alone also offers an observer for that + /// failure, [`Self::with_read_only_shared_snapshot_and_unlock_diagnostic`], + /// because a provider crossing may need to report it; the locked read and write + /// paths discard it. pub fn with_read_only_shared_snapshot( &self, action: impl FnOnce(Vec) -> Result, @@ -711,19 +838,44 @@ impl SingleFileMessagePackBackingStore { } fn read_all_unlocked(&self) -> Result> { - if !self.path.exists() { - return Ok(Vec::new()); - } - let bytes = fs::read(&self.path) - .with_context(|| format!("failed to read {}", self.path.display()))?; - if bytes.is_empty() { - return Ok(Vec::new()); - } + // Only a store that is not there is empty. A file that is there and empty is + // not a store (no writer leaves one), and any other failure to reach the file, + // such as a symlink loop, a file where a directory should be, or no permission, + // stays an io::Error: reading it as empty would let a writer replace a store it + // never saw. + // A dangling symbolic link is something at the path, not nothing: reading it as + // empty would let the writer rename a file over the link and move the store off + // the volume the link pointed to. + let bytes = match fs::read(&self.path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound && is_absent(&self.path)? => { + return Ok(Vec::new()); + } + Err(error) => { + return Err(anyhow::Error::new(error) + .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.starts_with("cultcache.store.") => decode_store_snapshot(&bytes), - _ => rmp_serde::from_slice(&bytes).map_err(anyhow::Error::from), + 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( + CultCacheStoreUnreadableKind::UnsupportedFormat, + anyhow!("its header is {}; this runtime reads {STORE_FORMAT_V1:?} only", describe_header(&header)), + )), + _ => rmp_serde::from_slice(&bytes) + .context("failed to decode legacy CultCache envelope array") + .map_err(|error| unreadable(CultCacheStoreUnreadableKind::Undecodable, error)), } - .map_err(|error| anyhow!("failed to decode MessagePack {}: {error:#}", self.path.display())) } /// Reads one filesystem snapshot without creating or opening the sibling @@ -739,29 +891,39 @@ impl SingleFileMessagePackBackingStore { } fn write_all_unlocked(&self, entries: &[CultCacheEnvelope]) -> Result<()> { + let failed = |kind| CultCacheStoreWriteFailed { + path: self.path.clone(), + kind, + }; + let bytes = encode_store_snapshot(entries) + .and_then(|snapshot| rmp_serde::to_vec(&snapshot).context("failed to encode MessagePack")) + .map_err(|error| error.context(failed(CultCacheStoreWriteFailedKind::Rejected)))?; + self.replace_unlocked(&bytes) + .map_err(|error| error.context(failed(CultCacheStoreWriteFailedKind::NotReplaced)))?; + sync_parent_directory(&self.path) + .map_err(|error| error.context(failed(CultCacheStoreWriteFailedKind::ReplacedNotDurable))) + } + + /// Everything up to and including the rename over the store file. On failure the + /// store file is as it was. + fn replace_unlocked(&self, bytes: &[u8]) -> Result<()> { if let Some(parent) = self.path.parent() { fs::create_dir_all(parent) .with_context(|| format!("failed to create {}", parent.display()))?; } remove_abandoned_staging_files(&self.path)?; - let bytes = rmp_serde::to_vec(&encode_store_snapshot(entries)?) - .context("failed to encode MessagePack")?; let tmp_path = temporary_path_for(&self.path); - let mut staged = OpenOptions::new() + let staged = OpenOptions::new() .create_new(true) .write(true) .open(&tmp_path) .with_context(|| format!("failed to create {}", tmp_path.display()))?; - use std::io::Write; - staged - .write_all(&bytes) - .with_context(|| format!("failed to write {}", tmp_path.display()))?; - staged - .sync_all() - .with_context(|| format!("failed to sync {}", tmp_path.display()))?; - drop(staged); - replace_file_atomically(&tmp_path, &self.path)?; - sync_parent_directory(&self.path)?; + // From here the staging file is this write's to remove: a write that fails + // for lack of space must not keep holding the space it failed on. + if let Err(error) = stage_and_replace(staged, &tmp_path, bytes, &self.path) { + let _ = fs::remove_file(&tmp_path); + return Err(error); + } Ok(()) } @@ -769,20 +931,14 @@ impl SingleFileMessagePackBackingStore { let lock = self.open_lock_file()?; fs2::FileExt::lock_shared(&lock) .with_context(|| format!("failed to lock {}", self.lock_path().display()))?; - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result + release_lock(lock, action()) } fn with_exclusive_lock(&self, action: impl FnOnce() -> Result) -> Result { let lock = self.open_lock_file()?; fs2::FileExt::lock_exclusive(&lock) .with_context(|| format!("failed to lock {}", self.lock_path().display()))?; - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result + release_lock(lock, action()) } /// Replaces the named entries only if each still matches its expectation. @@ -899,10 +1055,7 @@ impl SingleFileMessagePackBackingStore { .with_context(|| format!("failed to lock {}", self.lock_path().display())); } } - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result.map(Some) + release_lock(lock, action()).map(Some) } fn open_lock_file(&self) -> Result { @@ -987,7 +1140,7 @@ fn unique_batch_ids( impl CacheBackingStore for SingleFileMessagePackBackingStore { fn pull_all(&self) -> Result> { - if !self.path.exists() && !self.lock_path().exists() { + if is_absent(&self.path)? && is_absent(&self.lock_path())? { return Ok(Vec::new()); } self.with_shared_lock(|| self.read_all_unlocked()) @@ -1092,10 +1245,7 @@ impl RedbMessagePackBackingStore { .with_context(|| format!("failed to open {}", lock_path.display()))?; fs2::FileExt::lock_exclusive(&lock) .with_context(|| format!("failed to lock {}", lock_path.display()))?; - let result = self.open_database().and_then(|database| action(&database)); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", lock_path.display()))?; - result + release_lock(lock, self.open_database().and_then(|database| action(&database))) } fn lock_path(&self) -> PathBuf { @@ -1875,7 +2025,9 @@ fn read_redb_entry( table .get(key.as_slice())? .map(|value| { - rmp_serde::from_slice(value.value()).context("failed to decode redb CultCache envelope") + rmp_serde::from_slice(value.value()) + .map_err(opaque_decode_error) + .context("failed to decode redb CultCache envelope") }) .transpose() } @@ -1897,7 +2049,9 @@ fn read_all_redb( .iter()? .map(|row| { let (_, value) = row?; - rmp_serde::from_slice(value.value()).context("failed to decode redb CultCache envelope") + rmp_serde::from_slice(value.value()) + .map_err(opaque_decode_error) + .context("failed to decode redb CultCache envelope") }) .collect::>>()?; entries.sort_by_key(entry_id); @@ -2128,7 +2282,7 @@ impl CultCache { let Some(entry) = self.entries.get(&entry_id_parts(T::TYPE, key)) else { return Ok(None); }; - let payload = rmp_serde::from_slice(&entry.payload).with_context(|| { + let payload = rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2161,7 +2315,7 @@ impl CultCache { if entry.r#type != T::TYPE { continue; } - values.push(rmp_serde::from_slice(&entry.payload).with_context(|| { + values.push(rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2182,7 +2336,7 @@ impl CultCache { } values.push(( entry.key.clone(), - rmp_serde::from_slice(&entry.payload).with_context(|| { + rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to decode CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2264,7 +2418,7 @@ impl CultCache { key: String, payload: Vec, ) -> Result<(CultCacheEnvelope, T)> { - let parsed: T = rmp_serde::from_slice(&payload).with_context(|| { + let parsed: T = rmp_serde::from_slice(&payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to validate CultCache entry {:?} at key {:?} as {}", T::TYPE, @@ -2400,7 +2554,7 @@ impl CultCache { )); } - rmp_serde::from_slice(&entry.payload).with_context(|| { + rmp_serde::from_slice(&entry.payload).map_err(opaque_decode_error).with_context(|| { format!( "failed to validate CultCache envelope {:?} at key {:?} as {}", T::TYPE, @@ -2574,20 +2728,14 @@ fn encode_store_snapshot(entries: &[CultCacheEnvelope]) -> Result Result> { - if let Some(header) = store_header(bytes) { - if header != "cultcache.store.v1" { - return Err(anyhow!( - "CultCache store format {header:?} is not readable; this runtime reads \"cultcache.store.v1\" only. The store needs a runtime that resolves document variants." - )); - } - } let snapshot: PersistedStoreSnapshot = rmp_serde::from_slice(bytes).context("failed to decode CultCache v1 snapshot")?; @@ -2727,6 +2875,47 @@ fn temporary_path_for(path: &Path) -> PathBuf { path.with_file_name(file_name) } +/// Writes and syncs the staged snapshot, closes it, and renames it over `destination`. +fn stage_and_replace(mut staged: File, staged_path: &Path, bytes: &[u8], destination: &Path) -> Result<()> { + use std::io::Write; + staged + .write_all(bytes) + .with_context(|| format!("failed to write {}", staged_path.display()))?; + injected_write_fault(WriteStep::StagingSync) + .and_then(|()| staged.sync_all()) + .with_context(|| format!("failed to sync {}", staged_path.display()))?; + drop(staged); + replace_file_atomically(staged_path, destination) +} + +/// Releases a lock taken around `result`'s action and returns that result. +/// +/// The policy for every lock in this crate: an unlock failure never replaces the action's +/// result, so a write that landed is reported as landed. The failure is discarded here +/// because these paths have no observer; the read-only snapshot, whose API offers one, +/// hands it to [`SingleFileMessagePackBackingStore::with_read_only_shared_snapshot_and_unlock_diagnostic`]. +/// +/// Dropping `lock` closes this descriptor, which releases the lock only when no other +/// descriptor shares its open file description. A child forked without exec holds such a +/// duplicate, and the lock then stays held until the child closes it or exits. +fn release_lock(lock: File, result: Result) -> Result { + let _unlock_failure_released_by_close = + injected_write_fault(WriteStep::Unlock).and_then(|()| fs2::FileExt::unlock(&lock)); + drop(lock); + result +} + +/// True only when nothing is at `path`; a symbolic link, dangling or not, is something. +/// A path that cannot be inspected (a file where a directory should be, no permission) +/// is an error, never absent. +fn is_absent(path: &Path) -> Result { + match fs::symlink_metadata(path) { + Ok(_) => Ok(false), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(true), + Err(error) => Err(anyhow::Error::new(error).context(format!("failed to inspect {}", path.display()))), + } +} + fn remove_abandoned_staging_files(destination: &Path) -> Result<()> { let Some(parent) = destination.parent() else { return Ok(()); @@ -2870,11 +3059,41 @@ fn sync_parent_directory(path: &Path) -> Result<()> { let Some(parent) = path.parent() else { return Ok(()); }; - File::open(parent) + injected_write_fault(WriteStep::DirectorySync) + .and_then(|()| File::open(parent)) .and_then(|directory| directory.sync_all()) .with_context(|| format!("failed to sync {}", parent.display())) } +/// The steps no test can make fail for real: fsync and unlock do not fail on a healthy +/// filesystem. Tests arm one step on their own thread; builds outside tests never fail here. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[allow(dead_code)] +enum WriteStep { + StagingSync, + DirectorySync, + Unlock, +} + +#[cfg(test)] +thread_local! { + static INJECTED_WRITE_FAULT: std::cell::Cell> = const { std::cell::Cell::new(None) }; +} + +#[cfg(test)] +fn injected_write_fault(step: WriteStep) -> std::io::Result<()> { + if INJECTED_WRITE_FAULT.with(|fault| fault.get()) == Some(step) { + return Err(std::io::Error::other("injected write fault")); + } + Ok(()) +} + +#[cfg(not(test))] +#[inline(always)] +fn injected_write_fault(_step: WriteStep) -> std::io::Result<()> { + Ok(()) +} + #[cfg(windows)] fn sync_parent_directory(_path: &Path) -> Result<()> { // MoveFileExW with MOVEFILE_WRITE_THROUGH owns publication durability. @@ -2993,6 +3212,8 @@ mod tests { store.push(&released)?; let writer_path = store_path.clone(); + let gate_path = store_path.clone(); + let gate_released = released.clone(); let (started_tx, started_rx) = mpsc::channel(); let (finished_tx, finished_rx) = mpsc::channel(); store.with_read_only_shared_snapshot(|captured| { @@ -3007,15 +3228,21 @@ mod tests { writer.push(&engaged).unwrap(); finished_tx.send(()).unwrap(); }); - started_rx.recv_timeout(Duration::from_secs(1))?; + started_rx.recv_timeout(Duration::from_secs(30))?; assert!( finished_rx .recv_timeout(Duration::from_millis(100)) .is_err() ); + // The writer is still waiting on the gate: the file on disk is untouched. + assert_eq!( + SingleFileMessagePackBackingStore::new(&gate_path).pull_all_read_only_snapshot()?, + vec![gate_released.clone()] + ); Ok(()) })?; - finished_rx.recv_timeout(Duration::from_secs(1))?; + // The writer fsyncs twice once the gate opens; under IO pressure that is slow. + finished_rx.recv_timeout(Duration::from_secs(60))?; assert_eq!(store.pull_all()?[0].payload, b"engaged"); Ok(()) } @@ -4698,4 +4925,657 @@ mod tests { assert_eq!(envelopes[1].payload, b"\x92\xa4beta\x02"); Ok(()) } + + // Store read and write faults: what a consumer can tell apart, and what a failed + // write leaves on disk. + + fn unreadable_kind(error: &anyhow::Error) -> Option { + error + .downcast_ref::() + .map(|unreadable| unreadable.kind) + } + + fn io_error_in(error: &anyhow::Error) -> Option<&std::io::Error> { + error.chain().find_map(|cause| cause.downcast_ref::()) + } + + /// A store of three records written by this runtime, returned with its bytes. + fn written_store(path: &Path) -> Result<(SingleFileMessagePackBackingStore, Vec)> { + let store = SingleFileMessagePackBackingStore::new(path); + let records = vec![ + snapshot_envelope("alpha", &[1; 64]), + snapshot_envelope("beta", &[2; 64]), + snapshot_envelope("gamma", &[3; 64]), + ]; + assert!(store.compare_exchange_snapshot(&[], &records)?); + let bytes = fs::read(path)?; + Ok((store, bytes)) + } + + #[test] + fn a_truncated_store_is_undecodable_on_every_read_and_is_left_as_it_was() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, bytes) = written_store(&path)?; + let truncated = &bytes[..bytes.len() / 2]; + fs::write(&path, truncated)?; + + let failures = [ + store.pull_all().unwrap_err(), + store.pull_all_read_only_snapshot().unwrap_err(), + store + .compare_exchange_snapshot(&[], &[snapshot_envelope("delta", b"4")]) + .unwrap_err(), + ]; + for error in &failures { + assert_eq!(unreadable_kind(error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); + assert_eq!(error.downcast_ref::().unwrap().path, path); + assert!(error.to_string().contains(&path.display().to_string()), "{error}"); + assert!(io_error_in(error).is_none(), "{error:#}"); + } + assert_eq!(fs::read(&path)?, truncated); + Ok(()) + } + + #[test] + fn bytes_that_are_not_a_store_are_undecodable() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + // The last opens like a store, with a string where the header goes, but only a + // `cultcache.store.*` header claims a format this runtime might not read. + for bytes in [ + &b"not a store at all"[..], + &[0x92, 0xa3, b'a', b'b'][..], + &[0xc1][..], + &[0x92, 0xa2, b'a', b'b', 0x90][..], + &[0x93, 0xb0, b'c', b'u', b'l', b't', b'c', b'a', b'c', b'h', b'e', b'.', b's', b't', b'o', b'r', b'e', b'X', 0x90, 0x90][..], + ] { + fs::write(&path, bytes)?; + let error = store.pull_all().unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{bytes:?}: {error:#}"); + } + Ok(()) + } + + #[test] + fn a_v1_store_holding_a_record_v1_does_not_allow_is_undecodable() { + let error = pull_vector("extra-slot-full-payload.msgpack").unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); + } + + #[test] + fn a_store_in_a_format_this_runtime_does_not_read_is_unsupported_even_when_cut_short() -> Result<()> { + for name in ["unknown-header.msgpack", "variant-v2.msgpack", "../document-variants-c1/variant-store.msgpack"] { + let error = pull_vector(name).unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::UnsupportedFormat), "{name}: {error:#}"); + } + + // The header is all this runtime judges: a newer store cut short is still a + // newer store, not a damaged one this runtime could replace. + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.msgpack"); + let vector = fs::read( + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors/document-variants-c0/unknown-header.msgpack"), + )?; + let truncated = &vector[..vector.len() / 2]; + fs::write(&path, truncated)?; + let store = SingleFileMessagePackBackingStore::new(&path); + for error in [store.pull_all().unwrap_err(), store.pull_all_read_only_snapshot().unwrap_err()] { + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::UnsupportedFormat), "{error:#}"); + assert!(format!("{error:#}").contains("cultcache.store.v9"), "{error:#}"); + assert!(io_error_in(&error).is_none(), "{error:#}"); + } + assert_eq!(fs::read(&path)?, truncated); + Ok(()) + } + + #[test] + fn a_store_that_cannot_be_read_is_an_io_error_and_never_unreadable() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + fs::create_dir(&path)?; + let store = SingleFileMessagePackBackingStore::new(&path); + for error in [store.pull_all().unwrap_err(), store.pull_all_read_only_snapshot().unwrap_err()] { + assert!(io_error_in(&error).is_some(), "{error:#}"); + assert_eq!(unreadable_kind(&error), None, "{error:#}"); + } + Ok(()) + } + + /// Runs one write in a child process whose file-size limit is far below the + /// snapshot, with SIGXFSZ ignored so the write fails with EFBIG instead of + /// killing the child: a real mid-write failure after the staging file exists, + /// reached without permissions (the verification container runs as root). + #[cfg(unix)] + #[test] + fn a_write_that_fails_part_way_leaves_no_staging_file_and_keeps_the_store() -> Result<()> { + const CHILD_STORE: &str = "CULTCACHE_RS_FILE_SIZE_LIMITED_STORE"; + if let Ok(path) = std::env::var(CHILD_STORE) { + let mut store = SingleFileMessagePackBackingStore::new(path); + let error = store + .push(&snapshot_envelope("large", &vec![7; 1 << 20])) + .unwrap_err(); + let io = io_error_in(&error).expect("a size-limited write fails with an io::Error"); + assert_eq!(io.kind(), std::io::ErrorKind::FileTooLarge, "{error:#}"); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::NotReplaced), "{error:#}"); + assert_eq!(unreadable_kind(&error), None, "{error:#}"); + return Ok(()); + } + + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, before) = written_store(&path)?; + let child = std::process::Command::new("sh") + .arg("-c") + .arg("trap '' XFSZ; ulimit -f 64; exec \"$0\" --exact --test-threads=1 --quiet tests::a_write_that_fails_part_way_leaves_no_staging_file_and_keeps_the_store") + .arg(std::env::current_exe()?) + .env(CHILD_STORE, &path) + .output()?; + assert!( + child.status.success(), + "the size-limited child did not fail its write as expected: {} +{}{}", + child.status, + String::from_utf8_lossy(&child.stdout), + String::from_utf8_lossy(&child.stderr) + ); + + let staging: Vec<_> = fs::read_dir(temp.path())? + .map(|entry| entry.map(|entry| entry.file_name().to_string_lossy().into_owned())) + .collect::>>()? + .into_iter() + .filter(|name| name.starts_with("store.cc.") && name.ends_with(".tmp")) + .collect(); + assert_eq!(staging, Vec::::new()); + assert_eq!(fs::read(&path)?, before); + assert_eq!(store.pull_all()?.len(), 3); + Ok(()) + } + + /// Every read of a single-file store: the locked pull, the lock-free provider + /// snapshot, and the exchange a writer reads through. + fn every_read(store: &SingleFileMessagePackBackingStore) -> [Result>; 3] { + [ + store.pull_all(), + store.pull_all_read_only_snapshot(), + store + .compare_exchange_snapshot(&[], &[snapshot_envelope("delta", b"4")]) + .map(|_| Vec::new()), + ] + } + + fn assert_io_failure_on_every_read(store: &SingleFileMessagePackBackingStore) { + for result in every_read(store) { + let error = result.expect_err("a store that cannot be reached must not read as empty"); + let io = io_error_in(&error).expect("the failure to reach the store is an io::Error"); + assert_ne!(io.kind(), std::io::ErrorKind::NotFound, "{error:#}"); + assert_eq!(unreadable_kind(&error), None, "{error:#}"); + } + } + + #[test] + fn only_a_missing_store_reads_as_empty() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + assert!(store.pull_all()?.is_empty()); + assert!(store.pull_all_read_only_snapshot()?.is_empty()); + assert!(store.compare_exchange_snapshot(&[], &[snapshot_envelope("delta", b"4")])?); + + // A store whose lock file is gone is still a store, not an absent one. + fs::remove_file(store.lock_path())?; + assert_eq!(store.pull_all()?, vec![snapshot_envelope("delta", b"4")]); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn a_store_behind_a_symlink_loop_is_an_io_error_on_every_read() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + std::os::unix::fs::symlink(&path, &path)?; + assert_io_failure_on_every_read(&SingleFileMessagePackBackingStore::new(&path)); + assert!(fs::symlink_metadata(&path)?.file_type().is_symlink()); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn a_store_under_a_file_instead_of_a_directory_is_an_io_error_on_every_read() -> Result<()> { + let temp = tempfile::tempdir()?; + let parent = temp.path().join("body"); + fs::write(&parent, b"not a directory")?; + assert_io_failure_on_every_read(&SingleFileMessagePackBackingStore::new(parent.join("store.cc"))); + assert_eq!(fs::read(&parent)?, b"not a directory"); + Ok(()) + } + + #[test] + fn an_existing_empty_file_is_undecodable_and_is_left_as_it_was() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + fs::write(&path, b"")?; + let store = SingleFileMessagePackBackingStore::new(&path); + for result in every_read(&store) { + let error = result.expect_err("no CultCache writer leaves an empty file, so it is not a store"); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); + assert!(io_error_in(&error).is_none(), "{error:#}"); + } + assert_eq!(fs::read(&path)?, b""); + + // An empty array is the legacy envelope array with no envelopes: it has no + // header, so it claims no format, and it is an empty store. + fs::write(&path, [0x90])?; + assert!(store.pull_all()?.is_empty()); + assert!(store.pull_all_read_only_snapshot()?.is_empty()); + Ok(()) + } + + #[test] + fn a_newer_store_is_named_once_its_header_string_is_complete() -> Result<()> { + let vector = fs::read( + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors/document-variants-c0/unknown-header.msgpack"), + )?; + // One byte of array header, one of string header, then the 18-byte header string. + let header = "cultcache.store.v9"; + let header_end = 2 + header.len(); + assert_eq!(header_end, 20); + assert_eq!(&vector[2..header_end], header.as_bytes()); + + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + for length in 1..=vector.len() { + fs::write(&path, &vector[..length])?; + let error = store.pull_all().unwrap_err(); + let expected = if length < header_end { + CultCacheStoreUnreadableKind::Undecodable + } else { + CultCacheStoreUnreadableKind::UnsupportedFormat + }; + assert_eq!(unreadable_kind(&error), Some(expected), "cut to {length}: {error:#}"); + } + Ok(()) + } + + #[test] + fn refusals_say_what_was_found_without_blaming_variants() { + let unsupported = format!("{:#}", pull_vector("unknown-header.msgpack").unwrap_err()); + assert!(unsupported.contains("\"cultcache.store.v9\""), "{unsupported}"); + assert!(unsupported.contains("does not read"), "{unsupported}"); + assert!(!unsupported.contains("variant"), "{unsupported}"); + + // The C# reference refuses the same bytes: a v1 store with a longer record is + // not a store any runtime reads, so no upgrade is suggested. + let invalid = format!("{:#}", pull_vector("extra-slot-full-payload.msgpack").unwrap_err()); + assert!(invalid.contains("not a valid store"), "{invalid}"); + assert!(!invalid.contains("variant"), "{invalid}"); + assert!(!invalid.contains("runtime that"), "{invalid}"); + } + + /// The store path is a directory, so the staging file is written and synced and + /// then cannot be renamed over it: a failure after the write, reached without + /// permissions. + #[test] + fn a_write_that_fails_at_the_rename_leaves_no_staging_file() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + fs::create_dir(&path)?; + fs::write(path.join("kept"), b"kept")?; + let mut store = SingleFileMessagePackBackingStore::new(&path); + + let error = store + .push_all(&[snapshot_envelope("alpha", b"1")], PushAllOptions::default()) + .unwrap_err(); + assert!(format!("{error:#}").contains("failed to atomically replace"), "{error:#}"); + + let staging: Vec<_> = fs::read_dir(temp.path())? + .map(|entry| entry.map(|entry| entry.file_name().to_string_lossy().into_owned())) + .collect::>>()? + .into_iter() + .filter(|name| name.starts_with("store.cc.") && name.ends_with(".tmp")) + .collect(); + assert_eq!(staging, Vec::::new()); + assert_eq!(fs::read(path.join("kept"))?, b"kept"); + Ok(()) + } + + /// MessagePack bytes of a fixstr, for store fixtures built by hand. + fn fixstr(text: &str) -> Vec { + assert!(text.len() < 32); + let mut bytes = vec![0xa0 | text.len() as u8]; + bytes.extend_from_slice(text.as_bytes()); + bytes + } + + #[test] + fn a_refusal_never_echoes_a_value_from_the_store() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + + // A v1 store whose catalog holds a string where an entry goes, and a legacy + // envelope array holding a string where an envelope goes: the decoder's own + // message would quote both strings. + let mut v1 = vec![0x93]; + v1.extend(fixstr(STORE_FORMAT_V1)); + v1.push(0x91); + v1.extend(fixstr("SECRET-CATALOG-TEXT")); + v1.push(0x90); + let mut legacy = vec![0x91]; + legacy.extend(fixstr("SECRET-LEGACY-TEXT")); + for bytes in [v1, legacy] { + fs::write(&path, &bytes)?; + for result in every_read(&store) { + let error = result.unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); + assert!(!format!("{error:#}").contains("SECRET"), "{error:#}"); + assert!(!format!("{error:?}").contains("SECRET"), "{error:?}"); + } + } + Ok(()) + } + + #[test] + fn a_header_is_echoed_only_in_the_known_shape() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + for (header, echoed) in [ + ("cultcache.store.v12", true), + ("cultcache.store.SECRET-HEADER", false), + ("cultcache.store.v12SECRET", false), + ("cultcache.store.v", false), + ] { + let mut bytes = vec![0x93]; + bytes.extend(fixstr(header)); + bytes.extend([0x90, 0x90]); + fs::write(&path, &bytes)?; + let error = store.pull_all().unwrap_err(); + let text = format!("{error:#}"); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::UnsupportedFormat), "{text}"); + if echoed { + assert!(text.contains(&format!("\"{header}\"")), "{text}"); + } else { + assert!(!text.contains(&format!("\"{header}\"")), "{text}"); + assert!(!text.contains("SECRET"), "{text}"); + assert!(text.contains(&format!("of {} bytes", header.len())), "{text}"); + } + } + Ok(()) + } + + /// Nothing at the store path is an empty store; a symbolic link whose target is + /// gone is something, and a writer must not rename a file over it. + #[cfg(unix)] + #[test] + fn a_dangling_symlink_at_the_store_path_is_an_error_and_nothing_is_written() -> Result<()> { + let temp = tempfile::tempdir()?; + let volume = temp.path().join("volume"); + fs::create_dir(&volume)?; + let target = volume.join("store.cc"); + let path = temp.path().join("store.cc"); + std::os::unix::fs::symlink(&target, &path)?; + let store = SingleFileMessagePackBackingStore::new(&path); + for result in every_read(&store) { + let error = result.expect_err("a dangling link must not read as an empty store"); + assert!(io_error_in(&error).is_some(), "{error:#}"); + assert_eq!(unreadable_kind(&error), None, "{error:#}"); + } + assert!(fs::symlink_metadata(&path)?.file_type().is_symlink()); + assert!(!target.exists()); + assert_eq!(fs::read_dir(&volume)?.count(), 0); + Ok(()) + } + + fn write_failure_kind(error: &anyhow::Error) -> Option { + error + .downcast_ref::() + .map(|failed| failed.kind) + } + + /// Runs `write` with `step` made to fail on this thread. + fn with_write_fault(step: WriteStep, write: impl FnOnce() -> T) -> T { + INJECTED_WRITE_FAULT.with(|fault| fault.set(Some(step))); + let result = write(); + INJECTED_WRITE_FAULT.with(|fault| fault.set(None)); + result + } + + #[test] + fn a_write_that_fails_before_the_rename_is_not_replaced_and_the_store_is_intact() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, before) = written_store(&path)?; + let error = with_write_fault(WriteStep::StagingSync, || { + store.compare_exchange_snapshot(&store.pull_all().unwrap(), &[snapshot_envelope("delta", b"4")]) + }) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::NotReplaced), "{error:#}"); + assert!(io_error_in(&error).is_some(), "{error:#}"); + assert!(error.to_string().contains(&format!("{} was not replaced", path.display())), "{error}"); + assert_eq!(fs::read(&path)?, before); + assert_eq!(fs::read_dir(temp.path())?.count(), 2, "only the store and its lock remain"); + + // The rename fails when the store path is a directory. + let blocked = temp.path().join("blocked.cc"); + fs::create_dir(&blocked)?; + let mut blocked_store = SingleFileMessagePackBackingStore::new(&blocked); + let error = blocked_store + .push_all(&[snapshot_envelope("alpha", b"1")], PushAllOptions::default()) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::NotReplaced), "{error:#}"); + assert!(fs::metadata(&blocked)?.is_dir()); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn a_write_whose_directory_sync_fails_is_replaced_but_not_durable() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, _) = written_store(&path)?; + let replacement = vec![snapshot_envelope("delta", b"4")]; + let error = with_write_fault(WriteStep::DirectorySync, || { + store.compare_exchange_snapshot(&store.pull_all().unwrap(), &replacement) + }) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::ReplacedNotDurable), "{error:#}"); + assert!(error.to_string().contains(&format!("{} was replaced, but the replacement is not yet durable", path.display())), "{error}"); + assert!(io_error_in(&error).is_some(), "{error:#}"); + assert_eq!(store.pull_all()?, replacement); + Ok(()) + } + + #[test] + fn a_write_failure_names_no_value_from_the_snapshot() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let store = SingleFileMessagePackBackingStore::new(&path); + let error = with_write_fault(WriteStep::StagingSync, || { + store.compare_exchange_snapshot(&[], &[snapshot_envelope("SECRET-KEY", b"SECRET-PAYLOAD")]) + }) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::NotReplaced), "{error:#}"); + assert!(!format!("{error:#}").contains("SECRET"), "{error:#}"); + Ok(()) + } + + #[test] + fn a_write_that_landed_is_reported_as_landed_when_the_unlock_after_it_fails() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (mut store, _) = written_store(&path)?; + let replacement = vec![snapshot_envelope("delta", b"4")]; + let exchanged = with_write_fault(WriteStep::Unlock, || { + store.compare_exchange_snapshot(&store.pull_all().unwrap(), &replacement) + })?; + assert!(exchanged); + assert_eq!(store.pull_all()?, replacement); + + let pushed = snapshot_envelope("epsilon", b"5"); + with_write_fault(WriteStep::Unlock, || store.push(&pushed))?; + assert!(with_write_fault(WriteStep::Unlock, || store.pull_all())?.contains(&pushed)); + + // The lock was released: another writer takes it at once. + assert!(matches!( + store.try_compare_exchange_snapshot(&store.pull_all()?, &replacement)?, + TryCompareExchangeSnapshotOutcome::Exchanged + )); + + // The keyed store releases its lock the same way. + let mut keyed = RedbMessagePackBackingStore::new(temp.path().join("keyed.redb"))?; + with_write_fault(WriteStep::Unlock, || keyed.push(&pushed))?; + assert_eq!(keyed.pull_all()?, vec![pushed]); + Ok(()) + } + + // What a write does through a symbolic link today. R3 decides whether a write + // resolves the link or refuses it; until then these pin the current behaviour so the + // change is deliberate. Reads refuse a dangling link (see the test above). + #[cfg(unix)] + #[test] + fn r3_decides_resolve_or_refuse_a_push_through_a_live_link_replaces_the_link() -> Result<()> { + let temp = tempfile::tempdir()?; + let volume = temp.path().join("volume"); + fs::create_dir(&volume)?; + let target = volume.join("store.cc"); + SingleFileMessagePackBackingStore::new(&target).push(&snapshot_envelope("a", b"1"))?; + let before = fs::read(&target)?; + let link = temp.path().join("store.cc"); + std::os::unix::fs::symlink(&target, &link)?; + + SingleFileMessagePackBackingStore::new(&link).push(&snapshot_envelope("b", b"2"))?; + assert!(fs::symlink_metadata(&link)?.file_type().is_file(), "the link was replaced by a regular file"); + assert_eq!(fs::read(&target)?, before, "the store the link pointed to is unchanged"); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn r3_decides_resolve_or_refuse_a_push_all_through_a_dangling_link_replaces_the_link() -> Result<()> { + let temp = tempfile::tempdir()?; + let volume = temp.path().join("volume"); + fs::create_dir(&volume)?; + let link = temp.path().join("store.cc"); + std::os::unix::fs::symlink(volume.join("store.cc"), &link)?; + + SingleFileMessagePackBackingStore::new(&link).push_all(&[snapshot_envelope("a", b"1")], PushAllOptions::default())?; + assert!(fs::symlink_metadata(&link)?.file_type().is_file(), "the link was replaced by a regular file"); + assert_eq!(fs::read_dir(&volume)?.count(), 0, "nothing reached the link's volume"); + Ok(()) + } + + #[test] + fn a_keyed_store_refusal_never_echoes_a_stored_value() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("keyed.redb"); + let envelope = snapshot_envelope("k", b"1"); + RedbMessagePackBackingStore::new(&path)?.push(&envelope)?; + { + let database = redb::Database::create(&path)?; + let write = database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + let keys: Vec> = table + .iter()? + .map(|row| row.map(|(key, _)| key.value().to_vec())) + .collect::>()?; + let mut value = vec![0xab]; + value.extend_from_slice(b"SECRET-REDB"); + for key in keys { + table.insert(key.as_slice(), value.as_slice())?; + } + } + write.commit()?; + } + let store = RedbMessagePackBackingStore::new(&path)?; + for error in [store.pull_all().unwrap_err(), store.compare_and_swap_entry(&envelope, snapshot_envelope("k", b"2")).unwrap_err()] { + assert!(format!("{error:#}").contains("failed to decode redb CultCache envelope"), "{error:#}"); + assert!(!format!("{error:#}").contains("SECRET"), "{error:#}"); + } + Ok(()) + } + + #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] + enum Choice { + Only, + } + + #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] + struct SkippedChoice { + #[serde(default, skip_serializing_if = "Option::is_none")] + choice: Option, + text: String, + } + + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] + #[cultcache(type = "skipped-choice")] + struct ChoiceNote { + #[cultcache(key = 0)] + value: SkippedChoice, + } + + #[test] + fn a_document_decode_refusal_names_the_entry_and_never_a_stored_value() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + // A compact SkippedChoice with no choice is [text], so decoding it reads the text + // where the enum goes, and serde's own message would quote it as an unknown variant. + let secret = ChoiceNote { + value: SkippedChoice { + choice: None, + text: "SECRET-CHOICE".into(), + }, + }; + let mut payload = vec![0x91, 0x91, 0xad]; + payload.extend_from_slice(b"SECRET-CHOICE"); + let malformed = CultCacheEnvelope { + key: "k".into(), + r#type: ::TYPE.into(), + payload, + stored_at: "2026-09-30T00:00:00Z".into(), + schema_id: None, + }; + SingleFileMessagePackBackingStore::new(&path).push(&malformed)?; + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + + let refusals = [ + ("get", cache.get::("k").unwrap_err()), + ("get_all", cache.get_all::().unwrap_err()), + ("get_all_with_keys", cache.get_all_with_keys::().unwrap_err()), + ("put_envelope", cache.put_envelope::(malformed.clone()).unwrap_err()), + ("prepare_entry", cache.prepare_entry("compact", &secret).unwrap_err()), + ]; + for (path, error) in refusals { + let text = format!("{error:#}"); + assert!(!text.contains("SECRET"), "{path}: {text}"); + assert!(text.contains("CultCache"), "{path}: {text}"); + } + Ok(()) + } + + #[test] + fn a_snapshot_the_store_cannot_hold_is_rejected_before_any_write() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let (store, before) = written_store(&path)?; + let mut first = snapshot_envelope("x", b"1"); + first.schema_id = Some("shared".into()); + let mut second = snapshot_envelope("y", b"2"); + second.r#type = "other".into(); + second.schema_id = Some("shared".into()); + let error = store + .compare_exchange_snapshot(&store.pull_all()?, &[first, second]) + .unwrap_err(); + assert_eq!(write_failure_kind(&error), Some(CultCacheStoreWriteFailedKind::Rejected), "{error:#}"); + assert!(error.to_string().contains("refused a snapshot it cannot store"), "{error}"); + assert!(io_error_in(&error).is_none(), "{error:#}"); + assert_eq!(fs::read(&path)?, before); + Ok(()) + } } 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 b7d9684c..b7df1454 100644 --- a/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts +++ b/packages/cultcache-ts/src/single-file-messagepack-backing-store.ts @@ -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 { - 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 { @@ -124,16 +117,8 @@ export class SingleFileMessagePackBackingStore implements CacheBackingStore { async pushAll(entries: CultCacheEnvelope[], options: PushAllOptions = {}): Promise { 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 { + 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 { let delayMs = 10; let lastError: unknown; diff --git a/packages/cultcache-ts/src/store-format.ts b/packages/cultcache-ts/src/store-format.ts index 02de60e7..4f579505 100644 --- a/packages/cultcache-ts/src/store-format.ts +++ b/packages/cultcache-ts/src/store-format.ts @@ -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`, + * 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.`, ); } } diff --git a/packages/cultcache-ts/test/cult-cache.test.ts b/packages/cultcache-ts/test/cult-cache.test.ts index 8753af47..b7c51cc1 100644 --- a/packages/cultcache-ts/test/cult-cache.test.ts +++ b/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, 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); diff --git a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs index f591d764..51193806 100644 --- a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs +++ b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs @@ -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, 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); diff --git a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs index 181eca56..e9d8797d 100644 --- a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs +++ b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs @@ -167,6 +167,61 @@ namespace GameCult.Caching.Tests Throws.TypeOf().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()); + 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(), 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; 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(() => 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() {