crates/mimir-room/src/lib.rs
//! # mimir-room — the Mímir v2 code room, read side (lane 8a)
//!
//! ```text
//! skade mimir_v2 event log ──fold (scope: repo → seq, universe on/off)──▶ snapshot
//! source stream code_node_events · code_edge_events · code_span_events
//! binary stream bin_node_events · bin_edge_events · code_symbol_events
//! snapshot = node table (moniker → dense ordinal, lexicographic; name,
//! parent, cfg, type, is_test, every further payload column by name)
//! + node spans and edge sites (file, lines) of the scope's repos
//! + one boolean CSR GrB_Matrix per EdgeKind (sorted fast path)
//! + Aᵀ / kind unions built on first use and kept
//! + provenance per node, DERIVED (mimir_schema::derive_provenance)
//! + the linked symbols (size, section, role, artifact)
//! query API (typed, no Cypher) → Answer { value, evidence }
//! ```
//!
//! | question | call |
//! |---|---|
//! | what is this node | [`Room::node`], [`Room::attr`], [`Room::display_path`], [`Room::span`], [`Room::sites`] |
//! | which node is `a::b::c` / this symbol | [`Room::lookup`] |
//! | node sets | [`Room::mask`], [`Room::mask_of`], [`Mask::union`] … |
//! | possible system calls (the max a test must stay inside) | [`Room::possible_syscalls`] → the Graph JSON dwarves-tune-graph reads |
//! | walks | [`Room::reach`], [`Room::dependents`], [`Room::dependents_by`], [`Room::hops_from`], [`Room::min_hops_to`], [`Room::path`] |
//! | weights | [`Room::bottleneck`] (max, min), [`Room::worst_on_path`] (max) |
//! | edges out | [`Room::edges`], [`Room::quotient`] (crate / module / repo / file) |
//! | binary | [`Room::symbols`], [`Room::symbols_of`], [`Room::instances_of`] |
//! | identity | [`Room::duplicate_keys`], [`Room::diff`] |
//!
//! Every answer carries an [`Evidence`] record: which tables fed the room
//! (events read, rows open), which edge kinds the query walked, and what is
//! MISSING — a repo with no binary stream, a kind with no edge, a seed the
//! room does not know. An empty answer is never silent (v1's
//! `evidence_note` law).
//!
//! A built snapshot is cached per process, keyed by (log identity = the
//! event tables' UUIDs, resolved scope); a new commit to any event table
//! moves the log's [`mimir_ingest::Stamp`] and drops that log's entries.
mod attrs;
mod diff;
pub mod evidence;
mod export;
mod lookup;
mod mir;
mod path;
pub mod possible;
mod query;
mod ring;
pub mod scope;
pub mod snapshot;
mod symbols;
mod window;
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::time::Instant;
use anyhow::Result;
use mimir_ingest::{EventLog, Stamp};
use mimir_schema::{EdgeKind, EffectiveVisibility, Level, Origin, Provenance};
use serde::Serialize;
pub use attrs::{AttrValue, FileView, Span};
pub use diff::{Diff, EdgeDiff};
pub use evidence::{Answer, AnswerState, Capped, Evidence, Input, Missing};
pub use export::{EdgeList, EdgeOut, EdgeSel, Group, GroupBy, GroupEdge, Quotient};
pub use lookup::{NameMatch, NameQuery};
pub use mir::{MirBlockView, MirBody, MirSide};
pub use path::PathFound;
pub use possible::{PossibleGraph, PossibleOpts};
pub use query::{JIKES_KINDS, Mask, MaskSpec, Seeds, Walk, Weights};
pub use scope::{ResolvedScope, Scope, ScopeError};
pub use snapshot::{BuildStats, DupKey, RepoRows};
pub use symbols::SymbolView;
pub use window::{CommitWindow, WindowNode};
use snapshot::{F_BIN_NODE, F_DEFINED, F_SRC, F_SYMBOL, F_TEST, F_UNIVERSE_DEF, NONE, NONE8, Snapshot};
/// A room: a shared, immutable snapshot of the code graph under one scope.
#[derive(Clone)]
pub struct Room {
snap: Arc<Snapshot>,
}
/// How [`Room::open`] got its room.
pub struct Opened {
pub room: Room,
/// Served from the process cache (no fold, no matrix build).
pub cached: bool,
/// Built by ADVANCING a kept fold with the lake's delta instead of
/// refolding the log's history.
pub advanced: bool,
/// Wall clock of the open (scope resolution + identity/stamp + build or hit).
pub ms: u64,
}
/// A closed-vocabulary value as its column string (`Some("item")`).
pub(crate) fn ser_str<T: std::fmt::Display, S: serde::Serializer>(v: &Option<T>, s: S) -> std::result::Result<S::Ok, S::Error> {
match v {
Some(v) => s.collect_str(v),
None => s.serialize_none(),
}
}
/// A map keyed by a closed vocabulary, with the column strings as keys.
pub(crate) fn ser_map<K: std::fmt::Display, V: Serialize, S: serde::Serializer>(m: &BTreeMap<K, V>, s: S) -> std::result::Result<S::Ok, S::Error> {
s.collect_map(m.iter().map(|(k, v)| (k.to_string(), v)))
}
/// One node of the room. More fields appear as producers add columns
/// (`#[non_exhaustive]`); a column this build has no field for is read by
/// name with [`Room::attr`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[non_exhaustive]
pub struct NodeView<'a> {
pub ordinal: u64,
pub moniker: &'a str,
pub kind: Option<&'a str>,
#[serde(serialize_with = "ser_str")]
pub level: Option<Level>,
#[serde(rename = "crate")]
pub krate: Option<&'a str>,
pub version: Option<&'a str>,
/// The repo whose row described it (`@universe` for the ring).
pub repo: Option<&'a str>,
#[serde(serialize_with = "ser_str")]
pub origin: Option<Origin>,
#[serde(serialize_with = "ser_str")]
pub provenance: Option<Provenance>,
/// The item's own name (`open`, `Table`, the crate name for a crate node).
pub name: Option<&'a str>,
/// Ordinal of the containing node (module, impl, fn of a param …), when
/// the room holds it.
pub parent: Option<u64>,
/// The file its span lies in, and the lines — `None` without a span row
/// (a stub, a ring item, a binary-only node).
pub file: Option<&'a str>,
pub start_line: Option<u32>,
pub end_line: Option<u32>,
/// A `#[test]` / bench fn of a harness unit.
pub is_test: bool,
/// Printed type (params, fields, consts, statics, aliases).
pub ty: Option<&'a str>,
/// The `cfg` gate as written (`feature = "x" && unix`). An item the
/// configuration compiled OUT is a `cfg_gated_out` node (a token-level
/// stub, not a resolved item) when the producer scanned for them.
pub cfg: Option<&'a str>,
/// Nominal visibility as rustc resolves it (`pub`, `pub(crate)`,
/// `private`, `pub(in crate::a)`); `None` for kinds that have none, for
/// stubs, and in a log written before the column existed
/// ([`Room::has_attr`] tells the last case apart).
pub visibility: Option<&'a str>,
/// How far OUTSIDE its crate the item is visible (`direct` /
/// `reexported` = the nameable API).
#[serde(serialize_with = "ser_str")]
pub effective_visibility: Option<EffectiveVisibility>,
/// First line of the doc comment.
pub doc: Option<&'a str>,
/// The item carries a doc comment; `None` = not known here.
pub has_doc: Option<bool>,
/// A source definition (in scope, or in the ring via the provenance lookup).
pub defined: bool,
/// Only reference stubs describe it in the source stream.
pub stub_only: bool,
pub binary_node: bool,
/// A linked symbol carries it.
pub linked: bool,
/// No node row in any stream: only an edge endpoint.
pub endpoint_only: bool,
}
/// What a room holds.
#[derive(Debug, Clone, Serialize)]
pub struct RoomStats {
pub scope: String,
pub nodes: u64,
#[serde(serialize_with = "ser_map")]
pub edges: BTreeMap<EdgeKind, u64>,
#[serde(serialize_with = "ser_map")]
pub provenance: BTreeMap<Provenance, u64>,
pub node_kinds: BTreeMap<String, u64>,
pub endpoint_only: u64,
pub repo_rows: BTreeMap<String, RepoRows>,
pub universe_rows: u64,
pub unknown_kinds: BTreeMap<String, u64>,
pub inputs: Vec<Input>,
pub build: BuildStats,
/// Derived matrices cached so far: (kind unions, transposes).
pub cached_matrices: (usize, usize),
/// Nodes with a span row / distinct files / positioned edge sites.
pub spans: u64,
pub files: u64,
pub sites: u64,
/// Span rows naming a node or edge the room does not hold.
pub orphan_spans: u64,
/// `is_test` nodes.
pub tests: u64,
/// Linked symbol rows (binary stream).
pub symbols: u64,
/// Node ids several rows describe differently, or a dump repeated.
pub duplicate_keys: u64,
/// Node payload columns read by name ([`Room::attr`]).
pub extra_columns: Vec<String>,
/// Node payload columns this build has no reader for.
pub unread_columns: Vec<String>,
}
struct Entry {
identity: String,
/// The log's stamp when last checked valid (fast path: unchanged = hit).
stamp: Stamp,
key: (BTreeMap<String, i64>, bool),
/// What the snapshot read beyond its repos' source events at seq ≤ N:
/// the ring (`universe_manifest` snapshot) and each repo's binary releases.
manifest: Option<i64>,
ring_used: bool,
bin_revs: BTreeMap<String, i64>,
snap: Arc<Snapshot>,
/// The kept fold: what new events ADVANCE instead of a refold. A ring
/// view's fold carries the universe rows too; its advance is gated on an
/// unchanged `universe_manifest` snapshot and dependency closure.
fold: Option<snapshot::FoldState>,
used: u64,
/// Wall clock of the last hit or build (the idle ring eviction).
used_at: Instant,
}
/// Snapshots kept per process (least recently used goes first); default 4.
/// Of them at most ONE ring view (`Scope::universe`): a second evicts the first.
static CACHE_CAP: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(4);
/// The env knob of [`Room::set_cache_capacity`] a server reads at start.
pub const CACHE_CAP_ENV: &str = "MIMIR_ROOM_CACHE_CAP";
static CACHE: Mutex<Vec<Entry>> = Mutex::new(Vec::new());
/// Does a ring view keep its fold to advance (`MIMIR_RING_KEEP_FOLD=1`)?
/// Off by default: see the build in [`Room::open`].
fn keeps_ring_fold() -> bool {
match RING_FOLD.load(std::sync::atomic::Ordering::Relaxed) {
0 => {
let on = std::env::var("MIMIR_RING_KEEP_FOLD").is_ok_and(|v| v == "1");
RING_FOLD.store(if on { 2 } else { 1 }, std::sync::atomic::Ordering::Relaxed);
on
}
v => v == 2,
}
}
/// 0 = not read yet, 1 = off, 2 = on ([`Room::set_ring_keeps_fold`]).
static RING_FOLD: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
/// See [`Room::entries_at_last_fold`].
static ENTRIES_AT_FOLD: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
static TICK: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
/// Each repo's last binary release (`rev`, 0 = none).
async fn bin_revs(log: &EventLog, repos: impl Iterator<Item = &String>) -> Result<BTreeMap<String, i64>> {
let mut out = BTreeMap::new();
for r in repos {
out.insert(r.clone(), log.bin_offsets(r).await?.values().map(|o| o.rev).max().unwrap_or(0));
}
Ok(out)
}
/// Can events written since a snapshot was built change it? Source events
/// only ever land at seqs past a repo's offset (> N), so they never can.
/// A binary release lands at an existing seq (a re-ingest): it matters when
/// its seq ≤ N. A ring write matters to snapshots that read the ring (ring
/// on, or the ring lookup for linked items). Returns the refreshed binary
/// revs when the snapshot is still valid.
async fn still_valid(log: &EventLog, e: (&BTreeMap<String, i64>, Option<i64>, bool, &BTreeMap<String, i64>)) -> Result<Option<BTreeMap<String, i64>>> {
let (repos, manifest, ring_used, revs) = e;
if ring_used && ring::manifest_snapshot(log).await? != manifest {
return Ok(None);
}
let now = bin_revs(log, repos.keys()).await?;
for (repo, n) in repos {
let (old, cur) = (revs.get(repo).copied().unwrap_or(0), now.get(repo).copied().unwrap_or(0));
if cur > old && log.bin_runs(repo).await?.iter().any(|r| r.rev > old && r.seq <= *n) {
return Ok(None);
}
}
Ok(Some(now))
}
/// Folds in flight, per (log identity, scope key): the gate a second open
/// of the same room waits at.
type BuildKey = (String, (BTreeMap<String, i64>, bool));
static BUILDING: Mutex<BTreeMap<BuildKey, Arc<tokio::sync::Mutex<()>>>> = Mutex::new(BTreeMap::new());
/// The cached snapshot of (identity, key) when nothing written since it was
/// built can change it (an invalid one is dropped).
async fn cached(log: &EventLog, identity: &str, key: &(BTreeMap<String, i64>, bool), stamp: &Stamp) -> Result<Option<Arc<Snapshot>>> {
let tick = TICK.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let found = {
let c = CACHE.lock().unwrap_or_else(|e| e.into_inner());
c.iter()
.find(|e| e.identity == identity && &e.key == key)
.map(|e| (&e.stamp == stamp, e.manifest, e.ring_used, e.bin_revs.clone(), e.snap.clone()))
};
let Some((same_stamp, manifest, ring_used, revs, snap)) = found else { return Ok(None) };
let fresh = if same_stamp { Some(revs) } else { still_valid(log, (&key.0, manifest, ring_used, &revs)).await? };
let mut c = CACHE.lock().unwrap_or_else(|e| e.into_inner());
match fresh {
Some(revs) => {
if let Some(e) = c.iter_mut().find(|e| e.identity == identity && &e.key == key) {
e.used = tick;
e.used_at = Instant::now();
e.stamp = stamp.clone();
e.bin_revs = revs;
}
Ok(Some(snap))
}
None => {
c.retain(|e| !(e.identity == identity && &e.key == key));
Ok(None)
}
}
}
impl Room {
/// The room for `scope`, from the process cache when nothing written
/// since it was built can change it. A repo the log holds no offset for
/// is a [`ScopeError::NotConsumed`] (downcast the error) — a state, not
/// an internal failure.
pub async fn at(log: &EventLog, scope: &Scope) -> Result<Room> {
Ok(Self::open(log, scope).await?.room)
}
/// [`Room::at`] for a caller with no async context (wiring map R14): runs
/// the open on a thread of its own, on the LAKE's runtime
/// ([`EventLog::block_on`]), and blocks until it is done. Callable from
/// anywhere — a plain thread, a blocking-pool thread, or (blocking it) an
/// async task. It never starts a runtime of its own: a table handle built
/// on a short-lived runtime would outlive it in the catalog's cache.
pub fn at_blocking(log: &EventLog, scope: &Scope) -> Result<Room> {
Ok(Self::open_blocking(log, scope)?.room)
}
/// [`Room::open`] for a caller with no async context (see [`Room::at_blocking`]).
pub fn open_blocking(log: &EventLog, scope: &Scope) -> Result<Opened> {
std::thread::scope(|s| {
s.spawn(|| log.block_on(Self::open(log, scope)))
.join()
.unwrap_or_else(|p| std::panic::resume_unwind(p))
})
}
/// [`Room::at`], telling whether the cache served it and how long it took.
///
/// The whole open reads ONE committed state of the lake
/// ([`EventLog::pin`]): it may run while a consumer writes the same log —
/// the room is the last release that landed before the open began, never
/// half of the one in flight — and it never waits for the writer. A scope
/// that names a repo the log never consumed is a
/// [`ScopeError::NotConsumed`] (downcast the error).
pub async fn open(log: &EventLog, scope: &Scope) -> Result<Opened> {
Self::open_or_peek(log, scope, true).await.map(|o| o.expect("an open that may build always has a room"))
}
/// The room for `scope` ONLY when the process cache already holds a
/// valid one — never a fold (a scope that holds no consumed repo is
/// empty and built at once). `None` = it would have to be built (a
/// caller that must answer at once queues the build instead).
pub async fn peek(log: &EventLog, scope: &Scope) -> Result<Option<Room>> {
Ok(Self::open_or_peek(log, scope, false).await?.map(|o| o.room))
}
/// [`Room::peek`] for a caller with no async context (see [`Room::at_blocking`]).
pub fn peek_blocking(log: &EventLog, scope: &Scope) -> Result<Option<Room>> {
std::thread::scope(|s| {
s.spawn(|| log.block_on(Self::peek(log, scope)))
.join()
.unwrap_or_else(|p| std::panic::resume_unwind(p))
})
}
async fn open_or_peek(log: &EventLog, scope: &Scope, build: bool) -> Result<Option<Opened>> {
let t = Instant::now();
let log = &log.pin().await?;
let resolved = scope.resolve(log).await?;
let identity = log.identity().await?;
let stamp = log.stamp().await?;
let key = resolved.key();
if let Some(snap) = cached(log, &identity, &key, &stamp).await? {
return Ok(Some(Opened { room: Room { snap }, cached: true, advanced: false, ms: t.elapsed().as_millis() as u64 }));
}
// a scope with no consumed repo folds nothing: peek builds that one
if !build && !resolved.repos.is_empty() {
return Ok(None);
}
// ONE fold per (log, scope) at a time: a second open of the same
// room while the first folds waits for it and takes its snapshot
// (panes asking at once after a restart folded side by side — the
// 114 s first answer on dwarf, 2026-10-15)
let gate = {
let mut b = BUILDING.lock().unwrap_or_else(|e| e.into_inner());
b.entry((identity.clone(), key.clone())).or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))).clone()
};
let held = gate.lock().await;
let built = async {
if let Some(snap) = cached(log, &identity, &key, &stamp).await? {
return Ok::<_, anyhow::Error>((snap, true, false));
}
// read BEFORE the fold: a release landing in between only makes a
// later check conservative, never stale
let revs = bin_revs(log, resolved.repos.keys()).await?;
// the same room at older seqs may have left a kept fold: ADVANCE
// it by the lake's delta instead of refolding the whole history.
// Only a room AT THE LOG'S HEAD advances — a time-travel scope
// would skip the events past its seqs while the track's read
// mark moved to the lake's current snapshot (a poisoned fold) —
// and only the FOLD leaves the old entry: its snapshot stays,
// still serving its own (time-travel) key.
let at_head = Scope { repos: resolved.repos.keys().map(|r| (r.clone(), None)).collect(), universe: resolved.universe }
.resolve(log)
.await
.is_ok_and(|h| h.repos == resolved.repos);
let stale = if at_head {
let mut c = CACHE.lock().unwrap_or_else(|e| e.into_inner());
let same_room = |e: &Entry| {
e.identity == identity
&& e.key.1 == key.1
&& e.key.0.len() == key.0.len()
&& e.key.0.iter().all(|(r, n)| key.0.get(r).is_some_and(|m| m >= n))
};
// the old head's ENTRY goes with its fold: holding its
// snapshot beside the advanced one was a second whole room
// in memory after every consume (dwarf, 2026-10-22: the
// OOM killer 20 s after each consumed commit once MIR
// facts made the room several times bigger); a time-travel
// ask for the old seqs builds it again
let at = c.iter().position(|e| e.fold.is_some() && same_room(e));
at.and_then(|i| c.remove(i).fold)
} else {
None
};
let had_fold = stale.is_some();
// a ring view about to fold puts every other ring view of this
// log out FIRST — the new one evicts them when it lands anyway
// (at most one is kept), but until then an old ring view sat
// beside the fold: two dependency universes in memory for the
// six minutes of every post-consume ring refold (dwarf
// 2026-10-22: OOM-killed at 38.7 GB right after each)
if key.1 {
CACHE.lock().unwrap_or_else(|e| e.into_inner()).retain(|e| !(e.identity == identity && e.key.1));
}
ENTRIES_AT_FOLD.store(CACHE.lock().unwrap_or_else(|e| e.into_inner()).iter().filter(|e| e.identity == identity && e.key.1 == key.1).count(), std::sync::atomic::Ordering::Relaxed);
let mut advanced = false;
let mut kept: Option<(Arc<Snapshot>, Option<snapshot::FoldState>)> = None;
if let Some(mut st) = stale {
if snapshot::advance(log, &resolved, &mut st).await? {
let (snap, fold) = snapshot::assemble(log, resolved.clone(), st, true).await?;
kept = Some((Arc::new(snap), fold));
advanced = true;
}
// a refused advance half-applied: the fold is dropped here
}
let (snap, fold) = match kept {
Some(k) => k,
None => {
// a ring view keeps no fold unless asked: its advance is
// refused whenever the universe manifest moved, which a
// consume that dumps dependencies does every time (live:
// "no kept ring fold" after every consume), so the kept
// fold was a second copy of the dependency universe's
// records held for nothing
let keep = at_head && (!resolved.universe || keeps_ring_fold());
let (s, f) = snapshot::build_kept(log, resolved, keep).await?;
(Arc::new(s), f)
}
};
// the ring view's build decision, NAMED (the 160 s live refolds
// on dwarf 2026-10-20 were silent: nothing said whether a kept
// fold existed or why it did not carry)
if key.1 {
let b = &snap.stats;
if advanced {
eprintln!("mimir-room: ring view ADVANCED by the lake's delta in {} ms (delta {} ms) — {}", b.total_ms, b.fold_ms, snap.scope);
} else if had_fold {
eprintln!("mimir-room: ring view REFOLDED in {} ms (fold {} ms) — a kept fold was there, the advance refused (reason above) — {}", b.total_ms, b.fold_ms, snap.scope);
} else if !at_head {
eprintln!("mimir-room: ring view folded afresh in {} ms (fold {} ms) — time travel, nothing kept — {}", b.total_ms, b.fold_ms, snap.scope);
} else {
eprintln!("mimir-room: ring view folded afresh in {} ms (fold {} ms) — no kept ring fold — {}", b.total_ms, b.fold_ms, snap.scope);
}
}
let tick = TICK.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let mut c = CACHE.lock().unwrap_or_else(|e| e.into_inner());
c.retain(|e| !(e.identity == identity && e.key == key));
// at most ONE ring view (dependency ring on) is kept: it is the
// big one (1.9 GB for dwarf's 21 repos, the current room 360 MB);
// a newer ring view puts the older out, whatever the count cap
if key.1 {
c.retain(|e| !e.key.1);
}
let ring_used = snap.scope.universe || snap.ring_lookup.is_some();
c.push(Entry { identity: identity.clone(), stamp, key: key.clone(), manifest: snap.manifest, ring_used, bin_revs: revs, snap: snap.clone(), fold, used: tick, used_at: Instant::now() });
// the ring view is PINNED and does not count against the cap:
// count pressure evicts the least recently used NON-ring entry,
// and the non-ring rooms keep the FULL pool (a pinned ring view
// inside the cap taxed every other room by one slot and threw
// kept pane rooms out early). The ring view (minutes to refold;
// its kept fold is what the advance carries) leaves only via a
// newer ring view or invalidation. Live 2026-10-20: the panes'
// per-repo rooms cycled it out between asks and every ring ask
// refolded the whole dependency universe.
while c.iter().filter(|e| !e.key.1).count() > CACHE_CAP.load(std::sync::atomic::Ordering::Relaxed).max(1) {
let Some(oldest) = c.iter().enumerate().filter(|(_, e)| !e.key.1).min_by_key(|(_, e)| e.used).map(|(i, _)| i) else { break };
c.remove(oldest);
}
Ok((snap, false, advanced))
}
.await;
drop(held);
{
// the gate goes with its last user
let mut b = BUILDING.lock().unwrap_or_else(|e| e.into_inner());
if b.get(&(identity.clone(), key.clone())).is_some_and(|g| Arc::strong_count(g) == 2) {
b.remove(&(identity, key));
}
}
let (snap, cached, advanced) = built?;
Ok(Some(Opened { room: Room { snap }, cached, advanced, ms: t.elapsed().as_millis() as u64 }))
}
/// How many snapshots the process keeps (least recently used go first).
pub fn set_cache_capacity(n: usize) {
CACHE_CAP.store(n.max(1), std::sync::atomic::Ordering::Relaxed);
}
/// [`Self::set_cache_capacity`] from `$MIMIR_ROOM_CACHE_CAP` when it is a
/// number (a box short of memory keeps fewer rooms); returns the cap.
pub fn cache_capacity_from_env() -> usize {
if let Some(n) = std::env::var(CACHE_CAP_ENV).ok().and_then(|v| v.trim().parse::<usize>().ok()) {
Self::set_cache_capacity(n);
}
CACHE_CAP.load(std::sync::atomic::Ordering::Relaxed)
}
/// Entries of the same log and ring flag the cache still held when the
/// last fold started (a test hook: an at-head fold drops the old head's
/// entries first, so this is what a fold has to share memory with).
#[doc(hidden)]
pub fn entries_at_last_fold() -> usize {
ENTRIES_AT_FOLD.load(std::sync::atomic::Ordering::Relaxed)
}
/// **Put out a ring view nobody asked for in `idle`.** The ring view is
/// the big room (the whole dependency universe beside every repo) and
/// only the boundary lenses read it: kept between consumes it was the
/// difference between the server's 10 GB and its 25 GB. A server calls
/// this on its loop tick; the next ring ask folds it again. Returns how
/// many entries went.
pub fn evict_idle_ring(idle: std::time::Duration) -> usize {
let mut c = CACHE.lock().unwrap_or_else(|e| e.into_inner());
let before = c.len();
c.retain(|e| !(e.key.1 && e.used_at.elapsed() >= idle));
before - c.len()
}
/// Does a ring view keep its fold to advance? Default off
/// (`MIMIR_RING_KEEP_FOLD=1` turns it on at start).
pub fn set_ring_keeps_fold(on: bool) {
RING_FOLD.store(if on { 2 } else { 1 }, std::sync::atomic::Ordering::Relaxed);
}
/// The idle time [`Room::evict_idle_ring`] uses on a server:
/// `MIMIR_RING_IDLE_SECS` (default 600; 0 = never put out).
pub fn ring_idle_from_env() -> Option<std::time::Duration> {
let secs = std::env::var("MIMIR_RING_IDLE_SECS").ok().and_then(|v| v.trim().parse::<u64>().ok()).unwrap_or(600);
(secs > 0).then(|| std::time::Duration::from_secs(secs))
}
/// How many snapshots the cache holds now (ring view included).
pub fn cached_count() -> usize {
CACHE.lock().unwrap_or_else(|e| e.into_inner()).len()
}
/// Drop the cached ring definitions (the next ring-off open with linked
/// ring items folds the ring again).
pub fn clear_ring_cache() {
ring::clear();
}
/// The HEAD scope of the kept ring view's fold OF THIS LOG, when the
/// process holds one (`None` otherwise) — what a server opens right
/// after a consume so the universe fold ADVANCES on its own lane and the
/// next ring ask is a cache hit instead of a refold of the whole
/// dependency universe. Keyed by the log's identity: a process serves
/// several workspaces and another log's ring view is not this one's.
pub async fn kept_universe_scope(log: &EventLog) -> Result<Option<Scope>> {
let identity = log.identity().await?;
let c = CACHE.lock().unwrap_or_else(|e| e.into_inner());
Ok(c.iter()
.find(|e| e.identity == identity && e.key.1 && (e.fold.is_some() || !keeps_ring_fold()))
.map(|e| Scope { repos: e.key.0.keys().map(|r| (r.clone(), None)).collect(), universe: true }))
}
/// [`Room::kept_universe_scope`] for a caller with no async context (see
/// [`Room::at_blocking`]).
pub fn kept_universe_scope_blocking(log: &EventLog) -> Result<Option<Scope>> {
std::thread::scope(|s| {
s.spawn(|| log.block_on(Self::kept_universe_scope(log)))
.join()
.unwrap_or_else(|p| std::panic::resume_unwind(p))
})
}
/// Build without touching the cache (benches: the cold path).
pub async fn build_uncached(log: &EventLog, scope: &Scope) -> Result<Room> {
let log = &log.pin().await?;
let resolved = scope.resolve(log).await?;
Ok(Room { snap: Arc::new(snapshot::build(log, resolved).await?) })
}
/// Drop every cached snapshot.
pub fn clear_cache() {
CACHE.lock().unwrap_or_else(|e| e.into_inner()).clear();
}
pub fn scope(&self) -> &ResolvedScope {
&self.snap.scope
}
/// Number of nodes (ordinals are `0..len`).
pub fn len(&self) -> usize {
self.snap.nodes.len()
}
pub fn is_empty(&self) -> bool {
self.snap.nodes.is_empty()
}
pub fn ordinal(&self, moniker: &str) -> Option<u64> {
self.snap.nodes.names.find(moniker)
}
/// The node(s) `id` names: the exact moniker, or — when `id` is a STABLE
/// id (the ingest's deltas key by them since mimir-schema 5, versions as
/// `*`) — every node whose moniker has that stable form.
pub fn resolve(&self, id: &str) -> Vec<u64> {
match self.ordinal(id) {
Some(o) => vec![o],
None => self.stable_ordinals(id),
}
}
pub fn moniker(&self, ordinal: u64) -> &str {
self.snap.nodes.names.get(ordinal)
}
/// Monikers of `ordinals`, in order.
pub fn monikers(&self, ordinals: &[u64]) -> Vec<&str> {
ordinals.iter().map(|o| self.moniker(*o)).collect()
}
pub fn node(&self, ordinal: u64) -> NodeView<'_> {
let t = &self.snap.nodes;
let i = ordinal as usize;
fn s(v: u32, list: &[String]) -> Option<&str> {
(v != NONE).then(|| list[v as usize].as_str())
}
let f = t.flags[i];
NodeView {
ordinal,
moniker: t.names.get(ordinal),
kind: s(t.kind[i], &t.kinds),
level: (t.level[i] != NONE8).then(|| Level::ALL[t.level[i] as usize]),
krate: s(t.krate[i], &t.crates),
version: s(t.version[i], &t.versions),
repo: s(t.repo[i], &t.repos),
origin: (t.origin[i] != NONE8).then(|| Origin::ALL[t.origin[i] as usize]),
provenance: t.provenance(i),
name: s(t.name[i], &t.strs),
parent: (t.parent[i] != NONE).then(|| t.parent[i] as u64),
file: (t.span_file[i] != NONE).then(|| t.files[t.span_file[i] as usize].path.as_str()),
start_line: (t.span_file[i] != NONE).then(|| t.span_start[i]),
end_line: (t.span_file[i] != NONE).then(|| t.span_end[i]),
is_test: f & F_TEST != 0,
ty: s(t.ty[i], &t.strs),
cfg: s(t.cfg[i], &t.strs),
visibility: t.extra_str(i, "visibility"),
effective_visibility: t.extra_str(i, "effective_visibility").and_then(EffectiveVisibility::parse),
doc: t.extra_str(i, "doc"),
has_doc: t.extra_bool(i, "has_doc"),
defined: f & (F_DEFINED | F_UNIVERSE_DEF) != 0,
stub_only: f & F_SRC != 0 && f & (F_DEFINED | F_UNIVERSE_DEF) == 0,
binary_node: f & F_BIN_NODE != 0,
linked: f & F_SYMBOL != 0,
endpoint_only: f & (F_SRC | F_BIN_NODE | F_SYMBOL) == 0,
}
}
/// What the room holds.
pub fn stats(&self) -> RoomStats {
let s = &self.snap;
let t = &s.nodes;
let mut provenance = BTreeMap::new();
// counted by id, named once (no String per node)
let mut by_kind = vec![0u64; t.kinds.len() + 1];
for i in 0..t.len() {
if let Some(p) = t.provenance(i) {
*provenance.entry(p).or_default() += 1;
}
by_kind[if t.kind[i] == NONE { t.kinds.len() } else { t.kind[i] as usize }] += 1;
}
let mut node_kinds: BTreeMap<String, u64> = BTreeMap::new();
for (k, n) in by_kind.into_iter().enumerate().filter(|(_, n)| *n > 0) {
*node_kinds.entry(if k == t.kinds.len() { "(none)".to_string() } else { t.kinds[k].clone() }).or_default() += n;
}
RoomStats {
scope: s.scope.to_string(),
nodes: t.len() as u64,
edges: s.edge_counts.clone(),
provenance,
node_kinds,
endpoint_only: s.endpoint_only,
repo_rows: s.repo_rows.clone(),
universe_rows: s.universe_rows,
unknown_kinds: s.unknown_kinds.clone(),
inputs: s.inputs.clone(),
build: s.stats.clone(),
cached_matrices: s.graphs.lock().unwrap_or_else(|e| e.into_inner()).cached(),
spans: t.span_file.iter().filter(|f| **f != NONE).count() as u64,
files: t.files.len() as u64,
sites: s.sites.len() as u64,
orphan_spans: s.orphan_spans,
tests: t.flags.iter().filter(|f| **f & F_TEST != 0).count() as u64,
symbols: s.symbols.len() as u64,
duplicate_keys: s.dups.len() as u64,
extra_columns: t.extras.iter().map(|e| e.name.clone()).collect(),
unread_columns: s.unread_columns.clone(),
}
}
/// Which build of the room this is: unique per fold in this process, the
/// same for every room the cache serves from that fold. An answer
/// computed from a room is valid exactly as long as the room's version
/// is the one a new open returns.
pub fn version(&self) -> u64 {
self.snap.version
}
/// The lake state this room is 1:1 with — STABLE across processes (the
/// log identity, the scope and the event tables' snapshot ids), unlike
/// [`Room::version`]'s per-process counter. What an answer kept in the
/// lake is keyed by: a fresh process's counter restarts at 1 and would
/// read an older state's rows.
pub fn state_key(&self) -> &str {
&self.snap.state_key
}
/// The identity of the dependency universe this room folded in: the
/// `universe_manifest` snapshot and the ring owners its scan read (the
/// DEPENDS_ON closure). Equal keys = the same ring rows — every ring
/// write moves the manifest, and the closure names the owners read —
/// so what a reader computed over the universe half still holds.
/// `None` = the ring is off.
pub fn universe_key(&self) -> Option<u64> {
if !self.snap.scope.universe {
return None;
}
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
let m = self.snap.manifest.map_or([0xffu8; 9], |m| {
let mut b = [0u8; 9];
b[..8].copy_from_slice(&m.to_le_bytes());
b
});
for b in m.iter().chain(self.snap.ring_closure.unwrap_or(u64::MAX).to_le_bytes().iter()) {
h = (h ^ *b as u64).wrapping_mul(0x100_0000_01b3);
}
Some(h)
}
/// The nodes an edge row of a LOCAL repo (not the universe ring)
/// leaves, sorted: every out-edge a delta of the consumed repos can
/// move starts at one of them — the ring's own edges move only with
/// [`Room::universe_key`].
pub fn local_out(&self) -> Vec<u64> {
let f = &self.snap.nodes.flags;
(0..f.len()).filter(|&i| f[i] & snapshot::F_LOCAL_OUT != 0).map(|i| i as u64).collect()
}
/// The `(from, to)` pairs of `kind` leaving the sorted, distinct
/// ordinals `rows` — the rows' entries only, never the whole matrix.
pub fn out_pairs(&self, kind: EdgeKind, rows: &[u64]) -> Result<Vec<(u64, u64)>> {
let g = self.graphs();
match g.a.get(&kind) {
Some(m) => Ok(m.row_edges(rows)?),
None => Ok(Vec::new()),
}
}
/// The identity of the log this room was folded from: which lake —
/// two workspaces' rooms never share it, whatever their repos are called.
pub fn log_identity(&self) -> &str {
&self.snap.identity
}
/// Same snapshot (the cache served both).
pub fn same_snapshot(&self, other: &Room) -> bool {
Arc::ptr_eq(&self.snap, &other.snap)
}
}