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

Odin: a catalog query reads Idunn's projection once

Browse source 
The file-backed projection source no longer answers lookups itself, so no
caller can read the file once per lookup: every lookup is made against one
snapshot. The catalog query takes one per query, as the refresh pass does;
startup, activation and the own-projection check take one each.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:24:32 +02:00
commit 620efb85d8
2 changed files with 88 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

52
crates/odin-daemon/src/lib.rs
View file

@ -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<Vec<IdunnRuntimeProjection>> {
self.snapshot()?.projections(target)
}
fn projection(&self, incarnation: &IncarnationRef) -> Result<Option<IdunnRuntimeProjection>> {
self.snapshot()?.projection(incarnation)
}
}
impl CultCacheIdunnProjectionSource {
/// Read the projection file once. An absent file projects nothing.
pub fn snapshot(&self) -> Result<IdunnProjectionSnapshot> {
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<CultCacheEnvelope>,
}
@ -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<RecordId, CultCacheEnvelope>> {
self.records.borrow()
@ -1704,7 +1685,7 @@ fn parse_rfc3339_millis(value: &str) -> Result<u64> {
#[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(&current, &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(&current, &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<OdinTopologyIdentity>,
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);

75
crates/odin-daemon/src/main.rs
View file

@ -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::<IdunnServiceIdentity>(&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<Vec<CultNetRawDocumentRecord>> {
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<Option<CultNetRawDocumentRecord>> {
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<u64> {
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]