GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge hands/cultnet-rs-door

Browse source 
cultnet-rs 0.2.0: every received MessagePack byte decodes once through
cultcache-rs's door; the dynamic document lane is parked at
parked/cultnet-rs-dynamic-document-lane; one decoder per message family;
duplicate keys refused; refusals name schema and fields, never values;
cultmesh-rs 0.2.0. Cut cultnet-rs-door r2, held by Soul s1-s3.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-16 06:07:12 +02:00
commit 8df4528707
17 changed files with 976 additions and 500 deletions
Download patch file Download diff file Expand all files Collapse all files

27
packages/cultcache-rs/src/lib.rs
View file

@ -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.

67
packages/cultcache-rs/tests/union_parity.rs
View file

@ -293,6 +293,73 @@ 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::<Probe>(&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::<Probe>(&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::<Probe>(&mistyped).unwrap_err().to_string();
assert!(!message.contains("CANARY"), "{message}");
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<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let message = <String as serde::Deserialize>::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::<Raises>(&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.

4
packages/cultmesh-rs/Cargo.lock generated
View file

@ -200,7 +200,7 @@ dependencies = [
[[package]]
name = "cultmesh-rs"
version = "0.1.0"
version = "0.2.0"
dependencies = [
"anyhow",
"cultcache-rs",
@ -213,7 +213,7 @@ dependencies = [
[[package]]
name = "cultnet-rs"
version = "0.1.0"
version = "0.2.0"
dependencies = [
"aes-gcm",
"anyhow",

2
packages/cultmesh-rs/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "cultmesh-rs"
version = "0.1.0"
version = "0.2.0"
edition = "2024"
license = "MIT"
publish = false

2
packages/cultnet-rs/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "cultnet-rs"
version = "0.1.0"
version = "0.2.0"
edition = "2024"
publish = false
license = "MIT"

9
packages/cultnet-rs/README.md
View file

@ -11,9 +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.
- document put/delete/snapshot messages move typed CultCache entries.
- raw document put/snapshot messages preserve canonical MessagePack payload
bytes for bit-compatible neighbors.
- 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.
@ -41,7 +41,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
@ -164,7 +163,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

19
packages/cultnet-rs/contracts/cultnet/cultnet.document-put.schema.json
View file

@ -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"
}
}
}

22
packages/cultnet-rs/contracts/cultnet/cultnet.snapshot-response.schema.json
View file

@ -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"
}
}
}
}

309
packages/cultnet-rs/src/contracts.rs
View file

@ -1,10 +1,11 @@
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;
use serde::Serialize;
use serde_json::Value;
use crate::{Edge, RawDocumentHeader, Selection, SelectionDocumentRecord};
@ -30,23 +31,6 @@ pub enum CultNetRawPayloadEncoding {
Messagepack,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultNetDocumentRecord<TPayload = Value> {
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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source_agent_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source_role: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tags: Option<Vec<String>>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum CultNetDocumentOperation {
@ -284,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")),
}
}
}
@ -353,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 `serde_json::Value` path other tooling may still reach for.
// 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,
@ -370,17 +354,13 @@ 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<Value>,
},
#[serde(rename = "cultnet.document_delete.v0", rename_all = "camelCase")]
DocumentDelete {
message_id: String,
schema_id: String,
record_key: String,
},
#[serde(skip_deserializing)]
#[serde(rename = "cultnet.document_put_raw.v0", rename_all = "camelCase")]
DocumentPutRaw {
message_id: String,
@ -394,11 +374,7 @@ pub enum CultNetMessage {
#[serde(default, skip_serializing_if = "Option::is_none")]
record_keys: Option<Vec<String>>,
},
#[serde(rename = "cultnet.snapshot_response.v0", rename_all = "camelCase")]
SnapshotResponse {
message_id: String,
documents: Vec<CultNetDocumentRecord<Value>>,
},
#[serde(skip_deserializing)]
#[serde(rename = "cultnet.snapshot_response_raw.v0", rename_all = "camelCase")]
SnapshotResponseRaw {
message_id: String,
@ -430,10 +406,10 @@ pub enum CultNetMessage {
#[serde(default, skip_serializing_if = "Option::is_none")]
supported_body_transports: Option<Vec<String>>,
},
/// `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,
@ -515,6 +491,7 @@ pub enum CultNetMessage {
source_runtime_id: Option<String>,
},
/// `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,
@ -524,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,
@ -544,33 +522,6 @@ fn true_default() -> bool {
true
}
pub fn parse_cultnet_message(
wire_value: &rmpv::Value,
contract: CultNetWireContract,
) -> Result<CultNetMessage> {
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 +535,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 +556,25 @@ pub fn decode_cultnet_message_from_slice(
contract: CultNetWireContract,
) -> Result<CultNetMessage> {
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"
| "cultmesh.content_chunk_request.v1"
| "cultmesh.content_chunk_response.v1"
| "cultnet.error.v0",
) => parse_raw_cultnet_schema_message(&wire_value)
.with_context(|| received_version_label(&wire_value))?,
_ => cultcache_rs::decode_payload::<CultNetMessage>(bytes)
.with_context(|| received_version_label(&wire_value))?,
};
validate_message(&message)?;
Ok(message)
}
CultNetWireContract::GameCultNetworkingV0 => parse_gamecult_networking_message(&wire_value),
}
}
fn validate_message(message: &CultNetMessage) -> Result<()> {
@ -669,13 +638,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 +663,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 +827,6 @@ fn validate_mutation_contract(contract: &CultNetDocumentMutationContract) -> Res
Ok(())
}
fn validate_document_record(document: &CultNetDocumentRecord<Value>) -> 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,10 +918,52 @@ fn schema_version_from_wire_value(input: &rmpv::Value) -> Result<Option<&str>> {
}))
}
/// 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
}
/// 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<CultNetMessage> {
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)| {
@ -1011,69 +995,6 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result<CultNetMessag
documents,
})
}
"cultnet.snapshot_response_raw.v1" => {
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::<RawDocumentHeader>(Some(value), &format!("headers[{index}]"))
})
.collect::<Result<Vec<_>>>()
})
.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::<SelectionDocumentRecord>(
Some(value),
&format!("documents[{index}]"),
)
})
.collect::<Result<Vec<_>>>()
})
.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::<Edge>(Some(value), &format!("edges[{index}]"))
})
.collect::<Result<Vec<_>>>()
})
.transpose()?;
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")?,
@ -1109,7 +1030,7 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result<CultNetMessag
})
}
_ => Err(anyhow!(
"Unsupported raw cultnet.schema.v0 schemaVersion {schema_version}"
"Unsupported raw cultnet.schema.v0 message"
)),
}
}
@ -1127,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()
@ -1368,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)| {
@ -1390,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"
));
}
},
@ -1469,41 +1392,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<T: serde::de::DeserializeOwned>(
value: Option<&rmpv::Value>,
field_name: &str,
) -> Result<T> {
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: the enum's derive reads them from the original bytes.
fn encode_typed<T: Serialize>(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")
}
fn require_legacy_u32(value: Option<&rmpv::Value>, field_name: &str) -> Result<u32> {
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"))
cultcache_rs::decode_payload(&bytes).expect("named-map MessagePack bytes always decode into rmpv::Value")
}
fn require_legacy_i32(value: Option<&rmpv::Value>, field_name: &str) -> Result<i32> {
@ -1513,28 +1411,6 @@ fn require_legacy_i32(value: Option<&rmpv::Value>, field_name: &str) -> Result<i
i32::try_from(number).map_err(|_| anyhow!("{field_name} does not fit in i32"))
}
fn require_legacy_u64(value: Option<&rmpv::Value>, field_name: &str) -> Result<u64> {
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<Option<i64>> {
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<CultNetMessage> {
let items = input.as_array().ok_or_else(|| {
anyhow!("gamecult.networking.v0 messages must be a 2-element union array")
@ -1613,7 +1489,7 @@ fn parse_gamecult_networking_message(input: &rmpv::Value) -> Result<CultNetMessa
)?,
}),
_ => Err(anyhow!(
"Unsupported gamecult.networking.v0 union tag {tag}"
"Unsupported gamecult.networking.v0 union tag"
)),
}
}
@ -1782,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::<Result<Vec<_>>>()
.map(Some)
@ -1811,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)| {
@ -1844,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"
));
}
};

2
packages/cultnet-rs/src/idunn_lifecycle_brake.rs
View file

@ -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");

9
packages/cultnet-rs/src/provider_health_contracts.rs
View file

@ -123,7 +123,7 @@ impl IdunnSignedDaemonHealthRecord {
pub fn decode_canonical_signed_payload(payload: &[u8]) -> Result<(Self, Vec<u8>)> {
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");
@ -643,6 +643,11 @@ mod tests {
rmp_serde::from_slice::<IdunnSignedDaemonHealthRecord>(&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::<GameCultServiceTrustAnchorRecord>(&rmp_serde::to_vec(
&anchor

195
packages/cultnet-rs/src/replication.rs
View file

@ -1,4 +1,3 @@
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use cultcache_rs::CultCache;
@ -18,7 +17,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 +150,6 @@ impl CultNetDocumentRegistry {
.collect()
}
pub fn create_document_put_message<T>(
&self,
message_id: impl Into<String>,
record_key: impl Into<String>,
value: &T,
options: CultNetDocumentPutOptions,
) -> Result<CultNetMessage>
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<String>,
@ -230,38 +201,6 @@ impl CultNetDocumentRegistry {
})
}
pub fn create_snapshot_response(
&self,
cache: &CultCache,
message_id: impl Into<String>,
schema_ids: Option<&[String]>,
record_keys: Option<&[String]>,
) -> Result<CultNetMessage> {
let requested_schema_ids = schema_ids.map(|items| items.iter().collect::<BTreeSet<_>>());
let requested_record_keys = record_keys.map(|items| items.iter().collect::<BTreeSet<_>>());
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 +233,6 @@ impl CultNetDocumentRegistry {
})
}
pub fn apply_document_put_message<T>(
&self,
cache: &mut CultCache,
message: &CultNetMessage,
) -> Result<T>
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<T>(
&self,
cache: &mut CultCache,
@ -381,68 +290,6 @@ impl CultNetDocumentRegistry {
})
}
pub fn apply_snapshot_response<T>(
&self,
cache: &mut CultCache,
response: &CultNetMessage,
) -> Result<Vec<T>>
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::<T>(
cache,
&CultNetMessage::DocumentPut {
message_id: "snapshot-apply".to_string(),
document: document.clone(),
},
)?);
}
Ok(applied)
}
pub fn sync_document_from_snapshot_response<T>(
&self,
cache: &mut CultCache,
response: &CultNetMessage,
record_key: &str,
) -> Result<T>
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::<T>(
cache,
&CultNetMessage::DocumentPut {
message_id: "snapshot-sync".to_string(),
document: document.clone(),
},
)
}
pub fn apply_raw_snapshot_response<T>(
&self,
cache: &mut CultCache,
@ -517,29 +364,6 @@ impl CultNetDocumentRegistry {
CultNetReactiveDocument::new(self.clone(), cache, record_key, options)
}
fn document_record_from_envelope(
&self,
envelope: &CultCacheEnvelope,
) -> Result<CultNetDocumentRecord<Value>> {
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 +556,6 @@ where
flush_reactive_document::<T>(&self.cache, &self.current, &self.state, &self.record_key)
}
pub fn apply_document_put_message(&self, message: &CultNetMessage) -> Result<T> {
let canonical = {
let mut cache = self
.cache
.lock()
.map_err(|_| anyhow!("CultCache mutex poisoned"))?;
self.registry
.apply_document_put_message::<T>(&mut cache, message)?
};
self.apply_canonical_value(canonical)
}
pub fn apply_raw_document_put_message(&self, message: &CultNetMessage) -> Result<T> {
let canonical = {
let mut cache = self
@ -1002,13 +814,6 @@ fn numeric_delta(predicted: &Value, canonical: &Value) -> Option<Value> {
serde_json::Number::from_f64(predicted - canonical).map(Value::Number)
}
fn round_trip_typed<T>(value: &T) -> Result<T>
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)
}

10
packages/cultnet-rs/src/runtime_authority_contracts.rs
View file

@ -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");
}

14
packages/cultnet-rs/src/schema_discovery.rs
View file

@ -284,13 +284,6 @@ pub fn builtin_schema_registry() -> Result<CultNetSchemaRegistry> {
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<CultNetSchemaRegistry> {
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,

38
packages/cultnet-rs/src/service_identity.rs
View file

@ -285,11 +285,10 @@ pub fn open_service_identity_credential_reader<P: ServiceIdentityProfile>(
pub fn open_service_identity_credential_bytes<P: ServiceIdentityProfile>(
bytes: &[u8],
) -> Result<ServiceIdentitySigner<P>> {
let mut decoder = rmp_serde::Deserializer::new(std::io::Cursor::new(bytes));
let entries = Vec::<CultCacheEnvelope>::deserialize(&mut decoder)
let entries = cultcache_rs::decode_payload::<Vec<CultCacheEnvelope>>(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::<P>(&entries)
}
@ -769,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::<IdunnServiceIdentity>(&path)?;
let bytes = std::fs::read(path)?;
open_service_identity_credential_bytes::<IdunnServiceIdentity>(&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<CultCacheEnvelope> = 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<CultCacheEnvelope> = rmp_serde::from_slice(&forged)?;
assert_eq!(reread, envelopes, "the forgery decodes to identical entries");
let error = open_service_identity_credential_bytes::<IdunnServiceIdentity>(&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<()>
{

30
packages/cultnet-rs/tests/cultnet.rs
View file

@ -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::<GhostlightAgentStateFixture>()?;
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::<GhostlightAgentStateFixture>()?;
let error = registry
.apply_raw_snapshot_response::<GhostlightAgentStateFixture>(&mut target, &past)
.unwrap_err();
assert_eq!(
error.downcast_ref::<cultcache_rs::PayloadTooDeep>(),
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::<GhostlightAgentStateFixture>()?;
target.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&target_store))?;
target.pull_all_backing_stores()?;
let applied =
registry.apply_snapshot_response::<GhostlightAgentStateFixture>(&mut target, &snapshot)?;
let synced = registry.sync_document_from_snapshot_response::<GhostlightAgentStateFixture>(
registry.apply_raw_snapshot_response::<GhostlightAgentStateFixture>(&mut target, &snapshot)?;
let synced = registry.sync_raw_document_from_snapshot_response::<GhostlightAgentStateFixture>(
&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::<GhostlightAgentStateFixture>(&mut target, &direct_put)?;
.apply_raw_document_put_message::<GhostlightAgentStateFixture>(&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();

717
packages/cultnet-rs/tests/decode_door.rs Normal file
View file

@ -0,0 +1,717 @@
//! 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<Box<Wide32>>,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "cultnet.decode-door.wide", schema = "DecodeDoorWide")]
struct WideDocument {
#[cultcache(key = 0)]
chain: Wide32,
}
fn on_stack<R: Send + 'static>(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<u8> {
let mut bytes = vec![0x91; levels];
bytes.push(0xc0);
bytes
}
/// Swaps the one sentinel string in `bytes` for `with`.
fn splice(bytes: Vec<u8>, with: &[u8]) -> Vec<u8> {
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<u8> {
entries.push(("x".into(), "DEEP".into()));
rmp_serde::to_vec(&rmpv::Value::Map(entries)).unwrap()
}
fn hello_with(unknown: &[u8]) -> Vec<u8> {
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<u8>, stack: usize) -> Result<CultNetMessage> {
*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 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", || {
let error = decode_message(hello_with(&nested(MAX_PAYLOAD_DEPTH + 1)), SMALL).unwrap_err();
let refusal = error.downcast_ref::<PayloadTooDeep>().expect("typed refusal");
assert_eq!(refusal.limit, MAX_PAYLOAD_DEPTH);
});
}
fn raw_v1_bytes(header_unknown: Option<&[u8]>) -> Vec<u8> {
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<Box<Wide32>> = 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::<WideDocument>("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<u8> {
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<u8>) -> Result<WideDocument> {
*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::<WideDocument>()?;
registry().apply_raw_document_put_message::<WideDocument>(&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::<PayloadTooDeep>().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 names the schema the sender spelled, shaped like one, and nothing else of it.
assert!(format!("{error:#}").contains("cultnet.document_put.v0"), "{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::<IdunnServiceIdentity>(&path).unwrap();
let mut envelopes: rmpv::Value = rmp_serde::from_slice(&std::fs::read(path).unwrap()).unwrap();
let rmpv::Value::Array(items) = &mut envelopes else {
panic!("credential is an array");
};
// 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");
};
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));
let refused = on_stack(SMALL, move || {
open_service_identity_credential_bytes::<IdunnServiceIdentity>(&bytes).is_err()
});
assert!(refused, "{levels} levels deep");
}
});
}
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<u8> {
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<CultNetMessage> {
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::<CultNetMessage>(&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::<CultNetMessage>(&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",
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",
),
("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"));
}
#[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<u8>| {
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}");
}
#[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));
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}");
}