diff --git a/crates/odin-daemon/src/lib.rs b/crates/odin-daemon/src/lib.rs index 7c2f115..7549b75 100644 --- a/crates/odin-daemon/src/lib.rs +++ b/crates/odin-daemon/src/lib.rs @@ -4,14 +4,15 @@ //! read-only projection, a trusted receive time with any service-owned //! presence, and narrow clock/store/signer implementations. -use std::collections::BTreeMap; -use std::path::{Path, PathBuf}; +use std::cell::{Cell, Ref, RefCell}; +use std::collections::{BTreeMap, BTreeSet}; +use std::path::PathBuf; use std::time::{SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result, bail, ensure}; use chrono::{DateTime, SecondsFormat, Utc}; use cultcache_rs::{ - CultCacheEnvelope, CultCacheExpectedEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, + CacheBackingStore, CultCacheEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, }; use cultnet_rs::{ GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, GAMECULT_RUNTIME_PRESENCE_HEALTH_SIGNING_PURPOSE, @@ -29,8 +30,6 @@ use cultnet_rs::{ verify_service_identity_signature_with_public_key, }; -const CAS_ATTEMPTS: usize = 8; - /// Odin's durable high-water mark for one target's correlation sequence. /// /// The correlation's `publisher_sequence` must strictly increase for as long as @@ -178,7 +177,9 @@ pub trait IdunnProjectionSource { } } -/// Read-only adapter for the atomic projection file published by Idunn. +/// Read-only adapter for the atomic projection file published by Idunn. It +/// answers no lookup itself: each [`snapshot`](Self::snapshot) is one read of +/// the file, and every lookup is made against a snapshot. pub struct CultCacheIdunnProjectionSource { path: PathBuf, } @@ -188,23 +189,61 @@ impl CultCacheIdunnProjectionSource { Self { path: path.into() } } - pub fn path(&self) -> &Path { - &self.path + /// Read the projection file once. An absent file projects nothing. + pub fn snapshot(&self) -> Result { + if !self.path.is_file() { + return Ok(IdunnProjectionSnapshot { + entries: Vec::new(), + }); + } + SingleFileMessagePackBackingStore::new(&self.path) + .pull_all_read_only_snapshot() + .map(|entries| IdunnProjectionSnapshot { entries }) + .map_err(|error| { + // The store's decode step reports its failures as text, so an + // I/O error in the chain here is from opening or reading the + // file and from nothing the file said. + if error.chain().any(|cause| cause.is::()) { + error.context(ProjectionUnreadable) + } else { + error + } + }) } } -impl IdunnProjectionSource for CultCacheIdunnProjectionSource { +/// One reading of Idunn's projection file. Every lookup made against it is +/// answered from the same bytes. The default projects nothing. +#[derive(Default)] +pub struct IdunnProjectionSnapshot { + entries: Vec, +} + +impl IdunnProjectionSnapshot { + /// Every incarnation Idunn projects an Expected for, of every target. Only + /// the keys are read: decoding a record is the refresh of that one + /// incarnation, so a record that will not decode cannot make the list + /// unreadable. A record keyed by anything but an incarnation key is not + /// this contract's and is skipped. + pub fn incarnations(&self) -> BTreeSet { + self.entries + .iter() + .filter(|entry| entry.r#type == IdunnExpectedIncarnationRecord::TYPE) + .filter_map(|entry| IncarnationRef::parse_key(&entry.key)) + .collect() + } +} + +impl IdunnProjectionSource for IdunnProjectionSnapshot { fn projections(&self, target: &str) -> Result> { - let Some(entries) = self.entries()? else { - return Ok(Vec::new()); - }; - let anchor = provider_anchor(&entries, target)?; + let entries = &self.entries; + 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 // this reader will act on, rather than something it will misread. let mut projections = Vec::new(); - for envelope in &entries { + for envelope in entries { if envelope.r#type != IdunnExpectedIncarnationRecord::TYPE { continue; } @@ -215,7 +254,7 @@ impl IdunnProjectionSource for CultCacheIdunnProjectionSource { continue; } projections.push(assemble_projection( - &entries, + entries, anchor.clone(), envelope, &incarnation, @@ -229,40 +268,17 @@ impl IdunnProjectionSource for CultCacheIdunnProjectionSource { /// 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 entries = &self.entries; let Some(envelope) = unique_envelope( - &entries, + 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); - } - SingleFileMessagePackBackingStore::new(&self.path) - .pull_all_read_only_snapshot() - .map(Some) - .map_err(|error| { - // The store's decode step reports its failures as text, so an - // I/O error in the chain here is from opening or reading the - // file and from nothing the file said. - if error.chain().any(|cause| cause.is::()) { - error.context(ProjectionUnreadable) - } else { - error - } - }) + let anchor = provider_anchor(entries, &incarnation.target)?; + assemble_projection(entries, anchor, envelope, incarnation).map(Some) } } @@ -410,110 +426,261 @@ pub struct OdinStoreSnapshot { pub trait OdinTopologyStore { fn read(&self, query: OdinStoreQuery) -> Result; - fn compare_exchange( + fn commit( &self, - observed: &OdinStoreSnapshot, + query: &OdinStoreQuery, replacement_presence: Option<&AdmittedPresence>, replacement_correlation: &StoredCorrelation, - ) -> Result; + ) -> Result<()>; fn withdraw_correlation(&self, incarnation: &IncarnationRef) -> Result<()>; /// Claim the next correlation sequence for this target, never returning a - /// value already used. `prior` is the sequence of the currently stored + /// value already published. `prior` is the sequence of the currently stored /// correlation, or 0 when there is none. /// - /// The mark is advanced before the correlation that uses it is published, - /// so a crash in between leaves it ahead. That is the safe direction: - /// sequences may skip, they may never repeat. + /// A sequence is published when the store is flushed. The mark is written + /// in the same atomic write as every correlation that uses it, so the file + /// never holds a correlation ahead of its mark. A sequence reserved and + /// never flushed was never published, and may be claimed again after a + /// restart. fn reserve_publisher_sequence(&self, target: &str, prior: u64) -> Result; } -/// Odin's durable replay and correlation store. It persists existing CultNet -/// documents directly instead of inventing a parallel admission schema. -pub struct CultCacheOdinTopologyStore { - path: PathBuf, -} - -impl CultCacheOdinTopologyStore { - pub fn new(path: impl Into) -> Self { - Self { path: path.into() } +impl OdinTopologyStore for &T { + fn read(&self, query: OdinStoreQuery) -> Result { + (*self).read(query) } - pub fn path(&self) -> &Path { - &self.path + fn commit( + &self, + query: &OdinStoreQuery, + replacement_presence: Option<&AdmittedPresence>, + replacement_correlation: &StoredCorrelation, + ) -> Result<()> { + (*self).commit(query, replacement_presence, replacement_correlation) + } + + fn withdraw_correlation(&self, incarnation: &IncarnationRef) -> Result<()> { + (*self).withdraw_correlation(incarnation) + } + + fn reserve_publisher_sequence(&self, target: &str, prior: u64) -> Result { + (*self).reserve_publisher_sequence(target, prior) + } +} + +/// One record's identity in a CultCache store: its type, then its key. +pub type RecordId = (String, String); + +fn record_id(envelope: &CultCacheEnvelope) -> RecordId { + (envelope.r#type.clone(), envelope.key.clone()) +} + +/// Odin's store, held in memory for as long as Odin holds its write lease. It +/// persists existing CultNet documents directly instead of inventing a +/// parallel admission schema. +/// +/// The file is read once, by [`MemoryOdinTopologyStore::load`] when the lease +/// is granted, and never again while Odin serves: every lookup is answered +/// from memory. The file is Odin's durability checkpoint and Idunn's read-only +/// crossing. It is written by [`MemoryOdinTopologyStore::flush`] and nothing +/// else. +pub struct MemoryOdinTopologyStore { + path: PathBuf, + records: RefCell>, + /// Exactly what the file held when it was loaded or last written. + flushed: RefCell>, + /// What every write since the last one that succeeded tried to write. A + /// write can fail after it replaced the file (syncing the directory) or + /// before (anything earlier), and the error does not say which, so the + /// file may hold any of these: each is Odin's own, not another writer's. + /// One is kept per failed write whose records differ from the last. + attempted: RefCell>>, + /// The records differ from what was last loaded or written. + dirty: Cell, +} + +impl MemoryOdinTopologyStore { + /// Read the store file under its lock. An absent file is an empty store. + /// Records are loaded as they are stored: one that does not decode is that + /// record's failure, met where the record is used, and never stops the + /// load. + pub fn load(path: impl Into) -> Result { + let path = path.into(); + let flushed = SingleFileMessagePackBackingStore::new(&path).pull_all()?; + let mut records = BTreeMap::new(); + for envelope in &flushed { + if records.insert(record_id(envelope), envelope.clone()).is_some() { + eprintln!( + "Odin's store holds {} {} more than once; the last is kept", + envelope.r#type, envelope.key + ); + } + } + Ok(Self { + path, + records: RefCell::new(records), + flushed: RefCell::new(flushed), + attempted: RefCell::new(Vec::new()), + dirty: Cell::new(false), + }) + } + + pub fn records(&self) -> Ref<'_, BTreeMap> { + self.records.borrow() + } + + /// Store one record in place of any with the same type and key. Storing + /// the record already held changes nothing. + pub fn put(&self, envelope: CultCacheEnvelope) { + let id = record_id(&envelope); + let mut records = self.records.borrow_mut(); + if records.get(&id) != Some(&envelope) { + records.insert(id, envelope); + self.dirty.set(true); + } + } + + /// Remove the record with this type and key, if one is held. + pub fn remove(&self, record_type: &str, key: &str) { + if self + .records + .borrow_mut() + .remove(&(record_type.to_owned(), key.to_owned())) + .is_some() + { + self.dirty.set(true); + } + } + + /// Whether Odin's correlation of this incarnation says it is Ready. No + /// correlation, or one that does not decode, is not Ready. + pub fn correlation_ready(&self, incarnation: &IncarnationRef) -> bool { + self.records + .borrow() + .get(&( + OdinRuntimeTopologyCorrelationRecord::TYPE.to_owned(), + incarnation.key(), + )) + .and_then(|envelope| decode_correlation(&envelope.payload).ok()) + .is_some_and(|correlation| correlation.ready) + } + + pub fn correlated_incarnations(&self) -> BTreeSet { + self.records + .borrow() + .keys() + .filter(|(record_type, _)| record_type == OdinRuntimeTopologyCorrelationRecord::TYPE) + .filter_map(|(_, key)| IncarnationRef::parse_key(key)) + .collect() } /// Remove every correlation not keyed by an incarnation. Those were written /// by the previous, target-keyed Odin; nothing reads them under this - /// contract, and left in place `stored_snapshot` would serve them to the - /// Verse as current. Presence records are kept whatever their key: the - /// self publisher sequence is continued from them. - pub fn retire_legacy_correlations(&self) -> Result { - for _ in 0..CAS_ATTEMPTS { - if !self.path.is_file() { - return Ok(0); + /// contract, and left in place the catalog would serve them to the Verse + /// as current. + pub fn retire_legacy_correlations(&self) -> usize { + let legacy: Vec = self + .records + .borrow() + .keys() + .filter(|(record_type, key)| { + record_type == OdinRuntimeTopologyCorrelationRecord::TYPE + && IncarnationRef::parse_key(key).is_none() + }) + .map(|(_, key)| key.clone()) + .collect(); + for key in &legacy { + self.remove(OdinRuntimeTopologyCorrelationRecord::TYPE, key); + } + legacy.len() + } + + pub fn is_dirty(&self) -> bool { + self.dirty.get() + } + + /// Write every record to the file as one atomic snapshot, if anything + /// changed since the last write; returns whether it wrote. + /// + /// The file must still hold exactly what Odin last loaded or wrote, or what + /// one of the failed writes since then tried to write. Anything else was + /// written by another process, and the write is refused with + /// [`ForeignStoreWrite`] rather than made over it. A failed write leaves + /// the records dirty, so the next write retries it. + pub fn flush(&self) -> Result { + if !self.dirty.get() { + return Ok(false); + } + let current: Vec = self.records.borrow().values().cloned().collect(); + let file = SingleFileMessagePackBackingStore::new(&self.path); + let mut exchanged = file.compare_exchange_snapshot(&self.flushed.borrow(), ¤t); + for attempted in self.attempted.borrow().iter().rev() { + if !matches!(exchanged, Ok(false)) { + break; } - let entries = - SingleFileMessagePackBackingStore::new(&self.path).pull_all_read_only_snapshot()?; - let retained = entries - .iter() - .filter(|entry| { - !(entry.r#type == OdinRuntimeTopologyCorrelationRecord::TYPE - && IncarnationRef::parse_key(&entry.key).is_none()) - }) - .cloned() - .collect::>(); - let removed = entries.len() - retained.len(); - if removed == 0 { - return Ok(0); - } - if SingleFileMessagePackBackingStore::new(&self.path) - .compare_exchange_snapshot(&entries, &retained)? - { - return Ok(removed); + exchanged = file.compare_exchange_snapshot(attempted, ¤t); + } + match exchanged { + Ok(true) => {} + Ok(false) => return Err(ForeignStoreWrite.into()), + Err(error) => { + let mut attempted = self.attempted.borrow_mut(); + if attempted.last() != Some(¤t) { + attempted.push(current); + } + return Err(error); } } - bail!("Odin topology store changed repeatedly while retiring legacy correlations") + *self.flushed.borrow_mut() = current; + self.attempted.borrow_mut().clear(); + self.dirty.set(false); + Ok(true) } } -impl OdinTopologyStore for CultCacheOdinTopologyStore { +/// Odin's store file no longer holds what Odin last loaded or wrote, so another +/// process wrote it. While Odin holds its write lease it is the file's only +/// writer: this is a second writer, and Odin must not write over it. +#[derive(Debug)] +pub struct ForeignStoreWrite; + +impl std::fmt::Display for ForeignStoreWrite { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("Odin's store file was written by another process") + } +} + +impl std::error::Error for ForeignStoreWrite {} + +impl OdinTopologyStore for MemoryOdinTopologyStore { fn read(&self, mut query: OdinStoreQuery) -> Result { query.dependencies.sort(); query.dependencies.dedup(); - let entries = if self.path.is_file() { - SingleFileMessagePackBackingStore::new(&self.path).pull_all_read_only_snapshot()? - } else { - Vec::new() - }; - let presence_key = - presence_store_key(&query.incarnation, &query.provider_signer_identity_id); - let presence = unique_envelope( - &entries, + let records = self.records.borrow(); + let find = + |record_type: &str, key: &str| records.get(&(record_type.to_owned(), key.to_owned())); + let presence = find( GameCultRuntimePresenceHealthRecord::TYPE, - &presence_key, - )? + &presence_store_key(&query.incarnation, &query.provider_signer_identity_id), + ) .map(decode_presence_envelope) .transpose()?; - let correlation = unique_envelope( - &entries, + let correlation = find( OdinRuntimeTopologyCorrelationRecord::TYPE, &query.incarnation.key(), - )? + ) .map(decode_correlation_envelope) .transpose()?; let mut dependency_correlations = BTreeMap::new(); for dependency in &query.dependencies { - let current = unique_envelope( - &entries, - OdinRuntimeTopologyCorrelationRecord::TYPE, - &dependency.key(), - )? - .map(decode_correlation_envelope) - .transpose()?; + let current = find(OdinRuntimeTopologyCorrelationRecord::TYPE, &dependency.key()) + .map(decode_correlation_envelope) + .transpose()?; dependency_correlations.insert(dependency.clone(), current); } + drop(records); Ok(OdinStoreSnapshot { query, presence, @@ -522,128 +689,50 @@ impl OdinTopologyStore for CultCacheOdinTopologyStore { }) } - fn compare_exchange( + fn commit( &self, - observed: &OdinStoreSnapshot, + query: &OdinStoreQuery, replacement_presence: Option<&AdmittedPresence>, replacement_correlation: &StoredCorrelation, - ) -> Result { - let presence_key = presence_store_key( - &observed.query.incarnation, - &observed.query.provider_signer_identity_id, - ); - let correlation_key = observed.query.incarnation.key(); - let mut conditions = BTreeMap::new(); - insert_condition( - &mut conditions, - CultCacheExpectedEnvelope { - key: presence_key.clone(), - r#type: GameCultRuntimePresenceHealthRecord::TYPE.into(), - current: observed - .presence - .as_ref() - .map(|presence| presence_envelope(&presence_key, presence)), - }, - )?; - insert_condition( - &mut conditions, - CultCacheExpectedEnvelope { - key: correlation_key.clone(), - r#type: OdinRuntimeTopologyCorrelationRecord::TYPE.into(), - current: observed - .correlation - .as_ref() - .map(|correlation| correlation_envelope(&correlation_key, correlation)), - }, - )?; - for (dependency, correlation) in &observed.dependency_correlations { - let key = dependency.key(); - insert_condition( - &mut conditions, - CultCacheExpectedEnvelope { - key: key.clone(), - r#type: OdinRuntimeTopologyCorrelationRecord::TYPE.into(), - current: correlation - .as_ref() - .map(|correlation| correlation_envelope(&key, correlation)), - }, - )?; - } - let mut replacements = vec![correlation_envelope( - &correlation_key, + ) -> Result<()> { + self.put(correlation_envelope( + &query.incarnation.key(), replacement_correlation, - )]; + )); if let Some(presence) = replacement_presence { - replacements.push(presence_envelope(&presence_key, presence)); + self.put(presence_envelope( + &presence_store_key(&query.incarnation, &query.provider_signer_identity_id), + presence, + )); } - if let Some(parent) = self.path.parent() { - std::fs::create_dir_all(parent)?; - } - SingleFileMessagePackBackingStore::new(&self.path) - .compare_exchange(&conditions.into_values().collect::>(), &replacements) + Ok(()) } fn withdraw_correlation(&self, incarnation: &IncarnationRef) -> Result<()> { - let key = incarnation.key(); - for _ in 0..CAS_ATTEMPTS { - if !self.path.is_file() { - return Ok(()); - } - let entries = - SingleFileMessagePackBackingStore::new(&self.path).pull_all_read_only_snapshot()?; - let current = - unique_envelope(&entries, OdinRuntimeTopologyCorrelationRecord::TYPE, &key)?; - let Some(current) = current else { - return Ok(()); - }; - let replacement = entries - .iter() - .filter(|entry| *entry != current) - .cloned() - .collect::>(); - if SingleFileMessagePackBackingStore::new(&self.path) - .compare_exchange_snapshot(&entries, &replacement)? - { - return Ok(()); - } - } - bail!("Odin topology store changed repeatedly while withdrawing correlation") + self.remove(OdinRuntimeTopologyCorrelationRecord::TYPE, &incarnation.key()); + Ok(()) } fn reserve_publisher_sequence(&self, target: &str, prior: u64) -> Result { - for _ in 0..CAS_ATTEMPTS { - let entries = if self.path.is_file() { - SingleFileMessagePackBackingStore::new(&self.path).pull_all_read_only_snapshot()? - } else { - Vec::new() - }; - let current = unique_envelope(&entries, TOPOLOGY_PUBLISHER_WATERMARK_TYPE, target)?; - let stored = current - .map(|envelope| decode_publisher_watermark(target, &envelope.payload)) - .transpose()? - .unwrap_or(0); - // The stored correlation can be ahead of the mark on the first pass - // after this record was introduced, so take whichever is higher. - let next = stored - .max(prior) - .checked_add(1) - .context("Odin topology publisher sequence exhausted")?; - let replacement = publisher_watermark_envelope(target, next, &Utc::now().to_rfc3339())?; - if let Some(parent) = self.path.parent() { - std::fs::create_dir_all(parent)?; - } - if SingleFileMessagePackBackingStore::new(&self.path).compare_exchange( - &[CultCacheExpectedEnvelope { - key: target.into(), - r#type: TOPOLOGY_PUBLISHER_WATERMARK_TYPE.into(), - current: current.cloned(), - }], - &[replacement], - )? { - return Ok(next); - } - } - bail!("Odin topology store changed repeatedly while reserving a publisher sequence") + let stored = self + .records + .borrow() + .get(&(TOPOLOGY_PUBLISHER_WATERMARK_TYPE.to_owned(), target.to_owned())) + .map(|envelope| decode_publisher_watermark(target, &envelope.payload)) + .transpose()? + .unwrap_or(0); + // The stored correlation can be ahead of the mark on the first pass + // after this record was introduced, so take whichever is higher. + let next = stored + .max(prior) + .checked_add(1) + .context("Odin topology publisher sequence exhausted")?; + self.put(publisher_watermark_envelope( + target, + next, + &Utc::now().to_rfc3339(), + )?); + Ok(next) } } @@ -796,184 +885,177 @@ where dependencies: managed_dependency_incarnations(&projection.expected.dependencies), }; - for _ in 0..CAS_ATTEMPTS { - let observed = self.store.read(query.clone())?; - let admitted = self.select_presence(&observed, incoming, authority.as_ref())?; - let dependencies = - self.dependency_evidence(&projection.expected.dependencies, &observed, now)?; - let (presence, mut disagreements, replacement_presence) = match admitted { - SelectedPresence::None => (None, projection_disagreements.clone(), None), - SelectedPresence::Stale(stored) => { - let record = decode_presence(&stored.canonical_bytes)?; - let mut disagreements = projection_disagreements.clone(); - disagreements.push(OdinTopologyDisagreement { - code: "stored-presence-not-current".into(), - expected: projection - .activation - .as_ref() - .map(IdunnRuntimeActivationRecord::canonical_sha256) - .transpose()?, - observed: Some(record.activation_witness_sha256), - }); - (None, disagreements, None) - } - SelectedPresence::Current { - correlation, - replacement, - } => { - let mut disagreements = projection_disagreements.clone(); - disagreements.extend_from_slice(correlation.disagreements()); - (Some(correlation), disagreements, replacement) - } - }; - if let Some(disagreement) = &lease_disagreement { - disagreements.push(disagreement.clone()); - } - if presence.is_none() { + let observed = self.store.read(query)?; + let admitted = self.select_presence(&observed, incoming, authority.as_ref())?; + let dependencies = + self.dependency_evidence(&projection.expected.dependencies, &observed, now)?; + let (presence, mut disagreements, replacement_presence) = match admitted { + SelectedPresence::None => (None, projection_disagreements.clone(), None), + SelectedPresence::Stale(stored) => { + let record = decode_presence(&stored.canonical_bytes)?; + let mut disagreements = projection_disagreements.clone(); disagreements.push(OdinTopologyDisagreement { - code: "signed-presence-missing".into(), - expected: Some("dual-authenticated-runtime-presence".into()), - observed: None, + code: "stored-presence-not-current".into(), + expected: projection + .activation + .as_ref() + .map(IdunnRuntimeActivationRecord::canonical_sha256) + .transpose()?, + observed: Some(record.activation_witness_sha256), }); - disagreements.extend(projection.expected.capabilities.iter().enumerate().map( - |(index, capability)| OdinTopologyDisagreement { - code: format!("expected-capability-{index:03}-missing"), - expected: Some(format!( - "{}/{}/{} capacity>={}", - capability.capability, - capability.schema, - capability.compatibility, - capability.minimum_capacity - )), - observed: None, - }, - )); + (None, disagreements, None) } + SelectedPresence::Current { + correlation, + replacement, + } => { + let mut disagreements = projection_disagreements.clone(); + disagreements.extend_from_slice(correlation.disagreements()); + (Some(correlation), disagreements, replacement) + } + }; + if let Some(disagreement) = &lease_disagreement { + disagreements.push(disagreement.clone()); + } + if presence.is_none() { + disagreements.push(OdinTopologyDisagreement { + code: "signed-presence-missing".into(), + expected: Some("dual-authenticated-runtime-presence".into()), + observed: None, + }); + disagreements.extend(projection.expected.capabilities.iter().enumerate().map( + |(index, capability)| OdinTopologyDisagreement { + code: format!("expected-capability-{index:03}-missing"), + expected: Some(format!( + "{}/{}/{} capacity>={}", + capability.capability, + capability.schema, + capability.compatibility, + capability.minimum_capacity + )), + observed: None, + }, + )); + } - let observed_write_lease = if let Some(presence) = &presence { - correlate_write_lease( - &projection.expected, - lease_sha256.as_deref(), - presence.claim().record().write_lease_sha256.as_deref(), - &presence.claim().record().state, - &mut disagreements, - ) - } else { - None - }; - disagreements.sort_by(|left, right| left.code.cmp(&right.code)); + let observed_write_lease = if let Some(presence) = &presence { + correlate_write_lease( + &projection.expected, + lease_sha256.as_deref(), + presence.claim().record().write_lease_sha256.as_deref(), + &presence.claim().record().state, + &mut disagreements, + ) + } else { + None + }; + disagreements.sort_by(|left, right| left.code.cmp(&right.code)); - let prior = observed - .correlation + let prior = observed + .correlation + .as_ref() + .map(|stored| decode_correlation(&stored.canonical_bytes)) + .transpose()?; + let presence_record = presence.as_ref().map(|value| value.claim().record()); + let ready = presence_record.is_some_and(|record| record.state == "active") + && disagreements.is_empty() + && dependencies + .iter() + .all(|dependency| dependency_permits_ready(&dependency.kind, dependency.ready)); + let mut correlation = OdinRuntimeTopologyCorrelationRecord { + schema_version: ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA.into(), + target: projection.expected.target.clone(), + expected_projection_sha256: projection.expected.canonical_sha256()?, + expected: true, + current_activation_sha256: projection + .activation .as_ref() - .map(|stored| decode_correlation(&stored.canonical_bytes)) - .transpose()?; - let presence_record = presence.as_ref().map(|value| value.claim().record()); - let ready = presence_record.is_some_and(|record| record.state == "active") - && disagreements.is_empty() - && dependencies - .iter() - .all(|dependency| dependency_permits_ready(&dependency.kind, dependency.ready)); - let mut correlation = OdinRuntimeTopologyCorrelationRecord { - schema_version: ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA.into(), - target: projection.expected.target.clone(), - expected_projection_sha256: projection.expected.canonical_sha256()?, - expected: true, - current_activation_sha256: projection - .activation - .as_ref() - .map(IdunnRuntimeActivationRecord::canonical_sha256) - .transpose()?, - signed_presence_sha256: presence - .as_ref() - .map(|value| value.claim().signed_presence_sha256().into()), - observed_presence_state: presence_record.map(|record| record.state.clone()), - observed_presence_publisher_sequence: presence_record - .map(|record| record.publisher_sequence), - observed_write_lease_sha256: observed_write_lease, - observed_capabilities: presence_record - .map_or_else(Vec::new, |record| record.capabilities.clone()), - runtime_id: projection.expected.runtime_id.clone(), - runtime_instance_id: projection - .activation - .as_ref() - .map(|activation| activation.runtime_instance_id.clone()), - present: presence.is_some(), - ready, - dependencies, - disagreements, - signer_identity_id: self.signer.identity_id().into(), - // Placeholder. A sequence is only reserved once the facts are - // known to have changed, below; same_correlation_facts compares - // against the prior's sequence regardless of what sits here. - publisher_sequence: prior.as_ref().map_or(0, |record| record.publisher_sequence), - observed_at_unix_millis: now, - signature_algorithm: "ed25519".into(), - signature: Vec::new(), - }; - if let Some(prior) = &prior - && same_correlation_facts(prior, &correlation) - { - return Ok(observed - .correlation - .context("decoded prior correlation disappeared")? - .canonical_bytes); - } - // Reserved only after the dedupe: a refresh that observes nothing - // new must not burn a sequence or fsync the store. Reserved from a - // mark that survives withdrawal, not derived from the stored - // correlation, so a withdrawn correlation does not take the count - // with it. See TOPOLOGY_PUBLISHER_WATERMARK_SCHEMA. - correlation.publisher_sequence = self.store.reserve_publisher_sequence( - &observed.query.incarnation.target, - prior.as_ref().map_or(0, |record| record.publisher_sequence), - )?; - correlation.signature = self - .signer - .sign_correlation(&correlation.unsigned_signature_payload()?)?; - let canonical = correlation.canonical_bytes()?; - correlation.validate_against_expected(&projection.expected, lease_sha256.as_deref())?; - let (signed, unsigned) = - OdinRuntimeTopologyCorrelationRecord::decode_canonical_signed_payload(&canonical)?; - verify_service_identity_signature_with_public_key::< - OdinTopologyIdentity, - OdinRuntimeTopologyCorrelationPurpose, - >( + .map(IdunnRuntimeActivationRecord::canonical_sha256) + .transpose()?, + signed_presence_sha256: presence + .as_ref() + .map(|value| value.claim().signed_presence_sha256().into()), + observed_presence_state: presence_record.map(|record| record.state.clone()), + observed_presence_publisher_sequence: presence_record + .map(|record| record.publisher_sequence), + observed_write_lease_sha256: observed_write_lease, + observed_capabilities: presence_record + .map_or_else(Vec::new, |record| record.capabilities.clone()), + runtime_id: projection.expected.runtime_id.clone(), + runtime_instance_id: projection + .activation + .as_ref() + .map(|activation| activation.runtime_instance_id.clone()), + present: presence.is_some(), + ready, + dependencies, + disagreements, + signer_identity_id: self.signer.identity_id().into(), + // Placeholder. A sequence is only reserved once the facts are + // known to have changed, below; same_correlation_facts compares + // against the prior's sequence regardless of what sits here. + publisher_sequence: prior.as_ref().map_or(0, |record| record.publisher_sequence), + observed_at_unix_millis: now, + signature_algorithm: "ed25519".into(), + signature: Vec::new(), + }; + if let Some(prior) = &prior + && same_correlation_facts(prior, &correlation) + { + return Ok(observed + .correlation + .context("decoded prior correlation disappeared")? + .canonical_bytes); + } + // Reserved only after the dedupe: a refresh that observes nothing + // new must not burn a sequence or change the store. Reserved from a + // mark that survives withdrawal, not derived from the stored + // correlation, so a withdrawn correlation does not take the count + // with it. See TOPOLOGY_PUBLISHER_WATERMARK_SCHEMA. + correlation.publisher_sequence = self.store.reserve_publisher_sequence( + &observed.query.incarnation.target, + prior.as_ref().map_or(0, |record| record.publisher_sequence), + )?; + correlation.signature = self + .signer + .sign_correlation(&correlation.unsigned_signature_payload()?)?; + let canonical = correlation.canonical_bytes()?; + correlation.validate_against_expected(&projection.expected, lease_sha256.as_deref())?; + let (signed, unsigned) = + OdinRuntimeTopologyCorrelationRecord::decode_canonical_signed_payload(&canonical)?; + verify_service_identity_signature_with_public_key::< + OdinTopologyIdentity, + OdinRuntimeTopologyCorrelationPurpose, + >( + self.signer.public_key(), + &unsigned, + &ServiceIdentitySignature { + identity_id: signed.signer_identity_id, + signature: signed.signature, + }, + )?; + if let Some(authority) = &authority { + authenticate_odin_runtime_topology_correlation( + &canonical, + authority, + lease_sha256.as_deref(), self.signer.public_key(), - &unsigned, - &ServiceIdentitySignature { - identity_id: signed.signer_identity_id, - signature: signed.signature, + OdinTopologyAuthenticationContext { + trusted_received_at_unix_millis: now, + maximum_age_millis: self.policy.correlation_maximum_age_millis, + maximum_future_skew_millis: self + .policy + .correlation_maximum_future_skew_millis, }, )?; - if let Some(authority) = &authority { - authenticate_odin_runtime_topology_correlation( - &canonical, - authority, - lease_sha256.as_deref(), - self.signer.public_key(), - OdinTopologyAuthenticationContext { - trusted_received_at_unix_millis: now, - maximum_age_millis: self.policy.correlation_maximum_age_millis, - maximum_future_skew_millis: self - .policy - .correlation_maximum_future_skew_millis, - }, - )?; - } - let stored_correlation = StoredCorrelation { - canonical_bytes: canonical.clone(), - stored_at: rfc3339_millis(now)?, - }; - if self.store.compare_exchange( - &observed, - replacement_presence.as_ref(), - &stored_correlation, - )? { - return Ok(canonical); - } } - bail!("Odin topology store changed repeatedly during correlation") + let stored_correlation = StoredCorrelation { + canonical_bytes: canonical.clone(), + stored_at: rfc3339_millis(now)?, + }; + self.store + .commit(&observed.query, replacement_presence.as_ref(), &stored_correlation)?; + Ok(canonical) } fn select_presence( @@ -1619,22 +1701,6 @@ fn correlation_envelope(target: &str, correlation: &StoredCorrelation) -> CultCa } } -fn insert_condition( - conditions: &mut BTreeMap<(String, String), CultCacheExpectedEnvelope>, - condition: CultCacheExpectedEnvelope, -) -> Result<()> { - let identity = (condition.r#type.clone(), condition.key.clone()); - if let Some(current) = conditions.get(&identity) { - ensure!( - current.current == condition.current, - "store read set is incoherent" - ); - } else { - conditions.insert(identity, condition); - } - Ok(()) -} - fn rfc3339_millis(value: u64) -> Result { let value: i64 = value .try_into() @@ -1653,7 +1719,7 @@ fn parse_rfc3339_millis(value: &str) -> Result { #[cfg(test)] mod tests { - use std::path::PathBuf; + use std::path::{Path, PathBuf}; use cultnet_rs::{ GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA, GameCultProviderHealthIdentity, @@ -1699,7 +1765,9 @@ mod tests { ); } let projections = - CultCacheIdunnProjectionSource::new(&projection_path).projections(&target)?; + CultCacheIdunnProjectionSource::new(&projection_path) + .snapshot()? + .projections(&target)?; if projections.is_empty() { println!("== no incarnation-keyed projection for {target}"); return Ok(()); @@ -1800,10 +1868,12 @@ mod tests { candidate.activation = None; candidate.current_lease = None; world.publish_projection(&candidate)?; + let engine = world.engine(NOW); let candidate_incarnation = IncarnationRef::of(&candidate.expected)?; assert_ne!(incumbent.incarnation()?, candidate_incarnation); assert_eq!( CultCacheIdunnProjectionSource::new(&world.projection_path) + .snapshot()? .projections("odin")? .len(), 2 @@ -1856,7 +1926,7 @@ mod tests { }) .collect(); ensure!(store.compare_exchange_snapshot(¤t, &legacy)?); - let source = CultCacheIdunnProjectionSource::new(&world.projection_path); + let source = CultCacheIdunnProjectionSource::new(&world.projection_path).snapshot()?; assert!(source.projections("ghostlight")?.is_empty()); assert!(source.projection(&service.incarnation()?)?.is_none()); Ok(()) @@ -1879,7 +1949,7 @@ mod tests { schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), }); ensure!(store.compare_exchange_snapshot(¤t, &with_ghost)?); - let source = CultCacheIdunnProjectionSource::new(&world.projection_path); + let source = CultCacheIdunnProjectionSource::new(&world.projection_path).snapshot()?; assert!(source.projection(&service.incarnation()?)?.is_some()); assert!( source @@ -1891,16 +1961,19 @@ mod tests { } /// A store the target-keyed Odin wrote holds correlations under bare - /// target keys. They are retired; presence history under any key stays. + /// target keys. They are retired from the loaded set; presence history + /// under any key stays. #[test] fn legacy_target_keyed_correlations_are_retired_and_presence_is_kept() -> Result<()> { let world = TestWorld::new()?; let service = world.service("ghostlight", Vec::new(), false)?; let presence = service.signed_presence(1, |_| {})?; - let engine = world.engine(NOW); - let correlation = engine.admit_presence("ghostlight", &presence, NOW)?; - let store = SingleFileMessagePackBackingStore::new(&world.topology_path); - let current = store.pull_all_read_only_snapshot()?; + let correlation = world + .engine(NOW) + .admit_presence("ghostlight", &presence, NOW)?; + assert!(world.store.flush()?); + let file = SingleFileMessagePackBackingStore::new(&world.topology_path); + let current = file.pull_all_read_only_snapshot()?; let mut with_legacy = current.clone(); with_legacy.push(CultCacheEnvelope { key: "ghostlight".into(), @@ -1916,23 +1989,21 @@ mod tests { stored_at: rfc3339_millis(NOW - 1_000)?, schema_id: Some(GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()), }); - ensure!(store.compare_exchange_snapshot(¤t, &with_legacy)?); + ensure!(file.compare_exchange_snapshot(¤t, &with_legacy)?); - let odin_store = CultCacheOdinTopologyStore::new(&world.topology_path); - assert_eq!(odin_store.retire_legacy_correlations()?, 1); - assert_eq!(odin_store.retire_legacy_correlations()?, 0); - let after = store.pull_all_read_only_snapshot()?; - assert!(after.iter().all(|entry| { - entry.r#type != OdinRuntimeTopologyCorrelationRecord::TYPE - || IncarnationRef::parse_key(&entry.key).is_some() + let loaded = MemoryOdinTopologyStore::load(&world.topology_path)?; + assert_eq!(loaded.retire_legacy_correlations(), 1); + assert_eq!(loaded.retire_legacy_correlations(), 0); + let after = loaded.records(); + assert!(after.keys().all(|(record_type, key)| { + record_type != OdinRuntimeTopologyCorrelationRecord::TYPE + || IncarnationRef::parse_key(key).is_some() })); + assert!(after.keys().any(|(_, key)| key == "ghostlight/legacy-signer")); + drop(after); assert!( - after - .iter() - .any(|entry| entry.key == "ghostlight/legacy-signer") - ); - assert!( - engine + world + .engine_on(&loaded, NOW) .current_signed_correlation(&service.incarnation()?)? .is_some() ); @@ -1942,7 +2013,7 @@ mod tests { #[test] fn withdrawing_a_correlation_does_not_reset_its_publisher_sequence() -> Result<()> { let temp = TempDir::new()?; - let store = CultCacheOdinTopologyStore::new(temp.path().join("topology.cc")); + let store = MemoryOdinTopologyStore::load(temp.path().join("topology.cc"))?; assert_eq!(store.reserve_publisher_sequence("heimdall", 0)?, 1); assert_eq!(store.reserve_publisher_sequence("heimdall", 1)?, 2); @@ -1965,6 +2036,275 @@ mod tests { Ok(()) } + /// The store file is read once, when the store is loaded. Replaced behind + /// the loaded store with bytes that do not decode, it changes nothing the + /// store answers: admission, refresh and withdrawal all run from memory. + #[test] + fn a_loaded_store_never_reads_its_file_again() -> Result<()> { + let world = TestWorld::new()?; + let service = world.service("ghostlight", Vec::new(), false)?; + let engine = world.engine(NOW); + let first = engine.admit_presence("ghostlight", &service.signed_presence(1, |_| {})?, NOW)?; + assert!(world.store.flush()?); + + std::fs::write(&world.topology_path, b"not a cultcache store")?; + assert_eq!( + engine.current_signed_correlation(&service.incarnation()?)?, + Some(first) + ); + let second = decode_signed(&engine.admit_presence( + "ghostlight", + &service.signed_presence(2, |_| {})?, + NOW + 1, + )?)?; + assert_eq!(second.publisher_sequence, 2); + assert_eq!( + std::fs::read(&world.topology_path)?, + b"not a cultcache store", + "nothing was written without a flush" + ); + Ok(()) + } + + /// A write is made only when something changed, and only over the file the + /// store last loaded or wrote. A file another process wrote is refused + /// and left as that process wrote it. + #[test] + fn a_flush_writes_only_changes_and_never_over_a_foreign_file() -> Result<()> { + let temp = TempDir::new()?; + let path = temp.path().join("topology.cc"); + let store = MemoryOdinTopologyStore::load(&path)?; + assert!(!store.flush()?); + assert!(!path.exists(), "an unchanged empty store writes nothing"); + + let document = CultCacheEnvelope { + key: "doc-1".into(), + r#type: "ghostlight.doc".into(), + payload: vec![1], + stored_at: rfc3339_millis(NOW)?, + schema_id: Some("ghostlight.doc.v1".into()), + }; + store.put(document.clone()); + assert!(store.flush()?); + store.put(document.clone()); + assert!(!store.flush()?, "storing the record already held is no change"); + + let file = SingleFileMessagePackBackingStore::new(&path); + let mut foreign = document.clone(); + foreign.key = "foreign".into(); + ensure!(file.compare_exchange_snapshot(&[document.clone()], &[foreign.clone()])?); + let mut changed = document; + changed.payload = vec![2]; + store.put(changed); + let error = store.flush().unwrap_err(); + assert!(error.downcast_ref::().is_some(), "{error:#}"); + assert_eq!(file.pull_all_read_only_snapshot()?, vec![foreign]); + Ok(()) + } + + fn peer_document(key: &str) -> Result { + Ok(CultCacheEnvelope { + key: key.into(), + r#type: "ghostlight.doc".into(), + payload: vec![1], + stored_at: rfc3339_millis(NOW)?, + schema_id: Some("ghostlight.doc.v1".into()), + }) + } + + /// A write that fails leaves its change to be written: the next write + /// lands it with nothing changed in between. + #[test] + fn a_failed_write_is_made_by_the_next_without_another_change() -> Result<()> { + let temp = TempDir::new()?; + let path = temp.path().join("topology.cc"); + let store = MemoryOdinTopologyStore::load(&path)?; + store.put(peer_document("doc-1")?); + std::fs::write(&path, b"not a cultcache store")?; + assert!(store.flush().is_err(), "the injected fault fails the write"); + + // The fault clears: the file holds what Odin loaded, which was nothing. + std::fs::remove_file(&path)?; + assert!(store.flush()?, "the failed change is written"); + assert_eq!( + SingleFileMessagePackBackingStore::new(&path).pull_all_read_only_snapshot()?, + vec![peer_document("doc-1")?] + ); + assert!(!store.flush()?, "and nothing is left to write"); + Ok(()) + } + + /// A write can fail after it replaced the file (the directory sync), and + /// the file then holds Odin's own records: the next write is made over + /// them. Once a write lands, those records are no longer Odin's to expect. + #[test] + fn a_write_that_failed_after_replacing_the_file_is_not_another_writer() -> Result<()> { + let temp = TempDir::new()?; + let path = temp.path().join("topology.cc"); + let file = SingleFileMessagePackBackingStore::new(&path); + let store = MemoryOdinTopologyStore::load(&path)?; + store.put(peer_document("doc-1")?); + std::fs::write(&path, b"not a cultcache store")?; + assert!(store.flush().is_err()); + // The state a failed directory sync leaves: the replace landed. + std::fs::remove_file(&path)?; + ensure!(file.compare_exchange_snapshot(&[], &[peer_document("doc-1")?])?); + + store.put(peer_document("doc-2")?); + assert!(store.flush()?, "Odin's own records are written over"); + let written = file.pull_all_read_only_snapshot()?; + assert_eq!(written, vec![peer_document("doc-1")?, peer_document("doc-2")?]); + + // Another writer puts back what the failed write had tried to write. + ensure!(file.compare_exchange_snapshot(&written, &[peer_document("doc-1")?])?); + store.put(peer_document("doc-3")?); + let error = store.flush().unwrap_err(); + assert!(error.downcast_ref::().is_some(), "{error:#}"); + assert_eq!(file.pull_all_read_only_snapshot()?, vec![peer_document("doc-1")?]); + Ok(()) + } + + /// Two failed writes in a row: the first after it replaced the file, the + /// second before (the lock cannot be taken), so the file still holds the + /// first one's records. Both are Odin's own, and the next write is made + /// over the first's. + #[test] + fn a_write_that_failed_before_replacing_keeps_every_earlier_attempt_odins() -> Result<()> { + let temp = TempDir::new()?; + let path = temp.path().join("topology.cc"); + let file = SingleFileMessagePackBackingStore::new(&path); + let store = MemoryOdinTopologyStore::load(&path)?; + store.put(peer_document("doc-1")?); + std::fs::write(&path, b"not a cultcache store")?; + assert!(store.flush().is_err()); + // The state a failed directory sync leaves: the replace landed. + std::fs::remove_file(&path)?; + ensure!(file.compare_exchange_snapshot(&[], &[peer_document("doc-1")?])?); + + store.put(peer_document("doc-2")?); + let lock = temp.path().join("topology.cc.lock"); + std::fs::remove_file(&lock)?; + std::fs::create_dir(&lock)?; + assert!(store.flush().is_err(), "the lock cannot be taken"); + std::fs::remove_dir(&lock)?; + assert_eq!(file.pull_all_read_only_snapshot()?, vec![peer_document("doc-1")?]); + + assert!(store.flush()?, "the first attempt is Odin's own, not a foreign write"); + assert_eq!( + file.pull_all_read_only_snapshot()?, + vec![peer_document("doc-1")?, peer_document("doc-2")?] + ); + Ok(()) + } + + /// Two writes fail before replacing the file, and the file then comes to + /// hold the later attempt's records (the state a failed directory sync of + /// that attempt would leave). Any failed attempt may be what the file + /// holds, not only the first: the next write is made over it. + #[test] + fn the_file_may_hold_a_later_failed_attempt_not_only_the_first() -> Result<()> { + let temp = TempDir::new()?; + let path = temp.path().join("topology.cc"); + let file = SingleFileMessagePackBackingStore::new(&path); + let store = MemoryOdinTopologyStore::load(&path)?; + std::fs::write(&path, b"not a cultcache store")?; + store.put(peer_document("doc-1")?); + assert!(store.flush().is_err()); + store.put(peer_document("doc-2")?); + assert!(store.flush().is_err()); + std::fs::remove_file(&path)?; + ensure!(file.compare_exchange_snapshot( + &[], + &[peer_document("doc-1")?, peer_document("doc-2")?] + )?); + + store.put(peer_document("doc-3")?); + assert!(store.flush()?, "the later attempt is Odin's own"); + assert_eq!( + file.pull_all_read_only_snapshot()?, + vec![peer_document("doc-1")?, peer_document("doc-2")?, peer_document("doc-3")?] + ); + Ok(()) + } + + /// Removing a record is a change of its own: the next flush writes the + /// file without it, with nothing else changed. + #[test] + fn a_removed_record_is_written_out() -> Result<()> { + let temp = TempDir::new()?; + let path = temp.path().join("topology.cc"); + let file = SingleFileMessagePackBackingStore::new(&path); + let store = MemoryOdinTopologyStore::load(&path)?; + store.put(peer_document("doc-1")?); + store.put(peer_document("doc-2")?); + assert!(store.flush()?); + + store.remove("ghostlight.doc", "doc-1"); + assert!(store.flush()?, "the removal is written"); + assert_eq!(file.pull_all_read_only_snapshot()?, vec![peer_document("doc-2")?]); + store.remove("ghostlight.doc", "doc-1"); + assert!(!store.flush()?, "removing what is not held changes nothing"); + Ok(()) + } + + fn file_sequences(path: &Path) -> Result<(u64, Vec)> { + let entries = SingleFileMessagePackBackingStore::new(path).pull_all_read_only_snapshot()?; + let mark = unique_envelope(&entries, TOPOLOGY_PUBLISHER_WATERMARK_TYPE, "ghostlight")? + .map(|envelope| decode_publisher_watermark("ghostlight", &envelope.payload)) + .transpose()? + .unwrap_or(0); + let correlations = entries + .iter() + .filter(|entry| entry.r#type == OdinRuntimeTopologyCorrelationRecord::TYPE) + .map(|entry| Ok(decode_correlation(&entry.payload)?.publisher_sequence)) + .collect::>>()?; + Ok((mark, correlations)) + } + + /// A correlation sequence is published when it is written. The mark is + /// written with the correlations that use it, so the file never holds a + /// correlation ahead of its mark, and after a crash (the working set lost + /// without a final write) and a reload, the next correlation is above + /// every sequence the file holds -- even with the correlation withdrawn. + #[test] + fn a_crash_never_republishes_a_written_sequence() -> Result<()> { + let world = TestWorld::new()?; + let service = world.service("ghostlight", Vec::new(), false)?; + for sequence in 1..=2 { + world.engine(NOW).admit_presence( + "ghostlight", + &service.signed_presence(sequence, |_| {})?, + NOW + sequence, + )?; + } + world.withdraw_projection(&service.incarnation()?)?; + world.engine(NOW + 3).refresh(&service.incarnation()?)?; + assert!(world.store.flush()?); + let (mark, correlations) = file_sequences(&world.topology_path)?; + assert_eq!((mark, correlations.len()), (2, 0), "withdrawn, the mark stays"); + + // Changes made and never written, then the crash. + world.publish_projection(&service.projection)?; + world.engine(NOW + 4).admit_presence( + "ghostlight", + &service.signed_presence(3, |_| {})?, + NOW + 4, + )?; + let restarted = MemoryOdinTopologyStore::load(&world.topology_path)?; + assert_eq!(file_sequences(&world.topology_path)?.0, mark); + + let next = decode_signed(&world.engine_on(&restarted, NOW + 5).admit_presence( + "ghostlight", + &service.signed_presence(4, |_| {})?, + NOW + 5, + )?)?; + assert!(next.publisher_sequence > mark, "{} repeats", next.publisher_sequence); + assert!(restarted.flush()?); + let (mark, correlations) = file_sequences(&world.topology_path)?; + assert!(correlations.iter().all(|sequence| *sequence <= mark)); + Ok(()) + } + #[derive(Clone, Copy)] struct FixedClock(u64); @@ -2050,6 +2390,7 @@ mod tests { _temp: TempDir, projection_path: PathBuf, topology_path: PathBuf, + store: MemoryOdinTopologyStore, identity_root: PathBuf, idunn_signer: ServiceIdentitySigner, idunn_anchor: ServiceIdentityTrustAnchor, @@ -2066,9 +2407,11 @@ mod tests { let idunn_anchor = idunn_signer.trust_anchor()?; let odin_signer = enroll_service_identity_at::(&identity_root.join("odin.cc"))?; + let topology_path = temp.path().join("odin-topology.cc"); Ok(Self { projection_path: temp.path().join("idunn-projection.cc"), - topology_path: temp.path().join("odin-topology.cc"), + store: MemoryOdinTopologyStore::load(&topology_path)?, + topology_path, identity_root, idunn_signer, idunn_anchor, @@ -2259,18 +2602,18 @@ mod tests { Ok(()) } - fn engine( - &self, - now: u64, - ) -> OdinTopologyAuthority< - CultCacheIdunnProjectionSource, - CultCacheOdinTopologyStore, - &ServiceIdentitySigner, - FixedClock, - > { + fn engine(&self, now: u64) -> TestEngine<'_> { + self.engine_on(&self.store, now) + } + + /// An engine over another store, as a restarted Odin loads one. It reads + /// the projection once, when it is made, as each serving operation does. + fn engine_on<'a>(&'a self, store: &'a MemoryOdinTopologyStore, now: u64) -> TestEngine<'a> { OdinTopologyAuthority::new( - CultCacheIdunnProjectionSource::new(&self.projection_path), - CultCacheOdinTopologyStore::new(&self.topology_path), + CultCacheIdunnProjectionSource::new(&self.projection_path) + .snapshot() + .expect("the test projection reads"), + store, &self.odin_signer, FixedClock(now), self.idunn_anchor.clone(), @@ -2314,6 +2657,13 @@ mod tests { } } + type TestEngine<'a> = OdinTopologyAuthority< + IdunnProjectionSnapshot, + &'a MemoryOdinTopologyStore, + &'a ServiceIdentitySigner, + FixedClock, + >; + fn digest(byte: char) -> String { format!("sha256-{}", byte.to_string().repeat(64)) } @@ -2455,19 +2805,9 @@ mod tests { let reused = service.signed_presence(2, |record| record.detail = "different".into())?; assert!(engine.admit_presence("ghostlight", &reused, NOW).is_err()); - let entries = SingleFileMessagePackBackingStore::new(&world.topology_path) - .pull_all_read_only_snapshot()?; - let stored = unique_envelope( - &entries, - GameCultRuntimePresenceHealthRecord::TYPE, - &presence_store_key( - &service.incarnation()?, - &service.projection.expected.expected_signer_identity_id, - ), - )? - .unwrap(); - assert_eq!(stored.payload, sequence_two); - assert_eq!(parse_rfc3339_millis(&stored.stored_at)?, NOW); + let (payload, stored_at) = stored_presence(&world, &service)?; + assert_eq!(payload, sequence_two); + assert_eq!(parse_rfc3339_millis(&stored_at)?, NOW); Ok(()) } @@ -2475,20 +2815,21 @@ mod tests { /// received at `NOW`, beyond the skew window, as on yggdrasil 2026-09-29. const RESTART: u64 = NOW + 10_000; + /// The service's presence as the store holds it: payload and receipt time. + fn stored_presence(world: &TestWorld, service: &TestService) -> Result<(Vec, String)> { + let key = presence_store_key( + &service.incarnation()?, + &service.projection.expected.expected_signer_identity_id, + ); + let records = world.store.records(); + let stored = records + .get(&(GameCultRuntimePresenceHealthRecord::TYPE.to_owned(), key)) + .context("no stored presence")?; + Ok((stored.payload.clone(), stored.stored_at.clone())) + } + fn stored_presence_payload(world: &TestWorld, service: &TestService) -> Result> { - let entries = SingleFileMessagePackBackingStore::new(&world.topology_path) - .pull_all_read_only_snapshot()?; - Ok(unique_envelope( - &entries, - GameCultRuntimePresenceHealthRecord::TYPE, - &presence_store_key( - &service.incarnation()?, - &service.projection.expected.expected_signer_identity_id, - ), - )? - .context("no stored presence")? - .payload - .clone()) + Ok(stored_presence(world, service)?.0) } #[test] @@ -2800,9 +3141,11 @@ mod tests { let first = world .engine(NOW) .admit_presence("ghostlight", &presence, NOW)?; + assert!(world.store.flush()?); + let restarted = MemoryOdinTopologyStore::load(&world.topology_path)?; let after_restart = world - .engine(NOW + 100) + .engine_on(&restarted, NOW + 100) .refresh(&service.incarnation()?)? .unwrap(); assert_eq!(after_restart, first); @@ -2825,18 +3168,18 @@ mod tests { ); std::fs::remove_file(&world.projection_path)?; + let engine = world.engine(NOW); assert!( engine .current_signed_correlation(&service.incarnation()?)? .is_none() ); - let entries = SingleFileMessagePackBackingStore::new(&world.topology_path) - .pull_all_read_only_snapshot()?; - assert!(entries.iter().any(|entry| { + let records = world.store.records(); + assert!(records.values().any(|entry| { entry.r#type == GameCultRuntimePresenceHealthRecord::TYPE && entry.payload == presence })); - assert!(!entries.iter().any(|entry| { + assert!(!records.values().any(|entry| { entry.r#type == OdinRuntimeTopologyCorrelationRecord::TYPE && entry.key.starts_with("ghostlight@") })); @@ -2880,6 +3223,7 @@ mod tests { assert!(ready.dependencies[0].provider_evidence_sha256.is_some()); world.replace_provider_generation(&mut provider)?; + let engine = world.engine(NOW); let no_longer_exact = decode_signed(&engine.refresh(&consumer.incarnation()?)?.unwrap())?; assert!(no_longer_exact.present); assert!(!no_longer_exact.ready); diff --git a/crates/odin-daemon/src/main.rs b/crates/odin-daemon/src/main.rs index d7d5c39..509e499 100644 --- a/crates/odin-daemon/src/main.rs +++ b/crates/odin-daemon/src/main.rs @@ -12,31 +12,31 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result, bail, ensure}; use chrono::{DateTime, Utc}; -use cultcache_rs::{ - CultCacheEnvelope, CultCacheExpectedEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, -}; +use cultcache_rs::{CultCacheEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore}; use cultmesh_rs::{ CultMeshRudpDocumentServer, CultMeshRudpDocumentServerOptions, CultMeshRudpPollOutcome, CultMeshRudpRawDocumentReceipt, CultMeshRudpRawDocumentSink, CultMeshRudpSnapshotQuery, CultMeshRudpSnapshotSource, CultMeshSystemClock, }; use cultnet_rs::{ - CultNetRawDocumentRecord, CultNetRawPayloadEncoding, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, + CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, CultNetWireContract, + 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, + IdunnRuntimeActivationSigner, IdunnServiceIdentity, OdinTopologyIdentity, + ServiceIdentityProfile, ServiceIdentitySigner, ServiceIdentityTrustAnchor, derive_service_identity_id, - open_service_identity_credential_reader, verify_runtime_authority, + encode_cultnet_message_to_vec, open_service_identity_credential_reader, + verify_runtime_authority, }; use fs2::FileExt; use odin_daemon::{ - AuthenticationPolicy, CultCacheIdunnProjectionSource, CultCacheOdinTopologyStore, - IdunnProjectionSource, IncarnationRef, OdinTopologyAuthority, PresenceAuthorityRefused, - ProjectionUnreadable, SystemClock, + AuthenticationPolicy, CultCacheIdunnProjectionSource, ForeignStoreWrite, + IdunnProjectionSnapshot, IdunnProjectionSource, IncarnationRef, MemoryOdinTopologyStore, + OdinTopologyAuthority, PresenceAuthorityRefused, ProjectionUnreadable, SystemClock, }; const TARGET: &str = "odin"; @@ -70,17 +70,15 @@ 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); +// Odin's own changes (its heartbeat, correlations, the watermark) are written +// at most once per interval, and not at all when nothing changed. A crash loses +// at most this much of them; a correlation sequence that was never written was +// never published. A peer's put is written before Odin accepts it +// (`accept_raw_document`). Operator ruling Q3 B, docs/write-pattern-cut.md. +const FLUSH_INTERVAL: Duration = Duration::from_secs(1); const IDLE_POLL_INTERVAL: Duration = Duration::from_millis(2); const MAX_RECENT_WARMING_PROOFS: usize = 64; const WARMING_PROOF_LIFETIME_MILLIS: u64 = 60_000; -const CAS_ATTEMPTS: usize = 8; - -type TopologyAuthority = OdinTopologyAuthority< - CultCacheIdunnProjectionSource, - CultCacheOdinTopologyStore, - ServiceIdentitySigner, - SystemClock, ->; /// Relative path of Odin's durable store inside the state root. The recipe's /// `topology` state slot in `deployment/idunn/recipe.toml` is the authority for @@ -110,16 +108,53 @@ struct ProcessWriteLeaseGuard { sha256: String, } +/// Odin's topology authority once its write lease is held: the store it loaded +/// and the credentials that sign for it. An `OdinTopologyAuthority` is built +/// over it for each operation, with the projection read for that operation. +struct Topology { + store: MemoryOdinTopologyStore, + signer: ServiceIdentitySigner, + idunn_anchor: ServiceIdentityTrustAnchor, +} + +impl Topology { + fn authority( + &self, + projections: P, + ) -> OdinTopologyAuthority< + P, + &MemoryOdinTopologyStore, + &ServiceIdentitySigner, + SystemClock, + > { + OdinTopologyAuthority::new( + projections, + &self.store, + &self.signer, + SystemClock, + self.idunn_anchor.clone(), + AuthenticationPolicy::default(), + ) + } +} + struct RuntimeState { options: Options, candidate: SocketAddr, authority_material: RuntimeAuthority, idunn_anchor: Option, topology_signer: Option>, - topology: Option, + topology: Option, write_lease: Option, write_lease_path: PathBuf, + /// The largest snapshot response the server sends: its own + /// `max_snapshot_response_bytes`, set by `serve_with` from the options it + /// builds the server with. `None` until then, and Odin cannot activate + /// without it. + max_snapshot_response_bytes: Option, recent_warming_proofs: VecDeque<(String, u64)>, + /// Counts from this launch. Every launch is a fresh activation, and + /// presences are ordered only within one activation. publisher_sequence: u64, log_gate: RefCell, } @@ -153,7 +188,8 @@ impl RuntimeState { require_expected_contract(&authority_material.expected, candidate)?; let idunn_anchor = read_trust_anchor::(&options.idunn_anchor)?; - let projection_source = CultCacheIdunnProjectionSource::new(&options.idunn_projection); + let projection_source = + CultCacheIdunnProjectionSource::new(&options.idunn_projection).snapshot()?; // This process is one exact incarnation, named by the Expected digest // in its immutable runtime bundle. The projection may carry other // incarnations of `odin` at the same time -- the one being replaced, @@ -198,14 +234,6 @@ impl RuntimeState { ) })?, )?; - let store_sequence = prior_self_publisher_sequence( - &options.store, - authority_material - .provider_signer - .entry() - .identity_id - .as_str(), - )?; Ok(Self { options, @@ -216,8 +244,9 @@ impl RuntimeState { topology: None, write_lease: None, write_lease_path: PathBuf::from(required_environment(PROCESS_WRITE_LEASE_ENVIRONMENT)?), + max_snapshot_response_bytes: None, recent_warming_proofs: VecDeque::new(), - publisher_sequence: store_sequence, + publisher_sequence: 0, log_gate: RefCell::default(), }) } @@ -230,6 +259,9 @@ impl RuntimeState { if self.activated() { return Ok(true); } + let limit = self + .max_snapshot_response_bytes + .context("Odin cannot activate before its server is built")?; let Some(lease) = acquire_process_write_lease( &self.write_lease_path, &self.authority_material, @@ -239,6 +271,7 @@ impl RuntimeState { return Ok(false); }; let Some(projected) = CultCacheIdunnProjectionSource::new(&self.options.idunn_projection) + .snapshot()? .projection(&self_incarnation(&self.authority_material))? else { return Ok(false); @@ -246,6 +279,18 @@ impl RuntimeState { if projected.current_lease.as_ref() != Some(&lease.record) { return Ok(false); } + // The one read of the store file while this process lives: from here + // on the working set is in memory, and the file is only written. + let store = MemoryOdinTopologyStore::load(&self.options.store)?; + // Correlations the previous, target-keyed Odin left in this store are + // not this contract's and would otherwise be served to the Verse as + // current. + store.retire_legacy_correlations(); + for (record_type, key, size) in drop_unservable_documents(&store, limit) { + eprintln!( + "Odin dropped {record_type} {key} from its store: its snapshot response is {size} bytes, over the {limit} the server sends" + ); + } let signer = self .topology_signer .take() @@ -255,17 +300,55 @@ impl RuntimeState { .take() .context("Idunn trust anchor was already consumed")?; self.write_lease = Some(lease); - self.topology = Some(OdinTopologyAuthority::new( - CultCacheIdunnProjectionSource::new(&self.options.idunn_projection), - CultCacheOdinTopologyStore::new(&self.options.store), + self.topology = Some(Topology { + store, signer, - SystemClock, idunn_anchor, - AuthenticationPolicy::default(), - )); + }); Ok(true) } + fn topology(&self) -> Result<&Topology> { + self.topology + .as_ref() + .context("Odin topology authority is absent") + } + + /// Write Odin's store file from the working set, if it changed; returns + /// whether it wrote. This is the only write of the file, made on the + /// interval, for a peer's put, and on the way out. The lease is + /// checked immediately before it, so nothing is written once the lease is + /// lost, and a file another process wrote is refused (`ForeignStoreWrite`) + /// rather than written over. Both end Odin (see `survive`). + fn flush(&self) -> Result { + let store = &self.topology()?.store; + if !store.is_dirty() { + return Ok(false); + } + self.require_current_write_lease()?; + store.flush() + } + + /// A peer's put is written before Odin accepts it: this returns `Ok` only + /// once the store holding the put is on disk. The write is the one `flush`, + /// so it carries everything else that changed too. A write that fails + /// refuses the put, and the server answers a refusal with an application + /// rejection naming the put. + /// + /// What this promises is Odin's answer, not the transport's. The RUDP + /// acknowledgement of the datagram carrying a put is sent when it is + /// received, before ordered delivery hands the put here (a later put can + /// be acknowledged while an earlier one is still being resent), so it says + /// nothing about durability. A put accepted here is on disk; a put refused + /// here is rejected by name. + /// + /// A refusal runs one way only. A put refused because its write failed may + /// still become durable: it stays in the working set, so the catalog + /// serves it and the next write lands it (or the failed write had already + /// replaced the file). That is sound because a put is the latest value for + /// its type and key, not an event: a publisher that retries a refused put + /// stores the same value, and one that does not has lost nothing it was + /// promised. fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()> { ensure!( self.activated(), @@ -273,10 +356,12 @@ impl RuntimeState { ); validate_raw_document_shape(&receipt.document)?; if receipt.document.schema_id == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA { - return self - .admit_presence_document(&receipt.document, receipt.received_at_unix_millis); + self.admit_presence_document(&receipt.document, receipt.received_at_unix_millis)?; + } else { + persist_generic_document(&self.topology()?.store, &receipt.document)?; } - persist_generic_document(&self.options.store, &receipt.document) + self.flush()?; + Ok(()) } /// The one admission path for a runtime-presence document, whether a @@ -291,10 +376,13 @@ impl RuntimeState { 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)?; + let projections = + CultCacheIdunnProjectionSource::new(&self.options.idunn_projection).snapshot()?; + self.topology()?.authority(projections).admit_presence( + &presence.target, + &document.payload, + received_at_unix_millis, + )?; Ok(()) } @@ -436,7 +524,8 @@ impl RuntimeState { fn require_own_projection(&self) -> Result<()> { let path = &self.options.idunn_projection; match CultCacheIdunnProjectionSource::new(path) - .projection(&self_incarnation(&self.authority_material)) + .snapshot() + .and_then(|projections| projections.projection(&self_incarnation(&self.authority_material))) { Err(error) => Err(own_projection_failure(error)), // No file is an I/O condition (Idunn replaces it atomically, so it @@ -473,24 +562,22 @@ impl RuntimeState { Ok(()) } - fn refresh_all_correlations(&mut self) -> Result<()> { + /// Refresh every incarnation Idunn projects, and every incarnation Odin + /// holds a correlation for, projected or not (the latter are refreshed so + /// their correlations are withdrawn). The projection is read once for the + /// whole pass. + fn refresh_all_correlations(&self) -> Result<()> { self.require_current_write_lease()?; - let topology = self - .topology - .as_ref() - .context("Odin topology authority is absent")?; - let mut incarnations = incarnation_keys( - &self.options.idunn_projection, - IdunnExpectedIncarnationRecord::TYPE, - )?; - incarnations.extend(incarnation_keys( - &self.options.store, - OdinRuntimeTopologyCorrelationRecord::TYPE, - )?); + let topology = self.topology()?; + let projections = + CultCacheIdunnProjectionSource::new(&self.options.idunn_projection).snapshot()?; + let mut incarnations = projections.incarnations(); + incarnations.extend(topology.store.correlated_incarnations()); + let authority = topology.authority(projections); // 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 { - if let Err(error) = topology.refresh(&incarnation) { + if let Err(error) = authority.refresh(&incarnation) { self.log_repeating( &format!("refresh of incarnation {}", incarnation.key()), format!("failed; the rest are unaffected: {error:#}"), @@ -501,8 +588,14 @@ impl RuntimeState { } /// The catalog serves what is stored, one record at a time: a record that - /// does not decode, or whose projection cannot be read, is skipped and - /// logged, and is never a reason to refuse every other record. Odin's + /// does not decode, or whose incarnation's projection does not, is skipped + /// and logged, and is never a reason to refuse every other record. Idunn's + /// projection is read once per query; a projection that cannot be read at + /// all is every presence's failure and no one else's, so the query is + /// answered as if Idunn projected nothing: peer documents are served and + /// presences skipped. Two incarnations of one target holding presences + /// under the same identity are never a refusal either: one is served (see + /// `presence_precedence`) and the other skipped. Odin's /// correlations are not part of the catalog: Idunn reads them from the /// store file, and a target has one per incarnation, so keying them by /// target would collide for the whole deploy window. @@ -510,15 +603,18 @@ impl RuntimeState { &self, query: &CultMeshRudpSnapshotQuery, ) -> Result> { - let entries = if self.options.store.is_file() { - SingleFileMessagePackBackingStore::new(&self.options.store) - .pull_all_read_only_snapshot()? - } else { - Vec::new() - }; - let projections = CultCacheIdunnProjectionSource::new(&self.options.idunn_projection); + let records = self.topology()?.store.records(); + let projections = CultCacheIdunnProjectionSource::new(&self.options.idunn_projection) + .snapshot() + .unwrap_or_else(|error| { + self.log_repeating( + "catalog skips every presence", + format!("Idunn's projection cannot be read: {error:#}"), + ); + IdunnProjectionSnapshot::default() + }); let mut selected = BTreeMap::new(); - for envelope in &entries { + for envelope in records.values() { let document = match self.public_document(&projections, envelope) { Ok(Some(document)) => document, Ok(None) => continue, @@ -534,38 +630,56 @@ impl RuntimeState { continue; } let identity = (document.schema_id.clone(), document.record_key.clone()); - ensure!( - selected.insert(identity, document).is_none(), - "Odin catalog contains duplicate public document identities" - ); + let precedence = self.presence_precedence(envelope); + match selected.entry(identity) { + std::collections::btree_map::Entry::Vacant(slot) => { + slot.insert((document, precedence)); + } + std::collections::btree_map::Entry::Occupied(mut slot) => { + let (Some(this), Some(held)) = (precedence, slot.get().1) else { + bail!("Odin catalog contains duplicate public document identities"); + }; + self.log_repeating( + &format!("catalog serves one presence of {}", document.record_key), + "more than one incarnation holds an admitted presence; the Ready one, or else the latest, is served" + .into(), + ); + if this > held { + slot.insert((document, precedence)); + } + } + } } - Ok(selected.into_values().collect()) + Ok(selected.into_values().map(|(document, _)| document).collect()) + } + + /// The order in which presences of one target are preferred when more than + /// one incarnation of it holds an admitted presence, as every target does + /// while it is being replaced: the incarnation whose correlation is Ready + /// first, then the latest observation. `None` for anything but a presence, + /// which has one record per identity and never competes. + fn presence_precedence(&self, envelope: &CultCacheEnvelope) -> Option<(bool, u64)> { + if envelope.r#type != GameCultRuntimePresenceHealthRecord::TYPE { + return None; + } + let presence = decode_presence(&envelope.payload).ok()?; + let incarnation = IncarnationRef::new(presence.target, presence.expected_projection_sha256); + let ready = self.topology().ok()?.store.correlation_ready(&incarnation); + Some((ready, presence.observed_at_unix_millis)) } /// One stored record as the Verse sees it; `None` when it is not public. fn public_document( &self, - projections: &CultCacheIdunnProjectionSource, + projections: &IdunnProjectionSnapshot, envelope: &CultCacheEnvelope, ) -> Result> { + if envelope.r#type != GameCultRuntimePresenceHealthRecord::TYPE { + return Ok(peer_document(envelope)); + } let Some(schema_id) = envelope.schema_id.clone() else { return Ok(None); }; - if envelope.r#type != GameCultRuntimePresenceHealthRecord::TYPE { - return Ok(is_peer_document_type(&envelope.r#type).then(|| { - CultNetRawDocumentRecord { - schema_id, - record_key: envelope.key.clone(), - stored_at: envelope.stored_at.clone(), - payload_encoding: CultNetRawPayloadEncoding::Messagepack, - payload: envelope.payload.clone(), - source_runtime_id: None, - source_agent_id: None, - source_role: None, - tags: None, - } - })); - } let presence = decode_presence(&envelope.payload)?; let Some(projection) = projections.projection(&IncarnationRef::new( presence.target.clone(), @@ -694,13 +808,10 @@ fn main() -> Result<()> { ); let socket = UdpSocket::bind(candidate) .with_context(|| format!("binding Odin CultNet RUDP candidate {candidate}"))?; - let state = Rc::new(RefCell::new(RuntimeState::open(options, candidate)?)); - let mut server = CultMeshRudpDocumentServer::new( + let (state, mut server) = serve_with( socket, - SinkHandle(state.clone()), - SnapshotHandle(state.clone()), - CultMeshSystemClock::default(), CultMeshRudpDocumentServerOptions::default(), + RuntimeState::open(options, candidate)?, )?; // Idunn runs this process as PID 1 of its own PID namespace, and a @@ -730,20 +841,59 @@ fn main() -> Result<()> { } } - // Correlations the previous, target-keyed Odin left in this store are not - // 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()?; - // Once serving, only losing the write lease ends Odin (see `survive`). - let mut timers = ServingTimers::default(); - loop { + serve(&state, &mut server, &mut ServingTimers::default(), &stopping) +} + +/// Build the server over `socket` and hand it `state`, from one set of +/// options: the largest response the server sends is the limit `state` +/// sizes stored documents against when it activates, so the two cannot +/// disagree. +fn serve_with( + socket: UdpSocket, + options: CultMeshRudpDocumentServerOptions, + mut state: RuntimeState, +) -> Result<(Rc>, OdinServer)> { + state.max_snapshot_response_bytes = Some(options.max_snapshot_response_bytes); + let state = Rc::new(RefCell::new(state)); + let server = CultMeshRudpDocumentServer::new( + socket, + SinkHandle(state.clone()), + SnapshotHandle(state.clone()), + CultMeshSystemClock::default(), + options, + )?; + Ok((state, server)) +} + +/// Serve until stopped on request or ended by a condition `survive` returns +/// (or a dead socket). Every way out makes the last write (`stop`). +fn serve( + state: &Rc>, + server: &mut OdinServer, + timers: &mut ServingTimers, + stopping: &AtomicBool, +) -> Result<()> { + let ended = loop { if stopping.load(Ordering::Relaxed) { eprintln!("Odin stopping on request"); - return Ok(()); + break Ok(()); } - if !serving_pass(&state, &mut server, &mut timers)? { - thread::sleep(IDLE_POLL_INTERVAL); + match serving_pass(state, server, timers) { + Ok(true) => {} + Ok(false) => thread::sleep(IDLE_POLL_INTERVAL), + Err(error) => break Err(error), } + }; + stop(&state.borrow()); + ended +} + +/// The last write before Odin ends, under `flush`'s own checks: after +/// `WriteLeaseLost` or `ForeignStoreWrite` it writes nothing. A failure is +/// logged, not returned: Odin is ending either way. +fn stop(state: &RuntimeState) { + if let Err(error) = state.flush() { + eprintln!("Odin's final store write was not made: {error:#}"); } } @@ -753,14 +903,17 @@ fn main() -> Result<()> { struct ServingTimers { last_heartbeat: Option, last_projection_refresh: Option, + /// When the store file was last written, or a write last failed. + last_flush: 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. +/// One turn of the serving loop: serve a datagram, refresh the topology, +/// publish Odin's own presence, and write the store file, when each is due. +/// Returns whether the poll made progress. It fails only for the conditions +/// `survive` returns, or a dead socket. fn serving_pass( state: &Rc>, server: &mut OdinServer, @@ -777,6 +930,15 @@ fn serving_pass( survive(&state.borrow(), "self-presence publication", published)?; timers.last_heartbeat = Some(Instant::now()); } + if is_due(timers.last_flush, FLUSH_INTERVAL) { + let flushed = state.borrow().flush(); + // Only an attempt starts the interval: with nothing to write, the next + // change is written on the first pass after it. + if !matches!(flushed, Ok(false)) { + timers.last_flush = Some(Instant::now()); + } + survive(&state.borrow(), "store write", flushed)?; + } Ok(progressed) } @@ -800,8 +962,10 @@ 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 +/// allowed to end the daemon. Only `WriteLeaseLost` and `ForeignStoreWrite`, +/// wherever in the error chain they sit, are returned: each means another +/// writer holds Odin's state, and writing on would be a second writer. As a +/// named temporary rule, so is /// `PresenceAuthorityRefused`: a failure to establish Odin's OWN authority /// (`require_own_projection`), or a self-presence that authority will never /// admit (no verifiable authority, a signer that does not match the anchor, a @@ -817,6 +981,7 @@ fn survive(state: &RuntimeState, what: &str, result: Result) -> Result Ok(Some(value)), Err(error) if error.downcast_ref::().is_some() + || error.downcast_ref::().is_some() || error.downcast_ref::().is_some() => { Err(error) @@ -1177,39 +1342,6 @@ fn read_process_write_lease(path: &Path) -> Result Result { - if !path.is_file() { - return Ok(0); - } - Ok(SingleFileMessagePackBackingStore::new(path) - .pull_all_read_only_snapshot()? - .into_iter() - .filter(|entry| { - entry.r#type == GameCultRuntimePresenceHealthRecord::TYPE - && entry.key.starts_with(TARGET) - }) - .filter_map(|entry| match decode_presence(&entry.payload) { - Ok(presence) => Some(presence), - Err(error) => { - eprintln!( - "Odin startup skipped its stored presence {}: {error:#}", - entry.key - ); - None - } - }) - .filter(|presence| { - presence.target == TARGET && presence.signer_identity_id == signer_identity_id - }) - .map(|presence| presence.publisher_sequence) - .max() - .unwrap_or(0)) -} - /// The document namespaces providers publish to the rendezvous. A provider's /// document is stored and served back only under one of these; Odin's own /// records (correlations, the publisher watermark, every `odin.*` type) and @@ -1263,49 +1395,87 @@ fn is_peer_document_type(document_type: &str) -> bool { }) } -fn persist_generic_document(path: &Path, document: &CultNetRawDocumentRecord) -> Result<()> { +/// One stored peer document as the catalog serves it; `None` for any other +/// record. +fn peer_document(envelope: &CultCacheEnvelope) -> Option { + let schema_id = envelope.schema_id.clone()?; + is_peer_document_type(&envelope.r#type).then(|| CultNetRawDocumentRecord { + schema_id, + record_key: envelope.key.clone(), + stored_at: envelope.stored_at.clone(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: envelope.payload.clone(), + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + }) +} + +/// CultNet refuses an empty message id, so a response carries at least one +/// byte of it. +const SHORTEST_MESSAGE_ID: &str = "0"; + +/// Drop every stored peer document the server could never serve, and return +/// each one's type, key and response size. A document is unservable when a +/// snapshot response holding it alone, encoded as the server encodes one and +/// with the shortest message id CultNet admits, is larger than the largest +/// response the server sends (`limit`, the server's own +/// `max_snapshot_response_bytes`). Nobody can read such a document, and every +/// write of the store would carry it. Presences are Odin's own admitted +/// records, never this large, and are not checked; nor is anything the +/// catalog does not serve. +fn drop_unservable_documents( + store: &MemoryOdinTopologyStore, + limit: usize, +) -> Vec<(String, String, usize)> { + let unservable: Vec<(String, String, usize)> = store + .records() + .values() + .filter_map(|envelope| { + let response = CultNetMessage::SnapshotResponseRaw { + message_id: SHORTEST_MESSAGE_ID.into(), + documents: vec![peer_document(envelope)?], + }; + let size = + match encode_cultnet_message_to_vec(&response, CultNetWireContract::CultNetSchemaV0) { + Ok(payload) => payload.len(), + // Kept: the catalog meets the same failure when it serves + // the record, and logs it there. + Err(error) => { + eprintln!( + "Odin cannot size {} {} as a snapshot response; kept: {error:#}", + envelope.r#type, envelope.key + ); + return None; + } + }; + (size > limit).then(|| (envelope.r#type.clone(), envelope.key.clone(), size)) + }) + .collect(); + for (record_type, key, _) in &unservable { + store.remove(record_type, key); + } + unservable +} + +fn persist_generic_document( + store: &MemoryOdinTopologyStore, + document: &CultNetRawDocumentRecord, +) -> Result<()> { let document_type = document_type_for_schema(&document.schema_id)?; ensure!( is_peer_document_type(&document_type), "Odin accepts no {document_type} document from a provider" ); - let replacement = CultCacheEnvelope { + store.put(CultCacheEnvelope { key: document.record_key.clone(), - r#type: document_type.clone(), + r#type: document_type, payload: document.payload.clone(), stored_at: document.stored_at.clone(), schema_id: Some(document.schema_id.clone()), - }; - let store = SingleFileMessagePackBackingStore::new(path); - for _ in 0..CAS_ATTEMPTS { - let entries = if path.is_file() { - store.pull_all_read_only_snapshot()? - } else { - Vec::new() - }; - let mut matches = entries - .iter() - .filter(|entry| entry.r#type == document_type && entry.key == document.record_key); - let current = matches.next().cloned(); - ensure!( - matches.next().is_none(), - "generic CultCache identity is ambiguous" - ); - if current.as_ref() == Some(&replacement) { - return Ok(()); - } - if store.compare_exchange( - &[CultCacheExpectedEnvelope { - key: document.record_key.clone(), - r#type: document_type.clone(), - current, - }], - &[replacement.clone()], - )? { - return Ok(()); - } - } - bail!("Odin catalog changed repeatedly while persisting a provider document") + }); + Ok(()) } fn document_type_for_schema(schema_id: &str) -> Result { @@ -1349,28 +1519,6 @@ fn decode_presence(payload: &[u8]) -> Result Result> { - if !path.is_file() { - return Ok(BTreeSet::new()); - } - Ok(SingleFileMessagePackBackingStore::new(path) - .pull_all_read_only_snapshot()? - .into_iter() - .filter(|entry| entry.r#type == record_type) - .filter_map(|entry| IncarnationRef::parse_key(&entry.key)) - .collect()) -} - fn validate_snapshot_filters(query: &CultMeshRudpSnapshotQuery) -> Result<()> { ensure!( !query.message_id.is_empty() && query.message_id.trim() == query.message_id, @@ -1448,12 +1596,12 @@ 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, + CultNetRudpReliableSendStatus, CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, GAMECULT_RUNTIME_PRESENCE_HEALTH_SIGNING_PURPOSE, + GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA, GameCultServiceTrustAnchorRecord, + IdunnExpectedCapability, IdunnExpectedRoute, IdunnRuntimeActivationLaunch, + ODIN_RUNTIME_TOPOLOGY_CORRELATION_SCHEMA, OdinRuntimeTopologyCorrelationRecord, + enroll_service_identity_at, }; use super::*; @@ -1578,11 +1726,15 @@ mod tests { _temp: tempfile::TempDir, state: Rc>, server: OdinServer, + /// The server's own socket, shared: what is done to it is done to the server. + socket: UdpSocket, timers: ServingTimers, store: PathBuf, lease_path: PathBuf, provider_identity_id: String, lease: IdunnProcessWriteLeaseRecord, + /// Idunn's signer, for activations of further incarnations. + idunn_signer: ServiceIdentitySigner, } fn write_single_record(path: &Path, envelope: CultCacheEnvelope) -> Result<()> { @@ -1615,6 +1767,16 @@ mod tests { /// The same world with the production server bound under `options`. fn activated_odin_serving(options: CultMeshRudpDocumentServerOptions) -> Result { + activated_odin_with(options, |_| Ok(Vec::new())) + } + + /// The same world, with Odin's store file holding what `seed` returns when + /// Odin activates. `seed` runs before the lease is granted, with the + /// runtime that is about to activate, so it can sign Odin's own presences. + fn activated_odin_with( + options: CultMeshRudpDocumentServerOptions, + seed: impl FnOnce(&mut RuntimeState) -> Result>, + ) -> Result { let temp = tempfile::tempdir()?; let root = temp.path(); let socket = UdpSocket::bind("127.0.0.1:0")?; @@ -1708,10 +1870,18 @@ mod tests { topology: None, write_lease: None, write_lease_path: lease_path.clone(), + max_snapshot_response_bytes: None, recent_warming_proofs: VecDeque::new(), publisher_sequence: 0, log_gate: RefCell::default(), }; + let seeded = seed(&mut runtime)?; + if !seeded.is_empty() { + ensure!( + SingleFileMessagePackBackingStore::new(&store).compare_exchange_snapshot(&[], &seeded)?, + "test store seed failed" + ); + } // The lease names the Warming presence Odin signed for Idunn's probe. let warming = runtime.signed_presence_document("warming", "test warming")?; @@ -1767,27 +1937,31 @@ mod tests { "test projection CAS failed" ); + // Without its server Odin has no limit to size stored documents + // against, so it does not activate. ensure!( - runtime.try_activate()?, + runtime.try_activate().is_err(), + "Odin activated before its server was built" + ); + // Built as `main` builds it, so the activation below sizes stored + // documents against this server's own limit. + let server_socket = socket.try_clone()?; + let (state, server) = serve_with(socket, options, runtime)?; + ensure!( + state.borrow_mut().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(), - options, - )?; Ok(OdinWorld { _temp: temp, state, server, + socket: server_socket, timers: ServingTimers::default(), store, lease_path, provider_identity_id, lease, + idunn_signer, }) } @@ -1796,6 +1970,22 @@ mod tests { serving_pass(&self.state, &mut self.server, &mut self.timers) } + /// Serve as `main` does, with a stop already requested or not. + fn serve(&mut self, stop_requested: bool) -> Result<()> { + serve( + &self.state, + &mut self.server, + &mut self.timers, + &AtomicBool::new(stop_requested), + ) + } + + /// One heartbeat of Odin's own: a change to its bookkeeping, which is + /// written on the interval and not at once. + fn heartbeat(&self) -> Result<()> { + self.state.borrow_mut().publish_self_presence("test heartbeat") + } + /// 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. @@ -1861,18 +2051,43 @@ mod tests { }) } + /// Every record in Odin's working set. + fn working_set(&self) -> Vec { + let state = self.state.borrow(); + let records = state.topology.as_ref().unwrap().store.records(); + records.values().cloned().collect() + } + + /// Every record in Odin's store file. + fn file_records(&self) -> Result> { + SingleFileMessagePackBackingStore::new(&self.store).pull_all_read_only_snapshot() + } + fn stored_keys(&self, record_type: &str) -> Result> { - Ok(SingleFileMessagePackBackingStore::new(&self.store) - .pull_all_read_only_snapshot()? + Ok(self + .working_set() .into_iter() .filter(|entry| entry.r#type == record_type) .map(|entry| entry.key) .collect()) } - /// Odin's own publisher sequence as it is durably stored. + /// The publisher sequence of Odin's own admitted presence, 0 when none + /// is held. fn stored_sequence(&self) -> Result { - prior_self_publisher_sequence(&self.store, &self.provider_identity_id) + let key = format!( + "{}/{}", + self_incarnation(&self.state.borrow().authority_material).key(), + self.provider_identity_id + ); + self.working_set() + .into_iter() + .find(|entry| { + entry.r#type == GameCultRuntimePresenceHealthRecord::TYPE && entry.key == key + }) + .map_or(Ok(0), |entry| { + Ok(decode_presence(&entry.payload)?.publisher_sequence) + }) } /// Run a peer on its own thread while Odin polls its socket, and nothing @@ -1983,28 +2198,30 @@ mod tests { Ok(()) } - /// A publication or refresh that fails is logged and retried; it never - /// ends Odin, and the next attempt lands once the fault clears. + /// A store write 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<()> { + fn a_failed_store_write_does_not_end_odin() -> Result<()> { let mut odin = activated_odin()?; odin.pass()?; - assert!(odin.stored_sequence()? > 0, "the first pass publishes"); + assert_eq!(odin.file_records()?, odin.working_set(), "the first pass writes"); + let written = std::fs::read(&odin.store)?; 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" + provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1]).is_err(), + "a put whose write fails is refused" + ); + assert!( + odin.state.borrow().flush().is_err(), + "the injected fault must actually fail the write" ); odin.timers = ServingTimers::default(); odin.pass()?; let suppressed = |odin: &OdinWorld| odin.state.borrow().log_gate.borrow().suppressed_total; assert_eq!(suppressed(&odin), 0, "each distinct failure is logged"); - // The same faults on later passes are counted, not printed again. + // The same fault on later passes is counted, not printed again. for repeat in 1..=3 { odin.timers = ServingTimers::default(); odin.pass()?; @@ -2014,12 +2231,13 @@ mod tests { ); } - std::fs::remove_file(&odin.store)?; + std::fs::write(&odin.store, written)?; odin.timers = ServingTimers::default(); odin.pass()?; - assert!( - odin.stored_sequence()? > 0, - "the publication after the fault clears lands" + assert_eq!( + odin.file_records()?, + odin.working_set(), + "the write after the fault clears lands" ); Ok(()) } @@ -2235,24 +2453,25 @@ mod tests { /// A stored presence of Odin's own activation that no longer authenticates /// makes its publisher sequence unknowable; the claim is refused, closed. + /// The store is Odin's alone while it serves, so the bad record is in the + /// file when Odin activates. #[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() + let mut odin = activated_odin_with(CultMeshRudpDocumentServerOptions::default(), |runtime| { + let warming = runtime.signed_presence_document("warming", "earlier launch")?; + // Received two minutes after it was observed: outside the trusted window. + Ok(vec![CultCacheEnvelope { + key: format!( + "{}/{}", + self_incarnation(&runtime.authority_material).key(), + runtime.authority_material.provider_signer.entry().identity_id + ), + r#type: GameCultRuntimePresenceHealthRecord::TYPE.into(), + payload: warming.payload, + stored_at: rfc3339_millis(unix_millis()? + 120_000)?, + schema_id: Some(GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()), + }]) })?; - odin.timers = ServingTimers::default(); assert_refused(&odin.pass().unwrap_err()); Ok(()) } @@ -2292,6 +2511,127 @@ mod tests { Ok(incarnation) } + /// A candidate incarnation of Odin that Idunn has activated, with its + /// Warming presence admitted beside the incumbent's, as every target has + /// for the moment it is being replaced. The candidate's presence is + /// observed after every presence already held. + fn admit_candidate_presence(odin: &OdinWorld) -> Result { + let now = unix_millis()?; + let mut candidate = odin.state.borrow().authority_material.expected.clone(); + candidate.incarnation_id = "odin/generation-2".into(); + candidate.sealed_release_id = digest('9'); + candidate.validate()?; + let incarnation = IncarnationRef::of(&candidate)?; + let launch = + IdunnRuntimeActivationLaunch::issue(&candidate, digest('8'), now - 10, &odin.idunn_signer)?; + let activation = launch.activation().clone(); + let mut credential = Vec::new(); + launch.write_credential(&mut credential)?; + odin.append_projection(vec![ + CultCacheEnvelope { + key: incarnation.key(), + r#type: IdunnExpectedIncarnationRecord::TYPE.into(), + payload: candidate.canonical_bytes()?, + stored_at: rfc3339_millis(now - 30)?, + schema_id: Some(IDUNN_EXPECTED_INCARNATION_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()), + }, + ])?; + // Signed as the candidate process would sign it: its own Expected and + // activation, no lease yet. + thread::sleep(Duration::from_millis(5)); + let document = { + let mut state = odin.state.borrow_mut(); + let authority = &mut state.authority_material; + let expected_sha256 = candidate.canonical_sha256()?; + let activation_sha256 = activation.canonical_sha256()?; + let signer = IdunnRuntimeActivationSigner::from_credential_reader(credential.as_slice())?; + let incumbent = ( + std::mem::replace(&mut authority.expected, candidate), + std::mem::replace(&mut authority.expected_sha256, expected_sha256), + std::mem::replace(&mut authority.activation, activation), + std::mem::replace(&mut authority.activation_sha256, activation_sha256), + std::mem::replace(&mut authority.activation_signer, signer), + ); + let lease = state.write_lease.take(); + let signed = state.signed_presence_document("warming", "test candidate"); + state.write_lease = lease; + let authority = &mut state.authority_material; + ( + authority.expected, + authority.expected_sha256, + authority.activation, + authority.activation_sha256, + authority.activation_signer, + ) = incumbent; + signed? + }; + provider_put_document(odin, document)?; + Ok(incarnation) + } + + /// The presence of `target` the catalog serves, by the incarnation it names. + fn served_presence_incarnation(odin: &OdinWorld) -> Result { + let catalog = odin.catalog()?; + let presences: Vec<_> = catalog + .iter() + .filter(|document| document.schema_id == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA) + .collect(); + ensure!(presences.len() == 1, "{} presences served", presences.len()); + Ok(decode_presence(&presences[0].payload)?.expected_projection_sha256) + } + + /// While a target is being replaced, its incumbent and its candidate each + /// hold an admitted presence under the same catalog identity. The catalog + /// serves one: the incarnation whose correlation is Ready, even though the + /// other is newer. + #[test] + fn a_target_being_replaced_serves_the_presence_of_its_ready_incarnation() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + let incumbent = self_incarnation(&odin.state.borrow().authority_material); + let candidate = admit_candidate_presence(&odin)?; + { + let state = odin.state.borrow(); + let store = &state.topology.as_ref().unwrap().store; + assert!(store.correlation_ready(&incumbent)); + assert!(!store.correlation_ready(&candidate), "the candidate is still Warming"); + } + assert_eq!(served_presence_incarnation(&odin)?, incumbent.expected_sha256); + Ok(()) + } + + /// With no correlation to tell them apart, the later presence is served, + /// whichever incarnation holds it. + #[test] + fn a_target_being_replaced_serves_the_later_presence_when_neither_is_ready() -> Result<()> { + let odin = activated_odin()?; + let incumbent = self_incarnation(&odin.state.borrow().authority_material); + odin.heartbeat()?; + let candidate = admit_candidate_presence(&odin)?; + let forget_correlations = || { + let state = odin.state.borrow(); + let store = &state.topology.as_ref().unwrap().store; + for incarnation in [&incumbent, &candidate] { + store.remove(OdinRuntimeTopologyCorrelationRecord::TYPE, &incarnation.key()); + } + }; + forget_correlations(); + assert_eq!(served_presence_incarnation(&odin)?, candidate.expected_sha256); + + thread::sleep(Duration::from_millis(5)); + odin.heartbeat()?; + forget_correlations(); + assert_eq!(served_presence_incarnation(&odin)?, incumbent.expected_sha256); + Ok(()) + } + /// Idunn projects the incumbent and the candidate of a target for the whole /// deploy window, and Odin writes a correlation for each. The catalog does /// not serve correlations, so two of one target cannot collide in it, and @@ -2335,44 +2675,103 @@ mod tests { Ok(()) } - /// One record that does not decode is skipped where it is met: the catalog - /// still serves everything else, and startup still finds Odin's sequence. + /// One record that does not decode is skipped where it is met: Odin still + /// starts and serves, and the catalog serves everything else. The records + /// are in the file when Odin activates. #[test] fn records_that_do_not_decode_never_stop_the_catalog_or_startup() -> Result<()> { - let mut odin = activated_odin()?; - odin.pass()?; - let before = odin.stored_sequence()?; - assert!(before > 0); - let stamp = rfc3339_millis(unix_millis()?)?; - let own_signer = odin.provider_identity_id.clone(); - odin.tamper_store(|mut entries| { - for key in [ - "ghost".to_owned(), - format!("{TARGET}@{}/{own_signer}", digest('c')), - ] { - entries.push(CultCacheEnvelope { - key, + let mut odin = activated_odin_with(CultMeshRudpDocumentServerOptions::default(), |runtime| { + let own_signer = runtime + .authority_material + .provider_signer + .entry() + .identity_id + .clone(); + let stamp = rfc3339_millis(unix_millis()?)?; + let bad_presence = |key: String| CultCacheEnvelope { + key, + r#type: GameCultRuntimePresenceHealthRecord::TYPE.into(), + payload: vec![0xc1], + stored_at: stamp.clone(), + schema_id: Some(GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()), + }; + let earlier = runtime.signed_presence_document("warming", "earlier incarnation")?; + Ok(vec![ + bad_presence("ghost".into()), + bad_presence(format!("{TARGET}@{}/{own_signer}", digest('c'))), + CultCacheEnvelope { + key: format!("{TARGET}@{}/{own_signer}", digest('d')), r#type: GameCultRuntimePresenceHealthRecord::TYPE.into(), - payload: vec![0xc1], + payload: earlier.payload, stored_at: stamp.clone(), schema_id: Some(GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()), - }); - } - entries.push(generic_document().unwrap()); - entries + }, + generic_document()?, + ]) })?; let catalog = odin.catalog()?; assert_eq!( documents_of(&catalog, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA), 1, - "Odin's own presence is served" + "the presence that decodes is served" ); assert_eq!(documents_of(&catalog, "ghostlight.doc.v1"), 1); - assert_eq!( - prior_self_publisher_sequence(&odin.store, &odin.provider_identity_id)?, - before, - "the undecodable presences do not change the sequence Odin continues from" + odin.pass()?; + assert!(odin.stored_sequence()? > 0, "Odin publishes past them"); + Ok(()) + } + + /// The bytes this thread has read so far, as the kernel counts them. + fn thread_bytes_read() -> Result { + let io = std::fs::read_to_string("/proc/thread-self/io")?; + Ok(io + .lines() + .find_map(|line| line.strip_prefix("rchar: ")) + .context("no rchar in /proc/thread-self/io")? + .parse()?) + } + + /// A catalog query reads Idunn's projection once, however many stored + /// presences it has to look up there. + #[test] + fn a_catalog_query_reads_the_projection_once() -> Result<()> { + const PRESENCES: usize = 8; + let odin = activated_odin_with(CultMeshRudpDocumentServerOptions::default(), |runtime| { + let stamp = rfc3339_millis(unix_millis()?)?; + (0..PRESENCES) + .map(|index| { + Ok(CultCacheEnvelope { + key: format!("stored-{index}"), + r#type: GameCultRuntimePresenceHealthRecord::TYPE.into(), + payload: runtime.signed_presence_document("warming", "stored")?.payload, + stored_at: stamp.clone(), + schema_id: Some(GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()), + }) + }) + .collect() + })?; + let projection = odin.state.borrow().options.idunn_projection.clone(); + let projection_bytes = std::fs::metadata(&projection)?.len(); + // Every stored presence is looked up in the projection; the filter + // then keeps none of them, so none collides with another in the reply. + let query = CultMeshRudpSnapshotQuery { + session: cultmesh_rs::CultMeshRudpSessionKey { + remote_addr: "127.0.0.1:1".parse()?, + connection_id: 7, + }, + message_id: "lookups".into(), + requested_at_unix_millis: 1, + schema_ids: None, + record_keys: Some(vec!["absent".into()]), + }; + let before = thread_bytes_read()?; + assert!(odin.state.borrow_mut().raw_snapshot(&query)?.is_empty()); + let read = thread_bytes_read()? - before; + assert!(read >= projection_bytes, "the projection was read: {read} bytes"); + assert!( + read < 2 * projection_bytes, + "and only once: {read} bytes read, the projection is {projection_bytes}" ); Ok(()) } @@ -2383,10 +2782,7 @@ mod tests { fn a_presence_whose_projection_cannot_be_read_is_skipped_not_fatal() -> Result<()> { let mut odin = activated_odin()?; odin.pass()?; - odin.tamper_store(|mut entries| { - entries.push(generic_document().unwrap()); - entries - })?; + provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1])?; odin.tamper_projection(|entries| { entries .into_iter() @@ -2408,6 +2804,26 @@ mod tests { Ok(()) } + /// A projection file that cannot be read at all skips every presence and + /// nothing else: the catalog still serves the peer documents. + #[test] + fn a_projection_file_that_cannot_be_read_skips_presences_not_the_catalog() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1])?; + assert_eq!( + documents_of(&odin.catalog()?, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA), + 1, + "Odin's own presence is served while the projection reads" + ); + let projection = odin.state.borrow().options.idunn_projection.clone(); + std::fs::write(projection, b"not a cultcache store")?; + let catalog = odin.catalog()?; + assert_eq!(documents_of(&catalog, "ghostlight.doc.v1"), 1); + assert_eq!(documents_of(&catalog, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA), 0); + Ok(()) + } + /// Every failure to establish Odin's OWN authority ends Odin (temporary /// rule, see `survive`), whatever shape the failure takes: an anchor that /// does not decode or names another schema, an Expected Idunn no longer @@ -2579,6 +2995,23 @@ mod tests { key: &str, payload: Vec, ) -> Result<()> { + provider_put_document( + odin, + CultNetRawDocumentRecord { + schema_id: schema.into(), + record_key: key.into(), + stored_at: rfc3339_millis(unix_millis()?)?, + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload, + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + }, + ) + } + + fn provider_put_document(odin: &OdinWorld, document: CultNetRawDocumentRecord) -> Result<()> { odin.state .borrow_mut() .accept_raw_document(CultMeshRudpRawDocumentReceipt { @@ -2589,17 +3022,7 @@ mod tests { message_id: "put".into(), transport_sequence: 1, received_at_unix_millis: unix_millis()?, - document: CultNetRawDocumentRecord { - schema_id: schema.into(), - record_key: key.into(), - stored_at: rfc3339_millis(unix_millis()?)?, - payload_encoding: CultNetRawPayloadEncoding::Messagepack, - payload, - source_runtime_id: None, - source_agent_id: None, - source_role: None, - tags: None, - }, + document, }) } @@ -2613,8 +3036,7 @@ mod tests { let mut odin = activated_odin()?; odin.pass()?; let before = odin.stored_sequence()?; - let stored_before = - SingleFileMessagePackBackingStore::new(&odin.store).pull_all_read_only_snapshot()?; + let stored_before = odin.working_set(); // Every kind Odin itself stored (its watermark and correlations, found // rather than named), and the kinds Idunn projects to it. @@ -2650,7 +3072,7 @@ mod tests { assert!(format!("{refused:#}").contains("accepts no"), "{refused:#}"); } assert_eq!( - SingleFileMessagePackBackingStore::new(&odin.store).pull_all_read_only_snapshot()?, + odin.working_set(), stored_before, "no refused put touched the store" ); @@ -2672,12 +3094,11 @@ mod tests { let mut odin = activated_odin()?; odin.pass()?; provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1])?; - let stored: BTreeSet = - SingleFileMessagePackBackingStore::new(&odin.store) - .pull_all_read_only_snapshot()? - .into_iter() - .map(|entry| entry.r#type) - .collect(); + let stored: BTreeSet = odin + .working_set() + .into_iter() + .map(|entry| entry.r#type) + .collect(); assert!( stored.len() >= 4, "Odin's store holds more than what it serves: {stored:?}" @@ -2783,6 +3204,8 @@ mod tests { ); let lost = anyhow::Error::new(WriteLeaseLost("gone".into())).context("refreshing"); assert!(survive::<()>(&state, "x", Err(lost)).is_err()); + let foreign = anyhow::Error::new(ForeignStoreWrite).context("writing"); + assert!(survive::<()>(&state, "x", Err(foreign)).is_err()); Ok(()) } @@ -2912,4 +3335,524 @@ mod tests { assert!(!settle_poll(fault(), &mut since, restart).unwrap()); assert!(settle_poll(fault(), &mut since, restart + POLL_FAILURE_LIMIT).is_err()); } + + // ---- one write per interval ---------------------------------------------- + + /// The store file as a reader sees it: which file, and what it holds. Each + /// write replaces the file atomically, so it is a new file. + fn file_identity(path: &Path) -> Result<(u64, Vec)> { + use std::os::unix::fs::MetadataExt; + Ok((std::fs::metadata(path)?.ino(), std::fs::read(path)?)) + } + + fn ago(millis: u64) -> Option { + Instant::now().checked_sub(Duration::from_millis(millis)) + } + + /// Twenty-five of Odin's own changes (heartbeats) inside one interval are + /// one write, made once the interval has passed: not at 0.9 s after the + /// last write, at 1.1 s. + #[test] + fn changes_inside_one_interval_are_one_write() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + let first = file_identity(&odin.store)?; + for _ in 0..25 { + odin.timers.last_heartbeat = None; + odin.timers.last_flush = Some(Instant::now()); + odin.pass()?; + } + assert!(odin.stored_sequence()? >= 26, "every heartbeat was admitted"); + assert_eq!(file_identity(&odin.store)?, first, "no write inside the interval"); + + odin.timers.last_flush = ago(900); + odin.pass()?; + assert_eq!(file_identity(&odin.store)?, first, "0.9 s is inside the interval"); + + odin.timers.last_flush = ago(1_100); + odin.pass()?; + let second = file_identity(&odin.store)?; + assert_ne!(second.0, first.0, "one write once the interval passed"); + assert_eq!(odin.file_records()?, odin.working_set()); + + // The write itself starts the next interval: a change right after it + // waits. + odin.heartbeat()?; + odin.pass()?; + assert_eq!(file_identity(&odin.store)?, second, "and only one"); + Ok(()) + } + + /// A peer's put is on disk when Odin accepts it, inside the interval, and + /// the write carries Odin's own unwritten changes with it. + #[test] + fn a_put_is_written_before_it_is_accepted() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + odin.heartbeat()?; + odin.timers.last_flush = Some(Instant::now()); + provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1])?; + assert!( + odin.file_records()? + .iter() + .any(|entry| entry.r#type == "ghostlight.doc" && entry.key == "doc-1") + ); + assert_eq!(odin.file_records()?, odin.working_set()); + Ok(()) + } + + /// Two puts pipelined on one session with the first datagram lost on the + /// way: the transport resends it and delivers both in order, and Odin + /// accepts, and so writes, both. The transport may acknowledge the second + /// before the first is resent (see `accept_raw_document`); what holds is + /// that neither put is lost and both are on disk once delivered. + #[test] + fn pipelined_puts_with_a_lost_datagram_are_both_written() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + let server = odin.server.local_addr()?; + let acknowledged = odin.serve_while(move || -> Result { + // A relay that drops the client's first reliable data datagram. + let relay_socket = UdpSocket::bind("127.0.0.1:0")?; + relay_socket.set_read_timeout(Some(Duration::from_millis(5)))?; + let relay_addr = relay_socket.local_addr()?; + let stop = Arc::new(AtomicBool::new(false)); + let relay = thread::spawn({ + let stop = stop.clone(); + move || -> Result<()> { + let mut client = None; + let mut dropped = false; + let mut buffer = vec![0u8; 65_535]; + while !stop.load(Ordering::Relaxed) { + let Ok((length, from)) = relay_socket.recv_from(&mut buffer) else { + continue; + }; + let datagram = &buffer[..length]; + if from == server { + if let Some(client) = client { + relay_socket.send_to(datagram, client)?; + } + continue; + } + client = Some(from); + let reliable_data = cultnet_rs::decode_rudp_packet(datagram).is_ok_and( + |packet| { + packet.packet_type == cultnet_rs::CultNetRudpPacketType::Data + && packet.reliable + }, + ); + if reliable_data && !dropped { + dropped = true; + continue; + } + relay_socket.send_to(datagram, server)?; + } + Ok(()) + } + }); + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_read_timeout(Some(Duration::from_millis(5)))?; + let mut transport = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("pipeliner", socket, relay_addr, 0x6100_0001), + )?; + transport.connect(Vec::new())?; + let deadline = Instant::now() + Duration::from_secs(3); + while !transport.connected() { + let _ = transport.receive_once()?; + transport.poll_resends()?; + ensure!(Instant::now() < deadline, "timed out connecting"); + } + let put = |key: &str| -> Result> { + let message = CultNetMessage::DocumentPutRaw { + message_id: format!("put-{key}"), + document: CultNetRawDocumentRecord { + schema_id: "ghostlight.doc.v1".into(), + record_key: key.into(), + stored_at: rfc3339_millis(unix_millis()?)?, + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: vec![7; 64], + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + }, + }; + Ok(encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?) + }; + let receipts = [ + transport.send_reliable("schema", put("first")?)?, + transport.send_reliable("schema", put("second")?)?, + ]; + let deadline = Instant::now() + Duration::from_secs(5); + let all_acknowledged = |transport: &CultNetRudpSocketTransportConnection| { + receipts.iter().all(|receipt| { + transport.reliable_send_status(receipt) + == CultNetRudpReliableSendStatus::Acknowledged + }) + }; + while !all_acknowledged(&transport) && Instant::now() < deadline { + let _ = transport.receive_once()?; + transport.poll_resends()?; + } + stop.store(true, Ordering::Relaxed); + relay.join().expect("relay thread panicked")?; + Ok(all_acknowledged(&transport)) + })??; + assert!(acknowledged, "both puts were delivered"); + let written: Vec = odin + .file_records()? + .into_iter() + .filter(|entry| entry.r#type == "ghostlight.doc") + .map(|entry| entry.key) + .collect(); + assert_eq!(written, vec!["first".to_owned(), "second".to_owned()]); + assert_eq!(odin.file_records()?, odin.working_set()); + Ok(()) + } + + /// A put whose write is not made is refused, so it is never accepted: + /// with the write failing, and with the lease lost. + #[test] + fn a_put_whose_write_is_not_made_is_refused() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + std::fs::write(&odin.store, b"not a cultcache store")?; + assert!(provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1]).is_err()); + + let mut odin = activated_odin()?; + odin.pass()?; + let written = file_identity(&odin.store)?; + odin.swap_lease()?; + let error = provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1]).unwrap_err(); + assert!(error.downcast_ref::().is_some(), "{error:#}"); + assert_eq!(file_identity(&odin.store)?, written); + Ok(()) + } + + /// A correlation for an incarnation Idunn no longer projects is found + /// through the correlations Odin holds, and the refresh pass withdraws it. + #[test] + fn a_correlation_whose_incarnation_is_no_longer_projected_is_withdrawn() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + let candidate = project_candidate_incarnation(&odin)?; + odin.state.borrow().refresh_all_correlations()?; + let correlations = || odin.stored_keys(OdinRuntimeTopologyCorrelationRecord::TYPE); + assert!(correlations()?.contains(&candidate.key())); + + odin.tamper_projection(|entries| { + entries + .into_iter() + .filter(|entry| entry.key != candidate.key()) + .collect() + })?; + odin.state.borrow().refresh_all_correlations()?; + assert!(!correlations()?.contains(&candidate.key())); + assert_eq!(correlations()?.len(), 1, "Odin's own stays"); + Ok(()) + } + + /// An interval in which nothing changed writes nothing, however often the + /// write is due. With nothing to write the write step reads nothing + /// either, not even the lease: it is due on every pass until something + /// changes. + #[test] + fn nothing_changed_is_no_write() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + let written = file_identity(&odin.store)?; + for _ in 0..5 { + odin.timers.last_projection_refresh = None; + odin.timers.last_heartbeat = Some(Instant::now()); + odin.timers.last_flush = None; + odin.pass()?; + } + assert_eq!(file_identity(&odin.store)?, written); + + std::fs::write(&odin.lease_path, b"not a cultcache store")?; + odin.timers.last_projection_refresh = Some(Instant::now()); + odin.timers.last_heartbeat = Some(Instant::now()); + odin.timers.last_flush = None; + odin.pass()?; + assert_eq!(file_identity(&odin.store)?, written); + Ok(()) + } + + /// A peer document whose single-document snapshot response, encoded as the + /// server encodes one with a one-byte message id, is `response_bytes` long. + fn document_answering_in(key: &str, response_bytes: usize) -> Result { + let document = |payload_bytes: usize| -> Result { + Ok(CultCacheEnvelope { + key: key.into(), + r#type: "ghostlight.doc".into(), + payload: vec![7; payload_bytes], + stored_at: rfc3339_millis(unix_millis()?)?, + schema_id: Some("ghostlight.doc.v1".into()), + }) + }; + let response = |envelope: &CultCacheEnvelope| -> Result { + let served = CultNetRawDocumentRecord { + schema_id: "ghostlight.doc.v1".into(), + record_key: envelope.key.clone(), + stored_at: envelope.stored_at.clone(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: envelope.payload.clone(), + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + }; + let message = CultNetMessage::SnapshotResponseRaw { + message_id: "m".into(), + documents: vec![served], + }; + Ok(encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?.len()) + }; + let overhead = response(&document(1_000)?)? - 1_000; + let sized = document(response_bytes - overhead)?; + ensure!(response(&sized)? == response_bytes, "the document was not sized exactly"); + Ok(sized) + } + + /// A stored peer document the server could never serve is dropped when + /// Odin activates, everything else is served, and the first write removes + /// it from the file. The bound is the server's own limit, from the options + /// it was built with: a document answering in exactly that many bytes is + /// kept, one byte more is dropped. + #[test] + fn a_stored_document_the_server_could_never_serve_is_dropped_at_activation() -> Result<()> { + const LIMIT: usize = 4_096; + let options = CultMeshRudpDocumentServerOptions { + max_snapshot_response_bytes: LIMIT, + ..CultMeshRudpDocumentServerOptions::default() + }; + let mut odin = activated_odin_with(options, |_| { + Ok(vec![ + document_answering_in("at-limit", LIMIT)?, + document_answering_in("over-limit", LIMIT + 1)?, + document_answering_in("small", 2_000)?, + ]) + })?; + let held = odin.stored_keys("ghostlight.doc")?; + assert_eq!(held, vec!["at-limit".to_owned(), "small".to_owned()]); + let catalog = odin.catalog()?; + assert_eq!(documents_of(&catalog, "ghostlight.doc.v1"), 2); + assert!( + odin.file_records()?.iter().any(|entry| entry.key == "over-limit"), + "activation itself writes nothing" + ); + + odin.pass()?; + let written = odin.file_records()?; + assert!(!written.iter().any(|entry| entry.key == "over-limit")); + assert_eq!(written, odin.working_set()); + Ok(()) + } + + /// A due write with nothing to write does not start the interval: the next + /// change is written on the first pass after it. + #[test] + fn with_nothing_to_write_the_next_change_is_written_on_the_next_pass() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + let written = file_identity(&odin.store)?; + odin.timers.last_projection_refresh = Some(Instant::now()); + odin.timers.last_heartbeat = Some(Instant::now()); + odin.timers.last_flush = ago(1_100); + odin.pass()?; + assert_eq!(file_identity(&odin.store)?, written, "nothing to write"); + + odin.heartbeat()?; + odin.pass()?; + assert_ne!(file_identity(&odin.store)?, written, "the change is written at once"); + assert_eq!(odin.file_records()?, odin.working_set()); + Ok(()) + } + + /// The lease is checked at the write itself: lost between a change and + /// its write, the write is not made and Odin ends. + #[test] + fn a_lease_lost_before_the_write_writes_nothing_and_ends_odin() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + let written = file_identity(&odin.store)?; + odin.heartbeat()?; + odin.swap_lease()?; + odin.timers.last_projection_refresh = Some(Instant::now()); + odin.timers.last_heartbeat = Some(Instant::now()); + odin.timers.last_flush = None; + let error = odin.pass().unwrap_err(); + assert!( + error.downcast_ref::().is_some(), + "{error:#}" + ); + assert_eq!(file_identity(&odin.store)?, written); + Ok(()) + } + + /// A file another process wrote between Odin's writes ends Odin, and is + /// not written over. + #[test] + fn a_foreign_write_ends_odin_and_is_not_written_over() -> Result<()> { + let mut odin = activated_odin()?; + odin.pass()?; + odin.tamper_store(|entries| { + entries + .into_iter() + .filter(|entry| entry.r#type != GameCultRuntimePresenceHealthRecord::TYPE) + .collect() + })?; + let foreign = file_identity(&odin.store)?; + odin.heartbeat()?; + odin.timers.last_flush = None; + let error = odin.pass().unwrap_err(); + assert!( + error.downcast_ref::().is_some(), + "{error:#}" + ); + assert_eq!(file_identity(&odin.store)?, foreign); + Ok(()) + } + + /// Odin serving, written once, with a change of its own not yet written. + fn odin_with_an_unwritten_change() -> Result { + let mut odin = activated_odin()?; + odin.pass()?; + odin.heartbeat()?; + assert_ne!(odin.file_records()?, odin.working_set()); + Ok(odin) + } + + /// Stopping on request makes the last write. + #[test] + fn a_stop_request_writes_what_changed() -> Result<()> { + let mut odin = odin_with_an_unwritten_change()?; + odin.serve(true)?; + assert_eq!(odin.file_records()?, odin.working_set()); + Ok(()) + } + + /// Ending because Odin's own authority is refused, the lease is still + /// current: the last write is made. + #[test] + fn ending_on_a_refused_authority_writes_what_changed() -> Result<()> { + let mut odin = odin_with_an_unwritten_change()?; + odin.tamper_projection(|entries| { + entries + .into_iter() + .filter(|entry| entry.r#type != IdunnRuntimeActivationRecord::TYPE) + .collect() + })?; + odin.timers = ServingTimers::default(); + assert_refused(&odin.serve(false).unwrap_err()); + assert_eq!(odin.file_records()?, odin.working_set()); + Ok(()) + } + + /// Ending on a dead socket, the lease is still current: the last write is + /// made. The socket's own failure is real (a refused datagram, reported on + /// the server's next receive); only the length of the failing run is set. + #[test] + fn ending_on_a_dead_socket_writes_what_changed() -> Result<()> { + let mut odin = odin_with_an_unwritten_change()?; + let closed = UdpSocket::bind("127.0.0.1:0")?.local_addr()?; + odin.socket.connect(closed)?; + odin.socket.send(b"to nobody")?; + thread::sleep(Duration::from_millis(50)); + odin.timers.poll_failing_since = ago(POLL_FAILURE_LIMIT.as_millis() as u64 + 1_000); + let error = odin.serve(false).unwrap_err(); + assert!(format!("{error:#}").contains("failed every poll"), "{error:#}"); + assert_eq!(odin.file_records()?, odin.working_set()); + Ok(()) + } + + /// Ending because the lease was lost, nothing more is written. + #[test] + fn ending_on_a_lost_lease_writes_nothing() -> Result<()> { + let mut odin = odin_with_an_unwritten_change()?; + let written = file_identity(&odin.store)?; + odin.swap_lease()?; + odin.timers = ServingTimers::default(); + let error = odin.serve(false).unwrap_err(); + assert!(error.downcast_ref::().is_some(), "{error:#}"); + assert_eq!(file_identity(&odin.store)?, written); + Ok(()) + } + + /// Ending because another process wrote the file, nothing is written over + /// it. + #[test] + fn ending_on_a_foreign_write_writes_nothing() -> Result<()> { + let mut odin = odin_with_an_unwritten_change()?; + odin.tamper_store(|entries| { + entries + .into_iter() + .filter(|entry| entry.r#type != GameCultRuntimePresenceHealthRecord::TYPE) + .collect() + })?; + let foreign = file_identity(&odin.store)?; + odin.timers.last_flush = None; + let error = odin.serve(false).unwrap_err(); + assert!(error.downcast_ref::().is_some(), "{error:#}"); + assert_eq!(file_identity(&odin.store)?, foreign); + Ok(()) + } + + // ---- the store file is not working memory ----------------------------- + + /// Once Odin holds its lease, its working set is in memory. Rewriting the + /// store file behind its back changes nothing Odin serves or refreshes; + /// only a restart, which loads the file again, would see it. + #[test] + fn rewriting_the_store_file_behind_odin_changes_nothing_it_serves() -> Result<()> { + let mut odin = activated_odin()?; + provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1])?; + odin.pass()?; + let served = odin.catalog()?; + let working = odin.working_set(); + assert_eq!(documents_of(&served, "ghostlight.doc.v1"), 1); + + odin.tamper_store(|entries| { + let mut entries: Vec<_> = entries + .into_iter() + .filter(|entry| entry.r#type == "odin.topology_publisher_watermark") + .collect(); + let mut foreign = generic_document().unwrap(); + foreign.key = "foreign".into(); + entries.push(foreign); + entries + })?; + for _ in 0..3 { + odin.state.borrow().refresh_all_correlations()?; + } + assert_eq!(odin.catalog()?, served); + assert_eq!(odin.working_set(), working); + assert_ne!( + MemoryOdinTopologyStore::load(&odin.store)?.records().len(), + working.len(), + "a restart would load what the file now holds" + ); + Ok(()) + } + + /// Refresh passes and catalog reads never read the store file. With the + /// file replaced by bytes that do not decode, a reader would fail at once; + /// every pass succeeds and logs nothing. + #[test] + fn refresh_passes_and_catalog_reads_never_read_the_store_file() -> Result<()> { + let mut odin = activated_odin()?; + provider_put(&odin, "ghostlight.doc.v1", "doc-1", vec![1])?; + odin.pass()?; + let served = odin.catalog()?; + std::fs::write(&odin.store, b"not a cultcache store")?; + for _ in 0..5 { + odin.state.borrow().refresh_all_correlations()?; + assert_eq!(odin.catalog()?, served); + } + assert!( + odin.state.borrow().log_gate.borrow().seen.is_empty(), + "a pass failed and was logged" + ); + Ok(()) + } } diff --git a/crates/odin-daemon/tests/daemon_process.rs b/crates/odin-daemon/tests/daemon_process.rs new file mode 100644 index 0000000..e0bb98a --- /dev/null +++ b/crates/odin-daemon/tests/daemon_process.rs @@ -0,0 +1,350 @@ +//! The real odin-daemon binary, run under an Idunn-shaped world: its bundle, +//! signer descriptors, projection and write lease on disk, and Idunn's warming +//! probe and lease grant made over RUDP as Idunn makes them. It reaches what +//! the in-process tests cannot: `main`'s signal wiring and a process that dies. +//! Unix only, like the daemon. + +use std::fs::File; +use std::net::{SocketAddr, UdpSocket}; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, ExitStatus, Stdio}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use anyhow::{Result, bail, ensure}; +use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore}; +use cultmesh_rs::{ + CultMeshRudpDocumentPublishOptions, CultMeshRudpSnapshotOptions, + publish_cultnet_message_to_rudp_catalog, request_raw_snapshot_from_rudp_catalog, +}; +use cultnet_rs::*; +use odin_daemon::IncarnationRef; + +fn digest(byte: char) -> String { + format!("sha256-{}", byte.to_string().repeat(64)) +} + +fn now() -> u64 { + SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64 +} + +fn rfc3339(millis: u64) -> String { + chrono::DateTime::::from_timestamp_millis(millis as i64) + .unwrap() + .to_rfc3339() +} + +/// Replace the whole store at `path` with `entries`. +fn write_store(path: &Path, entries: &[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, entries)?, "store write lost a race"); + Ok(()) +} + +fn credential_bytes(store: &Path) -> Result> { + Ok(rmp_serde::to_vec(&SingleFileMessagePackBackingStore::new(store).pull_all()?)?) +} + +struct Odin { + _temp: tempfile::TempDir, + root: PathBuf, + candidate: SocketAddr, + child: Child, + store: PathBuf, +} + +impl Odin { + fn log(&self) -> String { + std::fs::read_to_string(self.root.join("odin.log")).unwrap_or_default() + } + + /// Put one provider document and return once the server acknowledged it. + fn put(&self, key: &str) -> Result<()> { + let message = CultNetMessage::DocumentPutRaw { + message_id: format!("put-{key}"), + document: CultNetRawDocumentRecord { + schema_id: "ghostlight.doc.v1".into(), + record_key: key.into(), + stored_at: rfc3339(now()), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: vec![1], + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + }, + }; + let mut options = CultMeshRudpDocumentPublishOptions::odin(self.candidate, "test-publisher"); + options.flush_timeout = Duration::from_secs(3); + publish_cultnet_message_to_rudp_catalog(&message, options) + } + + fn holds(&self, key: &str) -> Result { + Ok(SingleFileMessagePackBackingStore::new(&self.store) + .pull_all_read_only_snapshot()? + .iter() + .any(|entry| entry.r#type == "ghostlight.doc" && entry.key == key)) + } + + fn signal(&mut self, signal: &str) -> Result { + let pid = self.child.id().to_string(); + ensure!(Command::new("kill").args([signal, &pid]).status()?.success()); + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if let Some(status) = self.child.try_wait()? { + return Ok(status); + } + ensure!(Instant::now() < deadline, "Odin did not exit on {signal}"); + std::thread::sleep(Duration::from_millis(10)); + } + } +} + +impl Drop for Odin { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +/// Launch Odin as Idunn does and grant it its lease; returns once Odin has +/// written its store, which it does on its first pass after activating. +fn start_odin() -> Result { + let temp = tempfile::tempdir()?; + let root = temp.path().to_path_buf(); + let candidate = UdpSocket::bind("127.0.0.1:0")?.local_addr()?; + let t0 = now(); + let idunn = enroll_service_identity_at::(&root.join("idunn.cc"))?; + enroll_service_identity_at::(&root.join("odin-topology.cc"))?; + let provider = + enroll_service_identity_at::(&root.join("odin-provider.cc"))?; + std::fs::write(root.join("topology-cred"), credential_bytes(&root.join("odin-topology.cc"))?)?; + std::fs::write(root.join("provider-cred"), credential_bytes(&root.join("odin-provider.cc"))?)?; + export_service_identity_trust_anchor::(&idunn, &root.join("idunn-anchor.cc"))?; + let expected = IdunnExpectedIncarnationRecord { + schema_version: IDUNN_EXPECTED_INCARNATION_SCHEMA.into(), + target: "odin".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.entry().identity_id.clone(), + health_contract: "odin.runtime-health.v1".into(), + artifact_sha256: digest('5'), + state_schema_generation: Some("odin-v2".into()), + state_contract_sha256: Some( + "sha256-4f2f2dcd931d16f6b02bf295f41227b867aa982765661d55fa9f29fb2db7e449".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: "odin.verse-rendezvous".into(), + schema: "odin.verse-topology.v1".into(), + compatibility: "v1".into(), + minimum_capacity: 1, + }], + dependencies: Vec::new(), + }; + expected.validate()?; + let launch = IdunnRuntimeActivationLaunch::issue(&expected, digest('7'), t0 - 20, &idunn)?; + let activation = launch.activation().clone(); + let mut activation_credential = Vec::new(); + launch.write_credential(&mut activation_credential)?; + std::fs::write(root.join("activation-cred"), &activation_credential)?; + + let expected_envelope = |key: &str| -> Result { + Ok(CultCacheEnvelope { + key: key.into(), + r#type: IdunnExpectedIncarnationRecord::TYPE.into(), + payload: expected.canonical_bytes()?, + stored_at: rfc3339(t0 - 30), + schema_id: Some(IDUNN_EXPECTED_INCARNATION_SCHEMA.into()), + }) + }; + let activation_envelope = |key: &str| -> Result { + Ok(CultCacheEnvelope { + key: key.into(), + r#type: IdunnRuntimeActivationRecord::TYPE.into(), + payload: activation.canonical_bytes()?, + stored_at: rfc3339(activation.issued_at_unix_millis), + schema_id: Some(IDUNN_RUNTIME_ACTIVATION_SCHEMA.into()), + }) + }; + let bundle = root.join("bundle"); + std::fs::create_dir_all(&bundle)?; + write_store(&bundle.join("expected.cc"), &[expected_envelope("odin")?])?; + write_store(&bundle.join("activation.cc"), &[activation_envelope("odin")?])?; + + let anchor = GameCultServiceTrustAnchorRecord { + schema_version: GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA.into(), + trust_anchor_id: "root/odin/runtime-presence".into(), + service_id: "odin".into(), + runtime_id: expected.runtime_id.clone(), + signer_identity_id: provider.entry().identity_id.clone(), + signer_public_key: provider.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: t0 - 100, + expires_at_unix_millis: None, + private_state_exposed: false, + }; + let incarnation_key = IncarnationRef::of(&expected)?.key(); + let projection = root.join("projection.cc"); + let mut projected = vec![ + expected_envelope(&incarnation_key)?, + CultCacheEnvelope { + key: anchor.trust_anchor_id.clone(), + r#type: GameCultServiceTrustAnchorRecord::TYPE.into(), + payload: rmp_serde::to_vec(&anchor)?, + stored_at: rfc3339(anchor.bound_at_unix_millis), + schema_id: Some(GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA.into()), + }, + activation_envelope(&incarnation_key)?, + ]; + write_store(&projection, &projected)?; + let lease_path = root.join("lease.cc"); + File::create(root.join("lease.cc.lock"))?; + let state_root = root.join("state"); + std::fs::create_dir_all(&state_root)?; + + // Descriptors 3 and 4 carry the signer credentials, as systemd passes them. + let child = Command::new("bash") + .arg("-c") + .arg(format!( + "exec 3<{activation} 4<{provider}; exec {bin} --state-root {state} --idunn-projection {projection} --idunn-anchor {anchor}", + activation = root.join("activation-cred").display(), + provider = root.join("provider-cred").display(), + bin = env!("CARGO_BIN_EXE_odin-daemon"), + state = state_root.display(), + projection = projection.display(), + anchor = root.join("idunn-anchor.cc").display(), + )) + .env("GAMECULT_IDUNN_RUNTIME_BUNDLE", &bundle) + .env("GAMECULT_IDUNN_CANDIDATE_BIND", candidate.to_string()) + .env("GAMECULT_IDUNN_PROCESS_WRITE_LEASE", &lease_path) + .env("ODIN_TOPOLOGY_IDENTITY", root.join("topology-cred")) + .env("LISTEN_PID", "1") + .env("LISTEN_FDS", "2") + .env( + "LISTEN_FDNAMES", + format!("{IDUNN_RUNTIME_ACTIVATION_CREDENTIAL_NAME}:gamecult-runtime-presence-identity"), + ) + .stdout(Stdio::null()) + .stderr(Stdio::from(File::create(root.join("odin.log"))?)) + .spawn()?; + let mut odin = Odin { + _temp: temp, + root, + candidate, + child, + store: state_root.join("topology.cc"), + }; + + // Idunn's warming probe, then the lease naming that exact warming presence. + let deadline = Instant::now() + Duration::from_secs(20); + let response = loop { + match request_raw_snapshot_from_rudp_catalog(CultMeshRudpSnapshotOptions { + target: candidate, + runtime_id: "test-idunn".into(), + message_id: format!("warming-{}", now()), + schema_ids: Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]), + record_keys: Some(vec!["odin".into()]), + ..Default::default() + }) { + Ok(response) => break response, + Err(error) => { + if let Some(status) = odin.child.try_wait()? { + bail!("Odin exited while warming ({status}): {}", odin.log()); + } + ensure!(Instant::now() < deadline, "Odin never answered warming: {error:#}"); + std::thread::sleep(Duration::from_millis(100)); + } + } + }; + let CultNetMessage::SnapshotResponseRaw { documents, .. } = response else { + bail!("the warming probe was not answered with raw documents") + }; + let presence: GameCultRuntimePresenceHealthRecord = rmp_serde::from_slice(&documents[0].payload)?; + ensure!(presence.state == "warming", "{}", presence.state); + let lease = IdunnProcessWriteLeaseRecord { + schema_version: IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into(), + target: "odin".into(), + expected_projection_sha256: expected.canonical_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: "odin-v2".into(), + state_contract_sha256: expected.state_contract_sha256.clone().unwrap(), + runtime_id: expected.runtime_id.clone(), + runtime_instance_id: activation.runtime_instance_id.clone(), + warming_presence_sha256: presence.canonical_sha256()?, + lease_epoch: 1, + issued_at_unix_millis: now() - 5, + }; + let lease_envelope = |key: &str| -> Result { + Ok(CultCacheEnvelope { + key: key.into(), + r#type: IdunnProcessWriteLeaseRecord::TYPE.into(), + payload: lease.canonical_bytes()?, + stored_at: rfc3339(lease.issued_at_unix_millis), + schema_id: Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into()), + }) + }; + write_store(&lease_path, &[lease_envelope("odin")?])?; + projected.push(lease_envelope(&incarnation_key)?); + write_store(&projection, &projected)?; + + while !odin.store.is_file() { + if let Some(status) = odin.child.try_wait()? { + bail!("Odin exited while activating ({status}): {}", odin.log()); + } + ensure!(Instant::now() < deadline, "Odin never activated: {}", odin.log()); + std::thread::sleep(Duration::from_millis(20)); + } + Ok(odin) +} + +/// A put Odin acknowledged is on disk: killed outright right after the +/// acknowledgement, inside the interval its own changes wait for, Odin loses +/// none of them. +#[test] +fn an_acknowledged_put_survives_sigkill() -> Result<()> { + for round in 0..3 { + let mut odin = start_odin()?; + // Inside the interval that Odin's activation write started. + std::thread::sleep(Duration::from_millis(300)); + let key = format!("doc-{round}"); + odin.put(&key)?; + odin.signal("-KILL")?; + ensure!(odin.holds(&key)?, "round {round}: the acknowledged put was lost: {}", odin.log()); + } + Ok(()) +} + +/// SIGTERM reaches the serving loop as a stop request: Odin makes its last +/// write and exits cleanly, rather than waiting to be killed. +#[test] +fn a_stop_signal_ends_odin_cleanly() -> Result<()> { + let mut odin = start_odin()?; + let status = odin.signal("-TERM")?; + let log = odin.log(); + ensure!(status.success(), "Odin exited {status}: {log}"); + ensure!(log.contains("Odin stopping on request"), "{log}"); + ensure!(!log.contains("final store write was not made"), "{log}"); + Ok(()) +}