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

Add the content plane to cultnet-rs: chunk messages, pack, answer, verified fetch

Browse source 
ContentChunkRequest/Response join CultNetMessage on the raw path, byte
identical to the reference's field order. content.rs owns normalize_hash,
pack_content, validate_manifest, answer_content_chunk_request and
fetch_content, mirroring CultMeshCdn and CultMeshLegacyRudpContentServer.
The C# vector test is added here; the vector files and Rust tests follow.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 00:48:13 +02:00
commit 29c3e4c2b5
4 changed files with 744 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

365
packages/cultnet-rs/src/content.rs Normal file
View file

@ -0,0 +1,365 @@
//! CultMesh content plane: chunking, hashing, manifest shape and validation, answering one chunk
//! request, and verifying a fetched body. Mirrors `CultMeshCdn` (`CultMeshCdn.cs`) and
//! `CultMeshLegacyRudpContentServer` (`CultMeshContentSessions.cs`); the two chunk messages are
//! `CultNetMessage::ContentChunkRequest`/`ContentChunkResponse` in `contracts.rs`.
//!
//! This is the only owner of these rules in Rust. A consumer hands `fetch_content` a transport
//! closure and gets verified bytes; it never hashes, chunks or validates a manifest itself.
use std::collections::BTreeMap;
use anyhow::{Result, anyhow, bail};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::CultNetMessage;
/// One chunk of a packed body. `chunk_hash` is lowercase SHA-256 hex of `payload`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CultMeshCdnChunk {
pub chunk_hash: String,
pub payload: Vec<u8>,
}
/// `CultMeshCdnChunkRef`. Encodes as the reference's positional array under every encoder.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(from = "ChunkRefWire", into = "ChunkRefWire")]
pub struct CultMeshCdnChunkRef {
pub chunk_hash: String,
pub offset: i64,
pub size_bytes: i32,
pub record_key: String,
}
/// `CultMeshCdnArtifactManifest`. Encodes as the reference's positional array (keys 0-9).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(from = "ManifestWire", into = "ManifestWire")]
pub struct CultMeshCdnArtifactManifest {
pub artifact_id: String,
pub kind: String,
pub version: String,
pub content_hash: String,
pub size_bytes: i64,
pub mime_type: String,
pub created_at_utc: String,
pub chunks: Vec<CultMeshCdnChunkRef>,
pub tags: Vec<String>,
pub metadata: BTreeMap<String, String>,
}
// The tuple mirrors: `to_vec_named` would write a map for a named-field struct, but a tuple
// struct is an array under every serde encoder.
#[derive(Serialize, Deserialize)]
struct ChunkRefWire(String, i64, i32, String);
#[derive(Serialize, Deserialize)]
struct ManifestWire(
String,
String,
String,
String,
i64,
String,
String,
Vec<CultMeshCdnChunkRef>,
Vec<String>,
BTreeMap<String, String>,
);
impl From<CultMeshCdnChunkRef> for ChunkRefWire {
fn from(value: CultMeshCdnChunkRef) -> Self {
Self(value.chunk_hash, value.offset, value.size_bytes, value.record_key)
}
}
impl From<ChunkRefWire> for CultMeshCdnChunkRef {
fn from(value: ChunkRefWire) -> Self {
Self {
chunk_hash: value.0,
offset: value.1,
size_bytes: value.2,
record_key: value.3,
}
}
}
impl From<CultMeshCdnArtifactManifest> for ManifestWire {
fn from(value: CultMeshCdnArtifactManifest) -> Self {
Self(
value.artifact_id,
value.kind,
value.version,
value.content_hash,
value.size_bytes,
value.mime_type,
value.created_at_utc,
value.chunks,
value.tags,
value.metadata,
)
}
}
impl From<ManifestWire> for CultMeshCdnArtifactManifest {
fn from(value: ManifestWire) -> Self {
Self {
artifact_id: value.0,
kind: value.1,
version: value.2,
content_hash: value.3,
size_bytes: value.4,
mime_type: value.5,
created_at_utc: value.6,
chunks: value.7,
tags: value.8,
metadata: value.9,
}
}
}
fn sha256_hex(bytes: &[u8]) -> String {
format!("{:x}", Sha256::digest(bytes))
}
fn chunk_record_key(hash: &str) -> String {
format!("mesh:cdn:chunk:{hash}")
}
/// `CultMeshCdn.NormalizeHash`: an optional case-insensitive `sha256:` prefix is stripped, then the
/// rest is trimmed and lowercased. A blank hash is refused.
pub fn normalize_hash(hash: &str) -> Result<String> {
if hash.trim().is_empty() {
bail!("Hash must be non-empty.");
}
let stripped = match hash.get(..7) {
Some(prefix) if prefix.eq_ignore_ascii_case("sha256:") => &hash[7..],
_ => hash,
};
Ok(stripped.trim().to_lowercase())
}
/// `CultMeshCdn.PackArtifact`: splits `bytes` into content-addressed chunks and builds the manifest.
/// An empty body packs as one zero-length chunk, as the reference does.
pub fn pack_content(
artifact_id: &str,
kind: &str,
version: &str,
mime_type: &str,
created_at_utc: &str,
bytes: &[u8],
chunk_size: usize,
) -> Result<(CultMeshCdnArtifactManifest, Vec<CultMeshCdnChunk>)> {
if artifact_id.trim().is_empty() {
bail!("Artifact id must be non-empty.");
}
if chunk_size == 0 || i32::try_from(chunk_size).is_err() {
bail!("ChunkSizeBytes must be greater than zero and fit in an int.");
}
let pieces: Vec<&[u8]> = if bytes.is_empty() {
vec![bytes]
} else {
bytes.chunks(chunk_size).collect()
};
let mut offset = 0i64;
let mut chunks = Vec::with_capacity(pieces.len());
let mut refs = Vec::with_capacity(pieces.len());
for piece in pieces {
let chunk_hash = sha256_hex(piece);
refs.push(CultMeshCdnChunkRef {
record_key: chunk_record_key(&chunk_hash),
chunk_hash: chunk_hash.clone(),
offset,
// A piece is at most `chunk_size` bytes, which fits an i32 (checked above).
size_bytes: piece.len() as i32,
});
offset += piece.len() as i64;
chunks.push(CultMeshCdnChunk {
chunk_hash,
payload: piece.to_vec(),
});
}
let manifest = CultMeshCdnArtifactManifest {
artifact_id: artifact_id.to_string(),
kind: non_blank_or(kind, "asset"),
version: version.to_string(),
content_hash: sha256_hex(bytes),
size_bytes: bytes.len() as i64,
mime_type: non_blank_or(mime_type, "application/octet-stream"),
created_at_utc: created_at_utc.to_string(),
chunks: refs,
tags: Vec::new(),
metadata: BTreeMap::new(),
};
validate_manifest(&manifest)?;
Ok((manifest, chunks))
}
fn non_blank_or(value: &str, default: &str) -> String {
if value.trim().is_empty() {
default.to_string()
} else {
value.to_string()
}
}
/// The chunk list in offset order, as the reference reads it (`OrderBy(chunk => chunk.Offset)`).
fn in_offset_order(manifest: &CultMeshCdnArtifactManifest) -> Vec<&CultMeshCdnChunkRef> {
let mut refs: Vec<&CultMeshCdnChunkRef> = manifest.chunks.iter().collect();
refs.sort_by_key(|chunk| chunk.offset);
refs
}
/// `CultMeshCdn.ValidateManifestShape`: non-negative size, chunks contiguous from offset 0 once
/// sorted by offset, non-negative chunk sizes, and chunk sizes summing to the manifest size.
pub fn validate_manifest(manifest: &CultMeshCdnArtifactManifest) -> Result<()> {
if manifest.size_bytes < 0 {
bail!("CDN artifact manifest has a negative size.");
}
let mut expected_offset = 0i64;
for chunk in in_offset_order(manifest) {
if chunk.offset != expected_offset {
bail!("CDN artifact chunks are not contiguous.");
}
if chunk.size_bytes < 0 {
bail!("CDN artifact chunk has a negative size.");
}
expected_offset = expected_offset
.checked_add(i64::from(chunk.size_bytes))
.ok_or_else(|| anyhow!("CDN artifact chunk sizes overflow."))?;
}
if expected_offset != manifest.size_bytes {
bail!("CDN artifact chunk sizes do not sum to the manifest size.");
}
Ok(())
}
/// `CultMeshCdn.ValidateChunk`: the chunk's size and full SHA-256 must be what the reference and
/// the chunk's own claim say. `ref_hash` is already normalized.
fn validate_chunk(
ref_hash: &str,
ref_size: i32,
claimed_hash: &str,
claimed_size: i64,
payload: &[u8],
) -> Result<()> {
if i64::try_from(payload.len()).ok() != Some(i64::from(ref_size)) || claimed_size != i64::from(ref_size) {
bail!("CDN artifact chunk payload metadata does not match its manifest reference.");
}
let actual = sha256_hex(payload);
if ref_hash != actual || normalize_hash(claimed_hash)? != actual {
bail!("CDN artifact chunk hash does not match its payload.");
}
Ok(())
}
/// `CultMeshLegacyRudpContentServer.HandleAsync`, in its validation order: message id, hash
/// normalisation, record-key agreement with `mesh:cdn:chunk:<hash>`, lookup, then size and hash of
/// what is served. A failure is `found: false` with an empty payload and the reference's
/// `<ExceptionKind>: <message>` in `error`; it is never silence. `lookup` takes the normalized hash.
pub fn answer_content_chunk_request<'a>(
request: &CultNetMessage,
lookup: impl Fn(&str) -> Option<&'a [u8]>,
) -> CultNetMessage {
let CultNetMessage::ContentChunkRequest {
message_id,
chunk_hash,
record_key,
expected_size_bytes,
} = request
else {
return CultNetMessage::Error {
error: "not a cultmesh.content_chunk_request.v1 message".to_string(),
code: None,
details: None,
};
};
let served = (|| -> std::result::Result<(String, Vec<u8>), String> {
if message_id.trim().is_empty() {
return Err("InvalidDataException: Content chunk request requires a message identity.".to_string());
}
let hash = normalize_hash(chunk_hash)
.map_err(|_| "ArgumentException: Hash must be non-empty. (Parameter 'ChunkHash')".to_string())?;
if !record_key.trim().is_empty() && *record_key != chunk_record_key(&hash) {
return Err("InvalidDataException: Content chunk record key disagrees with its content hash.".to_string());
}
let payload = lookup(&hash).ok_or_else(|| "FileNotFoundException: Content chunk is not available.".to_string())?;
validate_chunk(&hash, *expected_size_bytes, &hash, payload.len() as i64, payload)
.map_err(|error| format!("InvalidDataException: {error}"))?;
Ok((hash, payload.to_vec()))
})();
match served {
Ok((hash, payload)) => CultNetMessage::ContentChunkResponse {
message_id: message_id.clone(),
found: true,
chunk_hash: hash,
size_bytes: *expected_size_bytes,
payload,
error: String::new(),
},
Err(error) => CultNetMessage::ContentChunkResponse {
message_id: message_id.clone(),
found: false,
chunk_hash: chunk_hash.clone(),
size_bytes: 0,
payload: Vec::new(),
error,
},
}
}
/// The client half: fetches every chunk of `manifest` through `ask`, one in flight, in offset
/// order, and returns the body only after each chunk and then the whole body verify. `max_bytes`
/// is checked against the manifest before the first request. `ask` sends a request on the
/// caller's session and returns the answer; it owns transport, this function owns verification.
pub fn fetch_content(
manifest: &CultMeshCdnArtifactManifest,
max_bytes: u64,
mut ask: impl FnMut(CultNetMessage) -> Result<CultNetMessage>,
) -> Result<Vec<u8>> {
validate_manifest(manifest)?;
if manifest.size_bytes as u64 > max_bytes {
bail!(
"CDN artifact of {} bytes exceeds the {max_bytes}-byte cap.",
manifest.size_bytes
);
}
let mut body = Vec::new();
for chunk in in_offset_order(manifest) {
let hash = normalize_hash(&chunk.chunk_hash)?;
let message_id = uuid::Uuid::new_v4().simple().to_string();
let record_key = if chunk.record_key.trim().is_empty() {
chunk_record_key(&hash)
} else {
chunk.record_key.clone()
};
let answer = ask(CultNetMessage::ContentChunkRequest {
message_id: message_id.clone(),
chunk_hash: hash.clone(),
record_key,
expected_size_bytes: chunk.size_bytes,
})?;
let CultNetMessage::ContentChunkResponse {
message_id: answered,
found,
chunk_hash,
size_bytes,
payload,
error,
} = answer
else {
bail!("chunk '{hash}' was answered with something other than a content chunk response");
};
if answered != message_id {
bail!("chunk '{hash}' was answered for message '{answered}', not '{message_id}'");
}
if !found {
bail!("content provider rejected chunk '{hash}': {error}");
}
validate_chunk(&hash, chunk.size_bytes, &chunk_hash, i64::from(size_bytes), &payload)?;
body.extend_from_slice(&payload);
}
if sha256_hex(&body) != normalize_hash(&manifest.content_hash)? {
bail!("CDN artifact content hash does not match its manifest.");
}
Ok(body)
}

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

@ -508,6 +508,26 @@ pub enum CultNetMessage {
#[serde(default, skip_serializing_if = "Option::is_none")]
source_runtime_id: Option<String>,
},
/// `cultmesh.content_chunk_request.v1`. Mirrors `CultMeshContentChunkRequestMessage`; raw path.
#[serde(rename = "cultmesh.content_chunk_request.v1", rename_all = "camelCase")]
ContentChunkRequest {
message_id: String,
chunk_hash: String,
record_key: String,
expected_size_bytes: i32,
},
/// `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(rename = "cultmesh.content_chunk_response.v1", rename_all = "camelCase")]
ContentChunkResponse {
message_id: String,
found: bool,
chunk_hash: String,
size_bytes: i32,
#[serde(with = "serde_bytes")]
payload: Vec<u8>,
error: String,
},
}
fn messagepack_base64_encoding() -> String {
@ -529,6 +549,8 @@ pub fn parse_cultnet_message(
"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)?,
_ => {
@ -553,6 +575,8 @@ pub fn encode_cultnet_message_for_wire(
CultNetMessage::DocumentPutRaw { .. }
| CultNetMessage::SnapshotResponseRaw { .. }
| CultNetMessage::SnapshotResponseRawV1 { .. }
| CultNetMessage::ContentChunkRequest { .. }
| CultNetMessage::ContentChunkResponse { .. }
| CultNetMessage::Error { .. } => encode_raw_cultnet_schema_message(message),
_ => Ok(rmp_serde::from_slice(&rmp_serde::to_vec(
&serde_json::to_value(message)?,
@ -777,6 +801,40 @@ fn validate_message(message: &CultNetMessage) -> Result<()> {
validate_shard_descriptor(shard)?;
}
}
CultNetMessage::ContentChunkRequest {
message_id,
chunk_hash,
record_key: _,
expected_size_bytes,
} => {
require_non_empty(message_id, "messageId")?;
if crate::normalize_hash(chunk_hash)?.is_empty() {
return Err(anyhow!("CultNet field chunkHash must be non-empty"));
}
if *expected_size_bytes < 0 {
return Err(anyhow!("CultNet field expectedSizeBytes must not be negative"));
}
}
CultNetMessage::ContentChunkResponse {
message_id,
found,
chunk_hash: _,
size_bytes,
payload,
error,
} => {
require_non_empty(message_id, "messageId")?;
if *found {
if i64::try_from(payload.len()).ok() != Some(i64::from(*size_bytes)) {
return Err(anyhow!("CultNet field payload length must equal sizeBytes"));
}
if !error.is_empty() {
return Err(anyhow!("CultNet field error must be empty when found"));
}
} else if !payload.is_empty() {
return Err(anyhow!("CultNet field payload must be empty when not found"));
}
}
CultNetMessage::OperationRequest {
message_id,
service_id,
@ -1044,6 +1102,23 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result<CultNetMessag
)?,
})
}
"cultmesh.content_chunk_request.v1" => Ok(CultNetMessage::ContentChunkRequest {
message_id: require_legacy_string(get("messageId"), "messageId")?,
chunk_hash: require_legacy_string(get("chunkHash"), "chunkHash")?,
record_key: require_legacy_string(get("recordKey"), "recordKey")?,
expected_size_bytes: require_legacy_i32(get("expectedSizeBytes"), "expectedSizeBytes")?,
}),
"cultmesh.content_chunk_response.v1" => Ok(CultNetMessage::ContentChunkResponse {
message_id: require_legacy_string(get("messageId"), "messageId")?,
found: require_legacy_bool(get("found"), "found")?,
chunk_hash: require_legacy_string(get("chunkHash"), "chunkHash")?,
size_bytes: require_legacy_i32(get("sizeBytes"), "sizeBytes")?,
payload: get("payload")
.and_then(rmpv::Value::as_slice)
.ok_or_else(|| anyhow!("payload must be binary data"))?
.to_vec(),
error: require_legacy_string(get("error"), "error")?,
}),
"cultnet.error.v0" => {
let code = get("code")
.filter(|value| !value.is_nil())
@ -1209,6 +1284,43 @@ fn encode_raw_cultnet_schema_message(message: &CultNetMessage) -> Result<rmpv::V
.unwrap_or(rmpv::Value::Nil),
),
],
CultNetMessage::ContentChunkRequest {
message_id,
chunk_hash,
record_key,
expected_size_bytes,
} => vec![
(
rmpv::Value::from("schemaVersion"),
rmpv::Value::from("cultmesh.content_chunk_request.v1"),
),
(rmpv::Value::from("messageId"), rmpv::Value::from(message_id.as_str())),
(rmpv::Value::from("chunkHash"), rmpv::Value::from(chunk_hash.as_str())),
(rmpv::Value::from("recordKey"), rmpv::Value::from(record_key.as_str())),
(
rmpv::Value::from("expectedSizeBytes"),
rmpv::Value::from(*expected_size_bytes),
),
],
CultNetMessage::ContentChunkResponse {
message_id,
found,
chunk_hash,
size_bytes,
payload,
error,
} => vec![
(
rmpv::Value::from("schemaVersion"),
rmpv::Value::from("cultmesh.content_chunk_response.v1"),
),
(rmpv::Value::from("messageId"), rmpv::Value::from(message_id.as_str())),
(rmpv::Value::from("found"), rmpv::Value::from(*found)),
(rmpv::Value::from("chunkHash"), rmpv::Value::from(chunk_hash.as_str())),
(rmpv::Value::from("sizeBytes"), rmpv::Value::from(*size_bytes)),
(rmpv::Value::from("payload"), rmpv::Value::Binary(payload.clone())),
(rmpv::Value::from("error"), rmpv::Value::from(error.as_str())),
],
// R-N: matches the landed C# `CultNetErrorMessage`
// (src/GameCult.Networking/CultNetSchemaMessages.cs) key for key and in its declared
// order - schemaVersion, error, routingHint, code, details - with every optional key
@ -1422,6 +1534,13 @@ fn require_legacy_u32(value: Option<&rmpv::Value>, field_name: &str) -> Result<u
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)
.ok_or_else(|| anyhow!("{field_name} must be an integer"))?;
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)

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

@ -1,3 +1,4 @@
mod content;
mod contracts;
mod idunn_deployment_brake;
mod idunn_lifecycle_brake;
@ -19,6 +20,7 @@ mod selection;
mod shard_catalog;
mod transport;
pub use content::*;
pub use contracts::*;
pub use idunn_deployment_brake::*;
pub use idunn_lifecycle_brake::*;

258
tests/GameCult.Mesh.Tests/CultMeshContentVectorTests.cs Normal file
View file

@ -0,0 +1,258 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text.Json;
using System.Threading.Tasks;
using FluentAssertions;
using GameCult.Caching;
using GameCult.Networking;
using MessagePack;
using NUnit.Framework;
namespace GameCult.Mesh.Tests;
/// <summary>
/// Content-plane parity vectors shared with <c>packages/cultnet-rs/tests/content.rs</c>. Each runtime
/// writes its own file under <c>contracts/cultmesh</c> and judges both. The chunk answers in these
/// files are produced by <see cref="CultMeshLegacyRudpContentServer"/> itself, so the failure
/// spellings are the reference's own bytes.
/// </summary>
[TestFixture]
public sealed class CultMeshContentVectorTests
{
private const int BodyLength = 1_315_551;
private const int ChunkSize = 262_144;
private const string CreatedAtUtc = "2026-09-22T00:00:00.0000000+00:00";
private const string ArtifactId = "eureka/content-vector";
private sealed record Vector(string Label, string Kind, byte[] Bytes);
[Test]
public void WriteContentVectors()
{
if (Environment.GetEnvironmentVariable("CULTNET_WRITE_VECTORS") != "1")
Assert.Ignore("Set CULTNET_WRITE_VECTORS=1 to (re)write the committed vector file.");
File.WriteAllText(VectorPath("content-vectors.cs-written.json"), ToJson(BuildVectors(out _, out _)));
}
[Test]
public void CommittedCsVectorsEqualAFreshEncode()
{
var fresh = BuildVectors(out _, out _);
var committed = ReadVectors("content-vectors.cs-written.json");
committed.Select(v => (v.Label, v.Kind)).Should().Equal(fresh.Select(v => (v.Label, v.Kind)));
foreach (var (expected, actual) in fresh.Zip(committed))
actual.Bytes.Should().Equal(expected.Bytes, $"vector '{actual.Label}' drifted from the reference's encode");
}
[Test]
public void RustVectorsDecodeInTheReference()
{
BuildVectors(out var cache, out var messages);
var rust = ReadVectors("content-vectors.rs-written.json");
rust.Should().NotBeEmpty();
rust.Select(v => v.Label).Should().Equal(messages.Keys, "both runtimes write the same labelled vectors");
foreach (var vector in rust)
{
var expected = messages[vector.Label];
switch (vector.Kind)
{
case "request":
var request = MessagePackSerializer.Deserialize<CultMeshContentChunkRequestMessage>(vector.Bytes, CultNetSchemaMessageSerialization.Options);
request.Should().BeEquivalentTo((CultMeshContentChunkRequestMessage)expected);
Reencode(request).Should().Equal(vector.Bytes, vector.Label);
break;
case "response":
var response = MessagePackSerializer.Deserialize<CultMeshContentChunkResponseMessage>(vector.Bytes, CultNetSchemaMessageSerialization.Options);
response.Should().BeEquivalentTo((CultMeshContentChunkResponseMessage)expected);
Reencode(response).Should().Equal(vector.Bytes, vector.Label);
break;
case "manifest":
var manifest = MessagePackSerializer.Deserialize<CultMeshCdnArtifactManifest>(vector.Bytes, CultNetSchemaMessageSerialization.Options);
manifest.Should().BeEquivalentTo((CultMeshCdnArtifactManifest)expected);
Reencode(manifest).Should().Equal(vector.Bytes, vector.Label);
break;
default:
Assert.Fail($"unknown vector kind '{vector.Kind}'");
break;
}
}
// Rust's answers to Rust's requests must be what the reference answers to the same bytes.
foreach (var vector in rust.Where(v => v.Label.EndsWith(".request", StringComparison.Ordinal)))
{
var request = MessagePackSerializer.Deserialize<CultMeshContentChunkRequestMessage>(vector.Bytes, CultNetSchemaMessageSerialization.Options);
var rustAnswer = rust.Single(v => v.Label == vector.Label[..^".request".Length] + ".response");
Reencode(Answer(cache, request)).Should().Equal(rustAnswer.Bytes, vector.Label);
}
}
[Test]
public void PackArtifactOfTheFixtureBodyEqualsTheFixtureManifest()
{
var artifact = CultMeshCdn.PackArtifact(ArtifactId, Body(), Options());
var committed = ReadVectors("content-vectors.cs-written.json").Single(v => v.Label == "manifest");
Reencode(artifact.Manifest).Should().Equal(committed.Bytes);
var rust = ReadVectors("content-vectors.rs-written.json").Single(v => v.Label == "manifest");
rust.Bytes.Should().Equal(committed.Bytes, "Rust's pack_content and the reference pack the same bytes to the same manifest");
}
private static byte[] Body() =>
Enumerable.Range(0, BodyLength).Select(i => (byte)(i % 251)).ToArray();
private static CultMeshCdnPackOptions Options() => new()
{
ChunkSizeBytes = ChunkSize,
Version = "1",
MimeType = "application/x-cultnet-vector",
CreatedAtUtc = CreatedAtUtc
};
private static byte[] Reencode(CultMeshContentChunkRequestMessage message) =>
CultNetSchemaMessageSerialization.Serialize(message);
private static byte[] Reencode(CultMeshContentChunkResponseMessage message) =>
CultNetSchemaMessageSerialization.Serialize(message);
private static byte[] Reencode(CultMeshCdnArtifactManifest manifest) =>
MessagePackSerializer.Serialize(manifest, CultNetSchemaMessageSerialization.Options);
private static CultMeshContentChunkRequestMessage Request(string id, string hash, string key, int size) => new()
{
MessageId = id,
ChunkHash = hash,
RecordKey = key,
ExpectedSizeBytes = size
};
/// <summary>The cs-written file's vectors, in file order. Messages are keyed by label.</summary>
private static List<Vector> BuildVectors(out CultCache cache, out Dictionary<string, object> messages)
{
var artifact = CultMeshCdn.PackArtifact(ArtifactId, Body(), Options());
var content = new CultCache(CultMesh.CreateCultCacheDocumentRegistry(
typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk)));
CultMeshCdn.PublishAsync(content, artifact).GetAwaiter().GetResult();
cache = content;
var last = artifact.Chunks[^1];
var key = "mesh:cdn:chunk:" + last.ChunkHash;
var lastDigit = last.ChunkHash[^1] == '0' ? '1' : '0';
var vectors = new List<Vector>();
var map = new Dictionary<string, object>(StringComparer.Ordinal);
messages = map;
void AddRequest(string label, CultMeshContentChunkRequestMessage message)
{
map[label] = message;
vectors.Add(new Vector(label, "request", Reencode(message)));
}
void AddResponse(string label, CultMeshContentChunkResponseMessage message)
{
map[label] = message;
vectors.Add(new Vector(label, "response", Reencode(message)));
}
void AddAnswer(string name, CultMeshContentChunkRequestMessage request)
{
AddRequest(name + ".request", request);
AddResponse(name + ".response", Answer(content, request));
}
AddAnswer("answer_found", Request("vector-found", "SHA256:" + last.ChunkHash, key, last.SizeBytes));
AddAnswer("answer_not_found", Request("vector-missing", new string('0', 64), string.Empty, 100));
AddAnswer("answer_record_key_disagrees", Request("vector-key", last.ChunkHash, "mesh:cdn:chunk:other", last.SizeBytes));
AddAnswer("answer_size_mismatch", Request("vector-size", last.ChunkHash, key, last.SizeBytes - 1));
AddResponse("response_hash_last_digit", new CultMeshContentChunkResponseMessage
{
MessageId = "vector-digit",
Found = true,
ChunkHash = last.ChunkHash[..^1] + lastDigit,
SizeBytes = last.SizeBytes,
Payload = last.Payload,
Error = string.Empty
});
void AddManifest(string label, CultMeshCdnArtifactManifest manifest)
{
map[label] = manifest;
vectors.Add(new Vector(label, "manifest", Reencode(manifest)));
}
AddManifest("manifest", artifact.Manifest);
var shuffled = CultMeshCdn.PackArtifact(ArtifactId, Body(), Options()).Manifest;
shuffled.Chunks = new[] { 3, 0, 5, 1, 4, 2 }.Select(i => shuffled.Chunks[i]).ToArray();
AddManifest("manifest_out_of_order", shuffled);
AddManifest("manifest_empty_body", CultMeshCdn.PackArtifact(ArtifactId, Array.Empty<byte>(), Options()).Manifest);
return vectors;
}
private static CultMeshContentChunkResponseMessage Answer(CultCache cache, CultMeshContentChunkRequestMessage request)
{
var host = new ReferenceHost();
using var server = new CultMeshLegacyRudpContentServer(host, cache);
var peer = new ReferencePeer();
host.Dispatch(request, peer).GetAwaiter().GetResult();
return peer.Sent ?? throw new InvalidOperationException("The reference server answered with silence.");
}
private sealed class ReferenceHost : ICultNetSchemaServer
{
private Func<CultMeshContentChunkRequestMessage, ICultNetSchemaServerPeer, Task>? _handler;
public void OnCultNet<TMessage>(Func<TMessage, ICultNetSchemaServerPeer, Task> callback)
where TMessage : ICultNetSchemaMessage =>
_handler = (Func<CultMeshContentChunkRequestMessage, ICultNetSchemaServerPeer, Task>)(object)callback;
public void RemoveCultNetMessageListener<TMessage>(Delegate callback)
where TMessage : ICultNetSchemaMessage => _handler = null;
public Task Dispatch(CultMeshContentChunkRequestMessage request, ICultNetSchemaServerPeer peer) =>
_handler!(request, peer);
}
private sealed class ReferencePeer : ICultNetSchemaServerPeer
{
public CultMeshContentChunkResponseMessage? Sent { get; private set; }
public void SendCultNet<TMessage>(TMessage message) where TMessage : ICultNetSchemaMessage =>
Sent = (CultMeshContentChunkResponseMessage)(object)message;
}
private static string ToJson(List<Vector> vectors) => JsonSerializer.Serialize(
new
{
vectors = vectors.Select(v => new
{
label = v.Label,
kind = v.Kind,
messagePackBase64 = Convert.ToBase64String(v.Bytes)
})
},
new JsonSerializerOptions { WriteIndented = true });
private static List<Vector> ReadVectors(string name)
{
var path = VectorPath(name);
if (!File.Exists(path))
Assert.Inconclusive($"{path} is missing - write it first (CULTNET_WRITE_VECTORS=1).");
using var document = JsonDocument.Parse(File.ReadAllText(path));
return document.RootElement.GetProperty("vectors").EnumerateArray()
.Select(v => new Vector(
v.GetProperty("label").GetString()!,
v.GetProperty("kind").GetString()!,
Convert.FromBase64String(v.GetProperty("messagePackBase64").GetString()!)))
.ToList();
}
private static string VectorPath(string name) => Path.Combine(RepoRoot(), "contracts", "cultmesh", name);
private static string RepoRoot()
{
var directory = new DirectoryInfo(TestContext.CurrentContext.TestDirectory);
while (directory != null && !File.Exists(Path.Combine(directory.FullName, "CultLib.sln")))
directory = directory.Parent;
return directory?.FullName ?? throw new InvalidOperationException("CultLib.sln was not found above the test directory.");
}
}