src/viz/action_log.rs
//! TOTAL user-action debug log — the headless-debuggability contract for the
//! viz. Every interaction (tab switch, button click, query, RPC call, error) is
//! pushed here with a millisecond timestamp so that "when I fire it up you
//! really follow every little bug".
//!
//! **Four** sinks, all always-on and cheap:
//! 1. an in-memory **ring buffer** (last [`CAP`] entries) the app paints in a
//! bottom overlay panel,
//! 2. a **per-entry stderr line** (`nornir-viz ACTION …`) so the operator
//! watching the launching terminal / server log sees the trail live, and
//! 3. an appended **file** at `$NORNIR_VIZ_ACTIONLOG` (default
//! `/tmp/nornir_viz_actions.log`) — externally observable + greppable for
//! bug-hunting, mirroring how `$NORNIR_VIZ_STATE` dumps the UI state, and
//! 4. **N5** — the durable warehouse `viz_actions` table (when a warehouse is
//! attached). The `/tmp` file stays the fast LOCAL live-channel; the
//! warehouse is the DURABLE, queryable sink (write both, per the
//! persist-to-warehouse-stream-not-`/tmp` law). The viz also reads its own
//! recent action history **back** from this table (see
//! [`ActionLog::warehouse_recent_json`]).
//!
//! Cheap by design: a `Mutex<VecDeque>` push + a `writeln!` + a non-blocking
//! channel send to a background drain thread (which owns the warehouse and
//! batches a burst into one Iceberg snapshot — the viz frame never blocks on
//! I/O). The instrumented call sites (see `app.rs`) only push on real
//! edge-triggered events (a click, a tab change, an RPC firing), never every
//! frame, so it never floods.
use std::collections::VecDeque;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::sync::mpsc::{Sender, channel};
use crate::warehouse::iceberg::IcebergWarehouse;
use crate::warehouse::viz_actions::{
ActionSelector, VizActionRow, append_viz_actions_blocking, query_viz_actions, rows_to_json,
};
/// Max entries retained in the in-memory ring (older ones drop off the front).
const CAP: usize = 256;
/// One timestamped action-log entry.
#[derive(Clone, Debug)]
pub struct Entry {
/// Monotonic sequence number (since app start) — stable ordering even when
/// two entries land in the same millisecond.
pub seq: u64,
/// `HH:MM:SS.mmm` wall-clock stamp (local), for the overlay + file.
pub stamp: String,
/// Coarse category — drives the overlay color + the grep prefix.
pub kind: Kind,
/// Free-form detail, e.g. `tab=Bench`, `Search.Query q="foo"`, the error text.
pub detail: String,
}
/// Action category. Kept small + greppable.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Kind {
/// App lifecycle (launch, workspace switch, reload).
Life,
/// A tab became active.
Tab,
/// A button / control was clicked.
Click,
/// A user-issued query (search / lookup / call path).
Query,
/// An outgoing gRPC call to the server.
Rpc,
/// An error surfaced anywhere in the flow.
Error,
}
impl Kind {
pub fn tag(self) -> &'static str {
match self {
Kind::Life => "LIFE",
Kind::Tab => "TAB",
Kind::Click => "CLICK",
Kind::Query => "QUERY",
Kind::Rpc => "RPC",
Kind::Error => "ERROR",
}
}
}
/// Central action log: a ring buffer + the file/stderr sinks + the durable
/// warehouse sink (N5). Interior-mutable (`Mutex`) so it can be shared `&self`
/// across every instrumented call site without threading `&mut` through the
/// whole app.
pub struct ActionLog {
inner: Mutex<Inner>,
/// File sink path (surfaced in the overlay header).
file_path: String,
/// This launch's identity — the `viz_actions.session_id` every row carries,
/// so the warehouse trail is groupable + readable back per session.
session_id: String,
/// Warehouse root the read-back queries open (set when a durable sink is
/// attached). `None` keeps the log file/stderr-only (e.g. remote mode or a
/// warehouse that wouldn't open).
warehouse_root: Option<PathBuf>,
/// Join handle for the background drain thread, so [`flush`](Self::flush)
/// can close the channel and wait for the final snapshot to land (used by
/// the inject-assert test; the live viz just lets it run for its lifetime).
drain: Mutex<Option<std::thread::JoinHandle<()>>>,
}
struct Inner {
ring: VecDeque<Entry>,
seq: u64,
file: Option<std::fs::File>,
/// Active context stamped onto each warehouse row (cheap to update each
/// frame from the app: the current workspace + tab).
workspace: String,
tab: String,
/// Non-blocking channel into the background warehouse-drain thread. `None`
/// until/unless a warehouse is attached.
wh_tx: Option<Sender<VizActionRow>>,
}
impl Default for ActionLog {
fn default() -> Self {
Self::new()
}
}
impl ActionLog {
pub fn new() -> Self {
// Default under the user's OWN state dir, never `/tmp`. `/tmp` is
// world-writable and this filename is fixed, so on a shared box any local
// user can pre-create (or symlink) `/tmp/nornir_viz_actions.log` and the
// next viz launch happily `File::create`s straight through the symlink —
// truncating whatever it points at, under our uid. Same class as korp's
// move off `/tmp` to `~/.korp`.
let file_path = std::env::var("NORNIR_VIZ_ACTIONLOG").unwrap_or_else(|_| {
let base = std::env::var_os("XDG_STATE_HOME")
.map(std::path::PathBuf::from)
.or_else(|| {
std::env::var_os("HOME")
.map(|h| std::path::PathBuf::from(h).join(".local/state"))
})
// No HOME either: stay in the workdir we own rather than /tmp.
.unwrap_or_else(|| std::path::PathBuf::from("."));
let dir = base.join("nornir");
let _ = std::fs::create_dir_all(&dir);
dir.join("viz_actions.log").to_string_lossy().into_owned()
});
// Truncate on launch so each session's trail starts clean (matches the
// single-snapshot semantics of $NORNIR_VIZ_STATE).
// A failure here is NOT silent: the ring buffer still works, but an
// operator who asked for a file trail must be told they are not getting
// one, with the reason and the path (see `nornir-workspace-*` class:
// "an empty surface that passes").
let file = match std::fs::File::create(&file_path) {
Ok(f) => Some(f),
Err(e) => {
eprintln!(
"nornir viz: action-log file trail DISABLED — cannot create {file_path}: {e} \
(the in-memory ring still records; set $NORNIR_VIZ_ACTIONLOG to a writable path)"
);
None
}
};
let log = Self {
inner: Mutex::new(Inner {
ring: VecDeque::with_capacity(CAP),
seq: 0,
file,
workspace: String::new(),
tab: String::new(),
wh_tx: None,
}),
file_path,
session_id: uuid::Uuid::new_v4().to_string(),
warehouse_root: None,
drain: Mutex::new(None),
};
log.push(
Kind::Life,
format!("action-log started → {}", log.file_path),
);
log
}
/// Attach the durable warehouse sink (N5): spawn a background drain thread
/// that owns one [`IcebergWarehouse`] for the log's lifetime and batches
/// incoming actions into one Iceberg snapshot per burst, and remember the
/// `root` so the viz can read its own recent history **back** from the
/// `viz_actions` table. Best-effort: if the warehouse can't open, telemetry
/// stays off (file/stderr keep working) and read-back returns empty.
///
/// Call once, right after [`new`](Self::new). The `/tmp` file remains the
/// fast local live-channel; this is the durable sink (write both).
/// The dedicated warehouse directory for the action-log durable sink, derived
/// from the data warehouse `root`. Kept separate so the sink's exclusive redb
/// writer lock never collides with the data tabs' read-only opens.
fn actionlog_warehouse_root(root: &Path) -> PathBuf {
root.join("viz_actionlog")
}
pub fn attach_warehouse(&mut self, root: PathBuf) {
// The durable sink keeps its OWN warehouse in a dedicated subdirectory so
// it never holds redb's single-writer lock on the *data* `catalog.redb`.
// Otherwise this thread's exclusive `open()` would force every data tab's
// `open_read_only()` into the degraded copy-aside snapshot path (which can
// read mid-write / stale state → 0 rows). The `viz_actions` table is the
// viz's own click-trail telemetry; nothing else reads it from the data
// warehouse, so isolating it is invisible to every other surface. Both the
// write (this drain thread) and the read-back (`warehouse_recent_json`)
// use this same dedicated root, so the trail stays self-consistent.
let log_root = Self::actionlog_warehouse_root(&root);
let _ = std::fs::create_dir_all(&log_root);
let (tx, rx) = channel::<VizActionRow>();
let wh_root = log_root.clone();
let spawned = std::thread::Builder::new()
.name("nornir-viz-actionlog".into())
.spawn(move || {
let wh = match IcebergWarehouse::open(&wh_root) {
Ok(w) => w,
Err(e) => {
eprintln!(
"nornir-viz action-log: warehouse open failed; durable sink off: {e:#}"
);
return;
}
};
// Drain-batch: take the first, then greedily coalesce whatever
// else is queued so a burst of clicks is ONE Iceberg snapshot.
while let Ok(first) = rx.recv() {
let mut batch = vec![first];
while let Ok(more) = rx.try_recv() {
batch.push(more);
if batch.len() >= 256 {
break;
}
}
if let Err(e) = append_viz_actions_blocking(&wh, &batch) {
eprintln!(
"nornir-viz action-log: append {} row(s) failed (non-fatal): {e:#}",
batch.len()
);
}
}
});
match spawned {
Ok(handle) => {
if let Ok(mut g) = self.inner.lock() {
g.wh_tx = Some(tx);
}
if let Ok(mut d) = self.drain.lock() {
*d = Some(handle);
}
self.warehouse_root = Some(log_root);
// The first row (the LIFE "started" line) was pushed before the
// sink existed; re-emit a LIFE marker so the durable trail opens
// with this session's start too.
self.push(
Kind::Life,
format!(
"durable sink → warehouse viz_actions (session {})",
self.session_id
),
);
}
Err(e) => {
eprintln!(
"nornir-viz action-log: drain thread spawn failed; durable sink off: {e}"
);
}
}
}
/// Path of the file sink (surfaced in the overlay header).
pub fn file_path(&self) -> &str {
&self.file_path
}
/// This launch's `viz_actions.session_id`.
pub fn session_id(&self) -> &str {
&self.session_id
}
/// Set the active context (workspace + tab) stamped onto subsequent
/// warehouse rows. Cheap; the app calls it each frame (only the lock cost).
pub fn set_context(&self, workspace: &str, tab: &str) {
if let Ok(mut g) = self.inner.lock() {
if g.workspace != workspace {
g.workspace = workspace.to_string();
}
if g.tab != tab {
g.tab = tab.to_string();
}
}
}
/// Record one action. Always-on; pushes to the ring + stderr + file + (when
/// attached) the durable warehouse sink.
pub fn push(&self, kind: Kind, detail: impl Into<String>) {
let detail = detail.into();
let stamp = now_stamp();
let ts_micros = chrono::Utc::now().timestamp_micros();
let Ok(mut g) = self.inner.lock() else { return };
g.seq += 1;
let seq = g.seq;
let entry = Entry {
seq,
stamp: stamp.clone(),
kind,
detail: detail.clone(),
};
// stderr: live trail for the launching terminal / server log.
eprintln!("nornir-viz ACTION {} [{}] {}", stamp, kind.tag(), detail);
// file: greppable, externally observable (mirrors $NORNIR_VIZ_STATE).
if let Some(f) = g.file.as_mut() {
let _ = writeln!(f, "{} {:>5} [{}] {}", stamp, seq, kind.tag(), detail);
let _ = f.flush();
}
// warehouse: durable, queryable sink (N5). Non-blocking send — the frame
// never waits on Iceberg I/O; a dropped channel just disables the sink.
if let Some(tx) = g.wh_tx.as_ref() {
let row = VizActionRow {
session_id: self.session_id.clone(),
seq: seq as i64,
ts_micros,
kind: kind.tag().to_string(),
workspace: g.workspace.clone(),
tab: g.tab.clone(),
detail: detail.clone(),
};
if tx.send(row).is_err() {
// Drain thread gone (warehouse open failed) — drop the sink so we
// stop trying.
g.wh_tx = None;
}
}
if g.ring.len() == CAP {
g.ring.pop_front();
}
g.ring.push_back(entry);
}
/// Read this session's recent actions **back** from the durable warehouse
/// `viz_actions` table as a JSON array (oldest→newest, last `n`). This is
/// the read-back history surfaced in `state_json` — it proves the trail is
/// durable + queryable, not just a `/tmp` file. Returns `"[]"` when no
/// warehouse is attached or the read fails (never panics, never blocks a
/// frame for long: a handful of rows scoped to one `session_id` partition).
pub fn warehouse_recent_json(&self, n: usize) -> String {
let Some(root) = self.warehouse_root.as_ref() else {
return "[]".to_string();
};
let wh = match IcebergWarehouse::open_read_only(root) {
Ok(w) => w,
Err(_) => return "[]".to_string(),
};
let sel = ActionSelector::Session(self.session_id.clone());
match wh.block_on(query_viz_actions(&wh, &sel, Some(n))) {
Ok(rows) => rows_to_json(&rows),
Err(_) => "[]".to_string(),
}
}
/// Close the durable-sink channel and **block** until the background drain
/// thread has flushed every queued action into the warehouse. After this
/// returns, the `viz_actions` table reflects all actions pushed so far — and
/// no more can be appended (the sink is closed). Intended for the
/// inject-assert test (deterministic, no sleeps) and a clean shutdown; the
/// live viz never calls it (the drain runs for the process lifetime).
pub fn flush(&self) {
// Drop the sender so the drain thread's `rx.recv()` returns `Err` and the
// loop exits after appending its last batch.
if let Ok(mut g) = self.inner.lock() {
g.wh_tx = None;
}
let handle = self.drain.lock().ok().and_then(|mut d| d.take());
if let Some(h) = handle {
let _ = h.join();
}
}
/// Snapshot the most recent `n` entries (oldest→newest) for the overlay.
pub fn recent(&self, n: usize) -> Vec<Entry> {
let Ok(g) = self.inner.lock() else {
return Vec::new();
};
let len = g.ring.len();
let start = len.saturating_sub(n);
g.ring.iter().skip(start).cloned().collect()
}
/// Total entries recorded this session (the live counter, for the header).
pub fn count(&self) -> u64 {
self.inner.lock().map(|g| g.seq).unwrap_or(0)
}
/// Tail the action-log **file** — the single shared source of truth. Unlike
/// [`recent`](Self::recent) (this viz's own in-memory ring), this reads back
/// the file, so it also surfaces lines appended by **other** nornir processes
/// to the same `$NORNIR_VIZ_ACTIONLOG` — most importantly a running
/// `nornir bench`'s `[BENCH]` progress, so the operator watches the benchmark
/// live in the viz's action-trail overlay. Bounded to the last `max_bytes`
/// (cheap even when a long bench floods the file) and the last `max_lines`.
pub fn tail_lines(&self, max_bytes: u64, max_lines: usize) -> Vec<String> {
use std::io::{Read, Seek, SeekFrom};
let Ok(mut f) = std::fs::File::open(&self.file_path) else {
return Vec::new();
};
let len = f.metadata().map(|m| m.len()).unwrap_or(0);
let start = len.saturating_sub(max_bytes);
if start > 0 && f.seek(SeekFrom::Start(start)).is_err() {
return Vec::new();
}
let mut bytes = Vec::new();
if f.read_to_end(&mut bytes).is_err() {
return Vec::new();
}
// from_utf8_lossy: the seek may land mid-multibyte; never panic on it.
let text = String::from_utf8_lossy(&bytes);
let mut lines: Vec<&str> = text.lines().collect();
// Drop the (likely partial) first line when we seeked into the middle.
if start > 0 && !lines.is_empty() {
lines.remove(0);
}
let n = lines.len();
lines[n.saturating_sub(max_lines)..]
.iter()
.map(|s| s.to_string())
.collect()
}
}
/// `HH:MM:SS.mmm` local-time stamp. Uses `chrono` (already a dep) so it matches
/// the rest of the codebase's time formatting.
fn now_stamp() -> String {
chrono::Local::now().format("%H:%M:%S%.3f").to_string()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ring_caps_and_orders() {
let _g = crate::serial_guard(); // global env NORNIR_VIZ_ACTIONLOG — serialize
// Don't clobber a real operator file during tests.
unsafe { std::env::set_var("NORNIR_VIZ_ACTIONLOG", "/tmp/nornir_viz_actions_test.log") };
let log = ActionLog::new();
for i in 0..(CAP + 50) {
log.push(Kind::Click, format!("click {i}"));
}
let recent = log.recent(10);
assert_eq!(recent.len(), 10);
// Newest last; seq strictly increasing.
for w in recent.windows(2) {
assert!(w[1].seq > w[0].seq);
}
// The ring never exceeds CAP even after CAP+50 (+1 the LIFE start) pushes.
assert!(log.recent(usize::MAX).len() <= CAP);
// Count tracks every push including the LIFE start line.
assert_eq!(log.count(), CAP as u64 + 50 + 1);
}
#[test]
fn kinds_have_distinct_tags() {
let tags = [
Kind::Life.tag(),
Kind::Tab.tag(),
Kind::Click.tag(),
Kind::Query.tag(),
Kind::Rpc.tag(),
Kind::Error.tag(),
];
let mut uniq = tags.to_vec();
uniq.sort_unstable();
uniq.dedup();
assert_eq!(uniq.len(), tags.len());
}
}