From ba2ef582d2cba5893d728679001bb602b69519ca Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 19 Jul 2026 00:49:27 +0100 Subject: [PATCH] Add Rust provider session transport --- packages/cultnet-rs/src/lib.rs | 2 + packages/cultnet-rs/src/provider_session.rs | 625 ++++++++++++++++++ packages/cultnet-rs/src/rudp.rs | 500 +++++++++++++- packages/cultnet-rs/tests/provider_session.rs | 365 ++++++++++ packages/cultnet-rs/tests/rudp_server_hub.rs | 459 +++++++++++++ 5 files changed, 1939 insertions(+), 12 deletions(-) create mode 100644 packages/cultnet-rs/src/provider_session.rs create mode 100644 packages/cultnet-rs/tests/provider_session.rs create mode 100644 packages/cultnet-rs/tests/rudp_server_hub.rs diff --git a/packages/cultnet-rs/src/lib.rs b/packages/cultnet-rs/src/lib.rs index 0b9fbb55..6fb1d38d 100644 --- a/packages/cultnet-rs/src/lib.rs +++ b/packages/cultnet-rs/src/lib.rs @@ -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::*; diff --git a/packages/cultnet-rs/src/provider_session.rs b/packages/cultnet-rs/src/provider_session.rs new file mode 100644 index 00000000..9965b0ef --- /dev/null +++ b/packages/cultnet-rs/src/provider_session.rs @@ -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 { + 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, +} + +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, +} + +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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub publication_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub command_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub receipt_id: Option, +} + +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(value: &T) -> Result +where + T: CultMeshProviderSessionPayload + Serialize, +{ + value.validate()?; + Ok(STANDARD.encode(rmp_serde::to_vec_named(value)?)) +} + +pub fn encode_provider_connect_evidence( + evidence: &CultMeshProviderConnectEvidenceWire, +) -> Result> { + evidence.validate()?; + Ok(rmp_serde::to_vec_named(evidence)?) +} + +pub fn decode_provider_connect_evidence( + payload: &[u8], +) -> Result { + 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(payload: &str) -> Result +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 { + 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( + message_id: impl Into, + operation: impl Into, + payload_schema: impl Into, + payload: &T, + source_runtime_id: Option, + target_runtime_id: Option, +) -> Result +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( + request: &CultNetMessage, + expected_operation: &str, + expected_payload_schema: &str, +) -> Result +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( + request: &CultNetMessage, + status: CultMeshProviderOperationStatus, + payload_schema: impl Into, + payload: &T, + source_runtime_id: Option, +) -> Result +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, + source_runtime_id: Option, +) -> Result { + 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( + 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> { + require_text(value, field)?; + DateTime::parse_from_rfc3339(value) + .map_err(|error| anyhow!("Provider-session {field} must be RFC3339: {error}")) +} diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 8b47f9e7..ae1ef00b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -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, + max_payload_bytes: Option, + 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 { + 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 { self.last_received_at_ms } + pub fn set_max_payload_bytes(&mut self, max_payload_bytes: Option) { + 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) -> 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> { 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::() + }) + .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>, } -impl CultNetRudpSocketTransportConnection { - pub fn new(options: CultNetRudpSocketTransportOptions) -> Result { +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, + pub max_payload_bytes: Option, + pub max_fragment_bytes: Option, + pub max_pending_reliable_packets: Option, + pub max_peers: usize, +} + +impl CultNetRudpServerHubOptions { + pub fn new(runtime_id: impl Into, 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, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum CultNetRudpServerEvent { + Connected { + session: CultNetRudpServerSessionContext, + }, + Frame { + session: CultNetRudpServerSessionContext, + frame: CultNetTransportFrame, + }, + Pong { + session: CultNetRudpServerSessionContext, + payload: Vec, + }, + Disconnected { + session: CultNetRudpServerSessionContext, + reason: Vec, + }, +} + +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, + max_payload_bytes: usize, + max_fragment_bytes: Option, + max_peers: usize, + next_session_generation: u64, + peers: BTreeMap, + pending_events: VecDeque, + pub profile: CultNetTransportProfile, + stats: CultNetTransportStats, +} + +impl CultNetRudpServerHub { + pub fn new(options: CultNetRudpServerHubOptions) -> Result { 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 { + Ok(self.socket.local_addr()?) + } + + pub fn stats(&self) -> CultNetTransportStats { + self.stats.clone() + } + + pub fn sessions(&self) -> Vec { + 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, + ) -> 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, + ) -> Result { + 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> { + 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 { + 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::>(); + 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 { + 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) diff --git a/packages/cultnet-rs/tests/provider_session.rs b/packages/cultnet-rs/tests/provider_session.rs new file mode 100644 index 00000000..4ad7c23a --- /dev/null +++ b/packages/cultnet-rs/tests/provider_session.rs @@ -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(®istration())?; + assert_eq!(encoded, REGISTRATION_GOLDEN); + assert_eq!( + decode_provider_session_payload::(&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, + ®istration(), + 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::( + &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::( + &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, + ®istration(), + 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::( + &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::( + &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::( + &encode_provider_session_payload(&receipt)? + )?, + receipt + ); + + let withdrawal = CultMeshProviderWithdrawalWire { + lease_id: "lease-1".to_string(), + }; + assert_eq!( + decode_provider_session_payload::( + &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::( + &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, + ®istration(), + 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::(&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(()) +} diff --git a/packages/cultnet-rs/tests/rudp_server_hub.rs b/packages/cultnet-rs/tests/rudp_server_hub.rs new file mode 100644 index 00000000..a8a2c114 --- /dev/null +++ b/packages/cultnet-rs/tests/rudp_server_hub.rs @@ -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 { + 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 { + 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 { + 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 { + 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 { + 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::( + &first_response, + CULTMESH_PROVIDER_REGISTER_OPERATION, + CULTMESH_PROVIDER_LEASE_SCHEMA, + )?; + let (_, second_lease) = decode_provider_session_response::( + &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(()) +}