src/code_room/reader/wired/trace.rs
//! **A traced run on the code map** — the room side of the replay.
//!
//! Two halves:
//!
//! * [`RoomTraceMap`]: which room function a traced function IS — by where its
//! code starts (`(repo, file, line)` inside the innermost local fn / method
//! whose span holds it: a closure lands on the function around it, a generic
//! instance on its generic), else by its symbol read as the room's path
//! (`nornir_trace::room_path`: `<T as Trait>::m` → `T::m`, generics dropped)
//! when exactly one local function has that path. The filer
//! (`Telemetry.SubmitStepFeed`) writes the stable id it says onto each event.
//! * the `Viz.CodeMap` lods the replay asks (queued like every map answer):
//!
//! | `lod` | answer |
//! |---|---|
//! | `trace_runs` | the traced runs of `repo` (all when empty), newest first: target, commit, events, threads, span, the share of functions / events the room knew |
//! | `trace` | one run's events in a window (`run_id`, `from_ns`, `to_ns`, paged by `after_seq` / `limit`), each on a HEAD of this map (`heads[i]`: card, `at`) with an entry's CALLER head (the innermost placed function open on its thread, for the pulse along the call spline); `stacks: true` adds each thread's open heads at `from_ns` (for a scrub), `heat: true` the run's total time per head |
//! | `profile` | the newest `fn_profile` of `repo` (or `run_id`): self / total time per head |
//!
//! An event of a function the room does not know (std, a registry crate) is
//! counted, never placed: `unplaced` says how many.
use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex};
use mimir_room::{MaskSpec, NameMatch, NameQuery, Room};
use mimir_schema::{Origin, kind as nk};
use nornir_trace::wire::FiledFn;
use nornir_trace::EventKind;
use serde_json::{Value, json};
use super::codemap::MapLayout;
use super::{internal, invalid, text};
use crate::code_room::{Reach, Refusal};
use crate::warehouse::fn_trace;
/// The room's local functions by `(repo, file)`: `(start, end, ordinal)`.
struct SpanIndex {
version: u64,
by_file: HashMap<(String, String), Vec<(u32, u32, u64)>>,
}
fn span_index(room: &Room) -> Result<Arc<SpanIndex>, Refusal> {
static KEPT: Mutex<Vec<Arc<SpanIndex>>> = Mutex::new(Vec::new());
let v = room.version();
if let Some(i) = KEPT.lock().unwrap_or_else(|e| e.into_inner()).iter().find(|i| i.version == v) {
return Ok(i.clone());
}
let fns = room
.mask(&MaskSpec { origins: vec![Origin::Local], defined_only: true, node_kinds: vec![nk::FN.into(), nk::METHOD.into()], ..Default::default() })
.map_err(internal)?
.value;
let mut by_file: HashMap<(String, String), Vec<(u32, u32, u64)>> = HashMap::new();
for &o in fns.ordinals().iter() {
let n = room.node(o);
let (Some(repo), Some(file), Some(line)) = (n.repo, n.file, n.start_line) else { continue };
let end = n.end_line.unwrap_or(line).max(line);
by_file.entry((repo.to_string(), file.to_string())).or_default().push((line, end, o));
}
let idx = Arc::new(SpanIndex { version: v, by_file });
let mut kept = KEPT.lock().unwrap_or_else(|e| e.into_inner());
kept.retain(|i| i.version != v);
kept.push(idx.clone());
if kept.len() > 4 {
kept.remove(0);
}
Ok(idx)
}
/// How a traced function found its room function.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum How {
/// Its source location lies in the function's span.
Span,
/// Its symbol, read as a path, names exactly one local function.
Path,
}
/// **Map traced functions to room functions** (ordinal + how), `None` = not in
/// this room.
pub fn map_fns(room: &Room, fns: &[FiledFn]) -> Result<Vec<Option<(u64, How)>>, Refusal> {
let idx = span_index(room)?;
let mut by_path: HashMap<String, Option<u64>> = HashMap::new();
Ok(fns
.iter()
.map(|f| {
if !f.repo.is_empty() && f.line > 0 {
let hit = idx.by_file.get(&(f.repo.clone(), f.file.clone())).and_then(|c| {
c.iter()
.filter(|(s, e, _)| *s <= f.line && f.line <= *e)
.max_by(|a, b| a.0.cmp(&b.0).then(b.1.cmp(&a.1)))
.map(|&(_, _, o)| o)
});
if let Some(o) = hit {
return Some((o, How::Span));
}
}
let path = nornir_trace::room_path(&f.symbol);
if !path.contains("::") {
return None;
}
let o = *by_path.entry(path.clone()).or_insert_with(|| {
let q = NameQuery { text: path, mode: NameMatch::Path, node_kinds: vec![nk::FN.into(), nk::METHOD.into()], defined_only: true, ..Default::default() };
room.lookup(&q).ok().map(|a| a.value).filter(|v| v.len() == 1).map(|v| v[0])
});
o.map(|o| (o, How::Path))
})
.collect())
}
/// The server's [`crate::warehouse::step_feed::RoomMap`]: the room of the
/// workspace a filing lands in.
pub struct RoomTraceMap {
pub room: Room,
}
impl crate::warehouse::step_feed::RoomMap for RoomTraceMap {
fn stable_ids(&self, fns: &[FiledFn]) -> Vec<String> {
match map_fns(&self.room, fns) {
Ok(m) => m
.into_iter()
.map(|o| o.map(|(o, _)| mimir_schema::moniker::stable(self.room.moniker(o)).into_owned()).unwrap_or_default())
.collect(),
Err(_) => vec![String::new(); fns.len()],
}
}
}
fn warehouse<'a>(reach: &Reach<'a>) -> Result<&'a crate::warehouse::iceberg::IcebergWarehouse, Refusal> {
reach.warehouse.ok_or_else(|| Refusal::new(crate::code_room::State::NotWired, "no warehouse in reach: traced runs live in the workspace's warehouse"))
}
fn share(n: u64, of: u64) -> Value {
if of == 0 { Value::Null } else { json!(((n as f64 / of as f64) * 1000.0).round() / 1000.0) }
}
/// `lod: trace_runs`.
pub fn runs(reach: &Reach<'_>, p: &Value) -> Result<Value, Refusal> {
let wh = warehouse(reach)?;
let repo = text(p, "repo");
let runs = wh.block_on(fn_trace::query_runs(wh, repo)).map_err(internal)?;
Ok(json!({
"lod": "trace_runs",
"repo": repo,
"runs": runs.iter().map(|r| json!({
"run_id": r.run_id, "repo": r.repo, "commit": r.commit, "target": r.target, "source": r.source,
"ts_micros": r.ts_micros, "chunks": r.chunks, "chunks_filed": r.chunks_filed,
"events": r.events, "total_events": r.total_events, "truncated": r.truncated,
"threads": r.threads, "duration_ns": r.duration_ns,
"functions": r.functions, "mapped_functions": r.mapped_functions,
"local_functions": r.local_functions, "mapped_local_functions": r.mapped_local_functions,
"mapped_events": r.mapped_events,
"mapped_share": share(r.mapped_functions, r.functions),
"mapped_local_share": share(r.mapped_local_functions, r.local_functions),
"mapped_event_share": share(r.mapped_events, r.events),
})).collect::<Vec<_>>(),
}))
}
/// Where a stable id stands on this map: (card, head) of its first local node
/// that has a card.
fn place(room: &Room, map: &MapLayout, stable_id: &str) -> Option<(u32, [f32; 2])> {
room.resolve(stable_id).into_iter().find_map(|o| map.card_of.get(&o).map(|&c| (c, map.head_at(c))))
}
/// Interns heads for one answer.
struct Heads<'a> {
room: &'a Room,
map: &'a MapLayout,
of: HashMap<String, Option<usize>>,
list: Vec<Value>,
}
impl<'a> Heads<'a> {
fn new(room: &'a Room, map: &'a MapLayout) -> Self {
Self { room, map, of: HashMap::new(), list: Vec::new() }
}
fn get(&mut self, stable_id: &str, symbol: &str) -> Option<usize> {
if stable_id.is_empty() {
return None;
}
if let Some(h) = self.of.get(stable_id) {
return *h;
}
let h = place(self.room, self.map, stable_id).map(|(card, at)| {
let o = self.room.resolve(stable_id)[0];
self.list.push(json!({ "card": card, "at": at, "stable_id": stable_id, "path": self.room.display_path(o), "symbol": symbol }));
self.list.len() - 1
});
self.of.insert(stable_id.to_string(), h);
h
}
}
/// `lod: trace`.
pub fn window(reach: &Reach<'_>, room: &Room, map: &MapLayout, p: &Value) -> Result<Value, Refusal> {
let wh = warehouse(reach)?;
let run_id = text(p, "run_id");
if run_id.is_empty() {
return Err(invalid("`run_id` (see lod `trace_runs`) for lod `trace`"));
}
let from = p.get("from_ns").and_then(Value::as_u64).unwrap_or(0);
let to = p.get("to_ns").and_then(Value::as_u64).unwrap_or(u64::MAX);
if to < from {
return Err(invalid(format!("an empty window: `to_ns` {to} is before `from_ns` {from}")));
}
let after = p.get("after_seq").and_then(Value::as_u64);
let limit = p.get("limit").and_then(Value::as_u64).filter(|&n| n > 0).unwrap_or(20_000).min(200_000) as usize;
let (rows, more) = wh.block_on(fn_trace::query_window(wh, run_id, from, to, after, limit)).map_err(internal)?;
let mut heads = Heads::new(room, map);
let mut events = Vec::with_capacity(rows.len());
let mut unplaced = 0u64;
let need_all = p.get("heat").and_then(Value::as_bool).unwrap_or(false)
|| (p.get("stacks").and_then(Value::as_bool).unwrap_or(false) && from > 0)
|| (!rows.is_empty() && (from > 0 || after.is_some()));
let all = if need_all { wh.block_on(fn_trace::query_all_events(wh, run_id)).map_err(internal)? } else { Vec::new() };
// each thread's open placed functions (stable ids, outermost first): the
// caller an entry's pulse runs from is the innermost placed one open on its
// thread — seeded from the run before this window
let mut open: HashMap<u32, Vec<String>> = HashMap::new();
let mut walk = |open: &mut HashMap<u32, Vec<String>>, e: &fn_trace::EventRow| {
if e.stable_id.is_empty() {
return;
}
let st = open.entry(e.thread).or_default();
match e.kind {
EventKind::Enter => st.push(e.stable_id.clone()),
EventKind::Exit => {
if let Some(i) = st.iter().rposition(|s| *s == e.stable_id) {
st.truncate(i);
}
}
}
};
if let Some(first) = rows.first() {
for e in all.iter().take_while(|e| e.seq_no < first.seq_no) {
walk(&mut open, e);
}
}
let sym: HashMap<String, String> = all.iter().chain(rows.iter()).map(|e| (e.stable_id.clone(), e.symbol.clone())).collect();
for r in &rows {
match heads.get(&r.stable_id, &r.symbol) {
// [seq, t_ns, thread, kind (0 enter, 1 exit), head, caller head | null]
Some(h) => {
let caller = if r.kind == EventKind::Enter {
open.get(&r.thread).and_then(|st| st.iter().rev().find_map(|s| heads.get(s, sym.get(s).map(String::as_str).unwrap_or(""))))
} else {
None
};
events.push(json!([r.seq_no, r.t_ns, r.thread, u8::from(r.kind == EventKind::Exit), h, caller]));
}
None => unplaced += 1,
}
walk(&mut open, r);
}
let mut body = json!({
"lod": "trace",
"run_id": run_id,
"from_ns": from,
"to_ns": if to == u64::MAX { Value::Null } else { json!(to) },
"events": events,
"unplaced": unplaced,
"more": more,
"next_after_seq": rows.last().map(|r| r.seq_no),
});
if need_all {
if p.get("stacks").and_then(Value::as_bool).unwrap_or(false) {
// each thread's open placed heads at `from` (outermost first)
let mut open: BTreeMap<u32, Vec<String>> = BTreeMap::new();
for e in all.iter().take_while(|e| e.t_ns < from) {
if e.stable_id.is_empty() {
continue;
}
let st = open.entry(e.thread).or_default();
match e.kind {
EventKind::Enter => st.push(e.stable_id.clone()),
EventKind::Exit => {
if let Some(i) = st.iter().rposition(|s| *s == e.stable_id) {
st.truncate(i);
}
}
}
}
let sym: HashMap<&str, &str> = all.iter().map(|e| (e.stable_id.as_str(), e.symbol.as_str())).collect();
let stacks: Vec<Value> = open
.into_iter()
.filter(|(_, st)| !st.is_empty())
.map(|(t, st)| json!({ "thread": t, "heads": st.iter().filter_map(|s| heads.get(s, sym.get(s.as_str()).copied().unwrap_or(""))).collect::<Vec<_>>() }))
.collect();
body["stacks"] = json!(stacks);
}
if p.get("heat").and_then(Value::as_bool).unwrap_or(false) {
let heat = fn_trace::heat_of(&all);
let sym: HashMap<&str, &str> = all.iter().map(|e| (e.stable_id.as_str(), e.symbol.as_str())).collect();
let mut placed: Vec<(usize, u64)> = heat
.iter()
.filter(|(k, _)| !k.starts_with('?'))
.filter_map(|(k, ns)| heads.get(k, sym.get(k.as_str()).copied().unwrap_or("")).map(|h| (h, *ns)))
.collect();
placed.sort_by(|a, b| b.1.cmp(&a.1).then(a.0.cmp(&b.0)));
body["heat"] = json!(placed.iter().map(|(h, ns)| json!([h, ns])).collect::<Vec<_>>());
body["duration_ns"] = json!(all.last().map(|e| e.t_ns).unwrap_or(0));
}
}
body["heads"] = Value::Array(heads.list);
Ok(body)
}
/// `lod: profile`.
pub fn profile(reach: &Reach<'_>, room: &Room, map: &MapLayout, p: &Value) -> Result<Value, Refusal> {
let wh = warehouse(reach)?;
let repo = text(p, "repo");
let run_id = text(p, "run_id");
let Some((id, target, samples, total_ns, rows)) = wh.block_on(fn_trace::query_profile(wh, repo, run_id)).map_err(internal)? else {
return Ok(json!({ "lod": "profile", "repo": repo, "run_id": Value::Null, "heat": [], "heads": [], "note": "no profile filed (a Runa `Profile` step files one)" }));
};
let mut heads = Heads::new(room, map);
let mut heat = Vec::new();
let mut unplaced = 0u64;
for r in &rows {
match heads.get(&r.stable_id, &r.symbol) {
// [head, self_ns, total_ns]
Some(h) => heat.push(json!([h, r.self_ns, r.total_ns])),
None => unplaced += 1,
}
}
Ok(json!({
"lod": "profile", "repo": repo, "run_id": id, "target": target,
"samples": samples, "total_ns": total_ns, "heat": heat, "unplaced": unplaced,
"heads": heads.list,
}))
}