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 <noreply@anthropic.com>
This commit is contained in:
parent
c0a2b890f6
commit
377c24b8cd
4 changed files with 365 additions and 83 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<Vec<String>>,
|
||||
},
|
||||
#[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<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,
|
||||
|
|
@ -487,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,
|
||||
|
|
@ -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::<CultNetMessage>(bytes)?,
|
||||
) => 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)
|
||||
|
|
@ -911,24 +918,52 @@ fn schema_version_from_wire_value(input: &rmpv::Value) -> Result<Option<&str>> {
|
|||
}))
|
||||
}
|
||||
|
||||
/// 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<Vec<RawDocumentHeader>>,
|
||||
#[serde(default)]
|
||||
documents: Option<Vec<SelectionDocumentRecord>>,
|
||||
#[serde(default)]
|
||||
edges: Option<Vec<Edge>>,
|
||||
/// 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<CultNetMessage> {
|
||||
/// 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)| {
|
||||
|
|
@ -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<T: Serialize>(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<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"))
|
||||
}
|
||||
|
||||
fn require_legacy_i32(value: Option<&rmpv::Value>, field_name: &str) -> Result<i32> {
|
||||
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<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")
|
||||
|
|
@ -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)| {
|
||||
|
|
|
|||
|
|
@ -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::<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<()>
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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<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", 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"));
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue