src/code_room/status.rs
//! What a workspace's code room holds and what its consumer is doing — the
//! state every status surface reports (the `Repopulate` report, the job
//! ledger, `workspace status`): offsets, ledger rows, feeds, what is behind,
//! what is in flight, and why nothing runs when nothing does.
//!
//! Read through ONE pinned state of the lake, so a status taken while a
//! release lands never shows an offset without its ledger row.
use std::collections::BTreeMap;
use mimir_ingest::{FeedRow, Offset, RunRow};
use serde_json::{Value, json};
use super::consumer_loop::{InFlight, Member, feed_json, short};
use super::{CodeRoom, Refusal, State};
/// One member's row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberRow {
pub member: String,
/// Is there a clone under `git/` to consume from?
pub has_source: bool,
/// The commit the server's clone is at.
pub head: String,
pub offset: Option<Offset>,
/// No offset, or the offset is not the head.
pub behind: bool,
/// The last ledger row (`ok` / `partial` / `error` / `skipped_docs_only`).
pub last_run: Option<RunRow>,
/// Ledger rows of the member in all.
pub runs: usize,
/// Commits whose code did not fully compile (`partial` / `error` rows).
pub partial: usize,
/// Per feed: its last ledger row.
pub feeds: BTreeMap<String, FeedRow>,
/// Why the consumer does not try this member again, when it does not: its
/// runs keep failing for a reason outside its commits (`MemberStatus::Blocked`).
pub blocked: Option<String>,
}
impl MemberRow {
/// `skade: seq 7 @ 0b07d30f37 → head 0b07d30f37 (ok)` / `… (never consumed)`.
pub fn line(&self) -> String {
// a run that did not go well says why: the ledger row's own sentence
let last = match &self.last_run {
None => "no run".to_string(),
Some(r) => match r.error.as_deref().and_then(|e| e.lines().next()) {
Some(why) => format!("{}: {why}", r.status),
None => r.status.clone(),
},
};
let line = self.line_of(&last);
match &self.blocked {
Some(why) => format!("{line} — BLOCKED, {why}"),
None => line,
}
}
fn line_of(&self, last: &str) -> String {
match &self.offset {
Some(o) if !self.behind => format!("{}: seq {} @ {} ({last})", self.member, o.seq, short(&o.sha)),
Some(o) => format!("{}: seq {} @ {} → head {} ({last})", self.member, o.seq, short(&o.sha), short(&self.head)),
None if !self.has_source => format!("{}: no clone yet", self.member),
None => format!("{}: never consumed → head {}", self.member, short(&self.head)),
}
}
pub fn to_json(&self) -> Value {
json!({
"member": self.member,
"has_source": self.has_source,
"head": self.head,
"offset": self.offset.as_ref().map(|o| json!({ "seq": o.seq, "sha": o.sha, "branch": o.branch })),
"behind": self.behind,
"runs": self.runs,
"partial": self.partial,
"last_run": self.last_run.as_ref().map(|r| json!({
"seq": r.seq, "sha": r.sha, "status": r.status, "crates_dumped": r.crates_dumped,
"rows_plus": r.rows_plus, "rows_minus": r.rows_minus, "ms": r.ms, "error": r.error,
})),
"feeds": self.feeds.values().map(feed_json).collect::<Vec<_>>(),
"blocked": self.blocked,
})
}
}
/// The code room of one workspace, as of one committed state of its lake.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Status {
pub workspace: String,
/// The lake's identity (its event tables' UUIDs): a wiped and re-created
/// lake is another one.
pub identity: String,
pub members: Vec<MemberRow>,
/// What the consumer is doing right now.
pub in_flight: Option<InFlight>,
/// Why the last attempt to consume could not start, if it could not.
pub refusal: Option<Refusal>,
}
impl Status {
/// The members with something to consume.
pub fn behind(&self) -> Vec<&MemberRow> {
self.members.iter().filter(|m| m.behind && m.has_source).collect()
}
/// `member@seq` of every member that has an offset — the registry's
/// `current_snapshot` (the same form as `ConsumeReport::stamp`).
pub fn stamp(&self) -> String {
self.members.iter().filter_map(|m| m.offset.as_ref().map(|o| format!("{}@{}", m.member, o.seq))).collect::<Vec<_>>().join(",")
}
/// `lake 3f9a… · skade@7 0b07d30f37 · znippy@12 9c1e…` — which state is served.
pub fn generation(&self) -> String {
let id = self.identity.split(',').next().unwrap_or_default();
let mut parts = vec![format!("lake {}", &id[..id.len().min(8)])];
parts.extend(self.members.iter().filter_map(|m| m.offset.as_ref().map(|o| format!("{}@{} {}", m.member, o.seq, short(&o.sha)))));
parts.join(" · ")
}
pub fn to_json(&self) -> Value {
json!({
"workspace": self.workspace,
"identity": self.identity,
"generation": self.generation(),
"members": self.members.iter().map(MemberRow::to_json).collect::<Vec<_>>(),
"behind": self.behind().iter().map(|m| m.member.as_str()).collect::<Vec<_>>(),
"in_flight": self.in_flight.as_ref().map(InFlight::line),
"refusal": self.refusal.as_ref().map(Refusal::to_json),
})
}
}
/// The room over every consumed repo at its offset, as a status line: what a
/// reader gets RIGHT NOW (also while the consumer is mid-commit).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoomLine {
/// `skade@7 (0b07d30f37) · universe off`
pub scope: String,
pub nodes: u64,
pub edges: u64,
/// Served from the process cache (nothing landed since it was built).
pub cached: bool,
/// Wall clock of the open.
pub ms: u64,
}
impl RoomLine {
/// `room 988 856 nodes / 1 204 113 edges at heimdal@1 (7087c3202f) (opened in 812 ms)`
pub fn line(&self) -> String {
format!(
"room {} node(s) / {} edge(s) at {} ({} in {} ms)",
self.nodes,
self.edges,
self.scope.split(" · ").next().unwrap_or(&self.scope),
if self.cached { "cached" } else { "built" },
self.ms
)
}
}
impl CodeRoom {
/// Open the room every reader of this workspace gets now — every consumed
/// repo at its offset — and say what it holds. `None`: nothing was
/// consumed yet (there is no room to speak of, which is not an empty one).
pub fn room_line(&self) -> Result<Option<RoomLine>, Refusal> {
let log = self.log()?;
let internal = |e: anyhow::Error| Refusal::new(State::Internal, format!("code room of `{}`: {e:#}", self.workspace));
if log.block_on(log.offsets()).map_err(internal)?.is_empty() {
return Ok(None);
}
let opened = mimir_room::Room::open_blocking(&log, &mimir_room::Scope::current()).map_err(internal)?;
let stats = opened.room.stats();
Ok(Some(RoomLine { scope: stats.scope, nodes: stats.nodes, edges: stats.edges.values().sum(), cached: opened.cached, ms: opened.ms }))
}
/// The status of `members` (see [`Status`]). Never waits for the consumer.
pub fn status(&self, members: &[Member]) -> Result<Status, Refusal> {
let log = self.log()?;
let internal = |e: anyhow::Error| Refusal::new(State::Internal, format!("code room of `{}`: {e:#}", self.workspace));
let (identity, rows) = log
.block_on(async {
let pinned = log.pin().await?;
let offsets = pinned.offsets().await?;
let mut rows = Vec::new();
for m in members {
let runs = pinned.runs(&m.name).await?;
let offset = offsets.get(&m.name).cloned();
rows.push(MemberRow {
member: m.name.clone(),
has_source: m.has_source(),
head: m.head.clone(),
behind: offset.as_ref().is_none_or(|o| o.sha != m.head),
offset,
partial: runs.iter().filter(|r| r.status == mimir_ingest::schema::STATUS_PARTIAL || r.status == mimir_ingest::schema::STATUS_ERROR).count(),
runs: runs.len(),
last_run: runs.last().cloned(),
feeds: pinned.feed_last(&m.name).await?,
blocked: self.blocked(members, &m.name),
});
}
anyhow::Ok((pinned.identity().await?, rows))
})
.map_err(internal)?;
Ok(Status { workspace: self.workspace.clone(), identity, members: rows, in_flight: self.in_flight(), refusal: self.last_refusal() })
}
}
/// The v1 tables Mímir v2 orphaned (PLAN-MIMIR-V2 §7.5: not migrated, never
/// read again): the nine the `syn` / rust-analyzer roads wrote, and the jera
/// chain's ledgers. Every `lens_<name>` result table goes with them.
pub const V1_TABLES: [&str; 11] = [
"symbol_facts",
"call_edges",
"feature_gate_facts",
"scip_occurrences",
"test_inventory",
"warehouse_access_edges",
"architecture_wiring",
"knowledge_scans",
"semantic_clusters",
"lens_runs",
"feed_runs",
];
/// Is `table` one of the orphaned v1 tables?
pub fn is_v1_table(table: &str) -> bool {
V1_TABLES.contains(&table) || table.starts_with("lens_")
}
/// The v1 tables still in `wh`'s catalog.
pub fn v1_tables_present(wh: &crate::warehouse::iceberg::IcebergWarehouse) -> anyhow::Result<Vec<String>> {
let mut names: Vec<String> = wh.table_names()?.into_iter().filter(|t| is_v1_table(t)).collect();
names.sort();
Ok(names)
}
/// Drop the orphaned v1 tables that are still there (wiring map I6). Returns
/// what was dropped; a second call drops nothing. The Mímir v2 lake is another
/// catalog and is not touched.
pub fn drop_v1_tables(wh: &crate::warehouse::iceberg::IcebergWarehouse) -> anyhow::Result<Vec<String>> {
let present = v1_tables_present(wh)?;
if present.is_empty() {
return Ok(present);
}
wh.drop_tables(&present)
}