From 4136167b417718f2ff517e145eb7d8b57cb9724f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 29 Oct 2026 18:27:12 +0100 Subject: [PATCH] cultnet-rs: IdunnRuntimePresencePublisher, Odin RUDP delivery of signed presence and the Warming lease wait Endpoint from Expected's one odin.verse-rendezvous shared-infrastructure dependency; one RUDP session per publish; ack is delivery, an Odin error frame is a refusal by code; message id from the record's sequence; the signer is borrowed per call. Seam tests run against cultmesh-rs's real document server. Co-Authored-By: Claude Opus 5.5 --- packages/cultnet-rs/Cargo.toml | 1 + .../cultnet-rs/src/idunn_runtime_provider.rs | 24 +- .../idunn_runtime_provider/odin_publisher.rs | 424 ++++++++++++ .../src/idunn_runtime_provider/tests.rs | 13 +- .../tests/odin_publisher.rs | 635 ++++++++++++++++++ 5 files changed, 1088 insertions(+), 9 deletions(-) create mode 100644 packages/cultnet-rs/src/idunn_runtime_provider/odin_publisher.rs create mode 100644 packages/cultnet-rs/src/idunn_runtime_provider/tests/odin_publisher.rs diff --git a/packages/cultnet-rs/Cargo.toml b/packages/cultnet-rs/Cargo.toml index 77f368d4..a26a3eca 100644 --- a/packages/cultnet-rs/Cargo.toml +++ b/packages/cultnet-rs/Cargo.toml @@ -37,6 +37,7 @@ libc = "0.2" windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Security_Cryptography"] } [dev-dependencies] +cultmesh-rs = { path = "../cultmesh-rs" } pretty_assertions = "1.4" tempfile = "3" diff --git a/packages/cultnet-rs/src/idunn_runtime_provider.rs b/packages/cultnet-rs/src/idunn_runtime_provider.rs index 6e4173ed..21b00997 100644 --- a/packages/cultnet-rs/src/idunn_runtime_provider.rs +++ b/packages/cultnet-rs/src/idunn_runtime_provider.rs @@ -4,8 +4,10 @@ //! runtime bundle with the two signing credentials Idunn hands it, reports its //! health as it changes, and signs presence records that Odin can authenticate //! and Idunn's route challenge can read. This module owns that provider half and -//! owns no network; publishing the signed records is a separate concern that -//! consumes [`IdunnRuntimePresenceSigner::sign`]. +//! owns no network; [`IdunnRuntimePresencePublisher`] delivers the signed records +//! to Odin and waits for Idunn's write lease while Warming. A target whose +//! Expected declares no Odin dependency is route-proof: `for_signer` returns +//! `None` and `answer_route_observation` is its whole presence. //! //! The module compiles and runs on every platform an Idunn driver launches: //! Idunn's systemd driver (unix) delivers the credentials as inherited @@ -14,9 +16,11 @@ //! lock are unix-only. //! //! ```no_run +//! use std::time::Duration; +//! //! use cultnet_rs::{ //! GameCultRuntimeCapability, IdunnRuntimeAuthority, IdunnRuntimeCredentials, -//! IdunnRuntimePresenceSigner, RuntimePresenceState, +//! IdunnRuntimePresencePublisher, IdunnRuntimePresenceSigner, RuntimePresenceState, //! }; //! //! # fn main() -> anyhow::Result<()> { @@ -40,10 +44,13 @@ //! capacity: 1, //! }], //! ); -//! let warming = signer.sign("starting")?; // the signer starts Warming -//! signer.report_health(RuntimePresenceState::Active); -//! let active = signer.sign("ready")?; -//! # let _ = (warming, active); +//! // Odin's endpoint comes from Expected; the publisher borrows the signer per call. +//! if let Some(publisher) = IdunnRuntimePresencePublisher::for_signer(&signer)? { +//! // The signer starts Warming; heartbeats until Idunn writes the lease. +//! let _lease = publisher.wait_for_write_lease(&mut signer, Duration::from_secs(60))?; +//! signer.report_health(RuntimePresenceState::Active); +//! publisher.publish(&mut signer, "ready")?; +//! } //! # Ok(()) //! # } //! ``` @@ -1400,5 +1407,8 @@ fn unix_millis() -> Result { .context("Unix time exceeds u64") } +mod odin_publisher; +pub use odin_publisher::IdunnRuntimePresencePublisher; + #[cfg(test)] mod tests; diff --git a/packages/cultnet-rs/src/idunn_runtime_provider/odin_publisher.rs b/packages/cultnet-rs/src/idunn_runtime_provider/odin_publisher.rs new file mode 100644 index 00000000..15d65973 --- /dev/null +++ b/packages/cultnet-rs/src/idunn_runtime_provider/odin_publisher.rs @@ -0,0 +1,424 @@ +//! Delivery of signed runtime presence to Odin's document catalog. +//! +//! [`IdunnRuntimePresencePublisher`] is the network half of the Idunn provider. +//! It owns where signed presence goes and how it gets there; the signer it +//! borrows on every call stays the one owner of health, sequence, Warming proofs +//! and the write-lease guard. The publisher keeps no state of its own: Odin's +//! endpoint is read once from the one `shared-infrastructure` +//! `odin.verse-rendezvous` dependency in Expected, and each publish is one +//! fresh RUDP session, so an Odin restart needs no reconnect machinery. +//! +//! Blocking `std::net` only. It builds and runs wherever the provider does. + +use std::net::{ToSocketAddrs, UdpSocket}; +use std::time::{Duration, Instant}; + +use anyhow::{Result, bail, ensure}; + +use super::{ + IdunnRuntimePresenceSigner, ProcessWriteLeaseGuard, RuntimePresenceState, + SignedRuntimePresence, +}; +use crate::{ + CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, CultNetErrorCode, CultNetMessage, + CultNetRudpReliableSendStatus, CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, CultNetWireContract, encode_cultnet_message_to_vec, +}; + +const ODIN_DEPENDENCY_KIND: &str = "shared-infrastructure"; +const ODIN_CAPABILITY: &str = "odin.verse-rendezvous"; +/// Names the field in refusals; never the value. +const ENDPOINT_FIELD: &str = "Expected odin.verse-rendezvous dependency provider_endpoint"; +const RUDP_SCHEME: &str = "rudp://"; +/// How long one socket read waits, so the delivery loops poll without spinning. +const RECEIVE_SLICE: Duration = Duration::from_millis(25); +const RESEND_DELAY_MS: u64 = 100; + +/// The publisher's timing. Crate-private: tests shorten it, consumers do not +/// tune it. A fixed 5 s heartbeat under Idunn's own Warming deadline needs no +/// backoff. +#[derive(Clone, Copy, Debug)] +pub(super) struct Cadence { + pub(super) heartbeat: Duration, + pub(super) lease_poll: Duration, + pub(super) connect: Duration, + pub(super) ack: Duration, +} + +impl Cadence { + const PRODUCTION: Self = Self { + heartbeat: Duration::from_secs(5), + lease_poll: Duration::from_millis(100), + connect: Duration::from_secs(2), + ack: Duration::from_secs(2), + }; +} + +/// Why one delivery failed. Each variant is a fixed code: nothing here carries +/// Odin's text, an address or a path. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum DeliveryFailure { + Unresolved, + NoSocket, + ConnectTimeout, + SendFailed, + AckTimeout, + SendInvalidated, + ReplyUnreadable, + SessionEnded, + Refused(Option), +} + +impl std::fmt::Display for DeliveryFailure { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("Odin delivery failed: ")?; + match self { + Self::Unresolved => formatter.write_str("endpoint_unresolved"), + Self::NoSocket => formatter.write_str("socket_unavailable"), + Self::ConnectTimeout => formatter.write_str("connect_timeout"), + Self::SendFailed => formatter.write_str("send_failed"), + Self::AckTimeout => formatter.write_str("ack_timeout"), + Self::SendInvalidated => formatter.write_str("send_invalidated"), + Self::ReplyUnreadable => formatter.write_str("reply_unreadable"), + Self::SessionEnded => formatter.write_str("session_ended"), + Self::Refused(Some(code)) => write!(formatter, "odin_refused code {code:?}"), + Self::Refused(None) => formatter.write_str("odin_refused without a code"), + } + } +} + +impl std::error::Error for DeliveryFailure {} + +/// Delivers an Idunn-managed runtime's signed presence to Odin and waits for +/// Idunn's write lease while Warming. +/// +/// Built from a signer, it borrows that signer on every call, so route answers +/// and publishes share one signer and one sequence. Health is whatever the +/// signer was last told with +/// [`report_health`](IdunnRuntimePresenceSigner::report_health); the publisher +/// has no state parameter. +pub struct IdunnRuntimePresencePublisher { + host: String, + port: u16, + expected_sha256: String, + runtime_instance_id: String, + cadence: Cadence, +} + +impl std::fmt::Debug for IdunnRuntimePresencePublisher { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("IdunnRuntimePresencePublisher") + .finish_non_exhaustive() + } +} + +impl IdunnRuntimePresencePublisher { + /// The publisher for `signer`'s Odin, or `None` when Expected declares no + /// `shared-infrastructure` `odin.verse-rendezvous` dependency. Such a + /// target is route-proof: Idunn reads its presence through the route + /// challenge, which [`answer_route_observation`](IdunnRuntimePresenceSigner::answer_route_observation) + /// answers with no Odin. + /// + /// Refuses two such dependencies, a missing `provider_endpoint`, a scheme + /// other than `rudp://` and an endpoint that is not `host:port`. A refusal + /// names the field, never the value. The host is resolved at each publish. + pub fn for_signer(signer: &IdunnRuntimePresenceSigner) -> Result> { + let records = &signer.authority.records; + let mut odin = records.expected.dependencies.iter().filter(|dependency| { + dependency.kind == ODIN_DEPENDENCY_KIND && dependency.capability == ODIN_CAPABILITY + }); + let Some(dependency) = odin.next() else { + return Ok(None); + }; + ensure!( + odin.next().is_none(), + "Expected declares more than one odin.verse-rendezvous shared-infrastructure dependency" + ); + let Some(endpoint) = dependency.provider_endpoint.as_deref() else { + bail!("{ENDPOINT_FIELD} is missing"); + }; + let Some((host, port)) = parse_endpoint(endpoint) else { + bail!("{ENDPOINT_FIELD} must be rudp://host:port or host:port"); + }; + Ok(Some(Self { + host, + port, + expected_sha256: records.expected_sha256.clone(), + runtime_instance_id: records.activation.runtime_instance_id.clone(), + cadence: Cadence::PRODUCTION, + })) + } + + #[cfg(test)] + pub(super) fn with_cadence(mut self, cadence: Cadence) -> Self { + self.cadence = cadence; + self + } + + /// Signs one presence in the signer's reported health and delivers it. It + /// returns after Odin acknowledged the document. The ack proves delivery; + /// Odin's correlation owns admission. A CultNet error frame from Odin is a + /// refusal naming its code, never its text. + pub fn publish( + &self, + signer: &mut IdunnRuntimePresenceSigner, + detail: &str, + ) -> Result { + let (signed, delivery) = self.sign_and_deliver(signer, detail)?; + delivery?; + Ok(signed) + } + + /// Waits for Idunn's process write lease, publishing a Warming heartbeat on + /// entry and every 5 s and polling for the lease every 100 ms. The signer + /// must report Warming; the wait is refused before any publish otherwise. + /// Returns `Ok(None)` at once when Expected requires no lease. + /// + /// The caller sets `timeout` because Idunn owns the Warming deadline. A + /// heartbeat that fails delivery is retried at the next heartbeat and does + /// not end the wait: the signer remembers its Warming proofs whether or not + /// delivery succeeded. A lease record that binds another incarnation ends + /// the wait at once. A timeout names the count of failed deliveries and the + /// last one's code. + pub fn wait_for_write_lease( + &self, + signer: &mut IdunnRuntimePresenceSigner, + timeout: Duration, + ) -> Result> { + self.require_own_signer(signer)?; + if !signer.authority.records.expected.write_lease_required { + return Ok(None); + } + ensure!( + signer.health() == RuntimePresenceState::Warming, + "waiting for the write lease requires the signer to report Warming" + ); + let started = Instant::now(); + let mut next_heartbeat = started; + let mut failed_deliveries = 0_u32; + let mut last_failure = None; + loop { + let now = Instant::now(); + if now >= next_heartbeat { + let (_, delivery) = self.sign_and_deliver(signer, "waiting-for-process-write-lease")?; + if let Err(failure) = delivery { + failed_deliveries += 1; + last_failure = Some(failure); + } + next_heartbeat = now + self.cadence.heartbeat; + } + if let Some(guard) = signer.try_acquire_write_lease()? { + return Ok(Some(guard)); + } + let Some(remaining) = timeout.checked_sub(started.elapsed()) else { + bail!( + "timed out waiting for Idunn's process write lease: {failed_deliveries} failed deliveries{}", + last_failure.map_or(String::new(), |failure| format!(", last: {failure}")) + ); + }; + std::thread::sleep(self.cadence.lease_poll.min(remaining)); + } + } + + /// Every presence that reaches Odin goes through here. The outer error is a + /// refusal before delivery; the inner result is the delivery. + fn sign_and_deliver( + &self, + signer: &mut IdunnRuntimePresenceSigner, + detail: &str, + ) -> Result<(SignedRuntimePresence, Result<(), DeliveryFailure>)> { + self.require_own_signer(signer)?; + let signed = signer.sign(detail)?; + let delivery = self.deliver(&signed); + Ok((signed, delivery)) + } + + fn require_own_signer(&self, signer: &IdunnRuntimePresenceSigner) -> Result<()> { + let records = &signer.authority.records; + ensure!( + records.expected_sha256 == self.expected_sha256 + && records.activation.runtime_instance_id == self.runtime_instance_id, + "this publisher was built for another runtime authority" + ); + Ok(()) + } + + /// One fresh session: connect, one reliable `DocumentPutRaw` on `schema`, + /// wait for the ack. The message id comes from the signed record, so the + /// publisher keeps no counter. + fn deliver(&self, signed: &SignedRuntimePresence) -> Result<(), DeliveryFailure> { + let remote = (self.host.as_str(), self.port) + .to_socket_addrs() + .ok() + .and_then(|mut addresses| addresses.next()) + .ok_or(DeliveryFailure::Unresolved)?; + let socket = UdpSocket::bind(if remote.is_ipv4() { + "0.0.0.0:0" + } else { + "[::]:0" + }) + .map_err(|_| DeliveryFailure::NoSocket)?; + socket + .set_read_timeout(Some(RECEIVE_SLICE)) + .map_err(|_| DeliveryFailure::NoSocket)?; + let mut options = CultNetRudpSocketTransportOptions::client( + signed.record.runtime_id.clone(), + socket, + remote, + CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, + ); + options.resend_delay_ms = RESEND_DELAY_MS; + let mut transport = CultNetRudpSocketTransportConnection::new(options) + .map_err(|_| DeliveryFailure::NoSocket)?; + transport + .connect(Vec::new()) + .map_err(|_| DeliveryFailure::SendFailed)?; + let connect_deadline = Instant::now() + self.cadence.connect; + loop { + pump(&mut transport)?; + if transport.connected() { + break; + } + if Instant::now() >= connect_deadline { + return Err(DeliveryFailure::ConnectTimeout); + } + } + let message = CultNetMessage::DocumentPutRaw { + message_id: format!( + "runtime-presence:{}:{}:{}", + signed.record.target, + signed.record.runtime_instance_id, + signed.record.publisher_sequence + ), + document: signed.document.clone(), + }; + let payload = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0) + .map_err(|_| DeliveryFailure::SendFailed)?; + let receipt = transport + .send_reliable("schema", payload) + .map_err(|_| DeliveryFailure::SendFailed)?; + let ack_deadline = Instant::now() + self.cadence.ack; + loop { + pump(&mut transport)?; + match transport.reliable_send_status(&receipt) { + CultNetRudpReliableSendStatus::Acknowledged => break, + CultNetRudpReliableSendStatus::Invalidated => { + return Err(DeliveryFailure::SendInvalidated); + } + CultNetRudpReliableSendStatus::Pending => {} + } + if Instant::now() >= ack_deadline { + return Err(DeliveryFailure::AckTimeout); + } + } + // Best effort: the session is discarded either way. + let _ = transport.disconnect(Vec::new()); + Ok(()) + } +} + +/// Reads what Odin has sent, resends what is due and reports a refusal or an +/// ended session. Waits at most one socket read slice. +fn pump(transport: &mut CultNetRudpSocketTransportConnection) -> Result<(), DeliveryFailure> { + loop { + match transport.receive_schema_message_once() { + Ok(Some(CultNetMessage::Error { code, .. })) => { + return Err(DeliveryFailure::Refused(code)); + } + Ok(Some(_)) => {} + Ok(None) => break, + Err(_) => return Err(DeliveryFailure::ReplyUnreadable), + } + } + if transport.disconnect_reason().is_some() { + return Err(DeliveryFailure::SessionEnded); + } + transport + .poll_resends() + .map_err(|_| DeliveryFailure::SendFailed) +} + +/// `rudp://host:port` or bare `host:port` (an IPv6 host in brackets) as host +/// and port. Any other scheme, a path, a blank host or a zero port is `None`. +fn parse_endpoint(endpoint: &str) -> Option<(String, u16)> { + let rest = match endpoint.strip_prefix(RUDP_SCHEME) { + Some(rest) => rest, + None if endpoint.contains("://") => return None, + None => endpoint, + }; + let (host, port) = if let Some(bracketed) = rest.strip_prefix('[') { + let (host, after) = bracketed.split_once(']')?; + (host, after.strip_prefix(':')?) + } else { + let (host, port) = rest.rsplit_once(':')?; + if host.contains(':') { + return None; + } + (host, port) + }; + if host.is_empty() || host.contains(|c: char| c.is_whitespace() || c == '/') { + return None; + } + let port: u16 = port.parse().ok().filter(|port| *port != 0)?; + Some((host.to_owned(), port)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn parsed(endpoint: &str) -> Option<(&'static str, u16)> { + parse_endpoint(endpoint).map(|(host, port)| (Box::leak(host.into_boxed_str()) as &str, port)) + } + + #[test] + fn endpoints_are_rudp_or_bare_host_and_port() { + assert_eq!(parsed("rudp://odin.example:7001"), Some(("odin.example", 7001))); + assert_eq!(parsed("10.0.0.5:7001"), Some(("10.0.0.5", 7001))); + assert_eq!(parsed("rudp://[::1]:7001"), Some(("::1", 7001))); + assert_eq!(parsed("[fd00::5]:65535"), Some(("fd00::5", 65535))); + for bad in [ + "", + "odin.example", + "odin.example:0", + "odin.example:65536", + "odin.example:port", + ":7001", + "udp://odin.example:7001", + "http://odin.example:7001", + "rudp://odin.example:7001/x", + "odin example:7001", + "::1:7001", + "[::1]", + "[::1]7001", + "rudp://rudp://odin.example:7001", + ] { + assert_eq!(parse_endpoint(bad), None, "{bad:?}"); + } + } + + #[test] + fn production_cadence_is_the_ruled_one() { + let cadence = Cadence::PRODUCTION; + assert_eq!(cadence.heartbeat, Duration::from_secs(5)); + assert_eq!(cadence.lease_poll, Duration::from_millis(100)); + assert_eq!(cadence.connect, Duration::from_secs(2)); + assert_eq!(cadence.ack, Duration::from_secs(2)); + } + + /// Odin's server sends error frames without a code, so a coded refusal is + /// reachable only here. + #[test] + fn a_refusal_names_its_code_and_nothing_else() { + let coded = DeliveryFailure::Refused(Some(CultNetErrorCode::WriteNotAuthorized)); + assert_eq!( + coded.to_string(), + "Odin delivery failed: odin_refused code WriteNotAuthorized" + ); + assert_eq!( + DeliveryFailure::Refused(None).to_string(), + "Odin delivery failed: odin_refused without a code" + ); + } +} diff --git a/packages/cultnet-rs/src/idunn_runtime_provider/tests.rs b/packages/cultnet-rs/src/idunn_runtime_provider/tests.rs index 9f2d53ec..3d9248a1 100644 --- a/packages/cultnet-rs/src/idunn_runtime_provider/tests.rs +++ b/packages/cultnet-rs/src/idunn_runtime_provider/tests.rs @@ -5,7 +5,7 @@ use cultcache_rs::{CacheBackingStore, CultCacheEnvelope}; use super::*; use crate::{ - IdunnExpectedCapability, IdunnExpectedRoute, IdunnRuntimeActivationLaunch, + IdunnExpectedCapability, IdunnExpectedDependency, IdunnExpectedRoute, IdunnRuntimeActivationLaunch, IdunnServiceIdentity, RUNTIME_PRESENCE_IDENTITY_FD_NAME, RuntimePresenceAuthenticationContext, VerifiedRuntimeAuthority, authenticate_runtime_presence_claim, correlate_runtime_presence_claim, enroll_service_identity_at, verify_runtime_authority, @@ -114,6 +114,13 @@ struct Rig { impl Rig { fn new(write_lease_required: bool) -> Result { + Self::with_dependencies(write_lease_required, Vec::new()) + } + + fn with_dependencies( + write_lease_required: bool, + dependencies: Vec, + ) -> Result { let dir = tempfile::tempdir()?; let provider_path = dir.path().join("provider.cc"); let provider = @@ -152,7 +159,7 @@ impl Rig { minimum_capacity: 1, }) .collect(), - dependencies: Vec::new(), + dependencies, }; let (activation, activation_seed) = Self::issue(&expected, &idunn)?; let rig = Self { @@ -2118,3 +2125,5 @@ fn host_credentials_keep_the_seed_when_the_provider_is_a_directory() -> Result<( assert!(seed.exists(), "a refused delivery leaves the one-shot seed"); Ok(()) } + +mod odin_publisher; diff --git a/packages/cultnet-rs/src/idunn_runtime_provider/tests/odin_publisher.rs b/packages/cultnet-rs/src/idunn_runtime_provider/tests/odin_publisher.rs new file mode 100644 index 00000000..f18622b0 --- /dev/null +++ b/packages/cultnet-rs/src/idunn_runtime_provider/tests/odin_publisher.rs @@ -0,0 +1,635 @@ +//! The publisher against cultmesh-rs's real RUDP document server, the receiver +//! Odin runs. cultmesh-rs links the non-test build of this crate, so the +//! server's types are a second instance of ours: only bytes and strings cross. + +use std::net::{SocketAddr, UdpSocket}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +use cultmesh_rs::{ + CultMeshRudpDocumentServer, CultMeshRudpDocumentServerOptions, CultMeshRudpPollOutcome, + CultMeshRudpRawDocumentReceipt, CultMeshRudpSnapshotQuery, CultMeshSystemClock, +}; + +use super::*; +use crate::idunn_runtime_provider::odin_publisher::Cadence; + +const ODIN_CAPABILITY: &str = "odin.verse-rendezvous"; +/// Text a refusing sink carries; Odin's error frame must not repeat it. +const SINK_CANARY: &str = "SINK-CANARY-8841"; + +const TEST_CADENCE: Cadence = Cadence { + heartbeat: Duration::from_millis(100), + lease_poll: Duration::from_millis(5), + connect: Duration::from_millis(500), + ack: Duration::from_millis(500), +}; + +// --------------------------------------------------------------- Odin harness + +#[derive(Clone, Debug)] +struct Delivered { + message_id: String, + record_key: String, + schema_id: String, + payload: Vec, + from: SocketAddr, +} + +impl Delivered { + fn record(&self) -> GameCultRuntimePresenceHealthRecord { + rmp_serde::from_slice(&self.payload).expect("a delivered presence record decodes") + } +} + +#[derive(Clone, Copy)] +enum Sink { + Accept, + Refuse, +} + +/// cultmesh-rs's document server on a thread, with a sink that records or refuses. +struct Odin { + addr: SocketAddr, + received: Arc>>, + stop: Arc, + thread: Option>, +} + +impl Odin { + fn start(sink: Sink) -> Self { + Self::start_on("127.0.0.1:0".parse().unwrap(), sink) + } + + fn start_on(addr: SocketAddr, sink: Sink) -> Self { + let socket = UdpSocket::bind(addr).expect("the Odin test socket binds"); + let addr = socket.local_addr().unwrap(); + let received = Arc::new(Mutex::new(Vec::new())); + let stop = Arc::new(AtomicBool::new(false)); + let record = Arc::clone(&received); + let mut server = CultMeshRudpDocumentServer::new( + socket, + move |receipt: CultMeshRudpRawDocumentReceipt| -> Result<()> { + match sink { + Sink::Refuse => bail!(SINK_CANARY), + Sink::Accept => { + record.lock().unwrap().push(Delivered { + message_id: receipt.message_id, + record_key: receipt.document.record_key, + schema_id: receipt.document.schema_id, + payload: receipt.document.payload, + from: receipt.session.remote_addr, + }); + Ok(()) + } + } + }, + |_: &CultMeshRudpSnapshotQuery| -> Result> { Ok(Vec::new()) }, + CultMeshSystemClock::default(), + CultMeshRudpDocumentServerOptions::default(), + ) + .expect("the Odin test server starts"); + let running = Arc::clone(&stop); + let thread = std::thread::spawn(move || { + while !running.load(Ordering::Relaxed) { + if matches!(server.poll_once(), Ok(CultMeshRudpPollOutcome::Idle) | Err(_)) { + std::thread::sleep(Duration::from_millis(1)); + } + } + }); + Self { + addr, + received, + stop, + thread: Some(thread), + } + } + + fn deliveries(&self) -> Vec { + self.received.lock().unwrap().clone() + } + + fn count(&self) -> usize { + self.received.lock().unwrap().len() + } + + fn wait_for(&self, count: usize) -> Vec { + let give_up = Instant::now() + Duration::from_secs(10); + while self.count() < count { + assert!(Instant::now() < give_up, "Odin never received {count}"); + std::thread::sleep(Duration::from_millis(2)); + } + self.deliveries() + } + + fn rudp_endpoint(&self) -> String { + format!("rudp://{}", self.addr) + } +} + +impl Drop for Odin { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +/// An address nothing listens on. +fn dead_address() -> SocketAddr { + UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap() +} + +// ----------------------------------------------------------------------- rigs + +fn odin_dependency(schema: &str, endpoint: Option) -> IdunnExpectedDependency { + IdunnExpectedDependency { + kind: "shared-infrastructure".into(), + capability: ODIN_CAPABILITY.into(), + schema: schema.into(), + compatibility: "v1".into(), + minimum_capacity: 1, + startup: "before-start".into(), + provider_id: Some("odin-ygg-1".into()), + provider_authority: Some("managed-incarnation".into()), + provider_expected_projection_sha256: Some(digest('9')), + provider_endpoint: endpoint, + } +} + +fn rig_with( + write_lease_required: bool, + dependencies: Vec, +) -> Result { + Rig::with_dependencies(write_lease_required, dependencies) +} + +fn odin_rig(write_lease_required: bool, endpoint: &str) -> Result { + rig_with( + write_lease_required, + vec![odin_dependency("odin.verse.v1", Some(endpoint.into()))], + ) +} + +fn publisher_for( + signer: &IdunnRuntimePresenceSigner, + cadence: Cadence, +) -> Result { + Ok(IdunnRuntimePresencePublisher::for_signer(signer)? + .expect("Expected declares an Odin dependency") + .with_cadence(cadence)) +} + +// ------------------------------------------------------------------- endpoint + +#[test] +fn for_signer_reads_endpoint_from_expected() -> Result<()> { + let odin = Odin::start(Sink::Accept); + // `rudp://host:port` and bare `host:port` both reach the dependency's server. + for endpoint in [odin.rudp_endpoint(), odin.addr.to_string()] { + let rig = odin_rig(false, &endpoint)?; + let mut signer = rig.signer()?; + let publisher = publisher_for(&signer, TEST_CADENCE)?; + publisher.publish(&mut signer, "hello")?; + } + assert_eq!(odin.count(), 2); + + // No Odin dependency: the target is route-proof and has no publisher. + let rig = Rig::new(false)?; + assert!(IdunnRuntimePresencePublisher::for_signer(&rig.signer()?)?.is_none()); + + // Only a shared-infrastructure dependency on Odin is Odin. + let mut bootstrap = odin_dependency("odin.verse.v1", Some(odin.rudp_endpoint())); + bootstrap.kind = "bootstrap".into(); + let mut other = odin_dependency("odin.verse.v1", Some(odin.rudp_endpoint())); + other.capability = "telemetry".into(); + let rig = rig_with(false, vec![bootstrap])?; + assert!(IdunnRuntimePresencePublisher::for_signer(&rig.signer()?)?.is_none()); + let rig = rig_with(false, vec![other])?; + assert!(IdunnRuntimePresencePublisher::for_signer(&rig.signer()?)?.is_none()); + + // IPv6 hosts are bracketed. + let rig = odin_rig(false, "rudp://[::1]:9000")?; + assert!(IdunnRuntimePresencePublisher::for_signer(&rig.signer()?)?.is_some()); + Ok(()) +} + +#[test] +fn for_signer_refuses_what_is_not_one_usable_endpoint() -> Result<()> { + let canary = "canary-host.invalid"; + let two = rig_with( + false, + vec![ + odin_dependency("odin.verse.v1", Some(format!("{canary}:9000"))), + odin_dependency("odin.verse.v2", Some(format!("{canary}:9001"))), + ], + )?; + let refused = refusal(IdunnRuntimePresencePublisher::for_signer(&two.signer()?)); + assert!(refused.contains(ODIN_CAPABILITY), "{refused}"); + assert!(!refused.contains(canary), "{refused}"); + + let missing = rig_with(false, vec![odin_dependency("odin.verse.v1", None)])?; + let refused = refusal(IdunnRuntimePresencePublisher::for_signer(&missing.signer()?)); + assert!(refused.contains("provider_endpoint"), "{refused}"); + + for bad in [ + format!("udp://{canary}:9000"), + format!("http://{canary}:9000"), + format!("tcp://{canary}:9000"), + canary.to_string(), + format!("{canary}:0"), + format!("{canary}:notaport"), + format!("{canary}:9000/path"), + format!(":{}", 9000), + ] { + let rig = rig_with(false, vec![odin_dependency("odin.verse.v1", Some(bad.clone()))])?; + let refused = refusal(IdunnRuntimePresencePublisher::for_signer(&rig.signer()?)); + assert!( + refused.contains("provider_endpoint"), + "{bad} was not refused by field: {refused}" + ); + assert!(!refused.contains(canary), "{refused}"); + } + Ok(()) +} + +// -------------------------------------------------------------------- publish + +#[test] +fn publish_delivers_to_odin_catalog_server() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(false, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let publisher = publisher_for(&signer, TEST_CADENCE)?; + + let signed = publisher.publish(&mut signer, "hello")?; + + // The sink ran before the ack, so the document is already there. + let delivered = odin.deliveries(); + assert_eq!(delivered.len(), 1); + let delivered = &delivered[0]; + assert_eq!(delivered.payload, signed.document.payload); + assert_eq!(delivered.record_key, TARGET); + assert_eq!(delivered.schema_id, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA); + assert_eq!( + delivered.message_id, + format!( + "runtime-presence:{TARGET}:{}:{}", + signed.record.runtime_instance_id, signed.record.publisher_sequence + ) + ); + assert_eq!(signed.record.publisher_sequence, 1); + assert_odin_accepts(&rig, &signed, signed.record.observed_at_unix_millis + 1)?; + Ok(()) +} + +#[test] +fn publish_surfaces_odin_refusal() -> Result<()> { + let odin = Odin::start(Sink::Refuse); + let rig = odin_rig(false, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let publisher = publisher_for(&signer, TEST_CADENCE)?; + + let refused = refusal(publisher.publish(&mut signer, "hello")); + + assert!(refused.contains("odin_refused"), "{refused}"); + assert!(!refused.contains(SINK_CANARY), "{refused}"); + assert!(!refused.contains("127.0.0.1"), "{refused}"); + assert!(!refused.contains(&odin.addr.port().to_string()), "{refused}"); + Ok(()) +} + +#[test] +fn publish_times_out_without_echoing_endpoint() -> Result<()> { + let dead = dead_address(); + let rig = odin_rig(false, &format!("rudp://{dead}"))?; + let mut signer = rig.signer()?; + let cadence = Cadence { + connect: Duration::from_millis(300), + ..TEST_CADENCE + }; + let publisher = publisher_for(&signer, cadence)?; + + let started = Instant::now(); + let refused = refusal(publisher.publish(&mut signer, "hello")); + let elapsed = started.elapsed(); + + assert!(refused.contains("connect_timeout"), "{refused}"); + assert!(!refused.contains("127.0.0.1"), "{refused}"); + assert!(!refused.contains(&dead.port().to_string()), "{refused}"); + assert!(elapsed >= Duration::from_millis(300), "{elapsed:?}"); + assert!(elapsed < Duration::from_millis(1_500), "{elapsed:?}"); + Ok(()) +} + +#[test] +fn publisher_states_signer_health() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(false, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let publisher = publisher_for(&signer, TEST_CADENCE)?; + + publisher.publish(&mut signer, "first")?; + signer.report_health(RuntimePresenceState::Degraded); + let signed = publisher.publish(&mut signer, "second")?; + + let delivered = odin.deliveries(); + assert_eq!(delivered[0].record().state, "warming"); + assert_eq!(delivered[1].record().state, "degraded"); + assert_eq!(signed.record.state, "degraded"); + Ok(()) +} + +#[test] +fn two_publishes_strictly_increase_sequence_in_fresh_sessions() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(false, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let publisher = publisher_for(&signer, TEST_CADENCE)?; + + publisher.publish(&mut signer, "one")?; + // A route answer signs from the same signer, so it takes sequence 2. + let answer = signer.answer_route_observation(&CultNetMessage::SnapshotRequest { + message_id: "challenge".into(), + schema_ids: None, + record_keys: None, + })?; + assert!(matches!(answer, CultNetMessage::SnapshotResponseRaw { .. })); + publisher.publish(&mut signer, "two")?; + + let delivered = odin.deliveries(); + assert_eq!(delivered.len(), 2); + assert_eq!(delivered[0].record().publisher_sequence, 1); + assert_eq!(delivered[1].record().publisher_sequence, 3); + assert!(delivered[0].message_id.ends_with(":1"), "{:?}", delivered[0]); + assert!(delivered[1].message_id.ends_with(":3"), "{:?}", delivered[1]); + assert_ne!( + delivered[0].from, delivered[1].from, + "each publish is its own RUDP session" + ); + Ok(()) +} + +#[test] +fn publisher_refuses_foreign_signer() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(false, &odin.rudp_endpoint())?; + let other = odin_rig(false, &odin.rudp_endpoint())?; + let publisher = publisher_for(&rig.signer()?, TEST_CADENCE)?; + let mut foreign = other.signer()?; + + let refused = refusal(publisher.publish(&mut foreign, "hello")); + assert!(refused.contains("another runtime authority"), "{refused}"); + let refused = refusal(publisher.wait_for_write_lease(&mut foreign, Duration::from_secs(1))); + assert!(refused.contains("another runtime authority"), "{refused}"); + + assert_eq!(odin.count(), 0, "nothing was delivered"); + assert_eq!( + foreign.sign("untouched")?.record.publisher_sequence, + 1, + "a refused signer signed nothing" + ); + Ok(()) +} + +// ----------------------------------------------------------------- lease wait + +#[test] +fn lease_wait_not_required_returns_none_without_publishing() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(false, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let publisher = publisher_for(&signer, TEST_CADENCE)?; + + let guard = publisher.wait_for_write_lease(&mut signer, Duration::from_secs(5))?; + + assert!(guard.is_none()); + assert_eq!(odin.count(), 0); + assert_eq!(signer.sign("untouched")?.record.publisher_sequence, 1); + Ok(()) +} + +#[test] +fn catalog_connection_id_has_one_spelling() { + assert_eq!( + cultmesh_rs::CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, + crate::CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID + ); + assert_eq!( + crate::CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, + 0x0d1d_0002, + "Odin's catalog id is wire contract" + ); +} + +/// The lease holder needs the lock files of a unix host; Windows refuses a +/// lease requirement when it opens the authority. +#[cfg(unix)] +mod lease_wait { + use super::*; + + /// Writes Idunn's lease for `warming_sha256` once `count` heartbeats arrived, + /// then stamps when it finished. + fn grant_after( + rig: &Rig, + odin: &Odin, + count: usize, + which: usize, + written: &Mutex>, + ) { + let delivered = odin.wait_for(count); + let sha = delivered[which].record().canonical_sha256().unwrap(); + rig.write_lease(&sha, 1).unwrap(); + *written.lock().unwrap() = Some(Instant::now()); + } + + #[test] + fn lease_wait_heartbeats_and_returns_guard() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(true, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let cadence = Cadence { + heartbeat: Duration::from_millis(300), + ..TEST_CADENCE + }; + let publisher = publisher_for(&signer, cadence)?; + let written = Mutex::new(None); + + let started = Instant::now(); + let guard = std::thread::scope(|scope| { + scope.spawn(|| grant_after(&rig, &odin, 2, 1, &written)); + publisher.wait_for_write_lease(&mut signer, Duration::from_secs(10)) + })? + .expect("the wait returns the lease guard"); + let returned = Instant::now(); + + let delivered = odin.deliveries(); + assert_eq!(delivered.len(), 2, "no heartbeat after the lease"); + for heartbeat in &delivered { + let record = heartbeat.record(); + assert_eq!(record.state, "warming"); + assert_eq!(record.write_lease_sha256, None); + } + // The first heartbeat is sent on entry, the next one heartbeat later. + let gap = delivered[1].record().observed_at_unix_millis + - delivered[0].record().observed_at_unix_millis; + assert!((290..600).contains(&gap), "heartbeat gap {gap} ms"); + assert!(returned - started >= Duration::from_millis(290)); + // The lease is polled far faster than heartbeats are sent. + let written = written.lock().unwrap().expect("the lease was written"); + assert!( + returned - written < Duration::from_millis(150), + "the wait noticed the lease after {:?}", + returned - written + ); + // The guard is the one the signer now signs under. + signer.report_health(RuntimePresenceState::Active); + let active = publisher.publish(&mut signer, "ready")?; + assert_eq!( + active.record.write_lease_sha256.as_deref(), + Some(guard.lease_sha256()) + ); + Ok(()) + } + + #[test] + fn lease_wait_refuses_non_warming_signer() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(true, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let publisher = publisher_for(&signer, TEST_CADENCE)?; + signer.report_health(RuntimePresenceState::Active); + + let refused = refusal(publisher.wait_for_write_lease(&mut signer, Duration::from_secs(5))); + + assert!(refused.contains("Warming"), "{refused}"); + assert_eq!(odin.count(), 0, "refused before any publish"); + Ok(()) + } + + #[test] + fn lease_wait_survives_failed_heartbeat() -> Result<()> { + let addr = dead_address(); + let rig = odin_rig(true, &format!("rudp://{addr}"))?; + let mut signer = rig.signer()?; + let cadence = Cadence { + connect: Duration::from_millis(60), + ..TEST_CADENCE + }; + let publisher = publisher_for(&signer, cadence)?; + let written = Mutex::new(None); + + let started = Instant::now(); + let guard = std::thread::scope(|scope| { + // Odin comes up after the first heartbeats failed. + let late = scope.spawn(|| { + std::thread::sleep(Duration::from_millis(350)); + let odin = Odin::start_on(addr, Sink::Accept); + grant_after(&rig, &odin, 1, 0, &written); + odin + }); + let guard = publisher.wait_for_write_lease(&mut signer, Duration::from_secs(10)); + late.join().unwrap(); + guard + })?; + + assert!(guard.is_some()); + assert!(started.elapsed() >= Duration::from_millis(350)); + Ok(()) + } + + #[test] + fn lease_wait_times_out_with_counts() -> Result<()> { + let dead = dead_address(); + let rig = odin_rig(true, &format!("rudp://{dead}"))?; + let mut signer = rig.signer()?; + let cadence = Cadence { + heartbeat: Duration::from_millis(50), + connect: Duration::from_millis(30), + ..TEST_CADENCE + }; + let publisher = publisher_for(&signer, cadence)?; + + let started = Instant::now(); + let refused = refusal(publisher.wait_for_write_lease(&mut signer, Duration::from_millis(500))); + let elapsed = started.elapsed(); + + assert!(elapsed >= Duration::from_millis(500), "{elapsed:?}"); + assert!(elapsed < Duration::from_millis(2_000), "{elapsed:?}"); + let count: u32 = refused + .split(" failed deliveries") + .next() + .and_then(|head| head.rsplit(' ').next()) + .and_then(|count| count.parse().ok()) + .unwrap_or_else(|| panic!("no failure count in {refused}")); + assert!((3..=11).contains(&count), "{count} failed deliveries"); + assert!(refused.contains("connect_timeout"), "{refused}"); + assert!(!refused.contains("127.0.0.1"), "{refused}"); + assert!(!refused.contains(&dead.port().to_string()), "{refused}"); + assert!( + !refused.contains(rig.dir.path().to_str().unwrap()), + "{refused}" + ); + Ok(()) + } + + #[test] + fn lease_wait_timeout_counts_only_failed_deliveries() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(true, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let cadence = Cadence { + heartbeat: Duration::from_millis(50), + ..TEST_CADENCE + }; + let publisher = publisher_for(&signer, cadence)?; + + let refused = refusal(publisher.wait_for_write_lease(&mut signer, Duration::from_millis(400))); + + assert!(refused.contains(": 0 failed deliveries"), "{refused}"); + assert!(!refused.contains("last:"), "{refused}"); + assert!(odin.count() >= 3, "heartbeats kept arriving"); + Ok(()) + } + + #[test] + fn lease_wait_ends_at_once_on_a_foreign_incarnation_lease() -> Result<()> { + let odin = Odin::start(Sink::Accept); + let rig = odin_rig(true, &odin.rudp_endpoint())?; + let mut signer = rig.signer()?; + let publisher = publisher_for(&signer, TEST_CADENCE)?; + + let started = Instant::now(); + let refused = std::thread::scope(|scope| { + scope.spawn(|| { + let sha = odin.wait_for(1)[0].record().canonical_sha256().unwrap(); + // Staged elsewhere, so the wait never sees a lease that binds it. + let staged = rig.dir.path().join("staged-lease.cc"); + rig.write_lease_at(&staged, &sha, 1).unwrap(); + let stored = SingleFileMessagePackBackingStore::new(&staged) + .pull_all() + .unwrap() + .remove(0); + let mut lease = + IdunnProcessWriteLeaseRecord::decode_canonical(&stored.payload).unwrap(); + lease.runtime_instance_id = digest('1'); + write_record( + &rig.lease_path(), + TARGET, + IdunnProcessWriteLeaseRecord::TYPE, + IDUNN_PROCESS_WRITE_LEASE_SCHEMA, + lease.canonical_bytes().unwrap(), + ) + .unwrap(); + }); + refusal(publisher.wait_for_write_lease(&mut signer, Duration::from_secs(10))) + }); + + assert!(refused.contains("provider-owned Warming proof"), "{refused}"); + assert!(started.elapsed() < Duration::from_secs(3)); + Ok(()) + } +}