crates/mimir-room/src/ring.rs
//! The universe ring, read side.
//!
//! * [`RingDefs`]: every DEFINED ring node (id + attributes + payload hash),
//! folded once per (log identity, `universe_manifest` snapshot) and kept
//! per process. A ring-off room whose binary stream links ring items looks
//! their definitions up here instead of folding the ring's ~10⁶ node
//! events on every open. The ring is append-only and every ring write
//! writes manifest rows in the same atomic release, so the manifest's
//! snapshot id moves exactly when the ring does.
//! * [`allowed_owners`]: which ring crates belong to a scope — the DEPENDS_ON
//! closure of the scope's local crates at their seqs, plus every ring
//! crate the scope's binary stream LINKS (the compiler links crates no
//! DEPENDS_ON edge names: panic_unwind / panic_abort, found on znippy-
//! zoomies), with their own closure. Ring events carry no
//! meaningful seq (each crate@version is dumped once, by whichever commit
//! first needed it), so the log answers "what did seq N depend on" through
//! the local crates' DEPENDS_ON edges folded at N, never through ring seqs.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::{Arc, Mutex};
use anyhow::Result;
use mimir_ingest::schema::UNIVERSE_REPO;
use mimir_ingest::{EventLog, FoldScope, Kind};
use mimir_schema::EdgeKind;
use skade::arrow_array::Array;
use crate::snapshot::{EdgeCols, Monikers, NODE_READ, NodeCols, hash64, node_cols};
/// What a ring definition keeps: kind, level, crate, version, origin, name.
const RING_ATTRS: usize = 6;
fn opt(a: &skade::arrow_array::StringArray, r: usize) -> Option<&str> {
(!a.is_null(r)).then(|| a.value(r))
}
/// The current snapshot id of `universe_manifest` (None: nothing dumped yet).
pub(crate) async fn manifest_snapshot(log: &EventLog) -> Result<Option<i64>> {
// through the handle, so a pinned log answers for ITS committed state
log.manifest_snapshot().await
}
/// Every defined ring node, sorted by id.
pub(crate) struct RingDefs {
pub(crate) ids: Monikers,
/// Per id: indexes into `strings` for kind, level, crate, version, origin, name.
pub(crate) attrs: Vec<[u32; RING_ATTRS]>,
pub(crate) hash: Vec<u64>,
pub(crate) strings: Vec<String>,
/// Ring node events folded to build this.
pub(crate) events_read: u64,
}
impl RingDefs {
pub(crate) fn s(&self, i: u32) -> Option<&str> {
(i != u32::MAX).then(|| self.strings[i as usize].as_str())
}
}
struct RingEntry {
identity: String,
manifest: Option<i64>,
defs: Arc<RingDefs>,
}
static RING: Mutex<Vec<RingEntry>> = Mutex::new(Vec::new());
const RING_CAP: usize = 2;
/// Drop the cached ring definitions (benches: the truly cold path).
pub(crate) fn clear() {
RING.lock().unwrap_or_else(|e| e.into_inner()).clear();
}
/// The ring definitions of `log` as of its manifest snapshot; `true` = cached.
pub(crate) async fn defs(log: &EventLog, identity: &str, manifest: Option<i64>) -> Result<(Arc<RingDefs>, bool)> {
{
let c = RING.lock().unwrap_or_else(|e| e.into_inner());
if let Some(e) = c.iter().find(|e| e.identity == identity && e.manifest == manifest) {
return Ok((e.defs.clone(), true));
}
}
let mut read: Vec<&str> = NODE_READ.to_vec();
if log.spec(Kind::Nodes).schema.field_with_name("name").is_ok() {
read.push("name");
}
let rows = log.fold_rows(Kind::Nodes, &FoldScope { repos: BTreeMap::new(), universe: true }, &read).await?;
let cols = node_cols(&rows, &[])?;
let mut strings: Vec<String> = Vec::new();
let mut sid: HashMap<String, u32> = HashMap::new();
let mut id_of = |s: Option<&str>| -> u32 {
let Some(s) = s else { return u32::MAX };
if let Some(i) = sid.get(s) {
return *i;
}
let i = strings.len() as u32;
strings.push(s.to_string());
sid.insert(s.to_string(), i);
i
};
let mut defs: Vec<(&str, [u32; RING_ATTRS], u64)> = Vec::new();
for &(b, r) in &rows.open {
let (c, r) = (&cols[b as usize], r as usize);
if c.defined.is_null(r) || !c.defined.value(r) {
continue;
}
let attrs = [
id_of(opt(&c.kind, r)),
id_of(opt(&c.level, r)),
id_of(opt(&c.krate, r)),
id_of(opt(&c.version, r)),
id_of(opt(&c.origin, r)),
id_of(c.name.as_ref().and_then(|n| opt(n, r))),
];
defs.push((c.id.value(r), attrs, hash64(&c.hash, r)));
}
// one definition per id (several owners may define one id: keep the
// smallest hash, deterministic)
defs.sort_unstable_by(|a, b| a.0.cmp(b.0).then(a.2.cmp(&b.2)));
defs.dedup_by(|a, b| a.0 == b.0);
let n = defs.len();
let ids = Monikers::from_sorted(defs.iter().map(|d| d.0), n);
let built = Arc::new(RingDefs {
ids,
attrs: defs.iter().map(|d| d.1).collect(),
hash: defs.iter().map(|d| d.2).collect(),
strings,
events_read: rows.events_read,
});
let mut c = RING.lock().unwrap_or_else(|e| e.into_inner());
c.retain(|e| e.identity != identity);
c.push(RingEntry { identity: identity.to_string(), manifest, defs: built.clone() });
while c.len() > RING_CAP {
c.remove(0);
}
Ok((built, false))
}
/// The ring owners (`crate@version:kind`) a scope's local crates depend on:
/// seeds = `to` of every LOCAL open DEPENDS_ON edge, closed over the ring's
/// own DEPENDS_ON edges; a reached crate node maps to the owner whose dump
/// defines it. Seeds also: the crate node of every ring owner defining an id
/// in `linked` (the binary stream's ids). Returns (allowed owners, every ring
/// owner seen).
pub(crate) fn allowed_owners<'a>(
nodes: &[(u32, u32)],
nc: &'a [NodeCols],
edges: &[(u32, u32)],
ec: &'a [EdgeCols],
linked: &HashSet<&str>,
) -> (HashSet<&'a str>, HashSet<&'a str>) {
let mut crate_owner: HashMap<&str, Vec<&str>> = HashMap::new();
let mut owner_crate: HashMap<&str, &str> = HashMap::new();
let mut linked_owners: HashSet<&str> = HashSet::new();
let mut all: HashSet<&str> = HashSet::new();
for &(b, r) in nodes {
let (c, r) = (&nc[b as usize], r as usize);
if c.repo.value(r) != UNIVERSE_REPO {
continue;
}
all.insert(c.owner.value(r));
let defined = !c.defined.is_null(r) && c.defined.value(r);
if defined && !c.kind.is_null(r) && c.kind.value(r) == mimir_schema::kind::CRATE {
crate_owner.entry(c.id.value(r)).or_default().push(c.owner.value(r));
owner_crate.insert(c.owner.value(r), c.id.value(r));
}
if defined && linked.contains(c.id.value(r)) {
linked_owners.insert(c.owner.value(r));
}
}
let dep = EdgeKind::DependsOn.as_str();
let mut ring_adj: HashMap<&str, Vec<&str>> = HashMap::new();
let mut frontier: Vec<&str> = Vec::new();
for &(b, r) in edges {
let (c, r) = (&ec[b as usize], r as usize);
let ring = c.repo.value(r) == UNIVERSE_REPO;
if ring {
all.insert(c.owner.value(r));
}
if c.kind.value(r) != dep {
continue;
}
if ring {
ring_adj.entry(c.from.value(r)).or_default().push(c.to.value(r));
} else {
frontier.push(c.to.value(r));
}
}
frontier.extend(linked_owners.iter().filter_map(|o| owner_crate.get(o)).copied());
let mut seen: HashSet<&str> = HashSet::new();
while let Some(m) = frontier.pop() {
if !seen.insert(m) {
continue;
}
if let Some(next) = ring_adj.get(m) {
frontier.extend(next.iter().copied());
}
}
let mut allowed: HashSet<&str> = seen.iter().filter_map(|m| crate_owner.get(m)).flatten().copied().collect();
// a linked owner without a crate node still belongs
allowed.extend(linked_owners);
(allowed, all)
}