src/warehouse/dep_graph.rs
//! nornir's warehouse-side of the cross-repo dependency graph.
//!
//! The compute + query + layout **core now lives in the shared `nornir-depgraph`
//! edda leaf** (so nornir AND dwarves render the same suite dependency view);
//! this module RE-EXPORTS it and keeps the two nornir-only halves that need
//! nornir's own subsystems:
//! * [`build`] — the `cargo_metadata` builder that resolves a
//! [`WorkspaceDescriptor`] through nornir's `workspace::resolve` (the leaf's
//! descriptor-free [`WorkspaceGraph::build_from_members`] is the other path).
//! * [`record_dep_graph`] / [`query_dep_graph_snapshots`] — persistence of a
//! [`DepGraphSnapshot`] into the suite warehouse's `dep_graph_edges` Iceberg
//! table.
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use chrono::Utc;
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use uuid::Uuid;
pub use nornir_depgraph::{
CrossRepoEdge, DepGraphSnapshot, DepKind, RepoFacts, WorkspaceGraph, inspect_repo,
parallel_map, topo_order_from_edges,
};
use super::iceberg::IcebergWarehouse;
use crate::workspace::descriptor::WorkspaceDescriptor;
/// Build the cross-repo graph by probing every repo the descriptor resolves with
/// `cargo metadata --no-deps` ([`inspect_repo`]), then assembling the graph via
/// [`WorkspaceGraph::from_facts`].
///
/// Each repo's probe is independent, so they fan across cores through the shared
/// [`parallel_map`] pool (the dependency-Mímir build hot path,
/// `nornir.dep_graph_build`). The pool sizes itself to `available_parallelism()`
/// and honours a positive `RAYON_NUM_THREADS` cap, so the bench's single-core
/// (`_st`) pass collapses to a serial loop and stays a genuine single-core figure
/// while the multi-core arm saturates. `from_facts` gets the same
/// `(name → RepoFacts)` map the old serial loop produced, so the graph is
/// identical. The `cargo metadata`-free twin is [`WorkspaceGraph::build_from_members`].
pub fn build(desc: &WorkspaceDescriptor) -> Result<WorkspaceGraph> {
let resolved = crate::workspace::resolve::resolve_sources(desc)?;
let entries: Vec<(String, PathBuf)> = resolved.into_iter().collect();
let probed = parallel_map(&entries, |(name, root)| {
inspect_repo(name, root).map(|f| (name.clone(), f))
});
let mut facts: BTreeMap<String, RepoFacts> = BTreeMap::new();
for probe in probed {
let (name, f) = probe?;
facts.insert(name, f);
}
let graph = WorkspaceGraph::from_facts(facts)?;
// ── test-matrix self-report (L4) ────────────────────────────────────────
//
// The dependency Mímir answers blast-radius questions, so the answer everyone
// downstream trusts is only as complete as its MEMBER SET. `resolve_sources`
// is allowed to yield fewer entries than the descriptor declares (a repo
// whose source cannot be resolved simply does not appear), and every query
// from there on is silently scoped to whatever survived — `affected_by_change`
// would return a smaller, entirely plausible re-run set.
//
// Falsifiable by construction: the descriptor is the declared truth and the
// graph is what was built, so the row goes RED exactly when a declared member
// is missing from the graph, and names it. Emitted after a successful build
// only; the `?` paths above return before it rather than reporting green.
let declared = desc.repos.len();
let built = graph.facts.len();
let missing: Vec<&str> = desc
.repos
.keys()
.filter(|r| !graph.facts.contains_key(*r))
.map(String::as_str)
.collect();
nornir_testmatrix::functional_status(
"nornir/dep-graph",
"every_declared_member_resolved",
missing.is_empty(),
&format!(
"workspace={} declared={declared} built={built} edges={} missing={missing:?}",
desc.workspace.name,
graph.edges.len()
),
);
Ok(graph)
}
// ─── persistence (suite warehouse) ───────────────────────────────────────
/// Append a graph snapshot to the warehouse. Returns the snapshot UUID.
/// Edges with no `via` crates (shouldn't happen — guarded by build)
/// are written as a single placeholder row to preserve the edge.
pub async fn record_dep_graph(
wh: &IcebergWarehouse,
workspace_name: &str,
graph: &WorkspaceGraph,
) -> Result<Uuid> {
let snapshot_id = Uuid::new_v4();
let ts = Utc::now();
let id_str = snapshot_id.to_string();
let mut snapshot_ids = Vec::new();
let mut ws_names = Vec::new();
let mut ts_vals: Vec<i64> = Vec::new();
let mut from_repos = Vec::new();
let mut to_repos = Vec::new();
let mut via_crates = Vec::new();
let mut kinds: Vec<String> = Vec::new();
for e in &graph.edges {
if e.via.is_empty() {
// Unresolvable/placeholder edge: keep the `from→to` topology and its
// `kind` label as a SINGLE placeholder row (empty `via_crate`) so the
// reader reconstructs it as a via-less edge rather than dropping it.
snapshot_ids.push(id_str.clone());
ws_names.push(workspace_name.to_string());
ts_vals.push(ts.timestamp_micros());
from_repos.push(e.from.clone());
to_repos.push(e.to.clone());
via_crates.push(String::new());
kinds.push(e.kind.as_str().to_string());
continue;
}
for via in &e.via {
snapshot_ids.push(id_str.clone());
ws_names.push(workspace_name.to_string());
ts_vals.push(ts.timestamp_micros());
from_repos.push(e.from.clone());
to_repos.push(e.to.clone());
via_crates.push(via.clone());
kinds.push(e.kind.as_str().to_string());
}
}
if snapshot_ids.is_empty() {
// Empty snapshot — still record a no-edges marker.
snapshot_ids.push(id_str);
ws_names.push(workspace_name.to_string());
ts_vals.push(ts.timestamp_micros());
from_repos.push(String::new());
to_repos.push(String::new());
via_crates.push(String::new());
kinds.push(DepKind::Normal.as_str().to_string());
}
let table = wh
.catalog()
.load_table(&wh.table_ident(super::iceberg::TABLE_DEP_GRAPH_EDGES))
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let mut cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(snapshot_ids)),
Arc::new(StringArray::from(ws_names)),
Arc::new(TimestampMicrosecondArray::from(ts_vals).with_timezone("+00:00")),
Arc::new(StringArray::from(from_repos)),
Arc::new(StringArray::from(to_repos)),
Arc::new(StringArray::from(via_crates)),
];
// Only append the `kind` column when the (possibly pre-existing) table schema
// actually carries it — a warehouse created before kinds were tracked has a
// 6-column `dep_graph_edges`, and `arrow_schema` mirrors the table.
if table
.metadata()
.current_schema()
.field_by_name("kind")
.is_some()
{
cols.push(Arc::new(StringArray::from(kinds)));
}
// ECOSYSTEM AXIS — which package manager's graph this edge belongs to. Same
// presence guard as `kind`. Every edge here is resolved from `cargo metadata`,
// so `cargo` is the correct value, not a placeholder; a Maven/Conan crust
// writes its own `Ecosystem::name()`. Deliberately NOT the knowledge tables'
// `language` column — Java and Kotlin share `maven`, and a Rust cdylib
// consumed through `conan` is `rust` in a non-cargo ecosystem.
// See `workspace_nordisk/.nornir/mimir-language-axis-design.md` §2.2.
if table
.metadata()
.current_schema()
.field_by_name("ecosystem")
.is_some()
{
let n = cols.first().map(|c| c.len()).unwrap_or(0);
let eco = vec![super::iceberg_schema::ECOSYSTEM_DEFAULT.to_string(); n];
cols.push(Arc::new(StringArray::from(eco)));
}
let batch = RecordBatch::try_new(arrow_schema, cols)?;
super::iceberg::append_batch(wh.catalog(), table, batch).await?;
Ok(snapshot_id)
}
/// Read every dep-graph snapshot row for a workspace, optionally
/// limited to the most recent N snapshots (after grouping).
pub async fn query_dep_graph_snapshots(
wh: &IcebergWarehouse,
workspace_name: &str,
limit: Option<usize>,
) -> Result<Vec<DepGraphSnapshot>> {
// Push the workspace filter into the scan and project only the columns we
// rebuild. `kind` was added later: try WITH kind, fall back WITHOUT — an old
// (6-col) snapshot then reads every edge as `Normal` (its prior behavior).
let filter = skade::ScanFilter::eq("workspace_name", workspace_name);
let base_cols = [
"snapshot_id",
"workspace_name",
"ts_micros",
"from_repo",
"to_repo",
"via_crate",
];
let mut with_kind: Vec<&str> = base_cols.to_vec();
with_kind.push("kind");
let (batches, has_kind): (Vec<RecordBatch>, bool) =
match super::iceberg::load_and_read_filtered(
wh,
super::iceberg::TABLE_DEP_GRAPH_EDGES,
&filter,
&with_kind,
)
.await
{
Ok(b) => (b, true),
Err(_) => (
super::iceberg::load_and_read_filtered(
wh,
super::iceberg::TABLE_DEP_GRAPH_EDGES,
&filter,
&base_cols,
)
.await?,
false,
),
};
// (snapshot_id, ts) → (workspace_name, BTreeMap<(from,to), (BTreeSet<via>, kind)>)
let mut by_snapshot: BTreeMap<
(Uuid, i64),
(
String,
BTreeMap<(String, String), (BTreeSet<String>, DepKind)>,
),
> = BTreeMap::new();
for batch in &batches {
let ids = col::<StringArray>(batch, "snapshot_id")?;
let wss = col::<StringArray>(batch, "workspace_name")?;
let tss = col::<TimestampMicrosecondArray>(batch, "ts_micros")?;
let froms = col::<StringArray>(batch, "from_repo")?;
let tos = col::<StringArray>(batch, "to_repo")?;
let vias = col::<StringArray>(batch, "via_crate")?;
let kinds = if has_kind {
col::<StringArray>(batch, "kind").ok()
} else {
None
};
for i in 0..batch.num_rows() {
// Residual guard: pushdown prunes at file/row-group granularity.
if wss.value(i) != workspace_name {
continue;
}
let uid = Uuid::parse_str(ids.value(i))?;
let key = (uid, tss.value(i));
let entry = by_snapshot
.entry(key)
.or_insert_with(|| (wss.value(i).to_string(), BTreeMap::new()));
let f = froms.value(i).to_string();
let t = tos.value(i).to_string();
// Kind column is optional (null on a row written before kinds) → Normal.
let kind = kinds
.filter(|k| !k.is_null(i))
.map(|k| DepKind::from_str_lenient(k.value(i)))
.unwrap_or(DepKind::Normal);
if !f.is_empty() || !t.is_empty() {
// Init the edge as `Dev` (most restrictive); ANY order-gating row
// promotes it to `Normal` — the normal edge wins for ordering (#8).
let slot = entry
.1
.entry((f, t))
.or_insert_with(|| (BTreeSet::new(), DepKind::Dev));
let via = vias.value(i);
if !via.is_empty() {
slot.0.insert(via.to_string());
}
if kind.is_order_gating() {
slot.1 = DepKind::Normal;
}
}
}
}
let mut out: Vec<DepGraphSnapshot> = by_snapshot
.into_iter()
.map(|((snapshot_id, ts_micros), (ws, edge_map))| {
let edges = edge_map
.into_iter()
.map(|((from, to), (via, kind))| CrossRepoEdge {
from,
to,
via,
kind,
})
.collect();
let timestamp = chrono::TimeZone::timestamp_micros(&Utc, ts_micros)
.single()
.unwrap_or_else(Utc::now);
DepGraphSnapshot {
snapshot_id,
workspace_name: ws,
timestamp,
edges,
}
})
.collect();
out.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
if let Some(n) = limit {
let drop_n = out.len().saturating_sub(n);
out.drain(..drop_n);
}
Ok(out)
}
/// Downcast a column **by name** — required once a scan uses `.select`,
/// since projection reorders/drops columns and positional access would
/// read the wrong array. Mirrors `index::snapshot::col`.
fn col<'a, T: 'static>(batch: &'a RecordBatch, name: &str) -> Result<&'a T> {
batch
.column_by_name(name)
.ok_or_else(|| anyhow!("projected batch missing column `{name}`"))?
.as_any()
.downcast_ref::<T>()
.ok_or_else(|| anyhow!("column `{name}` has unexpected arrow type"))
}
#[cfg(test)]
mod parity_tests {
use super::*;
use crate::workspace::descriptor::{RepoSpec, WorkspaceDescriptor, WorkspaceMeta};
/// Write a minimal member crate `<root>/<dir>/` with the given package name,
/// `publish` flag, and path-dep crate names.
fn member(root: &Path, dir: &str, pkg: &str, publish_false: bool, path_deps: &[&str]) {
let crate_dir = root.join(dir);
std::fs::create_dir_all(crate_dir.join("src")).unwrap();
let mut toml =
format!("[package]\nname = \"{pkg}\"\nversion = \"0.1.0\"\nedition = \"2021\"\n");
if publish_false {
toml.push_str("publish = false\n");
}
if !path_deps.is_empty() {
toml.push_str("\n[dependencies]\n");
for d in path_deps {
toml.push_str(&format!("{d} = {{ path = \"../{d}\" }}\n"));
}
}
std::fs::write(crate_dir.join("Cargo.toml"), toml).unwrap();
std::fs::write(crate_dir.join("src/lib.rs"), "// fixture\n").unwrap();
}
fn resolved(root: &Path, members: &[&str]) -> BTreeMap<String, PathBuf> {
members
.iter()
.map(|m| (m.to_string(), root.join(m)))
.collect()
}
fn edge_set(g: &WorkspaceGraph) -> BTreeSet<(String, String, Vec<String>)> {
g.edges
.iter()
.map(|e| {
(
e.from.clone(),
e.to.clone(),
e.via.iter().cloned().collect::<Vec<_>>(),
)
})
.collect()
}
/// Parity gate: on the SAME on-disk fixture, the `cargo metadata` graph
/// ([`build`]) and the direct-manifest graph ([`WorkspaceGraph::build_from_members`])
/// agree on produces + cross-repo edges. Skipped gracefully when `cargo
/// metadata` can't run (no cargo / offline).
#[test]
fn manifest_graph_matches_cargo_metadata_graph() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path();
member(root, "app", "app", false, &["liba"]);
member(root, "liba", "liba", false, &["util"]);
member(root, "util", "util", false, &[]);
let names = ["app", "liba", "util"];
let mut repos = BTreeMap::new();
for m in names {
repos.insert(
m.to_string(),
RepoSpec {
path: Some(root.join(m).to_string_lossy().into_owned()),
git: None,
branch: None,
// `release`/`publish` arrived in edda `29159ef` ("five release
// opt-outs in the live descriptor were parsed and thrown on the
// floor"). `None` is the descriptor's own default for both, so
// this fixture keeps meaning exactly what it meant before.
release: None,
publish: None,
},
);
}
let desc = WorkspaceDescriptor {
workspace: WorkspaceMeta {
name: "fixture".into(),
deep_scan: false,
},
repos,
descriptor_dir: root.to_path_buf(),
};
let meta_graph = match build(&desc) {
Ok(g) => g,
Err(e) => {
eprintln!("skip parity: cargo metadata unavailable: {e:#}");
return;
}
};
let manifest_graph = WorkspaceGraph::build_from_members(&resolved(root, &names)).unwrap();
assert_eq!(
manifest_graph.component_names(),
meta_graph.component_names(),
"same component set"
);
for m in names {
assert_eq!(
manifest_graph.facts.get(m).map(|f| &f.produces),
meta_graph.facts.get(m).map(|f| &f.produces),
"produces parity for {m}"
);
}
assert_eq!(
edge_set(&manifest_graph),
edge_set(&meta_graph),
"cross-repo edge parity"
);
}
}
#[cfg(test)]
mod warehouse_roundtrip_tests {
use super::*;
use crate::warehouse::iceberg::IcebergWarehouse;
/// A dep-graph snapshot carrying an UNRESOLVABLE edge — one with an empty
/// `via` set — round-trips through the warehouse: the placeholder row keeps
/// the edge's `from→to` topology AND its `kind` label, and the resolved
/// edges alongside it are unaffected.
#[test]
fn placeholder_edge_survives_round_trip_and_keeps_kind_label() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let edges = vec![
CrossRepoEdge::normal("app", "core", ["core_c".to_string()].into_iter().collect()),
CrossRepoEdge {
from: "app".to_string(),
to: "ghost".to_string(),
via: BTreeSet::new(),
kind: DepKind::Dev,
},
];
let graph = WorkspaceGraph::from_query_parts(BTreeMap::new(), edges);
wh.block_on(record_dep_graph(&wh, "ws-placeholder", &graph))
.unwrap();
let snaps = wh
.block_on(query_dep_graph_snapshots(&wh, "ws-placeholder", None))
.unwrap();
assert_eq!(snaps.len(), 1, "one snapshot recorded");
let read: BTreeMap<(String, String), &CrossRepoEdge> = snaps[0]
.edges
.iter()
.map(|e| ((e.from.clone(), e.to.clone()), e))
.collect();
let ghost = read
.get(&("app".to_string(), "ghost".to_string()))
.expect("placeholder edge app→ghost survived the round-trip");
assert!(ghost.via.is_empty(), "placeholder edge has no via crate");
assert_eq!(
ghost.kind,
DepKind::Dev,
"placeholder edge keeps its kind label"
);
let resolved = read
.get(&("app".to_string(), "core".to_string()))
.expect("resolved edge app→core round-trips");
assert_eq!(
resolved.via.iter().cloned().collect::<Vec<_>>(),
vec!["core_c".to_string()]
);
assert_eq!(resolved.kind, DepKind::Normal);
assert!(resolved.kind.is_order_gating());
}
}