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

cultnet-rs: IdunnRuntimePresencePublisher, Odin RUDP delivery of signed presence and the Warming lease wait

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-29 18:27:12 +01:00
commit 4136167b41
5 changed files with 1088 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/cultnet-rs/Cargo.toml
View file

@ -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"

24
packages/cultnet-rs/src/idunn_runtime_provider.rs
View file

@ -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<u64> {
.context("Unix time exceeds u64")
}
mod odin_publisher;
pub use odin_publisher::IdunnRuntimePresencePublisher;
#[cfg(test)]
mod tests;

424
packages/cultnet-rs/src/idunn_runtime_provider/odin_publisher.rs Normal file
View file

@ -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<CultNetErrorCode>),
}
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<Option<Self>> {
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<SignedRuntimePresence> {
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<Option<ProcessWriteLeaseGuard>> {
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"
);
}
}

13
packages/cultnet-rs/src/idunn_runtime_provider/tests.rs
View file

@ -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> {
Self::with_dependencies(write_lease_required, Vec::new())
}
fn with_dependencies(
write_lease_required: bool,
dependencies: Vec<IdunnExpectedDependency>,
) -> Result<Self> {
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;

635
packages/cultnet-rs/src/idunn_runtime_provider/tests/odin_publisher.rs Normal file
View file

@ -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<u8>,
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<Mutex<Vec<Delivered>>>,
stop: Arc<AtomicBool>,
thread: Option<JoinHandle<()>>,
}
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<Vec<_>> { 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<Delivered> {
self.received.lock().unwrap().clone()
}
fn count(&self) -> usize {
self.received.lock().unwrap().len()
}
fn wait_for(&self, count: usize) -> Vec<Delivered> {
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<String>) -> 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<IdunnExpectedDependency>,
) -> Result<Rig> {
Rig::with_dependencies(write_lease_required, dependencies)
}
fn odin_rig(write_lease_required: bool, endpoint: &str) -> Result<Rig> {
rig_with(
write_lease_required,
vec![odin_dependency("odin.verse.v1", Some(endpoint.into()))],
)
}
fn publisher_for(
signer: &IdunnRuntimePresenceSigner,
cadence: Cadence,
) -> Result<IdunnRuntimePresencePublisher> {
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<Option<Instant>>,
) {
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(())
}
}