From bdae8b831a6226a7639c52620965e900614a12bb Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 00:33:31 +0200 Subject: [PATCH 01/13] cultnet-rs: retire the dynamic document lane; every received byte decodes once through the door Deletes CultNetDocumentRecord, the DocumentPut and SnapshotResponse variants, parse_cultnet_message, require_typed, round_trip_typed, the eight replication functions and the two catalog schemas (parked at tag parked/cultnet-rs-dynamic-document-lane). decode_cultnet_message_from_slice is the one decode; raw v1 rows, identity credentials and the signed-record readers decode through cultcache_rs::decode_payload. Adds tests/decode_door.rs. 0.2.0. Not yet compiled at this commit (verification runs on Yggdrasil). Co-Authored-By: Claude Opus 5.5 --- packages/cultmesh-rs/Cargo.lock | 2 +- packages/cultnet-rs/Cargo.toml | 2 +- packages/cultnet-rs/README.md | 2 +- .../cultnet/cultnet.document-put.schema.json | 19 - .../cultnet.snapshot-response.schema.json | 22 -- packages/cultnet-rs/src/contracts.rs | 200 +++------- .../cultnet-rs/src/idunn_lifecycle_brake.rs | 2 +- .../src/provider_health_contracts.rs | 4 +- packages/cultnet-rs/src/replication.rs | 194 ---------- .../src/runtime_authority_contracts.rs | 10 +- packages/cultnet-rs/src/schema_discovery.rs | 14 - packages/cultnet-rs/src/service_identity.rs | 7 +- packages/cultnet-rs/tests/cultnet.rs | 30 +- packages/cultnet-rs/tests/decode_door.rs | 349 ++++++++++++++++++ 14 files changed, 425 insertions(+), 432 deletions(-) delete mode 100644 packages/cultnet-rs/contracts/cultnet/cultnet.document-put.schema.json delete mode 100644 packages/cultnet-rs/contracts/cultnet/cultnet.snapshot-response.schema.json create mode 100644 packages/cultnet-rs/tests/decode_door.rs diff --git a/packages/cultmesh-rs/Cargo.lock b/packages/cultmesh-rs/Cargo.lock index 945bc65b..42893bb7 100644 --- a/packages/cultmesh-rs/Cargo.lock +++ b/packages/cultmesh-rs/Cargo.lock @@ -213,7 +213,7 @@ dependencies = [ [[package]] name = "cultnet-rs" -version = "0.1.0" +version = "0.2.0" dependencies = [ "aes-gcm", "anyhow", diff --git a/packages/cultnet-rs/Cargo.toml b/packages/cultnet-rs/Cargo.toml index f87af6cb..13d87867 100644 --- a/packages/cultnet-rs/Cargo.toml +++ b/packages/cultnet-rs/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cultnet-rs" -version = "0.1.0" +version = "0.2.0" edition = "2024" publish = false license = "MIT" diff --git a/packages/cultnet-rs/README.md b/packages/cultnet-rs/README.md index 7ade81dc..8bea2899 100644 --- a/packages/cultnet-rs/README.md +++ b/packages/cultnet-rs/README.md @@ -164,7 +164,7 @@ cadence and flushes changed state back through the typed cache. `update(...)`, wants a more explicit Rust shape. When a canonical document put arrives while local state is dirty, -`apply_document_put_message(...)` or `apply_raw_document_put_message(...)` +`apply_raw_document_put_message(...)` records a `CultNetReactiveDocumentReconciliation`: the canonical value, the predicted local value, and a delta that can be smoothed by the caller. CultCache payloads serialize by numeric field slot, so Rust tuple-backed deltas use slot diff --git a/packages/cultnet-rs/contracts/cultnet/cultnet.document-put.schema.json b/packages/cultnet-rs/contracts/cultnet/cultnet.document-put.schema.json deleted file mode 100644 index 2b92051a..00000000 --- a/packages/cultnet-rs/contracts/cultnet/cultnet.document-put.schema.json +++ /dev/null @@ -1,19 +0,0 @@ -{ - "$schema": "https://json-schema.org/draft/2020-12/schema", - "$id": "https://github.com/GameCult/CultLib/contracts/cultnet/cultnet.document-put.schema.json", - "title": "CultNet Document Put Message", - "type": "object", - "required": [ - "schemaVersion", - "messageId", - "document" - ], - "additionalProperties": false, - "properties": { - "schemaVersion": { "const": "cultnet.document_put.v0" }, - "messageId": { "type": "string", "minLength": 1 }, - "document": { - "$ref": "./cultnet.document-record.schema.json" - } - } -} diff --git a/packages/cultnet-rs/contracts/cultnet/cultnet.snapshot-response.schema.json b/packages/cultnet-rs/contracts/cultnet/cultnet.snapshot-response.schema.json deleted file mode 100644 index 87daf147..00000000 --- a/packages/cultnet-rs/contracts/cultnet/cultnet.snapshot-response.schema.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "$schema": "https://json-schema.org/draft/2020-12/schema", - "$id": "https://github.com/GameCult/CultLib/contracts/cultnet/cultnet.snapshot-response.schema.json", - "title": "CultNet Snapshot Response Message", - "type": "object", - "required": [ - "schemaVersion", - "messageId", - "documents" - ], - "additionalProperties": false, - "properties": { - "schemaVersion": { "const": "cultnet.snapshot_response.v0" }, - "messageId": { "type": "string", "minLength": 1 }, - "documents": { - "type": "array", - "items": { - "$ref": "./cultnet.document-record.schema.json" - } - } - } -} diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index 6d9a2d77..fa42131a 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -4,7 +4,6 @@ use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use serde::Deserialize; use serde::Serialize; -use serde_json::Value; use crate::{Edge, RawDocumentHeader, Selection, SelectionDocumentRecord}; @@ -30,23 +29,6 @@ pub enum CultNetRawPayloadEncoding { Messagepack, } -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct CultNetDocumentRecord { - pub schema_id: String, - pub record_key: String, - pub stored_at: String, - pub payload: TPayload, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub source_runtime_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub source_agent_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub source_role: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub tags: Option>, -} - #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub enum CultNetDocumentOperation { @@ -357,7 +339,7 @@ pub enum CultNetMessage { // `cultnet.schema.v0` path (parse_raw_cultnet_schema_message/encode_raw_cultnet_schema_message) // like the other messages that need exact wire control, not through this derive - the derive // attributes below exist only so the enum as a whole still compiles and round-trips through the - // generic `serde_json::Value` path other tooling may still reach for. + // generic variants, which decode through the door from the message bytes. #[serde(rename = "cultnet.error.v0", rename_all = "camelCase")] Error { error: String, @@ -370,11 +352,6 @@ pub enum CultNetMessage { SampleChangeName { name: String }, #[serde(rename = "cultnet.sample.chat.v0", rename_all = "camelCase")] SampleChat { text: String }, - #[serde(rename = "cultnet.document_put.v0", rename_all = "camelCase")] - DocumentPut { - message_id: String, - document: CultNetDocumentRecord, - }, #[serde(rename = "cultnet.document_delete.v0", rename_all = "camelCase")] DocumentDelete { message_id: String, @@ -394,11 +371,6 @@ pub enum CultNetMessage { #[serde(default, skip_serializing_if = "Option::is_none")] record_keys: Option>, }, - #[serde(rename = "cultnet.snapshot_response.v0", rename_all = "camelCase")] - SnapshotResponse { - message_id: String, - documents: Vec>, - }, #[serde(rename = "cultnet.snapshot_response_raw.v0", rename_all = "camelCase")] SnapshotResponseRaw { message_id: String, @@ -544,33 +516,6 @@ fn true_default() -> bool { true } -pub fn parse_cultnet_message( - wire_value: &rmpv::Value, - contract: CultNetWireContract, -) -> Result { - match contract { - CultNetWireContract::CultNetSchemaV0 => { - let message = match schema_version_from_wire_value(wire_value)? { - Some( - "cultnet.document_put_raw.v0" - | "cultnet.snapshot_response_raw.v0" - | "cultnet.snapshot_response_raw.v1" - | "cultmesh.content_chunk_request.v1" - | "cultmesh.content_chunk_response.v1" - | "cultnet.error.v0", - ) => parse_raw_cultnet_schema_message(wire_value)?, - _ => { - let json_value: Value = rmp_serde::from_slice(&rmp_serde::to_vec(wire_value)?)?; - serde_json::from_value(json_value)? - } - }; - validate_message(&message)?; - Ok(message) - } - CultNetWireContract::GameCultNetworkingV0 => parse_gamecult_networking_message(wire_value), - } -} - pub fn encode_cultnet_message_for_wire( message: &CultNetMessage, contract: CultNetWireContract, @@ -584,7 +529,7 @@ pub fn encode_cultnet_message_for_wire( | CultNetMessage::ContentChunkRequest { .. } | CultNetMessage::ContentChunkResponse { .. } | CultNetMessage::Error { .. } => encode_raw_cultnet_schema_message(message), - _ => Ok(rmp_serde::from_slice(&rmp_serde::to_vec( + _ => Ok(cultcache_rs::decode_payload(&rmp_serde::to_vec( &serde_json::to_value(message)?, )?)?), }, @@ -605,7 +550,24 @@ pub fn decode_cultnet_message_from_slice( contract: CultNetWireContract, ) -> Result { let wire_value: rmpv::Value = cultcache_rs::decode_payload(bytes)?; - parse_cultnet_message(&wire_value, contract) + match contract { + CultNetWireContract::CultNetSchemaV0 => { + let message = match schema_version_from_wire_value(&wire_value)? { + Some( + "cultnet.document_put_raw.v0" + | "cultnet.snapshot_response_raw.v0" + | "cultnet.snapshot_response_raw.v1" + | "cultmesh.content_chunk_request.v1" + | "cultmesh.content_chunk_response.v1" + | "cultnet.error.v0", + ) => parse_raw_cultnet_schema_message(bytes, &wire_value)?, + _ => cultcache_rs::decode_payload::(bytes)?, + }; + validate_message(&message)?; + Ok(message) + } + CultNetWireContract::GameCultNetworkingV0 => parse_gamecult_networking_message(&wire_value), + } } fn validate_message(message: &CultNetMessage) -> Result<()> { @@ -669,13 +631,6 @@ fn validate_message(message: &CultNetMessage) -> Result<()> { CultNetMessage::Error { error, code: _, details: _ } => require_non_empty(error, "error")?, CultNetMessage::SampleChangeName { name } => require_non_empty(name, "name")?, CultNetMessage::SampleChat { text } => require_non_empty(text, "text")?, - CultNetMessage::DocumentPut { - message_id, - document, - } => { - require_non_empty(message_id, "messageId")?; - validate_document_record(document)?; - } CultNetMessage::DocumentDelete { message_id, schema_id, @@ -701,15 +656,6 @@ fn validate_message(message: &CultNetMessage) -> Result<()> { require_optional_string_vec(schema_ids.as_deref(), "schemaIds")?; require_optional_string_vec(record_keys.as_deref(), "recordKeys")?; } - CultNetMessage::SnapshotResponse { - message_id, - documents, - } => { - require_non_empty(message_id, "messageId")?; - for document in documents { - validate_document_record(document)?; - } - } CultNetMessage::SnapshotResponseRaw { message_id, documents, @@ -874,17 +820,6 @@ fn validate_mutation_contract(contract: &CultNetDocumentMutationContract) -> Res Ok(()) } -fn validate_document_record(document: &CultNetDocumentRecord) -> Result<()> { - require_non_empty(&document.schema_id, "schemaId")?; - require_non_empty(&document.record_key, "recordKey")?; - require_non_empty(&document.stored_at, "storedAt")?; - require_optional_non_empty(document.source_runtime_id.as_deref(), "sourceRuntimeId")?; - require_optional_non_empty(document.source_agent_id.as_deref(), "sourceAgentId")?; - require_optional_non_empty(document.source_role.as_deref(), "sourceRole")?; - require_optional_string_vec(document.tags.as_deref(), "tags")?; - Ok(()) -} - fn validate_raw_document_record(document: &CultNetRawDocumentRecord) -> Result<()> { require_non_empty(&document.schema_id, "schemaId")?; require_non_empty(&document.record_key, "recordKey")?; @@ -976,7 +911,21 @@ fn schema_version_from_wire_value(input: &rmpv::Value) -> Result> { })) } -fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result { +/// The row arrays of `cultnet.snapshot_response_raw.v1`, decoded from the original bytes so each +/// row is typed in one pass through the door; nil and absent both read as `None`, other keys are +/// ignored. +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct SnapshotResponseRawV1Rows { + #[serde(default)] + headers: Option>, + #[serde(default)] + documents: Option>, + #[serde(default)] + edges: Option>, +} + +fn parse_raw_cultnet_schema_message(bytes: &[u8], input: &rmpv::Value) -> Result { let object = input .as_map() .ok_or_else(|| anyhow!("cultnet.schema.v0 raw messages must be objects"))?; @@ -1012,51 +961,8 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result { - let headers = get("headers") - .filter(|value| !value.is_nil()) - .map(|value| { - value - .as_array() - .ok_or_else(|| anyhow!("headers must be an array"))? - .iter() - .enumerate() - .map(|(index, value)| { - require_typed::(Some(value), &format!("headers[{index}]")) - }) - .collect::>>() - }) - .transpose()?; - let documents = get("documents") - .filter(|value| !value.is_nil()) - .map(|value| { - value - .as_array() - .ok_or_else(|| anyhow!("documents must be an array"))? - .iter() - .enumerate() - .map(|(index, value)| { - require_typed::( - Some(value), - &format!("documents[{index}]"), - ) - }) - .collect::>>() - }) - .transpose()?; - let edges = get("edges") - .filter(|value| !value.is_nil()) - .map(|value| { - value - .as_array() - .ok_or_else(|| anyhow!("edges must be an array"))? - .iter() - .enumerate() - .map(|(index, value)| { - require_typed::(Some(value), &format!("edges[{index}]")) - }) - .collect::>>() - }) - .transpose()?; + let SnapshotResponseRawV1Rows { headers, documents, edges } = + cultcache_rs::decode_payload(bytes).map_err(|err| anyhow!("rows: {err}"))?; Ok(CultNetMessage::SnapshotResponseRawV1 { message_id: require_legacy_string(get("messageId"), "messageId")?, @@ -1469,34 +1375,16 @@ fn encode_raw_document_record(document: &CultNetRawDocumentRecord) -> rmpv::Valu ]) } -// R-M: `SelectionDocumentRecord`, `RawDocumentHeader`, `Edge` and `RecordRef` already derive -// `Serialize`/`Deserialize` (selection.rs) with the same camelCase field names this hand-written -// codec used to spell out by hand - about 335 lines duplicating what those derives already do. -// -// `rmpv::ext::to_value`/`from_value` looked like the natural generic bridge, but they serialize a -// struct positionally (`serialize_struct` as a MessagePack array, one slot per field in -// declaration order, breaking as soon as a `skip_serializing_if` field is actually skipped) - -// wrong for these types twice over: it is not the named-map encoding this wire always uses -// (section 10's rule; `rmp_serde::to_vec_named` elsewhere in this crate), and it does not -// round-trip at all once a field is omitted. These two helpers instead go through -// `rmp_serde::to_vec_named`/`from_slice` (named-map bytes, proven both ways by -// `tests/selection.rs`'s `snapshot_response_raw_v1_round_trips_*` tests) and hand the bytes to -// `rmpv::Value`'s own `Serialize`/`Deserialize` (already used throughout this file, e.g. to decode -// the outer message envelope) to shuttle to/from the `rmpv::Value` tree the rest of this codec -// builds by hand. - -fn require_typed( - value: Option<&rmpv::Value>, - field_name: &str, -) -> Result { - let value = value.ok_or_else(|| anyhow!("{field_name} is required"))?; - let bytes = rmp_serde::to_vec(value).map_err(|err| anyhow!("{field_name}: {err}"))?; - rmp_serde::from_slice(&bytes).map_err(|err| anyhow!("{field_name}: {err}")) -} +// R-M: `SelectionDocumentRecord`, `RawDocumentHeader`, `Edge` and `RecordRef` derive +// `Serialize`/`Deserialize` (selection.rs) with the camelCase field names this wire uses. +// `encode_typed` goes through `rmp_serde::to_vec_named` (the named-map encoding this wire always +// uses; `rmpv::ext::to_value` would write a struct positionally) and reads the bytes back into the +// `rmpv::Value` tree the rest of this codec builds by hand, through the decode door. Received rows +// are never re-encoded to become typed: `SnapshotResponseRawV1Rows` reads them from the original bytes. fn encode_typed(value: &T) -> rmpv::Value { let bytes = rmp_serde::to_vec_named(value).expect("selection wire types always encode to MessagePack"); - rmp_serde::from_slice(&bytes).expect("named-map MessagePack bytes always decode into rmpv::Value") + cultcache_rs::decode_payload(&bytes).expect("named-map MessagePack bytes always decode into rmpv::Value") } fn require_legacy_u32(value: Option<&rmpv::Value>, field_name: &str) -> Result { diff --git a/packages/cultnet-rs/src/idunn_lifecycle_brake.rs b/packages/cultnet-rs/src/idunn_lifecycle_brake.rs index 86eefa64..cd28cf2f 100644 --- a/packages/cultnet-rs/src/idunn_lifecycle_brake.rs +++ b/packages/cultnet-rs/src/idunn_lifecycle_brake.rs @@ -68,7 +68,7 @@ impl IdunnLifecycleBrakeRecord { if messagepack_array_len(payload) != Some(9) { bail!("lifecycle brake is not the 9-field positional contract"); } - let value: Self = rmp_serde::from_slice(payload)?; + let value: Self = cultcache_rs::decode_payload(payload)?; value.validate()?; if rmp_serde::to_vec(&value)? != payload { bail!("lifecycle brake encoding is not canonical"); diff --git a/packages/cultnet-rs/src/provider_health_contracts.rs b/packages/cultnet-rs/src/provider_health_contracts.rs index 632c74f2..f4ca108c 100644 --- a/packages/cultnet-rs/src/provider_health_contracts.rs +++ b/packages/cultnet-rs/src/provider_health_contracts.rs @@ -123,7 +123,7 @@ impl IdunnSignedDaemonHealthRecord { pub fn decode_canonical_signed_payload(payload: &[u8]) -> Result<(Self, Vec)> { match messagepack_array_len(payload) { Some(18) => { - let statement: Self = rmp_serde::from_slice(payload) + let statement: Self = cultcache_rs::decode_payload(payload) .context("decoding generic signed daemon health")?; if rmp_serde::to_vec(&statement)? != payload { bail!("signed daemon health payload is not canonical positional MessagePack"); @@ -133,7 +133,7 @@ impl IdunnSignedDaemonHealthRecord { Ok((statement, rmp_serde::to_vec(&unsigned)?)) } Some(17) => { - let legacy: LegacyIdunnSignedDaemonHealthWire = rmp_serde::from_slice(payload) + let legacy: LegacyIdunnSignedDaemonHealthWire = cultcache_rs::decode_payload(payload) .context("decoding legacy generic signed daemon health")?; if rmp_serde::to_vec(&legacy)? != payload { bail!("signed daemon health payload is not canonical positional MessagePack"); diff --git a/packages/cultnet-rs/src/replication.rs b/packages/cultnet-rs/src/replication.rs index 7b070fbd..85f9574d 100644 --- a/packages/cultnet-rs/src/replication.rs +++ b/packages/cultnet-rs/src/replication.rs @@ -18,7 +18,6 @@ use std::thread::JoinHandle; use std::time::Duration; use crate::CultNetDocumentMutationContract; -use crate::CultNetDocumentRecord; use crate::CultNetMessage; use crate::CultNetRawDocumentRecord; use crate::CultNetRawPayloadEncoding; @@ -152,33 +151,6 @@ impl CultNetDocumentRegistry { .collect() } - pub fn create_document_put_message( - &self, - message_id: impl Into, - record_key: impl Into, - value: &T, - options: CultNetDocumentPutOptions, - ) -> Result - where - T: DatabaseEntry + Serialize + DeserializeOwned, - { - let binding = self.require_binding(T::TYPE)?; - let payload = serde_json::to_value(round_trip_typed(value)?)?; - Ok(CultNetMessage::DocumentPut { - message_id: message_id.into(), - document: CultNetDocumentRecord { - schema_id: binding.schema_id.clone(), - record_key: record_key.into(), - stored_at: options.stored_at.unwrap_or_else(now_utc_second), - payload, - source_runtime_id: options.source_runtime_id, - source_agent_id: options.source_agent_id, - source_role: options.source_role, - tags: options.tags, - }, - }) - } - pub fn create_document_delete_message( &self, message_id: impl Into, @@ -230,38 +202,6 @@ impl CultNetDocumentRegistry { }) } - pub fn create_snapshot_response( - &self, - cache: &CultCache, - message_id: impl Into, - schema_ids: Option<&[String]>, - record_keys: Option<&[String]>, - ) -> Result { - let requested_schema_ids = schema_ids.map(|items| items.iter().collect::>()); - let requested_record_keys = record_keys.map(|items| items.iter().collect::>()); - let mut documents = Vec::new(); - for envelope in cache.snapshot() { - let binding = self.require_binding(&envelope.r#type)?; - if requested_schema_ids - .as_ref() - .is_some_and(|ids| !ids.contains(&binding.schema_id)) - { - continue; - } - if requested_record_keys - .as_ref() - .is_some_and(|keys| !keys.contains(&envelope.key)) - { - continue; - } - documents.push(self.document_record_from_envelope(&envelope)?); - } - Ok(CultNetMessage::SnapshotResponse { - message_id: message_id.into(), - documents, - }) - } - pub fn create_raw_snapshot_response( &self, cache: &CultCache, @@ -294,36 +234,6 @@ impl CultNetDocumentRegistry { }) } - pub fn apply_document_put_message( - &self, - cache: &mut CultCache, - message: &CultNetMessage, - ) -> Result - where - T: DatabaseEntry + Serialize + DeserializeOwned, - { - let CultNetMessage::DocumentPut { document, .. } = message else { - return Err(anyhow!("expected cultnet.document_put.v0")); - }; - let binding = self.require_binding(T::TYPE)?; - if document.schema_id != binding.schema_id { - return Err(anyhow!( - "schema id {:?} does not match registered Rust type {:?} schema {:?}", - document.schema_id, - T::TYPE, - binding.schema_id - )); - } - let value: T = serde_json::from_value(document.payload.clone()).with_context(|| { - format!( - "failed to decode CultNet payload schema {:?} as {}", - binding.schema_id, - T::SCHEMA_NAME - ) - })?; - cache.put(&document.record_key, &value) - } - pub fn apply_document_delete_message( &self, cache: &mut CultCache, @@ -381,68 +291,6 @@ impl CultNetDocumentRegistry { }) } - pub fn apply_snapshot_response( - &self, - cache: &mut CultCache, - response: &CultNetMessage, - ) -> Result> - where - T: DatabaseEntry + Serialize + DeserializeOwned, - { - let CultNetMessage::SnapshotResponse { documents, .. } = response else { - return Err(anyhow!("expected cultnet.snapshot_response.v0")); - }; - let mut applied = Vec::new(); - let binding = self.require_binding(T::TYPE)?; - for document in documents { - if document.schema_id != binding.schema_id { - continue; - } - applied.push(self.apply_document_put_message::( - cache, - &CultNetMessage::DocumentPut { - message_id: "snapshot-apply".to_string(), - document: document.clone(), - }, - )?); - } - Ok(applied) - } - - pub fn sync_document_from_snapshot_response( - &self, - cache: &mut CultCache, - response: &CultNetMessage, - record_key: &str, - ) -> Result - where - T: DatabaseEntry + Serialize + DeserializeOwned, - { - let CultNetMessage::SnapshotResponse { documents, .. } = response else { - return Err(anyhow!("expected cultnet.snapshot_response.v0")); - }; - let binding = self.require_binding(T::TYPE)?; - let document = documents - .iter() - .find(|document| { - document.schema_id == binding.schema_id && document.record_key == record_key - }) - .ok_or_else(|| { - anyhow!( - "No CultNet snapshot document for schema {:?} and record key {:?}", - binding.schema_id, - record_key - ) - })?; - self.apply_document_put_message::( - cache, - &CultNetMessage::DocumentPut { - message_id: "snapshot-sync".to_string(), - document: document.clone(), - }, - ) - } - pub fn apply_raw_snapshot_response( &self, cache: &mut CultCache, @@ -517,29 +365,6 @@ impl CultNetDocumentRegistry { CultNetReactiveDocument::new(self.clone(), cache, record_key, options) } - fn document_record_from_envelope( - &self, - envelope: &CultCacheEnvelope, - ) -> Result> { - let binding = self.require_binding(&envelope.r#type)?; - let payload: Value = cultcache_rs::decode_payload(&envelope.payload).with_context(|| { - format!( - "failed to decode CultCache envelope {:?} at {:?} as generic CultNet payload", - envelope.r#type, envelope.key - ) - })?; - Ok(CultNetDocumentRecord { - schema_id: binding.schema_id.clone(), - record_key: envelope.key.clone(), - stored_at: envelope.stored_at.clone(), - payload, - source_runtime_id: None, - source_agent_id: None, - source_role: None, - tags: None, - }) - } - pub fn raw_document_record_from_envelope( &self, envelope: &CultCacheEnvelope, @@ -732,18 +557,6 @@ where flush_reactive_document::(&self.cache, &self.current, &self.state, &self.record_key) } - pub fn apply_document_put_message(&self, message: &CultNetMessage) -> Result { - let canonical = { - let mut cache = self - .cache - .lock() - .map_err(|_| anyhow!("CultCache mutex poisoned"))?; - self.registry - .apply_document_put_message::(&mut cache, message)? - }; - self.apply_canonical_value(canonical) - } - pub fn apply_raw_document_put_message(&self, message: &CultNetMessage) -> Result { let canonical = { let mut cache = self @@ -1002,13 +815,6 @@ fn numeric_delta(predicted: &Value, canonical: &Value) -> Option { serde_json::Number::from_f64(predicted - canonical).map(Value::Number) } -fn round_trip_typed(value: &T) -> Result -where - T: Serialize + DeserializeOwned, -{ - Ok(rmp_serde::from_slice(&rmp_serde::to_vec(value)?)?) -} - fn now_utc_second() -> String { chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true) } diff --git a/packages/cultnet-rs/src/runtime_authority_contracts.rs b/packages/cultnet-rs/src/runtime_authority_contracts.rs index 426be1fb..0549a5c1 100644 --- a/packages/cultnet-rs/src/runtime_authority_contracts.rs +++ b/packages/cultnet-rs/src/runtime_authority_contracts.rs @@ -390,7 +390,7 @@ impl IdunnExpectedIncarnationRecord { bail!("expected incarnation is not the 18-field positional contract"); } let expected: Self = - rmp_serde::from_slice(payload).context("decoding expected incarnation")?; + cultcache_rs::decode_payload(payload).context("decoding expected incarnation")?; if rmp_serde::to_vec(&expected)? != payload { bail!("expected incarnation is not canonical positional MessagePack"); } @@ -629,7 +629,7 @@ impl OdinRuntimeTopologyCorrelationRecord { bail!("topology correlation is not the 21-field positional contract"); } let receipt: Self = - rmp_serde::from_slice(payload).context("decoding runtime topology correlation")?; + cultcache_rs::decode_payload(payload).context("decoding runtime topology correlation")?; if rmp_serde::to_vec(&receipt)? != payload { bail!("topology correlation is not canonical positional MessagePack"); } @@ -1173,7 +1173,7 @@ pub fn authenticate_runtime_presence_claim( bail!("runtime presence payload is not the 24-field positional contract"); } let presence: GameCultRuntimePresenceHealthRecord = - rmp_serde::from_slice(canonical_presence).context("decoding runtime presence health")?; + cultcache_rs::decode_payload(canonical_presence).context("decoding runtime presence health")?; if rmp_serde::to_vec(&presence)? != canonical_presence { bail!("runtime presence payload is not canonical positional MessagePack"); } @@ -1470,7 +1470,7 @@ impl IdunnRuntimeActivationRecord { bail!("runtime activation is not the 10-field positional contract"); } let activation: Self = - rmp_serde::from_slice(payload).context("decoding runtime activation")?; + cultcache_rs::decode_payload(payload).context("decoding runtime activation")?; if rmp_serde::to_vec(&activation)? != payload { bail!("runtime activation is not canonical positional MessagePack"); } @@ -1604,7 +1604,7 @@ impl IdunnProcessWriteLeaseRecord { if messagepack_array_len(payload) != Some(14) { bail!("process write lease is not the 14-field positional contract"); } - let lease: Self = rmp_serde::from_slice(payload).context("decoding process write lease")?; + let lease: Self = cultcache_rs::decode_payload(payload).context("decoding process write lease")?; if rmp_serde::to_vec(&lease)? != payload { bail!("process write lease is not canonical positional MessagePack"); } diff --git a/packages/cultnet-rs/src/schema_discovery.rs b/packages/cultnet-rs/src/schema_discovery.rs index 43211c3c..e061d5f9 100644 --- a/packages/cultnet-rs/src/schema_discovery.rs +++ b/packages/cultnet-rs/src/schema_discovery.rs @@ -284,13 +284,6 @@ pub fn builtin_schema_registry() -> Result { Some("cultnet.sample.chat.v0"), None, )?, - schema_registration( - include_str!("../contracts/cultnet/cultnet.document-put.schema.json"), - CultNetSchemaKind::WireMessage, - vec![CultNetWireContract::CultNetSchemaV0], - Some("cultnet.document_put.v0"), - None, - )?, schema_registration( include_str!("../contracts/cultnet/cultnet.document-delete.schema.json"), CultNetSchemaKind::WireMessage, @@ -319,13 +312,6 @@ pub fn builtin_schema_registry() -> Result { Some("cultnet.snapshot_request.v0"), None, )?, - schema_registration( - include_str!("../contracts/cultnet/cultnet.snapshot-response.schema.json"), - CultNetSchemaKind::WireMessage, - vec![CultNetWireContract::CultNetSchemaV0], - Some("cultnet.snapshot_response.v0"), - None, - )?, schema_registration( include_str!("../contracts/cultnet/cultnet.snapshot-response-raw.schema.json"), CultNetSchemaKind::WireMessage, diff --git a/packages/cultnet-rs/src/service_identity.rs b/packages/cultnet-rs/src/service_identity.rs index 3d37adfa..744c4fa3 100644 --- a/packages/cultnet-rs/src/service_identity.rs +++ b/packages/cultnet-rs/src/service_identity.rs @@ -285,11 +285,10 @@ pub fn open_service_identity_credential_reader( pub fn open_service_identity_credential_bytes( bytes: &[u8], ) -> Result> { - let mut decoder = rmp_serde::Deserializer::new(std::io::Cursor::new(bytes)); - let entries = Vec::::deserialize(&mut decoder) + let entries = cultcache_rs::decode_payload::>(bytes) .context("immutable service identity credential is malformed MessagePack")?; - if decoder.position() != bytes.len() as u64 { - bail!("immutable service identity credential contains trailing data"); + if Zeroizing::new(rmp_serde::to_vec(&entries)?).as_slice() != bytes { + bail!("immutable service identity credential is not its canonical encoding"); } open_service_identity_envelopes::

(&entries) } diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index cac4d9dd..df348cee 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -1780,7 +1780,7 @@ fn document_registry_advertises_mutation_contracts_with_bindings() -> Result<()> } #[test] -fn a_stored_payload_nested_past_the_bound_is_not_replicated_as_a_document() -> Result<()> { +fn a_replicated_payload_nested_past_the_bound_is_refused_typed_on_apply() -> Result<()> { let temp = tempfile::tempdir()?; let store = temp.path().join("foreign.msgpack"); let limit = cultcache_rs::MAX_PAYLOAD_DEPTH; @@ -1803,9 +1803,10 @@ fn a_stored_payload_nested_past_the_bound_is_not_replicated_as_a_document() -> R cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store))?; cache.pull_all_backing_stores()?; - let at_limit = registry.create_snapshot_response(&cache, "at", None, None)?; - let CultNetMessage::SnapshotResponse { documents, .. } = at_limit else { - panic!("expected a snapshot response"); + // The raw lane copies payload bytes without decoding; the receiver's apply is the door. + let at_limit = registry.create_raw_snapshot_response(&cache, "at", None, None)?; + let CultNetMessage::SnapshotResponseRaw { documents, .. } = &at_limit else { + panic!("expected a raw snapshot response"); }; assert_eq!(documents.len(), 1); @@ -1814,7 +1815,12 @@ fn a_stored_payload_nested_past_the_bound_is_not_replicated_as_a_document() -> R cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store))?; cache.pull_all_backing_stores()?; - let error = registry.create_snapshot_response(&cache, "past", None, None).unwrap_err(); + let past = registry.create_raw_snapshot_response(&cache, "past", None, Some(&["past".to_string()]))?; + let mut target = CultCache::new(); + target.register_entry_type::()?; + let error = registry + .apply_raw_snapshot_response::(&mut target, &past) + .unwrap_err(); assert_eq!( error.downcast_ref::(), Some(&cultcache_rs::PayloadTooDeep { depth: limit + 1, limit }) @@ -1844,15 +1850,15 @@ fn document_registry_replicates_typed_cultcache_state() -> Result<()> { origin.pull_all_backing_stores()?; origin.put("epiphany.persona", &payload)?; - let snapshot = registry.create_snapshot_response(&origin, "snapshot-1", None, None)?; + let snapshot = registry.create_raw_snapshot_response(&origin, "snapshot-1", None, None)?; let mut target = CultCache::new(); target.register_entry_type::()?; target.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&target_store))?; target.pull_all_backing_stores()?; let applied = - registry.apply_snapshot_response::(&mut target, &snapshot)?; - let synced = registry.sync_document_from_snapshot_response::( + registry.apply_raw_snapshot_response::(&mut target, &snapshot)?; + let synced = registry.sync_raw_document_from_snapshot_response::( &mut target, &snapshot, "epiphany.persona", @@ -1864,7 +1870,7 @@ fn document_registry_replicates_typed_cultcache_state() -> Result<()> { payload ); - let direct_put = registry.create_document_put_message( + let direct_put = registry.create_raw_document_put_message( "put-1", "epiphany.persona", &GhostlightAgentStateFixture { @@ -1874,7 +1880,7 @@ fn document_registry_replicates_typed_cultcache_state() -> Result<()> { CultNetDocumentPutOptions::default(), )?; let updated = registry - .apply_document_put_message::(&mut target, &direct_put)?; + .apply_raw_document_put_message::(&mut target, &direct_put)?; assert_eq!(updated.display_name, "Persona Prime"); Ok(()) } @@ -2260,7 +2266,7 @@ fn reactive_document_tracks_canonical_reconciliation_delta() -> Result<()> { note.body = "local-prediction".to_string(); note.revision = 2; })?; - let authoritative = registry.create_document_put_message( + let authoritative = registry.create_raw_document_put_message( "put-authoritative", "note:reconcile", &GhostlightReactiveNoteFixture { @@ -2270,7 +2276,7 @@ fn reactive_document_tracks_canonical_reconciliation_delta() -> Result<()> { CultNetDocumentPutOptions::default(), )?; - reactive.apply_document_put_message(&authoritative)?; + reactive.apply_raw_document_put_message(&authoritative)?; let current = reactive.current(); let current = current.lock().expect("reactive current mutex").clone(); diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs new file mode 100644 index 00000000..40432367 --- /dev/null +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -0,0 +1,349 @@ +//! Every MessagePack byte cultnet-rs receives decodes once through `cultcache_rs::decode_payload`: +//! a message, the row arrays of `snapshot_response_raw.v1`, a raw document payload applied to a cache, +//! and an immutable identity credential. Each test decodes in a child test process on a thread of a +//! stated size, so an abort fails one named test; values are built, encoded and compared on a large +//! thread, and only the decode runs on the small one. + +use anyhow::Result; +use cultcache_rs::CultCache; +use cultcache_rs::CultSlots; +use cultcache_rs::DatabaseEntry; +use cultcache_rs::MAX_PAYLOAD_DEPTH; +use cultcache_rs::PayloadTooDeep; +use cultnet_rs::*; + +const CHILD_MARKER: &str = "CULTNET_DECODE_DOOR_CHILD"; +const SMALL: usize = 256 * 1024; +const BIG: usize = 64 * 1024 * 1024; +/// A five-byte MessagePack string standing where a test wants its nested value spliced in. +const SENTINEL: [u8; 5] = [0xa4, b'D', b'E', b'E', b'P']; + +#[derive(Clone, Debug, PartialEq, CultSlots)] +struct Wide32 { + #[cultcache(key = 0)] + a0: [u64; 32], + #[cultcache(key = 1)] + a1: [u64; 32], + #[cultcache(key = 2)] + a2: [u64; 32], + #[cultcache(key = 3)] + a3: [u64; 32], + #[cultcache(key = 4)] + a4: [u64; 32], + #[cultcache(key = 5)] + a5: [u64; 32], + #[cultcache(key = 6)] + a6: [u64; 32], + #[cultcache(key = 7)] + a7: [u64; 32], + #[cultcache(key = 8)] + a8: [u64; 32], + #[cultcache(key = 9)] + a9: [u64; 32], + #[cultcache(key = 10)] + a10: [u64; 32], + #[cultcache(key = 11)] + a11: [u64; 32], + #[cultcache(key = 12)] + a12: [u64; 32], + #[cultcache(key = 13)] + a13: [u64; 32], + #[cultcache(key = 14)] + a14: [u64; 32], + #[cultcache(key = 15)] + a15: [u64; 32], + #[cultcache(key = 16)] + a16: [u64; 32], + #[cultcache(key = 17)] + a17: [u64; 32], + #[cultcache(key = 18)] + a18: [u64; 32], + #[cultcache(key = 19)] + a19: [u64; 32], + #[cultcache(key = 20)] + a20: [u64; 32], + #[cultcache(key = 21)] + a21: [u64; 32], + #[cultcache(key = 22)] + a22: [u64; 32], + #[cultcache(key = 23)] + a23: [u64; 32], + #[cultcache(key = 24)] + a24: [u64; 32], + #[cultcache(key = 25)] + a25: [u64; 32], + #[cultcache(key = 26)] + a26: [u64; 32], + #[cultcache(key = 27)] + a27: [u64; 32], + #[cultcache(key = 28)] + a28: [u64; 32], + #[cultcache(key = 29)] + a29: [u64; 32], + #[cultcache(key = 30)] + a30: [u64; 32], + #[cultcache(key = 31)] + a31: [u64; 32], + #[cultcache(key = 32)] + next: Option>, +} + +#[derive(Clone, Debug, PartialEq, DatabaseEntry)] +#[cultcache(type = "cultnet.decode-door.wide", schema = "DecodeDoorWide")] +struct WideDocument { + #[cultcache(key = 0)] + chain: Wide32, +} + +fn on_stack(bytes: usize, body: impl FnOnce() -> R + Send + 'static) -> R { + std::thread::Builder::new().stack_size(bytes).spawn(body).unwrap().join().unwrap() +} + +/// Runs `body` in a child process that runs only the test `name`; the parent passes when the child +/// exits cleanly having run exactly that one test. An abort in the child fails the parent. +fn in_child(name: &str, body: impl FnOnce()) { + if std::env::var_os(CHILD_MARKER).is_some() { + body(); + return; + } + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([name, "--exact", "--nocapture", "--test-threads=1"]) + .env(CHILD_MARKER, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + output.status.success() && stdout.contains("1 passed"), + "child {name} failed: {:?}\n{stdout}\n{stderr}", + output.status + ); +} + +/// `levels` one-element arrays around a nil, from marker bytes: the encoders recurse too. +fn nested(levels: usize) -> Vec { + let mut bytes = vec![0x91; levels]; + bytes.push(0xc0); + bytes +} + +/// Swaps the one sentinel string in `bytes` for `with`. +fn splice(bytes: Vec, with: &[u8]) -> Vec { + let at = bytes + .windows(SENTINEL.len()) + .position(|window| window == SENTINEL) + .expect("sentinel present"); + [&bytes[..at], with, &bytes[at + SENTINEL.len()..]].concat() +} + +fn map_with_sentinel(mut entries: Vec<(rmpv::Value, rmpv::Value)>) -> Vec { + entries.push(("x".into(), "DEEP".into())); + rmp_serde::to_vec(&rmpv::Value::Map(entries)).unwrap() +} + +fn hello_with(unknown: &[u8]) -> Vec { + splice( + map_with_sentinel(vec![ + ("schemaVersion".into(), "cultnet.hello.v0".into()), + ("runtimeId".into(), "rt".into()), + ("runtimeKind".into(), "test".into()), + ]), + unknown, + ) +} + +fn decode_message(bytes: Vec, stack: usize) -> Result { + *on_stack(stack, move || { + Box::new(decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0)) + }) +} + +#[test] +fn generic_message_with_deep_unknown_key_decodes_on_a_small_thread() { + in_child("generic_message_with_deep_unknown_key_decodes_on_a_small_thread", || { + let plain = decode_message(hello_with(&nested(0)), SMALL).unwrap(); + let deep = decode_message(hello_with(&nested(MAX_PAYLOAD_DEPTH - 2)), SMALL).unwrap(); + assert_eq!(deep, plain); + }); +} + +#[test] +fn message_past_the_bound_is_refused_typed() { + in_child("message_past_the_bound_is_refused_typed", || { + let error = decode_message(hello_with(&nested(MAX_PAYLOAD_DEPTH + 1)), SMALL).unwrap_err(); + let refusal = error.downcast_ref::().expect("typed refusal"); + assert_eq!(refusal.limit, MAX_PAYLOAD_DEPTH); + }); +} + +fn raw_v1_bytes(header_unknown: Option<&[u8]>) -> Vec { + let mut header: Vec<(rmpv::Value, rmpv::Value)> = vec![ + ("schemaId".into(), "s".into()), + ("recordKey".into(), "k".into()), + ("storedAt".into(), "2026-10-16T00:00:00Z".into()), + ]; + if header_unknown.is_some() { + header.push(("x".into(), "DEEP".into())); + } + let message = rmpv::Value::Map(vec![ + ("schemaVersion".into(), "cultnet.snapshot_response_raw.v1".into()), + ("messageId".into(), "m".into()), + ("matched".into(), 1.into()), + ("asOf".into(), 7.into()), + ("headers".into(), rmpv::Value::Array(vec![rmpv::Value::Map(header)])), + ]); + let bytes = rmp_serde::to_vec(&message).unwrap(); + match header_unknown { + Some(nested) => splice(bytes, nested), + None => bytes, + } +} + +#[test] +fn raw_v1_rows_with_deep_unknown_key_decode_on_a_small_thread() { + in_child("raw_v1_rows_with_deep_unknown_key_decode_on_a_small_thread", || { + let plain = decode_message(raw_v1_bytes(None), SMALL).unwrap(); + // The message map, the headers array and the header map sit above the unknown key. + let deep = decode_message(raw_v1_bytes(Some(&nested(MAX_PAYLOAD_DEPTH - 3))), SMALL).unwrap(); + let CultNetMessage::SnapshotResponseRawV1 { headers: Some(rows), .. } = &deep else { + panic!("expected raw v1 rows"); + }; + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].record_key, "k"); + assert_eq!(deep, plain); + }); +} + +fn wide_chain(levels: usize) -> Wide32 { + let mut chain: Option> = None; + for level in (0..levels).rev() { + let mut node = Wide32 { next: chain, ..empty_wide() }; + node.a0[0] = level as u64; + chain = Some(Box::new(node)); + } + *chain.unwrap() +} + +fn empty_wide() -> Wide32 { + Wide32 { + a0: [0; 32], + a1: [0; 32], + a2: [0; 32], + a3: [0; 32], + a4: [0; 32], + a5: [0; 32], + a6: [0; 32], + a7: [0; 32], + a8: [0; 32], + a9: [0; 32], + a10: [0; 32], + a11: [0; 32], + a12: [0; 32], + a13: [0; 32], + a14: [0; 32], + a15: [0; 32], + a16: [0; 32], + a17: [0; 32], + a18: [0; 32], + a19: [0; 32], + a20: [0; 32], + a21: [0; 32], + a22: [0; 32], + a23: [0; 32], + a24: [0; 32], + a25: [0; 32], + a26: [0; 32], + a27: [0; 32], + a28: [0; 32], + a29: [0; 32], + a30: [0; 32], + a31: [0; 32], + next: None, + } +} + +fn registry() -> CultNetDocumentRegistry { + let mut registry = CultNetDocumentRegistry::new(); + registry.register(CultNetDocumentBinding::for_entry::("cultnet.decode-door.wide".to_string())); + registry +} + +/// The wire bytes of a raw put of a `levels`-deep Wide32 document, built on a large thread. +fn raw_put_bytes(levels: usize) -> Vec { + on_stack(BIG, move || { + let document = WideDocument { chain: wide_chain(levels) }; + let message = registry() + .create_raw_document_put_message("m", "k", &document, CultNetDocumentPutOptions::default()) + .unwrap(); + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap() + }) +} + +/// Receives and applies on the small thread; the applied record is boxed there. +fn receive_and_apply(bytes: Vec) -> Result { + *on_stack(SMALL, move || { + Box::new((|| { + let message = decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0)?; + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + registry().apply_raw_document_put_message::(&mut cache, &message) + })()) + }) +} + +#[test] +fn raw_put_of_a_wide32_chain_applies_on_a_small_thread() { + in_child("raw_put_of_a_wide32_chain_applies_on_a_small_thread", || { + let levels = MAX_PAYLOAD_DEPTH - 2; + let applied = receive_and_apply(raw_put_bytes(levels)).unwrap(); + on_stack(BIG, move || assert!(applied.chain == wide_chain(levels))); + let error = receive_and_apply(raw_put_bytes(levels + 1)).unwrap_err(); + assert!(error.downcast_ref::().is_some(), "typed refusal"); + }); +} + +#[test] +fn retired_dynamic_put_is_refused_typed() { + in_child("retired_dynamic_put_is_refused_typed", || { + let document = rmpv::Value::Map(vec![ + ("schemaId".into(), "s".into()), + ("recordKey".into(), "k".into()), + ("storedAt".into(), "2026-10-16T00:00:00Z".into()), + ("payload".into(), "DEEP".into()), + ]); + let bytes = splice( + rmp_serde::to_vec(&rmpv::Value::Map(vec![ + ("schemaVersion".into(), "cultnet.document_put.v0".into()), + ("messageId".into(), "m".into()), + ("document".into(), document), + ])) + .unwrap(), + &nested(253), + ); + let error = decode_message(bytes, SMALL).unwrap_err(); + // The refusal keeps its kind and never quotes the received variant name. + assert!(!format!("{error:#}").contains("document_put"), "{error:#}"); + }); +} + +#[test] +fn identity_credential_with_deep_key_is_refused_on_a_small_thread() { + in_child("identity_credential_with_deep_key_is_refused_on_a_small_thread", || { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("identity.cc"); + enroll_service_identity_at::(&path).unwrap(); + let mut envelopes: rmpv::Value = rmp_serde::from_slice(&std::fs::read(path).unwrap()).unwrap(); + let rmpv::Value::Map(envelope) = &mut envelopes.as_array_mut().unwrap()[0] else { + panic!("envelope is a map"); + }; + envelope.push(("zz".into(), "DEEP".into())); + let credential = rmp_serde::to_vec(&envelopes).unwrap(); + for levels in [1000, 250] { + let bytes = splice(credential.clone(), &nested(levels)); + let refused = on_stack(SMALL, move || { + open_service_identity_credential_bytes::(&bytes).is_err() + }); + assert!(refused, "{levels} levels deep"); + } + }); +} From a38768202af4e54d78a1aa2b42e1fdc7c466086b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 00:34:55 +0200 Subject: [PATCH 02/13] cultnet-rs: fix decode_door credential test and unused import Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/src/replication.rs | 1 - packages/cultnet-rs/tests/decode_door.rs | 5 ++++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/packages/cultnet-rs/src/replication.rs b/packages/cultnet-rs/src/replication.rs index 85f9574d..d29714fe 100644 --- a/packages/cultnet-rs/src/replication.rs +++ b/packages/cultnet-rs/src/replication.rs @@ -1,4 +1,3 @@ -use anyhow::Context; use anyhow::Result; use anyhow::anyhow; use cultcache_rs::CultCache; diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index 40432367..813112da 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -333,7 +333,10 @@ fn identity_credential_with_deep_key_is_refused_on_a_small_thread() { let path = temp.path().join("identity.cc"); enroll_service_identity_at::(&path).unwrap(); let mut envelopes: rmpv::Value = rmp_serde::from_slice(&std::fs::read(path).unwrap()).unwrap(); - let rmpv::Value::Map(envelope) = &mut envelopes.as_array_mut().unwrap()[0] else { + let rmpv::Value::Array(items) = &mut envelopes else { + panic!("credential is an array"); + }; + let rmpv::Value::Map(envelope) = &mut items[0] else { panic!("envelope is a map"); }; envelope.push(("zz".into(), "DEEP".into())); From 7fb7474cc98f68c060833f4f01b8bd96e61054df Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 00:40:08 +0200 Subject: [PATCH 03/13] cultnet-rs: decode_door credential test adds an unknown slot to the positional envelope Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/tests/decode_door.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index 813112da..4d58a6a9 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -336,10 +336,11 @@ fn identity_credential_with_deep_key_is_refused_on_a_small_thread() { let rmpv::Value::Array(items) = &mut envelopes else { panic!("credential is an array"); }; - let rmpv::Value::Map(envelope) = &mut items[0] else { - panic!("envelope is a map"); + // The writer encodes an envelope positionally; one slot more is the unknown one. + let rmpv::Value::Array(envelope) = &mut items[0] else { + panic!("envelope is a positional array"); }; - envelope.push(("zz".into(), "DEEP".into())); + envelope.push("DEEP".into()); let credential = rmp_serde::to_vec(&envelopes).unwrap(); for levels in [1000, 250] { let bytes = splice(credential.clone(), &nested(levels)); From bca8a70a5ae3fabc527da34a731a2a2363252685 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 00:44:28 +0200 Subject: [PATCH 04/13] cultnet-rs: decode_door credential test spells the envelope as a map with one unknown key Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/tests/decode_door.rs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index 4d58a6a9..29488503 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -336,11 +336,18 @@ fn identity_credential_with_deep_key_is_refused_on_a_small_thread() { let rmpv::Value::Array(items) = &mut envelopes else { panic!("credential is an array"); }; - // The writer encodes an envelope positionally; one slot more is the unknown one. - let rmpv::Value::Array(envelope) = &mut items[0] else { + // The writer encodes an envelope positionally; the reader also takes the named map form, + // where an unknown key is skipped, so the test spells the envelope that way plus one key. + let rmpv::Value::Array(slots) = std::mem::replace(&mut items[0], rmpv::Value::Nil) else { panic!("envelope is a positional array"); }; - envelope.push("DEEP".into()); + let mut named: Vec<(rmpv::Value, rmpv::Value)> = ["key", "type", "payload", "storedAt", "schemaId"] + .into_iter() + .map(rmpv::Value::from) + .zip(slots) + .collect(); + named.push(("zz".into(), "DEEP".into())); + items[0] = rmpv::Value::Map(named); let credential = rmp_serde::to_vec(&envelopes).unwrap(); for levels in [1000, 250] { let bytes = splice(credential.clone(), &nested(levels)); From a3a03be589ad21284aaf560a7591a4abfe54f6bd Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 00:57:53 +0200 Subject: [PATCH 05/13] cultnet-rs: pin message validation after the door and the current 18-slot health wire Mutation survivors of the first run: validate_message -> Ok(()) and the Some(18) arm of decode_canonical_signed_payload. Co-Authored-By: Claude Opus 5.5 --- .../cultnet-rs/src/provider_health_contracts.rs | 5 +++++ packages/cultnet-rs/tests/decode_door.rs | 14 ++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/packages/cultnet-rs/src/provider_health_contracts.rs b/packages/cultnet-rs/src/provider_health_contracts.rs index f4ca108c..8764f651 100644 --- a/packages/cultnet-rs/src/provider_health_contracts.rs +++ b/packages/cultnet-rs/src/provider_health_contracts.rs @@ -643,6 +643,11 @@ mod tests { rmp_serde::from_slice::(&encoded)?, health ); + let (decoded, unsigned) = + IdunnSignedDaemonHealthRecord::decode_canonical_signed_payload(&encoded)?; + assert_eq!(decoded, health); + assert_eq!(&unsigned[..3], &[0xdc, 0, 18]); + assert_ne!(unsigned, encoded, "the signed payload excludes the signature"); assert_eq!( rmp_serde::from_slice::(&rmp_serde::to_vec( &anchor diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index 29488503..99d82cad 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -167,6 +167,20 @@ fn generic_message_with_deep_unknown_key_decodes_on_a_small_thread() { }); } +#[test] +fn a_decoded_message_that_fails_validation_is_refused() { + let bytes = splice( + map_with_sentinel(vec![ + ("schemaVersion".into(), "cultnet.hello.v0".into()), + ("runtimeId".into(), "".into()), + ("runtimeKind".into(), "test".into()), + ]), + &nested(0), + ); + let error = decode_message(bytes, BIG).unwrap_err(); + assert!(format!("{error:#}").contains("runtimeId"), "{error:#}"); +} + #[test] fn message_past_the_bound_is_refused_typed() { in_child("message_past_the_bound_is_refused_typed", || { From c0a2b890f62306ca0371605f7ae82df4f50e40ea Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 04:22:22 +0200 Subject: [PATCH 06/13] cultcache-rs: the decode door names a missing or duplicated field and never quotes a value describe_decode_error keeps serde's own value-free texts for missing and duplicate fields (static names from the type), maps unknown-variant and invalid-type to kind descriptions, and leaves everything else opaque. Co-Authored-By: Claude Opus 5.5 --- packages/cultcache-rs/src/lib.rs | 27 ++++++++++++++++++- packages/cultcache-rs/tests/union_parity.rs | 30 +++++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index d58f1242..ecb69145 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -732,6 +732,7 @@ fn opaque_decode_error(error: rmp_serde::decode::Error) -> anyhow::Error { anyhow!(describe_decode_error(&error)) } +/// The text of a decode failure. Field names in it come from the type, never from the bytes. fn describe_decode_error(error: &rmp_serde::decode::Error) -> String { use rmp_serde::decode::Error; match error { @@ -744,11 +745,35 @@ fn describe_decode_error(error: &rmp_serde::decode::Error) -> String { 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", + Error::Syntax(message) | Error::Uncategorized(message) => { + return describe_serde_message(message); + } } .to_string() } +/// serde's own texts, kept only where they cannot carry a received value: a missing or duplicated +/// field is named by the type's static field name (checked to look like one), an unknown tag and a +/// mistyped value are described by kind. Everything else, `unknown field` included, is opaque. +fn describe_serde_message(message: &str) -> String { + for lead in ["missing field `", "duplicate field `"] { + if let Some(name) = message.strip_prefix(lead).and_then(|rest| rest.strip_suffix('`')) { + let is_name = (1..=64).contains(&name.len()) + && name.bytes().all(|byte| byte.is_ascii_alphanumeric() || byte == b'_'); + if is_name { + return message.to_string(); + } + } + } + if message.starts_with("unknown variant ") { + "a tag names no variant this type knows".to_string() + } else if message.starts_with("invalid type: ") { + "a value has a MessagePack type the store format does not allow".to_string() + } else { + "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. diff --git a/packages/cultcache-rs/tests/union_parity.rs b/packages/cultcache-rs/tests/union_parity.rs index a566e627..0caa8ee9 100644 --- a/packages/cultcache-rs/tests/union_parity.rs +++ b/packages/cultcache-rs/tests/union_parity.rs @@ -293,6 +293,36 @@ fn a_union_that_is_not_a_key_and_slots_pair_is_refused_naming_the_enum() { } } +// A plain derive struct: the door names a missing or duplicated field (a static name from the type) +// and describes a mistyped value by kind, never quoting it. +#[derive(Debug, serde::Deserialize)] +#[allow(dead_code)] +struct Probe { + name: String, + flag: bool, +} + +#[test] +fn a_missing_or_duplicate_field_is_named_and_a_mistyped_value_is_not() { + let name = |value: &str| [&[0xa4][..], b"name", &[0xa0 | value.len() as u8][..], value.as_bytes()].concat(); + let flag_true = [&[0xa4][..], b"flag", &[0xc3][..]].concat(); + + let missing = [&[0x81][..], &name("a")].concat(); + let message = cultcache_rs::decode_payload::(&missing).unwrap_err().to_string(); + assert!(message.contains("missing field `flag`"), "{message}"); + + let duplicated = [&[0x83][..], &name("a"), &name("b"), &flag_true].concat(); + let message = cultcache_rs::decode_payload::(&duplicated).unwrap_err().to_string(); + assert!(message.contains("duplicate field `name`"), "{message}"); + + let canary = "CANARY-3d9e"; + let flag_text = [&[0xa4][..], b"flag", &[0xa0 | canary.len() as u8][..], canary.as_bytes()].concat(); + let mistyped = [&[0x82][..], &name("a"), &flag_text].concat(); + let message = cultcache_rs::decode_payload::(&mistyped).unwrap_err().to_string(); + assert!(!message.contains("CANARY"), "{message}"); + assert!(message.contains("MessagePack type"), "{message}"); +} + #[test] fn the_key_decides_the_arm() { // Tool and Part carry different slots under different keys in the two unions. From 377c24b8cd760cf58d2630f961a1f7dcc031f9d3 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 04:22:27 +0200 Subject: [PATCH 07/13] cultnet-rs: one decoder per family, duplicate keys refused, value-free receive errors Raw v1 is decoded by the enum's derive alone (SnapshotResponseRawV1Rows, the v1 arm and its helpers deleted); the five raw families carry skip_deserializing so the hand reader is their only decoder. The four hand readers refuse a duplicated key; refusals carry a schema-shaped schemaVersion context and never a received value. README describes only the raw lane. A same-length str-payload credential forgery is pinned. Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/README.md | 6 +- packages/cultnet-rs/src/contracts.rs | 143 +++++------ packages/cultnet-rs/src/service_identity.rs | 31 +++ packages/cultnet-rs/tests/decode_door.rs | 268 +++++++++++++++++++- 4 files changed, 365 insertions(+), 83 deletions(-) diff --git a/packages/cultnet-rs/README.md b/packages/cultnet-rs/README.md index 8bea2899..3c7488ba 100644 --- a/packages/cultnet-rs/README.md +++ b/packages/cultnet-rs/README.md @@ -11,9 +11,8 @@ The contract is intentionally boring: - `gamecult.networking.v0` maps the legacy C# union shape explicitly. - schema discovery uses explicit catalog request/response messages; no inbound auto-detect sludge. -- document put/delete/snapshot messages move typed CultCache entries. -- raw document put/snapshot messages preserve canonical MessagePack payload - bytes for bit-compatible neighbors. +- put/delete/snapshot travel only as raw messages that carry the canonical + MessagePack payload bytes; the receiver decodes them through registered types. - hello messages advertise `supportedMutationContracts`, so callers discover which document types are read-only, which accept typed intents, which authority owns the mutation, and which receipt documents prove the outcome. @@ -41,7 +40,6 @@ The initial tests prove: - legacy `gamecult.networking.v0` login mapping - schema discovery catalog responses with canonical JSON schema hashes - shard catalog request/response filtering and response application -- CultCache snapshot replication through registered typed documents - raw snapshot replication that preserves the original payload bytes - raw snapshot replication into a local same-schema Rust alias type - document mutation contract advertisement through hello frames and registries diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index fa42131a..0a8f657a 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -1,5 +1,7 @@ +use anyhow::Context; use anyhow::Result; use anyhow::anyhow; +use anyhow::bail; use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use serde::Deserialize; @@ -266,7 +268,7 @@ impl CultNetErrorCode { "reference_outside_target" => Ok(Self::ReferenceOutsideTarget), "variant_unsupported" => Ok(Self::VariantUnsupported), "unowned_schema" => Ok(Self::UnownedSchema), - other => Err(anyhow!("ErrorMessage.Code {other:?} is not a recognized cultnet.error.v0 code")), + _ => Err(anyhow!("ErrorMessage.Code is not a recognized cultnet.error.v0 code")), } } } @@ -335,11 +337,11 @@ pub enum CultNetMessage { // R-N (docs/cultnet-selection-cut.md, fix batch 3): `code`/`details` mirror the landed C# // reference (CultNetSchemaMessages.cs) exactly, including its always-five-key map shape // (`schemaVersion`, `error`, `routingHint`, `code`, `details`, every optional key nil rather - // than omitted). `Error` is therefore encoded/decoded through the raw - // `cultnet.schema.v0` path (parse_raw_cultnet_schema_message/encode_raw_cultnet_schema_message) - // like the other messages that need exact wire control, not through this derive - the derive - // attributes below exist only so the enum as a whole still compiles and round-trips through the - // generic variants, which decode through the door from the message bytes. + // than omitted). `Error` is read and written only by the raw `cultnet.schema.v0` path + // (parse_raw_cultnet_schema_message/encode_raw_cultnet_schema_message): `skip_deserializing` + // makes the derive unable to decode it, so the raw reader is its one decoder and the derive + // attributes describe `Serialize` only. + #[serde(skip_deserializing)] #[serde(rename = "cultnet.error.v0", rename_all = "camelCase")] Error { error: String, @@ -358,6 +360,7 @@ pub enum CultNetMessage { schema_id: String, record_key: String, }, + #[serde(skip_deserializing)] #[serde(rename = "cultnet.document_put_raw.v0", rename_all = "camelCase")] DocumentPutRaw { message_id: String, @@ -371,6 +374,7 @@ pub enum CultNetMessage { #[serde(default, skip_serializing_if = "Option::is_none")] record_keys: Option>, }, + #[serde(skip_deserializing)] #[serde(rename = "cultnet.snapshot_response_raw.v0", rename_all = "camelCase")] SnapshotResponseRaw { message_id: String, @@ -402,10 +406,10 @@ pub enum CultNetMessage { #[serde(default, skip_serializing_if = "Option::is_none")] supported_body_transports: Option>, }, - /// `cultnet.snapshot_response_raw.v1` (section 2: `SelectionPage`). Binary payload, so this is - /// carried through `encode_raw_cultnet_schema_message`/`parse_raw_cultnet_schema_message` like - /// `SnapshotResponseRaw` above - a JSON intermediate cannot losslessly carry `documents[].payload` - /// or `edges[].payload`. Mirrors `CultNetSnapshotResponseRawV1Message`. + /// `cultnet.snapshot_response_raw.v1` (section 2: `SelectionPage`). Decoded by the derive from + /// the message bytes in one pass (rows and scalars, a duplicated key refused); encoded by + /// `encode_raw_cultnet_schema_message` for the reference's fixed nil-keyed shape. Mirrors + /// `CultNetSnapshotResponseRawV1Message`. #[serde(rename = "cultnet.snapshot_response_raw.v1", rename_all = "camelCase")] SnapshotResponseRawV1 { message_id: String, @@ -487,6 +491,7 @@ pub enum CultNetMessage { source_runtime_id: Option, }, /// `cultmesh.content_chunk_request.v1`. Mirrors `CultMeshContentChunkRequestMessage`; raw path. + #[serde(skip_deserializing)] #[serde(rename = "cultmesh.content_chunk_request.v1", rename_all = "camelCase")] ContentChunkRequest { message_id: String, @@ -496,6 +501,7 @@ pub enum CultNetMessage { }, /// `cultmesh.content_chunk_response.v1`. Mirrors `CultMeshContentChunkResponseMessage`. `payload` /// is `bin`, so it travels the raw path; the serde attribute never runs on the wire. + #[serde(skip_deserializing)] #[serde(rename = "cultmesh.content_chunk_response.v1", rename_all = "camelCase")] ContentChunkResponse { message_id: String, @@ -556,12 +562,13 @@ pub fn decode_cultnet_message_from_slice( Some( "cultnet.document_put_raw.v0" | "cultnet.snapshot_response_raw.v0" - | "cultnet.snapshot_response_raw.v1" | "cultmesh.content_chunk_request.v1" | "cultmesh.content_chunk_response.v1" | "cultnet.error.v0", - ) => parse_raw_cultnet_schema_message(bytes, &wire_value)?, - _ => cultcache_rs::decode_payload::(bytes)?, + ) => parse_raw_cultnet_schema_message(&wire_value) + .with_context(|| received_version_label(&wire_value))?, + _ => cultcache_rs::decode_payload::(bytes) + .with_context(|| received_version_label(&wire_value))?, }; validate_message(&message)?; Ok(message) @@ -911,24 +918,52 @@ fn schema_version_from_wire_value(input: &rmpv::Value) -> Result> { })) } -/// The row arrays of `cultnet.snapshot_response_raw.v1`, decoded from the original bytes so each -/// row is typed in one pass through the door; nil and absent both read as `None`, other keys are -/// ignored. -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct SnapshotResponseRawV1Rows { - #[serde(default)] - headers: Option>, - #[serde(default)] - documents: Option>, - #[serde(default)] - edges: Option>, +/// True for a name shaped like a schema version (`cultnet.hello.v0`): a received value is named +/// in a refusal only when it cannot be anything else. +fn looks_like_schema_name(candidate: &str) -> bool { + let mut parts = candidate.split('.'); + let segment = |part: &str| !part.is_empty() && part.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_'); + let first_ok = parts.next().is_some_and(|first| { + first.bytes().next().is_some_and(|b| b.is_ascii_lowercase()) && segment(first) + }); + let rest: Vec<&str> = parts.collect(); + first_ok && !rest.is_empty() && rest.iter().all(|part| segment(*part)) && candidate.len() <= 96 } -fn parse_raw_cultnet_schema_message(bytes: &[u8], input: &rmpv::Value) -> Result { +/// What a refusal says of the message it refused: its `schemaVersion` when that is shaped like a +/// schema name, else only the shape of what arrived. Never a received value that could be text. +fn received_version_label(wire_value: &rmpv::Value) -> String { + match schema_version_from_wire_value(wire_value) { + Ok(Some(version)) if looks_like_schema_name(version) => { + format!("cultnet.schema.v0 message {version} refused") + } + Ok(Some(version)) => format!( + "cultnet.schema.v0 message whose schemaVersion is not a schema name ({} bytes)", + version.len() + ), + _ => "cultnet.schema.v0 message with no schemaVersion".to_string(), + } +} + +/// Refuses a map that spells a string key twice, so no reader resolves a duplicate by position. +/// Names the context, never the key (it is received text). +fn refuse_duplicate_keys(object: &[(rmpv::Value, rmpv::Value)], context: &str) -> Result<()> { + let mut seen = std::collections::HashSet::new(); + for (key, _) in object { + if let Some(name) = key.as_str() { + if !seen.insert(name) { + bail!("{context} spells a key twice"); + } + } + } + Ok(()) +} + +fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result { let object = input .as_map() .ok_or_else(|| anyhow!("cultnet.schema.v0 raw messages must be objects"))?; + refuse_duplicate_keys(object, "cultnet.schema.v0 raw message")?; let get = |name: &str| -> Option<&rmpv::Value> { object.iter().find_map(|(key, value)| { @@ -960,26 +995,6 @@ fn parse_raw_cultnet_schema_message(bytes: &[u8], input: &rmpv::Value) -> Result documents, }) } - "cultnet.snapshot_response_raw.v1" => { - let SnapshotResponseRawV1Rows { headers, documents, edges } = - cultcache_rs::decode_payload(bytes).map_err(|err| anyhow!("rows: {err}"))?; - - Ok(CultNetMessage::SnapshotResponseRawV1 { - message_id: require_legacy_string(get("messageId"), "messageId")?, - matched: require_legacy_u32(get("matched"), "matched")?, - as_of: require_legacy_u64(get("asOf"), "asOf")?, - next: require_legacy_optional_string(get("next"), "next")?, - headers, - documents, - edges, - shard_id: require_legacy_optional_string(get("shardId"), "shardId")?, - shard_epoch: require_legacy_optional_i64(get("shardEpoch"), "shardEpoch")?, - shard_log_sequence: require_legacy_optional_i64( - get("shardLogSequence"), - "shardLogSequence", - )?, - }) - } "cultmesh.content_chunk_request.v1" => Ok(CultNetMessage::ContentChunkRequest { message_id: legacy_string_or_nil(get("messageId"), "messageId")?, chunk_hash: legacy_string_or_nil(get("chunkHash"), "chunkHash")?, @@ -1033,6 +1048,7 @@ fn require_error_details( let object = value .as_map() .ok_or_else(|| anyhow!("{field_name} must be an object"))?; + refuse_duplicate_keys(object, field_name)?; let get = |name: &str| -> Option<&rmpv::Value> { object.iter().find_map(|(key, value)| { key.as_str() @@ -1274,6 +1290,7 @@ fn require_raw_document_record( let object = value .and_then(rmpv::Value::as_map) .ok_or_else(|| anyhow!("{field_name} must be an object"))?; + refuse_duplicate_keys(object, field_name)?; let get = |name: &str| -> Option<&rmpv::Value> { object.iter().find_map(|(key, value)| { @@ -1296,7 +1313,7 @@ fn require_raw_document_record( "messagepack" => CultNetRawPayloadEncoding::Messagepack, _ => { return Err(anyhow!( - "{field_name}.payloadEncoding has unsupported value {payload_encoding}" + "{field_name}.payloadEncoding names an encoding other than messagepack" )); } }, @@ -1380,20 +1397,13 @@ fn encode_raw_document_record(document: &CultNetRawDocumentRecord) -> rmpv::Valu // `encode_typed` goes through `rmp_serde::to_vec_named` (the named-map encoding this wire always // uses; `rmpv::ext::to_value` would write a struct positionally) and reads the bytes back into the // `rmpv::Value` tree the rest of this codec builds by hand, through the decode door. Received rows -// are never re-encoded to become typed: `SnapshotResponseRawV1Rows` reads them from the original bytes. +// are never re-encoded to become typed: the enum's derive reads them from the original bytes. fn encode_typed(value: &T) -> rmpv::Value { let bytes = rmp_serde::to_vec_named(value).expect("selection wire types always encode to MessagePack"); cultcache_rs::decode_payload(&bytes).expect("named-map MessagePack bytes always decode into rmpv::Value") } -fn require_legacy_u32(value: Option<&rmpv::Value>, field_name: &str) -> Result { - let number = value - .and_then(rmpv::Value::as_u64) - .ok_or_else(|| anyhow!("{field_name} must be a non-negative integer"))?; - u32::try_from(number).map_err(|_| anyhow!("{field_name} does not fit in u32")) -} - fn require_legacy_i32(value: Option<&rmpv::Value>, field_name: &str) -> Result { let number = value .and_then(rmpv::Value::as_i64) @@ -1401,28 +1411,6 @@ fn require_legacy_i32(value: Option<&rmpv::Value>, field_name: &str) -> Result, field_name: &str) -> Result { - value - .and_then(rmpv::Value::as_u64) - .ok_or_else(|| anyhow!("{field_name} must be a non-negative integer")) -} - -fn require_legacy_optional_i64( - value: Option<&rmpv::Value>, - field_name: &str, -) -> Result> { - let Some(value) = value else { - return Ok(None); - }; - if value.is_nil() { - return Ok(None); - } - value - .as_i64() - .map(Some) - .ok_or_else(|| anyhow!("{field_name} must be an integer")) -} - fn parse_gamecult_networking_message(input: &rmpv::Value) -> Result { let items = input.as_array().ok_or_else(|| { anyhow!("gamecult.networking.v0 messages must be a 2-element union array") @@ -1699,6 +1687,7 @@ fn require_legacy_schema_descriptor( let object = value .as_map() .ok_or_else(|| anyhow!("{field_name} must be an object in gamecult.networking.v0"))?; + refuse_duplicate_keys(object, field_name)?; let get = |name: &str| -> Option<&rmpv::Value> { object.iter().find_map(|(key, value)| { diff --git a/packages/cultnet-rs/src/service_identity.rs b/packages/cultnet-rs/src/service_identity.rs index 744c4fa3..63ec29f6 100644 --- a/packages/cultnet-rs/src/service_identity.rs +++ b/packages/cultnet-rs/src/service_identity.rs @@ -768,6 +768,37 @@ mod tests { Ok(()) } + #[test] + fn a_same_length_credential_spelled_with_a_str_payload_is_refused() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("idunn.ccmp"); + enroll_service_identity_at::(&path)?; + let bytes = std::fs::read(path)?; + open_service_identity_credential_bytes::(&bytes)?; + + // The envelope's payload spelled as a MessagePack str of the same bytes (bin8 -> str8, bin16 -> + // str16): same length, same decoded entries, a different encoding. + let envelopes: Vec = rmp_serde::from_slice(&bytes)?; + let payload = &envelopes[0].payload; + let at = bytes + .windows(payload.len()) + .position(|window| window == payload.as_slice()) + .expect("the payload is in the credential"); + let mut forged = bytes.clone(); + let (bin, str_marker, header) = if payload.len() <= 0xff { (0xc4, 0xd9, 2) } else { (0xc5, 0xda, 3) }; + assert_eq!(forged[at - header], bin, "payload spelled bin"); + forged[at - header] = str_marker; + assert_eq!(forged.len(), bytes.len()); + let reread: Vec = rmp_serde::from_slice(&forged)?; + assert_eq!(reread, envelopes, "the forgery decodes to identical entries"); + + let error = open_service_identity_credential_bytes::(&forged) + .err() + .expect("a differently spelled credential is refused"); + assert!(format!("{error:#}").contains("canonical encoding"), "{error:#}"); + Ok(()) + } + #[test] fn immutable_credential_reader_opens_without_a_path_and_retains_signing_authority() -> Result<()> { diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index 99d82cad..09dddb91 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -335,8 +335,8 @@ fn retired_dynamic_put_is_refused_typed() { &nested(253), ); let error = decode_message(bytes, SMALL).unwrap_err(); - // The refusal keeps its kind and never quotes the received variant name. - assert!(!format!("{error:#}").contains("document_put"), "{error:#}"); + // The refusal names the schema the sender spelled, shaped like one, and nothing else of it. + assert!(format!("{error:#}").contains("cultnet.document_put.v0"), "{error:#}"); }); } @@ -372,3 +372,267 @@ fn identity_credential_with_deep_key_is_refused_on_a_small_thread() { } }); } + +type Entries = Vec<(rmpv::Value, rmpv::Value)>; + +fn pairs(entries: &[(&str, rmpv::Value)]) -> Entries { + entries.iter().map(|(key, value)| ((*key).into(), value.clone())).collect() +} + +fn encoded(entries: Entries) -> Vec { + rmp_serde::to_vec(&rmpv::Value::Map(entries)).unwrap() +} + +/// The refusal text of a message built from `entries`, as every receiver prints it. +fn refusal(entries: Entries) -> String { + format!("{:#}", decode_message(encoded(entries), BIG).unwrap_err()) +} + +fn wire_record(extra: &[(&str, rmpv::Value)]) -> rmpv::Value { + let mut entries = pairs(&[ + ("schemaId", "s".into()), + ("recordKey", "k".into()), + ("storedAt", "2026-10-16T00:00:00Z".into()), + ("payloadEncoding", "messagepack".into()), + ("payload", rmpv::Value::Binary(vec![0x90])), + ]); + entries.extend(pairs(extra)); + rmpv::Value::Map(entries) +} + +fn raw_record() -> CultNetRawDocumentRecord { + CultNetRawDocumentRecord { + schema_id: "s".into(), + record_key: "k".into(), + stored_at: "2026-10-16T00:00:00Z".into(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: vec![0x90], + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + } +} + +fn sv(name: &str) -> (&str, rmpv::Value) { + ("schemaVersion", rmpv::Value::from(name)) +} + +/// One valid message of every family the hand reader owns. +fn raw_family_messages() -> Vec { + vec![ + CultNetMessage::DocumentPutRaw { message_id: "m".into(), document: raw_record() }, + CultNetMessage::SnapshotResponseRaw { message_id: "m".into(), documents: vec![raw_record()] }, + CultNetMessage::ContentChunkRequest { + message_id: "m".into(), + chunk_hash: "h".into(), + record_key: "k".into(), + expected_size_bytes: 1, + }, + CultNetMessage::ContentChunkResponse { + message_id: "m".into(), + found: true, + chunk_hash: "h".into(), + size_bytes: 1, + payload: vec![1], + error: String::new(), + }, + CultNetMessage::Error { error: "e".into(), code: None, details: None }, + ] +} + +#[test] +fn every_raw_family_has_exactly_one_decoder() { + for message in raw_family_messages() { + let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap(); + // The derive cannot decode a raw-owned family: only the hand reader can. + assert!(cultcache_rs::decode_payload::(&bytes).is_err(), "{message:?}"); + let decoded = decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0).unwrap(); + assert_eq!(decoded, message); + } + // v1 belongs to the derive alone: it decodes the bytes, and the door returns the same message. + let v1 = CultNetMessage::SnapshotResponseRawV1 { + message_id: "m".into(), + matched: 1, + as_of: 7, + next: None, + headers: Some(vec![]), + documents: None, + edges: None, + shard_id: None, + shard_epoch: None, + shard_log_sequence: None, + }; + let bytes = encode_cultnet_message_to_vec(&v1, CultNetWireContract::CultNetSchemaV0).unwrap(); + assert_eq!(cultcache_rs::decode_payload::(&bytes).unwrap(), v1); + assert_eq!( + decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0).unwrap(), + v1 + ); +} + +#[test] +fn a_duplicated_key_is_refused_in_every_family() { + let bases: Vec<(&str, Vec<(&str, rmpv::Value)>)> = vec![ + ("put_raw", vec![sv("cultnet.document_put_raw.v0"), ("messageId", "m".into()), ("document", wire_record(&[]))]), + ( + "snapshot_raw", + vec![ + sv("cultnet.snapshot_response_raw.v0"), + ("messageId", "m".into()), + ("documents", rmpv::Value::Array(vec![wire_record(&[])])), + ], + ), + ( + "chunk_request", + vec![ + sv("cultmesh.content_chunk_request.v1"), + ("messageId", "m".into()), + ("chunkHash", "h".into()), + ("recordKey", "k".into()), + ("expectedSizeBytes", 1.into()), + ], + ), + ( + "chunk_response", + vec![ + sv("cultmesh.content_chunk_response.v1"), + ("messageId", "m".into()), + ("found", true.into()), + ("chunkHash", "h".into()), + ("sizeBytes", 1.into()), + ("payload", rmpv::Value::Binary(vec![1])), + ("error", "".into()), + ], + ), + ("error", vec![sv("cultnet.error.v0"), ("error", "e".into())]), + ( + "v1", + vec![ + sv("cultnet.snapshot_response_raw.v1"), + ("messageId", "m".into()), + ("matched", 1.into()), + ("asOf", 7.into()), + ("headers", rmpv::Value::Array(vec![])), + ], + ), + ("hello", vec![sv("cultnet.hello.v0"), ("runtimeId", "rt".into()), ("runtimeKind", "test".into())]), + ]; + for (label, base) in &bases { + decode_message(encoded(pairs(base)), BIG).unwrap_or_else(|error| panic!("{label} control: {error:#}")); + // Each key in turn spelled a second time, with a different value. + for index in 0..base.len() { + let mut doubled = base.clone(); + doubled.push((base[index].0, "second".into())); + let text = refusal(pairs(&doubled)); + assert!( + text.contains("twice") || text.contains("duplicate field"), + "{label} {}: {text}", + base[index].0 + ); + } + } + // A key spelled twice one level down: in a document record (both record lanes) and in error details. + let doubled_record = wire_record(&[("schemaId", "t".into())]); + for entries in [ + pairs(&[sv("cultnet.document_put_raw.v0"), ("messageId", "m".into()), ("document", doubled_record.clone())]), + pairs(&[ + sv("cultnet.snapshot_response_raw.v0"), + ("messageId", "m".into()), + ("documents", rmpv::Value::Array(vec![doubled_record.clone()])), + ]), + ] { + let text = refusal(entries); + assert!(text.contains("twice"), "{text}"); + } + let error_with = |details: Entries| { + pairs(&[ + sv("cultnet.error.v0"), + ("error", "e".into()), + ("code", "selection_invalid".into()), + ("details", rmpv::Value::Map(details)), + ]) + }; + decode_message(encoded(error_with(pairs(&[("field", "f".into())]))), BIG).unwrap(); + let text = refusal(error_with(pairs(&[("field", "f".into()), ("field", "g".into())]))); + assert!(text.contains("twice"), "{text}"); + // v1 rows come from the derive, which refuses a key spelled twice inside a row. + let v1_header = |extra: &[(&str, rmpv::Value)]| { + let mut header = pairs(&[ + ("schemaId", "s".into()), + ("recordKey", "k".into()), + ("storedAt", "2026-10-16T00:00:00Z".into()), + ]); + header.extend(pairs(extra)); + pairs(&[ + sv("cultnet.snapshot_response_raw.v1"), + ("messageId", "m".into()), + ("matched", 1.into()), + ("asOf", 7.into()), + ("headers", rmpv::Value::Array(vec![rmpv::Value::Map(header)])), + ]) + }; + decode_message(encoded(v1_header(&[])), BIG).unwrap(); + let text = refusal(v1_header(&[("recordKey", "second".into())])); + assert!(text.contains("duplicate field"), "{text}"); +} + +#[test] +fn the_schema_version_spelled_twice_is_refused_whichever_family_comes_first() { + for versions in [["cultnet.hello.v0", "cultnet.error.v0"], ["cultnet.error.v0", "cultnet.hello.v0"]] { + let entries = pairs(&[ + sv(versions[0]), + sv(versions[1]), + ("runtimeId", "rt".into()), + ("runtimeKind", "test".into()), + ("error", "e".into()), + ]); + decode_message(encoded(entries), BIG).unwrap_err(); + } +} + +#[test] +fn receive_errors_name_schema_never_values() { + let canary = "CANARY-5e1f"; + let hello = |extra: &[(&str, rmpv::Value)]| { + let mut entries = vec![sv("cultnet.hello.v0"), ("runtimeKind", "test".into())]; + entries.extend(extra.iter().cloned()); + pairs(&entries) + }; + let cases: Vec<(&str, Entries, &str)> = vec![ + ("unknown schema", pairs(&[sv("cultnet.bogus.v0"), ("messageId", "m".into())]), "cultnet.bogus.v0"), + ("missing runtimeId", hello(&[]), "runtimeId"), + ( + "mistyped bool", + hello(&[("runtimeId", "rt".into()), ("supportsSchemaCatalog", canary.into())]), + "cultnet.hello.v0", + ), + ( + "login without auth", + pairs(&[sv("cultnet.login.v0"), ("nonce", "n".into()), ("password", canary.into())]), + "auth", + ), + ( + "unrecognized error code", + pairs(&[sv("cultnet.error.v0"), ("error", "e".into()), ("code", canary.into())]), + "cultnet.error.v0", + ), + ( + "unknown payload encoding", + pairs(&[ + sv("cultnet.document_put_raw.v0"), + ("messageId", "m".into()), + ("document", wire_record(&[("payloadEncoding", canary.into())])), + ]), + "payloadEncoding", + ), + ("schemaVersion that is no schema name", pairs(&[sv(canary), ("messageId", "m".into())]), "schemaVersion"), + ]; + for (label, entries, names) in cases { + let text = refusal(entries); + assert!(text.contains(names), "{label}: {text}"); + assert!(!text.contains("CANARY") && !text.contains("canary"), "{label}: {text}"); + } + // A message with no schemaVersion at all says so. + assert!(refusal(pairs(&[("messageId", "m".into())])).contains("no schemaVersion")); +} From 020360e2385e3bf5736c0031993a66f1ed41ad4c Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 04:22:29 +0200 Subject: [PATCH 08/13] cultmesh-rs 0.2.0: its public functions take and return cultnet_rs::CultNetMessage, whose variants 0.2.0 removed Co-Authored-By: Claude Opus 5.5 --- packages/cultmesh-rs/Cargo.lock | 2 +- packages/cultmesh-rs/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/cultmesh-rs/Cargo.lock b/packages/cultmesh-rs/Cargo.lock index 42893bb7..8dfc33ed 100644 --- a/packages/cultmesh-rs/Cargo.lock +++ b/packages/cultmesh-rs/Cargo.lock @@ -200,7 +200,7 @@ dependencies = [ [[package]] name = "cultmesh-rs" -version = "0.1.0" +version = "0.2.0" dependencies = [ "anyhow", "cultcache-rs", diff --git a/packages/cultmesh-rs/Cargo.toml b/packages/cultmesh-rs/Cargo.toml index 820050c3..af75e419 100644 --- a/packages/cultmesh-rs/Cargo.toml +++ b/packages/cultmesh-rs/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cultmesh-rs" -version = "0.1.0" +version = "0.2.0" edition = "2024" license = "MIT" publish = false From d9d2ce4aea069e171c052c8441a0c47251e5ee5b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 04:30:57 +0200 Subject: [PATCH 09/13] cultnet-rs tests: the unknown-encoding case spells payloadEncoding once Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/tests/decode_door.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index 09dddb91..c421c040 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -622,7 +622,16 @@ fn receive_errors_name_schema_never_values() { pairs(&[ sv("cultnet.document_put_raw.v0"), ("messageId", "m".into()), - ("document", wire_record(&[("payloadEncoding", canary.into())])), + ( + "document", + rmpv::Value::Map(pairs(&[ + ("schemaId", "s".into()), + ("recordKey", "k".into()), + ("storedAt", "2026-10-16T00:00:00Z".into()), + ("payloadEncoding", canary.into()), + ("payload", rmpv::Value::Binary(vec![0x90])), + ])), + ), ]), "payloadEncoding", ), From 7ab3092550cc0d79783443fc99ecbec20377874f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 04:31:34 +0200 Subject: [PATCH 10/13] cultcache-rs tests: a field name is kept only when it looks like one Co-Authored-By: Claude Opus 5.5 --- packages/cultcache-rs/tests/union_parity.rs | 37 +++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/packages/cultcache-rs/tests/union_parity.rs b/packages/cultcache-rs/tests/union_parity.rs index 0caa8ee9..6b120791 100644 --- a/packages/cultcache-rs/tests/union_parity.rs +++ b/packages/cultcache-rs/tests/union_parity.rs @@ -323,6 +323,43 @@ fn a_missing_or_duplicate_field_is_named_and_a_mistyped_value_is_not() { assert!(message.contains("MessagePack type"), "{message}"); } +// A decoder whose error text is the string it reads, standing for any serde message. +struct Raises; + +impl<'de> serde::Deserialize<'de> for Raises { + fn deserialize>(deserializer: D) -> Result { + let message = ::deserialize(deserializer)?; + Err(serde::de::Error::custom(message)) + } +} + +#[test] +fn a_field_name_is_kept_only_when_it_looks_like_a_field_name() { + let refusal = |message: &str| { + let bytes = rmp_serde::to_vec(message).unwrap(); + match cultcache_rs::decode_payload::(&bytes) { + Ok(_) => panic!("Raises never decodes"), + Err(error) => error.to_string(), + } + }; + assert_eq!(refusal("missing field `a_1`"), "missing field `a_1`"); + assert_eq!(refusal("duplicate field `Z9`"), "duplicate field `Z9`"); + let long = format!("missing field `{}`", "a".repeat(64)); + assert_eq!(refusal(&long), long); + for message in [ + "missing field `has space CANARY`".to_string(), + "missing field `x-CANARY`".to_string(), + "missing field ``".to_string(), + format!("duplicate field `{}`", "a".repeat(65)), + "unknown field `CANARY`, expected `a`".to_string(), + ] { + let text = refusal(&message); + assert_eq!(text, "a value does not fit the store format", "{message}"); + } + assert_eq!(refusal("unknown variant `CANARY`, expected `a`"), "a tag names no variant this type knows"); + assert!(!refusal("invalid type: string \"CANARY\", expected a bool").contains("CANARY")); +} + #[test] fn the_key_decides_the_arm() { // Tool and Part carry different slots under different keys in the two unions. From fd944072b6ada8585ae0d1adcbc0767448bef412 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 04:33:27 +0200 Subject: [PATCH 11/13] cultnet-rs: three more refusals stop quoting received values; the networking descriptor's repeated key is pinned Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/src/contracts.rs | 6 ++--- packages/cultnet-rs/tests/decode_door.rs | 29 ++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index 0a8f657a..bb1b1e57 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -1030,7 +1030,7 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result Err(anyhow!( - "Unsupported raw cultnet.schema.v0 schemaVersion {schema_version}" + "Unsupported raw cultnet.schema.v0 message" )), } } @@ -1489,7 +1489,7 @@ fn parse_gamecult_networking_message(input: &rmpv::Value) -> Result Err(anyhow!( - "Unsupported gamecult.networking.v0 union tag {tag}" + "Unsupported gamecult.networking.v0 union tag" )), } } @@ -1721,7 +1721,7 @@ fn require_legacy_schema_descriptor( "shared_contract" => CultNetSchemaKind::SharedContract, _ => { return Err(anyhow!( - "{field_name}.Kind has unsupported schema kind {kind}" + "{field_name}.Kind is not a supported schema kind" )); } }; diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index c421c040..03330beb 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -645,3 +645,32 @@ fn receive_errors_name_schema_never_values() { // A message with no schemaVersion at all says so. assert!(refusal(pairs(&[("messageId", "m".into())])).contains("no schemaVersion")); } + +#[test] +fn a_networking_schema_descriptor_with_a_repeated_key_is_refused_and_a_bad_kind_is_not_quoted() { + let canary = "CANARY-8c2d"; + let descriptor = |extra: &[(&str, rmpv::Value)]| { + let mut entries = pairs(&[("schemaId", "s".into()), ("kind", "wire_message".into()), ("contentHash", "h".into())]); + entries.extend(pairs(extra)); + rmpv::Value::Map(entries) + }; + let response = |descriptor: rmpv::Value| { + let union = rmpv::Value::Array(vec![ + 8.into(), + rmpv::Value::Array(vec!["m".into(), rmpv::Value::Array(vec![descriptor])]), + ]); + rmp_serde::to_vec(&union).unwrap() + }; + let decode = |bytes: Vec| { + on_stack(BIG, move || { + decode_cultnet_message_from_slice(&bytes, CultNetWireContract::GameCultNetworkingV0) + .map_err(|error| format!("{error:#}")) + }) + }; + decode(response(descriptor(&[]))).expect("the control decodes"); + let text = decode(response(descriptor(&[("schemaId", "t".into())]))).unwrap_err(); + assert!(text.contains("twice"), "{text}"); + let bad_kind = rmpv::Value::Map(pairs(&[("schemaId", "s".into()), ("kind", canary.into()), ("contentHash", "h".into())])); + let text = decode(response(bad_kind)).unwrap_err(); + assert!(text.contains("Kind") && !text.contains("CANARY"), "{text}"); +} From a83a7153cdd887fc423eea5a04502fb0118703dd Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 05:03:16 +0200 Subject: [PATCH 12/13] cultnet-rs tests: pin which schemaVersions a refusal may name Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/tests/decode_door.rs | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index 03330beb..8a25ef02 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -674,3 +674,24 @@ fn a_networking_schema_descriptor_with_a_repeated_key_is_refused_and_a_bad_kind_ let text = decode(response(bad_kind)).unwrap_err(); assert!(text.contains("Kind") && !text.contains("CANARY"), "{text}"); } + +#[test] +fn a_schema_version_that_is_not_shaped_like_a_schema_name_is_described_by_length() { + let long = format!("cultnet.{}.v0", "a".repeat(100)); + for version in [ + "CANARY-5e1f", + "1canary.v0", + "cultnet.CANARY.v0", + "cultnet..v0", + "canary", + "cultnet.canary-x.v0", + long.as_str(), + ] { + let text = refusal(pairs(&[sv(version), ("messageId", "m".into())])); + assert!(text.contains(&format!("is not a schema name ({} bytes)", version.len())), "{version}: {text}"); + assert!(!text.contains(version), "{version}: {text}"); + } + // Shaped like a schema name, it is named (the sender spelled a schema, not data). + let text = refusal(pairs(&[sv("cultnet.x_y9.v10"), ("messageId", "m".into())])); + assert!(text.contains("cultnet.x_y9.v10 refused"), "{text}"); +} From 0daa2f10179e55bcc243448001f00f856478a06b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 16 Oct 2026 05:55:05 +0200 Subject: [PATCH 13/13] cultnet-rs: kind-array refusal no longer quotes the value; README says delete carries ids Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/README.md | 5 +++-- packages/cultnet-rs/src/contracts.rs | 2 +- packages/cultnet-rs/tests/decode_door.rs | 20 ++++++++++++++++++++ 3 files changed, 24 insertions(+), 3 deletions(-) diff --git a/packages/cultnet-rs/README.md b/packages/cultnet-rs/README.md index 3c7488ba..48ac2dd3 100644 --- a/packages/cultnet-rs/README.md +++ b/packages/cultnet-rs/README.md @@ -11,8 +11,9 @@ The contract is intentionally boring: - `gamecult.networking.v0` maps the legacy C# union shape explicitly. - schema discovery uses explicit catalog request/response messages; no inbound auto-detect sludge. -- put/delete/snapshot travel only as raw messages that carry the canonical - MessagePack payload bytes; the receiver decodes them through registered types. +- put and snapshot travel only as raw messages that carry the canonical + MessagePack payload bytes; the receiver decodes them through registered + types. Delete carries ids. - hello messages advertise `supportedMutationContracts`, so callers discover which document types are read-only, which accept typed intents, which authority owns the mutation, and which receipt documents prove the outcome. diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index bb1b1e57..e4f67164 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -1658,7 +1658,7 @@ fn require_legacy_optional_schema_kind_array( "wire_message" => Ok(CultNetSchemaKind::WireMessage), "document_payload" => Ok(CultNetSchemaKind::DocumentPayload), "shared_contract" => Ok(CultNetSchemaKind::SharedContract), - _ => Err(anyhow!("{field_name} has unsupported schema kind {value}")), + _ => Err(anyhow!("{field_name} has an unsupported schema kind")), }) .collect::>>() .map(Some) diff --git a/packages/cultnet-rs/tests/decode_door.rs b/packages/cultnet-rs/tests/decode_door.rs index 8a25ef02..fd5bab6a 100644 --- a/packages/cultnet-rs/tests/decode_door.rs +++ b/packages/cultnet-rs/tests/decode_door.rs @@ -675,6 +675,26 @@ fn a_networking_schema_descriptor_with_a_repeated_key_is_refused_and_a_bad_kind_ assert!(text.contains("Kind") && !text.contains("CANARY"), "{text}"); } +#[test] +fn a_networking_catalog_request_with_a_bad_kind_element_is_refused_without_quoting_it() { + let union = rmpv::Value::Array(vec![ + 7.into(), + rmpv::Value::Array(vec![ + "m".into(), + false.into(), + rmpv::Value::Nil, + rmpv::Value::Array(vec!["CANARY-s2x".into()]), + ]), + ]); + let bytes = rmp_serde::to_vec(&union).unwrap(); + let text = on_stack(BIG, move || { + decode_cultnet_message_from_slice(&bytes, CultNetWireContract::GameCultNetworkingV0) + .map_err(|error| format!("{error:#}")) + }) + .unwrap_err(); + assert!(text.contains("Kinds") && !text.contains("CANARY"), "{text}"); +} + #[test] fn a_schema_version_that_is_not_shaped_like_a_schema_name_is_described_by_length() { let long = format!("cultnet.{}.v0", "a".repeat(100));