crates/nornir-warehouse-embed/src/viz.rs
//! **Viz READ client — the THIN gRPC leg for `Viz.Jobs`, the jera job ledger.**
//!
//! The rule is written one line above the answer in the upstream proto
//! (`edda/crates/nornir-wire/proto/nornir.proto:981-983`):
//!
//! > the server holds the redb lock, **SO CLIENTS GO THROUGH HERE.**
//!
//! nornir owns `<warehouse root>/jobs.redb` and keeps it open for the process
//! lifetime (`server_grpc.rs:6013`). A client that wants the ledger asks the server
//! for it. That is what this module does: it speaks `/nornir.v1.Viz/Jobs` and hands
//! back the server's JSON verbatim, exactly as [`crate::mimir`] does for the lens
//! folds — the SAME `<guest>-embed` / no-`nornir`-dep shape as the warehouse,
//! workspaces, Mímir and bench legs beside it.
//!
//! ## The file road, and why it is not merely "locked"
//!
//! The predecessor read the ledger off the filesystem
//! (`dwarves_embed::read_ledger_jobs` → `jera::jobs::JobStore::open_read_only`). The
//! standing diagnosis was "redb takes an EXCLUSIVE lock and refuses a second handle".
//! Measured on t14s 2026-08-25, that is **not** what fails:
//!
//! | probe | result |
//! |---|---|
//! | second plain `JobStore::open` against a live writer | `Database already open. Cannot acquire lock.` |
//! | `JobStore::open_read_only` against the SAME live writer | **succeeded**, 1 row — it snapshots the file aside |
//! | `open_read_only` on a `0o444` ledger | `I/O error: Permission denied (os error 13)` |
//!
//! So `open_read_only` genuinely IS lock-tolerant (its doc comment, and the one in
//! `dwarves-embed`'s `read_ledger_jobs`, are true). What it is not is
//! *permission*-tolerant: it routes through `redb::Database::create`, so a ledger the
//! reader may read but not WRITE fails before [`is_lock_error`] is ever consulted, and
//! the snapshot fallback is never reached. The installed forge runs as user `dwarves`
//! and writes `-rw-r--r--`; the UI does not run as `dwarves`. On top of that the
//! lock-tolerant path costs a whole-file byte copy per read, and the Jera tab polls
//! every 2 s.
//!
//! [`is_lock_error`]: https://git.gunnar.rs/nordisk/edda
//!
//! ## Absent vs empty over the wire
//!
//! The file reader distinguishes three states (`LedgerRead::{Absent,Empty,Rows}`)
//! because on a filesystem "no file" and "no rows" are different facts. Over the wire
//! they are not the client's to invent: a server that answers at all has opened the
//! served workspace's ledger, so `[]` means **the ledger is empty**, and "there is no
//! such workspace" arrives as a typed [`WarehouseReadError::NotServed`] — a NAMED
//! failure, never an empty list. The caller therefore still gets three distinguishable
//! answers; none of them is a silent empty. That is the invariant `1aa8f0f` bought with
//! a workspace that painted invented jobs, and it survives the move to gRPC.
use crate::{norm_endpoint, pb, status_to_read_error, WarehouseReadError};
use anyhow::Context;
/// The intercepted `Viz` client type (bearer + `nornir-workspace` header) — the exact
/// interceptor shape the warehouse and Mímir read legs use.
type VizClient = pb::viz_client::VizClient<
tonic::service::interceptor::InterceptedService<
tonic::transport::Channel,
crate::NornirInterceptor,
>,
>;
/// Run a blocking `Viz` gRPC call on a DEDICATED OS thread (never a nested-runtime
/// panic — an egui host may drive this from inside its own tokio runtime), with the
/// bearer + `nornir-workspace` header + a 3s connect timeout. A connect failure maps to
/// [`WarehouseReadError::Unreachable`]; the closure maps its RPC `Status` via
/// `status_to_read_error`. Mirrors `with_mimir_client` / `with_warehouse_client` — the
/// crate carries ONE spawn body per served nornir service, and this is the `Viz` one.
fn with_viz_client<T, F>(
endpoint: &str,
token: &str,
workspace: &str,
f: F,
) -> Result<T, WarehouseReadError>
where
T: Send + 'static,
F: for<'a> FnOnce(
&'a mut VizClient,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = Result<T, WarehouseReadError>> + 'a>,
> + Send,
{
let endpoint = norm_endpoint(endpoint);
let token = token.to_string();
let workspace = workspace.to_string();
std::thread::scope(|s| {
s.spawn(|| {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| {
WarehouseReadError::Unreachable(format!("build tokio runtime: {e}"))
})?;
rt.block_on(async move {
// The ONE shared client bearer-builder (Law #1) — no per-site
// `format!("Bearer …").parse()` twin. `None` ⇒ open (empty token).
let bearer = nornir_service::client::bearer(&token).map_err(|e| {
WarehouseReadError::AuthFailed(format!("bad bearer token: {e}"))
})?;
let ws_md: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> =
(!workspace.is_empty())
.then(|| workspace.parse().ok())
.flatten();
let channel = tonic::transport::Channel::from_shared(endpoint.clone())
.map_err(|e| {
WarehouseReadError::Unreachable(format!("invalid url `{endpoint}`: {e}"))
})?
.connect_timeout(std::time::Duration::from_secs(3))
.timeout(crate::READ_TIMEOUT)
.connect()
.await
.map_err(|e| {
WarehouseReadError::Unreachable(format!("connect to {endpoint}: {e}"))
})?;
let interceptor: crate::NornirInterceptor =
Box::new(move |mut req: tonic::Request<()>| {
if let Some(b) = &bearer {
req.metadata_mut().insert("authorization", b.clone());
}
if let Some(ws) = &ws_md {
req.metadata_mut().insert("nornir-workspace", ws.clone());
}
Ok(req)
});
let mut client = pb::viz_client::VizClient::with_interceptor(channel, interceptor)
.max_decoding_message_size(64 << 20);
f(&mut client).await
})
})
.join()
.map_err(|_| WarehouseReadError::Rpc("viz-read thread panicked".into()))?
})
}
/// **The ONE decode for a `Viz` JSON RPC that answers a ROW ARRAY.**
///
/// Every `Viz` read on this leg hands back `JsonResponse.json`, and every one of
/// them must treat a non-array body as a failure rather than as an empty list — a
/// server saying `Ok` and sending something that is not rows is a bug, and it must
/// never render as "nothing happened". That rule lived once, inline in [`jobs`];
/// with a second reader ([`clone_events`]) it becomes a rule two copies would have
/// to agree on, so it lives here instead (LAW 5).
///
/// `rpc` names the verb for the error text a person reads.
fn rows_of(rpc: &str, json: &str) -> Result<Vec<serde_json::Value>, WarehouseReadError> {
let v: serde_json::Value = serde_json::from_str(json)
.with_context(|| format!("parse {rpc} JsonResponse body"))
.map_err(|e| WarehouseReadError::Rpc(format!("{e:#}")))?;
match v {
serde_json::Value::Array(rows) => Ok(rows),
other => Err(WarehouseReadError::Rpc(format!(
"{rpc} answered Ok with a {} body, not a JSON array of rows",
match other {
serde_json::Value::Null => "null",
serde_json::Value::Bool(_) => "bool",
serde_json::Value::Number(_) => "number",
serde_json::Value::String(_) => "string",
serde_json::Value::Object(_) => "object",
serde_json::Value::Array(_) => unreachable!(),
}
))),
}
}
/// **`Viz.Jobs`** — the served workspace's jera job ledger, newest-first, as the JSON
/// array of `jera::jobs::JobRecord` the server serialized.
///
/// Returned as a [`serde_json::Value`] verbatim (the crate deliberately carries no
/// `jera` dependency — the consumer that already links `jera` deserializes it into the
/// real `JobRecord`, and because `nornir::jobs` is a re-export of `jera::jobs` those are
/// the SAME type, not two structurally-equal twins).
///
/// * `kind` — optional `jobs::kind` filter; empty = all kinds.
/// * `limit` — max rows; 0 = the server default (~200).
///
/// A non-array body is an [`WarehouseReadError::Rpc`], not an empty list: the server
/// saying `Ok` and sending something that is not a ledger is a failure, and a failure
/// must never render as "no jobs".
pub fn jobs(
endpoint: &str,
token: &str,
workspace: &str,
kind: &str,
limit: u32,
) -> Result<Vec<serde_json::Value>, WarehouseReadError> {
let (ws, kind) = (workspace.to_string(), kind.to_string());
with_viz_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.jobs(pb::JobsRequest {
workspace: ws,
kind,
limit,
})
.await
.map_err(status_to_read_error)?
.into_inner();
rows_of("Viz.Jobs", &resp.json)
})
})
}
/// **`Viz.CloneEvents`** — the served workspace's recent `clone_events` rows,
/// newest-first, as the JSON array of `nornir_wire::view::rows::CloneEventRow` the
/// server serialized.
///
/// **This is the verb that says WHY a member did not populate** (`ok` / `failed`
/// plus the error detail and when), and it is what a job row's
/// `result_ref: "clone_events:<workspace>"` points at. Until this existed the
/// question had no client on the `:7900` road at all: the pane could only NAME the
/// RPC, so every operator whose UI can reach the forge but not nornir directly —
/// dwarves-ui on t14s, for one — was told where the answer lives and given no way
/// to read it.
///
/// The server caps the row count itself (`VIZ_CLONE_EVENTS_CAP`, 2 000) because an
/// unbounded history overflowed the 4 MiB gRPC decode cap; there is deliberately no
/// `limit` here to re-decide a bound the upstream already owns.
///
/// The workspace travels in the `nornir-workspace` metadata header (the interceptor
/// sets it), so the request body's `workspace` stays empty — identical to
/// [`jobs`].
pub fn clone_events(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<Vec<serde_json::Value>, WarehouseReadError> {
with_viz_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.clone_events(pb::VizTimelineRequest {
workspace: String::new(),
})
.await
.map_err(status_to_read_error)?
.into_inner();
rows_of("Viz.CloneEvents", &resp.json)
})
})
}
#[cfg(test)]
mod tests {
use super::*;
/// LAW 2 — the decode must REFUSE a non-array body rather than render it as
/// "no rows". A reader that silently turns `{}` or `null` into an empty list is
/// exactly the silent-empty this whole leg exists to prevent.
#[test]
fn a_non_array_body_is_a_named_failure_not_an_empty_list() {
assert_eq!(
rows_of("Viz.CloneEvents", "[]").unwrap(),
Vec::<serde_json::Value>::new()
);
assert_eq!(
rows_of("Viz.CloneEvents", r#"[{"member":"skade"}]"#)
.unwrap()
.len(),
1
);
for body in ["null", "{}", r#""oops""#, "7", "true"] {
let err = rows_of("Viz.CloneEvents", body)
.expect_err("a {body} body must not decode as rows");
let msg = err.to_string();
assert!(
msg.contains("Viz.CloneEvents"),
"the failure must name the verb a person can retry: {msg}"
);
}
// Malformed JSON is a failure too, and names the verb.
let err = rows_of("Viz.Jobs", "[{").expect_err("truncated JSON must fail");
assert!(err.to_string().contains("Viz.Jobs"), "{err}");
}
}