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

Add Rust provider session transport

Browse source
This commit is contained in:
Meta Vens 2026-07-19 00:49:27 +01:00
commit ba2ef582d2
5 changed files with 1937 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -2,6 +2,7 @@ mod contracts;
mod cultmesh;
mod framing;
mod operation_service;
mod provider_session;
mod replication;
mod rudp;
mod schema_discovery;
@ -13,6 +14,7 @@ pub use contracts::*;
pub use cultmesh::*;
pub use framing::*;
pub use operation_service::*;
pub use provider_session::*;
pub use replication::*;
pub use rudp::*;
pub use schema_discovery::*;

625
packages/cultnet-rs/src/provider_session.rs Normal file
View file

@ -0,0 +1,625 @@
use anyhow::{Result, anyhow};
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use chrono::DateTime;
use rmpv::Value;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use crate::{CultNetMessage, CultNetRawDocumentRecord};
pub const CULTMESH_PROVIDER_SESSION_SERVICE_ID: &str = "gamecult.mesh.provider_session";
pub const CULTMESH_PROVIDER_SESSION_PAYLOAD_ENCODING: &str = "messagepack-base64";
pub const CULTMESH_PROVIDER_REGISTER_OPERATION: &str = "provider.register";
pub const CULTMESH_PROVIDER_RENEW_OPERATION: &str = "provider.renew";
pub const CULTMESH_PROVIDER_PUBLICATION_PUT_OPERATION: &str = "provider.publication.put";
pub const CULTMESH_PROVIDER_PUBLICATION_DELETE_OPERATION: &str = "provider.publication.delete";
pub const CULTMESH_PROVIDER_RECEIPT_PUT_OPERATION: &str = "provider.receipt.put";
pub const CULTMESH_PROVIDER_WITHDRAW_OPERATION: &str = "provider.withdraw";
pub const CULTMESH_PROVIDER_REGISTRATION_SCHEMA: &str = "gamecult.mesh.provider_registration.v1";
pub const CULTMESH_PROVIDER_LEASE_SCHEMA: &str = "gamecult.mesh.provider_lease.v1";
pub const CULTMESH_PROVIDER_LEASE_RENEWAL_SCHEMA: &str = "gamecult.mesh.provider_lease_renewal.v1";
pub const CULTMESH_PROVIDER_PUBLICATION_PUT_SCHEMA: &str =
"gamecult.mesh.provider_publication_put.v1";
pub const CULTMESH_PROVIDER_PUBLICATION_DELETE_SCHEMA: &str =
"gamecult.mesh.provider_publication_delete.v1";
pub const CULTMESH_PROVIDER_COMMAND_SCHEMA: &str = "gamecult.mesh.provider_command.v1";
pub const CULTMESH_PROVIDER_RECEIPT_PUT_SCHEMA: &str = "gamecult.mesh.provider_receipt_put.v1";
pub const CULTMESH_PROVIDER_WITHDRAWAL_SCHEMA: &str = "gamecult.mesh.provider_withdrawal.v1";
pub const CULTMESH_PROVIDER_MUTATION_ACCEPTANCE_SCHEMA: &str =
"gamecult.mesh.provider_mutation_acceptance.v1";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CultMeshProviderOperationStatus {
Ok,
Conflict,
Expired,
Denied,
Invalid,
}
impl CultMeshProviderOperationStatus {
pub fn as_str(self) -> &'static str {
match self {
Self::Ok => "ok",
Self::Conflict => "conflict",
Self::Expired => "expired",
Self::Denied => "denied",
Self::Invalid => "invalid",
}
}
pub fn parse(value: &str) -> Result<Self> {
match value {
"ok" => Ok(Self::Ok),
"conflict" => Ok(Self::Conflict),
"expired" => Ok(Self::Expired),
"denied" => Ok(Self::Denied),
"invalid" => Ok(Self::Invalid),
_ => Err(anyhow!("Unsupported provider-session status {value:?}")),
}
}
}
pub trait CultMeshProviderSessionPayload {
fn validate(&self) -> Result<()>;
}
/// Provider-owned evidence carried verbatim in the RUDP Connect payload.
/// `client_session_id` distinguishes process/session generations; the optional
/// token remains opaque authorization evidence for the receiving service.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderConnectEvidenceWire {
pub client_session_id: String,
#[serde(default)]
pub session_token: Option<String>,
}
impl CultMeshProviderSessionPayload for CultMeshProviderConnectEvidenceWire {
fn validate(&self) -> Result<()> {
require_text(&self.client_session_id, "clientSessionId")?;
require_optional_text(self.session_token.as_deref(), "sessionToken")
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderRegistrationWire {
pub provider_id: String,
pub service_instance_id: String,
pub endpoint_id: String,
pub verse_id: String,
pub requested_lease_duration_ms: i32,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub authority_lease_id: Option<String>,
}
impl CultMeshProviderSessionPayload for CultMeshProviderRegistrationWire {
fn validate(&self) -> Result<()> {
require_identity(
&self.provider_id,
&self.service_instance_id,
&self.endpoint_id,
&self.verse_id,
)?;
require_positive(self.requested_lease_duration_ms, "requestedLeaseDurationMs")?;
require_optional_text(self.authority_lease_id.as_deref(), "authorityLeaseId")
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderLeaseWire {
pub provider_id: String,
pub service_instance_id: String,
pub endpoint_id: String,
pub verse_id: String,
pub lease_id: String,
pub valid_from_utc: String,
pub expires_at_utc: String,
}
impl CultMeshProviderSessionPayload for CultMeshProviderLeaseWire {
fn validate(&self) -> Result<()> {
require_identity(
&self.provider_id,
&self.service_instance_id,
&self.endpoint_id,
&self.verse_id,
)?;
require_text(&self.lease_id, "leaseId")?;
let valid_from = require_timestamp(&self.valid_from_utc, "validFromUtc")?;
let expires_at = require_timestamp(&self.expires_at_utc, "expiresAtUtc")?;
if expires_at <= valid_from {
return Err(anyhow!("Provider lease must expire after validFromUtc"));
}
Ok(())
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderLeaseRenewalWire {
pub lease_id: String,
pub requested_lease_duration_ms: i32,
}
impl CultMeshProviderSessionPayload for CultMeshProviderLeaseRenewalWire {
fn validate(&self) -> Result<()> {
require_text(&self.lease_id, "leaseId")?;
require_positive(self.requested_lease_duration_ms, "requestedLeaseDurationMs")
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderPublicationPutWire {
pub lease_id: String,
pub publication_id: String,
pub document: CultNetRawDocumentRecord,
}
impl CultMeshProviderSessionPayload for CultMeshProviderPublicationPutWire {
fn validate(&self) -> Result<()> {
require_text(&self.lease_id, "leaseId")?;
require_text(&self.publication_id, "publicationId")?;
validate_raw_document(&self.document)
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderPublicationDeleteWire {
pub lease_id: String,
pub publication_id: String,
pub schema_id: String,
pub record_key: String,
}
impl CultMeshProviderSessionPayload for CultMeshProviderPublicationDeleteWire {
fn validate(&self) -> Result<()> {
require_text(&self.lease_id, "leaseId")?;
require_text(&self.publication_id, "publicationId")?;
require_text(&self.schema_id, "schemaId")?;
require_text(&self.record_key, "recordKey")
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderCommandWire {
pub command_id: String,
pub command_kind: String,
pub provider_id: String,
pub service_instance_id: String,
pub payload: Value,
}
impl CultMeshProviderSessionPayload for CultMeshProviderCommandWire {
fn validate(&self) -> Result<()> {
require_text(&self.command_id, "commandId")?;
require_text(&self.command_kind, "commandKind")?;
require_text(&self.provider_id, "providerId")?;
require_text(&self.service_instance_id, "serviceInstanceId")
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum CultMeshProviderReceiptStateWire {
Applied,
Rejected,
Failed,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderCommandReceiptWire {
pub receipt_id: String,
pub command_id: String,
pub command_kind: String,
pub provider_id: String,
pub service_instance_id: String,
pub state: CultMeshProviderReceiptStateWire,
pub completed_at_utc: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub result: Option<Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
impl CultMeshProviderSessionPayload for CultMeshProviderCommandReceiptWire {
fn validate(&self) -> Result<()> {
require_text(&self.receipt_id, "receiptId")?;
require_text(&self.command_id, "commandId")?;
require_text(&self.command_kind, "commandKind")?;
require_text(&self.provider_id, "providerId")?;
require_text(&self.service_instance_id, "serviceInstanceId")?;
require_timestamp(&self.completed_at_utc, "completedAtUtc")?;
require_optional_text(self.error.as_deref(), "error")
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderReceiptPutWire {
pub lease_id: String,
pub receipt: CultMeshProviderCommandReceiptWire,
}
impl CultMeshProviderSessionPayload for CultMeshProviderReceiptPutWire {
fn validate(&self) -> Result<()> {
require_text(&self.lease_id, "leaseId")?;
self.receipt.validate()
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderWithdrawalWire {
pub lease_id: String,
}
impl CultMeshProviderSessionPayload for CultMeshProviderWithdrawalWire {
fn validate(&self) -> Result<()> {
require_text(&self.lease_id, "leaseId")
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CultMeshProviderMutationAcceptanceWire {
pub accepted_at_utc: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub lease_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub publication_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub command_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub receipt_id: Option<String>,
}
impl CultMeshProviderSessionPayload for CultMeshProviderMutationAcceptanceWire {
fn validate(&self) -> Result<()> {
require_timestamp(&self.accepted_at_utc, "acceptedAtUtc")?;
require_optional_text(self.lease_id.as_deref(), "leaseId")?;
require_optional_text(self.publication_id.as_deref(), "publicationId")?;
require_optional_text(self.command_id.as_deref(), "commandId")?;
require_optional_text(self.receipt_id.as_deref(), "receiptId")
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct CultMeshProviderErrorWire {}
impl CultMeshProviderSessionPayload for CultMeshProviderErrorWire {
fn validate(&self) -> Result<()> {
Ok(())
}
}
pub fn encode_provider_session_payload<T>(value: &T) -> Result<String>
where
T: CultMeshProviderSessionPayload + Serialize,
{
value.validate()?;
Ok(STANDARD.encode(rmp_serde::to_vec_named(value)?))
}
pub fn encode_provider_connect_evidence(
evidence: &CultMeshProviderConnectEvidenceWire,
) -> Result<Vec<u8>> {
evidence.validate()?;
Ok(rmp_serde::to_vec_named(evidence)?)
}
pub fn decode_provider_connect_evidence(
payload: &[u8],
) -> Result<CultMeshProviderConnectEvidenceWire> {
if payload.is_empty() {
return Err(anyhow!(
"Provider Connect evidence must be a non-empty MessagePack map"
));
}
let evidence: CultMeshProviderConnectEvidenceWire = rmp_serde::from_slice(payload)?;
evidence.validate()?;
Ok(evidence)
}
pub fn decode_provider_session_payload<T>(payload: &str) -> Result<T>
where
T: CultMeshProviderSessionPayload + DeserializeOwned,
{
require_text(payload, "payload")?;
let bytes = STANDARD
.decode(payload)
.map_err(|error| anyhow!("Provider-session payload must be MessagePack base64: {error}"))?;
let value: T = rmp_serde::from_slice(&bytes)?;
value.validate()?;
Ok(value)
}
pub fn decode_provider_command_document(
document: &CultNetRawDocumentRecord,
) -> Result<CultMeshProviderCommandWire> {
if document.schema_id != CULTMESH_PROVIDER_COMMAND_SCHEMA {
return Err(anyhow!(
"Expected provider command schema {:?}, received {:?}",
CULTMESH_PROVIDER_COMMAND_SCHEMA,
document.schema_id
));
}
let command: CultMeshProviderCommandWire = rmp_serde::from_slice(&document.payload)?;
command.validate()?;
Ok(command)
}
pub fn create_provider_session_request<T>(
message_id: impl Into<String>,
operation: impl Into<String>,
payload_schema: impl Into<String>,
payload: &T,
source_runtime_id: Option<String>,
target_runtime_id: Option<String>,
) -> Result<CultNetMessage>
where
T: CultMeshProviderSessionPayload + Serialize,
{
let message_id = message_id.into();
let operation = operation.into();
let payload_schema = payload_schema.into();
require_text(&message_id, "messageId")?;
validate_operation_schema(&operation, &payload_schema)?;
Ok(CultNetMessage::OperationRequest {
message_id,
service_id: CULTMESH_PROVIDER_SESSION_SERVICE_ID.to_string(),
operation,
payload_schema,
payload_encoding: CULTMESH_PROVIDER_SESSION_PAYLOAD_ENCODING.to_string(),
payload: encode_provider_session_payload(payload)?,
source_runtime_id,
target_runtime_id,
})
}
pub fn decode_provider_session_request<T>(
request: &CultNetMessage,
expected_operation: &str,
expected_payload_schema: &str,
) -> Result<T>
where
T: CultMeshProviderSessionPayload + DeserializeOwned,
{
let CultNetMessage::OperationRequest {
service_id,
operation,
payload_schema,
payload_encoding,
payload,
..
} = request
else {
return Err(anyhow!("Expected a CultNet operation request"));
};
require_envelope(
service_id,
operation,
payload_schema,
payload_encoding,
expected_operation,
expected_payload_schema,
)?;
decode_provider_session_payload(payload)
}
pub fn create_provider_session_response<T>(
request: &CultNetMessage,
status: CultMeshProviderOperationStatus,
payload_schema: impl Into<String>,
payload: &T,
source_runtime_id: Option<String>,
) -> Result<CultNetMessage>
where
T: CultMeshProviderSessionPayload + Serialize,
{
let CultNetMessage::OperationRequest {
message_id,
operation,
..
} = request
else {
return Err(anyhow!("Expected a CultNet operation request"));
};
require_text(message_id, "messageId")?;
require_text(operation, "operation")?;
let payload_schema = payload_schema.into();
require_text(&payload_schema, "payloadSchema")?;
Ok(CultNetMessage::OperationResponse {
message_id: message_id.clone(),
service_id: CULTMESH_PROVIDER_SESSION_SERVICE_ID.to_string(),
operation: operation.clone(),
status: status.as_str().to_string(),
payload_schema,
payload_encoding: CULTMESH_PROVIDER_SESSION_PAYLOAD_ENCODING.to_string(),
payload: encode_provider_session_payload(payload)?,
diagnostics: Vec::new(),
source_runtime_id,
})
}
pub fn create_provider_session_error_response(
request: &CultNetMessage,
status: CultMeshProviderOperationStatus,
diagnostics: Vec<String>,
source_runtime_id: Option<String>,
) -> Result<CultNetMessage> {
if status == CultMeshProviderOperationStatus::Ok {
return Err(anyhow!(
"Provider-session error responses require a non-ok status"
));
}
if diagnostics.is_empty() {
return Err(anyhow!(
"Provider-session error responses require diagnostics"
));
}
for diagnostic in &diagnostics {
require_text(diagnostic, "diagnostics[]")?;
}
let mut response = create_provider_session_response(
request,
status,
CULTMESH_PROVIDER_MUTATION_ACCEPTANCE_SCHEMA,
&CultMeshProviderErrorWire {},
source_runtime_id,
)?;
let CultNetMessage::OperationResponse {
diagnostics: response_diagnostics,
..
} = &mut response
else {
unreachable!("provider-session response helper returned a request")
};
*response_diagnostics = diagnostics;
Ok(response)
}
pub fn decode_provider_session_response<T>(
response: &CultNetMessage,
expected_operation: &str,
expected_payload_schema: &str,
) -> Result<(CultMeshProviderOperationStatus, T)>
where
T: CultMeshProviderSessionPayload + DeserializeOwned,
{
let CultNetMessage::OperationResponse {
service_id,
operation,
status,
payload_schema,
payload_encoding,
payload,
..
} = response
else {
return Err(anyhow!("Expected a CultNet operation response"));
};
require_envelope(
service_id,
operation,
payload_schema,
payload_encoding,
expected_operation,
expected_payload_schema,
)?;
Ok((
CultMeshProviderOperationStatus::parse(status)?,
decode_provider_session_payload(payload)?,
))
}
fn validate_operation_schema(operation: &str, payload_schema: &str) -> Result<()> {
let expected = match operation {
CULTMESH_PROVIDER_REGISTER_OPERATION => CULTMESH_PROVIDER_REGISTRATION_SCHEMA,
CULTMESH_PROVIDER_RENEW_OPERATION => CULTMESH_PROVIDER_LEASE_RENEWAL_SCHEMA,
CULTMESH_PROVIDER_PUBLICATION_PUT_OPERATION => CULTMESH_PROVIDER_PUBLICATION_PUT_SCHEMA,
CULTMESH_PROVIDER_PUBLICATION_DELETE_OPERATION => {
CULTMESH_PROVIDER_PUBLICATION_DELETE_SCHEMA
}
CULTMESH_PROVIDER_RECEIPT_PUT_OPERATION => CULTMESH_PROVIDER_RECEIPT_PUT_SCHEMA,
CULTMESH_PROVIDER_WITHDRAW_OPERATION => CULTMESH_PROVIDER_WITHDRAWAL_SCHEMA,
_ => {
return Err(anyhow!(
"Unsupported provider-session operation {operation:?}"
));
}
};
if payload_schema != expected {
return Err(anyhow!(
"Provider-session operation {operation:?} requires payload schema {expected:?}"
));
}
Ok(())
}
fn require_envelope(
service_id: &str,
operation: &str,
payload_schema: &str,
payload_encoding: &str,
expected_operation: &str,
expected_payload_schema: &str,
) -> Result<()> {
if service_id != CULTMESH_PROVIDER_SESSION_SERVICE_ID {
return Err(anyhow!(
"Unexpected provider-session service id {service_id:?}"
));
}
if operation != expected_operation {
return Err(anyhow!(
"Unexpected provider-session operation {operation:?}"
));
}
if payload_schema != expected_payload_schema {
return Err(anyhow!(
"Unexpected provider-session payload schema {payload_schema:?}"
));
}
if payload_encoding != CULTMESH_PROVIDER_SESSION_PAYLOAD_ENCODING {
return Err(anyhow!(
"Unexpected provider-session payload encoding {payload_encoding:?}"
));
}
Ok(())
}
fn validate_raw_document(document: &CultNetRawDocumentRecord) -> Result<()> {
require_text(&document.schema_id, "document.schemaId")?;
require_text(&document.record_key, "document.recordKey")?;
Ok(())
}
fn require_identity(
provider_id: &str,
service_instance_id: &str,
endpoint_id: &str,
verse_id: &str,
) -> Result<()> {
require_text(provider_id, "providerId")?;
require_text(service_instance_id, "serviceInstanceId")?;
require_text(endpoint_id, "endpointId")?;
require_text(verse_id, "verseId")
}
fn require_text(value: &str, field: &str) -> Result<()> {
if value.trim().is_empty() {
return Err(anyhow!("Provider-session {field} must not be empty"));
}
Ok(())
}
fn require_optional_text(value: Option<&str>, field: &str) -> Result<()> {
if let Some(value) = value {
require_text(value, field)?;
}
Ok(())
}
fn require_positive(value: i32, field: &str) -> Result<()> {
if value <= 0 {
return Err(anyhow!(
"Provider-session {field} must be greater than zero"
));
}
Ok(())
}
fn require_timestamp(value: &str, field: &str) -> Result<DateTime<chrono::FixedOffset>> {
require_text(value, field)?;
DateTime::parse_from_rfc3339(value)
.map_err(|error| anyhow!("Provider-session {field} must be RFC3339: {error}"))
}

500
packages/cultnet-rs/src/rudp.rs
View file

@ -29,6 +29,8 @@ use crate::encode_cultnet_message_to_vec;
const RUDP_MAGIC: [u8; 4] = [0x43, 0x4e, 0x52, 0x30];
const RUDP_VERSION: u8 = 0;
const RUDP_FIXED_HEADER_BYTES: usize = 36;
pub const CULTNET_RUDP_DEFAULT_MAX_PAYLOAD_BYTES: usize = 16 * 1024 * 1024;
const RUDP_RECEIVED_SEQUENCE_WINDOW: usize = 4_096;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CultNetRudpPacketType {
@ -127,6 +129,8 @@ pub struct CultNetRudpSession {
connection_id: u32,
resend_delay_ms: u64,
max_pending_reliable_packets: Option<usize>,
max_payload_bytes: Option<usize>,
max_pending_fragment_sets: usize,
initial_sequence: u32,
next_sequence: u32,
next_fragment_id: u16,
@ -146,6 +150,8 @@ impl CultNetRudpSession {
connection_id: options.connection_id,
resend_delay_ms: options.resend_delay_ms,
max_pending_reliable_packets: options.max_pending_reliable_packets,
max_payload_bytes: None,
max_pending_fragment_sets: 64,
initial_sequence: options.initial_sequence,
next_sequence: options.initial_sequence,
next_fragment_id: 1,
@ -176,10 +182,36 @@ impl CultNetRudpSession {
self.pending_reliable.keys().copied().collect()
}
fn pending_accept_for_resend(&mut self, now_ms: u64) -> Option<CultNetRudpPacket> {
let pending = self
.pending_reliable
.values_mut()
.find(|pending| pending.packet.packet_type == CultNetRudpPacketType::Accept)?;
pending.last_sent_at_ms = now_ms;
Some(pending.packet.clone())
}
pub fn last_received_at_ms(&self) -> Option<u64> {
self.last_received_at_ms
}
pub fn set_max_payload_bytes(&mut self, max_payload_bytes: Option<usize>) {
self.max_payload_bytes = max_payload_bytes;
}
pub fn set_max_pending_fragment_sets(
&mut self,
max_pending_fragment_sets: usize,
) -> Result<()> {
if max_pending_fragment_sets == 0 {
return Err(anyhow!(
"RUDP max_pending_fragment_sets must be greater than zero"
));
}
self.max_pending_fragment_sets = max_pending_fragment_sets;
Ok(())
}
pub fn reset_peer_state(&mut self) {
self.next_sequence = self.initial_sequence;
self.next_fragment_id = 1;
@ -223,6 +255,7 @@ impl CultNetRudpSession {
self.ensure_reliable_capacity(1)?;
self.remember_received(packet.sequence);
self.last_received_at_ms = Some(now_ms);
self.connected = true;
let response = self.create_packet(
CultNetRudpPacketType::Accept,
@ -260,6 +293,7 @@ impl CultNetRudpSession {
"Cannot send RUDP data before the session is connected"
));
}
self.require_payload_size(payload.len())?;
if let Some(max_fragment_bytes) = max_fragment_bytes {
if max_fragment_bytes == 0 {
@ -400,7 +434,11 @@ impl CultNetRudpSession {
});
}
let duplicate = self.received_sequences.contains(&packet.sequence);
let duplicate = self.received_sequences.contains(&packet.sequence)
|| self.highest_received_sequence.is_some_and(|highest| {
packet.sequence < highest
&& highest - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW as u32
});
self.remember_received(packet.sequence);
if duplicate {
return Ok(CultNetRudpReceiveResult {
@ -457,6 +495,24 @@ impl CultNetRudpSession {
}
}
pub fn create_ack_for(&self, sequence: u32) -> CultNetRudpPacket {
CultNetRudpPacket {
packet_type: CultNetRudpPacketType::Ack,
connection_id: self.connection_id,
sequence: 0,
ack: sequence,
ack_mask: 0,
channel_id: "control".to_string(),
reliable: false,
ordered: false,
sequenced: false,
fragment_id: 0,
fragment_index: 0,
fragment_count: 0,
payload: Vec::new(),
}
}
pub fn create_ping(&mut self, payload: Vec<u8>) -> CultNetRudpPacket {
self.create_packet(
CultNetRudpPacketType::Ping,
@ -603,6 +659,13 @@ impl CultNetRudpSession {
{
self.highest_received_sequence = Some(sequence);
}
if self.received_sequences.len() > RUDP_RECEIVED_SEQUENCE_WINDOW {
let keep_from = self
.highest_received_sequence
.unwrap_or(sequence)
.saturating_sub(RUDP_RECEIVED_SEQUENCE_WINDOW as u32 - 1);
self.received_sequences = self.received_sequences.split_off(&keep_from);
}
}
fn ack_state(&self) -> (u32, u32) {
@ -621,6 +684,7 @@ impl CultNetRudpSession {
packet: &CultNetRudpPacket,
) -> Result<Option<(CultNetRudpDeliveredFrame, bool, u32)>> {
if packet.fragment_count == 0 {
self.require_payload_size(packet.payload.len())?;
return Ok(Some((
CultNetRudpDeliveredFrame {
channel_id: packet.channel_id.clone(),
@ -643,6 +707,27 @@ impl CultNetRudpSession {
}
let key = (packet.channel_id.clone(), packet.fragment_id);
if !self.fragment_buffers.contains_key(&key)
&& self.fragment_buffers.len() >= self.max_pending_fragment_sets
{
return Err(anyhow!("RUDP pending fragment-set limit reached"));
}
let buffered_bytes = self
.fragment_buffers
.get(&key)
.map(|buffer| {
buffer
.payloads
.iter()
.filter(|(index, _)| **index != packet.fragment_index)
.map(|(_, payload)| payload.len())
.sum::<usize>()
})
.unwrap_or(0);
if let Err(error) = self.require_payload_size(buffered_bytes + packet.payload.len()) {
self.fragment_buffers.remove(&key);
return Err(error);
}
let buffer = self
.fragment_buffers
.entry(key.clone())
@ -694,6 +779,16 @@ impl CultNetRudpSession {
)))
}
fn require_payload_size(&self, payload_bytes: usize) -> Result<()> {
if self
.max_payload_bytes
.is_some_and(|max_payload_bytes| payload_bytes > max_payload_bytes)
{
return Err(anyhow!("RUDP payload exceeds max_payload_bytes"));
}
Ok(())
}
fn deliver_ordered(
&mut self,
frame: CultNetRudpDeliveredFrame,
@ -889,16 +984,394 @@ pub struct CultNetRudpSocketTransportConnection {
pong_payloads: VecDeque<Vec<u8>>,
}
impl CultNetRudpSocketTransportConnection {
pub fn new(options: CultNetRudpSocketTransportOptions) -> Result<Self> {
pub struct CultNetRudpServerHubOptions {
pub runtime_id: String,
pub socket: UdpSocket,
pub connection_id: u32,
pub initial_sequence: u32,
pub resend_delay_ms: u64,
pub transport_id: Option<String>,
pub max_payload_bytes: Option<u32>,
pub max_fragment_bytes: Option<u32>,
pub max_pending_reliable_packets: Option<u32>,
pub max_peers: usize,
}
impl CultNetRudpServerHubOptions {
pub fn new(runtime_id: impl Into<String>, socket: UdpSocket, connection_id: u32) -> Self {
Self {
runtime_id: runtime_id.into(),
socket,
connection_id,
initial_sequence: 1,
resend_delay_ms: 250,
transport_id: None,
max_payload_bytes: None,
max_fragment_bytes: None,
max_pending_reliable_packets: None,
max_peers: 256,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CultNetRudpServerSessionContext {
pub remote_addr: SocketAddr,
pub connection_id: u32,
/// Server-minted generation used to fence work from an older session at
/// the same UDP endpoint.
pub session_generation: u64,
/// Exact bytes supplied by the peer's most recently accepted Connect packet.
/// CultNet does not interpret these bytes; higher layers may use them as
/// transport evidence when making their own authorization decision.
pub connect_payload: Vec<u8>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CultNetRudpServerEvent {
Connected {
session: CultNetRudpServerSessionContext,
},
Frame {
session: CultNetRudpServerSessionContext,
frame: CultNetTransportFrame,
},
Pong {
session: CultNetRudpServerSessionContext,
payload: Vec<u8>,
},
Disconnected {
session: CultNetRudpServerSessionContext,
reason: Vec<u8>,
},
}
struct CultNetRudpServerPeer {
session: CultNetRudpSession,
context: CultNetRudpServerSessionContext,
}
/// One UDP listener with an independent reliable session for every remote peer.
///
/// The hub owns packet/session mechanics only. Registration, leases, commands,
/// and all other application policy belong to the service using the hub.
pub struct CultNetRudpServerHub {
socket: UdpSocket,
connection_id: u32,
initial_sequence: u32,
resend_delay_ms: u64,
max_pending_reliable_packets: Option<usize>,
max_payload_bytes: usize,
max_fragment_bytes: Option<usize>,
max_peers: usize,
next_session_generation: u64,
peers: BTreeMap<SocketAddr, CultNetRudpServerPeer>,
pending_events: VecDeque<CultNetRudpServerEvent>,
pub profile: CultNetTransportProfile,
stats: CultNetTransportStats,
}
impl CultNetRudpServerHub {
pub fn new(options: CultNetRudpServerHubOptions) -> Result<Self> {
let local_addr = options.socket.local_addr()?;
if options.max_peers == 0 {
return Err(anyhow!(
"RUDP server hub max_peers must be greater than zero"
));
}
let max_payload_bytes = options
.max_payload_bytes
.or(Some(CULTNET_RUDP_DEFAULT_MAX_PAYLOAD_BYTES as u32));
let profile = create_rudp_transport_profile(
options.runtime_id,
RudpTransportProfileOptions {
transport_id: options.transport_id,
host: Some(local_addr.ip().to_string()),
port: Some(local_addr.port()),
max_payload_bytes: options.max_payload_bytes,
max_payload_bytes,
max_fragment_bytes: options.max_fragment_bytes,
max_pending_reliable_packets: options.max_pending_reliable_packets,
reconnect_policy: None,
},
);
Ok(Self {
socket: options.socket,
connection_id: options.connection_id,
initial_sequence: options.initial_sequence,
resend_delay_ms: options.resend_delay_ms,
max_pending_reliable_packets: options
.max_pending_reliable_packets
.map(|value| value as usize),
max_payload_bytes: max_payload_bytes.expect("RUDP hub has an effective payload limit")
as usize,
max_fragment_bytes: options.max_fragment_bytes.map(|value| value as usize),
max_peers: options.max_peers,
next_session_generation: 1,
peers: BTreeMap::new(),
pending_events: VecDeque::new(),
profile,
stats: CultNetTransportStats::default(),
})
}
pub fn local_addr(&self) -> Result<SocketAddr> {
Ok(self.socket.local_addr()?)
}
pub fn stats(&self) -> CultNetTransportStats {
self.stats.clone()
}
pub fn sessions(&self) -> Vec<CultNetRudpServerSessionContext> {
self.peers
.values()
.map(|peer| peer.context.clone())
.collect()
}
pub fn session(&self, remote_addr: SocketAddr) -> Option<&CultNetRudpServerSessionContext> {
self.peers.get(&remote_addr).map(|peer| &peer.context)
}
pub fn send(
&mut self,
session: &CultNetRudpServerSessionContext,
channel_id: &str,
payload: Vec<u8>,
) -> Result<()> {
let remote_addr = session.remote_addr;
let packets = {
let peer = self
.peers
.get_mut(&remote_addr)
.ok_or_else(|| anyhow!("Unknown RUDP server peer {remote_addr}"))?;
if peer.context.session_generation != session.session_generation {
return Err(anyhow!(
"RUDP server session generation {} is no longer active",
session.session_generation
));
}
peer.session.send_many(
channel_id,
payload,
channel_send_options(channel_id, now_ms()),
self.max_fragment_bytes,
)?
};
for packet in packets {
self.send_packet(remote_addr, &packet)?;
}
self.stats.frames_sent += 1;
Ok(())
}
pub fn send_schema_message(
&mut self,
session: &CultNetRudpServerSessionContext,
message: &CultNetMessage,
) -> Result<()> {
let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
self.send(session, "schema", payload)
}
pub fn disconnect(
&mut self,
session: &CultNetRudpServerSessionContext,
reason: Vec<u8>,
) -> Result<bool> {
let Some(peer) = self.peers.get(&session.remote_addr) else {
return Ok(false);
};
if peer.context.session_generation != session.session_generation {
return Ok(false);
}
let mut peer = self
.peers
.remove(&session.remote_addr)
.expect("validated RUDP peer exists");
let packet = peer.session.create_disconnect(reason);
self.send_packet(session.remote_addr, &packet)?;
Ok(true)
}
pub fn receive_event_once(&mut self) -> Result<Option<CultNetRudpServerEvent>> {
if let Some(event) = self.pending_events.pop_front() {
return Ok(Some(event));
}
let mut wire = vec![0_u8; 65_535];
let (received, remote_addr) = match self.socket.recv_from(&mut wire) {
Ok(value) => value,
Err(error)
if error.kind() == ErrorKind::WouldBlock || error.kind() == ErrorKind::TimedOut =>
{
return Ok(None);
}
Err(error) => return Err(error.into()),
};
wire.truncate(received);
self.stats.bytes_received += received as u64;
let packet = decode_rudp_packet(&wire)?;
if packet.connection_id != self.connection_id {
return Ok(None);
}
if packet.packet_type == CultNetRudpPacketType::Connect {
if let Some(peer) = self.peers.get_mut(&remote_addr)
&& peer.context.connect_payload == packet.payload
{
let _ = peer.session.receive(&packet, now_ms())?;
let reply = peer
.session
.pending_accept_for_resend(now_ms())
.unwrap_or_else(|| peer.session.create_ack_for(packet.sequence));
self.send_packet(remote_addr, &reply)?;
return Ok(None);
}
if !self.peers.contains_key(&remote_addr) && self.peers.len() >= self.max_peers {
return Err(anyhow!("RUDP server hub peer limit reached"));
}
let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: self.connection_id,
initial_sequence: self.initial_sequence,
resend_delay_ms: self.resend_delay_ms,
max_pending_reliable_packets: self.max_pending_reliable_packets,
});
session.set_max_payload_bytes(Some(self.max_payload_bytes));
let generation = self.next_session_generation;
self.next_session_generation = self
.next_session_generation
.checked_add(1)
.ok_or_else(|| anyhow!("RUDP server session generation exhausted"))?;
let context = CultNetRudpServerSessionContext {
remote_addr,
connection_id: packet.connection_id,
session_generation: generation,
connect_payload: packet.payload.clone(),
};
let accept = session.accept_connect(&packet, now_ms(), Vec::new())?;
self.send_packet(remote_addr, &accept)?;
let replaced = self.peers.insert(
remote_addr,
CultNetRudpServerPeer {
session,
context: context.clone(),
},
);
if let Some(replaced) = replaced {
self.pending_events
.push_back(CultNetRudpServerEvent::Disconnected {
session: replaced.context,
reason: b"replaced by a new Connect generation".to_vec(),
});
}
self.pending_events
.push_back(CultNetRudpServerEvent::Connected { session: context });
return Ok(self.pending_events.pop_front());
}
let Some(peer) = self.peers.get_mut(&remote_addr) else {
return Ok(None);
};
let result = peer.session.receive(&packet, now_ms())?;
let context = peer.context.clone();
let ack = if packet.reliable {
Some(peer.session.create_ack_for(packet.sequence))
} else {
None
};
if let Some(reply) = result.reply {
self.send_packet(remote_addr, &reply)?;
}
if let Some(ack) = ack {
self.send_packet(remote_addr, &ack)?;
}
if result.pong {
self.pending_events.push_back(CultNetRudpServerEvent::Pong {
session: context.clone(),
payload: result.pong_payload,
});
}
for delivered in result.delivered {
self.stats.frames_received += 1;
self.pending_events
.push_back(CultNetRudpServerEvent::Frame {
session: context.clone(),
frame: CultNetTransportFrame {
channel_id: delivered.channel_id,
payload: delivered.payload,
},
});
}
if result.disconnected {
self.pending_events
.push_back(CultNetRudpServerEvent::Disconnected {
session: context,
reason: result.disconnect_reason,
});
self.peers.remove(&remote_addr);
}
Ok(self.pending_events.pop_front())
}
pub fn poll_resends(&mut self) -> Result<()> {
let now = now_ms();
let mut packets = Vec::new();
for (remote_addr, peer) in &mut self.peers {
packets.extend(
peer.session
.due_resends(now)
.into_iter()
.map(|packet| (*remote_addr, packet)),
);
}
for (remote_addr, packet) in packets {
self.send_packet(remote_addr, &packet)?;
}
Ok(())
}
pub fn remove_timed_out_sessions(
&mut self,
timeout_ms: u64,
) -> Vec<CultNetRudpServerSessionContext> {
let now = now_ms();
let timed_out = self
.peers
.iter_mut()
.filter_map(|(remote_addr, peer)| {
peer.session
.check_timeout(now, timeout_ms)
.then_some((*remote_addr, peer.context.clone()))
})
.collect::<Vec<_>>();
for (remote_addr, _) in &timed_out {
self.peers.remove(remote_addr);
}
timed_out.into_iter().map(|(_, context)| context).collect()
}
fn send_packet(&mut self, remote_addr: SocketAddr, packet: &CultNetRudpPacket) -> Result<()> {
let wire = encode_rudp_packet(packet)?;
let sent = self.socket.send_to(&wire, remote_addr)?;
self.stats.bytes_sent += sent as u64;
Ok(())
}
}
impl CultNetRudpSocketTransportConnection {
pub fn new(options: CultNetRudpSocketTransportOptions) -> Result<Self> {
let local_addr = options.socket.local_addr()?;
let max_payload_bytes = options
.max_payload_bytes
.or(Some(CULTNET_RUDP_DEFAULT_MAX_PAYLOAD_BYTES as u32));
let profile = create_rudp_transport_profile(
options.runtime_id,
RudpTransportProfileOptions {
transport_id: options.transport_id,
host: Some(local_addr.ip().to_string()),
port: Some(local_addr.port()),
max_payload_bytes,
max_fragment_bytes: options.max_fragment_bytes,
max_pending_reliable_packets: options.max_pending_reliable_packets,
reconnect_policy: options.reconnect_policy,
@ -907,14 +1380,16 @@ impl CultNetRudpSocketTransportConnection {
let max_pending_reliable_packets = options
.max_pending_reliable_packets
.map(|value| value as usize);
let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: options.connection_id,
initial_sequence: options.initial_sequence,
resend_delay_ms: options.resend_delay_ms,
max_pending_reliable_packets,
});
session.set_max_payload_bytes(max_payload_bytes.map(|value| value as usize));
Ok(Self {
socket: options.socket,
session: CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: options.connection_id,
initial_sequence: options.initial_sequence,
resend_delay_ms: options.resend_delay_ms,
max_pending_reliable_packets,
}),
session,
mode: options.mode,
remote_addr: options.remote_addr,
profile,
@ -1065,8 +1540,9 @@ impl CultNetRudpSocketTransportConnection {
self.stats.frames_received += 1;
}
let frame = self.delivered_frames.pop_front();
if packet.packet_type == CultNetRudpPacketType::Accept || frame.is_some() {
let ack = self.session.create_ack();
if packet.reliable || packet.packet_type == CultNetRudpPacketType::Accept || frame.is_some()
{
let ack = self.session.create_ack_for(packet.sequence);
self.send_packet(&ack)?;
}
Ok(frame)

365
packages/cultnet-rs/tests/provider_session.rs Normal file
View file

@ -0,0 +1,365 @@
use anyhow::Result;
use cultnet_rs::*;
const REGISTRATION_GOLDEN: &str = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03";
const CONNECT_EVIDENCE_GOLDEN: &str =
"gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbrJvZGluLXNlc3Npb24tdG9rZW4=";
const TOKENLESS_CONNECT_EVIDENCE_GOLDEN: &str =
"gq9jbGllbnRTZXNzaW9uSWSwYW5vbnltb3VzLWNsaWVudKxzZXNzaW9uVG9rZW7A";
fn registration() -> CultMeshProviderRegistrationWire {
CultMeshProviderRegistrationWire {
provider_id: "aetheria".to_string(),
service_instance_id: "aetheria-42".to_string(),
endpoint_id: "aetheria-public".to_string(),
verse_id: "public".to_string(),
requested_lease_duration_ms: 30_000,
authority_lease_id: Some("authority-7".to_string()),
}
}
#[test]
fn registration_matches_csharp_and_typescript_golden_payload() -> Result<()> {
let encoded = encode_provider_session_payload(&registration())?;
assert_eq!(encoded, REGISTRATION_GOLDEN);
assert_eq!(
decode_provider_session_payload::<CultMeshProviderRegistrationWire>(&encoded)?,
registration()
);
Ok(())
}
#[test]
fn connect_evidence_is_a_validated_named_messagepack_map() -> Result<()> {
let evidence = CultMeshProviderConnectEvidenceWire {
client_session_id: "aetheria-client-42".to_string(),
session_token: Some("odin-session-token".to_string()),
};
let encoded = encode_provider_connect_evidence(&evidence)?;
assert_eq!(
base64::Engine::encode(&base64::engine::general_purpose::STANDARD, &encoded),
CONNECT_EVIDENCE_GOLDEN
);
assert_eq!(decode_provider_connect_evidence(&encoded)?, evidence);
let value: rmpv::Value = rmp_serde::from_slice(&encoded)?;
let map = value.as_map().expect("Connect evidence is a named map");
assert_eq!(
map.iter()
.find_map(|(key, value)| (key.as_str() == Some("clientSessionId")).then_some(value))
.and_then(rmpv::Value::as_str),
Some("aetheria-client-42")
);
assert_eq!(
map.iter()
.find_map(|(key, value)| (key.as_str() == Some("sessionToken")).then_some(value))
.and_then(rmpv::Value::as_str),
Some("odin-session-token")
);
assert!(
encode_provider_connect_evidence(&CultMeshProviderConnectEvidenceWire {
client_session_id: String::new(),
session_token: None,
})
.is_err()
);
assert!(
encode_provider_connect_evidence(&CultMeshProviderConnectEvidenceWire {
client_session_id: "client".to_string(),
session_token: Some(" ".to_string()),
})
.is_err()
);
assert!(decode_provider_connect_evidence(&[]).is_err());
let tokenless = CultMeshProviderConnectEvidenceWire {
client_session_id: "anonymous-client".to_string(),
session_token: None,
};
let tokenless_encoded = encode_provider_connect_evidence(&tokenless)?;
assert_eq!(
base64::Engine::encode(
&base64::engine::general_purpose::STANDARD,
&tokenless_encoded
),
TOKENLESS_CONNECT_EVIDENCE_GOLDEN
);
assert_eq!(
decode_provider_connect_evidence(&tokenless_encoded)?,
tokenless
);
Ok(())
}
#[test]
fn registration_round_trips_inside_canonical_operation_envelope() -> Result<()> {
let request = create_provider_session_request(
"register-1",
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_REGISTRATION_SCHEMA,
&registration(),
Some("aetheria-42".to_string()),
Some("odin".to_string()),
)?;
let bytes = encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0)?;
let decoded_request =
decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0)?;
let decoded: CultMeshProviderRegistrationWire = decode_provider_session_request(
&decoded_request,
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_REGISTRATION_SCHEMA,
)?;
assert_eq!(decoded, registration());
let lease = CultMeshProviderLeaseWire {
provider_id: "aetheria".to_string(),
service_instance_id: "aetheria-42".to_string(),
endpoint_id: "aetheria-public".to_string(),
verse_id: "public".to_string(),
lease_id: "lease-2".to_string(),
valid_from_utc: "2026-07-14T12:00:00Z".to_string(),
expires_at_utc: "2026-07-14T12:00:10Z".to_string(),
};
let response = create_provider_session_response(
&decoded_request,
CultMeshProviderOperationStatus::Ok,
CULTMESH_PROVIDER_LEASE_SCHEMA,
&lease,
Some("odin".to_string()),
)?;
let (status, decoded_lease) = decode_provider_session_response::<CultMeshProviderLeaseWire>(
&response,
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_LEASE_SCHEMA,
)?;
assert_eq!(status, CultMeshProviderOperationStatus::Ok);
assert_eq!(decoded_lease, lease);
Ok(())
}
#[test]
fn publication_and_command_preserve_typed_messagepack_values() -> Result<()> {
let publication = CultMeshProviderPublicationPutWire {
lease_id: "lease-1".to_string(),
publication_id: "surface:pilot".to_string(),
document: CultNetRawDocumentRecord {
schema_id: "gamecult.eve.surface.v1".to_string(),
record_key: "eve:surface:pilot".to_string(),
stored_at: String::new(),
payload_encoding: CultNetRawPayloadEncoding::Messagepack,
payload: vec![0x81, 0xa2, 0x69, 0x64, 0x01],
source_runtime_id: Some("aetheria-42".to_string()),
source_agent_id: None,
source_role: None,
tags: None,
},
};
let publication_encoded = encode_provider_session_payload(&publication)?;
assert_eq!(
decode_provider_session_payload::<CultMeshProviderPublicationPutWire>(
&publication_encoded
)?,
publication
);
let command = CultMeshProviderCommandWire {
command_id: "cmd-7".to_string(),
command_kind: "pilot.thrust".to_string(),
provider_id: "aetheria".to_string(),
service_instance_id: "aetheria-42".to_string(),
payload: rmpv::Value::Map(vec![(rmpv::Value::from("axis"), rmpv::Value::F64(0.75))]),
};
let command_bytes = base64::Engine::decode(
&base64::engine::general_purpose::STANDARD,
encode_provider_session_payload(&command)?,
)?;
let decoded_command = decode_provider_command_document(&CultNetRawDocumentRecord {
schema_id: CULTMESH_PROVIDER_COMMAND_SCHEMA.to_string(),
record_key: "provider-command:aetheria:aetheria-42:cmd-7".to_string(),
stored_at: String::new(),
payload_encoding: CultNetRawPayloadEncoding::Messagepack,
payload: command_bytes,
source_runtime_id: Some("odin".to_string()),
source_agent_id: None,
source_role: Some("provider-session-broker".to_string()),
tags: None,
})?;
assert_eq!(decoded_command, command);
Ok(())
}
#[test]
fn lifecycle_validation_rejects_split_or_malformed_authority() {
let mismatched = create_provider_session_request(
"bad-schema",
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_PUBLICATION_PUT_SCHEMA,
&registration(),
None,
None,
);
assert!(mismatched.is_err());
let malformed_lease = CultMeshProviderLeaseWire {
provider_id: "provider".to_string(),
service_instance_id: "instance".to_string(),
endpoint_id: "endpoint".to_string(),
verse_id: "verse".to_string(),
lease_id: "lease".to_string(),
valid_from_utc: "2026-07-14T12:00:10Z".to_string(),
expires_at_utc: "2026-07-14T12:00:00Z".to_string(),
};
assert!(encode_provider_session_payload(&malformed_lease).is_err());
let malformed_request = CultNetMessage::OperationRequest {
message_id: String::new(),
service_id: CULTMESH_PROVIDER_SESSION_SERVICE_ID.to_string(),
operation: String::new(),
payload_schema: CULTMESH_PROVIDER_REGISTRATION_SCHEMA.to_string(),
payload_encoding: CULTMESH_PROVIDER_SESSION_PAYLOAD_ENCODING.to_string(),
payload: REGISTRATION_GOLDEN.to_string(),
source_runtime_id: None,
target_runtime_id: None,
};
assert!(
create_provider_session_response(
&malformed_request,
CultMeshProviderOperationStatus::Invalid,
CULTMESH_PROVIDER_MUTATION_ACCEPTANCE_SCHEMA,
&CultMeshProviderMutationAcceptanceWire {
accepted_at_utc: "2026-07-14T12:00:00Z".to_string(),
lease_id: None,
publication_id: None,
command_id: None,
receipt_id: None,
},
None,
)
.is_err()
);
}
#[test]
fn remaining_lifecycle_payloads_round_trip_as_named_maps() -> Result<()> {
let renewal = CultMeshProviderLeaseRenewalWire {
lease_id: "lease-1".to_string(),
requested_lease_duration_ms: 30_000,
};
assert_eq!(
decode_provider_session_payload::<CultMeshProviderLeaseRenewalWire>(
&encode_provider_session_payload(&renewal)?
)?,
renewal
);
let deletion = CultMeshProviderPublicationDeleteWire {
lease_id: "lease-1".to_string(),
publication_id: "surface:pilot".to_string(),
schema_id: "gamecult.eve.surface.v1".to_string(),
record_key: "eve:surface:pilot".to_string(),
};
assert_eq!(
decode_provider_session_payload::<CultMeshProviderPublicationDeleteWire>(
&encode_provider_session_payload(&deletion)?
)?,
deletion
);
let receipt = CultMeshProviderReceiptPutWire {
lease_id: "lease-1".to_string(),
receipt: CultMeshProviderCommandReceiptWire {
receipt_id: "receipt-7".to_string(),
command_id: "cmd-7".to_string(),
command_kind: "pilot.thrust".to_string(),
provider_id: "aetheria".to_string(),
service_instance_id: "aetheria-42".to_string(),
state: CultMeshProviderReceiptStateWire::Applied,
completed_at_utc: "2026-07-14T12:01:02.345Z".to_string(),
result: Some(rmpv::Value::Map(vec![(
rmpv::Value::from("accepted"),
rmpv::Value::Boolean(true),
)])),
error: None,
},
};
assert_eq!(
decode_provider_session_payload::<CultMeshProviderReceiptPutWire>(
&encode_provider_session_payload(&receipt)?
)?,
receipt
);
let withdrawal = CultMeshProviderWithdrawalWire {
lease_id: "lease-1".to_string(),
};
assert_eq!(
decode_provider_session_payload::<CultMeshProviderWithdrawalWire>(
&encode_provider_session_payload(&withdrawal)?
)?,
withdrawal
);
let acceptance = CultMeshProviderMutationAcceptanceWire {
accepted_at_utc: "2026-07-14T12:01:03Z".to_string(),
lease_id: Some("lease-1".to_string()),
publication_id: Some("surface:pilot".to_string()),
command_id: None,
receipt_id: None,
};
assert_eq!(
decode_provider_session_payload::<CultMeshProviderMutationAcceptanceWire>(
&encode_provider_session_payload(&acceptance)?
)?,
acceptance
);
Ok(())
}
#[test]
fn non_ok_response_uses_correlated_empty_map_and_diagnostics() -> Result<()> {
let request = create_provider_session_request(
"register-denied",
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_REGISTRATION_SCHEMA,
&registration(),
Some("aetheria-42".to_string()),
Some("odin".to_string()),
)?;
let response = create_provider_session_error_response(
&request,
CultMeshProviderOperationStatus::Denied,
vec!["registration authority rejected the session".to_string()],
Some("odin".to_string()),
)?;
let CultNetMessage::OperationResponse {
message_id,
operation,
status,
payload_schema,
payload,
diagnostics,
..
} = response
else {
panic!("expected operation response");
};
assert_eq!(message_id, "register-denied");
assert_eq!(operation, CULTMESH_PROVIDER_REGISTER_OPERATION);
assert_eq!(status, "denied");
assert_eq!(payload_schema, CULTMESH_PROVIDER_MUTATION_ACCEPTANCE_SCHEMA);
assert_eq!(diagnostics.len(), 1);
let bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, payload)?;
assert_eq!(
rmp_serde::from_slice::<rmpv::Value>(&bytes)?,
rmpv::Value::Map(Vec::new())
);
assert!(
create_provider_session_error_response(
&request,
CultMeshProviderOperationStatus::Ok,
vec!["not an error".to_string()],
None,
)
.is_err()
);
Ok(())
}

459
packages/cultnet-rs/tests/rudp_server_hub.rs Normal file
View file

@ -0,0 +1,459 @@
use std::net::UdpSocket;
use std::time::Duration;
use anyhow::{Result, anyhow};
use cultnet_rs::*;
const CONNECTION_ID: u32 = 0x4355_4c54;
fn socket() -> Result<UdpSocket> {
let socket = UdpSocket::bind("127.0.0.1:0")?;
socket.set_read_timeout(Some(Duration::from_millis(20)))?;
Ok(socket)
}
fn client(remote: std::net::SocketAddr) -> Result<CultNetRudpSocketTransportConnection> {
let mut options =
CultNetRudpSocketTransportOptions::client("provider", socket()?, remote, CONNECTION_ID);
options.max_fragment_bytes = Some(2_048);
CultNetRudpSocketTransportConnection::new(options)
}
fn connect(
hub: &mut CultNetRudpServerHub,
client: &mut CultNetRudpSocketTransportConnection,
evidence: &[u8],
) -> Result<CultNetRudpServerSessionContext> {
client.connect(evidence.to_vec())?;
let mut connected = None;
for _ in 0..20 {
if let Some(CultNetRudpServerEvent::Connected { session }) = hub.receive_event_once()? {
connected = Some(session);
break;
}
}
let session = connected.ok_or_else(|| anyhow!("hub did not receive Connect"))?;
let _ = client.receive_once()?;
assert!(client.connected());
Ok(session)
}
fn receive_frame(hub: &mut CultNetRudpServerHub) -> Result<CultNetRudpServerEvent> {
for _ in 0..20 {
if let Some(event @ CultNetRudpServerEvent::Frame { .. }) = hub.receive_event_once()? {
return Ok(event);
}
}
Err(anyhow!("hub did not receive a frame"))
}
fn receive_message(client: &mut CultNetRudpSocketTransportConnection) -> Result<CultNetMessage> {
for _ in 0..20 {
if let Some(message) = client.receive_schema_message_once()? {
return Ok(message);
}
}
Err(anyhow!("client did not receive a schema message"))
}
#[test]
fn hub_keeps_independent_peer_sessions_and_exposes_connect_evidence() -> Result<()> {
let server_socket = socket()?;
let server_addr = server_socket.local_addr()?;
let mut hub = CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new(
"odin",
server_socket,
CONNECTION_ID,
))?;
let mut first = client(server_addr)?;
let mut second = client(server_addr)?;
let first_session = connect(&mut hub, &mut first, b"session-token-one")?;
let second_session = connect(&mut hub, &mut second, b"session-token-two")?;
assert_ne!(first_session.remote_addr, second_session.remote_addr);
assert_eq!(first_session.connect_payload, b"session-token-one");
assert_eq!(second_session.connect_payload, b"session-token-two");
assert_eq!(hub.sessions().len(), 2);
let first_request = create_provider_session_request(
"register-first",
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_REGISTRATION_SCHEMA,
&CultMeshProviderRegistrationWire {
provider_id: "first".to_string(),
service_instance_id: "first-1".to_string(),
endpoint_id: "first-public".to_string(),
verse_id: "public".to_string(),
requested_lease_duration_ms: 30_000,
authority_lease_id: None,
},
Some("first-1".to_string()),
Some("odin".to_string()),
)?;
let second_request = create_provider_session_request(
"register-second",
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_REGISTRATION_SCHEMA,
&CultMeshProviderRegistrationWire {
provider_id: "second".to_string(),
service_instance_id: "second-1".to_string(),
endpoint_id: "second-public".to_string(),
verse_id: "public".to_string(),
requested_lease_duration_ms: 30_000,
authority_lease_id: None,
},
Some("second-1".to_string()),
Some("odin".to_string()),
)?;
first.send_schema_message(&first_request)?;
second.send_schema_message(&second_request)?;
for _ in 0..2 {
let CultNetRudpServerEvent::Frame { session, frame } = receive_frame(&mut hub)? else {
unreachable!();
};
assert_eq!(frame.channel_id, "schema");
let request = decode_cultnet_message_from_slice(
&frame.payload,
CultNetWireContract::CultNetSchemaV0,
)?;
let CultNetMessage::OperationRequest { message_id, .. } = &request else {
return Err(anyhow!("expected operation request"));
};
let provider_id = if message_id == "register-first" {
"first"
} else {
"second"
};
let lease = CultMeshProviderLeaseWire {
provider_id: provider_id.to_string(),
service_instance_id: format!("{provider_id}-1"),
endpoint_id: format!("{provider_id}-public"),
verse_id: "public".to_string(),
lease_id: format!("lease-{provider_id}"),
valid_from_utc: "2026-07-14T12:00:00Z".to_string(),
expires_at_utc: "2026-07-14T12:00:30Z".to_string(),
};
let response = create_provider_session_response(
&request,
CultMeshProviderOperationStatus::Ok,
CULTMESH_PROVIDER_LEASE_SCHEMA,
&lease,
Some("odin".to_string()),
)?;
hub.send_schema_message(&session, &response)?;
}
let first_response = receive_message(&mut first)?;
let second_response = receive_message(&mut second)?;
let (_, first_lease) = decode_provider_session_response::<CultMeshProviderLeaseWire>(
&first_response,
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_LEASE_SCHEMA,
)?;
let (_, second_lease) = decode_provider_session_response::<CultMeshProviderLeaseWire>(
&second_response,
CULTMESH_PROVIDER_REGISTER_OPERATION,
CULTMESH_PROVIDER_LEASE_SCHEMA,
)?;
assert_eq!(first_lease.provider_id, "first");
assert_eq!(second_lease.provider_id, "second");
assert_eq!(hub.session(first_session.remote_addr), Some(&first_session));
assert_eq!(
hub.session(second_session.remote_addr),
Some(&second_session)
);
Ok(())
}
#[test]
fn hub_fences_replaced_generations_and_does_not_replace_connect_retransmits() -> Result<()> {
let server_socket = socket()?;
let server_addr = server_socket.local_addr()?;
let mut hub = CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new(
"odin",
server_socket,
CONNECTION_ID,
))?;
let mut provider = client(server_addr)?;
let original_evidence =
encode_provider_connect_evidence(&CultMeshProviderConnectEvidenceWire {
client_session_id: "provider-client-1".to_string(),
session_token: Some("shared-session-token".to_string()),
})?;
let replacement_evidence =
encode_provider_connect_evidence(&CultMeshProviderConnectEvidenceWire {
client_session_id: "provider-client-2".to_string(),
session_token: Some("shared-session-token".to_string()),
})?;
let original = connect(&mut hub, &mut provider, &original_evidence)?;
provider.connect(original_evidence.clone())?;
for _ in 0..3 {
assert!(hub.receive_event_once()?.is_none());
}
assert_eq!(hub.session(original.remote_addr), Some(&original));
provider.connect(replacement_evidence.clone())?;
let mut disconnected = None;
for _ in 0..20 {
if let Some(event @ CultNetRudpServerEvent::Disconnected { .. }) =
hub.receive_event_once()?
{
disconnected = Some(event);
break;
}
}
let disconnected =
disconnected.ok_or_else(|| anyhow!("replacement did not disconnect old generation"))?;
let CultNetRudpServerEvent::Disconnected { session, .. } = disconnected else {
return Err(anyhow!("expected old generation Disconnected event"));
};
assert_eq!(session, original);
let connected = hub
.receive_event_once()?
.ok_or_else(|| anyhow!("replacement did not connect new generation"))?;
let CultNetRudpServerEvent::Connected {
session: replacement,
} = connected
else {
return Err(anyhow!("expected replacement Connected event"));
};
assert_ne!(replacement.session_generation, original.session_generation);
let original_decoded = decode_provider_connect_evidence(&original.connect_payload)?;
let replacement_decoded = decode_provider_connect_evidence(&replacement.connect_payload)?;
assert_eq!(
original_decoded.session_token,
replacement_decoded.session_token
);
assert_ne!(
original_decoded.client_session_id,
replacement_decoded.client_session_id
);
assert!(
hub.send(&original, "schema", vec![0x80])
.unwrap_err()
.to_string()
.contains("no longer active")
);
assert_eq!(hub.session(replacement.remote_addr), Some(&replacement));
Ok(())
}
#[test]
fn identical_connect_retransmit_reuses_pending_accept_after_loss() -> Result<()> {
let server_socket = socket()?;
let server_addr = server_socket.local_addr()?;
let mut hub = CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new(
"odin",
server_socket,
CONNECTION_ID,
))?;
let client_socket = socket()?;
let mut client_session = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: CONNECTION_ID,
initial_sequence: 1,
resend_delay_ms: 250,
max_pending_reliable_packets: None,
});
let evidence = encode_provider_connect_evidence(&CultMeshProviderConnectEvidenceWire {
client_session_id: "provider-client-lost-accept".to_string(),
session_token: Some("shared-session-token".to_string()),
})?;
let connect_packet = client_session.create_connect(0, evidence)?;
let connect_wire = encode_rudp_packet(&connect_packet)?;
client_socket.send_to(&connect_wire, server_addr)?;
let CultNetRudpServerEvent::Connected { session } = hub
.receive_event_once()?
.ok_or_else(|| anyhow!("hub did not accept first Connect"))?
else {
return Err(anyhow!("hub did not emit Connected"));
};
let mut buffer = vec![0_u8; 65_535];
let (first_len, _) = client_socket.recv_from(&mut buffer)?;
let lost_accept = decode_rudp_packet(&buffer[..first_len])?;
assert_eq!(lost_accept.packet_type, CultNetRudpPacketType::Accept);
client_socket.send_to(&connect_wire, server_addr)?;
assert!(hub.receive_event_once()?.is_none());
let (resent_len, _) = client_socket.recv_from(&mut buffer)?;
let resent_accept = decode_rudp_packet(&buffer[..resent_len])?;
assert_eq!(resent_accept, lost_accept);
assert_eq!(hub.session(session.remote_addr), Some(&session));
client_session.receive(&resent_accept, 1)?;
assert!(client_session.connected());
Ok(())
}
#[test]
fn exact_reliable_acks_clear_more_than_one_ack_window_of_fragments() -> Result<()> {
let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: CONNECTION_ID,
initial_sequence: 1,
resend_delay_ms: 250,
max_pending_reliable_packets: None,
});
let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: CONNECTION_ID,
initial_sequence: 1,
resend_delay_ms: 250,
max_pending_reliable_packets: None,
});
let connect_packet = sender.create_connect(0, Vec::new())?;
let accept = receiver.accept_connect(&connect_packet, 0, Vec::new())?;
sender.receive(&accept, 0)?;
receiver.receive(&sender.create_ack_for(accept.sequence), 0)?;
let packets = sender.send_many(
"schema",
vec![0x5a; 40],
CultNetRudpSendOptions {
reliable: true,
ordered: true,
sequenced: false,
now_ms: 1,
},
Some(1),
)?;
assert_eq!(packets.len(), 40);
let mut delivered = Vec::new();
for packet in &packets {
let result = receiver.receive(packet, 2)?;
delivered.extend(result.delivered);
sender.receive(&receiver.create_ack_for(packet.sequence), 3)?;
}
assert_eq!(delivered.len(), 1);
assert_eq!(delivered[0].payload, vec![0x5a; 40]);
assert!(sender.pending_reliable_sequences().is_empty());
Ok(())
}
#[test]
fn session_and_hub_reject_configured_memory_bounds() -> Result<()> {
let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: CONNECTION_ID,
initial_sequence: 1,
resend_delay_ms: 250,
max_pending_reliable_packets: None,
});
receiver.set_max_payload_bytes(Some(4));
receiver.set_max_pending_fragment_sets(1)?;
let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: CONNECTION_ID,
initial_sequence: 1,
resend_delay_ms: 250,
max_pending_reliable_packets: None,
});
let connect_packet = sender.create_connect(0, Vec::new())?;
let accept = receiver.accept_connect(&connect_packet, 0, Vec::new())?;
sender.receive(&accept, 0)?;
let oversized = sender.send_many(
"schema",
vec![1, 2, 3, 4, 5],
CultNetRudpSendOptions {
reliable: true,
ordered: true,
sequenced: false,
now_ms: 1,
},
Some(2),
)?;
receiver.receive(&oversized[0], 1)?;
receiver.receive(&oversized[1], 1)?;
assert!(receiver.receive(&oversized[2], 1).is_err());
let fragment_options = CultNetRudpSendOptions {
reliable: true,
ordered: true,
sequenced: false,
now_ms: 2,
};
let first_set = sender.send_many(
"schema",
vec![1, 2, 3, 4],
fragment_options.clone(),
Some(2),
)?;
let second_set = sender.send_many("schema", vec![5, 6, 7, 8], fragment_options, Some(2))?;
receiver.receive(&first_set[0], 2)?;
assert!(receiver.receive(&second_set[0], 2).is_err());
let server_socket = socket()?;
let server_addr = server_socket.local_addr()?;
let mut options = CultNetRudpServerHubOptions::new("odin", server_socket, CONNECTION_ID);
options.max_peers = 1;
let mut hub = CultNetRudpServerHub::new(options)?;
let mut first = client(server_addr)?;
let mut second = client(server_addr)?;
connect(&mut hub, &mut first, b"first")?;
second.connect(b"second".to_vec())?;
for _ in 0..20 {
match hub.receive_event_once() {
Err(error) => {
assert!(error.to_string().contains("peer limit"));
return Ok(());
}
Ok(_) => continue,
}
}
Err(anyhow!("hub did not enforce peer limit"))
}
#[test]
fn replay_history_is_bounded_and_idle_hub_sessions_expire() -> Result<()> {
let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: CONNECTION_ID,
initial_sequence: 1,
resend_delay_ms: 250,
max_pending_reliable_packets: None,
});
let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: CONNECTION_ID,
initial_sequence: 1,
resend_delay_ms: 250,
max_pending_reliable_packets: None,
});
let connect_packet = sender.create_connect(0, Vec::new())?;
let accept = receiver.accept_connect(&connect_packet, 0, Vec::new())?;
sender.receive(&accept, 0)?;
let mut oldest = None;
for sequence in 0..4_100 {
let packet = sender.send(
"realtime",
vec![(sequence % 251) as u8],
CultNetRudpSendOptions {
reliable: false,
ordered: false,
sequenced: false,
now_ms: sequence,
},
)?;
if oldest.is_none() {
oldest = Some(packet.clone());
}
assert_eq!(receiver.receive(&packet, sequence)?.delivered.len(), 1);
}
assert!(
receiver
.receive(&oldest.expect("oldest packet"), 4_101)?
.delivered
.is_empty()
);
let server_socket = socket()?;
let server_addr = server_socket.local_addr()?;
let mut hub = CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new(
"odin",
server_socket,
CONNECTION_ID,
))?;
let mut provider = client(server_addr)?;
let context = connect(&mut hub, &mut provider, b"expiring-session")?;
std::thread::sleep(Duration::from_millis(2));
assert_eq!(hub.remove_timed_out_sessions(0), vec![context]);
assert!(hub.sessions().is_empty());
Ok(())
}