diff --git a/crates/odin-daemon/src/lib.rs b/crates/odin-daemon/src/lib.rs index beec1bc..67c8e7d 100644 --- a/crates/odin-daemon/src/lib.rs +++ b/crates/odin-daemon/src/lib.rs @@ -6,7 +6,7 @@ use std::cell::{Cell, Ref, RefCell}; use std::collections::{BTreeMap, BTreeSet}; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::time::{SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result, bail, ensure}; @@ -177,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, } @@ -187,22 +189,6 @@ impl CultCacheIdunnProjectionSource { Self { path: path.into() } } - pub fn path(&self) -> &Path { - &self.path - } -} - -impl IdunnProjectionSource for CultCacheIdunnProjectionSource { - fn projections(&self, target: &str) -> Result> { - self.snapshot()?.projections(target) - } - - fn projection(&self, incarnation: &IncarnationRef) -> Result> { - self.snapshot()?.projection(incarnation) - } -} - -impl CultCacheIdunnProjectionSource { /// Read the projection file once. An absent file projects nothing. pub fn snapshot(&self) -> Result { if !self.path.is_file() { @@ -227,8 +213,7 @@ impl CultCacheIdunnProjectionSource { } /// One reading of Idunn's projection file. Every lookup made against it is -/// answered from the same bytes, so a pass that refreshes every incarnation -/// reads the file once. +/// answered from the same bytes. pub struct IdunnProjectionSnapshot { entries: Vec, } @@ -536,10 +521,6 @@ impl MemoryOdinTopologyStore { }) } - pub fn path(&self) -> &Path { - &self.path - } - /// Every record, in the order the file stores them. pub fn records(&self) -> Ref<'_, BTreeMap> { self.records.borrow() @@ -1704,7 +1685,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, @@ -1750,7 +1731,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(()); @@ -1851,10 +1834,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 @@ -1907,7 +1892,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(()) @@ -1930,7 +1915,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 @@ -2442,10 +2427,13 @@ mod tests { self.engine_on(&self.store, now) } - /// An engine over another store, as a restarted Odin loads one. + /// 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), + CultCacheIdunnProjectionSource::new(&self.projection_path) + .snapshot() + .expect("the test projection reads"), store, &self.odin_signer, FixedClock(now), @@ -2491,7 +2479,7 @@ mod tests { } type TestEngine<'a> = OdinTopologyAuthority< - CultCacheIdunnProjectionSource, + IdunnProjectionSnapshot, &'a MemoryOdinTopologyStore, &'a ServiceIdentitySigner, FixedClock, @@ -3001,6 +2989,7 @@ mod tests { ); std::fs::remove_file(&world.projection_path)?; + let engine = world.engine(NOW); assert!( engine .current_signed_correlation(&service.incarnation()?)? @@ -3055,6 +3044,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 c9105d3..b1b461c 100644 --- a/crates/odin-daemon/src/main.rs +++ b/crates/odin-daemon/src/main.rs @@ -33,8 +33,8 @@ use cultnet_rs::{ use fs2::FileExt; use odin_daemon::{ AuthenticationPolicy, CultCacheIdunnProjectionSource, ForeignStoreWrite, - IdunnProjectionSource, IncarnationRef, MemoryOdinTopologyStore, OdinTopologyAuthority, - PresenceAuthorityRefused, ProjectionUnreadable, SystemClock, + IdunnProjectionSnapshot, IdunnProjectionSource, IncarnationRef, MemoryOdinTopologyStore, + OdinTopologyAuthority, PresenceAuthorityRefused, ProjectionUnreadable, SystemClock, }; const TARGET: &str = "odin"; @@ -180,7 +180,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, @@ -258,6 +259,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); @@ -483,7 +485,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 @@ -546,8 +549,9 @@ 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. 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. @@ -556,7 +560,8 @@ impl RuntimeState { query: &CultMeshRudpSnapshotQuery, ) -> Result> { let records = self.topology()?.store.records(); - let projections = CultCacheIdunnProjectionSource::new(&self.options.idunn_projection); + let projections = + CultCacheIdunnProjectionSource::new(&self.options.idunn_projection).snapshot()?; let mut selected = BTreeMap::new(); for envelope in records.values() { let document = match self.public_document(&projections, envelope) { @@ -585,7 +590,7 @@ impl RuntimeState { /// 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> { let Some(schema_id) = envelope.schema_id.clone() else { @@ -2406,6 +2411,60 @@ mod tests { 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(()) + } + /// A presence whose incarnation's projection cannot be read is skipped as /// that record's failure; the rest of the catalog is still served. #[test]