facett-pipeline/src/wire.rs
//! **The [`PipelineModel`] wire** — the ONE JSON encoding of the normalized model,
//! in both directions, beside each other.
//!
//! This is the shape a host posts to the `facet-wrapped` wasm bundle's `pipeline`
//! component, and the shape knut's `knutpipe web` streams over SSE. It used to live
//! in `facet-wrapped/src/ops.rs` (row 5, 2026-09-05). It moved here on row 6 for one
//! reason: knut's web server must ENCODE it, and the only way to reuse the encoder
//! from there was a dependency on `facet-wrapped` — an unpublished `cdylib` wrapper
//! that pulls every wrapped facett component (3D maps on wgpu, vello, arrow) into a
//! server whose only job is to hand a browser some bytes. `facett-pipeline` is the
//! crate that owns the model, is already knut's dependency (`facett-adapter`), and
//! is published. The wrapper now delegates to this module, so there is still exactly
//! one encoding and one decoding.
//!
//! **Why not serde derives on the model?** `PipelineModel`, [`Stage`],
//! [`StageMetrics`] and [`PipelineEdge`] derive `Clone, Default, Debug` and nothing
//! else, and that stays so: a derive would freeze the model's field names as a wire
//! contract and make every future field a breaking change to a payload other
//! repos ship. This module is the contract instead — explicit, versioned by its
//! tests, and tolerant on the way in (a hand-written payload may spell `"ok"` for
//! `done`).
//!
//! # The shape
//!
//! ```json
//! { "title": "sdp ingest",
//! "stages": [{ "id": "stage-0", "name": "scan_bronze", "status": "done",
//! "kind": "table", "error": null,
//! "metrics": { "rows": 1200000, "throughput_rows_s": 84000.0, "lag_s": null } }],
//! "edges": [{ "from": "stage-0", "to": "stage-1", "label": "12k rows/s" }] }
//! ```
//!
//! `id` is carried separately from `name` because an engine with opaque ids (Flink
//! vertex ids) needs them and the edges reference them, not the display label.
//!
//! # The status vocabulary
//!
//! `idle | running | done | failed | unknown:<reason>`, where the reason is the
//! snake_case spelling of every [`UnknownReason`]: `unknown:not_probed`,
//! `unknown:unreachable`, `unknown:probe_failed`, `unknown:not_authorised`,
//! `unknown:ambiguous`, `unknown:stale`, `unknown:not_applicable`. `Unknown` carries
//! its REASON and must survive the trip: [`StageStatus`]'s own doc makes the
//! distinction the point of the type — `Idle` means "we know it has not started",
//! `Unknown(reason)` means "we could not find out". A wire that collapsed the second
//! into the first would turn a permissions failure into a pipeline that merely looks
//! pending.
//!
//! (The render label a `state_json` reader sees — [`StageStatus::label`] — spells
//! the same reasons with hyphens, `unknown:not-probed`. That is a READ-BACK of what
//! the view drew, not this wire; [`status_of`] accepts both spellings so a payload
//! echoed from a `state_json` still decodes to the same status.)
use serde_json::Value;
use crate::model::{PipelineEdge, PipelineModel, Stage, StageMetrics, StageStatus, UnknownReason};
/// The wire token of a [`StageStatus`].
#[must_use]
pub fn status_token(s: StageStatus) -> &'static str {
use StageStatus as S;
use UnknownReason as R;
match s {
S::Idle => "idle",
S::Running => "running",
S::Done => "done",
S::Failed => "failed",
S::Unknown(R::NotProbed) => "unknown:not_probed",
S::Unknown(R::Unreachable) => "unknown:unreachable",
S::Unknown(R::ProbeFailed) => "unknown:probe_failed",
S::Unknown(R::NotAuthorised) => "unknown:not_authorised",
S::Unknown(R::Ambiguous) => "unknown:ambiguous",
S::Unknown(R::Stale) => "unknown:stale",
S::Unknown(R::NotApplicable) => "unknown:not_applicable",
// No `_` arm on the reasons on purpose: `#[non_exhaustive]` does NOT relax
// exhaustiveness inside the defining crate, so a new reason FAILS TO COMPILE
// here until it has a wire token — it can never silently encode as a bare
// `unknown` and lose itself on the trip.
}
}
/// The inverse of [`status_token`], plus the human spellings a hand-written payload
/// is likely to use (`ok`, `success`, `error`, `crashed`, `pending`, …).
///
/// Anything unrecognised lands on `Unknown(NotProbed)` — the component's neutral
/// state — never on a flattering `Done` and never on `Idle`.
#[must_use]
pub fn status_of(s: &str) -> StageStatus {
use StageStatus as S;
use UnknownReason as R;
match s.trim().to_ascii_lowercase().as_str() {
"unknown:not_probed" | "unknown:not-probed" | "unknown" => S::Unknown(R::NotProbed),
"unknown:unreachable" => S::Unknown(R::Unreachable),
"unknown:probe_failed" | "unknown:probe-failed" => S::Unknown(R::ProbeFailed),
"unknown:not_authorised"
| "unknown:not-authorised"
| "unknown:not_authorized"
| "unknown:not-authorized" => S::Unknown(R::NotAuthorised),
"unknown:ambiguous" => S::Unknown(R::Ambiguous),
"unknown:stale" => S::Unknown(R::Stale),
"unknown:not_applicable" | "unknown:not-applicable" => S::Unknown(R::NotApplicable),
"running" | "active" | "in_progress" | "started" => S::Running,
"done" | "ok" | "success" | "succeeded" | "complete" | "completed" | "passed" => S::Done,
"failed" | "fail" | "error" | "crashed" | "aborted" => S::Failed,
"idle" | "pending" | "queued" | "waiting" | "" => S::Idle,
_ => S::Unknown(R::NotProbed),
}
}
/// **Encode a [`PipelineModel`] onto the wire** — what an adapter's host (knut's
/// `knutpipe web`, korp, dwarves) posts to the wasm bundle.
///
/// Pipeline-level [`PipelineMetrics`](crate::PipelineMetrics) are NOT on the wire
/// (yet): the wasm `pipeline` component takes the per-stage facts and derives its
/// own progress; a host that wants the aggregate bar carries it separately.
#[must_use]
pub fn to_json(m: &PipelineModel) -> Value {
let stages: Vec<Value> = m
.stages
.iter()
.map(|s| {
serde_json::json!({
"id": s.id,
"name": s.name,
"status": status_token(s.status),
"kind": s.kind,
"error": s.error,
"metrics": {
"rows": s.metrics.rows,
"throughput_rows_s": s.metrics.throughput_rows_s,
"lag_s": s.metrics.lag_s,
},
})
})
.collect();
let edges: Vec<Value> = m
.edges
.iter()
.map(|e| serde_json::json!({ "from": e.from, "to": e.to, "label": e.label }))
.collect();
serde_json::json!({ "title": m.title, "stages": stages, "edges": edges })
}
/// **Decode the wire back into a [`PipelineModel`]** — the inverse of [`to_json`].
///
/// `None` when there is nothing renderable (no `stages` array, or every stage
/// nameless), so a loader shows its honest refusal rather than an empty DAG. A
/// stage with no name is dropped — an unnamed box on a DAG tells a reader nothing —
/// and the id falls back to the name, matching [`Stage::new`].
#[must_use]
pub fn from_json(v: &Value) -> Option<PipelineModel> {
let arr = v.get("stages")?.as_array()?;
let stages: Vec<Stage> = arr
.iter()
.filter_map(|s| {
let name = s
.get("name")
.or_else(|| s.get("id"))
.and_then(Value::as_str)
.filter(|t| !t.trim().is_empty())?;
let id = s.get("id").and_then(Value::as_str).filter(|t| !t.trim().is_empty());
let status = status_of(s.get("status").and_then(Value::as_str).unwrap_or("idle"));
let mut stage = Stage::new(name, status);
if let Some(i) = id {
stage = stage.with_id(i);
}
if let Some(k) = s.get("kind").and_then(Value::as_str).filter(|t| !t.is_empty()) {
stage = stage.with_kind(k);
}
if let Some(e) = s.get("error").and_then(Value::as_str).filter(|t| !t.is_empty()) {
stage = stage.with_error(e);
}
let met = s.get("metrics");
stage = stage.with_metrics(StageMetrics {
rows: met.and_then(|m| m.get("rows")).and_then(Value::as_u64),
throughput_rows_s: met
.and_then(|m| m.get("throughput_rows_s"))
.and_then(Value::as_f64)
.map(|f| f as f32),
lag_s: met.and_then(|m| m.get("lag_s")).and_then(Value::as_f64).map(|f| f as f32),
});
Some(stage)
})
.collect();
if stages.is_empty() {
return None;
}
let edges: Vec<PipelineEdge> = v
.get("edges")
.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(|e| {
let from = e.get("from").and_then(Value::as_str)?;
let to = e.get("to").and_then(Value::as_str)?;
let mut edge = PipelineEdge::new(from, to);
edge.label = e
.get("label")
.and_then(Value::as_str)
.filter(|s| !s.is_empty())
.map(str::to_string);
Some(edge)
})
.collect()
})
.unwrap_or_default();
Some(PipelineModel {
title: v.get("title").and_then(Value::as_str).unwrap_or("pipeline").to_string(),
stages,
edges,
metrics: None,
})
}
#[cfg(test)]
mod tests {
use super::*;
/// Every reason has a token, and every token decodes to the reason it came
/// from — including the three (`ambiguous`, `stale`, `not_applicable`) the
/// wrapper's copy of this codec used to flatten into a bare `unknown`.
#[test]
fn every_unknown_reason_round_trips_by_name() {
let reasons = [
UnknownReason::NotProbed,
UnknownReason::Unreachable,
UnknownReason::ProbeFailed,
UnknownReason::NotAuthorised,
UnknownReason::Ambiguous,
UnknownReason::Stale,
UnknownReason::NotApplicable,
];
for r in reasons {
let token = status_token(StageStatus::Unknown(r));
assert!(token.starts_with("unknown:"), "{r:?} → {token}");
assert_eq!(status_of(token), StageStatus::Unknown(r), "{token}");
// The hyphenated render label decodes to the same status.
assert_eq!(status_of(StageStatus::Unknown(r).label()), StageStatus::Unknown(r));
}
for s in [StageStatus::Idle, StageStatus::Running, StageStatus::Done, StageStatus::Failed] {
assert_eq!(status_of(status_token(s)), s);
}
}
#[test]
fn an_unrecognised_word_is_unknown_never_done_or_idle() {
assert_eq!(status_of("green"), StageStatus::Unknown(UnknownReason::NotProbed));
assert_eq!(status_of("ok"), StageStatus::Done);
assert_eq!(status_of("pending"), StageStatus::Idle);
}
#[test]
fn nothing_renderable_is_none() {
assert!(from_json(&serde_json::json!({})).is_none());
assert!(from_json(&serde_json::json!({ "stages": [] })).is_none());
assert!(from_json(&serde_json::json!({ "stages": [{ "status": "done" }] })).is_none());
}
}