nordisk/dwarves
crates/nornir-warehouse-embed/src/lib.rs
//! **nornir-warehouse-embed** — a finished dwarves build lands in nornir's
//! warehouse (backend (2), the WRITE leg).
//!
//! The reusable `<guest>-embed` adapter shape (`nornir-funnel-embed` is the
//! template): a **thin tonic gRPC client** ships a build's
//! [`nornir_build_thing::BuildInfoRow`] stream to a running `nornir-server`'s
//! [`Telemetry.SubmitTestResults`](pb) write RPC, so the rows land in the served
//! workspace's unified Iceberg warehouse (`test_results`) and show up in the viz
//! 🧪 Test pane. The server holds the single-writer redb catalog lock, so a thin
//! client can't open the warehouse to write it — it submits rows and the server
//! appends them (the exact contract in `nornir/.nornir/data-submit-rpcs.md`).
//!
//! **No `nornir` dependency.** nornir already depends on dwarves, so a `dwarves →
//! nornir` crate edge would be a cycle. Instead this crate compiles ONLY nornir's
//! `Telemetry.SubmitTestResults` proto (a trimmed copy under `proto/`, package
//! `nornir.v1`, so the gRPC method path `/nornir.v1.Telemetry/SubmitTestResults`
//! matches the live server) and speaks it directly.
//!
//! ## The seam
//! [`NornirWarehouseSink`] impls [`nornir_build_thing::BuildInfoSink`]: hand it to
//! `nornir_build_thing::build_variant_with_info` and every emitted build-info row
//! is shipped to the warehouse. It is **best-effort**: with no configured endpoint
//! it is a **no-op** (embedded/local behaviour is unchanged), and a submit failure
//! (server offline) is logged and swallowed — it NEVER fails the build (L2).
//!
//! ## The BUILD-RECORD producer (2026-07-31 — the fourth-recurrence fix)
//! [`NornirWarehouseSink::record_build`] is the **producer**: it ships a finished
//! build's [`nornir_build_thing::BuildProvenance`] over
//! `Telemetry.SubmitBuildProvenance` and its §f3 [`BuildInfoRow`] stream over
//! `Telemetry.SubmitBuildInfo`, so the rows land in the served workspace's
//! `build_provenance` + `build_info` tables.
//!
//! Both RPCs, and both server handlers, existed long before anything called them.
//! ⚒ Build Thing came up empty FOUR times and was misdiagnosed three of those as a
//! wiring/read problem, because the write side had **no producer at all**: the
//! ground truth on oden was `build_provenance` = 0 parquet files, `build_info` = 0,
//! `symbol_facts` = 984 — same warehouse, same snapshot mechanism. Reads were never
//! broken; those two tables had simply never received a single row.
//!
//! Every submit returns a [`BuildRecordReceipt`], not a swallowed `()`. Shipping is
//! still best-effort (a dead server never fails a build, L2) but the OUTCOME is a
//! value the caller must carry: [`BuildRecordReceipt::is_red`] is true whenever an
//! endpoint was configured and zero rows landed. That receipt is the guard — see
//! `dwarves::build_thing::BuildReport::warehouse`.
//!
//! The historical interim (build-info mapped onto `test_results` rows tagged
//! `aspect = build-<kind>`) survives as [`build_info_to_test_rows`] for the 🧪 Test
//! pane, but it is no longer the landing: a build's native home is `build_info`.
use anyhow::{Context, Result};
use nornir_build_thing::{BuildInfoRow, BuildInfoSink, BuildProvenance};
/// The generated thin client + (test) server + wire types for nornir's
/// `Telemetry.SubmitTestResults`.
pub mod pb {
tonic::include_proto!("nornir.v1");
}
/// **The Mimir LENS read client** (`Mimir.{GenericCost,ZombieFeatures,ChangeRisk,
/// SimilarCode,WeakTests,BenchGaps}`) — the THIN gRPC leg a dwarves host uses to ask
/// nornir to fold the six lenses over the warehouse IT owns, instead of the host
/// folding locally over a warehouse it does not have. Same no-`nornir`-dep shape as
/// the warehouse/workspaces legs above.
pub mod mimir;
/// **The Viz READ client** (`Viz.Jobs`) — the THIN gRPC leg a host uses to read the
/// served workspace's jera job ledger off the server that HOLDS it, instead of opening
/// a `jobs.redb` at a filesystem path it had to guess. Same no-`nornir`-dep shape as the
/// warehouse / workspaces / Mímir legs above; the rule it implements is stated in
/// `nornir.proto` one line above the verb: *the server holds the redb lock, so clients
/// go through here*.
pub mod viz;
/// **The code-room READ client** (`CodeRoom.Answer`) — the ONE door a thin host asks
/// nornir's code room through (the crate graph, the function tier, symbols, the source
/// of an item, what the tests reach, a snapshot diff). Every outcome is a
/// [`room::RoomAnswer`] with a named state; only a computed one hands out rows.
pub mod room;
/// The default `nornir-server` endpoint on this box ("dwarf") — same default the
/// funnel embed uses. `NORNIR_SERVER` overrides it.
pub const DEFAULT_ENDPOINT: &str = "http://127.0.0.1:7878";
/// Normalize an endpoint: prefix `http://` if the scheme is missing.
fn norm_endpoint(endpoint: &str) -> String {
if endpoint.starts_with("http") {
endpoint.to_string()
} else {
format!("http://{endpoint}")
}
}
/// **Map a build's [`BuildInfoRow`] stream onto `test_results` rows.**
///
/// Pure (no network) so it is unit-testable. Every build-info row becomes ONE
/// [`pb::TestResultPb`], tagged `aspect = "build-<kind>"` and carrying the row's
/// `name`/`value` — so the `provenance` / `dep` / `artifact` / `log` / `metric` /
/// `step` feed lands in the warehouse verbatim. The `status` is derived honestly:
/// a `step` row is `pass`/`fail` off its `level`; an `error`-level `log` is `fail`;
/// every other row is a *fact* (`pass`, it didn't fail anything). A numeric
/// `value` (the `metric` rows) rides in `metric`; the raw value also rides in
/// `message` so nothing is lost.
pub fn build_info_to_test_rows(
rows: &[BuildInfoRow],
repo: &str,
run_id: &str,
) -> Vec<pb::TestResultPb> {
rows.iter()
.map(|r| {
let status = match r.kind.as_str() {
"step" => match r.level.as_str() {
"done" => "pass",
"failed" => "fail",
_ => "info",
},
"log" => {
if r.level == "error" {
"fail"
} else {
"info"
}
}
// provenance / dep / artifact / metric are recorded facts, not
// pass/fail gates — never mark a build red for carrying one.
_ => "pass",
};
let test_name = if r.name.is_empty() {
r.kind.clone()
} else {
format!("{}:{}", r.kind, r.name)
};
pb::TestResultPb {
run_id: run_id.to_string(),
repo: repo.to_string(),
suite: "dwarves-build".to_string(),
test_name,
status: status.to_string(),
duration_ms: 0.0,
ts_micros: r.ts_micros,
message: r.value.clone(),
aspect: format!("build-{}", r.kind),
// metric rows carry a numeric value; non-numeric parses to 0.0.
metric: r.value.parse::<f64>().unwrap_or(0.0),
}
})
.collect()
}
/// **Submit `test_results` rows over the thin `Telemetry.SubmitTestResults` gRPC.**
///
/// Authenticates with `Bearer <token>` (empty ⇒ no real auth); a non-empty
/// `workspace` rides in the `nornir-workspace` metadata header (selects which
/// served workspace's warehouse the rows land in). Returns the server's accepted
/// count. Runs the whole call on a DEDICATED OS thread with its own current-thread
/// runtime, so it is safe to call from ANY context — sync, or nested inside an
/// existing tokio runtime (the dwarves-server serve loop) without the
/// "runtime within a runtime" panic. A 3s connect timeout keeps a dead server
/// from wedging the caller.
pub fn submit_test_results(
endpoint: &str,
token: &str,
workspace: &str,
rows: Vec<pb::TestResultPb>,
) -> Result<u32> {
with_telemetry_client(endpoint, token, workspace, |client| {
Box::pin(async move {
let accepted = client
.submit_test_results(pb::SubmitTestResultsRequest { rows })
.await
.context("Telemetry.SubmitTestResults RPC")?
.into_inner()
.accepted;
Ok(accepted)
})
})
}
/// **Submit `build_provenance` records over `Telemetry.SubmitBuildProvenance`.**
///
/// Each element of `records_json` is one [`BuildProvenance::to_json`]; the server
/// reconstructs it and appends one ledger row per record. Same transport contract
/// as [`submit_test_results`].
pub fn submit_build_provenance(
endpoint: &str,
token: &str,
workspace: &str,
records_json: Vec<String>,
) -> Result<u32> {
with_telemetry_client(endpoint, token, workspace, |client| {
Box::pin(async move {
let accepted = client
.submit_build_provenance(pb::SubmitBuildProvenanceRequest { records_json })
.await
.context("Telemetry.SubmitBuildProvenance RPC")?
.into_inner()
.accepted;
Ok(accepted)
})
})
}
/// Map the §f3 [`BuildInfoRow`]s onto the wire type. Pure (no network), so the
/// mapping — including the `has_blob` presence flag that keeps a NULL blob distinct
/// from an empty one — is unit-testable.
pub fn build_info_to_pb(rows: &[BuildInfoRow]) -> Vec<pb::BuildInfoRowPb> {
rows.iter()
.map(|r| pb::BuildInfoRowPb {
build_id: r.build_id.clone(),
step_id: r.step_id.clone(),
ts_micros: r.ts_micros,
kind: r.kind.clone(),
level: r.level.clone(),
name: r.name.clone(),
value: r.value.clone(),
content_id: r.content_id.clone(),
has_blob: r.blob.is_some(),
blob: r.blob.clone().unwrap_or_default(),
})
.collect()
}
/// **Submit `build_info` rows over `Telemetry.SubmitBuildInfo`.** The NATIVE landing
/// for a dwarves build's manifest + feed rows. Same transport contract as
/// [`submit_test_results`].
pub fn submit_build_info(
endpoint: &str,
token: &str,
workspace: &str,
rows: &[BuildInfoRow],
) -> Result<u32> {
let rows = build_info_to_pb(rows);
with_telemetry_client(endpoint, token, workspace, |client| {
Box::pin(async move {
let accepted = client
.submit_build_info(pb::SubmitBuildInfoRequest { rows })
.await
.context("Telemetry.SubmitBuildInfo RPC")?
.into_inner()
.accepted;
Ok(accepted)
})
})
}
/// **File what a Runa step measured over `Telemetry.SubmitStepFeed`** — a test
/// step's outcomes, a bench run or a coverage export, keyed by (repo, commit).
/// Returns the rows the server wrote (`0` = that run was already filed). Same
/// transport contract as [`submit_test_results`].
pub fn submit_step_feed(
endpoint: &str,
token: &str,
workspace: &str,
req: pb::SubmitStepFeedRequest,
) -> Result<u32> {
with_telemetry_client(endpoint, token, workspace, |client| {
Box::pin(async move {
let accepted = client
.submit_step_feed(req)
.await
.context("Telemetry.SubmitStepFeed RPC")?
.into_inner()
.accepted;
Ok(accepted)
})
})
}
// ═════════════════════════════════════════════════════════════════════════════
// The BENCH-RUN producer — the forge measures, the warehouse persists.
// ═════════════════════════════════════════════════════════════════════════════
/// **Map a finished [`nornir_bench::BenchRun`] onto the wire type.**
///
/// Pure (no network), so the mapping is unit-testable — and it needs to be,
/// because it is LOSSY in two documented ways that a caller must not discover in
/// production:
///
/// * **`BenchResult::source`** (measured vs static-capabilities) has no proto
/// field, so a `<repo>.capabilities` row crosses as an ordinary result. This is
/// not new: `nornir bench run` against a `NORNIR_SERVER` has always shipped
/// through this same `Bench.Submit` shape and lost the same tag. Named here so
/// the loss is a known property of the RPC rather than a surprise.
/// * **non-numeric metrics** are dropped, because `KVF.value` is a `double`. Every
/// `telem_*` metric — including `telem_proc_cores_max`, the input to the
/// `core_saturation` gate — is numeric and survives.
pub fn bench_run_to_pb(run: &nornir_bench::BenchRun) -> pb::BenchRun {
pb::BenchRun {
date: run.date.clone(),
timestamp: run.timestamp.clone().unwrap_or_default(),
version: run.version.clone(),
machine: run.machine.clone(),
cores: run.cores,
results: run
.results
.iter()
.map(|r| pb::BenchResult {
name: r.name.clone(),
metrics: r
.metrics
.iter()
.filter_map(|(k, v)| {
v.as_f64().map(|f| pb::Kvf {
key: k.clone(),
value: f,
})
})
.collect(),
})
.collect(),
tests: run
.tests
.iter()
.map(|t| pb::TestOutcome {
name: t.name.clone(),
passed: t.passed,
duration_ms: t.duration_ms.unwrap_or(0.0),
has_duration: t.duration_ms.is_some(),
message: t.message.clone().unwrap_or_default(),
})
.collect(),
// **The producer's own identity and the box's facts, carried.** Both
// are fields 8 and 9 of the warehouse's `BenchRun` (nornir-wire's
// nornir.proto) and this copy of the proto was simply behind. Dropping
// `id` here would put the run back where it was before 2026-09-20: the
// server mints one, a re-submit makes a SECOND row, and a failed
// submit cannot be retried. A nil id serialises to empty, which is
// exactly what tells the store "this one cannot be deduplicated".
id: if run.has_id() {
run.id.to_string()
} else {
Default::default()
},
machine_facts_json: run
.machine_facts
.as_ref()
.and_then(|m| serde_json::to_string(m).ok())
.unwrap_or_default(),
}
}
/// **Submit one finished `BenchRun` over `Bench.Submit`** — the forge's write leg
/// into nornir's `bench_runs` / `bench_results` / `bench_telemetry` tables.
///
/// The server holds the redb catalog lock, so the forge cannot append to the
/// Iceberg warehouse itself; it ships the run and the server appends it. Returns
/// the server-assigned run id — a receipt, never a swallowed `()`, because a
/// measurement that did not land must not read as one that did.
///
/// Same transport contract as [`submit_test_results`]: bearer, the
/// `nornir-workspace` header, a 3s connect timeout, its own OS thread.
pub fn submit_bench_run(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
run: &nornir_bench::BenchRun,
) -> Result<String> {
let repo = repo.to_string();
let pb_run = bench_run_to_pb(run);
with_bench_client(endpoint, token, workspace, |client| {
Box::pin(async move {
let run_id = client
.submit(pb::SubmitBenchRequest {
repo,
run: Some(pb_run),
})
.await
.context("Bench.Submit RPC")?
.into_inner()
.run_id;
Ok(run_id)
})
})
}
/// **Map a wire [`pb::BenchRun`] back onto the domain type** — the inverse of
/// [`bench_run_to_pb`], and pure for the same reason (the mapping must be
/// unit-testable off the network).
///
/// The proto flattens two `Option`s into "empty/zero means absent", so this maps
/// them back honestly: an empty `timestamp` is `None` (never `Some("")`), and a
/// `TestOutcome` whose `has_duration` is false comes back `duration_ms: None`
/// even though the wire carried `0.0` — that flag exists precisely so "0 ms" and
/// "not recorded" survive the round trip as different values. Two losses are
/// structural and inherited from the WRITE mapping, not introduced here:
/// `BenchResult::source` has no proto field (every row reads back `Measured`),
/// and `KVF.value` is a `double` (non-numeric metrics never crossed; a non-finite
/// double has no JSON number, so it is dropped rather than smuggled as `null`).
pub fn bench_run_from_pb(run: pb::BenchRun) -> nornir_bench::BenchRun {
nornir_bench::BenchRun {
// The inverse, and honest about a wire that carried neither: an empty
// `id` is `Uuid::nil` — "legacy, no identity", which `has_id` reports
// and which consumers must join heuristically rather than deduplicate
// — and `machine_facts_json` that is empty or that does not parse is
// `None`, never a half-built `Machine`.
id: run.id.parse().unwrap_or_default(),
machine_facts: (!run.machine_facts_json.is_empty())
.then(|| serde_json::from_str(&run.machine_facts_json).ok())
.flatten(),
date: run.date,
timestamp: (!run.timestamp.is_empty()).then_some(run.timestamp),
version: run.version,
machine: run.machine,
cores: run.cores,
results: run
.results
.into_iter()
.map(|r| nornir_bench::BenchResult {
name: r.name,
source: nornir_bench::BenchSource::Measured,
metrics: r
.metrics
.into_iter()
.filter_map(|kv| {
serde_json::Number::from_f64(kv.value)
.map(|n| (kv.key, serde_json::Value::Number(n)))
})
.collect(),
})
.collect(),
tests: run
.tests
.into_iter()
.map(|t| nornir_bench::TestOutcome {
name: t.name,
passed: t.passed,
duration_ms: t.has_duration.then_some(t.duration_ms),
message: (!t.message.is_empty()).then_some(t.message),
})
.collect(),
}
}
/// **Read a repo's recorded runs back over `Bench.History`** — the forge's READ
/// leg out of nornir's `bench_runs` warehouse tables, and the data the doc-book
/// MASHUP renderer composes into the docs.
///
/// The doctrine so far was write-only: the forge runs the measurement
/// (`dwarves bench run`), the warehouse keeps the numbers, and reading them back
/// was `nornir bench history-show`. The mashup renderer moving into dwarves is
/// what changes that — a docs verb that composes bench numbers needs to FETCH
/// them, and the host-only law says it must not grow a `nornir` link to do it.
/// So this is the same thin-client shape as [`submit_bench_run`]: the query
/// itself stays nornir's; dwarves only carries the call.
///
/// A repo with no recorded runs is an EMPTY VEC, never an error — "measured
/// nothing yet" is a normal state for the mashup to render. An unreachable or
/// refusing server IS an error (never a silent empty), same as every fetcher
/// here. Same transport contract too: bearer, the `nornir-workspace` header, a
/// 3s connect timeout, its own OS thread.
pub fn fetch_bench_history(
endpoint: &str,
token: &str,
workspace: &str,
repo: &str,
) -> Result<Vec<nornir_bench::BenchRun>> {
let repo = repo.to_string();
with_bench_client(endpoint, token, workspace, |client| {
Box::pin(async move {
let resp = client
.history(pb::RepoOnly { repo })
.await
.context("Bench.History RPC")?
.into_inner();
Ok(resp.runs.into_iter().map(bench_run_from_pb).collect())
})
})
}
/// The boxed request interceptor every nornir client here is built with
/// (bearer + `nornir-workspace` header).
pub type NornirInterceptor =
Box<dyn FnMut(tonic::Request<()>) -> Result<tonic::Request<()>, tonic::Status> + Send>;
/// The intercepted `Telemetry` client type (bearer + `nornir-workspace` header).
type TelemetryClient = pb::telemetry_client::TelemetryClient<
tonic::service::interceptor::InterceptedService<tonic::transport::Channel, NornirInterceptor>,
>;
/// The intercepted `Bench` client type — same channel, same interceptor, a
/// different generated stub.
type BenchClient = pb::bench_client::BenchClient<
tonic::service::interceptor::InterceptedService<tonic::transport::Channel, NornirInterceptor>,
>;
/// **The ONE connect-and-call body** for every submit in this crate.
///
/// Runs on a DEDICATED OS thread with its own current-thread runtime — safe from
/// ANY context, sync or nested inside an existing tokio runtime (the
/// dwarves-server serve loop) without the "runtime within a runtime" panic.
/// Bearer + `nornir-workspace` header + a 3s connect timeout, so a dead server can
/// never wedge a build.
///
/// It hands the closure the connected `Channel` and the built interceptor rather
/// than a finished client, because the CLIENT is the only thing that differs
/// between `Telemetry` and `Bench`. Copying the thread/runtime/bearer/timeout
/// dance per service is exactly the twin (L5) — and the twin is how the two build
/// tables ended up with a write path nobody exercised.
fn with_nornir_channel<T, Fut, F>(endpoint: &str, token: &str, workspace: &str, f: F) -> Result<T>
where
T: Send + 'static,
Fut: std::future::Future<Output = Result<T>>,
F: FnOnce(tonic::transport::Channel, NornirInterceptor) -> Fut + 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()
.context("build tokio runtime for warehouse submit")?;
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).context("parse bearer token")?;
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())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect_timeout(std::time::Duration::from_secs(3))
.timeout(crate::READ_TIMEOUT)
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let interceptor: 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)
});
f(channel, interceptor).await
})
})
.join()
.map_err(|_| anyhow::anyhow!("warehouse-submit thread panicked"))?
})
}
/// [`with_nornir_channel`] with the `Telemetry` stub on top — what the three
/// `Telemetry` submits call.
fn with_telemetry_client<T, F>(endpoint: &str, token: &str, workspace: &str, f: F) -> Result<T>
where
T: Send + 'static,
F: for<'a> FnOnce(
&'a mut TelemetryClient,
)
-> std::pin::Pin<Box<dyn std::future::Future<Output = Result<T>> + 'a>>
+ Send,
{
with_nornir_channel(
endpoint,
token,
workspace,
move |channel, interceptor| async move {
let mut client =
pb::telemetry_client::TelemetryClient::with_interceptor(channel, interceptor)
.max_decoding_message_size(64 << 20);
f(&mut client).await
},
)
}
/// [`with_nornir_channel`] with the `Bench` stub on top.
fn with_bench_client<T, F>(endpoint: &str, token: &str, workspace: &str, f: F) -> Result<T>
where
T: Send + 'static,
F: for<'a> FnOnce(
&'a mut BenchClient,
)
-> std::pin::Pin<Box<dyn std::future::Future<Output = Result<T>> + 'a>>
+ Send,
{
with_nornir_channel(
endpoint,
token,
workspace,
move |channel, interceptor| async move {
let mut client = pb::bench_client::BenchClient::with_interceptor(channel, interceptor)
.max_decoding_message_size(64 << 20);
f(&mut client).await
},
)
}
// ═════════════════════════════════════════════════════════════════════════════
// The BUILD-RECORD PRODUCER — the thing that was missing.
// ═════════════════════════════════════════════════════════════════════════════
/// **The receipt of a build's landing in the warehouse.** Never a swallowed `()`.
///
/// The whole reason ⚒ Build Thing came up empty four times is that the write side
/// had no producer AND no observable outcome: nothing said "this build recorded
/// nothing". A receipt is a value the caller must carry into its report, and
/// [`Self::is_red`] is the guard condition.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct BuildRecordReceipt {
/// The endpoint that was submitted to. `None` ⇒ no server configured, so
/// nothing was even attempted (an honest no-op, NOT a failure).
pub endpoint: Option<String>,
/// The served workspace the rows were addressed to.
pub workspace: String,
/// `build_provenance` records the server accepted.
pub provenance_accepted: u32,
/// `build_info` rows the server accepted.
pub build_info_accepted: u32,
/// `build_info` rows the build actually produced (the denominator — a build that
/// emitted nothing is a different fault from a server that accepted nothing).
pub build_info_offered: u32,
/// The first transport/RPC failure, if any.
pub error: Option<String>,
}
impl BuildRecordReceipt {
/// True when a server WAS configured and this build nonetheless recorded
/// nothing — a submit error, or zero accepted rows in either table.
///
/// **This is the guard.** A build that compiles a binary and files no record of
/// having done so is the exact failure that hid for four rounds.
pub fn is_red(&self) -> bool {
self.endpoint.is_some()
&& (self.error.is_some()
|| self.provenance_accepted == 0
|| self.build_info_accepted == 0)
}
/// True when rows genuinely landed.
pub fn landed(&self) -> bool {
self.endpoint.is_some() && !self.is_red()
}
/// The one line a log / a pane shows. The three states are worded so they can
/// never be mistaken for each other.
pub fn line(&self) -> String {
match &self.endpoint {
None => "build records: NOT SUBMITTED — no nornir server configured \
(set NORNIR_SERVER); this build filed no provenance"
.to_string(),
Some(ep) => {
if let Some(e) = &self.error {
format!("⚠ build records: SUBMIT FAILED to {ep} — {e}")
} else if self.is_red() {
format!(
"⚠ build records: ZERO ROWS LANDED at {ep} \
(build_provenance={}, build_info={}/{} offered) — \
dwarves-buildthing-10",
self.provenance_accepted, self.build_info_accepted, self.build_info_offered,
)
} else {
format!(
"build records: {} provenance + {} build_info row(s) landed in \
`{}` at {ep}",
self.provenance_accepted,
self.build_info_accepted,
if self.workspace.is_empty() {
"<default>"
} else {
&self.workspace
},
)
}
}
}
}
/// The observable shape (L4) a pane/robot reads back.
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"endpoint": self.endpoint,
"workspace": self.workspace,
"provenance_accepted": self.provenance_accepted,
"build_info_accepted": self.build_info_accepted,
"build_info_offered": self.build_info_offered,
"error": self.error,
"red": self.is_red(),
"landed": self.landed(),
"line": self.line(),
})
}
}
/// A [`BuildInfoSink`] that ships each emitted build's rows to nornir-server's
/// warehouse over `Telemetry.SubmitTestResults`.
///
/// Construct it from the environment ([`NornirWarehouseSink::from_env`]) or
/// explicitly. With **no** endpoint it is a **no-op** — so a build on a box with
/// no `NORNIR_SERVER` set behaves exactly as before (local ledger only). A submit
/// failure is logged and swallowed: it is best-effort and NEVER fails the build.
#[derive(Clone, Debug)]
pub struct NornirWarehouseSink {
/// The `nornir-server` endpoint; `None` ⇒ no-op sink.
endpoint: Option<String>,
token: String,
workspace: String,
/// The repo the build ran in (the `test_results.repo` column).
repo: String,
/// The build's run id (the `test_results.run_id` column).
run_id: String,
}
impl NornirWarehouseSink {
/// A sink pointed at `endpoint` (`None` ⇒ no-op), tagging rows with `repo` +
/// `run_id`.
pub fn new(
endpoint: Option<String>,
token: impl Into<String>,
workspace: impl Into<String>,
repo: impl Into<String>,
run_id: impl Into<String>,
) -> Self {
Self {
endpoint: endpoint.filter(|e| !e.trim().is_empty()),
token: token.into(),
workspace: workspace.into(),
repo: repo.into(),
run_id: run_id.into(),
}
}
/// Build the sink from the environment (the uniform nornir mode switch):
/// `NORNIR_SERVER` (endpoint — unset ⇒ no-op), `NORNIR_SERVER_TOKEN` (bearer),
/// `NORNIR_WORKSPACE` (served workspace). Tags rows with `repo` + `run_id`.
pub fn from_env(repo: impl Into<String>, run_id: impl Into<String>) -> Self {
let endpoint = std::env::var("NORNIR_SERVER")
.ok()
.filter(|e| !e.is_empty());
let token = std::env::var("NORNIR_SERVER_TOKEN").unwrap_or_default();
let workspace = std::env::var("NORNIR_WORKSPACE").unwrap_or_default();
Self::new(endpoint, token, workspace, repo, run_id)
}
/// **Bind the warehouse writes to the SELECTED warehouse** (the selected-warehouse
/// spine): a dwarves release/write targets the ACTIVE warehouse the user picked in
/// the selector, not just whatever `$NORNIR_WORKSPACE` happens to be. The endpoint
/// and token still come from the environment (`NORNIR_SERVER` / `NORNIR_SERVER_TOKEN`),
/// but the `workspace` the rows land in is the passed `selected` name. An empty
/// `selected` yields an empty workspace — the server then rejects (or defaults),
/// and the release UI should refuse to write with no warehouse selected. Tags rows
/// with `repo` + `run_id`.
pub fn from_selected_warehouse(
selected: impl Into<String>,
repo: impl Into<String>,
run_id: impl Into<String>,
) -> Self {
let endpoint = std::env::var("NORNIR_SERVER")
.ok()
.filter(|e| !e.is_empty());
let token = std::env::var("NORNIR_SERVER_TOKEN").unwrap_or_default();
Self::new(endpoint, token, selected, repo, run_id)
}
/// The warehouse/workspace this sink writes to (the selected warehouse) — the
/// robot-observable proof that a release/write binds to the selection.
pub fn workspace(&self) -> &str {
&self.workspace
}
/// True when an endpoint is configured (i.e. the sink will actually submit).
pub fn is_active(&self) -> bool {
self.endpoint.is_some()
}
/// **THE PRODUCER.** Land a finished build in the warehouse: the
/// [`BuildProvenance`] record over `Telemetry.SubmitBuildProvenance` and the §f3
/// [`BuildInfoRow`] stream over `Telemetry.SubmitBuildInfo`.
///
/// Call this once per build, from wherever builds genuinely execute
/// (`dwarves::build_thing::BuildRun` — the ⚒ Build Thing ▶ Build worker — and
/// `dwarves::runner::RealRunner` — the forge's cargo step). It returns a
/// [`BuildRecordReceipt`]; it does NOT return `Result`, because a submit failure
/// must never fail the build (L2) — but it must never be invisible either, which
/// is what the receipt is for.
///
/// With no endpoint configured this is a no-op that says so
/// ([`BuildRecordReceipt::endpoint`] is `None`) — distinguishable from a submit
/// that reached a server and landed nothing.
pub fn record_build(
&self,
prov: &BuildProvenance,
rows: &[BuildInfoRow],
) -> BuildRecordReceipt {
let mut receipt = BuildRecordReceipt {
endpoint: self.endpoint.clone(),
workspace: self.workspace.clone(),
build_info_offered: rows.len() as u32,
..Default::default()
};
let Some(endpoint) = &self.endpoint else {
return receipt;
};
match prov.to_json() {
Ok(json) => {
match submit_build_provenance(endpoint, &self.token, &self.workspace, vec![json]) {
Ok(n) => receipt.provenance_accepted = n,
Err(e) => receipt.error = Some(format!("{e:#}")),
}
}
// A record that will not serialize is a producer bug, not a transport
// one — it still has to be visible rather than dropped.
Err(e) => receipt.error = Some(format!("serialize BuildProvenance: {e:#}")),
}
if !rows.is_empty() {
match submit_build_info(endpoint, &self.token, &self.workspace, rows) {
Ok(n) => receipt.build_info_accepted = n,
Err(e) => {
if receipt.error.is_none() {
receipt.error = Some(format!("{e:#}"));
}
}
}
}
// One line on the build's own stderr, whichever way it went — the operator
// sees the landing (or its absence) without opening a warehouse.
eprintln!("dwarves-warehouse: {}", receipt.line());
receipt
}
}
impl BuildInfoSink for NornirWarehouseSink {
fn emit(&self, rows: &[BuildInfoRow]) -> Result<()> {
// No endpoint ⇒ no-op (local/embedded behaviour unchanged).
let Some(endpoint) = &self.endpoint else {
return Ok(());
};
if rows.is_empty() {
return Ok(());
}
let test_rows = build_info_to_test_rows(rows, &self.repo, &self.run_id);
let n = test_rows.len();
match submit_test_results(endpoint, &self.token, &self.workspace, test_rows) {
Ok(accepted) => {
eprintln!(
"dwarves-warehouse: submitted {accepted}/{n} build-info row(s) to {endpoint} (test_results)"
);
}
Err(e) => {
// BEST-EFFORT (L2): a submit failure never fails the build.
eprintln!(
"dwarves-warehouse: submit to {endpoint} failed (best-effort, build unaffected): {e:#}"
);
}
}
Ok(())
}
}
// ═════════════════════════════════════════════════════════════════════════════
// "Add warehouse (SSH) → Populate" — the dwarves control surface over nornir's
// `Workspaces.Register` (add + eager-populate) + `Workspaces.Fetch` (re-sync).
//
// ARCHITECTURE (amended doctrine): the warehouse is NORNIR's. dwarves' UI lets a
// user ADD a warehouse from an SSH git remote and POPULATE it, but the warehouse-
// WRITE (clone the SSH members, build the Iceberg warehouse) happens in nornir —
// dwarves is the thin gRPC CLIENT. This is the same `<guest>-embed` shape as the
// Telemetry submit above; NO `nornir` crate dep (no cycle).
// ═════════════════════════════════════════════════════════════════════════════
/// Validate an SSH git remote (LAW L6 — SSH-ONLY remotes). Accepts scp-style
/// `git@host:owner/repo(.git)` and `ssh://git@host[:port]/owner/repo(.git)`; REJECTS
/// `http(s)://…` and anything without an `owner/repo` path — so the "Add warehouse"
/// form can never register an HTTPS remote.
pub fn validate_ssh_remote(url: &str) -> Result<()> {
let u = url.trim();
if u.is_empty() {
anyhow::bail!("SSH remote must not be empty");
}
if u.starts_with("http://") || u.starts_with("https://") {
anyhow::bail!(
"HTTPS remotes are forbidden (LAW L6, SSH-only) — use git@host:owner/repo.git"
);
}
if let Some(rest) = u.strip_prefix("ssh://") {
// ssh://git@host[:port]/owner/repo(.git)
let after_host = rest.split_once('/').map(|x| x.1).unwrap_or("");
if after_host.trim_matches('/').is_empty() {
anyhow::bail!("SSH remote `{u}` has no owner/repo path");
}
return Ok(());
}
// scp-style: user@host:path — must have a single ':' separating host and path.
let Some((host, path)) = u.split_once(':') else {
anyhow::bail!("`{u}` is not an SSH remote (expected git@host:owner/repo.git or ssh://…)");
};
if !host.contains('@') {
anyhow::bail!("SSH remote `{u}` needs a user (e.g. git@{host})");
}
if path.trim_matches('/').is_empty() {
anyhow::bail!("SSH remote `{u}` has no owner/repo path");
}
Ok(())
}
/// Derive a bare member/repo name from an SSH remote's `owner/repo(.git)` tail
/// (e.g. `ssh://git@git.gunnar.rs:2222/nordisk/dwarves.git` → `dwarves`). Falls back to
/// `"repo"` if no usable segment is found.
pub fn repo_name_from_ssh(url: &str) -> String {
let tail = url.rsplit(['/', ':']).next().unwrap_or("");
let name = tail.strip_suffix(".git").unwrap_or(tail);
let name = name.trim().trim_matches('/');
if name.is_empty() {
"repo".to_string()
} else {
name.to_string()
}
}
/// Derive a SHORT workspace name from an SSH remote — the suggestion the Add form
/// pre-fills into `name` as the user types the url. The member/repo derivation
/// ([`repo_name_from_ssh`]) stays untouched: the descriptor must keep the repo's real
/// name (`workspace_nordisk`), while the WORKSPACE wants the short one (`nordisk`),
/// so a `workspace_`/`workspace-` prefix is stripped here and only here.
pub fn workspace_name_from_ssh(url: &str) -> String {
let repo = repo_name_from_ssh(url);
let short = repo
.strip_prefix("workspace_")
.or_else(|| repo.strip_prefix("workspace-"))
.unwrap_or(&repo);
if short.is_empty() {
repo.clone()
} else {
short.to_string()
}
}
/// Synthesize a minimal `nornir-workspace.toml` descriptor for a single SSH member —
/// the content dwarves ships to nornir via `Register{descriptor_content}` (the
/// server writes it to its own descriptors dir and clones the member). One member,
/// declared by its `git =` SSH remote (a `Git` source the monitor clones).
pub fn descriptor_for_ssh(workspace_name: &str, ssh_url: &str) -> String {
let member = repo_name_from_ssh(ssh_url);
format!("[workspace]\nname = \"{workspace_name}\"\n\n[repos.{member}]\ngit = \"{ssh_url}\"\n")
}
/// A workspace record as the dwarves UI sees it (the thin summary of nornir's
/// `WorkspaceRecord`).
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct RegisteredWorkspace {
pub name: String,
pub mode: String,
pub current_snapshot: String,
pub members: Vec<String>,
}
impl RegisteredWorkspace {
fn from_pb(r: pb::WorkspaceRecord) -> Self {
Self {
name: r.name,
mode: r.mode,
current_snapshot: r.current_snapshot,
members: r.members.into_iter().map(|m| m.name).collect(),
}
}
/// True once the workspace has built a warehouse snapshot (populated).
pub fn is_populated(&self) -> bool {
!self.current_snapshot.trim().is_empty()
}
}
/// The outcome of a `Populate/Sync` (`Workspaces.Fetch`).
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct FetchOutcome {
pub workspace: String,
pub fetched: u32,
pub changed: Vec<String>,
pub errors: Vec<String>,
pub snapshot: String,
}
/// **The bound on a roster READ — the boot-time call the window waits behind.**
///
/// NOT applied to the whole seam: `add_warehouse_by_ssh` / `populate_warehouse` come
/// through the same helper and legitimately run for MINUTES (nornir clones and builds
/// the members), so a blanket request timeout here would abort real work. This bounds
/// the calls that are pure reads of a registry, which are fast or broken.
const ROSTER_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
/// **The bound on a READ, which `connect_timeout` is not.** A server that ACCEPTS the
/// socket and then never answers clears the connect bound instantly and then waits
/// forever — the shape that left `dwarves-ui-gpu` with no window at all (see
/// `nornir-warehouse-embed/tests/roster_does_not_hang.rs`). Every seam below pairs the
/// connect bound with a request bound so an unresponsive host degrades to each pane's
/// honest unavailable state instead of hanging the caller.
pub(crate) const READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
/// A `Warehouse.Scan` legitimately moves far more than a summary read, so it gets its
/// own, larger bound. Larger — never absent: a scan that has not answered in two
/// minutes is a broken host, not a slow one.
pub(crate) const SCAN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);
/// Run a blocking gRPC call against nornir-server on a DEDICATED OS thread (never a
/// nested-runtime panic), with the bearer + `nornir-workspace` header, 3s connect
/// timeout. The closure gets a connected `WorkspacesClient`.
///
/// `request_timeout` bounds EACH REQUEST, which `connect_timeout` does not: a server
/// that ACCEPTS the socket and then never answers passes the connect bound instantly
/// and then hangs forever. That is not hypothetical — see
/// `tests/roster_does_not_hang.rs` for the measured oden case that left
/// `dwarves-ui-gpu` with no window at all. `None` ⇒ unbounded, for the write calls
/// (register/fetch) whose real work outlasts any read-shaped bound.
fn with_workspaces_client<T, F>(
endpoint: &str,
token: &str,
workspace: &str,
request_timeout: Option<std::time::Duration>,
f: F,
) -> Result<T>
where
T: Send + 'static,
F: for<'a> FnOnce(
&'a mut pb::workspaces_client::WorkspacesClient<
tonic::service::interceptor::InterceptedService<
tonic::transport::Channel,
NornirInterceptor,
>,
>,
)
-> std::pin::Pin<Box<dyn std::future::Future<Output = Result<T>> + '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()
.context("build tokio runtime for workspaces RPC")?;
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).context("parse bearer token")?;
let ws_md: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> =
(!workspace.is_empty())
.then(|| workspace.parse().ok())
.flatten();
let mut ep = tonic::transport::Channel::from_shared(endpoint.clone())
.with_context(|| format!("invalid server url `{endpoint}`"))?
.connect_timeout(std::time::Duration::from_secs(3));
if let Some(t) = request_timeout {
ep = ep.timeout(t);
}
let channel = ep
.connect()
.await
.with_context(|| format!("connect to nornir-server at {endpoint}"))?;
let interceptor: 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::workspaces_client::WorkspacesClient::with_interceptor(channel, interceptor)
.max_decoding_message_size(64 << 20);
f(&mut client).await
})
})
.join()
.map_err(|_| anyhow::anyhow!("workspaces RPC thread panicked"))?
})
}
/// **The name of the diagnostic RPC to reach for when nothing else can say why.**
/// `nornir workspace events <name>` / `Viz.CloneEvents` is the way to see *why* a
/// member did not populate (auth, vanished repo, network): one `clone_events` row per
/// member ATTEMPT with op / status / error / when.
///
/// ── 2026-08-27: it was a PROXY gap, not a server gap ──────────────────────────
///
/// This constant's name and its old doc both said the RPC "does not answer … returns
/// `status: Unimplemented` server-side (session-resume-2026-07-23-oden.md §9.2)".
/// MEASURED on oden against ONE process, seconds apart:
///
/// `--server http://127.0.0.1:7878` (the nornir embedded in the forge) → 20 REAL ROWS
/// `--server http://127.0.0.1:7900` (the forge's own port) → Unimplemented
///
/// nornir-server had implemented it all along. What did not carry it was the dwarves
/// forge's `Viz` PROXY, whose surface is the trimmed
/// `nornir-warehouse-embed/proto/viz.proto` — one rpc, `Jobs`. So the diagnosis was
/// unreachable for exactly the clients that see only the forge port, which is
/// `dwarves-ui --server <forge>`: the deployed thin UI. The rpc is now declared in that
/// proto and forwarded in `dwarves::warehouse_proxy` (`clone_events`).
///
/// The constant stays, and so does the pane naming it — a populate can still land with
/// no snapshot, no member error and nothing to say, and pointing at the RPC that WOULD
/// answer is still better than shrugging. What changed is the sentence: it no longer
/// blames another product for a gap that was here, which is the failure mode worth
/// remembering — a stale attribution PAINTED ON SCREEN outlives the gap it described.
pub const MISSING_DIAGNOSTIC_RPC: &str = "Viz.CloneEvents";
/// **The verdict of a WHOLE `add` action** — register *and* populate, judged together.
///
/// The bug this type exists to prevent: `workspace add` printing
/// `✓ populated now (eager register)` over an empty `git/`. The descriptor fetch
/// succeeded and the member clone failed, and the command reported green. So the
/// outcome is not "did Register return a record" — a record always comes back — it is
/// **"is there a warehouse snapshot"**, and if not, **why not**.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct AddOutcome {
/// The record `Workspaces.Register` returned.
pub registered: RegisteredWorkspace,
/// The populate report, when one was needed to reach or diagnose a snapshot.
pub fetch: Option<FetchOutcome>,
/// Per-member failures ("member: message"), from the fetch report.
pub errors: Vec<String>,
/// True iff a warehouse snapshot exists at the end of the action. THIS is the
/// success condition — not "the RPC returned Ok".
pub populated: bool,
/// True when the action failed and nothing available could say WHY: no member
/// error, no snapshot. The pane must show [`MISSING_DIAGNOSTIC_RPC`] here rather
/// than invent a cause or draw the silence as success.
pub cause_unknown: bool,
}
impl AddOutcome {
/// A one-line human verdict. Never green unless [`Self::populated`].
pub fn describe(&self) -> String {
if self.populated {
return format!(
"ok: added `{}` — {} member(s), populated (snapshot {})",
self.registered.name,
self.registered.members.len(),
if self.registered.current_snapshot.is_empty() {
self.fetch
.as_ref()
.map(|f| f.snapshot.clone())
.unwrap_or_default()
} else {
self.registered.current_snapshot.clone()
},
);
}
if !self.errors.is_empty() {
return format!(
"error: `{}` registered but NOT populated — {}",
self.registered.name,
self.errors.join("; ")
);
}
format!(
"error: `{}` registered but NOT populated, and this action returned neither \
a snapshot nor a member error — ask `{MISSING_DIAGNOSTIC_RPC}` (behind \
`nornir workspace events <name>`) for the per-member populate outcome",
self.registered.name,
)
}
}
/// **Add a warehouse from an SSH git remote — the WHOLE action.** Register *and*
/// populate, judged as one: there is no half-done "registered" state to report green.
///
/// Validates the SSH URL (L6), synthesizes a one-member descriptor, ships it as
/// `descriptor_content`, and registers it MONITORED + eager (`lazy=false`) so nornir
/// clones the member + builds the warehouse inline. Then it **checks that a snapshot
/// actually exists**; if it does not, it runs a `Fetch{force}` purely to obtain the
/// per-member `errors` list — the only diagnostic the served surface can give — and
/// reports THAT. A member clone that failed therefore fails the action.
///
/// Returns `Err` only when the wire call itself failed. A registered-but-empty
/// warehouse comes back as `Ok(AddOutcome { populated: false, .. })`, which the caller
/// must render as a failure — see [`AddOutcome::describe`].
pub fn add_warehouse_by_ssh(
endpoint: &str,
token: &str,
name: &str,
ssh_url: &str,
) -> Result<AddOutcome> {
let rec = register_warehouse_by_ssh(endpoint, token, name, ssh_url)?;
// THE SUCCESS CONDITION: a snapshot. `Register` returning a record is not one.
if rec.is_populated() {
return Ok(AddOutcome {
populated: true,
registered: rec,
fetch: None,
errors: Vec::new(),
cause_unknown: false,
});
}
// No snapshot ⇒ the populate leg did not happen or did not finish. Ask for the one
// diagnostic the served surface has: the fetch report's per-member error list.
// `force` so a workspace whose members did not move is still rebuilt (the empty
// warehouse is exactly the case where nothing moved).
match populate_warehouse(endpoint, token, &rec.name, true) {
Ok(out) => {
let populated = !out.snapshot.trim().is_empty();
let errors = out.errors.clone();
Ok(AddOutcome {
populated,
cause_unknown: !populated && errors.is_empty(),
registered: rec,
fetch: Some(out),
errors,
})
}
Err(e) => Ok(AddOutcome {
errors: vec![format!("populate: {e:#}")],
registered: rec,
fetch: None,
populated: false,
cause_unknown: false,
}),
}
}
/// The bare `Workspaces.Register` leg of [`add_warehouse_by_ssh`] — registers eagerly
/// and returns whatever record the server produced, WITHOUT judging it. Split out so
/// the judging lives in exactly one place (the whole action above); a caller that
/// wants the raw record can still have it.
pub fn register_warehouse_by_ssh(
endpoint: &str,
token: &str,
name: &str,
ssh_url: &str,
) -> Result<RegisteredWorkspace> {
validate_ssh_remote(ssh_url)?;
if name.trim().is_empty() {
anyhow::bail!("warehouse name must not be empty");
}
// ── A WORKSPACE-DESCRIPTOR REPO IS NOT A MEMBER ──────────────────────────────
//
// The URL is sent as `descriptor` FIRST, so the server clones the repo and reads a
// real `nornir-workspace.toml` out of it if one is there. Only when that fails do we
// fall back to `descriptor_for_ssh`, which synthesises a one-member descriptor from
// the URL.
//
// The fallback used to be the ONLY path, and it silently did the wrong thing for the
// exact repo the whole scheme is built around. Registering
// `ssh://git@forge.example/nordisk/workspace_nordisk.git` on 2026-08-08 produced a
// workspace whose single member was `workspace_nordisk` itself — so nornir cloned the
// descriptor repo, ran `rust-analyzer scip` over a pile of markdown, and reported a
// failed ingest. That repo does not BELONG to the workspace; it DECLARES it, in a
// `nornir-workspace.toml` naming fourteen `[repos.*]` (nornir, holger, modgunn,
// funnel-core, …). Its own header documents the intended registration:
// `nornir workspace register nordisk --descriptor nornir-workspace.toml --monitored`.
//
// The server has resolved a git-URL `descriptor` this way all along
// (`resolve_descriptor_content`: clone, then read `nornir-workspace.toml`); this
// client simply never asked it to, because it always pre-empted the question by
// shipping `descriptor_content`.
//
// Order matters and is deliberate: try the repo's OWN declaration, then synthesise.
// The other order would let a synthesised single-member descriptor mask a real one,
// which is the bug being fixed. Both legs stay because both are legitimate — "add
// this one repo as a warehouse" and "add the workspace this repo declares" arrive
// through the same box, and only the repo's contents can tell them apart.
let synthesised = descriptor_for_ssh(name, ssh_url);
let name = name.to_string();
let ws = name.clone();
let url = ssh_url.to_string();
let by_descriptor = {
let name = name.clone();
let url = url.clone();
with_workspaces_client(endpoint, token, &ws, None, move |client| {
let name = name.clone();
let url = url.clone();
Box::pin(async move {
let rec = client
.register(pb::RegisterWorkspaceRequest {
name,
// The git URL: the server clones it and reads its
// `nornir-workspace.toml`.
descriptor: url,
mode: "monitored".to_string(),
poll: String::new(),
lazy: false, // EAGER: clone + build the warehouse now
descriptor_content: String::new(),
})
.await
.context("Workspaces.Register RPC")?
.into_inner();
Ok(RegisteredWorkspace::from_pb(rec))
})
})
};
match by_descriptor {
Ok(rec) => Ok(rec),
// No readable `nornir-workspace.toml` in that repo — so it is an ordinary member
// repo, and the synthesised single-member descriptor is the right answer after
// all. The first error is dropped deliberately: it says "this repo declares no
// workspace", which is not a failure for the one-repo case.
Err(_) => {
let ws = name.clone();
with_workspaces_client(endpoint, token, &ws, None, move |client| {
let name = name.clone();
let descriptor_content = synthesised.clone();
Box::pin(async move {
let rec = client
.register(pb::RegisterWorkspaceRequest {
name,
descriptor: String::new(),
mode: "monitored".to_string(),
poll: String::new(),
lazy: false,
descriptor_content,
})
.await
.context("Workspaces.Register RPC")?
.into_inner();
Ok(RegisteredWorkspace::from_pb(rec))
})
})
}
}
}
/// **Register a workspace from a descriptor's TEXT** (`Workspaces.Register` with
/// `descriptor_content`), recorded LAZILY: the server writes the descriptor to its own
/// descriptors dir and its poll loop populates the members — the call does not clone
/// or build anything inline. What a Dwarves project register uses for a project
/// workspace the registry does not hold yet.
pub fn register_workspace_descriptor(
endpoint: &str,
token: &str,
name: &str,
descriptor_content: &str,
mode: &str,
) -> Result<RegisteredWorkspace> {
if name.trim().is_empty() {
anyhow::bail!("workspace name must not be empty");
}
let (name, content, mode) = (
name.to_string(),
descriptor_content.to_string(),
mode.to_string(),
);
let ws = name.clone();
with_workspaces_client(endpoint, token, &ws, None, move |client| {
Box::pin(async move {
let rec = client
.register(pb::RegisterWorkspaceRequest {
name,
descriptor: String::new(),
mode,
poll: String::new(),
lazy: true,
descriptor_content: content,
})
.await
.context("Workspaces.Register RPC")?
.into_inner();
Ok(RegisteredWorkspace::from_pb(rec))
})
})
}
/// **Populate / re-sync a warehouse** — `Workspaces.Fetch { force }`. Re-clones the
/// members and rebuilds the warehouse; returns what changed + the new snapshot. This
/// is the "Populate" button; it also fixes the funnel's `not_served` (once populated,
/// nornir serves the workspace and the funnel loads a real DAG).
pub fn populate_warehouse(
endpoint: &str,
token: &str,
name: &str,
force: bool,
) -> Result<FetchOutcome> {
let name = name.to_string();
let ws = name.clone();
with_workspaces_client(endpoint, token, &ws, None, move |client| {
let name = name.clone();
Box::pin(async move {
let rep = client
.fetch(pb::WorkspaceFetchRequest {
name: name.clone(),
force,
background: false,
})
.await
.context("Workspaces.Fetch RPC")?
.into_inner();
Ok(FetchOutcome {
workspace: rep.workspace,
fetched: rep.fetched,
changed: rep.changed,
errors: rep.errors,
snapshot: rep.snapshot,
})
})
})
}
/// **What ONE Refresh did** — the client-side mirror of nornir's
/// `WorkspaceRepopulateReport`, kept as a type so the pane renders facts rather
/// than re-deriving a sentence from loose fields.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct RepopulateOutcome {
pub workspace: String,
/// Whether a purge was asked for.
pub purged: bool,
/// What the purge dropped, named: the ORPHANED v1 tables still in the warehouse
/// (`symbol_facts`, `call_edges`, `lens_*` … — nothing writes them since Mímir v2).
/// Never the code room's lake. A second purge drops nothing.
pub dropped: Vec<String>,
/// Which state is served, in nornir's words: the lake's identity, `member@seq sha`
/// per consumed member, and the room a reader gets now.
pub generation: String,
/// The members BEHIND their offset, one line each — a consume for them was started
/// by this call. Empty when every member is at its offset (not a refusal).
pub queued: Vec<String>,
/// What the consumer is on right now (`member seq N @ sha`).
pub in_flight: Vec<String>,
/// nornir's NAMED refusal, verbatim, when nothing can be consumed: the driver is
/// not installed, the lake is held elsewhere, the workspace has no code room.
/// Empty on success.
pub refusal: String,
}
impl RepopulateOutcome {
/// True when the server refused (the dedup half — a gate collision arrives as
/// an `Err` from [`repopulate_warehouse`] instead).
pub fn is_refusal(&self) -> bool {
!self.refusal.is_empty()
}
/// The ONE line the button shows. On a refusal it is nornir's own sentence,
/// **verbatim** — the server named why nothing can be consumed, and re-wording
/// that here is how two products come to describe one event differently.
pub fn describe(&self) -> String {
if self.is_refusal() {
return self.refusal.clone();
}
let dropped = if self.purged {
format!(" · {} orphaned v1 table(s) dropped", self.dropped.len())
} else {
String::new()
};
let behind = if self.queued.is_empty() {
"every member is at its offset, nothing to consume".to_string()
} else {
let lines: Vec<&str> = self.queued.iter().take(3).map(String::as_str).collect();
let more = self.queued.len().saturating_sub(lines.len());
let tail = if more > 0 {
format!(" (+{more} more)")
} else {
String::new()
};
format!(
"{} member(s) behind, consume started [{}{tail}]",
self.queued.len(),
lines.join("; ")
)
};
let on = if self.in_flight.is_empty() {
"the consumer is idle".to_string()
} else {
format!("the consumer is on {}", self.in_flight.join("; "))
};
format!(
"ok: `{}` serves {}{dropped} — {behind}; {on}",
self.workspace, self.generation
)
}
}
/// **THE ONE REFRESH.** `Workspaces.Repopulate { name, purge }` — clear-and-repopulate
/// the derived half of a served workspace's warehouse and re-enqueue the chain that
/// rebuilds it.
///
/// This is the whole reason the warehouse/Mímir panes need no per-pane Scan buttons.
/// nornir already AUTO-POPULATES — a landed snapshot submits one jera chain job per
/// lens, deduped by key and superseded by generation — so a pane full of "Scan"
/// buttons was N ways to ask for work the server schedules by itself, each with its
/// own half-answer. What was genuinely missing was one verb for *clear it and build
/// it again*, and this is it.
///
/// It is NOT `nornir warehouse reset`: that opens `catalog.redb` itself, so it
/// refuses whenever a server is up — precisely when a Refresh is wanted — and it
/// re-enqueues nothing.
///
/// **The two refusals reach the caller differently, and deliberately.** A populate
/// gate held mid-flight is `Err` carrying nornir's FAILED_PRECONDITION message with
/// the **in-flight job id** in it; a chain already queued at this generation is
/// `Ok(outcome)` with [`RepopulateOutcome::refusal`] set and `in_flight` naming those
/// jobs. The first means *nothing happened, try later*; the second means *the work
/// you asked for is already running*. A pane that collapsed them into one red would
/// be lying about one of them.
pub fn repopulate_warehouse(
endpoint: &str,
token: &str,
name: &str,
purge: bool,
) -> Result<RepopulateOutcome> {
let name = name.trim().to_string();
if name.is_empty() {
anyhow::bail!("no warehouse selected — pick one to refresh");
}
let ws = name.clone();
with_workspaces_client(endpoint, token, &ws, None, move |client| {
let name = name.clone();
Box::pin(async move {
let rep = client
.repopulate(pb::WorkspaceRepopulateRequest { name, purge, affected: false })
.await
// The status MESSAGE is nornir's named refusal (it carries the
// in-flight job id); `{e}` would bury it in tonic's Debug shape.
.map_err(|e| anyhow::anyhow!("Workspaces.Repopulate RPC: {}", e.message()))?
.into_inner();
Ok(RepopulateOutcome {
workspace: rep.workspace,
purged: rep.purged,
dropped: rep.dropped,
generation: rep.generation,
queued: rep.queued,
in_flight: rep.in_flight,
refusal: rep.refusal,
})
})
})
}
/// **Remove a warehouse — the WHOLE action.** `Workspaces.Remove { name, purge }` with
/// `purge = true`: the registry row AND the workspace's on-disk `builds/` (warehouse +
/// `catalog.redb` + `jobs.redb`) go together, so "remove" does not leave a registered
/// entry pointing at data nobody deletes.
///
/// **It never touches a credential.** The RPC's whole reach is the server's workspace
/// data; there is no field on it that names a key, a token or a vault, and dwarves'
/// own ssh key (`~/.dwarves/ssh/…`) and vault (`~/.dwarves/vault.bin`) live on the
/// CLIENT, which this call does not read or write at all. This mirrors nornir's
/// `--purge --keep-identity` rule (`nornir/src/bin/nornir.rs`, `purge_except`): clear
/// the data, keep the identity.
pub fn remove_warehouse(endpoint: &str, token: &str, name: &str) -> Result<()> {
let name = name.trim().to_string();
if name.is_empty() {
anyhow::bail!("no warehouse selected — pick one to remove");
}
let ws = name.clone();
with_workspaces_client(endpoint, token, &ws, None, move |client| {
let name = name.clone();
Box::pin(async move {
client
.remove(pb::WorkspaceName { name, purge: true })
.await
.context("Workspaces.Remove RPC")?;
Ok(())
})
})
}
// ═════════════════════════════════════════════════════════════════════════════
// POPULATE PROGRESS — a bar whenever population happens, whoever started it.
//
// A populate is minutes of work. The UI must show it running, and must show it
// running EVEN WHEN NO BUTTON HERE STARTED IT: nornir's own poll loop, another
// client, or `nornir workspace populate` on the command line all move the same
// warehouse this shell is bound to.
//
// The live signal is per-member and already on the wire: `WorkspaceMember.
// sync_state` is `"fetching"` while a member is being cloned/fetched, `"ok"`
// once it lands, `"error: …"` when it did not (nornir/src/monitor.rs). Polling
// `Workspaces.Get` therefore sees a populate no matter who launched it.
//
// THE ESTIMATE MOVES, AND THAT IS THE POINT. The member count is known
// immediately from the registry, so there is a denominator within one round
// trip. The crate count is not — it only exists once the warehouse has been
// rebuilt — so it arrives later and refines. A bar that sits at 0% and jumps to
// 100% is not honest; one whose denominator visibly improves is.
// ═════════════════════════════════════════════════════════════════════════════
/// A running counter that knows the difference between *not yet* and *never*.
///
/// The three states are three. Collapsing "not measured yet" into 0 is how a bar
/// sits at zero and then jumps; collapsing "this seam cannot measure it" into 0 is
/// how a fabricated metric gets drawn beside a real one.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(tag = "state", rename_all = "lowercase")]
pub enum Counter {
/// Measured.
Known { value: u64 },
/// Not measured YET — it will be, on a later poll.
Pending,
/// This seam CANNOT measure it. `why` says what is missing, by name.
Unavailable { why: &'static str },
}
impl Counter {
/// The measured value, if there is one.
pub fn value(&self) -> Option<u64> {
match self {
Counter::Known { value } => Some(*value),
_ => None,
}
}
/// A short label for the pane: the number, `…`, or `n/a`.
pub fn label(&self) -> String {
match self {
Counter::Known { value } => value.to_string(),
Counter::Pending => "…".to_string(),
Counter::Unavailable { .. } => "n/a".to_string(),
}
}
}
/// **Why the byte counter is `Unavailable`, named precisely.**
///
/// Checked against nornir at 2026-07-23: `WorkspaceFetchReport` carries
/// `{workspace, fetched, changed, errors, snapshot}`; `WorkspaceMember` carries
/// `{name, remote, git_ref, last_seen_sha, last_synced, sync_state,
/// worktree_digest, worktree_dirty}`; and the `clone_events` row is
/// `{ts_micros, workspace, member, remote, op, status, detail, elapsed_ms}`.
/// **None of the three has a byte field**, and nothing else on the served surface
/// counts clone bytes, so a thin client cannot report them without inventing them.
pub const BYTES_UNAVAILABLE_WHY: &str =
"no byte counter exists on the served surface: WorkspaceFetchReport has \
{fetched, changed, errors, snapshot}, WorkspaceMember has no size field, and a \
clone_events row ends at elapsed_ms. Adding one is a nornir-side change.";
/// **What a populate looks like from here, right now.**
///
/// Built entirely from RPCs that answer today, and no invented numbers. It uses
/// `Workspaces.PopulateStatus` (a roll-up) rather than `Viz.CloneEvents` (the
/// unpaginated per-attempt ledger) on purpose — see [`MISSING_DIAGNOSTIC_RPC`] for why
/// the latter is now reachable through the forge and why this still does not poll it.
#[derive(Debug, Clone, serde::Serialize)]
pub struct PopulateProgress {
pub workspace: String,
/// True iff at least one member is mid-fetch (`sync_state == "fetching"`) —
/// which is true whether or not a button in this process started it.
pub running: bool,
/// Members the registry declares. Known on the FIRST poll: the rough
/// denominator that makes the bar a proportion straight away.
pub repos_total: usize,
/// Members whose latest sync landed (`sync_state == "ok"`).
pub repos_done: usize,
/// Members currently being fetched.
pub repos_fetching: usize,
/// Members whose latest sync failed.
pub repos_failed: usize,
/// Distinct crates in the warehouse — the REFINING estimate. `Pending` until the
/// warehouse has a dependency graph to count, then a real number that grows.
pub crates: Counter,
/// **Whether the CODE GRAPH has been consumed** — nornir's consumer has released
/// at least one commit of this workspace into its code room.
///
/// A populate is three phases, not one: the members are cloned, then a warehouse is
/// built from them, then nornir's consumer feeds the code graph from the clones
/// (Mímir v2: the rustc driver, one release per commit). [`crates`](Self::crates)
/// (`dep_graph_edges`) is the warehouse-build's product and appears at the end of
/// phase two; this is the consumer's product and appears in phase three. Without it
/// the two are indistinguishable from here and a step bar would have to guess at the
/// transition — see [`crate::CODE_GRAPH_PROBE_WHY`].
///
/// A presence probe, deliberately: `Known { value: 1 }` means "a stamp exists", not
/// a count of anything, so it is never drawn as a fraction.
pub code_graph: Counter,
/// Bytes pulled in. Always [`Counter::Unavailable`] over this seam — see
/// [`BYTES_UNAVAILABLE_WHY`]. It is drawn as `n/a` with the reason, never as 0.
pub bytes: Counter,
/// The rolled-up verdict from `Workspaces.PopulateStatus`: `green` / `red` /
/// `stale`, or empty when the RPC did not answer.
pub verdict: String,
/// The newest sync timestamp across members (`never` when none).
pub last_synced: String,
/// The first failing member + its error, from the roll-up.
pub failing_member: String,
pub last_error: String,
}
impl PopulateProgress {
/// The bar's fraction, or `None` for an INDETERMINATE bar (no denominator yet).
/// Once a denominator exists this becomes a proportion — and the denominator is
/// allowed to move as more becomes known.
pub fn fraction(&self) -> Option<f32> {
if self.repos_total == 0 {
return None;
}
Some((self.repos_done as f32 / self.repos_total as f32).clamp(0.0, 1.0))
}
/// The counters line: what has actually come in, with `n/a` where this seam
/// cannot see and `…` where it has not looked yet.
pub fn counters_line(&self) -> String {
format!(
"{}/{} repo(s) · {} crate(s) · {} bytes",
self.repos_done,
self.repos_total,
self.crates.label(),
self.bytes.label(),
)
}
}
/// **Fetch a workspace record** (`Workspaces.Get`) — the per-member sync states the
/// progress bar reads.
pub fn get_warehouse(endpoint: &str, token: &str, name: &str) -> Result<RegisteredWorkspace> {
let name = name.trim().to_string();
if name.is_empty() {
anyhow::bail!("no warehouse named");
}
let ws = name.clone();
with_workspaces_client(
endpoint,
token,
&ws,
Some(ROSTER_READ_TIMEOUT),
move |client| {
let name = name.clone();
Box::pin(async move {
let rec = client
.get(pb::WorkspaceName { name, purge: false })
.await
.context("Workspaces.Get RPC")?
.into_inner();
Ok(rec)
})
},
)
.map(RegisteredWorkspace::from_pb)
}
/// The member sync states behind the bar — the raw `(name, sync_state)` pairs.
fn member_sync_states(endpoint: &str, token: &str, name: &str) -> Result<Vec<(String, String)>> {
let name = name.trim().to_string();
let ws = name.clone();
with_workspaces_client(
endpoint,
token,
&ws,
Some(ROSTER_READ_TIMEOUT),
move |client| {
let name = name.clone();
Box::pin(async move {
let rec = client
.get(pb::WorkspaceName { name, purge: false })
.await
.context("Workspaces.Get RPC")?
.into_inner();
Ok(rec
.members
.into_iter()
.map(|m| (m.name, m.sync_state))
.collect::<Vec<_>>())
})
},
)
}
/// The `Workspaces.PopulateStatus` roll-up row for ONE workspace, if the server
/// answers. Best-effort: an older server that does not implement it leaves the
/// verdict empty rather than failing the poll.
fn populate_roster_row(endpoint: &str, token: &str, name: &str) -> Option<serde_json::Value> {
let json = with_workspaces_client(
endpoint,
token,
"",
Some(ROSTER_READ_TIMEOUT),
move |client| {
Box::pin(async move {
let r = client
.populate_status(pb::Empty {})
.await
.context("Workspaces.PopulateStatus RPC")?
.into_inner();
Ok(r.json)
})
},
)
.ok()?;
let rows: Vec<serde_json::Value> = serde_json::from_str(&json).ok()?;
rows.into_iter()
.find(|r| r.get("name").and_then(|n| n.as_str()) == Some(name))
}
/// **Why the code-graph phase needs a probe of its own, named precisely.**
///
/// `Workspaces.Get` reports per-member `sync_state`, which covers the CLONE phase and
/// nothing after it. `Workspaces.PopulateStatus` rolls up `clone_events`, which is also
/// the clone phase — it goes `green` as soon as the members landed, long before the
/// consumer has released a commit, so it cannot mark the end of anything later. The
/// later two phases are therefore observed by their products: `dep_graph_edges` is what
/// the warehouse BUILD writes, and the workspace record's `current_snapshot` is the
/// STAMP nornir's consumer sets when it releases a commit into the code room.
///
/// Until Mímir v2 the third product was a row in `call_edges`. That table has no writer
/// any more, so a bar probing it never finished on a new warehouse — and finished at
/// once on an old one.
pub const CODE_GRAPH_PROBE_WHY: &str =
"the clone phase is visible per-member (WorkspaceMember.sync_state) but nothing on \
the served surface reports the warehouse-build or code-graph phases directly: \
PopulateStatus rolls up clone_events, which turns green when the CLONES land. The \
later two phases are therefore observed by their products — dep_graph_edges for the \
build, the workspace's current_snapshot (the consumer's stamp) for the code graph.";
/// **Has nornir's consumer released anything of this workspace yet?** — the presence
/// probe behind the code-graph phase: the workspace record carries a non-empty
/// `current_snapshot` once it has.
///
/// `Pending` while the record cannot be read or carries no stamp, never
/// 0-as-if-measured: "nothing was consumed yet" and "this seam could not look" both
/// mean the phase is not known to be over.
pub fn code_graph_consumed(endpoint: &str, token: &str, workspace: &str) -> Counter {
match get_warehouse(endpoint, token, workspace) {
Ok(ws) if ws.is_populated() => Counter::Known { value: 1 },
_ => Counter::Pending,
}
}
/// **Count the distinct crates the warehouse knows about** — the refining estimate.
///
/// Read off `dep_graph_edges.via_crate`, which every warehouse build writes, so it
/// grows as the populate finishes rather than appearing all at once. `Pending` while
/// the table is absent or empty (a warehouse mid-build), never 0-as-if-measured.
pub fn count_warehouse_crates(endpoint: &str, token: &str, workspace: &str, limit: u32) -> Counter {
let Ok(scan) = scan_warehouse_table(endpoint, token, workspace, "dep_graph_edges", limit)
else {
return Counter::Pending;
};
let Some(col) = scan.columns.iter().position(|c| c == "via_crate") else {
return Counter::Pending;
};
let distinct: std::collections::BTreeSet<&str> = scan
.rows
.iter()
.filter_map(|r| r.get(col))
.map(|s| s.as_str())
.collect();
if distinct.is_empty() {
Counter::Pending
} else {
Counter::Known {
value: distinct.len() as u64,
}
}
}
/// **Poll the populate progress of ONE warehouse.**
///
/// One `Workspaces.Get` (always — it is the live signal and it is cheap) plus a
/// `PopulateStatus` roll-up. `crate_scan_limit > 0` additionally counts crates; pass
/// 0 on the fast ticks and a real limit on the slow ones, so the refining counter
/// does not cost a table scan every frame.
pub fn poll_populate_progress(
endpoint: &str,
token: &str,
workspace: &str,
crate_scan_limit: u32,
) -> Result<PopulateProgress> {
let members = member_sync_states(endpoint, token, workspace)?;
let repos_total = members.len();
let repos_done = members.iter().filter(|(_, s)| s == "ok").count();
let repos_fetching = members.iter().filter(|(_, s)| s == "fetching").count();
let repos_failed = members
.iter()
.filter(|(_, s)| s.starts_with("error"))
.count();
let row = populate_roster_row(endpoint, token, workspace);
let get_str = |k: &str| -> String {
row.as_ref()
.and_then(|r| r.get(k))
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string()
};
let crates = if crate_scan_limit > 0 {
count_warehouse_crates(endpoint, token, workspace, crate_scan_limit)
} else {
Counter::Pending
};
// The code-graph probe rides the same refine tick: one `Workspaces.Get`, asking
// only "has the consumer stamped this workspace?".
let code_graph = if crate_scan_limit > 0 {
code_graph_consumed(endpoint, token, workspace)
} else {
Counter::Pending
};
Ok(PopulateProgress {
workspace: workspace.to_string(),
running: repos_fetching > 0,
repos_total,
repos_done,
repos_fetching,
repos_failed,
crates,
code_graph,
// NEVER a number. There is no byte counter to read.
bytes: Counter::Unavailable {
why: BYTES_UNAVAILABLE_WHY,
},
verdict: get_str("populate"),
last_synced: get_str("last_synced"),
failing_member: get_str("failing_member"),
last_error: get_str("last_error"),
})
}
/// List the warehouses nornir already serves (`Workspaces.List`).
pub fn list_warehouses(endpoint: &str, token: &str) -> Result<Vec<RegisteredWorkspace>> {
with_workspaces_client(
endpoint,
token,
"",
Some(ROSTER_READ_TIMEOUT),
move |client| {
Box::pin(async move {
let list = client
.list(pb::Empty {})
.await
.context("Workspaces.List RPC")?
.into_inner();
Ok(list
.workspaces
.into_iter()
.map(RegisteredWorkspace::from_pb)
.collect())
})
},
)
}
/// **The headless "Add warehouse…" form model** — the drivable core the dwarves-gui
/// warehouse menu hosts and the robot test drives over `state_json`/`update_json`
/// (no egui, no live server needed to test the validation + status transitions).
///
/// The UI: type a `name` + an `ssh_url`, press **Add** (validates SSH → Register),
/// then **Populate** (Fetch). All server I/O is injected via the `*_result` setters
/// in tests; in the app the buttons call [`add_warehouse_by_ssh`] / [`populate_warehouse`].
#[derive(Debug, Clone, Default)]
pub struct AddWarehouse {
pub endpoint: String,
pub token: String,
pub name: String,
pub ssh_url: String,
/// Last status line ("", "ok: …", "error: …").
pub status: String,
/// The last successfully registered/populated record (if any).
pub registered: Option<RegisteredWorkspace>,
/// The last populate outcome (if any).
pub last_fetch: Option<FetchOutcome>,
/// The verdict of the last WHOLE `add` action — register + populate judged
/// together. `None` until one has been run. A `Some` whose `populated` is false is
/// a FAILED add, no matter that the `Register` RPC returned Ok.
pub last_add: Option<AddOutcome>,
/// The name the last successful `remove` cleared (for the status line).
pub last_removed: Option<String>,
/// What the last **Refresh** (`Workspaces.Repopulate`) did — including a named
/// refusal, which is a fact about the run and not an absence of one. `None`
/// until one has been driven.
pub last_repopulate: Option<RepopulateOutcome>,
/// True once the user has typed a non-empty `name` BY HAND — from then on the
/// ssh-url auto-derivation ([`workspace_name_from_ssh`]) never overwrites it.
/// Clearing the name field hands the suggestion back to the derivation.
pub name_edited: bool,
}
impl AddWarehouse {
pub fn new(endpoint: impl Into<String>, token: impl Into<String>) -> Self {
Self {
endpoint: endpoint.into(),
token: token.into(),
..Default::default()
}
}
/// The URL is a valid SSH remote AND the name is non-empty ⇒ Add is enabled.
pub fn can_add(&self) -> bool {
!self.name.trim().is_empty() && validate_ssh_remote(&self.ssh_url).is_ok()
}
/// Type into the `name` field. A non-empty hand-typed name pins itself against
/// the ssh auto-derivation; clearing it un-pins.
pub fn set_name(&mut self, name: &str) {
self.name = name.to_string();
self.name_edited = !self.name.trim().is_empty();
}
/// Type into the `ssh` field. Unless the user has pinned a name of their own,
/// the short workspace name is derived from the url and pre-filled — typing
/// `…/workspace_nordisk.git` fills `name` with `nordisk`.
pub fn set_ssh_url(&mut self, url: &str) {
self.ssh_url = url.to_string();
if !self.name_edited && validate_ssh_remote(url).is_ok() {
self.name = workspace_name_from_ssh(url);
}
}
/// Drive **Add** — the WHOLE action: validate, register, populate, and judge the
/// three together. Best-effort UI: a failure sets the status, never panics.
///
/// Returns `Ok(())` **iff a warehouse snapshot exists at the end**. A register that
/// succeeded over a warehouse with no snapshot — the failed-member-clone case that
/// printed `✓ populated now` over an empty `git/` — returns `Err` and leaves an
/// `error:` status, because a failed member clone must fail the action.
pub fn add(&mut self) -> Result<()> {
if let Err(e) = validate_ssh_remote(&self.ssh_url) {
self.status = format!("error: {e}");
return Err(e);
}
match add_warehouse_by_ssh(&self.endpoint, &self.token, &self.name, &self.ssh_url) {
Ok(out) => {
self.status = out.describe();
self.registered = Some(out.registered.clone());
if let Some(f) = &out.fetch {
self.last_fetch = Some(f.clone());
}
let populated = out.populated;
let verdict = out.describe();
self.last_add = Some(out);
if populated {
Ok(())
} else {
// NOT green. The action did not do what its name says.
Err(anyhow::anyhow!(verdict))
}
}
Err(e) => {
self.status = format!("error: {e:#}");
Err(e)
}
}
}
/// Drive **Remove** — the WHOLE action: the registry row and the on-disk warehouse
/// go together (`Workspaces.Remove { purge: true }`).
///
/// It clears warehouse/data and **never touches ssh keys or tokens** — the RPC has
/// no field that names one, and dwarves' key + vault live client-side where this
/// call does not reach.
pub fn remove_named(&mut self, name: &str) -> Result<()> {
match remove_warehouse(&self.endpoint, &self.token, name) {
Ok(()) => {
self.status = format!(
"ok: removed `{name}` — registry entry and its on-disk warehouse \
(warehouse + catalog.redb + jobs.redb) cleared. ssh keys and tokens untouched."
);
self.last_removed = Some(name.trim().to_string());
// The record we were holding describes a warehouse that no longer
// exists — drop it rather than keep drawing it.
if self.registered.as_ref().map(|r| r.name.as_str()) == Some(name.trim()) {
self.registered = None;
self.last_fetch = None;
self.last_add = None;
}
Ok(())
}
Err(e) => {
self.status = format!("error: {e:#}");
Err(e)
}
}
}
/// Drive **Refresh** — THE ONE REFRESH, on a named warehouse
/// (`Workspaces.Repopulate { purge }`).
///
/// This is the verb that replaced the per-pane Scan buttons. Those buttons asked
/// N times for work nornir already schedules by itself (a landed snapshot submits
/// one chain job per lens, deduped and superseded); what no button could ask for
/// was *clear the derived tables and build them again*, which is exactly this.
///
/// `purge = false` re-enqueues only. `purge = true` first drops the derived lens
/// tables — nornir reads that set off its own lens registry, so this side never
/// carries a table list that could go stale — and never touches a raw scan or a
/// snapshot.
///
/// **A refusal is not an error and is not a success.** A chain already in flight
/// comes back `Ok` with nornir's named sentence (and the in-flight job ids) in
/// `status`, and this returns `Err` so the caller draws it as "not done" — but the
/// sentence is nornir's own, verbatim, naming the job the operator can watch.
pub fn repopulate_named(&mut self, name: &str, purge: bool) -> Result<RepopulateOutcome> {
match repopulate_warehouse(&self.endpoint, &self.token, name, purge) {
Ok(out) => {
self.status = out.describe();
self.last_repopulate = Some(out.clone());
if out.is_refusal() {
return Err(anyhow::anyhow!("{}", out.refusal));
}
Ok(out)
}
Err(e) => {
// The gate half: nornir's FAILED_PRECONDITION message carries the
// in-flight job id. Show it, do not re-word it.
self.status = format!("error: {e:#}");
Err(e)
}
}
}
/// Drive **Populate** — re-sync the warehouse this form last registered
/// (`Fetch{force}`). A thin special case of [`populate_named`](Self::populate_named):
/// populate is an operation on an EXISTING warehouse, and the one just added is
/// only one of the warehouses a host may offer it on.
pub fn populate(&mut self, force: bool) -> Result<()> {
let Some(name) = self.registered.as_ref().map(|r| r.name.clone()) else {
let e = anyhow::anyhow!("no warehouse registered yet — Add one first");
self.status = format!("error: {e}");
return Err(e);
};
self.populate_named(&name, force)
}
/// Drive **Populate on a NAMED warehouse** — `Workspaces.Fetch { force }` against
/// any warehouse nornir already serves, not only the one this form registered.
///
/// This is the verb a warehouse PICKER binds to: the user selects a warehouse and
/// re-syncs THAT one. It needs nothing from the add-form except the endpoint +
/// token, so a host can offer Populate long after (and independently of) an Add.
pub fn populate_named(&mut self, name: &str, force: bool) -> Result<()> {
let name = name.trim().to_string();
if name.is_empty() {
let e = anyhow::anyhow!("no warehouse selected — pick one to populate");
self.status = format!("error: {e}");
return Err(e);
}
match populate_warehouse(&self.endpoint, &self.token, &name, force) {
Ok(out) => {
self.status = format!(
"ok: populated `{}` — {} fetched, {} changed{}",
out.workspace,
out.fetched,
out.changed.len(),
if out.snapshot.is_empty() {
String::new()
} else {
format!(", snapshot {}", out.snapshot)
},
);
self.last_fetch = Some(out);
Ok(())
}
Err(e) => {
self.status = format!("error: {e:#}");
Err(e)
}
}
}
/// The robot oracle.
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"name": self.name,
"ssh_url": self.ssh_url,
"ssh_valid": validate_ssh_remote(&self.ssh_url).is_ok(),
"can_add": self.can_add(),
"status": self.status,
"registered": self.registered,
"registered_name": self.registered.as_ref().map(|r| r.name.clone()),
// `populated` is the WHOLE-ACTION verdict when an add has been run (a
// register whose member clone failed is NOT populated), falling back to
// the record's own snapshot for a warehouse that arrived another way.
"populated": self
.last_add
.as_ref()
.map(|a| a.populated)
.or_else(|| self.registered.as_ref().map(|r| r.is_populated()))
.unwrap_or(false),
"last_fetch": self.last_fetch,
"last_add": self.last_add,
"last_removed": self.last_removed,
// True when an add failed and NOTHING available can say why. The pane must
// name the missing diagnostic rather than draw the silence as success.
"cause_unknown": self.last_add.as_ref().map(|a| a.cause_unknown).unwrap_or(false),
"missing_diagnostic_rpc": MISSING_DIAGNOSTIC_RPC,
})
}
/// The control seam — drive the form as the UI would (type name/url).
pub fn update_json(&mut self, msg: &str) {
let Ok(v) = serde_json::from_str::<serde_json::Value>(msg) else {
return;
};
if let Some(n) = v.get("name").and_then(|x| x.as_str()) {
self.set_name(n);
}
if let Some(u) = v.get("ssh_url").and_then(|x| x.as_str()) {
self.set_ssh_url(u);
}
}
}
// ═════════════════════════════════════════════════════════════════════════════
// Warehouse READ client — the CATALOG-DOCTRINE read leg (ratified 2026-07-15).
//
// nornir-server is the ONE live catalog (sole redb write-lock holder + read-RPC);
// every OTHER app (dwarves, korp, modgunn's view of `sbom_components`) READS
// nornir's Iceberg tables through its `Warehouse { Tables, Scan }` gRPC — never by
// opening `catalog.redb` (a 2nd exclusive lock → latent WOULDBLOCK/deadlock). This
// is the reusable THIN client for that read: SAME `<guest>-embed` / no-`nornir`-dep
// shape as the write legs above. modgunn consumes it for its SBOM view.
// ═════════════════════════════════════════════════════════════════════════════
/// One table scan — the `Warehouse.Scan` reply: stringified `columns` + `rows`
/// (each row is one cell per column). The server's `scan_preview` shape (all cells
/// rendered to display strings), so this is exact for string columns (e.g.
/// `sbom_components`) and a display view for others.
#[derive(Debug, Clone, Default)]
pub struct WarehouseScan {
pub columns: Vec<String>,
pub rows: Vec<Vec<String>>,
}
/// **Honest typed failure** of a warehouse READ over gRPC — never a silent empty.
/// The catalog doctrine forbids falling back to opening nornir's `catalog.redb`, so
/// a reader must be able to tell WHY a read failed and surface it (a consumer maps
/// these to its own error / retry policy). `std::error::Error` so it round-trips
/// through `anyhow::Error::downcast_ref`.
#[derive(Debug)]
pub enum WarehouseReadError {
/// Could not reach nornir-server (connect refused / timed out / bad URL).
Unreachable(String),
/// Server reached but rejected the credentials (`Unauthenticated`/`PermissionDenied`).
AuthFailed(String),
/// Server reached + authed, but the requested workspace/table isn't served
/// (`NotFound`/`Unavailable`/`FailedPrecondition`).
NotServed(String),
/// Any other RPC-level failure.
Rpc(String),
/// **The transport cannot carry this shape.** The server is reachable, authed and
/// serving the workspace, but does not implement the RPC (`UNIMPLEMENTED`).
///
/// This exists as its own variant because of one specific bug class: a thin client
/// that cannot fetch `embeddings` and a warehouse that holds no embeddings look
/// identical downstream (both end in "no vectors"), and the dupes SEMANTIC tier
/// rendered them the same for months — reading as "nothing similar found" when the
/// truth was "never looked". A caller must be able to tell the two apart.
Unsupported(String),
}
impl std::fmt::Display for WarehouseReadError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Unreachable(m) => write!(f, "nornir-server unreachable: {m}"),
Self::AuthFailed(m) => write!(f, "nornir-server auth_failed: {m}"),
Self::NotServed(m) => write!(f, "nornir warehouse not_served: {m}"),
Self::Rpc(m) => write!(f, "nornir Warehouse RPC failed: {m}"),
Self::Unsupported(m) => write!(f, "nornir warehouse transport cannot carry it: {m}"),
}
}
}
impl std::error::Error for WarehouseReadError {}
/// Map an RPC `Status` onto the honest read-error category.
fn status_to_read_error(s: tonic::Status) -> WarehouseReadError {
use tonic::Code;
let msg = format!("{} ({:?})", s.message(), s.code());
match s.code() {
Code::Unauthenticated | Code::PermissionDenied => WarehouseReadError::AuthFailed(msg),
Code::NotFound | Code::Unavailable | Code::FailedPrecondition => {
WarehouseReadError::NotServed(msg)
}
// A server too old to serve the RPC at all — the "this transport cannot carry
// it" state, which must NEVER collapse into an empty result.
Code::Unimplemented => WarehouseReadError::Unsupported(msg),
_ => WarehouseReadError::Rpc(msg),
}
}
/// The intercepted `Warehouse` client type (bearer + `nornir-workspace` header).
type WhClient = pb::warehouse_client::WarehouseClient<
tonic::service::interceptor::InterceptedService<tonic::transport::Channel, NornirInterceptor>,
>;
/// Run a blocking `Warehouse` gRPC call on a DEDICATED OS thread (never a nested-
/// runtime panic — modgunn drives this from inside its own multi-thread 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`].
fn with_warehouse_client<T, F>(
endpoint: &str,
token: &str,
workspace: &str,
f: F,
) -> Result<T, WarehouseReadError>
where
T: Send + 'static,
F: for<'a> FnOnce(
&'a mut WhClient,
) -> 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::SCAN_TIMEOUT)
.connect()
.await
.map_err(|e| {
WarehouseReadError::Unreachable(format!("connect to {endpoint}: {e}"))
})?;
let interceptor: 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::warehouse_client::WarehouseClient::with_interceptor(channel, interceptor)
.max_decoding_message_size(64 << 20);
f(&mut client).await
})
})
.join()
.map_err(|_| WarehouseReadError::Rpc("warehouse-read thread panicked".into()))?
})
}
/// One embedding, TYPED — the shape `Warehouse.Scan` cannot carry.
#[derive(Debug, Clone, PartialEq)]
pub struct VectorRow {
/// `embeddings.content_hash`.
pub key: String,
/// The vector itself. Exact `f32`, decoded from the wire's little-endian bytes —
/// never re-parsed from a rendered string.
pub vector: Vec<f32>,
/// The embedding SPACE (`embeddings.model_profile`). A cosine measured in one space
/// says nothing about another; carried so a consumer can see a mixed corpus rather
/// than average two vector spaces into confident nonsense.
pub model_profile: String,
}
/// The result of a [`scan_warehouse_vectors`] read.
///
/// An empty `vectors` here means exactly one thing: **the server answered and the
/// warehouse holds no embedding rows.** "The transport cannot carry vectors" is
/// [`WarehouseReadError::Unsupported`], never this.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct VectorScan {
pub vectors: Vec<VectorRow>,
/// Source rows the server read. `rows_scanned > vectors.len()` ⇒ a partially
/// embedded table, visible instead of silently looking like a smaller one.
pub rows_scanned: u64,
/// A non-zero `limit` bit: the corpus is INCOMPLETE and an all-pairs verdict over
/// it is not comparable with a full read.
pub clipped: bool,
}
/// **Read a vector column TYPED** (the `Warehouse.ScanVectors` server-streaming RPC).
///
/// This is the leg that un-blinds a thin client's semantic tier. `Warehouse.Scan`
/// stringifies every cell through one renderer whose catch-all prints the Arrow type
/// name, so `embeddings.vector` came back as the SAME constant placeholder on every
/// row — zero floats, indistinguishable from a table that was never populated. Here the
/// floats cross the wire as raw little-endian `f32` and arrive bit-identical to what a
/// fat (on-disk Parquet) reader sees, which is what lets a thin client reach the same
/// verdict as a fat one.
///
/// `limit` 0 = every row; the semantic tier is an all-pairs sweep, and a silently short
/// corpus is a different answer, not a cheaper one. A `limit` that bites sets
/// [`VectorScan::clipped`].
///
/// Empty table (`table: ""`) defaults to `embeddings`, `key_col` to `content_hash`,
/// `vec_col` to `vector` — the server applies those defaults.
pub fn scan_warehouse_vectors(
endpoint: &str,
token: &str,
workspace: &str,
table: &str,
key_col: &str,
vec_col: &str,
limit: u32,
) -> Result<VectorScan, WarehouseReadError> {
let req = pb::WarehouseVectorRequest {
table: table.to_string(),
key_col: key_col.to_string(),
vec_col: vec_col.to_string(),
limit,
};
with_warehouse_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let mut stream = client
.scan_vectors(req)
.await
.map_err(status_to_read_error)?
.into_inner();
let mut out = VectorScan::default();
while let Some(batch) = stream.message().await.map_err(status_to_read_error)? {
out.rows_scanned = out.rows_scanned.max(batch.rows_scanned);
out.clipped |= batch.clipped;
for v in batch.vectors {
out.vectors.push(VectorRow {
key: v.key,
vector: decode_le_f32(&v.values, v.dims)?,
model_profile: v.model_profile,
});
}
}
Ok(out)
})
})
}
/// Decode `dims` little-endian IEEE-754 `f32`s out of a wire payload.
///
/// A length that does not divide evenly is an ERROR, never a truncated vector: a short
/// vector still produces a plausible cosine, which is the confidently-wrong answer the
/// typed route exists to prevent.
pub fn decode_le_f32(bytes: &[u8], dims: u32) -> Result<Vec<f32>, WarehouseReadError> {
let dims = dims as usize;
if bytes.len() != dims * 4 {
return Err(WarehouseReadError::Rpc(format!(
"ScanVectors: declared dims={dims} needs {} bytes, got {} — refusing to \
truncate (a short vector still yields a plausible-looking cosine)",
dims * 4,
bytes.len()
)));
}
Ok(bytes
.as_chunks::<4>()
.0
.iter()
.map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
.collect())
}
/// Encode `f32`s as the wire's little-endian payload — the ONE encoder, shared by the
/// mock server and any producer, so encode and [`decode_le_f32`] cannot drift.
pub fn encode_le_f32(v: &[f32]) -> Vec<u8> {
let mut out = Vec::with_capacity(v.len() * 4);
for f in v {
out.extend_from_slice(&f.to_le_bytes());
}
out
}
/// **List the tables nornir-server owns** (the `Warehouse.Tables` RPC). An honest
/// typed error on unreachable/auth/not-served — never a silent empty.
pub fn fetch_warehouse_tables(
endpoint: &str,
token: &str,
workspace: &str,
) -> Result<Vec<String>, WarehouseReadError> {
with_warehouse_client(endpoint, token, workspace, |client| {
Box::pin(async move {
let resp = client
.tables(pb::Empty {})
.await
.map_err(status_to_read_error)?
.into_inner();
Ok(resp.names)
})
})
}
/// **Scan one nornir table** (the `Warehouse.Scan` RPC). `limit` 0 = server default
/// (~500 rows); pass a large value for a fleet read (the server honors a non-zero
/// limit). An honest typed error on failure.
pub fn scan_warehouse_table(
endpoint: &str,
token: &str,
workspace: &str,
table: &str,
limit: u32,
) -> Result<WarehouseScan, WarehouseReadError> {
let table = table.to_string();
with_warehouse_client(endpoint, token, workspace, move |client| {
Box::pin(async move {
let resp = client
.scan(pb::WarehouseScanRequest { table, limit })
.await
.map_err(status_to_read_error)?
.into_inner();
Ok(WarehouseScan {
columns: resp.columns,
rows: resp.rows.into_iter().map(|r| r.cells).collect(),
})
})
})
}
/// Reusable MOCK nornir gRPC servers for tests — ONE shared mock per service so a
/// consumer's test (this crate's, modgunn's gRPC-read test, dwarves' warehouse-picker
/// UI test) doesn't twin its own tonic server (L5). Available to this crate's own
/// `cfg(test)` and, for external consumers, behind the `testkit` feature.
#[cfg(any(test, feature = "testkit"))]
pub mod testkit {
use super::pb;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
/// A `Warehouse` server serving canned tables: name → (columns, rows).
///
/// `vectors` is served over `ScanVectors` (typed). `serve_vectors = false` makes the
/// server answer `UNIMPLEMENTED` — the "this transport cannot carry vectors" state,
/// which a test needs to be able to produce on purpose, because that state used to be
/// indistinguishable from an unpopulated warehouse.
#[derive(Clone, Default)]
pub struct MockWarehouse {
tables: Arc<HashMap<String, (Vec<String>, Vec<Vec<String>>)>>,
vectors: Arc<Vec<(String, Vec<f32>)>>,
serve_vectors: bool,
/// Hold every reply this long before answering. `Default` = answer at once.
/// The SAME knob [`MockMimir::reply_delay`] has, for the SAME reason and with
/// the same `pause()` shape — see [`spawn_mock_warehouse_slow`].
reply_delay: std::time::Duration,
}
impl MockWarehouse {
/// Hold the reply, if this server was asked to. One body, called by every RPC —
/// a second `if !delay.is_zero()` in a handler is how a mock ends up slow on
/// some calls and not others.
async fn pause(&self) {
if !self.reply_delay.is_zero() {
tokio::time::sleep(self.reply_delay).await;
}
}
}
#[tonic::async_trait]
impl pb::warehouse_server::Warehouse for MockWarehouse {
async fn tables(
&self,
_req: tonic::Request<pb::Empty>,
) -> Result<tonic::Response<pb::WarehouseTables>, tonic::Status> {
self.pause().await;
Ok(tonic::Response::new(pb::WarehouseTables {
names: self.tables.keys().cloned().collect(),
}))
}
async fn scan(
&self,
req: tonic::Request<pb::WarehouseScanRequest>,
) -> Result<tonic::Response<pb::WarehouseScan>, tonic::Status> {
self.pause().await;
let r = req.into_inner();
// An unknown table ⇒ NotFound — the mock's honest "not served" signal.
let (cols, rows) = self
.tables
.get(&r.table)
.ok_or_else(|| tonic::Status::not_found(format!("no table `{}`", r.table)))?;
let rows = rows
.iter()
.map(|cells| pb::WarehouseRow {
cells: cells.clone(),
})
.collect();
Ok(tonic::Response::new(pb::WarehouseScan {
columns: cols.clone(),
rows,
}))
}
type ScanVectorsStream = std::pin::Pin<
Box<
dyn tokio_stream::Stream<Item = Result<pb::WarehouseVectorBatch, tonic::Status>>
+ Send
+ 'static,
>,
>;
async fn scan_vectors(
&self,
req: tonic::Request<pb::WarehouseVectorRequest>,
) -> Result<tonic::Response<Self::ScanVectorsStream>, tonic::Status> {
self.pause().await;
if !self.serve_vectors {
// The pre-ScanVectors server. Distinct from an empty stream ON PURPOSE.
return Err(tonic::Status::unimplemented(
"Warehouse.ScanVectors not served by this build",
));
}
let limit = req.into_inner().limit as usize;
let clipped = limit > 0 && limit < self.vectors.len();
let rows_scanned = self.vectors.len() as u64;
let take = if limit == 0 {
self.vectors.len()
} else {
limit.min(self.vectors.len())
};
// Batched, so the streaming decode path is exercised rather than assumed.
let batches: Vec<Result<pb::WarehouseVectorBatch, tonic::Status>> = self.vectors
[..take]
.chunks(2)
.map(|c| {
Ok(pb::WarehouseVectorBatch {
vectors: c
.iter()
.map(|(k, v)| pb::WarehouseVector {
key: k.clone(),
dims: v.len() as u32,
values: super::encode_le_f32(v),
model_profile: "mock-profile".into(),
})
.collect(),
rows_scanned,
clipped,
})
})
.collect();
// Zero vectors still gets ONE frame: "answered, holds nothing" is a positive
// statement on the wire, not an absence a client has to interpret.
let batches = if batches.is_empty() {
vec![Ok(pb::WarehouseVectorBatch {
vectors: Vec::new(),
rows_scanned,
clipped,
})]
} else {
batches
};
Ok(tonic::Response::new(
Box::pin(tokio_stream::iter(batches)) as Self::ScanVectorsStream
))
}
}
/// A running mock server on an ephemeral port. Dropping it shuts the server
/// down and joins its thread.
pub struct MockWarehouseServer {
pub endpoint: String,
shutdown: Option<tokio::sync::oneshot::Sender<()>>,
join: Option<std::thread::JoinHandle<()>>,
}
impl Drop for MockWarehouseServer {
fn drop(&mut self) {
if let Some(tx) = self.shutdown.take() {
let _ = tx.send(());
}
if let Some(j) = self.join.take() {
let _ = j.join();
}
}
}
/// Bind an ephemeral port, serve `router` on its OWN OS thread + current-thread
/// runtime (so a caller with its own tokio runtime can drive it without a
/// nested-runtime panic), and wait until the port answers. The ONE spawn body every
/// mock in this module reuses — a second service must not twin it (L5). PUBLIC so a
/// sibling embed crate's testkit (nornir-funnel-embed's mock `Funnel`) mounts its
/// service through this same body instead of twinning the thread/runtime/wait dance.
pub fn spawn_router(
build: impl FnOnce(tonic::transport::Server) -> tonic::transport::server::Router
+ Send
+ 'static,
) -> MockWarehouseServer {
let addr = {
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
l.local_addr().unwrap()
};
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
let join = std::thread::spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(async move {
build(tonic::transport::Server::builder())
.serve_with_shutdown(addr, async {
let _ = rx.await;
})
.await
.unwrap();
});
});
for _ in 0..50 {
if std::net::TcpStream::connect(addr).is_ok() {
break;
}
std::thread::sleep(std::time::Duration::from_millis(20));
}
MockWarehouseServer {
endpoint: format!("http://{addr}"),
shutdown: Some(tx),
join: Some(join),
}
}
/// Spawn a mock `Warehouse` server serving `tables` (each `(name, columns, rows)`).
///
/// `ScanVectors` answers `UNIMPLEMENTED` — i.e. this is the **pre-typed-vector**
/// server, the shape every existing caller was talking to. Use
/// [`spawn_mock_warehouse_with_vectors`] for one that can carry vectors.
pub fn spawn_mock_warehouse(
tables: Vec<(String, Vec<String>, Vec<Vec<String>>)>,
) -> MockWarehouseServer {
spawn_mock(tables, Vec::new(), false, std::time::Duration::ZERO)
}
/// **A mock `Warehouse` that HOLDS every reply for `reply_delay` before answering.**
///
/// The knob [`MockMimir::reply_delay`] has already had, arriving here because the
/// panes that read `Warehouse.Scan` need the same proof the Lenses pane got: that a
/// frame comes back BEFORE the server does.
///
/// **Why this and not a slow injected source.** A test source that sleeps proves the
/// pane's own worker is off-thread. It does NOT prove the pane is — the read could
/// still be inline through a different seam, or through the real client, and a fake
/// source would never notice. A server that holds its reply puts the delay on the far
/// side of the REAL transport (`tonic` channel, real socket, the shipped
/// [`scan_warehouse_table`]), so what the frame is waiting on is the thing it waits
/// on in production. Prefer it over an injected slow source, and over bulking a
/// fixture to hundreds of thousands of rows to make the read take measurable time —
/// that works (`binview.rs` did it before this existed) but it needs a calibration
/// pass and a non-vacuity guard to not be machine-dependent, and it burns memory.
///
/// The delay applies to `Tables`, `Scan` and `ScanVectors` alike, through one
/// `pause()` body, so a mock cannot be slow on one call and instant on another.
pub fn spawn_mock_warehouse_slow(
tables: Vec<(String, Vec<String>, Vec<Vec<String>>)>,
reply_delay: std::time::Duration,
) -> MockWarehouseServer {
spawn_mock(tables, Vec::new(), false, reply_delay)
}
/// Spawn a mock `Warehouse` server that ALSO serves `vectors` typed over
/// `ScanVectors` — the seam under test when a thin client's semantic tier has to
/// agree with a fat one's.
///
/// An empty `vectors` list here is the OTHER honest state: the transport works, the
/// warehouse simply holds no embeddings.
pub fn spawn_mock_warehouse_with_vectors(
tables: Vec<(String, Vec<String>, Vec<Vec<String>>)>,
vectors: Vec<(String, Vec<f32>)>,
) -> MockWarehouseServer {
spawn_mock(tables, vectors, true, std::time::Duration::ZERO)
}
fn spawn_mock(
tables: Vec<(String, Vec<String>, Vec<Vec<String>>)>,
vectors: Vec<(String, Vec<f32>)>,
serve_vectors: bool,
reply_delay: std::time::Duration,
) -> MockWarehouseServer {
let map: HashMap<String, (Vec<String>, Vec<Vec<String>>)> =
tables.into_iter().map(|(n, c, r)| (n, (c, r))).collect();
let mock = MockWarehouse {
tables: Arc::new(map),
vectors: Arc::new(vectors),
serve_vectors,
reply_delay,
};
spawn_router(move |mut s| s.add_service(pb::warehouse_server::WarehouseServer::new(mock)))
}
// ── the `Telemetry` WRITE surface (test_results + the two BUILD tables) ──────
/// What a [`MockTelemetry`] server received, per table. The build-record
/// producer's red-prove test reads `build_provenance()` / `build_info()` back out
/// of this: a producer that files nothing leaves them EMPTY, which is exactly the
/// state that hid for four rounds.
#[derive(Clone, Default)]
pub struct TelemetryLog {
test_results: Arc<Mutex<Vec<pb::TestResultPb>>>,
build_provenance: Arc<Mutex<Vec<String>>>,
build_info: Arc<Mutex<Vec<pb::BuildInfoRowPb>>>,
step_feeds: Arc<Mutex<Vec<pb::SubmitStepFeedRequest>>>,
/// When set, EVERY build submit is rejected with this status message — the
/// knob that red-proves the guard against a reachable-but-refusing server.
reject_builds: Arc<Mutex<Option<String>>>,
}
impl TelemetryLog {
/// Every `test_results` row the mock served, in arrival order.
pub fn test_results(&self) -> Vec<pb::TestResultPb> {
self.test_results.lock().unwrap().clone()
}
/// Every step feed the mock served, in arrival order.
pub fn step_feeds(&self) -> Vec<pb::SubmitStepFeedRequest> {
self.step_feeds.lock().unwrap().clone()
}
/// Every `BuildProvenance` JSON the mock served, in arrival order.
pub fn build_provenance(&self) -> Vec<String> {
self.build_provenance.lock().unwrap().clone()
}
/// Every `build_info` row the mock served, in arrival order.
pub fn build_info(&self) -> Vec<pb::BuildInfoRowPb> {
self.build_info.lock().unwrap().clone()
}
/// Make the server REFUSE both build submits with `msg` — a reachable server
/// that records nothing, so the guard's red path is exercised for real
/// instead of asserted in theory.
pub fn reject_builds(&self, msg: &str) {
*self.reject_builds.lock().unwrap() = Some(msg.to_string());
}
fn rejection(&self) -> Option<tonic::Status> {
self.reject_builds
.lock()
.unwrap()
.as_ref()
.map(|m| tonic::Status::internal(m.clone()))
}
}
/// A `Telemetry` server recording into a [`TelemetryLog`].
#[derive(Clone, Default)]
pub struct MockTelemetry {
pub log: TelemetryLog,
}
#[tonic::async_trait]
impl pb::telemetry_server::Telemetry for MockTelemetry {
async fn submit_test_results(
&self,
req: tonic::Request<pb::SubmitTestResultsRequest>,
) -> Result<tonic::Response<pb::SubmitCountResponse>, tonic::Status> {
let rows = req.into_inner().rows;
let accepted = rows.len() as u32;
self.log.test_results.lock().unwrap().extend(rows);
Ok(tonic::Response::new(pb::SubmitCountResponse { accepted }))
}
async fn submit_build_provenance(
&self,
req: tonic::Request<pb::SubmitBuildProvenanceRequest>,
) -> Result<tonic::Response<pb::SubmitCountResponse>, tonic::Status> {
if let Some(s) = self.log.rejection() {
return Err(s);
}
let records = req.into_inner().records_json;
let accepted = records.len() as u32;
self.log.build_provenance.lock().unwrap().extend(records);
Ok(tonic::Response::new(pb::SubmitCountResponse { accepted }))
}
async fn submit_build_info(
&self,
req: tonic::Request<pb::SubmitBuildInfoRequest>,
) -> Result<tonic::Response<pb::SubmitCountResponse>, tonic::Status> {
if let Some(s) = self.log.rejection() {
return Err(s);
}
let rows = req.into_inner().rows;
let accepted = rows.len() as u32;
self.log.build_info.lock().unwrap().extend(rows);
Ok(tonic::Response::new(pb::SubmitCountResponse { accepted }))
}
async fn submit_step_feed(
&self,
req: tonic::Request<pb::SubmitStepFeedRequest>,
) -> Result<tonic::Response<pb::SubmitCountResponse>, tonic::Status> {
let r = req.into_inner();
let accepted = r.tests.len().max(1) as u32;
self.log.step_feeds.lock().unwrap().push(r);
Ok(tonic::Response::new(pb::SubmitCountResponse { accepted }))
}
}
/// Spawn a mock `Telemetry` server on an ephemeral port; the returned
/// [`TelemetryLog`] is what it recorded.
pub fn spawn_mock_telemetry() -> (MockWarehouseServer, TelemetryLog) {
let mock = MockTelemetry::default();
let log = mock.log.clone();
let srv = spawn_router(move |mut s| {
s.add_service(pb::telemetry_server::TelemetryServer::new(mock))
});
(srv, log)
}
// ── the `Bench` surface (Submit + History) ──────────────────────────────────
/// What a [`MockBench`] server received. `dwarves bench run`'s red-prove reads
/// the submitted request back out of this: what matters is the ROW THAT CROSSED
/// THE WIRE, not that the client's own struct round-tripped through itself.
#[derive(Clone, Default)]
pub struct BenchLog {
submitted: Arc<Mutex<Vec<pb::SubmitBenchRequest>>>,
/// When set, every submit is refused with this status message — the knob
/// that red-proves a reachable-but-refusing warehouse.
reject: Arc<Mutex<Option<String>>>,
}
impl BenchLog {
/// Every `Bench.Submit` request the mock served, in arrival order.
pub fn submitted(&self) -> Vec<pb::SubmitBenchRequest> {
self.submitted.lock().unwrap().clone()
}
/// Make the server REFUSE every submit with `msg`.
pub fn reject(&self, msg: &str) {
*self.reject.lock().unwrap() = Some(msg.to_string());
}
}
/// A `Bench` server that records what it was asked to persist and answers a
/// deterministic run id.
#[derive(Default)]
pub struct MockBench {
pub log: BenchLog,
}
/// The run id [`MockBench`] answers with, so a caller can assert the receipt it
/// prints is the SERVER's id and not something it made up.
pub const MOCK_BENCH_RUN_ID: &str = "mock-run-0001";
#[tonic::async_trait]
impl pb::bench_server::Bench for MockBench {
async fn submit(
&self,
req: tonic::Request<pb::SubmitBenchRequest>,
) -> Result<tonic::Response<pb::SubmitBenchResponse>, tonic::Status> {
if let Some(m) = self.log.reject.lock().unwrap().as_ref() {
return Err(tonic::Status::internal(m.clone()));
}
self.log.submitted.lock().unwrap().push(req.into_inner());
Ok(tonic::Response::new(pb::SubmitBenchResponse {
run_id: MOCK_BENCH_RUN_ID.to_string(),
}))
}
/// `History` answers what `Submit` recorded, FILTERED by repo — so a
/// round-trip test proves the row that comes back is the row that crossed
/// the wire in, and a repo nothing measured answers an EMPTY list (the
/// state the client must map to an empty vec, never an error). The
/// `reject` knob refuses reads too: a refusing warehouse refuses both legs.
async fn history(
&self,
req: tonic::Request<pb::RepoOnly>,
) -> Result<tonic::Response<pb::BenchHistoryResponse>, tonic::Status> {
if let Some(m) = self.log.reject.lock().unwrap().as_ref() {
return Err(tonic::Status::internal(m.clone()));
}
let repo = req.into_inner().repo;
let runs = self
.log
.submitted
.lock()
.unwrap()
.iter()
.filter(|s| s.repo == repo)
.filter_map(|s| s.run.clone())
.collect();
Ok(tonic::Response::new(pb::BenchHistoryResponse {
repo,
runs,
}))
}
}
/// Spawn a mock `Bench` server on an ephemeral port; the returned [`BenchLog`]
/// is what it recorded.
pub fn spawn_mock_bench() -> (MockWarehouseServer, BenchLog) {
let mock = MockBench::default();
let log = mock.log.clone();
let srv =
spawn_router(move |mut s| s.add_service(pb::bench_server::BenchServer::new(mock)));
(srv, log)
}
// ── the `Workspaces` control surface (Register / Fetch / List) ───────────────
//
// The mock behind dwarves' "Add warehouse (SSH)" + "Populate" affordances. It
// deliberately does NOT populate on Register: `Register` records the workspace with
// an EMPTY `current_snapshot` and `Fetch` is what mints one, so a test can tell the
// two steps apart on the wire — which is the whole point of a picker that offers
// Populate as a SEPARATE, later act on an existing warehouse.
/// The in-memory registry a [`MockWorkspacesServer`] accumulates: every
/// `Register`ed name (in call order) and every `Fetch`ed name (in call order).
#[derive(Clone, Default)]
pub struct WorkspacesLog {
registered: Arc<Mutex<Vec<String>>>,
fetched: Arc<Mutex<Vec<String>>>,
removed: Arc<Mutex<Vec<(String, bool)>>>,
/// Every `Workspaces.Repopulate` the mock served — `(name, purge)`. A UI test
/// asserts against THIS to prove the Refresh button actually reached the wire
/// with `purge = true`; an absent or disabled button leaves it empty.
repopulated: Arc<Mutex<Vec<(String, bool)>>>,
/// Which of the verb's three answers the mock gives — see
/// [`RepopulateBehaviour`].
repopulate_behaviour: Arc<Mutex<RepopulateBehaviour>>,
/// The failure mode to simulate on `Fetch` — see [`FetchBehaviour`].
behaviour: FetchBehaviour,
/// The per-member `(name, sync_state)` `Get` reports — the live signal the
/// populate progress bar reads. A test drives this to stage a populate that
/// NOTHING in the client started.
member_states: Arc<Mutex<Vec<(String, String)>>>,
/// The JSON `PopulateStatus` returns (`[]` when unset).
populate_status_json: Arc<Mutex<String>>,
}
/// What the mock's `Fetch` does — so a test can reproduce the exact shapes the
/// whole-action `add` has to tell apart.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum FetchBehaviour {
/// Populate works: a snapshot is minted.
#[default]
Populates,
/// The MEMBER CLONE FAILED — a report with an error and NO snapshot. This is
/// the njord shape that printed `✓ populated now` over an empty `git/`.
MemberCloneFails,
/// Neither a snapshot nor an error: the populate silently did nothing, so the
/// reply alone cannot say why and the per-member outcome has to be asked for
/// separately (`Viz.CloneEvents`).
SilentlyDoesNothing,
}
/// What the mock's `Repopulate` answers — THE ONE REFRESH has three outcomes and a
/// UI must be able to be driven into each. Two of them are refusals, and they do
/// not mean the same thing: `CannotConsume` says *members are behind and nothing
/// here can consume them* (a report), `NotServed` says *this server has no code
/// room for that workspace* (a status).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum RepopulateBehaviour {
/// Members are behind and a consume was started: their lines come back in
/// `queued`, `refusal` is empty.
#[default]
Queues,
/// An `Ok` report with nothing queued and nornir's named refusal sentence
/// (state in brackets): the driver is not installed, the lake is held elsewhere.
CannotConsume,
/// The workspace is not served — `FAILED_PRECONDITION`.
NotServed,
}
impl WorkspacesLog {
/// Every `Workspaces.Register` the mock served, in call order.
pub fn registered(&self) -> Vec<String> {
self.registered.lock().unwrap().clone()
}
/// Every `Workspaces.Fetch` the mock served, in call order. A UI test asserts
/// against THIS to prove a Populate control was actually hittable: an absent or
/// disabled button leaves it empty.
pub fn fetched(&self) -> Vec<String> {
self.fetched.lock().unwrap().clone()
}
/// Every `Workspaces.Remove` the mock served — `(name, purge)`. A test asserts
/// `purge == true` to prove Remove clears the warehouse, not only the row.
pub fn removed(&self) -> Vec<(String, bool)> {
self.removed.lock().unwrap().clone()
}
/// Every `Workspaces.Repopulate` the mock served — `(name, purge)`. THE ONE
/// REFRESH's wire proof: `purge == true` is what separates a Refresh that
/// discards the derived answers from one that merely re-asks for work the
/// chain would short-circuit.
pub fn repopulated(&self) -> Vec<(String, bool)> {
self.repopulated.lock().unwrap().clone()
}
/// Stage the **cannot-consume refusal**: the next `Repopulate` answers `Ok`
/// with nothing queued and nornir's named sentence. This is the shape a UI must
/// draw as "not done", not as a success with a zero in it.
pub fn refuse_repopulate_as_unconsumable(&self) {
*self.repopulate_behaviour.lock().unwrap() = RepopulateBehaviour::CannotConsume;
}
/// Stage the **not-served refusal**: the next `Repopulate` answers
/// `FAILED_PRECONDITION`, which reaches the client as an `Err` rather than a
/// report.
pub fn refuse_repopulate_as_not_served(&self) {
*self.repopulate_behaviour.lock().unwrap() = RepopulateBehaviour::NotServed;
}
/// Stage what `Workspaces.Get` reports per member — `(name, sync_state)`, where
/// `sync_state` is `"ok"` / `"fetching"` / `"error: …"` exactly as nornir's
/// monitor writes it. Call it between polls to make a populate advance.
pub fn set_member_states(&self, states: &[(&str, &str)]) {
*self.member_states.lock().unwrap() = states
.iter()
.map(|(n, s)| (n.to_string(), s.to_string()))
.collect();
}
/// Stage the `PopulateStatus` roll-up body.
pub fn set_populate_status_json(&self, json: &str) {
*self.populate_status_json.lock().unwrap() = json.to_string();
}
}
#[tonic::async_trait]
impl pb::workspaces_server::Workspaces for WorkspacesLog {
async fn list(
&self,
_req: tonic::Request<pb::Empty>,
) -> Result<tonic::Response<pb::WorkspaceList>, tonic::Status> {
let workspaces = self
.registered
.lock()
.unwrap()
.iter()
.map(|n| pb::WorkspaceRecord {
name: n.clone(),
mode: "monitored".into(),
..Default::default()
})
.collect();
Ok(tonic::Response::new(pb::WorkspaceList { workspaces }))
}
async fn register(
&self,
req: tonic::Request<pb::RegisterWorkspaceRequest>,
) -> Result<tonic::Response<pb::WorkspaceRecord>, tonic::Status> {
let r = req.into_inner();
if r.name.trim().is_empty() {
return Err(tonic::Status::invalid_argument(
"workspace name must not be empty",
));
}
self.registered.lock().unwrap().push(r.name.clone());
Ok(tonic::Response::new(pb::WorkspaceRecord {
name: r.name,
mode: if r.mode.is_empty() {
"pushed".into()
} else {
r.mode
},
// Recorded, NOT yet populated — Fetch is what mints a snapshot.
current_snapshot: String::new(),
members: vec![pb::WorkspaceMember {
name: "member".into(),
..Default::default()
}],
..Default::default()
}))
}
async fn get(
&self,
req: tonic::Request<pb::WorkspaceName>,
) -> Result<tonic::Response<pb::WorkspaceRecord>, tonic::Status> {
let r = req.into_inner();
if !self.registered.lock().unwrap().iter().any(|n| n == &r.name) {
return Err(tonic::Status::not_found(format!(
"no workspace `{}`",
r.name
)));
}
let members = self
.member_states
.lock()
.unwrap()
.iter()
.map(|(n, s)| pb::WorkspaceMember {
name: n.clone(),
sync_state: s.clone(),
..Default::default()
})
.collect();
Ok(tonic::Response::new(pb::WorkspaceRecord {
name: r.name,
mode: "monitored".into(),
members,
..Default::default()
}))
}
async fn populate_status(
&self,
_req: tonic::Request<pb::Empty>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
let json = self.populate_status_json.lock().unwrap().clone();
Ok(tonic::Response::new(pb::JsonResponse {
json: if json.is_empty() {
"[]".to_string()
} else {
json
},
}))
}
async fn remove(
&self,
req: tonic::Request<pb::WorkspaceName>,
) -> Result<tonic::Response<pb::Empty>, tonic::Status> {
let r = req.into_inner();
let mut reg = self.registered.lock().unwrap();
let Some(i) = reg.iter().position(|n| n == &r.name) else {
return Err(tonic::Status::not_found(format!(
"no workspace `{}`",
r.name
)));
};
reg.remove(i);
self.removed.lock().unwrap().push((r.name, r.purge));
Ok(tonic::Response::new(pb::Empty {}))
}
async fn fetch(
&self,
req: tonic::Request<pb::WorkspaceFetchRequest>,
) -> Result<tonic::Response<pb::WorkspaceFetchReport>, tonic::Status> {
let r = req.into_inner();
if !self.registered.lock().unwrap().iter().any(|n| n == &r.name) {
return Err(tonic::Status::not_found(format!(
"no workspace `{}`",
r.name
)));
}
self.fetched.lock().unwrap().push(r.name.clone());
let (changed, errors, snapshot) = match self.behaviour {
FetchBehaviour::Populates => {
(vec!["member".to_string()], vec![], "snap-1".to_string())
}
// The njord shape: the descriptor fetch worked, the member clone did
// not, and there is NO snapshot — but there IS an error to report.
FetchBehaviour::MemberCloneFails => (
vec![],
vec!["member: clone failed: Permission denied (publickey)".to_string()],
String::new(),
),
// Neither. Nothing on the served surface can say why.
FetchBehaviour::SilentlyDoesNothing => (vec![], vec![], String::new()),
};
Ok(tonic::Response::new(pb::WorkspaceFetchReport {
workspace: r.name,
fetched: u32::from(!changed.is_empty()),
changed,
errors,
snapshot,
}))
}
/// The mock's `Repopulate` — it records the call and answers in whichever of
/// the verb's THREE shapes the test staged, because a UI that renders only the
/// happy one has not been tested against the verb at all:
///
/// * **queued** (default) — the members behind come back and `refusal` is empty;
/// * **the cannot-consume refusal**
/// ([`WorkspacesLog::refuse_repopulate_as_unconsumable`]) — an `Ok` report
/// with nothing queued and nornir's named sentence;
/// * **the not-served refusal** ([`WorkspacesLog::refuse_repopulate_as_not_served`])
/// — `FAILED_PRECONDITION`.
async fn repopulate(
&self,
req: tonic::Request<pb::WorkspaceRepopulateRequest>,
) -> Result<tonic::Response<pb::WorkspaceRepopulateReport>, tonic::Status> {
let r = req.into_inner();
if !self.registered.lock().unwrap().iter().any(|n| n == &r.name) {
return Err(tonic::Status::not_found(format!(
"no workspace `{}`",
r.name
)));
}
self.repopulated
.lock()
.unwrap()
.push((r.name.clone(), r.purge));
match *self.repopulate_behaviour.lock().unwrap() {
RepopulateBehaviour::Queues => Ok(tonic::Response::new(
pb::WorkspaceRepopulateReport {
workspace: r.name,
purged: r.purge,
dropped: if r.purge {
vec!["call_edges".into(), "lens_change_risk".into(), "lens_runs".into()]
} else {
vec![]
},
generation: "lake 5f1c · edda@41 9a0b1c2 · nornir@7 77e2d10".into(),
queued: vec![
"edda behind: offset 41 @ 9a0b1c2, head 3c4d5e6".into(),
"nornir behind: offset 7 @ 77e2d10, head a1b2c3d".into(),
],
in_flight: vec![],
refusal: String::new(),
},
)),
RepopulateBehaviour::CannotConsume => Ok(tonic::Response::new(
pb::WorkspaceRepopulateReport {
workspace: r.name.clone(),
purged: r.purge,
dropped: vec![],
generation: "lake 5f1c · edda@41 9a0b1c2 · nornir@7 77e2d10".into(),
queued: vec![],
in_flight: vec![],
refusal: format!(
"repopulate `{}` [driver_not_available]: 2 member(s) behind and nothing \
can consume them — cargo-mimir-dump is not installed on this box \
(toolchain nightly-2026-10-05)",
r.name
),
},
)),
RepopulateBehaviour::NotServed => Err(tonic::Status::failed_precondition(format!(
"workspace `{}` is not served by this server — there is no open code \
room to repopulate. Register/populate it first.",
r.name
))),
}
}
}
/// A running mock `Workspaces` server + the log of what it was asked to do.
/// Dropping it shuts the server down and joins its thread.
pub struct MockWorkspacesServer {
pub endpoint: String,
/// What the mock served — `registered()` / `fetched()`.
pub log: WorkspacesLog,
_srv: MockWarehouseServer,
}
/// Spawn a mock `Workspaces` server (Register / Get / Remove / Fetch / List) on an
/// ephemeral port, whose `Fetch` populates normally.
pub fn spawn_mock_workspaces() -> MockWorkspacesServer {
spawn_mock_workspaces_with(FetchBehaviour::Populates)
}
/// The same mock with a chosen [`FetchBehaviour`], so a test can reproduce a failed
/// member clone or a populate that silently does nothing.
pub fn spawn_mock_workspaces_with(behaviour: FetchBehaviour) -> MockWorkspacesServer {
let log = WorkspacesLog {
behaviour,
..Default::default()
};
let svc = log.clone();
let srv = spawn_router(move |mut s| {
s.add_service(pb::workspaces_server::WorkspacesServer::new(svc))
});
MockWorkspacesServer {
endpoint: srv.endpoint.clone(),
log,
_srv: srv,
}
}
// ── the `Mimir` lens surface (GenericCost / ZombieFeatures / … / BenchGaps) ──
//
// The mock behind dwarves' THIN Lenses pane and the `local == thin` equivalence
// test. Each lens returns the JSON body STAGED for it (default `{}`), so a test
// can feed it the EXACT `state_json` the local fold produced and prove the wire
// round-trips it losslessly — the pane's thin path then renders the identical
// finding-set the fat/local fold would.
/// A `Mimir` server serving canned per-lens JSON. Empty ⇒ `{}` (valid, parses).
#[derive(Clone)]
pub struct MockMimir {
/// Hold every lens reply this long before answering — so a pane test can
/// prove its frame comes back BEFORE the server does (the async Scan).
/// `Default` = answer at once.
pub reply_delay: std::time::Duration,
pub generic_cost: String,
pub zombie_features: String,
pub change_risk: String,
pub similar_code: String,
pub weak_tests: String,
pub bench_gaps: String,
pub villa: String,
pub churn: String,
pub anomaly: String,
pub world_impact: String,
pub binary_drift: String,
/// The Type-4 semantic-clone lens (`2fb5a1d`). A lens verb that exists on the
/// server and NOT on this mock does not degrade a test — it stops the whole
/// crate's test build compiling, so every consumer of `testkit` goes red at once.
pub semantic_clones: String,
/// The SOURCE-door lens (sub-item clone regions): the body served when the
/// query's `repo` is not in [`Self::subclone_by_repo`] — stage nornir's
/// REFUSAL envelope here to test the empty-repo path.
pub subclone: String,
/// `Subclone` per repo (`repo` → body) — the per-repo answers, so a test can
/// prove the pane re-asks with the repo a click chose.
pub subclone_by_repo: std::collections::BTreeMap<String, String>,
/// The second SOURCE-door lens (`GroupTwins`): the body served when the
/// query's `repo` is not in [`Self::group_twins_by_repo`] — stage nornir's
/// REFUSAL envelope here to test the empty-repo path.
pub group_twins: String,
/// `GroupTwins` per repo (`repo` → body).
pub group_twins_by_repo: std::collections::BTreeMap<String, String>,
/// `ReimplementsShared` — a library re-implemented by a crate importing it.
pub reimplements_shared: String,
/// The six lenses that joined the trim on 2026-10-16.
pub data_reach: String,
pub trust: String,
pub name_collision: String,
pub symbol_identity: String,
pub undoc_api: String,
pub reachable_vuln: String,
/// The clone signals (2026-10-17).
pub exact_clones: String,
pub near_clones: String,
pub structural_clones: String,
pub role_neighbours: String,
pub same_intent: String,
pub semantic_twins: String,
/// Per lens (its nornir name, `churn`, `world_impact` …): REFUSE it with
/// this message (`Internal`) — one lens failing while the rest answer.
pub lens_fail: std::collections::BTreeMap<String, String>,
/// Per lens: hold ITS reply this much longer than the rest — one slow lens.
pub lens_delay: std::collections::BTreeMap<String, std::time::Duration>,
/// Per lens: answer `queued` (the server's jera envelope) this many times
/// before the staged body — what an asker must re-ask through.
pub queued_first: Arc<Mutex<std::collections::BTreeMap<String, u32>>>,
/// The dependency verbs (2026-10-15): `BuildOrder`'s body.
pub build_order: String,
/// `DepsOf` per repo (`repo` → body); a repo not here is refused as nornir
/// refuses one (`unknown repo`).
pub deps_of: std::collections::BTreeMap<String, String>,
/// `DependentsOf` per repo, as [`Self::deps_of`].
pub dependents_of: std::collections::BTreeMap<String, String>,
/// `AffectedByChange`'s body.
pub affected_by_change: String,
/// `ChangedSinceLastRelease`'s body; `Err(msg)` = the server refuses (Internal).
pub changed_since_last_release: Result<String, String>,
/// Every dependency-verb call as `<verb> <nornir-workspace header>` — what a
/// test reads to prove the asker forwarded the workspace.
pub calls: Arc<Mutex<Vec<String>>>,
}
impl Default for MockMimir {
fn default() -> Self {
Self {
reply_delay: std::time::Duration::ZERO,
generic_cost: String::new(),
zombie_features: String::new(),
change_risk: String::new(),
similar_code: String::new(),
weak_tests: String::new(),
bench_gaps: String::new(),
villa: String::new(),
churn: String::new(),
anomaly: String::new(),
world_impact: String::new(),
binary_drift: String::new(),
semantic_clones: String::new(),
subclone: String::new(),
subclone_by_repo: Default::default(),
group_twins: String::new(),
group_twins_by_repo: Default::default(),
reimplements_shared: String::new(),
data_reach: String::new(),
trust: String::new(),
name_collision: String::new(),
symbol_identity: String::new(),
undoc_api: String::new(),
reachable_vuln: String::new(),
exact_clones: String::new(),
near_clones: String::new(),
structural_clones: String::new(),
role_neighbours: String::new(),
same_intent: String::new(),
semantic_twins: String::new(),
lens_fail: Default::default(),
lens_delay: Default::default(),
queued_first: Arc::default(),
build_order: String::new(),
deps_of: Default::default(),
dependents_of: Default::default(),
affected_by_change: String::new(),
changed_since_last_release: Ok(String::new()),
calls: Arc::default(),
}
}
}
impl MockMimir {
fn note<T>(&self, verb: &str, req: &tonic::Request<T>) {
let ws = req
.metadata()
.get("nornir-workspace")
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string();
self.calls.lock().unwrap().push(format!("{verb} {ws}"));
}
fn per_repo(
map: &std::collections::BTreeMap<String, String>,
repo: &str,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
match map.get(repo) {
Some(b) => Ok(tonic::Response::new(pb::JsonResponse {
json: Self::body(b),
})),
None => Err(tonic::Status::internal(format!("unknown repo `{repo}`"))),
}
}
}
impl MockMimir {
fn body(s: &str) -> String {
if s.is_empty() {
"{}".to_string()
} else {
s.to_string()
}
}
}
impl MockMimir {
async fn pause(&self) {
if !self.reply_delay.is_zero() {
tokio::time::sleep(self.reply_delay).await;
}
}
/// One lens reply: the shared delay, then the lens's own knobs (delay,
/// `queued` first, refusal), then its staged body.
async fn lens(
&self,
lens: &str,
body: &str,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.pause().await;
if let Some(d) = self.lens_delay.get(lens) {
tokio::time::sleep(*d).await;
}
if let Some(m) = self.lens_fail.get(lens) {
return Err(tonic::Status::internal(m.clone()));
}
let queued = {
let mut q = self.queued_first.lock().unwrap();
match q.get_mut(lens) {
Some(n) if *n > 0 => {
*n -= 1;
true
}
_ => false,
}
};
let json = if queued {
serde_json::json!({
"kind": lens, "worst": [], "feeds": [],
"evidence_note": format!("{lens}: queued on jera — job mock-{lens}"),
"lens_meta": { "state": "queued", "lens": lens, "job_id": format!("mock-{lens}") },
})
.to_string()
} else {
Self::body(body)
};
Ok(tonic::Response::new(pb::JsonResponse { json }))
}
}
#[tonic::async_trait]
impl pb::mimir_server::Mimir for MockMimir {
async fn generic_cost(
&self,
_req: tonic::Request<pb::GenericCostQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("generic_cost", &self.generic_cost).await
}
async fn zombie_features(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("zombie_features", &self.zombie_features).await
}
async fn change_risk(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("change_risk", &self.change_risk).await
}
async fn similar_code(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("similar_code", &self.similar_code).await
}
async fn weak_tests(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("weak_tests", &self.weak_tests).await
}
async fn bench_gaps(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("bench_gaps", &self.bench_gaps).await
}
async fn villa(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("villa", &self.villa).await
}
async fn churn(
&self,
_req: tonic::Request<pb::ChurnQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("churn", &self.churn).await
}
async fn anomaly(
&self,
_req: tonic::Request<pb::AnomalyQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("anomaly", &self.anomaly).await
}
async fn world_impact(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("world_impact", &self.world_impact).await
}
async fn binary_drift(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("binary_drift", &self.binary_drift).await
}
async fn semantic_clones(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("semantic_clones", &self.semantic_clones).await
}
/// The SOURCE-door lens. Logs the asked repo (`Subclone repo=<repo>` in
/// [`MockMimir::calls`]) and serves the per-repo body when one is staged —
/// the fallback body is the empty-repo refusal a test stages.
async fn subclone(
&self,
req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
let q = req.into_inner();
self.calls
.lock()
.unwrap()
.push(format!("Subclone repo={}", q.repo));
let body = self
.subclone_by_repo
.get(&q.repo)
.cloned()
.unwrap_or_else(|| self.subclone.clone());
self.lens("subclone", &body).await
}
/// The second SOURCE-door lens, as [`Self::subclone`]: logs
/// `GroupTwins repo=<repo>` and serves the per-repo body when one is staged.
async fn group_twins(
&self,
req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
let q = req.into_inner();
self.calls
.lock()
.unwrap()
.push(format!("GroupTwins repo={}", q.repo));
let body = self
.group_twins_by_repo
.get(&q.repo)
.cloned()
.unwrap_or_else(|| self.group_twins.clone());
self.lens("group_twins", &body).await
}
async fn reimplements_shared(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
let body = self.reimplements_shared.clone();
self.lens("reimplements_shared", &body).await
}
async fn exact_clones(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("exact_clones", &self.exact_clones).await
}
async fn near_clones(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("near_clones", &self.near_clones).await
}
async fn structural_clones(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("structural_clones", &self.structural_clones).await
}
async fn role_neighbours(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("role_neighbours", &self.role_neighbours).await
}
async fn same_intent(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("same_intent", &self.same_intent).await
}
async fn semantic_twins(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("semantic_twins", &self.semantic_twins).await
}
async fn data_reach(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("data_reach", &self.data_reach).await
}
async fn trust(
&self,
_req: tonic::Request<pb::TrustQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("trust", &self.trust).await
}
async fn name_collision(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("name_collision", &self.name_collision).await
}
async fn symbol_identity(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("symbol_identity", &self.symbol_identity).await
}
async fn undoc_api(
&self,
_req: tonic::Request<pb::UndocApiQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("undoc_api", &self.undoc_api).await
}
async fn reachable_vuln(
&self,
_req: tonic::Request<pb::LensQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.lens("reachable_vuln", &self.reachable_vuln).await
}
async fn deps_of(
&self,
req: tonic::Request<pb::DepQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.note("DepsOf", &req);
self.pause().await;
Self::per_repo(&self.deps_of, &req.get_ref().repo)
}
async fn dependents_of(
&self,
req: tonic::Request<pb::DepQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.note("DependentsOf", &req);
self.pause().await;
Self::per_repo(&self.dependents_of, &req.get_ref().repo)
}
async fn affected_by_change(
&self,
req: tonic::Request<pb::AffectedQuery>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.note("AffectedByChange", &req);
self.pause().await;
Ok(tonic::Response::new(pb::JsonResponse {
json: Self::body(&self.affected_by_change),
}))
}
async fn build_order(
&self,
req: tonic::Request<pb::Empty>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.note("BuildOrder", &req);
self.pause().await;
Ok(tonic::Response::new(pb::JsonResponse {
json: Self::body(&self.build_order),
}))
}
async fn changed_since_last_release(
&self,
req: tonic::Request<pb::Empty>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
self.note("ChangedSinceLastRelease", &req);
self.pause().await;
match &self.changed_since_last_release {
Ok(b) => Ok(tonic::Response::new(pb::JsonResponse {
json: Self::body(b),
})),
Err(m) => Err(tonic::Status::internal(m.clone())),
}
}
}
/// Spawn a mock `Mimir` lens server on an ephemeral port, serving the canned
/// per-lens JSON in `mock`. Reuses the ONE `spawn_router` body (no twin server).
pub fn spawn_mock_mimir(mock: MockMimir) -> MockWarehouseServer {
spawn_router(move |mut s| s.add_service(pb::mimir_server::MimirServer::new(mock)))
}
// ── the `Viz` READ surface (`Viz.Jobs` — the jera job ledger) ────────────────
/// The gRPC status code, re-exported so a consumer's test can ask a mock to REFUSE
/// with a specific one ([`MockViz::refuse`]) without taking a `tonic`
/// dev-dependency of its own. A test crate linking tonic just to name
/// `Unimplemented` would be a second spelling of this crate's transport, and the
/// point of `testkit` is that there is one.
pub use tonic::Code;
/// A `Viz` server serving a canned `Viz.Jobs` body.
///
/// Three postures, because the pane under test has to be able to tell them apart and
/// two of them used to render identically:
///
/// * `jobs` holds a JSON array ⇒ the ledger has rows.
/// * `jobs` is empty (the [`Default`]) ⇒ the server answers `[]` — the ledger is
/// genuinely empty. **Not** "the workspace is gone".
/// * `refuse` is set ⇒ every call fails with that `Status`. This is how a test
/// produces the states that MUST be named rather than blanked: an unserved
/// workspace (`NotFound`), a forge whose proxy does not carry `Viz` yet
/// (`Unimplemented`), an unauthorized caller (`Unauthenticated`).
#[derive(Clone, Default)]
pub struct MockViz {
/// The `JsonResponse.json` body for `CloneEvents`. Empty ⇒ `[]`. Separate from
/// [`Self::jobs`] on purpose: a test that canned one and read the other would
/// pass on a proxy that forwarded both calls to the same upstream method.
pub clone_events: String,
/// The `JsonResponse.json` body for `Jobs`. Empty ⇒ `[]`.
pub jobs: String,
/// Hold every reply this long before answering — so a pane test can prove its
/// frame comes back BEFORE the server does. `Default` = answer at once.
pub reply_delay: std::time::Duration,
/// When set, EVERY call is refused with `(code, message)` instead of answered.
pub refuse: Option<(tonic::Code, String)>,
/// Every `nornir-workspace` header value the server was called with, in order —
/// so a test can prove the client actually stamped the selected workspace rather
/// than relying on a server-side default.
pub seen_workspaces: Arc<Mutex<Vec<String>>>,
}
#[tonic::async_trait]
impl pb::viz_server::Viz for MockViz {
async fn jobs(
&self,
req: tonic::Request<pb::JobsRequest>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
if !self.reply_delay.is_zero() {
tokio::time::sleep(self.reply_delay).await;
}
let ws = req
.metadata()
.get("nornir-workspace")
.and_then(|v| v.to_str().ok())
.unwrap_or_default()
.to_string();
self.seen_workspaces.lock().unwrap().push(ws);
if let Some((code, msg)) = &self.refuse {
return Err(tonic::Status::new(*code, msg.clone()));
}
let json = if self.jobs.is_empty() {
"[]".to_string()
} else {
self.jobs.clone()
};
Ok(tonic::Response::new(pb::JsonResponse { json }))
}
/// ── Added 2026-08-27, AS A REPAIR, and the sequence is the point ─────────
///
/// `viz.proto` grew `CloneEvents` and `warehouse_proxy` grew its forward in the
/// same commit — and this mock did not. The generated tonic trait has no default
/// bodies, so that left `cargo test -p nornir-warehouse-embed` at
/// `error[E0046]: not all trait items implemented, missing: clone_events` on
/// `origin/main`, while the RELEASE binaries built and installed fine: `testkit`
/// is not in a release build's graph, so the product was green and the tests were
/// red, which is the ONE ordering in which nobody notices.
///
/// The E0046 property the proto's comment advertises therefore did its job
/// exactly as designed — it made the omission a hard compile error rather than a
/// mock that silently answers `Unimplemented` on the new path. It just fired for
/// whoever compiled the tests next instead of for the author. Widening the
/// trimmed proto is TWO edits plus this one, not two.
///
/// The canned body is deliberately separate from [`Self::jobs`]: a test that
/// canned one field and read the other would pass against a proxy that forwarded
/// both calls to a single upstream method.
async fn clone_events(
&self,
req: tonic::Request<pb::VizTimelineRequest>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
if !self.reply_delay.is_zero() {
tokio::time::sleep(self.reply_delay).await;
}
let ws = req
.metadata()
.get("nornir-workspace")
.and_then(|v| v.to_str().ok())
.unwrap_or_default()
.to_string();
self.seen_workspaces.lock().unwrap().push(ws);
if let Some((code, msg)) = &self.refuse {
return Err(tonic::Status::new(*code, msg.clone()));
}
let json = if self.clone_events.is_empty() {
"[]".to_string()
} else {
self.clone_events.clone()
};
Ok(tonic::Response::new(pb::JsonResponse { json }))
}
}
/// Spawn a mock `Viz` server on an ephemeral port serving `mock`. Reuses the ONE
/// `spawn_router` body (no twin server).
pub fn spawn_mock_viz(mock: MockViz) -> MockWarehouseServer {
spawn_router(move |mut s| s.add_service(pb::viz_server::VizServer::new(mock)))
}
// ── the `CodeRoom` door ──────────────────────────────────────────────────────
/// A `CodeRoom` server answering canned bodies per surface.
///
/// * a surface in `answers` ⇒ that JSON body, verbatim (a computed answer or a
/// refusal body — the test decides);
/// * a surface NOT in `answers` ⇒ the reader's own refusal for a name it does not
/// know (`invalid_argument`), so a pane asking a surface the test forgot to feed
/// draws a named state instead of passing on an empty body;
/// * `refuse` set ⇒ EVERY call fails with that `Status` (an older forge:
/// `Unimplemented`; an unauthorized caller: `Unauthenticated`).
///
/// `asked` records `(workspace header, surface, params_json)` of every call, so a
/// test can prove WHAT the pane asked, not only that it drew.
#[derive(Clone, Default)]
pub struct MockCodeRoom {
pub answers: Arc<Mutex<HashMap<String, String>>>,
pub reply_delay: std::time::Duration,
pub refuse: Option<(tonic::Code, String)>,
pub asked: Arc<Mutex<Vec<(String, String, String)>>>,
}
impl MockCodeRoom {
/// A mock answering `answers` (surface → JSON body).
pub fn answering(answers: &[(&str, serde_json::Value)]) -> Self {
let m = Self::default();
for (surface, body) in answers {
m.set(surface, body.clone());
}
m
}
/// Replace (or add) the body of one surface while the server runs — how a
/// red-proof breaks the feed under a live pane.
pub fn set(&self, surface: &str, body: serde_json::Value) {
self.answers
.lock()
.unwrap()
.insert(surface.to_string(), body.to_string());
}
/// Stop answering one surface (it then refuses `invalid_argument`).
pub fn unset(&self, surface: &str) {
self.answers.lock().unwrap().remove(surface);
}
/// The surfaces asked so far, in order.
pub fn surfaces_asked(&self) -> Vec<String> {
self.asked
.lock()
.unwrap()
.iter()
.map(|a| a.1.clone())
.collect()
}
}
#[tonic::async_trait]
impl pb::code_room_server::CodeRoom for MockCodeRoom {
async fn answer(
&self,
req: tonic::Request<pb::CodeRoomQuestion>,
) -> Result<tonic::Response<pb::JsonResponse>, tonic::Status> {
if !self.reply_delay.is_zero() {
tokio::time::sleep(self.reply_delay).await;
}
let ws = req
.metadata()
.get("nornir-workspace")
.and_then(|v| v.to_str().ok())
.unwrap_or_default()
.to_string();
let q = req.into_inner();
self.asked
.lock()
.unwrap()
.push((ws, q.surface.clone(), q.params_json.clone()));
if let Some((code, msg)) = &self.refuse {
return Err(tonic::Status::new(*code, msg.clone()));
}
let json = match self.answers.lock().unwrap().get(&q.surface) {
Some(body) => body.clone(),
None => serde_json::json!({
"surface": q.surface,
"state": "invalid_argument",
"refused": true,
"note": format!("`{}` is not a surface this mock answers", q.surface),
})
.to_string(),
};
Ok(tonic::Response::new(pb::JsonResponse { json }))
}
}
/// Spawn a mock `CodeRoom` server on an ephemeral port serving `mock`. Reuses the
The file goes on; the console shows all of it.