src/warehouse/yggdrasil.rs
//! Yggdrasil — the **infra crown** (`infra_nodes` / `infra_edges`) + the
//! **runtime band** (`runtime_spans`) of the one super-graph.
//!
//! `yggdrasil-supergraph.md` describes the warehouse as one tree whose Iceberg
//! tables are bands of nodes/edges at every scale — instruction → symbol → call
//! → crate → repo → dependency → architecture, and, at the CROWN, **where the
//! code actually runs**: the machines / services / VMs / containers of the
//! constellation, plus the general runtime traces threaded through them. That
//! crown + runtime band was the last "MISSING ⛔" scale; this module lands it.
//!
//! Persistence mirrors the dependency-graph snapshot (`warehouse::dep_graph`):
//! each `record_*` call writes ONE snapshot (all rows share a fresh
//! `snapshot_id` + one `ts_micros`), scoped by `workspace_name`; each `query_*`
//! reads them back grouped into whole snapshots. It reuses the EXISTING
//! `IcebergWarehouse` (skade's warehouse) — no schema owned outside nornir, no
//! new skade dependency: the warehouse already stores arbitrary tables.
//!
//! **Real producer (`infra_from_jobs` / `spans_from_jobs`):** the jera
//! `JobRecord` ledger. Every long-running nornir op is a durable ledger row that
//! already carries *which node ran it* (`JobRecord::node`, the fleet machine)
//! and its lifecycle timing. From that we derive:
//! * a **machine** infra-node per distinct `node`, a **service** infra-node
//! per distinct job `kind`, and a `runs_on` **infra-edge** for every
//! `(service kind → machine)` pair that co-occurred;
//! * a **runtime span** per job (`span_id = job_id`, `parent = parent_id`,
//! `machine = node`, `dur_us = elapsed_ms * 1000`).
//!
//! So the crown is not dead schema — it is populated straight from the ledger
//! that jera's multi-machine scheduler already stamps.
use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{
Array, BooleanArray, Int32Array, Int64Array, RecordBatch, StringArray,
TimestampMicrosecondArray,
};
use chrono::Utc;
use iceberg::Catalog;
use iceberg::arrow::schema_to_arrow_schema;
use uuid::Uuid;
use super::iceberg::{
IcebergWarehouse, TABLE_INFRA_EDGES, TABLE_INFRA_NODES, TABLE_RUNTIME_SPANS, append_batch,
ensure_table_schema,
};
use crate::jobs::JobRecord;
/// A node in the infra crown — a machine, service, VM, or container.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InfraNode {
/// Stable id (host name for a machine, `service:<kind>` for a service, …).
pub id: String,
/// `machine` | `service` | `vm` | `container`.
pub kind: String,
/// The host this lives on (a machine hosts itself).
pub host: String,
/// Worker-group / cluster this belongs to.
pub group: String,
/// CPU cores (0 when unknown — the ledger doesn't carry roster specs).
pub cores: i32,
/// Memory in MiB (0 when unknown).
pub mem_mb: i64,
/// Free-form role label (`worker` / `control-plane` / the service kind).
pub role: String,
/// Whether it was seen live in this snapshot.
pub online: bool,
}
/// An edge in the infra crown (`runs_on` / `hosts` / `deploys` / `member_of`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InfraEdge {
pub from: String,
pub to: String,
pub rel: String,
}
/// One executed span in the runtime band (a jera job or a server op).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeSpan {
pub span_id: String,
/// Parent span id (`""` = a root/top-level span).
pub parent: String,
/// The service that ran it (the job `kind`).
pub service: String,
/// What it acted on (the job `target`).
pub op: String,
pub start_micros: i64,
/// Duration in microseconds (0 when the op is still running / untimed).
pub dur_us: i64,
pub job_id: String,
/// The fleet machine that executed it (`local` when unstamped).
pub machine: String,
pub status: String,
}
/// Machine placeholder for a job whose `node` is unset (ran on the local
/// control-plane, not dispatched across the fleet).
pub const LOCAL_MACHINE: &str = "local";
/// Default worker-group when the ledger carries no cluster label.
pub const CONSTELLATION_GROUP: &str = "constellation";
/// Derive the infra crown (nodes + edges) from a slice of jera job records —
/// the **real producer**. Pure and warehouse-free, so it unit-tests on its own.
///
/// * one `machine` node per distinct `node` (unset → [`LOCAL_MACHINE`]),
/// * one `service` node per distinct job `kind`,
/// * a `runs_on` edge `service → machine` for each `(kind, node)` that ran,
/// * a `member_of` edge `machine → group` per machine.
///
/// Nodes/edges are de-duplicated and returned in a stable (sorted) order so the
/// snapshot is deterministic.
pub fn infra_from_jobs(jobs: &[JobRecord]) -> (Vec<InfraNode>, Vec<InfraEdge>) {
let mut machines: BTreeSet<String> = BTreeSet::new();
let mut services: BTreeSet<String> = BTreeSet::new();
// (service_kind, machine) pairs that co-occurred.
let mut runs_on: BTreeSet<(String, String)> = BTreeSet::new();
for j in jobs {
let machine = j
.node
.clone()
.filter(|n| !n.is_empty())
.unwrap_or_else(|| LOCAL_MACHINE.to_string());
machines.insert(machine.clone());
if !j.kind.is_empty() {
services.insert(j.kind.clone());
runs_on.insert((j.kind.clone(), machine));
}
}
let mut nodes: Vec<InfraNode> = Vec::new();
for m in &machines {
nodes.push(InfraNode {
id: m.clone(),
kind: "machine".to_string(),
host: m.clone(),
group: CONSTELLATION_GROUP.to_string(),
cores: 0,
mem_mb: 0,
role: if m == LOCAL_MACHINE {
"control-plane".to_string()
} else {
"worker".to_string()
},
online: true,
});
}
for s in &services {
nodes.push(InfraNode {
id: format!("service:{s}"),
kind: "service".to_string(),
host: String::new(),
group: CONSTELLATION_GROUP.to_string(),
cores: 0,
mem_mb: 0,
role: s.clone(),
online: true,
});
}
let mut edges: Vec<InfraEdge> = Vec::new();
for (kind, machine) in &runs_on {
edges.push(InfraEdge {
from: format!("service:{kind}"),
to: machine.clone(),
rel: "runs_on".to_string(),
});
}
for m in &machines {
edges.push(InfraEdge {
from: m.clone(),
to: CONSTELLATION_GROUP.to_string(),
rel: "member_of".to_string(),
});
}
edges.sort_by(|a, b| (&a.from, &a.to, &a.rel).cmp(&(&b.from, &b.to, &b.rel)));
(nodes, edges)
}
/// Derive one runtime span per job record — the **real producer** for the
/// runtime band. Pure/warehouse-free. Spans are returned sorted by
/// `(start_micros, span_id)` so the snapshot is deterministic.
pub fn spans_from_jobs(jobs: &[JobRecord]) -> Vec<RuntimeSpan> {
let mut spans: Vec<RuntimeSpan> = jobs
.iter()
.map(|j| RuntimeSpan {
span_id: j.job_id.clone(),
parent: j.parent_id.clone().unwrap_or_default(),
service: j.kind.clone(),
op: j.target.clone(),
start_micros: j.ts_start_micros,
dur_us: j.elapsed_ms.map(|ms| ms.saturating_mul(1000)).unwrap_or(0),
job_id: j.job_id.clone(),
machine: j
.node
.clone()
.filter(|n| !n.is_empty())
.unwrap_or_else(|| LOCAL_MACHINE.to_string()),
status: j.status.clone(),
})
.collect();
spans.sort_by(|a, b| (a.start_micros, &a.span_id).cmp(&(b.start_micros, &b.span_id)));
spans
}
// ─── persistence (suite warehouse) ────────────────────────────────────────
/// Append one infra-node snapshot. All rows share the returned `snapshot_id`
/// and a single `ts_micros`. Empty `nodes` writes no snapshot.
pub async fn record_infra_nodes(
wh: &IcebergWarehouse,
workspace_name: &str,
nodes: &[InfraNode],
) -> Result<Uuid> {
let snapshot_id = Uuid::new_v4();
if nodes.is_empty() {
return Ok(snapshot_id);
}
let ts = Utc::now().timestamp_micros();
let sid = snapshot_id.to_string();
let n = nodes.len();
let ident = wh.table_ident(TABLE_INFRA_NODES);
let table = wh.catalog().load_table(&ident).await?;
let table = ensure_table_schema(
wh.catalog(),
&ident,
table,
&super::iceberg_schema::infra_nodes()?,
)
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![sid.as_str(); n])),
Arc::new(StringArray::from(vec![workspace_name; n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
Arc::new(StringArray::from(
nodes.iter().map(|x| x.id.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
nodes.iter().map(|x| x.kind.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
nodes.iter().map(|x| x.host.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
nodes.iter().map(|x| x.group.clone()).collect::<Vec<_>>(),
)),
Arc::new(Int32Array::from(
nodes.iter().map(|x| x.cores).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
nodes.iter().map(|x| x.mem_mb).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
nodes.iter().map(|x| x.role.clone()).collect::<Vec<_>>(),
)),
Arc::new(BooleanArray::from(
nodes.iter().map(|x| x.online).collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), table, batch).await?;
Ok(snapshot_id)
}
/// Append one infra-edge snapshot. Empty `edges` writes no snapshot.
pub async fn record_infra_edges(
wh: &IcebergWarehouse,
workspace_name: &str,
edges: &[InfraEdge],
) -> Result<Uuid> {
let snapshot_id = Uuid::new_v4();
if edges.is_empty() {
return Ok(snapshot_id);
}
let ts = Utc::now().timestamp_micros();
let sid = snapshot_id.to_string();
let n = edges.len();
let ident = wh.table_ident(TABLE_INFRA_EDGES);
let table = wh.catalog().load_table(&ident).await?;
let table = ensure_table_schema(
wh.catalog(),
&ident,
table,
&super::iceberg_schema::infra_edges()?,
)
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![sid.as_str(); n])),
Arc::new(StringArray::from(vec![workspace_name; n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
Arc::new(StringArray::from(
edges.iter().map(|x| x.from.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
edges.iter().map(|x| x.to.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
edges.iter().map(|x| x.rel.clone()).collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), table, batch).await?;
Ok(snapshot_id)
}
/// Append one runtime-span snapshot. Empty `spans` writes no snapshot.
pub async fn record_runtime_spans(
wh: &IcebergWarehouse,
workspace_name: &str,
spans: &[RuntimeSpan],
) -> Result<Uuid> {
let snapshot_id = Uuid::new_v4();
if spans.is_empty() {
return Ok(snapshot_id);
}
let ts = Utc::now().timestamp_micros();
let sid = snapshot_id.to_string();
let n = spans.len();
let ident = wh.table_ident(TABLE_RUNTIME_SPANS);
let table = wh.catalog().load_table(&ident).await?;
let table = ensure_table_schema(
wh.catalog(),
&ident,
table,
&super::iceberg_schema::runtime_spans()?,
)
.await?;
let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![sid.as_str(); n])),
Arc::new(StringArray::from(vec![workspace_name; n])),
Arc::new(TimestampMicrosecondArray::from(vec![ts; n]).with_timezone("+00:00")),
Arc::new(StringArray::from(
spans.iter().map(|x| x.span_id.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
spans.iter().map(|x| x.parent.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
spans.iter().map(|x| x.service.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
spans.iter().map(|x| x.op.clone()).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
spans.iter().map(|x| x.start_micros).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
spans.iter().map(|x| x.dur_us).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
spans.iter().map(|x| x.job_id.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
spans.iter().map(|x| x.machine.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
spans.iter().map(|x| x.status.clone()).collect::<Vec<_>>(),
)),
];
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), table, batch).await?;
Ok(snapshot_id)
}
/// Read every infra node for `workspace_name`, sorted by `(kind, id)`.
pub async fn query_infra_nodes(
wh: &IcebergWarehouse,
workspace_name: &str,
) -> Result<Vec<InfraNode>> {
let filter = skade::ScanFilter::eq("workspace_name", workspace_name);
let cols = [
"workspace_name",
"node_id",
"kind",
"host",
"node_group",
"cores",
"mem_mb",
"role",
"online",
];
let batches =
super::iceberg::load_and_read_filtered(wh, TABLE_INFRA_NODES, &filter, &cols).await?;
let mut out: Vec<InfraNode> = Vec::new();
for b in &batches {
let ws = col::<StringArray>(b, "workspace_name")?;
let id = col::<StringArray>(b, "node_id")?;
let kind = col::<StringArray>(b, "kind")?;
let host = col::<StringArray>(b, "host")?;
let group = col::<StringArray>(b, "node_group")?;
let cores = col::<Int32Array>(b, "cores")?;
let mem = col::<Int64Array>(b, "mem_mb")?;
let role = col::<StringArray>(b, "role")?;
let online = col::<BooleanArray>(b, "online")?;
for i in 0..b.num_rows() {
// Residual guard: pushdown prunes at file granularity, not per row.
if ws.value(i) != workspace_name {
continue;
}
out.push(InfraNode {
id: id.value(i).to_string(),
kind: kind.value(i).to_string(),
host: host.value(i).to_string(),
group: group.value(i).to_string(),
cores: cores.value(i),
mem_mb: mem.value(i),
role: role.value(i).to_string(),
online: online.value(i),
});
}
}
out.sort_by(|a, b| (&a.kind, &a.id).cmp(&(&b.kind, &b.id)));
Ok(out)
}
/// Read every infra edge for `workspace_name`, sorted by `(rel, from, to)`.
pub async fn query_infra_edges(
wh: &IcebergWarehouse,
workspace_name: &str,
) -> Result<Vec<InfraEdge>> {
let filter = skade::ScanFilter::eq("workspace_name", workspace_name);
let cols = ["workspace_name", "from_node", "to_node", "rel"];
let batches =
super::iceberg::load_and_read_filtered(wh, TABLE_INFRA_EDGES, &filter, &cols).await?;
let mut out: Vec<InfraEdge> = Vec::new();
for b in &batches {
let ws = col::<StringArray>(b, "workspace_name")?;
let from = col::<StringArray>(b, "from_node")?;
let to = col::<StringArray>(b, "to_node")?;
let rel = col::<StringArray>(b, "rel")?;
for i in 0..b.num_rows() {
if ws.value(i) != workspace_name {
continue;
}
out.push(InfraEdge {
from: from.value(i).to_string(),
to: to.value(i).to_string(),
rel: rel.value(i).to_string(),
});
}
}
out.sort_by(|a, b| (&a.rel, &a.from, &a.to).cmp(&(&b.rel, &b.from, &b.to)));
Ok(out)
}
/// Read every runtime span for `workspace_name`, sorted by `(start_micros, span_id)`.
pub async fn query_runtime_spans(
wh: &IcebergWarehouse,
workspace_name: &str,
) -> Result<Vec<RuntimeSpan>> {
let filter = skade::ScanFilter::eq("workspace_name", workspace_name);
let cols = [
"workspace_name",
"span_id",
"parent_span",
"service",
"op",
"start_micros",
"dur_us",
"job_id",
"machine",
"status",
];
let batches =
super::iceberg::load_and_read_filtered(wh, TABLE_RUNTIME_SPANS, &filter, &cols).await?;
let mut out: Vec<RuntimeSpan> = Vec::new();
for b in &batches {
let ws = col::<StringArray>(b, "workspace_name")?;
let span_id = col::<StringArray>(b, "span_id")?;
let parent = col::<StringArray>(b, "parent_span")?;
let service = col::<StringArray>(b, "service")?;
let op = col::<StringArray>(b, "op")?;
let start = col::<Int64Array>(b, "start_micros")?;
let dur = col::<Int64Array>(b, "dur_us")?;
let job_id = col::<StringArray>(b, "job_id")?;
let machine = col::<StringArray>(b, "machine")?;
let status = col::<StringArray>(b, "status")?;
for i in 0..b.num_rows() {
if ws.value(i) != workspace_name {
continue;
}
out.push(RuntimeSpan {
span_id: span_id.value(i).to_string(),
parent: parent.value(i).to_string(),
service: service.value(i).to_string(),
op: op.value(i).to_string(),
start_micros: start.value(i),
dur_us: dur.value(i),
job_id: job_id.value(i).to_string(),
machine: machine.value(i).to_string(),
status: status.value(i).to_string(),
});
}
}
out.sort_by(|a, b| (a.start_micros, &a.span_id).cmp(&(b.start_micros, &b.span_id)));
Ok(out)
}
/// One-call **producer + persist**: derive the infra crown + runtime band from
/// the jera job ledger and write all three snapshots. Returns the three
/// `snapshot_id`s `(infra_nodes, infra_edges, runtime_spans)`. This is the seam
/// a monitor / `nornir yggdrasil ingest` calls with `JobStore::list(&All)`.
pub async fn ingest_from_jobs(
wh: &IcebergWarehouse,
workspace_name: &str,
jobs: &[JobRecord],
) -> Result<(Uuid, Uuid, Uuid)> {
let (nodes, edges) = infra_from_jobs(jobs);
let spans = spans_from_jobs(jobs);
let node_snap = record_infra_nodes(wh, workspace_name, &nodes).await?;
let edge_snap = record_infra_edges(wh, workspace_name, &edges).await?;
let span_snap = record_runtime_spans(wh, workspace_name, &spans).await?;
Ok((node_snap, edge_snap, span_snap))
}
/// Downcast a projected column **by name** (a `.select` scan reorders/drops
/// columns, so positional access would read the wrong array). Mirrors
/// `dep_graph::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"))
}
/// Sync wrapper: derive + persist the Yggdrasil infra/runtime scales from a job
/// ledger on this warehouse's private runtime. Returns the three snapshot ids.
impl IcebergWarehouse {
pub fn record_yggdrasil_from_jobs(
&self,
workspace_name: &str,
jobs: &[JobRecord],
) -> Result<(Uuid, Uuid, Uuid)> {
self.block_on(ingest_from_jobs(self, workspace_name, jobs))
}
}
#[cfg(test)]
mod tests {
use super::*;
/// A job record with the fields the producer reads set explicitly.
fn job(
job_id: &str,
kind: &str,
target: &str,
node: Option<&str>,
parent: Option<&str>,
elapsed_ms: Option<i64>,
status: &str,
start: i64,
) -> JobRecord {
JobRecord {
job_id: job_id.to_string(),
kind: kind.to_string(),
target: target.to_string(),
workspace: "ws".to_string(),
status: status.to_string(),
ts_start_micros: start,
ts_end_micros: elapsed_ms.map(|_| start + 1),
elapsed_ms,
detail_json: String::new(),
result_ref: String::new(),
parent_id: parent.map(|s| s.to_string()),
node: node.map(|s| s.to_string()),
phase: None,
actor: None,
actor_method: None,
}
}
fn sample_jobs() -> Vec<JobRecord> {
vec![
job(
"j1",
"release",
"nornir",
Some("loki"),
None,
Some(1200),
"done",
1000,
),
job(
"j2",
"bench",
"nornir",
Some("odin"),
Some("j1"),
Some(50),
"done",
2000,
),
job(
"j3",
"release",
"holger",
Some("loki"),
None,
Some(900),
"failed",
3000,
),
// Unstamped node → LOCAL_MACHINE; still-running → dur 0.
job("j4", "docs", "skade", None, None, None, "running", 4000),
]
}
#[test]
fn producer_derives_machines_services_edges_and_spans() {
let jobs = sample_jobs();
let (nodes, edges) = infra_from_jobs(&jobs);
// Machines: loki, odin, local (j4 unstamped). Services: release, bench, docs.
let machines: BTreeSet<&str> = nodes
.iter()
.filter(|n| n.kind == "machine")
.map(|n| n.id.as_str())
.collect();
assert_eq!(machines, ["loki", "local", "odin"].into_iter().collect());
let services: BTreeSet<&str> = nodes
.iter()
.filter(|n| n.kind == "service")
.map(|n| n.id.as_str())
.collect();
assert_eq!(
services,
["service:release", "service:bench", "service:docs"]
.into_iter()
.collect()
);
// `release` ran on BOTH loki and odin? no — release on loki (j1,j3), bench on
// odin (j2), docs on local (j4). Assert the runs_on edges precisely.
let runs_on: BTreeSet<(&str, &str)> = edges
.iter()
.filter(|e| e.rel == "runs_on")
.map(|e| (e.from.as_str(), e.to.as_str()))
.collect();
assert_eq!(
runs_on,
[
("service:release", "loki"),
("service:bench", "odin"),
("service:docs", "local"),
]
.into_iter()
.collect()
);
// Every machine is member_of the constellation group.
let member_of: BTreeSet<&str> = edges
.iter()
.filter(|e| e.rel == "member_of" && e.to == CONSTELLATION_GROUP)
.map(|e| e.from.as_str())
.collect();
assert_eq!(member_of, ["loki", "odin", "local"].into_iter().collect());
// Spans: one per job, machine + dur derived.
let spans = spans_from_jobs(&jobs);
assert_eq!(spans.len(), 4);
let j2 = spans.iter().find(|s| s.span_id == "j2").unwrap();
assert_eq!(j2.machine, "odin");
assert_eq!(j2.parent, "j1");
assert_eq!(j2.dur_us, 50_000, "elapsed_ms 50 → 50_000 us");
let j4 = spans.iter().find(|s| s.span_id == "j4").unwrap();
assert_eq!(j4.machine, "local", "unstamped node → local");
assert_eq!(j4.dur_us, 0, "running job has no duration yet");
assert_eq!(j4.parent, "", "no parent → empty");
}
/// RED-when-broken warehouse round-trip: record the ledger-derived crown +
/// runtime band, read all three back, and assert field-identical to what the
/// pure producer built. A wrong column mapping, a dropped row, or a lost
/// numeric value flips an `assert_eq!`.
#[test]
fn ledger_scales_round_trip_field_identical() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let jobs = sample_jobs();
let (mut want_nodes, want_edges) = infra_from_jobs(&jobs);
let mut want_spans = spans_from_jobs(&jobs);
wh.record_yggdrasil_from_jobs("wsA", &jobs).unwrap();
// Nodes round-trip (sorted the same way the query sorts).
want_nodes.sort_by(|a, b| (&a.kind, &a.id).cmp(&(&b.kind, &b.id)));
let got_nodes = wh.block_on(query_infra_nodes(&wh, "wsA")).unwrap();
assert_eq!(
got_nodes, want_nodes,
"infra_nodes byte/field-identical round-trip"
);
// Edges round-trip.
let mut want_edges = want_edges;
want_edges.sort_by(|a, b| (&a.rel, &a.from, &a.to).cmp(&(&b.rel, &b.from, &b.to)));
let got_edges = wh.block_on(query_infra_edges(&wh, "wsA")).unwrap();
assert_eq!(
got_edges, want_edges,
"infra_edges byte/field-identical round-trip"
);
// Spans round-trip — the numeric dur_us / start_micros survive exactly.
want_spans.sort_by(|a, b| (a.start_micros, &a.span_id).cmp(&(b.start_micros, &b.span_id)));
let got_spans = wh.block_on(query_runtime_spans(&wh, "wsA")).unwrap();
assert_eq!(
got_spans, want_spans,
"runtime_spans byte/field-identical round-trip"
);
// Mutation check: a DIFFERENT workspace sees none of wsA's rows.
assert!(
wh.block_on(query_infra_nodes(&wh, "wsB"))
.unwrap()
.is_empty()
);
assert!(
wh.block_on(query_infra_edges(&wh, "wsB"))
.unwrap()
.is_empty()
);
assert!(
wh.block_on(query_runtime_spans(&wh, "wsB"))
.unwrap()
.is_empty()
);
// A specific numeric round-trips (guards a Long→Int truncation regression).
let bench_span = got_spans.iter().find(|s| s.span_id == "j2").unwrap();
assert_eq!(bench_span.dur_us, 50_000);
assert_eq!(bench_span.machine, "odin");
}
#[test]
fn empty_ledger_writes_no_rows() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
wh.record_yggdrasil_from_jobs("empty", &[]).unwrap();
assert!(
wh.block_on(query_infra_nodes(&wh, "empty"))
.unwrap()
.is_empty()
);
assert!(
wh.block_on(query_infra_edges(&wh, "empty"))
.unwrap()
.is_empty()
);
assert!(
wh.block_on(query_runtime_spans(&wh, "empty"))
.unwrap()
.is_empty()
);
}
}