diff --git a/crates/odin-daemon/src/lib.rs b/crates/odin-daemon/src/lib.rs index dc8ec24..dfe5112 100644 --- a/crates/odin-daemon/src/lib.rs +++ b/crates/odin-daemon/src/lib.rs @@ -195,23 +195,10 @@ impl CultCacheIdunnProjectionSource { impl IdunnProjectionSource for CultCacheIdunnProjectionSource { fn projections(&self, target: &str) -> Result> { - if !self.path.is_file() { + let Some(entries) = self.entries()? else { return Ok(Vec::new()); - } - let entries = - SingleFileMessagePackBackingStore::new(&self.path).pull_all_read_only_snapshot()?; - let anchor_envelope = unique_envelope( - &entries, - GameCultServiceTrustAnchorRecord::TYPE, - &runtime_presence_anchor_id(target), - )?; - let provider_anchor: Option = anchor_envelope - .map(|envelope| { - ensure_schema(envelope, GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA)?; - decode_canonical(&envelope.payload, "provider trust anchor") - }) - .transpose()?; - + }; + let anchor = provider_anchor(&entries, target)?; // A record keyed by anything but its own incarnation key is not this // contract's. That includes the target-keyed single slot the previous // projection used; an Idunn still writing that shape projects nothing @@ -227,38 +214,99 @@ impl IdunnProjectionSource for CultCacheIdunnProjectionSource { if incarnation.target != target { continue; } - ensure_schema(envelope, IDUNN_EXPECTED_INCARNATION_SCHEMA)?; - let expected = IdunnExpectedIncarnationRecord::decode_canonical(&envelope.payload)?; - ensure!( - IncarnationRef::of(&expected)? == incarnation, - "Expected projection key is substituted" - ); - let key = incarnation.key(); - let activation = unique_envelope(&entries, IdunnRuntimeActivationRecord::TYPE, &key)? - .map(|envelope| { - ensure_schema(envelope, IDUNN_RUNTIME_ACTIVATION_SCHEMA)?; - IdunnRuntimeActivationRecord::decode_canonical(&envelope.payload) - }) - .transpose()?; - let current_lease = - unique_envelope(&entries, IdunnProcessWriteLeaseRecord::TYPE, &key)? - .map(|envelope| { - ensure_schema(envelope, IDUNN_PROCESS_WRITE_LEASE_SCHEMA)?; - IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload) - }) - .transpose()?; - let projection = IdunnRuntimeProjection { - expected, - provider_anchor: provider_anchor.clone(), - activation, - current_lease, - }; - projection.validate()?; - projections.push(projection); + projections.push(assemble_projection( + &entries, + anchor.clone(), + envelope, + &incarnation, + )?); } projections.sort_by_key(|projection| projection.expected.incarnation_id.clone()); Ok(projections) } + + /// Only the named incarnation's own records are decoded: an undecodable + /// record under another incarnation's key is that incarnation's fault, and + /// never makes this one unreadable. + fn projection(&self, incarnation: &IncarnationRef) -> Result> { + let Some(entries) = self.entries()? else { + return Ok(None); + }; + let Some(envelope) = unique_envelope( + &entries, + IdunnExpectedIncarnationRecord::TYPE, + &incarnation.key(), + )? + else { + return Ok(None); + }; + let anchor = provider_anchor(&entries, &incarnation.target)?; + assemble_projection(&entries, anchor, envelope, incarnation).map(Some) + } +} + +impl CultCacheIdunnProjectionSource { + fn entries(&self) -> Result>> { + if !self.path.is_file() { + return Ok(None); + } + Ok(Some( + SingleFileMessagePackBackingStore::new(&self.path).pull_all_read_only_snapshot()?, + )) + } +} + +fn provider_anchor( + entries: &[CultCacheEnvelope], + target: &str, +) -> Result> { + unique_envelope( + entries, + GameCultServiceTrustAnchorRecord::TYPE, + &runtime_presence_anchor_id(target), + )? + .map(|envelope| { + ensure_schema(envelope, GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA)?; + decode_canonical(&envelope.payload, "provider trust anchor") + }) + .transpose() +} + +fn assemble_projection( + entries: &[CultCacheEnvelope], + provider_anchor: Option, + expected_envelope: &CultCacheEnvelope, + incarnation: &IncarnationRef, +) -> Result { + ensure_schema(expected_envelope, IDUNN_EXPECTED_INCARNATION_SCHEMA)?; + let expected = IdunnExpectedIncarnationRecord::decode_canonical(&expected_envelope.payload)?; + ensure!( + IncarnationRef::of(&expected)? == *incarnation, + "Expected projection key is substituted" + ); + let key = incarnation.key(); + let activation = unique_envelope(entries, IdunnRuntimeActivationRecord::TYPE, &key)? + .map(|envelope| { + ensure_schema(envelope, IDUNN_RUNTIME_ACTIVATION_SCHEMA)?; + IdunnRuntimeActivationRecord::decode_canonical(&envelope.payload) + }) + .transpose()?; + let current_lease = unique_envelope(entries, IdunnProcessWriteLeaseRecord::TYPE, &key)? + .map(|envelope| { + ensure_schema(envelope, IDUNN_PROCESS_WRITE_LEASE_SCHEMA)?; + IdunnProcessWriteLeaseRecord::decode_canonical(&envelope.payload) + }) + .transpose()?; + let projection = IdunnRuntimeProjection { + expected, + provider_anchor, + activation, + current_lease, + }; + projection + .validate() + .context(PresenceAuthorityRefused("Idunn's projection is invalid"))?; + Ok(projection) } pub trait OdinCorrelationSigner { @@ -599,6 +647,21 @@ fn decode_publisher_watermark(target: &str, payload: &[u8]) -> Result { Ok(sequence) } +/// Odin's own runtime presence cannot be admitted under the authority Idunn +/// projects: the authority cannot be established, or the claim, or the stored +/// presence of the same activation, does not authenticate under it. Marks the +/// failure for the serving loop; see `survive` in the daemon. +#[derive(Debug)] +pub struct PresenceAuthorityRefused(pub &'static str); + +impl std::fmt::Display for PresenceAuthorityRefused { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(self.0) + } +} + +impl std::error::Error for PresenceAuthorityRefused {} + pub struct OdinTopologyAuthority { projections: P, store: S, @@ -689,7 +752,10 @@ where projection.validate()?; let now = self.clock.now_unix_millis()?; let (authority, projection_disagreements) = - classify_runtime_authority(&projection, &self.idunn_anchor, now)?; + classify_runtime_authority(&projection, &self.idunn_anchor, now) + .context(PresenceAuthorityRefused( + "the projected runtime authority cannot be verified", + ))?; let (lease_sha256, lease_disagreement) = classify_current_lease(&projection)?; let query = OdinStoreQuery { incarnation: IncarnationRef::of(&projection.expected)?, @@ -884,10 +950,12 @@ where authority: Option<&cultnet_rs::VerifiedRuntimeAuthority>, ) -> Result { let Some(authority) = authority else { - ensure!( - incoming.is_none(), - "runtime presence has no exact current activation and provider anchor" - ); + if incoming.is_some() { + return Err(PresenceAuthorityRefused( + "runtime presence has no exact current activation and provider anchor", + ) + .into()); + } return Ok(observed .presence .as_ref() @@ -915,7 +983,10 @@ where bytes, authority, self.presence_context(received_at), - )?; + ) + .context(PresenceAuthorityRefused( + "runtime presence does not authenticate under the projected authority", + ))?; // Publisher sequence orders the claims of one publisher: the // activation whose key signs them. Each activation counts from its // own start, so the stored presence constrains the claim only when @@ -926,6 +997,16 @@ where // that names this activation is this publisher's, and it must // authenticate at the time Odin received it; failing to is an // error, never a reason to forget its sequence. + // + // The two ways a stored presence can fail to be the same + // publisher's are decided differently on purpose. One that cannot + // be decoded names no activation, so it cannot be this + // publisher's: it is treated as another publisher's and the claim + // supersedes it (fail-open; the store's own read refuses such a + // record earlier still). One that decodes and names this + // activation is this publisher's own count: if it no longer + // authenticates, the count is unknowable and the claim is + // refused (fail-closed). let same_publisher = match observed.presence.as_ref() { Some(stored) if decode_presence(&stored.canonical_bytes).is_ok_and(|record| { @@ -938,9 +1019,9 @@ where authority, self.presence_context(stored.trusted_received_at_unix_millis), ) - .context( + .context(PresenceAuthorityRefused( "stored presence of the current activation no longer authenticates", - )?, + ))?, ) } _ => None, @@ -1729,6 +1810,34 @@ mod tests { Ok(()) } + /// An Expected that will not decode is its own incarnation's fault: another + /// incarnation, even of the same target, is still read exactly. + #[test] + fn an_undecodable_expected_does_not_hide_its_sibling_incarnations() -> Result<()> { + let world = TestWorld::new()?; + let service = world.service("ghostlight", Vec::new(), false)?; + let store = SingleFileMessagePackBackingStore::new(&world.projection_path); + let current = store.pull_all_read_only_snapshot()?; + let mut with_ghost = current.clone(); + with_ghost.push(CultCacheEnvelope { + key: IncarnationRef::new("ghostlight", digest('a')).key(), + r#type: IdunnExpectedIncarnationRecord::TYPE.into(), + payload: vec![0xc1, 0x00], + stored_at: rfc3339_millis(NOW - 1_000)?, + schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), + }); + ensure!(store.compare_exchange_snapshot(¤t, &with_ghost)?); + let source = CultCacheIdunnProjectionSource::new(&world.projection_path); + assert!(source.projection(&service.incarnation()?)?.is_some()); + assert!( + source + .projection(&IncarnationRef::new("ghostlight", digest('a'))) + .is_err(), + "the ghost itself is refused" + ); + Ok(()) + } + /// A store the target-keyed Odin wrote holds correlations under bare /// target keys. They are retired; presence history under any key stays. #[test] @@ -2334,9 +2443,11 @@ mod tests { fn a_restarted_activation_counts_its_publisher_sequence_from_one() -> Result<()> { let world = TestWorld::new()?; let mut service = world.service("ghostlight", Vec::new(), false)?; - world - .engine(NOW) - .admit_presence("ghostlight", &service.signed_presence(5, |_| {})?, NOW)?; + world.engine(NOW).admit_presence( + "ghostlight", + &service.signed_presence(5, |_| {})?, + NOW, + )?; world.restart_activation(&mut service, RESTART)?; let restarted = service.signed_presence(1, |record| { @@ -2385,9 +2496,11 @@ mod tests { fn a_stale_stored_presence_never_blocks_a_current_claim() -> Result<()> { let world = TestWorld::new()?; let mut service = world.service("ghostlight", Vec::new(), false)?; - world - .engine(NOW) - .admit_presence("ghostlight", &service.signed_presence(3, |_| {})?, NOW)?; + world.engine(NOW).admit_presence( + "ghostlight", + &service.signed_presence(3, |_| {})?, + NOW, + )?; world.restart_activation(&mut service, RESTART)?; let before = decode_signed( @@ -2445,6 +2558,81 @@ mod tests { Ok(()) } + fn snapshot_with_stored_presence( + service: &TestService, + stored: AdmittedPresence, + ) -> Result { + Ok(OdinStoreSnapshot { + query: OdinStoreQuery { + incarnation: service.incarnation()?, + provider_signer_identity_id: service + .projection + .expected + .expected_signer_identity_id + .clone(), + dependencies: Vec::new(), + }, + presence: Some(stored), + correlation: None, + dependency_correlations: BTreeMap::new(), + }) + } + + /// A stored presence that names the current activation but no longer + /// authenticates at its own received time is an error, not "another + /// publisher": the claim must not slip past that publisher's sequence. + #[test] + fn a_same_activation_stored_presence_that_no_longer_authenticates_is_an_error() -> Result<()> { + let world = TestWorld::new()?; + let service = world.service("ghostlight", Vec::new(), false)?; + let later = NOW + 3_600_000; + let engine = world.engine(later); + let (authority, _) = + classify_runtime_authority(&service.projection, &world.idunn_anchor, later)?; + // Observed at NOW but received an hour later: outside the skew window + // at its own received time. + let stored = AdmittedPresence::new(service.signed_presence(5, |_| {})?, later)?; + let snapshot = snapshot_with_stored_presence(&service, stored)?; + let claim = service.signed_presence(6, |record| { + record.observed_at_unix_millis = later - 10; + })?; + + let error = engine + .select_presence(&snapshot, Some((&claim, later)), authority.as_ref()) + .err() + .context("a same-activation presence that fails authentication was ignored")?; + assert!( + format!("{error:#}").contains("no longer authenticates"), + "refused for the wrong reason: {error:#}" + ); + Ok(()) + } + + /// A stored presence that cannot be decoded names no activation, so it is + /// another publisher's and a fresh claim supersedes it. + #[test] + fn an_undecodable_stored_presence_is_another_publisher_and_is_superseded() -> Result<()> { + let world = TestWorld::new()?; + let service = world.service("ghostlight", Vec::new(), false)?; + let engine = world.engine(NOW); + let (authority, _) = + classify_runtime_authority(&service.projection, &world.idunn_anchor, NOW)?; + let stored = AdmittedPresence::new(vec![0xc1], NOW)?; + let snapshot = snapshot_with_stored_presence(&service, stored)?; + let claim = service.signed_presence(1, |_| {})?; + + let selected = + engine.select_presence(&snapshot, Some((&claim, NOW)), authority.as_ref())?; + assert!(matches!( + selected, + SelectedPresence::Current { + replacement: Some(_), + .. + } + )); + Ok(()) + } + #[test] fn a_refresh_that_observes_nothing_new_does_not_burn_a_publisher_sequence() -> Result<()> { let world = TestWorld::new()?; diff --git a/crates/odin-daemon/src/main.rs b/crates/odin-daemon/src/main.rs index be6ebec..a05d06c 100644 --- a/crates/odin-daemon/src/main.rs +++ b/crates/odin-daemon/src/main.rs @@ -10,34 +10,33 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; -use anyhow::{Context, Result, anyhow, bail, ensure}; +use anyhow::{Context, Result, bail, ensure}; use chrono::{DateTime, Utc}; use cultcache_rs::{ CultCacheEnvelope, CultCacheExpectedEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, }; use cultmesh_rs::{ - CultMeshRudpDocumentPublishOptions, CultMeshRudpDocumentServer, - CultMeshRudpDocumentServerOptions, CultMeshRudpPollOutcome, CultMeshRudpRawDocumentReceipt, - CultMeshRudpRawDocumentSink, CultMeshRudpSnapshotQuery, CultMeshRudpSnapshotSource, - CultMeshSystemClock, publish_cultnet_message_to_rudp_catalog, + CultMeshRudpDocumentServer, CultMeshRudpDocumentServerOptions, CultMeshRudpPollOutcome, + CultMeshRudpRawDocumentReceipt, CultMeshRudpRawDocumentSink, CultMeshRudpSnapshotQuery, + CultMeshRudpSnapshotSource, CultMeshSystemClock, }; use cultnet_rs::{ - CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, - GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, GameCultProviderHealthIdentity, - GameCultRuntimeCapability, GameCultRuntimePresenceHealthPurpose, - GameCultRuntimePresenceHealthRecord, IDUNN_EXPECTED_INCARNATION_SCHEMA, - IDUNN_PROCESS_WRITE_LEASE_SCHEMA, IDUNN_RUNTIME_ACTIVATION_CREDENTIAL_NAME, - IDUNN_RUNTIME_ACTIVATION_SCHEMA, IdunnExpectedIncarnationRecord, IdunnProcessWriteLeaseRecord, - IdunnRuntimeActivationRecord, IdunnRuntimeActivationSigner, IdunnServiceIdentity, - ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA, OdinRuntimeTopologyCorrelationRecord, - OdinTopologyIdentity, ServiceIdentityProfile, ServiceIdentitySigner, - ServiceIdentityTrustAnchor, derive_service_identity_id, + CultNetRawDocumentRecord, CultNetRawPayloadEncoding, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, + GameCultProviderHealthIdentity, GameCultRuntimeCapability, + GameCultRuntimePresenceHealthPurpose, GameCultRuntimePresenceHealthRecord, + IDUNN_EXPECTED_INCARNATION_SCHEMA, IDUNN_PROCESS_WRITE_LEASE_SCHEMA, + IDUNN_RUNTIME_ACTIVATION_CREDENTIAL_NAME, IDUNN_RUNTIME_ACTIVATION_SCHEMA, + IdunnExpectedIncarnationRecord, IdunnProcessWriteLeaseRecord, IdunnRuntimeActivationRecord, + IdunnRuntimeActivationSigner, IdunnServiceIdentity, + OdinRuntimeTopologyCorrelationRecord, OdinTopologyIdentity, ServiceIdentityProfile, + ServiceIdentitySigner, ServiceIdentityTrustAnchor, derive_service_identity_id, open_service_identity_credential_reader, verify_runtime_authority, }; use fs2::FileExt; use odin_daemon::{ AuthenticationPolicy, CultCacheIdunnProjectionSource, CultCacheOdinTopologyStore, - IdunnProjectionSource, IncarnationRef, OdinTopologyAuthority, SystemClock, + IdunnProjectionSource, IncarnationRef, OdinTopologyAuthority, PresenceAuthorityRefused, + SystemClock, }; const TARGET: &str = "odin"; @@ -65,7 +64,10 @@ const SYSTEMD_LISTEN_FDS_START: RawFd = 3; // candidate that never hears back is cleaned up by Idunn's own abort; this // only bounds how long it sits idle first. const BOOTSTRAP_LEASE_TIMEOUT: Duration = Duration::from_secs(300); -const SELF_PUBLISH_TIMEOUT: Duration = Duration::from_secs(6); +// How long the UDP socket may fail every poll before that is a dead socket +// rather than a hiccup. Odin then ends and Idunn restarts it. +const POLL_FAILURE_LIMIT: Duration = Duration::from_secs(30); +const POLL_FAILURE_BACKOFF: Duration = Duration::from_millis(100); const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5); const PROJECTION_REFRESH_INTERVAL: Duration = Duration::from_millis(250); const IDLE_POLL_INTERVAL: Duration = Duration::from_millis(2); @@ -269,24 +271,31 @@ impl RuntimeState { ); validate_raw_document_shape(&receipt.document)?; if receipt.document.schema_id == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA { - let presence = decode_presence(&receipt.document.payload)?; - ensure!( - receipt.document.record_key == presence.target, - "runtime-presence document key differs from its signed target" - ); - self.topology - .as_ref() - .context("Odin topology authority is absent")? - .admit_presence( - &presence.target, - &receipt.document.payload, - receipt.received_at_unix_millis, - )?; - return Ok(()); + return self + .admit_presence_document(&receipt.document, receipt.received_at_unix_millis); } persist_generic_document(&self.options.store, &receipt.document) } + /// The one admission path for a runtime-presence document, whether a + /// provider delivered it over RUDP or Odin signed it for itself. + fn admit_presence_document( + &self, + document: &CultNetRawDocumentRecord, + received_at_unix_millis: u64, + ) -> Result<()> { + let presence = decode_presence(&document.payload)?; + ensure!( + document.record_key == presence.target, + "runtime-presence document key differs from its signed target" + ); + self.topology + .as_ref() + .context("Odin topology authority is absent")? + .admit_presence(&presence.target, &document.payload, received_at_unix_millis)?; + Ok(()) + } + fn raw_snapshot( &mut self, query: &CultMeshRudpSnapshotQuery, @@ -305,7 +314,7 @@ impl RuntimeState { self.activated(), "Odin catalog is unavailable until Idunn grants its process-write lease" ); - self.refresh_all_correlations()?; + self.require_current_write_lease()?; self.stored_snapshot(query) } @@ -403,18 +412,15 @@ impl RuntimeState { }) } - fn active_document_put(&mut self, detail: &str) -> Result { + /// Odin is the store, so its own heartbeat is admitted into the store + /// directly. It never crosses RUDP to itself, and so never takes one of the + /// server's bounded session slots from the providers it serves: a table + /// full of lingering publisher sessions cannot make Odin miss its own + /// liveness. + fn publish_self_presence(&mut self, detail: &str) -> Result<()> { self.require_current_write_lease()?; let document = self.signed_presence_document("active", detail)?; - let sequence = self.publisher_sequence; - Ok(CultNetMessage::DocumentPutRaw { - message_id: format!( - "odin-presence:{}:{}:{sequence}", - self.authority_material.expected.runtime_id, - self.authority_material.activation.runtime_instance_id - ), - document, - }) + self.admit_presence_document(&document, unix_millis()?) } fn require_current_write_lease(&self) -> Result<()> { @@ -422,12 +428,17 @@ impl RuntimeState { .write_lease .as_ref() .context("Odin has no process-write lease")?; - let current = read_process_write_lease(&self.write_lease_path)? - .context("Odin process-write lease was withdrawn")?; - ensure!( - current == held.record && current.canonical_sha256()? == held.sha256, - "Odin process-write lease changed after admission" - ); + // A lease Odin cannot read is a lease Odin cannot show it still holds. + let current = read_process_write_lease(&self.write_lease_path) + .map_err(|error| { + WriteLeaseLost(format!("Odin process-write lease is unreadable: {error:#}")) + })? + .ok_or_else(|| WriteLeaseLost("Odin process-write lease was withdrawn".into()))?; + if current != held.record || current.canonical_sha256()? != held.sha256 { + return Err( + WriteLeaseLost("Odin process-write lease changed after admission".into()).into(), + ); + } Ok(()) } @@ -437,10 +448,23 @@ impl RuntimeState { .topology .as_ref() .context("Odin topology authority is absent")?; - let mut incarnations = projection_incarnations(&self.options.idunn_projection)?; - incarnations.extend(correlation_incarnations(&self.options.store)?); + let mut incarnations = incarnation_keys( + &self.options.idunn_projection, + IdunnExpectedIncarnationRecord::TYPE, + )?; + incarnations.extend(incarnation_keys( + &self.options.store, + OdinRuntimeTopologyCorrelationRecord::TYPE, + )?); + // Each incarnation is refreshed on its own: one whose records cannot be + // read is that incarnation's failure, and never stops the others. for incarnation in incarnations { - topology.refresh(&incarnation)?; + if let Err(error) = topology.refresh(&incarnation) { + eprintln!( + "Odin could not refresh incarnation {}; the rest are unaffected: {error:#}", + incarnation.key() + ); + } } Ok(()) } @@ -570,12 +594,13 @@ fn main() -> Result<()> { } let bootstrap_started = Instant::now(); + let mut poll_failing_since = None; while !state.borrow_mut().try_activate()? { if stopping.load(Ordering::Relaxed) { eprintln!("Odin stopping before activation on request"); return Ok(()); } - let progressed = poll_server(&mut server)?; + let progressed = poll_server(&mut server, &mut poll_failing_since)?; ensure!( bootstrap_started.elapsed() < BOOTSTRAP_LEASE_TIMEOUT, "timed out waiting for Idunn to admit Odin's exact Warming proof and grant its lease" @@ -589,87 +614,143 @@ fn main() -> Result<()> { // this contract's and would otherwise be served to the Verse as current. // Presence history is kept: the self publisher sequence continues from it. CultCacheOdinTopologyStore::new(&state.borrow().options.store).retire_legacy_correlations()?; - publish_self_presence(&state, &mut server, "ready")?; - state.borrow_mut().refresh_all_correlations()?; - let mut last_heartbeat = Instant::now(); - let mut last_projection_refresh = Instant::now(); + // Once serving, only losing the write lease ends Odin (see `survive`). + let mut timers = ServingTimers::default(); loop { if stopping.load(Ordering::Relaxed) { eprintln!("Odin stopping on request"); return Ok(()); } - let progressed = poll_server(&mut server)?; - if last_projection_refresh.elapsed() >= PROJECTION_REFRESH_INTERVAL { - state.borrow_mut().refresh_all_correlations()?; - last_projection_refresh = Instant::now(); - } - if last_heartbeat.elapsed() >= HEARTBEAT_INTERVAL { - publish_self_presence(&state, &mut server, "ready")?; - last_heartbeat = Instant::now(); - } - if !progressed { + if !serving_pass(&state, &mut server, &mut timers)? { thread::sleep(IDLE_POLL_INTERVAL); } } } -fn publish_self_presence( +/// A timer that has never fired is due, so the first pass refreshes and +/// publishes. +#[derive(Default)] +struct ServingTimers { + last_heartbeat: Option, + last_projection_refresh: Option, + poll_failing_since: Option, +} + +type OdinServer = CultMeshRudpDocumentServer; + +/// One turn of the serving loop: serve a datagram, refresh the topology, and +/// publish Odin's own presence when each is due. Returns whether the poll made +/// progress. It fails only for `WriteLeaseLost` or a dead socket. +fn serving_pass( state: &Rc>, - server: &mut CultMeshRudpDocumentServer, - detail: &str, -) -> Result<()> { - let (message, options) = { - let mut state = state.borrow_mut(); - let message = state.active_document_put(detail)?; - let mut options = CultMeshRudpDocumentPublishOptions::odin( - state.candidate, - state.authority_material.expected.runtime_id.clone(), - ); - options.connect_timeout = Duration::from_secs(2); - options.flush_timeout = Duration::from_secs(2); - options.source_agent_id = Some( - state - .authority_material - .provider_signer - .entry() - .identity_id - .clone(), - ); - options.source_role = Some("runtime-presence-health-publisher".into()); - options.tags = vec!["cultnet.transport.rudp.v0".into()]; - (message, options) - }; - let publisher = - thread::spawn(move || publish_cultnet_message_to_rudp_catalog(&message, options)); - let started = Instant::now(); - while !publisher.is_finished() { - let progressed = poll_server(server)?; - ensure!( - started.elapsed() < SELF_PUBLISH_TIMEOUT, - "Odin self-presence publication did not complete" - ); - if !progressed { - thread::sleep(IDLE_POLL_INTERVAL); - } + server: &mut OdinServer, + timers: &mut ServingTimers, +) -> Result { + let progressed = poll_server(server, &mut timers.poll_failing_since)?; + if is_due(timers.last_projection_refresh, PROJECTION_REFRESH_INTERVAL) { + survive( + "projection refresh", + state.borrow_mut().refresh_all_correlations(), + )?; + timers.last_projection_refresh = Some(Instant::now()); } - publisher - .join() - .map_err(|_| anyhow!("Odin self-presence publisher panicked"))??; - Ok(()) + if is_due(timers.last_heartbeat, HEARTBEAT_INTERVAL) { + survive( + "self-presence publication", + state.borrow_mut().publish_self_presence("ready"), + )?; + timers.last_heartbeat = Some(Instant::now()); + } + Ok(progressed) } -fn poll_server( - server: &mut CultMeshRudpDocumentServer, +fn is_due(last: Option, interval: Duration) -> bool { + last.is_none_or(|last| last.elapsed() >= interval) +} + +/// Odin held its process-write lease and no longer does: another incarnation +/// has been granted the state, or Idunn withdrew the grant. Serving on would be +/// a second writer, so this is the one serving-loop condition that ends Odin. +#[derive(Debug)] +struct WriteLeaseLost(String); + +impl std::fmt::Display for WriteLeaseLost { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(&self.0) + } +} + +impl std::error::Error for WriteLeaseLost {} + +/// The serving loop's failure policy. Odin's liveness is Odin's own: a failed +/// refresh, publication or read is logged and retried on the next pass, never +/// allowed to end the daemon. Only `WriteLeaseLost`, wherever in the error chain +/// it sits, is returned -- and, as a named temporary rule, so is +/// `PresenceAuthorityRefused`: a self-presence that Odin's own authority will +/// never admit (no verifiable authority, a signer that does not match the +/// anchor, a stored presence of this activation that no longer authenticates) +/// leaves a frozen presence that goes stale, and a stale Odin presence makes +/// `dependency_evidence` flip every dependent to not-Ready. Ending Odin lets +/// Idunn replace it. Deleted when Idunn proves Odin by its own route challenge +/// (Idunn audit cut A3, operator question Q-O5). +fn survive(what: &str, result: Result) -> Result> { + match result { + Ok(value) => Ok(Some(value)), + Err(error) + if error.downcast_ref::().is_some() + || error.downcast_ref::().is_some() => + { + Err(error) + } + Err(error) => { + eprintln!("Odin {what} failed; retrying on the next pass: {error:#}"); + Ok(None) + } + } +} + +fn poll_server(server: &mut OdinServer, failing_since: &mut Option) -> Result { + settle_poll(server.poll_once(), failing_since, Instant::now()) +} + +/// A failed poll is logged and retried. The socket failing every poll for +/// `POLL_FAILURE_LIMIT` is a dead socket, not a transient one; only then does +/// Odin end, so Idunn restarts it. Any successful poll clears the failure run. +fn settle_poll( + result: Result, + failing_since: &mut Option, + now: Instant, ) -> Result { - match server.poll_once()? { - CultMeshRudpPollOutcome::Idle => Ok(false), - CultMeshRudpPollOutcome::Handled => Ok(true), + match result { + Ok(outcome) => { + *failing_since = None; + Ok(progress_of(outcome)) + } + Err(error) => { + let since = *failing_since.get_or_insert(now); + ensure!( + now.duration_since(since) < POLL_FAILURE_LIMIT, + "Odin's RUDP socket has failed every poll for {POLL_FAILURE_LIMIT:?}: {error:#}" + ); + eprintln!("Odin RUDP poll failed; retrying: {error:#}"); + thread::sleep(POLL_FAILURE_BACKOFF); + Ok(false) + } + } +} + +/// An application rejection ends only the offending peer's session, which the +/// server has already done. It is logged and never ends Odin. +fn progress_of(outcome: CultMeshRudpPollOutcome) -> bool { + match outcome { + CultMeshRudpPollOutcome::Idle => false, + CultMeshRudpPollOutcome::Handled => true, CultMeshRudpPollOutcome::ApplicationRejected(rejection) => { eprintln!( "Odin rejected CultMesh RUDP application message {:?} {} from {:?}: {}", rejection.operation, rejection.message_id, rejection.session, rejection.reason ); - Ok(true) + true } } } @@ -1091,66 +1172,26 @@ fn decode_presence(payload: &[u8]) -> Result Result> { - let entries = SingleFileMessagePackBackingStore::new(path).pull_all_read_only_snapshot()?; - entries - .into_iter() - .filter(|entry| entry.r#type == IdunnExpectedIncarnationRecord::TYPE) - .filter_map(|entry| { - let incarnation = IncarnationRef::parse_key(&entry.key)?; - Some((|| -> Result { - ensure!( - entry.schema_id.as_deref() == Some(IDUNN_EXPECTED_INCARNATION_SCHEMA), - "Idunn projection contains an Expected record under the wrong schema" - ); - let expected = IdunnExpectedIncarnationRecord::decode_canonical(&entry.payload)?; - ensure!( - IncarnationRef::of(&expected)? == incarnation, - "Idunn Expected key is substituted" - ); - Ok(incarnation) - })()) - }) - .collect() -} - -/// Every incarnation Odin holds a correlation for, projected by Idunn or not; -/// the latter are refreshed so their correlations are withdrawn. -fn correlation_incarnations(path: &Path) -> Result> { +/// Records keyed by anything but an incarnation key are not this contract's and +/// are skipped, not refused: a projection written by an older Idunn projects +/// nothing this daemon acts on, and a correlation written by the previous, +/// target-keyed Odin is retired at activation (`retire_legacy_correlations`). +fn incarnation_keys(path: &Path, record_type: &str) -> Result> { if !path.is_file() { return Ok(BTreeSet::new()); } - SingleFileMessagePackBackingStore::new(path) + Ok(SingleFileMessagePackBackingStore::new(path) .pull_all_read_only_snapshot()? .into_iter() - .filter(|entry| entry.r#type == OdinRuntimeTopologyCorrelationRecord::TYPE) - // A correlation keyed by anything but an incarnation key was written - // by the previous, target-keyed Odin. It is retired at activation - // (`retire_legacy_correlations`), never refreshed, and never a reason - // to refuse to start. - .filter(|entry| IncarnationRef::parse_key(&entry.key).is_some()) - .map(|entry| { - ensure!( - entry.schema_id.as_deref() == Some(ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA), - "Odin correlation is stored under the wrong schema" - ); - let (record, _) = - OdinRuntimeTopologyCorrelationRecord::decode_canonical_signed_payload( - &entry.payload, - )?; - let incarnation = IncarnationRef::new(record.target, record.expected_projection_sha256); - ensure!( - entry.key == incarnation.key(), - "Odin correlation key is substituted" - ); - Ok(incarnation) - }) - .collect() + .filter(|entry| entry.r#type == record_type) + .filter_map(|entry| IncarnationRef::parse_key(&entry.key)) + .collect()) } fn validate_snapshot_filters(query: &CultMeshRudpSnapshotQuery) -> Result<()> { @@ -1229,6 +1270,15 @@ fn rfc3339_millis(value: u64) -> Result { #[cfg(test)] mod tests { + use cultnet_rs::{ + CultNetMessage, CultNetRudpReliableSendStatus, CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, CultNetWireContract, + GAMECULT_RUNTIME_PRESENCE_HEALTH_SIGNING_PURPOSE, GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA, + GameCultServiceTrustAnchorRecord, IdunnExpectedCapability, IdunnExpectedRoute, + IdunnRuntimeActivationLaunch, ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA, + encode_cultnet_message_to_vec, enroll_service_identity_at, + }; + use super::*; #[test] @@ -1337,4 +1387,729 @@ mod tests { assert_eq!(proofs.back().unwrap().0, "warming-64"); assert!(!proofs.iter().any(|(proof, _)| proof == "warming-one")); } + + // ---- serving-loop liveness ------------------------------------------- + + fn digest(byte: char) -> String { + format!("sha256-{}", byte.to_string().repeat(64)) + } + + /// Odin as Idunn admits it: a real activation, provider anchor, write + /// lease and projection on disk, `try_activate` run for real, and the + /// production RUDP server bound to the candidate socket the Expected names. + struct OdinWorld { + _temp: tempfile::TempDir, + state: Rc>, + server: OdinServer, + timers: ServingTimers, + store: PathBuf, + lease_path: PathBuf, + provider_identity_id: String, + lease: IdunnProcessWriteLeaseRecord, + } + + fn write_single_record(path: &Path, envelope: CultCacheEnvelope) -> Result<()> { + let store = SingleFileMessagePackBackingStore::new(path); + let current = if path.is_file() { + store.pull_all_read_only_snapshot()? + } else { + Vec::new() + }; + ensure!( + store.compare_exchange_snapshot(¤t, &[envelope])?, + "test store CAS failed" + ); + Ok(()) + } + + fn lease_envelope(lease: &IdunnProcessWriteLeaseRecord) -> Result { + Ok(CultCacheEnvelope { + key: lease.target.clone(), + r#type: IdunnProcessWriteLeaseRecord::TYPE.into(), + payload: lease.canonical_bytes()?, + stored_at: rfc3339_millis(lease.issued_at_unix_millis)?, + schema_id: Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into()), + }) + } + + fn activated_odin() -> Result { + let temp = tempfile::tempdir()?; + let root = temp.path(); + let socket = UdpSocket::bind("127.0.0.1:0")?; + let candidate = socket.local_addr()?; + let now = unix_millis()?; + let idunn_signer = + enroll_service_identity_at::(&root.join("idunn.cc"))?; + let idunn_anchor = idunn_signer.trust_anchor()?; + let odin_signer = + enroll_service_identity_at::(&root.join("odin-topology.cc"))?; + let provider_signer = enroll_service_identity_at::( + &root.join("odin-provider.cc"), + )?; + let expected = IdunnExpectedIncarnationRecord { + schema_version: IDUNN_EXPECTED_INCARNATION_SCHEMA.into(), + target: TARGET.into(), + plan_id: digest('1'), + incarnation_id: "odin/generation-1".into(), + sealed_release_id: digest('2'), + source_repository: "github.com/GameCult/Odin".into(), + source_revision: "3".repeat(40), + recipe_sha256: digest('4'), + runtime_id: "odin-runtime".into(), + expected_signer_identity_id: provider_signer.entry().identity_id.clone(), + health_contract: HEALTH_CONTRACT.into(), + artifact_sha256: digest('5'), + state_schema_generation: Some(STATE_SCHEMA_GENERATION.into()), + state_contract_sha256: Some(STATE_CONTRACT_SHA256.into()), + write_lease_required: true, + route: Some(IdunnExpectedRoute { + route_id: "odin-route".into(), + transport: "rudp".into(), + stable_endpoint: "rudp://odin.internal:1000".into(), + candidate_endpoint: format!("rudp://{candidate}"), + }), + capabilities: vec![IdunnExpectedCapability { + capability: RENDEZVOUS_CAPABILITY.into(), + schema: RENDEZVOUS_SCHEMA.into(), + compatibility: RENDEZVOUS_COMPATIBILITY.into(), + minimum_capacity: 1, + }], + dependencies: Vec::new(), + }; + expected.validate()?; + require_expected_contract(&expected, candidate)?; + let launch = + IdunnRuntimeActivationLaunch::issue(&expected, digest('7'), now - 20, &idunn_signer)?; + let activation = launch.activation().clone(); + let mut credential = Vec::new(); + launch.write_credential(&mut credential)?; + let activation_signer = + IdunnRuntimeActivationSigner::from_credential_reader(credential.as_slice())?; + let provider_anchor = GameCultServiceTrustAnchorRecord { + schema_version: GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA.into(), + trust_anchor_id: format!("root/{TARGET}/runtime-presence"), + service_id: TARGET.into(), + runtime_id: expected.runtime_id.clone(), + signer_identity_id: provider_signer.entry().identity_id.clone(), + signer_public_key: provider_signer.entry().public_key.clone(), + signature_algorithm: "ed25519".into(), + signing_purpose: GAMECULT_RUNTIME_PRESENCE_HEALTH_SIGNING_PURPOSE.into(), + signed_schema: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(), + binding_authority: "root".into(), + bound_at_unix_millis: now - 100, + expires_at_unix_millis: None, + private_state_exposed: false, + }; + let provider_identity_id = provider_signer.entry().identity_id.clone(); + let expected_sha256 = expected.canonical_sha256()?; + let authority_material = RuntimeAuthority { + activation_sha256: activation.canonical_sha256()?, + expected_sha256: expected_sha256.clone(), + expected: expected.clone(), + activation: activation.clone(), + activation_signer, + provider_signer, + }; + let store = root.join(TOPOLOGY_SLOT); + let projection_path = root.join("idunn-projection.cc"); + let lease_path = root.join("process-write-lease.cc"); + let mut runtime = RuntimeState { + options: Options { + store: store.clone(), + idunn_projection: projection_path.clone(), + idunn_anchor: root.join("unused-idunn-anchor.cc"), + }, + candidate, + authority_material, + idunn_anchor: Some(idunn_anchor), + topology_signer: Some(odin_signer), + topology: None, + write_lease: None, + write_lease_path: lease_path.clone(), + recent_warming_proofs: VecDeque::new(), + publisher_sequence: 0, + }; + + // The lease names the Warming presence Odin signed for Idunn's probe. + let warming = runtime.signed_presence_document("warming", "test warming")?; + let lease = IdunnProcessWriteLeaseRecord { + schema_version: IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into(), + target: TARGET.into(), + expected_projection_sha256: expected_sha256, + plan_id: expected.plan_id.clone(), + incarnation_id: expected.incarnation_id.clone(), + sealed_release_id: expected.sealed_release_id.clone(), + activation_witness_sha256: activation.canonical_sha256()?, + state_schema_generation: STATE_SCHEMA_GENERATION.into(), + state_contract_sha256: STATE_CONTRACT_SHA256.into(), + runtime_id: expected.runtime_id.clone(), + runtime_instance_id: activation.runtime_instance_id.clone(), + warming_presence_sha256: decode_presence(&warming.payload)?.canonical_sha256()?, + lease_epoch: 1, + issued_at_unix_millis: now - 5, + }; + write_single_record(&lease_path, lease_envelope(&lease)?)?; + File::create(sibling_lock_path(&lease_path)?)?; + + let incarnation_key = IncarnationRef::of(&expected)?.key(); + let mut projected_lease = lease_envelope(&lease)?; + projected_lease.key = incarnation_key.clone(); + let projection = vec![ + CultCacheEnvelope { + key: incarnation_key.clone(), + r#type: IdunnExpectedIncarnationRecord::TYPE.into(), + payload: expected.canonical_bytes()?, + stored_at: rfc3339_millis(now - 30)?, + schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), + }, + CultCacheEnvelope { + key: provider_anchor.trust_anchor_id.clone(), + r#type: GameCultServiceTrustAnchorRecord::TYPE.into(), + payload: rmp_serde::to_vec(&provider_anchor)?, + stored_at: rfc3339_millis(provider_anchor.bound_at_unix_millis)?, + schema_id: Some(GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA.into()), + }, + CultCacheEnvelope { + key: incarnation_key, + r#type: IdunnRuntimeActivationRecord::TYPE.into(), + payload: activation.canonical_bytes()?, + stored_at: rfc3339_millis(activation.issued_at_unix_millis)?, + schema_id: Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA.into()), + }, + projected_lease, + ]; + ensure!( + SingleFileMessagePackBackingStore::new(&projection_path) + .compare_exchange_snapshot(&[], &projection)?, + "test projection CAS failed" + ); + + ensure!( + runtime.try_activate()?, + "Odin did not activate in the fixture" + ); + let state = Rc::new(RefCell::new(runtime)); + let server = CultMeshRudpDocumentServer::new( + socket, + SinkHandle(state.clone()), + SnapshotHandle(state.clone()), + CultMeshSystemClock::default(), + CultMeshRudpDocumentServerOptions::default(), + )?; + Ok(OdinWorld { + _temp: temp, + state, + server, + timers: ServingTimers::default(), + store, + lease_path, + provider_identity_id, + lease, + }) + } + + impl OdinWorld { + fn pass(&mut self) -> Result { + serving_pass(&self.state, &mut self.server, &mut self.timers) + } + + /// Replace the lease under Odin with a later epoch. Odin holds a shared + /// lock on the lease's lifetime lock, which is what stops Idunn from + /// writing the lease under it; the test swaps the file underneath. + fn swap_lease(&self) -> Result<()> { + let mut replaced = self.lease.clone(); + replaced.lease_epoch += 1; + let scratch = self.lease_path.with_file_name("replacement-lease.cc"); + write_single_record(&scratch, lease_envelope(&replaced)?)?; + std::fs::copy(&scratch, &self.lease_path)?; + Ok(()) + } + + /// Add records to Idunn's projection as Idunn would publish them. + fn append_projection(&self, added: Vec) -> Result<()> { + self.tamper_projection(|mut entries| { + entries.extend(added); + entries + }) + } + + /// Rewrite Odin's own store as `change` sees it, in one exchange. + fn tamper_store( + &self, + change: impl FnOnce(Vec) -> Vec, + ) -> Result<()> { + let store = SingleFileMessagePackBackingStore::new(&self.store); + let current = store.pull_all_read_only_snapshot()?; + let next = change(current.clone()); + ensure!( + store.compare_exchange_snapshot(¤t, &next)?, + "test store CAS failed" + ); + Ok(()) + } + + /// Rewrite Idunn's projection as `change` sees it, in one exchange. + fn tamper_projection( + &self, + change: impl FnOnce(Vec) -> Vec, + ) -> Result<()> { + let path = self.state.borrow().options.idunn_projection.clone(); + let store = SingleFileMessagePackBackingStore::new(path); + let current = store.pull_all_read_only_snapshot()?; + let next = change(current.clone()); + ensure!( + store.compare_exchange_snapshot(¤t, &next)?, + "test projection CAS failed" + ); + Ok(()) + } + + /// The whole catalog, as a Verse peer reading it would get it. + fn catalog(&self) -> Result> { + self.state.borrow_mut().raw_snapshot(&CultMeshRudpSnapshotQuery { + session: cultmesh_rs::CultMeshRudpSessionKey { + remote_addr: "127.0.0.1:1".parse()?, + connection_id: 7, + }, + message_id: "catalog".into(), + requested_at_unix_millis: 1, + schema_ids: None, + record_keys: None, + }) + } + + fn stored_keys(&self, record_type: &str) -> Result> { + Ok(SingleFileMessagePackBackingStore::new(&self.store) + .pull_all_read_only_snapshot()? + .into_iter() + .filter(|entry| entry.r#type == record_type) + .map(|entry| entry.key) + .collect()) + } + + /// Odin's own publisher sequence as it is durably stored. + fn stored_sequence(&self) -> Result { + prior_self_publisher_sequence(&self.store, &self.provider_identity_id) + } + + /// Run a peer on its own thread while Odin serves. Every pass Odin + /// takes meanwhile must succeed. + fn serve_while( + &mut self, + peer: impl FnOnce() -> T + Send + 'static, + ) -> Result { + let peer = thread::spawn(peer); + while !peer.is_finished() { + if !self.pass()? { + thread::sleep(IDLE_POLL_INTERVAL); + } + } + Ok(peer.join().expect("peer thread panicked")) + } + } + + const PEER_CONNECTION_BASE: u32 = 0x7000_0000; + + /// A publisher the way Ghostlight and CodexConnector behave: connect, + /// optionally put one message, never Disconnect. + fn lingering_peer( + target: SocketAddr, + connection_id: u32, + message: Option, + ) -> Result<()> { + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_read_timeout(Some(Duration::from_millis(100)))?; + let mut transport = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("test-peer", socket, target, connection_id), + )?; + transport.connect(Vec::new())?; + let deadline = Instant::now() + Duration::from_millis(500); + while !transport.connected() { + let _ = transport.receive_once()?; + transport.poll_resends()?; + ensure!(Instant::now() < deadline, "timed out connecting"); + } + if let Some(message) = message { + let receipt = transport.send_reliable( + "schema", + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?, + )?; + let deadline = Instant::now() + Duration::from_millis(1500); + while transport.reliable_send_status(&receipt) == CultNetRudpReliableSendStatus::Pending + && Instant::now() < deadline + { + let _ = transport.receive_once()?; + transport.poll_resends()?; + } + } + Ok(()) + } + + /// Soul's 70-lingering-session probe: with the server's session table + /// full of publishers that never Disconnect, Odin's own heartbeat still + /// lands. The table is Odin's own production server bound to the candidate + /// socket, so a self-publication that went over RUDP would find it full. + #[test] + fn self_presence_lands_while_lingering_publishers_hold_every_session() -> Result<()> { + let mut odin = activated_odin()?; + let target = odin.server.local_addr()?; + let admitted = odin.serve_while(move || { + (0..70) + .filter(|index| lingering_peer(target, PEER_CONNECTION_BASE + index, None).is_ok()) + .count() + })?; + assert_eq!(admitted, 64, "the default session table is 64 wide"); + assert_eq!(odin.server.session_count(), 64, "the table is full"); + + let before = odin.stored_sequence()?; + for _ in 0..5 { + odin.timers.last_heartbeat = None; + odin.pass()?; + } + assert_eq!( + odin.stored_sequence()?, + before + 5, + "each heartbeat is admitted into Odin's store" + ); + assert_eq!(odin.server.session_count(), 64); + Ok(()) + } + + /// A publication or refresh that fails is logged and retried; it never + /// ends Odin, and the next attempt lands once the fault clears. + #[test] + fn a_failed_self_publication_or_refresh_does_not_end_odin() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + assert!(odin.stored_sequence()? > 0, "the first pass publishes"); + + std::fs::write(&odin.store, b"not a cultcache store")?; + assert!( + odin.state + .borrow_mut() + .publish_self_presence("probe") + .is_err(), + "the injected fault must actually fail the publication" + ); + odin.timers = ServingTimers::default(); + odin.pass()?; + + std::fs::remove_file(&odin.store)?; + odin.timers = ServingTimers::default(); + odin.pass()?; + assert!( + odin.stored_sequence()? > 0, + "the publication after the fault clears lands" + ); + Ok(()) + } + + /// Losing the write lease is the one condition that ends the serving loop. + #[test] + fn losing_the_write_lease_ends_odin() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + + odin.swap_lease()?; + odin.timers = ServingTimers::default(); + let error = odin.pass().unwrap_err(); + assert!( + error.downcast_ref::().is_some(), + "{error:#}" + ); + + std::fs::remove_file(&odin.lease_path)?; + odin.timers = ServingTimers::default(); + let error = odin.pass().unwrap_err(); + assert!( + error.downcast_ref::().is_some(), + "{error:#}" + ); + Ok(()) + } + + /// The heartbeat and the refresh each check the lease themselves: with only + /// one of them due, a swapped lease still ends Odin. + #[test] + fn each_timer_checks_the_write_lease_itself() -> Result<()> { + for heartbeat_due in [true, false] { + let mut odin = activated_odin()?; + odin.pass()?; + odin.swap_lease()?; + odin.timers = ServingTimers::default(); + if heartbeat_due { + odin.timers.last_projection_refresh = Some(Instant::now()); + } else { + odin.timers.last_heartbeat = Some(Instant::now()); + } + let error = odin.pass().unwrap_err(); + assert!( + error.downcast_ref::().is_some(), + "heartbeat_due={heartbeat_due}: {error:#}" + ); + } + Ok(()) + } + + /// A lease Odin cannot read is a lease it cannot show it still holds, so + /// it is lost, not a transient fault to retry. + #[test] + fn an_unreadable_write_lease_ends_odin() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + std::fs::write(&odin.lease_path, b"not a cultcache store")?; + odin.timers = ServingTimers::default(); + let error = odin.pass().unwrap_err(); + assert!( + error.downcast_ref::().is_some(), + "{error:#}" + ); + Ok(()) + } + + /// Soul's ghost probe: one Expected that will not decode is that + /// incarnation's own failure. The other incarnations still refresh, and + /// the catalog still serves. + #[test] + fn one_undecodable_incarnation_leaves_the_others_and_the_catalog_readable() -> Result<()> { + let mut odin = activated_odin()?; + let mut candidate = odin.state.borrow().authority_material.expected.clone(); + candidate.target = "sibling".into(); + candidate.validate()?; + let candidate_key = IncarnationRef::of(&candidate)?.key(); + // "aaghost" sorts before "odin", so a refresh that stops at its first + // failure never reaches the good incarnations. + let ghost_key = IncarnationRef::new("aaghost", digest('a')).key(); + odin.append_projection(vec![ + CultCacheEnvelope { + key: ghost_key, + r#type: IdunnExpectedIncarnationRecord::TYPE.into(), + payload: vec![0xc1, 0x00], + stored_at: rfc3339_millis(unix_millis()?)?, + schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), + }, + CultCacheEnvelope { + key: candidate_key.clone(), + r#type: IdunnExpectedIncarnationRecord::TYPE.into(), + payload: candidate.canonical_bytes()?, + stored_at: rfc3339_millis(unix_millis()?)?, + schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), + }, + ])?; + assert!( + !odin + .stored_keys(OdinRuntimeTopologyCorrelationRecord::TYPE)? + .contains(&candidate_key) + ); + + // A catalog read serves what is stored; refreshing is the timer's. + odin.catalog()?; + assert!( + !odin + .stored_keys(OdinRuntimeTopologyCorrelationRecord::TYPE)? + .contains(&candidate_key), + "a read does not refresh" + ); + + odin.pass()?; + assert!( + odin.stored_keys(OdinRuntimeTopologyCorrelationRecord::TYPE)? + .contains(&candidate_key), + "the good incarnation was refreshed past the ghost" + ); + + let catalog = odin.catalog()?; + assert!( + catalog + .iter() + .any(|document| document.schema_id == ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA), + "the catalog serves its correlations" + ); + Ok(()) + } + + fn assert_refused(error: &anyhow::Error) { + assert!( + error.downcast_ref::().is_some(), + "{error:#}" + ); + } + + /// Temporary rule (see `survive`): a self-presence Odin's authority will + /// never admit ends Odin. With no current activation Idunn projects, there + /// is no verifiable authority. + #[test] + fn a_projection_without_authority_ends_odin() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + odin.tamper_projection(|entries| { + entries + .into_iter() + .filter(|entry| entry.r#type != IdunnRuntimeActivationRecord::TYPE) + .collect() + })?; + odin.timers = ServingTimers::default(); + assert_refused(&odin.pass().unwrap_err()); + Ok(()) + } + + /// Idunn's activation of Odin carries a signature Idunn's anchor does not + /// verify: the authority cannot be verified. + #[test] + fn an_activation_that_does_not_verify_ends_odin() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + odin.tamper_projection(|entries| { + entries + .into_iter() + .map(|mut entry| { + if entry.r#type == IdunnRuntimeActivationRecord::TYPE { + let mut activation = + IdunnRuntimeActivationRecord::decode_canonical(&entry.payload) + .unwrap(); + activation.signature[0] ^= 1; + entry.payload = activation.canonical_bytes().unwrap(); + } + entry + }) + .collect() + })?; + odin.timers = ServingTimers::default(); + assert_refused(&odin.pass().unwrap_err()); + Ok(()) + } + + /// The projected provider anchor names Odin's signer identity but carries + /// a key that identity does not derive from: the authority cannot verify. + #[test] + fn a_signer_that_does_not_match_the_anchor_ends_odin() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + odin.tamper_projection(|entries| { + entries + .into_iter() + .map(|mut entry| { + if entry.r#type == GameCultServiceTrustAnchorRecord::TYPE { + let mut anchor: GameCultServiceTrustAnchorRecord = + rmp_serde::from_slice(&entry.payload).unwrap(); + anchor.signer_public_key = vec![7; 32]; + entry.payload = rmp_serde::to_vec(&anchor).unwrap(); + } + entry + }) + .collect() + })?; + odin.timers = ServingTimers::default(); + assert_refused(&odin.pass().unwrap_err()); + + // The other way round: the authority is sound, but the key Odin signs + // with is not the Expected signer. + let mut odin = activated_odin()?; + odin.pass()?; + let other = odin._temp.path().join("other-provider.cc"); + odin.state.borrow_mut().authority_material.provider_signer = + enroll_service_identity_at::(&other)?; + odin.timers = ServingTimers::default(); + assert_refused(&odin.pass().unwrap_err()); + Ok(()) + } + + /// A stored presence of Odin's own activation that no longer authenticates + /// makes its publisher sequence unknowable; the claim is refused, closed. + #[test] + fn a_stored_presence_that_no_longer_authenticates_ends_odin() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + // Received two minutes after it was observed: outside the trusted window. + let late = rfc3339_millis(unix_millis()? + 120_000)?; + odin.tamper_store(|entries| { + entries + .into_iter() + .map(|mut entry| { + if entry.r#type == GameCultRuntimePresenceHealthRecord::TYPE { + entry.stored_at = late.clone(); + } + entry + }) + .collect() + })?; + odin.timers = ServingTimers::default(); + assert_refused(&odin.pass().unwrap_err()); + Ok(()) + } + + /// An application rejection ends only the offending peer's session, and a + /// stray datagram is dropped; neither ends Odin's serving loop. + #[test] + fn an_application_rejection_or_a_stray_packet_does_not_end_odin() -> Result<()> { + let mut odin = activated_odin()?; + let target = odin.server.local_addr()?; + let not_a_presence = CultNetMessage::DocumentPutRaw { + message_id: "rejected".into(), + document: CultNetRawDocumentRecord { + schema_id: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(), + record_key: "ghostlight".into(), + stored_at: rfc3339_millis(unix_millis()?)?, + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: vec![0xc1], + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + }, + }; + odin.serve_while(move || { + lingering_peer(target, PEER_CONNECTION_BASE, Some(not_a_presence)) + })??; + assert_eq!( + odin.server.session_count(), + 0, + "the rejection ended the offending session" + ); + + UdpSocket::bind("127.0.0.1:0")?.send_to(b"junk", target)?; + for _ in 0..50 { + odin.pass()?; + thread::sleep(IDLE_POLL_INTERVAL); + } + assert!(odin.server.packets_dropped() >= 1); + + odin.serve_while(move || lingering_peer(target, PEER_CONNECTION_BASE + 1, None))??; + assert_eq!(odin.server.session_count(), 1, "Odin still admits peers"); + Ok(()) + } + + #[test] + fn only_the_write_lease_is_fatal_in_the_serving_loop() { + assert_eq!(survive("x", Ok(3)).unwrap(), Some(3)); + assert_eq!( + survive::<()>("x", Err(anyhow::anyhow!("io"))).unwrap(), + None + ); + let lost = anyhow::Error::new(WriteLeaseLost("gone".into())).context("refreshing"); + assert!(survive::<()>("x", Err(lost)).is_err()); + } + + #[test] + fn a_failing_socket_is_retried_and_ends_odin_only_after_the_failure_limit() { + let start = Instant::now(); + let mut since = None; + let fault = || Err(anyhow::anyhow!("socket fault")); + assert!(!settle_poll(fault(), &mut since, start).unwrap()); + assert_eq!(since, Some(start)); + assert!( + !settle_poll( + fault(), + &mut since, + start + POLL_FAILURE_LIMIT - Duration::from_millis(1) + ) + .unwrap() + ); + // A successful poll ends the run, so the same fault starts a new one. + assert!(settle_poll(Ok(CultMeshRudpPollOutcome::Handled), &mut since, start).unwrap()); + assert_eq!(since, None); + let restart = start + POLL_FAILURE_LIMIT; + assert!(!settle_poll(fault(), &mut since, restart).unwrap()); + assert!(settle_poll(fault(), &mut since, restart + POLL_FAILURE_LIMIT).is_err()); + } } diff --git a/deploy/cultlib.commit b/deploy/cultlib.commit deleted file mode 100644 index 8fd097a..0000000 --- a/deploy/cultlib.commit +++ /dev/null @@ -1 +0,0 @@ -542ddd3e5830ad7dc44e69298e5469f97f992422