src/mimir_lens/world_impact.rs
//! **Lens — `Mimir.WorldImpact`: changes ranked by how far they ripple.**
//!
//! Composes the blast-radius signals nornir owns:
//!
//! - **the CHANGE itself** — [`Change`]: what a repo's last consumed commit did to
//! each of its crates (`Room::diff` between two consumer ordinals): items added /
//! removed / changed, and how many of the removed ones were PUBLIC (rustc's
//! effective visibility). A removed public item is a BREAKING change — the drop
//! that makes downstream callers fail to compile.
//! - **the callers** — the other repos whose code depends on the changed or removed
//! items (`Room::dependents`, the Jikes walk over the whole workspace room).
//! - **`dep_graph_edges`** — the CONSTELLATION blast radius. The workspace's
//! cross-repo edges (`from_repo` consumes `to_repo` via `via_crate`), expanded
//! transitively into the set of workspace repos a change to the crate forces a
//! rebuild of — the same expansion `Mimir.AffectedByChange` /
//! `changed_since_last_release` serve, read from the warehouse table.
//!
//! **The rank** is `world_callers + constellation_dependents`, with both kept as their
//! own columns and breaking changes first — a change to a widely-depended, breaking
//! symbol is the highest world-impact. The composite is a SORT KEY, never a lone
//! verdict.
//!
//! **Honest about reach.** "World" is the consumed workspace: the callers are the
//! repos of this lake. Callers out on crates.io (the world ring) and a published
//! semver diff have no producer (P5 — the coderoom ingest is gone); the lens says so
//! rather than reading zero external callers as "nothing breaks".
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use serde::{Deserialize, Serialize};
use super::rows::Table;
use super::sources::{Feed, Sources, evidence_note};
/// One changed crate, ranked by its world-wide + constellation downstream.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorldChange {
/// The repo and the commit window the change was read from (`repo@seq-1→seq`).
pub world_repo: String,
pub crate_name: String,
pub added_items: i64,
pub removed_items: i64,
pub changed_items: i64,
pub suggested_bump: String,
/// A removed PUBLIC item — the drop that breaks callers.
pub breaking: bool,
/// The other repos of the workspace whose code depends on the changed items.
pub world_callers: Vec<String>,
/// Workspace repos a change to this crate forces a rebuild of (transitive
/// dependents from `dep_graph_edges`) — the constellation blast radius.
pub constellation_dependents: Vec<String>,
/// `world_callers + constellation_dependents`. A SORT KEY, read the two beside it.
pub world_impact: usize,
}
impl WorldChange {
pub fn explain(&self) -> String {
format!(
"{} world caller repo(s) + {} constellation dependent(s) = world-impact {}{}",
self.world_callers.len(),
self.constellation_dependents.len(),
self.world_impact,
if self.breaking {
" (BREAKING: removed public API)"
} else {
""
}
)
}
}
/// The whole reading.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorldImpactReport {
pub changes_seen: usize,
pub breaking_changes: usize,
pub worst: Vec<WorldChange>,
pub feeds: Vec<Feed>,
pub evidence_note: String,
pub model_attribution: String,
}
pub const ATTRIBUTION: &str = "Composition of nornir primitives: the code room's diff \
between two consumer ordinals (the change, with rustc's effective visibility saying \
what was public) + its dependents walk (the callers in the workspace's other repos) + \
the dependency Mimir's affected_by_change blast radius (dep_graph_edges).";
impl WorldImpactReport {
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"kind": "world_impact",
"changes_seen": self.changes_seen,
"breaking_changes": self.breaking_changes,
"evidence_note": self.evidence_note,
"model_attribution": self.model_attribution,
"feeds": self.feeds,
"worst": self.worst.iter().map(|c| serde_json::json!({
"world_repo": c.world_repo,
"crate_name": c.crate_name,
"added_items": c.added_items,
"removed_items": c.removed_items,
"changed_items": c.changed_items,
"suggested_bump": c.suggested_bump,
"breaking": c.breaking,
"world_callers": c.world_callers,
"world_caller_count": c.world_callers.len(),
"constellation_dependents": c.constellation_dependents,
"constellation_count": c.constellation_dependents.len(),
"world_impact": c.world_impact,
"explain": c.explain(),
})).collect::<Vec<_>>(),
})
}
}
/// Normalize a crate name for matching (`-`→`_`, lowercase) — a package and the
/// crate it builds spell the same crate differently otherwise.
fn norm(krate: &str) -> String {
krate.trim().to_ascii_lowercase().replace('-', "_")
}
/// **One repo's commit window (seq N−1 → N), folded once.** The per-crate
/// change (without its callers, which are read from the current room) and the
/// ids each crate changed or removed. A window never changes once computed,
/// so it is kept per (repo, seq) for the workspace's code room ([`Windows`]).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Window {
pub per_crate: Vec<Change>,
/// crate → the monikers of the items it changed or removed.
pub touched: BTreeMap<String, Vec<String>>,
}
/// The windows kept for one workspace, by (repo, seq). A repo's older
/// windows go when a newer one is kept (the lens asks the newest only).
#[derive(Debug, Default)]
pub struct Windows {
kept: BTreeMap<(String, i64), std::sync::Arc<Window>>,
}
impl Windows {
/// Windows kept per repo at most.
const PER_REPO: usize = 2;
pub fn get(&self, repo: &str, seq: i64) -> Option<std::sync::Arc<Window>> {
self.kept.get(&(repo.to_string(), seq)).cloned()
}
pub fn put(&mut self, repo: &str, seq: i64, w: std::sync::Arc<Window>) {
self.kept.insert((repo.to_string(), seq), w);
let mine: Vec<i64> = self.kept.keys().filter(|(r, _)| r == repo).map(|(_, s)| *s).collect();
for s in mine.iter().take(mine.len().saturating_sub(Self::PER_REPO)) {
self.kept.remove(&(repo.to_string(), *s));
}
}
pub fn len(&self) -> usize {
self.kept.len()
}
pub fn is_empty(&self) -> bool {
self.kept.is_empty()
}
}
/// What one repo's last consumed commit changed in ONE of its crates.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Change {
pub repo: String,
/// The commit window (`shop@1→2`).
pub window: String,
pub crate_name: String,
/// Item-level nodes added / removed / whose payload changed.
pub added_items: i64,
pub removed_items: i64,
pub changed_items: i64,
/// Of the added / removed ones, those another crate could name.
pub added_public: i64,
pub removed_public: i64,
/// The OTHER repos of the workspace whose code depends on a changed or
/// removed item.
pub callers: BTreeSet<String>,
}
/// **Rank the changes by workspace-wide + constellation downstream impact.** Pure
/// fold. `sources` carries `dep_graph_edges`; `repo_filter` keeps the changes of one
/// repo or crate (""=all).
pub fn analyze(changes: &[Change], sources: &Sources, feeds: Vec<Feed>, repo_filter: &str, top_n: usize) -> WorldImpactReport {
// ── the constellation dependency edges (dep_graph_edges) ────────────────────
let dep = dep_dependents(sources.get("dep_graph_edges"));
let mut out: Vec<WorldChange> = Vec::new();
for c in changes {
if !repo_filter.is_empty() && c.repo != repo_filter && c.crate_name != repo_filter {
continue;
}
let breaking = c.removed_public > 0;
let suggested_bump = if breaking {
"major"
} else if c.added_public > 0 {
"minor"
} else {
"patch"
};
let world_callers: Vec<String> = c.callers.iter().filter(|r| **r != c.repo).cloned().collect();
let constellation_dependents = dep.transitive_dependents_of_crate(&norm(&c.crate_name));
let world_impact = world_callers.len() + constellation_dependents.len();
out.push(WorldChange {
world_repo: c.window.clone(),
crate_name: c.crate_name.clone(),
added_items: c.added_items,
removed_items: c.removed_items,
changed_items: c.changed_items,
suggested_bump: suggested_bump.to_string(),
breaking,
world_callers,
constellation_dependents,
world_impact,
});
}
let changes_seen = out.len();
let breaking_changes = out.iter().filter(|c| c.breaking).count();
out.sort_by(|a, b| {
b.breaking
.cmp(&a.breaking)
.then_with(|| b.world_impact.cmp(&a.world_impact))
.then_with(|| b.world_callers.len().cmp(&a.world_callers.len()))
.then_with(|| a.world_repo.cmp(&b.world_repo))
.then_with(|| a.crate_name.cmp(&b.crate_name))
});
if top_n > 0 {
out.truncate(top_n);
}
let evidence_note = evidence_note(&feeds, "world impact of changes");
WorldImpactReport {
changes_seen,
breaking_changes,
worst: out,
feeds,
evidence_note,
model_attribution: ATTRIBUTION.to_string(),
}
}
/// The workspace dependency edges, indexed for transitive-dependent expansion.
struct DepIndex {
/// `via_crate` (normalized) → producer repos (`to_repo`) that export it.
producers_of_crate: BTreeMap<String, BTreeSet<String>>,
/// `repo` → repos that DEPEND on it (`from_repo`s whose `to_repo == repo`).
dependents_of_repo: BTreeMap<String, BTreeSet<String>>,
/// Direct consumers of a crate (`from_repo`s whose `via_crate == crate`).
consumers_of_crate: BTreeMap<String, BTreeSet<String>>,
}
impl DepIndex {
/// Transitive workspace dependents of a crate: the direct consumers of the crate,
/// plus everything that transitively depends on them (BFS over reverse edges) and
/// on the crate's producer repos.
fn transitive_dependents_of_crate(&self, crate_norm: &str) -> Vec<String> {
let mut seen: BTreeSet<String> = BTreeSet::new();
let mut q: VecDeque<String> = VecDeque::new();
// seed: direct consumers of the crate.
for c in self
.consumers_of_crate
.get(crate_norm)
.into_iter()
.flatten()
{
if seen.insert(c.clone()) {
q.push_back(c.clone());
}
}
// seed the producer repos too, so their dependents are pulled in.
for p in self
.producers_of_crate
.get(crate_norm)
.into_iter()
.flatten()
{
for d in self.dependents_of_repo.get(p).into_iter().flatten() {
if seen.insert(d.clone()) {
q.push_back(d.clone());
}
}
}
while let Some(repo) = q.pop_front() {
for d in self.dependents_of_repo.get(&repo).into_iter().flatten() {
if seen.insert(d.clone()) {
q.push_back(d.clone());
}
}
}
seen.into_iter().collect()
}
}
/// Parse `dep_graph_edges` into a [`DepIndex`]. `from_repo` consumes `to_repo` via
/// `via_crate`; the latest-snapshot rows are used (a stale generation of edges must
/// not inflate the blast radius).
fn dep_dependents(t: &Table) -> DepIndex {
let t = super::rows::latest_snapshot_rows(t);
let c = t.cursor(&["from_repo", "to_repo", "via_crate"]);
let mut producers_of_crate: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
let mut dependents_of_repo: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
let mut consumers_of_crate: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
for r in c.rows() {
let from = c.get(r, 0).to_string();
let to = c.get(r, 1).to_string();
let via = norm(c.get(r, 2));
if !to.is_empty() && !via.is_empty() {
producers_of_crate
.entry(via.clone())
.or_default()
.insert(to.clone());
}
if !from.is_empty() && !via.is_empty() {
consumers_of_crate
.entry(via)
.or_default()
.insert(from.clone());
}
if !from.is_empty() && !to.is_empty() {
dependents_of_repo.entry(to).or_default().insert(from);
}
}
DepIndex {
producers_of_crate,
dependents_of_repo,
consumers_of_crate,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn deps(rows: &[&[&str]]) -> Sources {
let t = Table::new(
["snapshot_id", "from_repo", "to_repo", "via_crate", "ts_micros"].iter().map(|s| s.to_string()).collect(),
rows.iter().map(|r| r.iter().map(|s| s.to_string()).collect()).collect(),
);
let mut s = Sources::default();
s.add_root("/t", [("dep_graph_edges".to_string(), t)].into());
s
}
fn change(repo: &str, krate: &str, added: i64, removed_public: i64, callers: &[&str]) -> Change {
Change {
repo: repo.into(),
window: format!("{repo}@1→2"),
crate_name: krate.into(),
added_items: added,
removed_items: removed_public,
changed_items: 1,
added_public: added,
removed_public,
callers: callers.iter().map(|c| c.to_string()).collect(),
}
}
fn run(changes: &[Change], s: &Sources, filter: &str) -> WorldImpactReport {
analyze(changes, s, s.feeds(&["dep_graph_edges"]), filter, 10)
}
/// A KNOWN world-impact change: a breaking drop in a widely-called crate outranks a
/// non-breaking change in a crate nobody calls.
#[test]
fn a_breaking_change_to_a_widely_called_crate_ranks_highest() {
let changes = [
// a leaf crate, only added items → not breaking, no callers.
change("leaf", "leaf", 5, 0, &[]),
// widely-used crate, removed a public item → breaking; three repos depend on it.
change("serde", "serde", 0, 3, &["appA", "appB", "appC"]),
];
let r = run(&changes, &deps(&[]), "");
assert_eq!(r.changes_seen, 2);
assert_eq!(r.breaking_changes, 1);
assert_eq!(r.worst[0].crate_name, "serde", "the breaking, widely-called change ranks first");
assert!(r.worst[0].breaking);
assert_eq!((r.worst[0].suggested_bump.as_str(), r.worst[1].suggested_bump.as_str()), ("major", "minor"));
assert_eq!(r.worst[0].world_callers.len(), 3);
assert!(r.worst[0].explain().contains("BREAKING"));
nornir_testmatrix::functional_status(
"nornir/mimir/world-impact",
"breaking_widely_called_ranks_first",
r.worst[0].crate_name == "serde" && r.worst[0].breaking && r.worst[0].world_callers.len() == 3,
&format!("top={} breaking={} callers={}", r.worst[0].crate_name, r.worst[0].breaking, r.worst[0].world_callers.len()),
);
}
/// Removing a PRIVATE item breaks nobody: only a removed public item is breaking.
#[test]
fn a_removed_private_item_is_not_breaking() {
let mut c = change("a", "a", 0, 0, &["b"]);
c.removed_items = 4;
let r = run(&[c], &deps(&[]), "");
assert!(!r.worst[0].breaking);
assert_eq!((r.worst[0].removed_items, r.worst[0].suggested_bump.as_str()), (4, "patch"));
}
/// A repo is never its own caller.
#[test]
fn the_changed_repo_is_not_counted_among_its_callers() {
let r = run(&[change("a", "a", 0, 1, &["a", "b"])], &deps(&[]), "");
assert_eq!(r.worst[0].world_callers, vec!["b".to_string()]);
}
/// The constellation blast radius: a change to a crate transitively rebuilds the
/// workspace repos that depend on it (dep_graph_edges), composed into the impact.
#[test]
fn constellation_dependents_are_expanded_transitively() {
// app → lib (via lib_c), lib → util (via util_c). A change to util_c rebuilds
// lib (direct consumer) and app (transitive).
let s = deps(&[&["s1", "app", "lib", "lib_c", "100"], &["s1", "lib", "util", "util_c", "100"]]);
let r = run(&[change("util", "util-c", 0, 1, &[])], &s, "");
let c = &r.worst[0];
let set: BTreeSet<&str> = c.constellation_dependents.iter().map(|s| s.as_str()).collect();
assert!(set.contains("lib"), "direct consumer lib rebuilds: {set:?}");
assert!(set.contains("app"), "transitive dependent app rebuilds: {set:?}");
assert!(c.world_impact >= 2);
}
#[test]
fn the_filter_keeps_one_repo_or_one_crate() {
let changes = [change("a", "a-core", 1, 0, &[]), change("b", "b", 1, 0, &[])];
assert_eq!(run(&changes, &deps(&[]), "a").changes_seen, 1);
assert_eq!(run(&changes, &deps(&[]), "a-core").changes_seen, 1);
assert_eq!(run(&changes, &deps(&[]), "").changes_seen, 2);
}
#[test]
fn an_empty_dep_graph_reads_as_missing_not_no_impact() {
let r = run(&[], &deps(&[]), "");
assert_eq!(r.changes_seen, 0);
assert!(r.evidence_note.contains("MISSING FEEDS"), "{}", r.evidence_note);
assert!(r.evidence_note.contains("dep_graph_edges"), "{}", r.evidence_note);
}
}