crates/knut-pipelines/src/web.rs
//! Web viewer (feature `web`): an `axum` server + Server-Sent Events feeding the
//! **Rust** `facett-pipeline` DAG viewer, compiled to wasm.
//!
//! A third frontend over the *same* model + `PipelineEventSink` seam as the
//! CLI and the egui GUI — never a replacement for them. knut draws **nothing in
//! JavaScript** here. The page is facett's `facet-wrapped` bundle — the real
//! `facett_pipeline::PipelineView` behind a wasm-bindgen boundary — and knut's job
//! is to hand a browser those bytes and a stream of `PipelineModel` frames. (This
//! module used to carry a hand-written SVG/JS drawing of the DAG; lane SW row 6,
//! 2026-09-05, deleted it: a JavaScript reimplementation of a facett component is
//! exactly the thing that went badly once already.)
//!
//! **Handing a browser bytes is not knut's expertise, and as of 2026-09-06 it is not
//! knut's code either.** Bundle discovery, validation, extension → content type, the
//! entry redirect and the SSE framing all moved to [`knut_wasm_host`], a crate that
//! knows nothing about facett, pipelines, maps or graphs. What is left in this file
//! is the part that IS knut's: the DAG, the `PipelineModel` wire, and the event
//! sources that drive it. The three file names below and the recipe for producing
//! them are facett's business, so knut declares them and the host quotes them back.
//!
//! It serves:
//!
//! * `GET /` — redirects to the loader, told to mount the
//! `pipeline` component and read frames from `/events`;
//! * `GET /index.html`, `GET /facet_wrapped.js`, `GET /facet_wrapped_bg.wasm` —
//! the facet-wrapped bundle, as bytes, from a directory named at run time
//! ([`load_bundle`]); a missing or malformed bundle answers `503` with the exact
//! file list and the build recipe, never a blank page;
//! * `GET /graph.json` — the dataflow DAG as JSON (reusing [`crate::dump_json`]);
//! * `GET /pipeline.json` — ONE frame: the all-idle [`PipelineModel`] on the wire;
//! * `GET /events` — a stream of frames over SSE, one per run event,
//! ended by `{"done":true}` (or `{"error":"…"}`).
//!
//! **The wire.** Every frame is [`crate::facett_adapter::to_pipeline_model`] encoded
//! by [`facett_pipeline::wire::to_json`] — the ONE codec of the model, owned by the
//! crate that owns the model, decoded by the same function inside the wasm. Nothing
//! here spells a field name, and the host does not re-encode what this file encoded.
//!
//! **Demoable with no Spark server:** `knutpipe web` (no `--connect`) serves the
//! built-in demo graph and replays a synthetic event sequence over SSE so the DAG
//! animates. With `--connect` (feature `grpc`) it streams a live run.
//!
//! The event source — synthetic or live — is driven through a
//! [`WebEventSink`], a [`PipelineEventSink`] that folds each event into a single
//! [`ViewerState`] (no duplicated model or attribution) and broadcasts the
//! resulting frame. The CLI `show`/`dump_json` stays the parity floor.
//!
//! [`PipelineModel`]: facett_pipeline::PipelineModel
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use axum::Router;
use axum::extract::State;
use axum::http::header;
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use knut_wasm_host::{Bundle, BundleError, BundleSpec, EntryPoint, FrameSender, WasmHost};
use crate::facett_adapter::to_pipeline_model;
use crate::model::{DataflowGraph, PipelineRunEvent};
use crate::progress::{ViewerState, ViewerUpdate};
use crate::viewer::PipelineEventSink;
// ─────────────────────────────────────────────────────────────────────────────
// The facet-wrapped bundle — the viewer's bytes. knut says WHICH files and HOW to
// build them; knut-wasm-host does the opening, the checking and the serving.
// ─────────────────────────────────────────────────────────────────────────────
/// Environment variable naming the bundle directory when the caller gives none
/// (`knutpipe web` without `--facet-dist`; korp's demo calling [`serve`]).
pub const BUNDLE_DIR_ENV: &str = "FACET_WRAPPED_DIST";
/// The three files a bundle directory must hold. `facet_wrapped.js` and
/// `facet_wrapped_bg.wasm` are what facett's `cargo xtask wrap-wasm` writes to
/// `facet-wrapped/dist/`; `index.html` is facet-wrapped's host-agnostic loader
/// (`pkg-facet-wrapped/src/facet_wrapped_web/assets/index.html`), the same file its
/// wheel ships beside the wasm.
pub const BUNDLE_FILES: [&str; 3] = ["index.html", "facet_wrapped.js", "facet_wrapped_bg.wasm"];
/// The facett component the loader is told to mount.
pub const COMPONENT: &str = "pipeline";
/// Where that component reads its frames from.
pub const EVENTS_URL: &str = "/events";
/// How a person produces the bundle. Printed verbatim by the host in every refusal
/// — it is the one thing a blank page never tells you.
const BUNDLE_RECIPE: &str = "\
The page is facett's Rust `pipeline` viewer compiled to wasm; knut serves it,
it does not draw it. Produce the bundle in the facett checkout:
cargo xtask wrap-wasm # -> facet-wrapped/dist/{facet_wrapped.js,facet_wrapped_bg.wasm}
then put pkg-facet-wrapped/src/facet_wrapped_web/assets/index.html beside them
(the loader; facett's make_wheel.py does the same copy the other way) and point
knutpipe at that directory: `knutpipe web --facet-dist DIR`.";
/// What knut expects of a facet-wrapped bundle directory, in the host's terms.
#[must_use]
pub fn bundle_spec() -> BundleSpec {
BundleSpec::new(
"knutpipe web: the facet-wrapped wasm bundle is not available.",
BUNDLE_FILES,
BUNDLE_RECIPE,
)
.with_env_var(BUNDLE_DIR_ENV)
// knut's own sanity check on knut's own file: a page that does not pull the
// glue module in cannot be facet-wrapped's loader, whatever else it is.
.requiring_content(
BUNDLE_FILES[0],
BUNDLE_FILES[1],
"not facet-wrapped's loader page",
)
}
/// Read the facet-wrapped bundle from `dir`, refusing with every problem at once.
pub fn load_bundle(dir: impl AsRef<Path>) -> Result<Bundle, BundleError> {
bundle_spec().load(dir)
}
/// [`load_bundle`] from the directory named by [`BUNDLE_DIR_ENV`]; an unset variable
/// is a refusal like any other (and it says which variable).
pub fn bundle_from_env() -> Result<Bundle, BundleError> {
bundle_spec().load_from_env()
}
/// Where `GET /` sends the browser: the loader, told which component to mount and
/// where its frames come from. The loader owns the `EventSource`; knut adds no
/// script of its own.
#[must_use]
pub fn entry_point() -> EntryPoint {
EntryPoint::new(format!("/{}", BUNDLE_FILES[0]), COMPONENT, EVENTS_URL)
}
/// The URL [`entry_point`] redirects to.
#[must_use]
pub fn loader_url() -> String {
entry_point().url()
}
/// Keep an idle SSE connection (and any proxy in front of it) from timing out
/// between run events — a Spark stage can be quiet for minutes.
const SSE_KEEP_ALIVE: Duration = Duration::from_secs(15);
// ─────────────────────────────────────────────────────────────────────────────
// The frames — one codec, upstream.
// ─────────────────────────────────────────────────────────────────────────────
/// Encode one viewer snapshot as a frame for the wasm `pipeline` component: the
/// Spark/SDP adapter ([`to_pipeline_model`]) then facett's own codec
/// ([`facett_pipeline::wire::to_json`]). This is the ONLY place a frame is made,
/// and it names no field.
pub fn wire_frame(state: &ViewerState) -> String {
facett_pipeline::wire::to_json(&to_pipeline_model(state)).to_string()
}
/// Shared server state for the DATA routes: the DAG to render and how to source its
/// run events. The viewer's bytes are not here — they are the host's.
#[derive(Clone)]
struct AppState {
/// The static dataflow graph (edges already completed via `populate_reads`).
graph: Arc<DataflowGraph>,
/// Delay between synthetic replay events (0 in tests → instant).
delay: Duration,
/// Spark Connect URL for a live run; `None` = synthetic demo replay.
connect: Option<String>,
}
/// A [`PipelineEventSink`] that folds run events into one [`ViewerState`] and
/// pushes each resulting frame ([`wire_frame`]) to an SSE channel. The wasm
/// renders purely from these frames, so attribution/status/rows come from the
/// shared model — the web view can show nothing the CLI/GUI can't.
struct WebEventSink {
state: ViewerState,
tx: FrameSender,
}
impl WebEventSink {
fn new(graph: DataflowGraph, tx: FrameSender) -> Self {
WebEventSink {
state: ViewerState::new(graph),
tx,
}
}
/// Broadcast the current viewer snapshot, on the wire, to the SSE client.
fn send_snapshot(&self) {
self.tx.frame(wire_frame(&self.state));
}
/// Fold a live [`ViewerUpdate`] (event **or** job progress) into the state
/// and broadcast the new frame. Used by the live (`grpc`) path.
#[cfg(feature = "grpc")]
fn apply_update(&mut self, update: ViewerUpdate) {
self.state.apply(update);
self.send_snapshot();
}
/// Broadcast the terminal control frame; the loader closes its `EventSource`
/// and leaves the last frame up. The host has no opinion about this string —
/// it is knut's protocol with facet-wrapped's loader.
fn send_done(&self) {
self.tx.finish(r#"{"done":true}"#);
}
/// Broadcast an error control frame (e.g. a live connection failed); the
/// loader shows it in its refusal panel.
#[cfg(feature = "grpc")]
fn send_error(&self, msg: &str) {
self.tx
.finish(serde_json::json!({ "error": msg }).to_string());
}
}
impl PipelineEventSink for WebEventSink {
fn on_event(&mut self, event: &PipelineRunEvent) -> crate::Result<()> {
// Through the seam: fold the event into the shared ViewerState, then
// publish the frame the wasm renders.
self.state.apply(ViewerUpdate::Event(event.clone()));
self.send_snapshot();
Ok(())
}
fn on_finish(&mut self) -> crate::Result<()> {
self.send_done();
Ok(())
}
}
/// Build the axum router for `graph`, sourcing events synthetically (`connect =
/// None`) or from a live Spark Connect run. `delay` paces the synthetic replay.
/// `bundle` is the viewer's bytes — or the refusal the page routes serve.
///
/// knut's three data routes are EXPLICIT and the host is the fallback, so `/` and
/// every bundle file go to the host while `/events` and friends stay knut's.
fn router(
graph: DataflowGraph,
connect: Option<String>,
delay: Duration,
bundle: Result<Bundle, BundleError>,
) -> Router {
let state = AppState {
graph: Arc::new(graph),
delay,
connect,
};
Router::new()
.route("/graph.json", get(graph_json))
.route("/pipeline.json", get(pipeline_json))
.route(EVENTS_URL, get(events))
.with_state(state)
.merge(knut_wasm_host::router(WasmHost::new(bundle, entry_point())))
}
/// The default synthetic-replay pace: one run event every 450 ms, so the demo
/// DAG animates at a human-watchable speed.
const DEFAULT_REPLAY_DELAY: Duration = Duration::from_millis(450);
/// Serve the web viewer on `addr` until the process is stopped, with the bundle
/// from [`BUNDLE_DIR_ENV`]. See [`serve_bundle`] to pass the bundle explicitly.
pub async fn serve(
graph: DataflowGraph,
connect: Option<String>,
addr: SocketAddr,
) -> crate::Result<()> {
serve_bundle(graph, connect, addr, bundle_from_env()).await
}
/// [`serve`], with the viewer's bytes (or the refusal to serve instead) chosen by
/// the caller — `knutpipe web --facet-dist DIR` comes through here.
pub async fn serve_bundle(
graph: DataflowGraph,
connect: Option<String>,
addr: SocketAddr,
bundle: Result<Bundle, BundleError>,
) -> crate::Result<()> {
let listener = tokio::net::TcpListener::bind(addr)
.await
.map_err(|e| crate::PipelineError::Other(format!("bind {addr}: {e}")))?;
serve_on_bundle(listener, graph, connect, DEFAULT_REPLAY_DELAY, bundle).await
}
/// Serve the web viewer on an **already-bound** `TcpListener` — the seam that
/// lets a caller pick an ephemeral OS port (`bind("127.0.0.1:0")`), recover the
/// concrete [`SocketAddr`] via [`TcpListener::local_addr`], and *then* hand the
/// live listener here with no bind race. The browser robot test
/// (`tests/robot_web_viewer.rs`) uses exactly this to point `fantoccini` at the
/// real server. `delay` paces the synthetic replay (pass [`Duration::ZERO`] for
/// an instant animation in tests). The bundle comes from [`BUNDLE_DIR_ENV`]; see
/// [`serve_on_bundle`] to pass it.
///
/// [`TcpListener`]: tokio::net::TcpListener
/// [`TcpListener::local_addr`]: tokio::net::TcpListener::local_addr
pub async fn serve_on(
listener: tokio::net::TcpListener,
graph: DataflowGraph,
connect: Option<String>,
delay: Duration,
) -> crate::Result<()> {
serve_on_bundle(listener, graph, connect, delay, bundle_from_env()).await
}
/// [`serve_on`] with the bundle chosen by the caller. A missing bundle does NOT
/// stop the server — the data routes still answer, and the page routes say
/// exactly what is missing — but it is said out loud here, at startup, too.
pub async fn serve_on_bundle(
listener: tokio::net::TcpListener,
graph: DataflowGraph,
connect: Option<String>,
delay: Duration,
bundle: Result<Bundle, BundleError>,
) -> crate::Result<()> {
match &bundle {
Ok(b) => {
tracing::info!(target: "knutpipe::web", dir = %b.dir().display(), "facet-wrapped bundle loaded");
eprintln!("# knutpipe web: facet-wrapped bundle from {}", b.dir().display());
}
Err(e) => {
tracing::warn!(target: "knutpipe::web", %e, "facet-wrapped bundle unavailable; page routes will answer 503");
eprintln!("# {e}");
}
}
let app = router(graph, connect, delay, bundle);
let bound = listener
.local_addr()
.map_err(|e| crate::PipelineError::Other(e.to_string()))?;
tracing::info!(target: "knutpipe::web", %bound, "web viewer listening");
eprintln!("# knutpipe web viewer on http://{bound}");
axum::serve(listener, app)
.await
.map_err(|e| crate::PipelineError::Other(format!("web serve: {e}")))
}
/// `GET /graph.json` — the DAG as JSON, reusing [`crate::dump_json`].
async fn graph_json(State(st): State<AppState>) -> impl IntoResponse {
match crate::dump_json(&st.graph) {
Ok(body) => (
[(header::CONTENT_TYPE, "application/json")],
body,
),
Err(e) => (
[(header::CONTENT_TYPE, "application/json")],
serde_json::json!({ "error": e.to_string() }).to_string(),
),
}
}
/// `GET /pipeline.json` — one frame, the all-idle model on the wire: what the first
/// `/events` frame carries, for a script (or a test) that wants the shape without
/// holding an SSE stream open.
async fn pipeline_json(State(st): State<AppState>) -> impl IntoResponse {
(
[(header::CONTENT_TYPE, "application/json")],
wire_frame(&ViewerState::new((*st.graph).clone())),
)
}
/// `GET /events` — SSE stream of frames. Each connection gets its own event
/// source (a synthetic replay, or a live run). The framing is the host's; the
/// frames are knut's, already encoded, and nothing re-encodes them.
async fn events(State(st): State<AppState>) -> Response {
let (tx, body) = knut_wasm_host::sse_channel();
tokio::spawn(run_source(st, tx));
knut_wasm_host::sse_response(body.with_keep_alive(SSE_KEEP_ALIVE))
}
/// Drive one connection's event source into the SSE channel via [`WebEventSink`].
async fn run_source(st: AppState, tx: FrameSender) {
let mut sink = WebEventSink::new((*st.graph).clone(), tx);
sink.send_snapshot(); // initial (all-idle) frame
#[cfg(feature = "grpc")]
if let Some(url) = st.connect.clone() {
drive_live(&url, &mut sink).await;
let _ = sink.on_finish();
return;
}
#[cfg(not(feature = "grpc"))]
let _ = &st.connect;
// Synthetic demo replay through the sink seam.
for ev in synthetic_run(&st.graph) {
if !st.delay.is_zero() {
tokio::time::sleep(st.delay).await;
}
let _ = sink.on_event(&ev);
}
let _ = sink.on_finish();
}
/// Stream a live Spark Connect run's updates into the sink (feature `grpc`).
#[cfg(feature = "grpc")]
async fn drive_live(url: &str, sink: &mut WebEventSink) {
use crate::PipelinesClient;
use crate::model::PipelineRun;
let mut client = match PipelinesClient::connect(url).await {
Ok(c) => c,
Err(e) => {
sink.send_error(&format!("connect {url}: {e}"));
return;
}
};
let run = PipelineRun::default();
let stream = match client.run_updates(&run).await {
Ok(s) => s,
Err(e) => {
sink.send_error(&format!("start run: {e}"));
return;
}
};
drive_updates(sink, stream).await;
}
/// The pure translation core of [`drive_live`]: fold a stream of live run
/// updates into the sink, publishing one frame per successful update and an
/// error frame (then stop) on the first failure. Factored out so the event →
/// SSE-frame fan-out can be unit-tested offline with a canned update stream —
/// no tonic channel, no network.
#[cfg(feature = "grpc")]
async fn drive_updates<S, E>(sink: &mut WebEventSink, mut stream: S)
where
S: futures_util::Stream<Item = Result<ViewerUpdate, E>> + Unpin,
E: std::fmt::Display,
{
use futures_util::StreamExt;
while let Some(item) = stream.next().await {
match item {
Ok(update) => sink.apply_update(update),
Err(e) => {
sink.send_error(&e.to_string());
break;
}
}
}
}
/// A deterministic synthetic run for the demo: in topological order, each flow
/// STARTs then COMPLETEs with a row count. Attribution (flow name → target
/// dataset) and status/row parsing are handled by [`ViewerState`], so this only
/// produces the message text a real Spark SDP run would emit.
pub(crate) fn synthetic_run(graph: &DataflowGraph) -> Vec<PipelineRunEvent> {
let order = graph
.topo_order()
.unwrap_or_else(|| graph.datasets.iter().map(|d| d.name.clone()).collect());
let mut events = Vec::new();
let mut rows = 1000u64;
for dataset in order {
for flow in graph.flows_into(&dataset) {
events.push(PipelineRunEvent::info(format!(
"Flow '{}' is STARTING.",
flow.name
)));
events.push(PipelineRunEvent::info(format!(
"Flow '{}' has COMPLETED and wrote {rows} rows.",
flow.name
)));
rows = (rows / 2).max(1);
}
}
events
}
#[cfg(test)]
mod tests {
use super::*;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use facett_pipeline::{StageStatus, UnknownReason};
use http_body_util::BodyExt;
use knut_wasm_host::WASM_MAGIC;
use std::path::PathBuf;
use tower::ServiceExt; // oneshot
fn demo() -> DataflowGraph {
use crate::model::{Dataset, Flow, OutputType};
let mut g = DataflowGraph::new()
.with_dataset(Dataset::new("bronze", OutputType::Table))
.with_dataset(Dataset::new("silver", OutputType::MaterializedView))
.with_dataset(Dataset::new("gold", OutputType::MaterializedView))
.with_flow(
Flow::new("ingest", "bronze", Vec::<String>::new())
.with_query("SELECT * FROM range(0, 100)"),
)
.with_flow(
Flow::new("clean", "silver", Vec::<String>::new())
.with_query("SELECT * FROM bronze"),
)
.with_flow(
Flow::new("rollup", "gold", Vec::<String>::new())
.with_query("SELECT * FROM silver"),
);
g.populate_reads();
g
}
async fn body_string(resp: axum::response::Response) -> String {
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
String::from_utf8(bytes.to_vec()).unwrap()
}
async fn get(app: Router, uri: &str) -> axum::response::Response {
app.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
.await
.unwrap()
}
fn content_type(resp: &axum::response::Response) -> String {
resp.headers()
.get(header::CONTENT_TYPE)
.unwrap()
.to_str()
.unwrap()
.to_string()
}
/// The REAL facet-wrapped loader page — what a served bundle carries. From the
/// bundle dir named by `FACET_WRAPPED_DIST` when one is set, else from the
/// sibling facett checkout (the same `../../knut` neighbourhood facett's own
/// test matrix relies on to reach knut). A copy of it in knut would be one more
/// thing to drift, so there is none.
fn real_loader_html() -> Vec<u8> {
if let Some(dir) = std::env::var_os(BUNDLE_DIR_ENV) {
if let Ok(b) = std::fs::read(Path::new(&dir).join("index.html")) {
return b;
}
}
let sibling = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../../facett/pkg-facet-wrapped/src/facet_wrapped_web/assets/index.html");
std::fs::read(&sibling).unwrap_or_else(|e| {
panic!(
"facet-wrapped's loader page is needed to test what knut serves: {} ({e}). \
Set {BUNDLE_DIR_ENV} to a bundle directory or check facett out beside knut.",
sibling.display()
)
})
}
/// A bundle directory under the OS temp dir: the REAL loader, a stub ES module
/// for the glue and a `\0asm`-headed stub for the wasm. The bytes only have to
/// be a bundle as far as [`load_bundle`] can tell — the router test is about
/// what knut serves and how, not about running wasm.
fn stub_bundle_dir(tag: &str, wasm: &[u8]) -> PathBuf {
let dir = std::env::temp_dir().join(format!("knut-web-bundle-{}-{tag}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
std::fs::write(dir.join("index.html"), real_loader_html()).unwrap();
std::fs::write(
dir.join("facet_wrapped.js"),
b"export default async function init() {}\n",
)
.unwrap();
std::fs::write(dir.join("facet_wrapped_bg.wasm"), wasm).unwrap();
dir
}
fn stub_bundle(tag: &str) -> Result<Bundle, BundleError> {
load_bundle(stub_bundle_dir(tag, &[0x00, b'a', b's', b'm', 1, 0, 0, 0]))
}
/// The `data:` payloads of an SSE body, parsed.
fn sse_frames(body: &str) -> Vec<serde_json::Value> {
body.lines()
.filter_map(|l| l.strip_prefix("data:"))
.map(|d| serde_json::from_str(d.trim()).unwrap())
.collect()
}
/// The sorted key set of a JSON object.
fn keys(v: &serde_json::Value) -> Vec<&str> {
let mut k: Vec<&str> = v.as_object().unwrap().keys().map(String::as_str).collect();
k.sort_unstable();
k
}
/// Every status token the wasm `pipeline` component accepts, as spelled by the
/// codec the frames are made with — no local copy of the vocabulary.
fn is_wire_status(s: &str) -> bool {
let all = [
StageStatus::Idle,
StageStatus::Running,
StageStatus::Done,
StageStatus::Failed,
StageStatus::Unknown(UnknownReason::NotProbed),
StageStatus::Unknown(UnknownReason::Unreachable),
StageStatus::Unknown(UnknownReason::ProbeFailed),
StageStatus::Unknown(UnknownReason::NotAuthorised),
];
all.iter().any(|st| facett_pipeline::wire::status_token(*st) == s)
}
/// Assert `frame` has the row-5 wire shape, field by field: exactly the keys the
/// wasm decodes, and nothing of `ViewerState::state_json`'s.
fn assert_wire_shape(frame: &serde_json::Value) {
assert_eq!(keys(frame), ["edges", "stages", "title"], "frame: {frame}");
assert!(frame["title"].is_string());
for s in frame["stages"].as_array().unwrap() {
assert_eq!(keys(s), ["error", "id", "kind", "metrics", "name", "status"], "stage: {s}");
assert!(s["id"].is_string() && s["name"].is_string());
assert!(s["kind"].is_string() || s["kind"].is_null());
assert!(s["error"].is_string() || s["error"].is_null());
let status = s["status"].as_str().unwrap();
assert!(is_wire_status(status), "status {status:?} is not in the wire vocabulary");
assert_eq!(keys(&s["metrics"]), ["lag_s", "rows", "throughput_rows_s"], "metrics: {s}");
assert!(s["metrics"]["rows"].is_u64() || s["metrics"]["rows"].is_null());
}
for e in frame["edges"].as_array().unwrap() {
assert_eq!(keys(e), ["from", "label", "to"], "edge: {e}");
assert!(e["from"].is_string() && e["to"].is_string());
}
}
fn stage<'a>(frame: &'a serde_json::Value, name: &str) -> &'a serde_json::Value {
frame["stages"]
.as_array()
.unwrap()
.iter()
.find(|s| s["name"] == name)
.unwrap_or_else(|| panic!("no stage {name:?} in {frame}"))
}
#[tokio::test]
async fn graph_json_endpoint_serves_the_dag() {
let app = router(demo(), None, Duration::ZERO, stub_bundle("graph"));
let resp = get(app, "/graph.json").await;
assert_eq!(resp.status(), StatusCode::OK);
let body = body_string(resp).await;
let v: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(v["datasets"].as_array().unwrap().len(), 3);
// Edges were completed by populate_reads() before serving.
let clean = v["flows"]
.as_array()
.unwrap()
.iter()
.find(|f| f["name"] == "clean")
.unwrap();
assert_eq!(clean["reads"], serde_json::json!(["bronze"]));
}
/// The data route: `/events` streams the row-5 `PipelineModel` wire, field by
/// field, from the all-idle first frame to the all-done last one, then the
/// `{"done":true}` control frame the loader closes on.
#[tokio::test]
async fn events_stream_the_pipeline_model_wire_to_completion() {
let app = router(demo(), None, Duration::ZERO, stub_bundle("events")); // instant replay
let resp = get(app, "/events").await;
assert_eq!(resp.status(), StatusCode::OK);
assert!(content_type(&resp).contains("text/event-stream"));
let body = body_string(resp).await;
let frames = sse_frames(&body);
// Terminal control frame last; every other frame is a model on the wire.
assert_eq!(frames.last().unwrap(), &serde_json::json!({ "done": true }));
let models = &frames[..frames.len() - 1];
// 1 idle frame + 3 flows × (STARTING + COMPLETED).
assert_eq!(models.len(), 7, "frames: {frames:#?}");
for m in models {
assert_wire_shape(m);
assert!(m.get("node_status").is_none(), "state_json leaked onto the wire: {m}");
}
// First frame: the DAG, all idle, kinds from the output types, edges by id.
let first = &models[0];
assert_eq!(first["title"], "pipeline");
let names: Vec<&str> = first["stages"].as_array().unwrap().iter().map(|s| s["name"].as_str().unwrap()).collect();
assert_eq!(names, ["bronze", "silver", "gold"]);
for s in first["stages"].as_array().unwrap() {
assert_eq!(s["id"], s["name"]);
assert_eq!(s["status"], "idle");
assert_eq!(s["error"], serde_json::Value::Null);
assert_eq!(s["metrics"], serde_json::json!({ "rows": null, "throughput_rows_s": null, "lag_s": null }));
}
assert_eq!(stage(first, "bronze")["kind"], "table");
assert_eq!(stage(first, "silver")["kind"], "materialized_view");
assert_eq!(
first["edges"],
serde_json::json!([
{ "from": "bronze", "to": "silver", "label": null },
{ "from": "silver", "to": "gold", "label": null },
])
);
// A mid-run frame: bronze done with its rows, silver running, gold idle.
let mid = &models[3];
assert_eq!(stage(mid, "bronze")["status"], "done");
assert_eq!(stage(mid, "bronze")["metrics"]["rows"], 1000);
assert_eq!(stage(mid, "silver")["status"], "running");
assert_eq!(stage(mid, "gold")["status"], "idle");
// Last frame: everything done, the rows knut parsed from each COMPLETED event.
let last = models.last().unwrap();
for (name, rows) in [("bronze", 1000), ("silver", 500), ("gold", 250)] {
assert_eq!(stage(last, name)["status"], "done", "{name}");
assert_eq!(stage(last, name)["metrics"]["rows"], rows, "{name}");
}
}
/// `/pipeline.json` is the first `/events` frame, byte for byte.
#[tokio::test]
async fn pipeline_json_is_the_first_frame() {
let one = body_string(get(router(demo(), None, Duration::ZERO, stub_bundle("pj1")), "/pipeline.json").await).await;
let stream = body_string(get(router(demo(), None, Duration::ZERO, stub_bundle("pj2")), "/events").await).await;
let first_line = stream.lines().find_map(|l| l.strip_prefix("data:")).unwrap().trim();
assert_eq!(one, first_line);
assert_wire_shape(&serde_json::from_str(&one).unwrap());
}
/// The frame IS facett's codec over knut's adapter, and the codec round-trips
/// every field — including an `Unknown(NotAuthorised)` stage, whose reason must
/// survive the trip rather than flatten into a merely-pending `idle`.
#[test]
fn wire_frame_round_trips_through_facetts_codec_including_not_authorised() {
let vs = ViewerState::new(demo());
let model = to_pipeline_model(&vs);
assert_eq!(wire_frame(&vs), facett_pipeline::wire::to_json(&model).to_string());
let mut m = model;
m.stages[1].status = StageStatus::Unknown(UnknownReason::NotAuthorised);
m.stages[1].error = Some("RBAC: no read on silver".to_string());
m.stages[2] = std::mem::take(&mut m.stages[2]).with_rows(42);
let v = facett_pipeline::wire::to_json(&m);
assert_wire_shape(&v);
assert_eq!(v["stages"][1]["status"], "unknown:not_authorised");
let back = facett_pipeline::wire::from_json(&v).expect("a renderable model");
assert_eq!(back.title, m.title);
assert_eq!(back.stages.len(), m.stages.len());
for (a, b) in back.stages.iter().zip(&m.stages) {
assert_eq!(a.id, b.id);
assert_eq!(a.name, b.name);
assert_eq!(a.status, b.status);
assert_eq!(a.kind, b.kind);
assert_eq!(a.error, b.error);
assert_eq!(a.metrics.rows, b.metrics.rows);
assert_eq!(a.metrics.throughput_rows_s, b.metrics.throughput_rows_s);
assert_eq!(a.metrics.lag_s, b.metrics.lag_s);
}
assert_eq!(back.stages[1].status, StageStatus::Unknown(UnknownReason::NotAuthorised));
assert_eq!(back.edges.len(), m.edges.len());
for (a, b) in back.edges.iter().zip(&m.edges) {
assert_eq!((&a.from, &a.to, &a.label), (&b.from, &b.to, &b.label));
}
}
/// The page knut serves is facet-wrapped's loader for the Rust `pipeline`
/// viewer — NOT a hand-drawn SVG. `GET /` sends the browser to the loader with
/// `component=pipeline` and the SSE URL; the loader has no `<svg`, no
/// `createElementNS`, no inline drawing; it imports `facet_wrapped.js` (which
/// pulls `facet_wrapped_bg.wasm`) and hands each frame to the wasm
/// (`facet_mount_with_data`, then `facet_update_data` in place).
#[tokio::test]
async fn index_serves_the_facet_wrapped_loader_not_a_drawing() {
let bundle = stub_bundle("loader");
let resp = get(router(demo(), None, Duration::ZERO, bundle.clone()), "/").await;
assert_eq!(resp.status(), StatusCode::SEE_OTHER);
let location = resp.headers()[header::LOCATION].to_str().unwrap();
assert_eq!(location, loader_url());
assert!(location.contains("component=pipeline") && location.contains("events=/events"));
let resp = get(router(demo(), None, Duration::ZERO, bundle.clone()), "/index.html").await;
assert_eq!(resp.status(), StatusCode::OK);
assert!(content_type(&resp).starts_with("text/html"));
let body = body_string(resp).await;
for forbidden in ["<svg", "createElementNS", "http://www.w3.org/2000/svg", "<script src"] {
assert!(!body.contains(forbidden), "page draws in JavaScript: found {forbidden:?}");
}
for required in [
"./facet_wrapped.js",
"EventSource",
"facet_mount_with_data",
"facet_update_data",
"q.get(\"events\")",
] {
assert!(body.contains(required), "page is not the facet-wrapped loader: missing {required:?}");
}
let resp = get(router(demo(), None, Duration::ZERO, bundle.clone()), "/facet_wrapped.js").await;
assert_eq!(resp.status(), StatusCode::OK);
assert!(content_type(&resp).starts_with("text/javascript"));
let resp = get(router(demo(), None, Duration::ZERO, bundle), "/facet_wrapped_bg.wasm").await;
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(content_type(&resp), "application/wasm");
let wasm = resp.into_body().collect().await.unwrap().to_bytes();
assert!(wasm.starts_with(&WASM_MAGIC));
}
/// No bundle → the page routes answer 503 with every missing file by name, the
/// directory looked in and the build recipe; the data routes keep working.
#[tokio::test]
async fn missing_bundle_refuses_loudly_and_keeps_the_data_routes() {
let dir = std::env::temp_dir().join("knut-web-bundle-does-not-exist");
let bundle = load_bundle(&dir);
let e = bundle.as_ref().unwrap_err().to_string();
for f in BUNDLE_FILES {
assert!(e.contains(f), "refusal does not name {f}: {e}");
}
assert!(e.contains(&dir.display().to_string()));
assert!(e.contains("cargo xtask wrap-wasm"));
for uri in ["/", "/index.html", "/facet_wrapped.js", "/facet_wrapped_bg.wasm"] {
let resp = get(router(demo(), None, Duration::ZERO, bundle.clone()), uri).await;
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE, "{uri}");
assert!(content_type(&resp).starts_with("text/plain"));
let body = body_string(resp).await;
assert!(body.contains("facet_wrapped_bg.wasm") && body.contains("wrap-wasm"), "{uri}: {body}");
}
for uri in ["/graph.json", "/pipeline.json", "/events"] {
let resp = get(router(demo(), None, Duration::ZERO, bundle.clone()), uri).await;
assert_eq!(resp.status(), StatusCode::OK, "{uri}");
}
}
/// A bundle whose `.wasm` is not a WebAssembly module is refused by name.
#[test]
fn bundle_load_rejects_bytes_that_are_not_wasm() {
let dir = stub_bundle_dir("notwasm", b"<!doctype html>");
let e = load_bundle(&dir).unwrap_err().to_string();
assert!(e.contains("facet_wrapped_bg.wasm: not a WebAssembly module"), "{e}");
// The loader page itself was fine.
assert!(!e.contains("index.html:"), "{e}");
}
/// `FACET_WRAPPED_DIST` unset is a refusal that names the variable.
#[test]
fn from_env_names_the_variable_when_unset() {
// Only meaningful when the variable is not set in this process; when a
// developer runs with it set, the loaded bundle is the answer instead.
match bundle_from_env() {
Err(e) => assert!(e.to_string().contains(BUNDLE_DIR_ENV)),
Ok(b) => assert!(b.dir().join("facet_wrapped_bg.wasm").exists()),
}
}
/// Offline coverage for the live gRPC → SSE bridge: drive [`drive_updates`]
/// (the translation core of `drive_live`) with a canned update stream and
/// assert the emitted frame sequence + terminal error/done frames. No tonic
/// channel, no network.
#[cfg(feature = "grpc")]
#[tokio::test]
async fn drive_live_translates_event_stream_to_frames() {
use crate::model::PipelineRunEvent;
use crate::progress::ViewerUpdate;
let (tx, body) = knut_wasm_host::sse_channel();
let mut sink = WebEventSink::new(demo(), tx);
// Two good events (a flow start, then a completion that writes rows), a
// failure, then a trailing event that must NOT be applied — the loop
// stops on the first error.
let updates: Vec<Result<ViewerUpdate, String>> = vec![
Ok(ViewerUpdate::Event(PipelineRunEvent::info(
"Flow 'ingest' is STARTING.",
))),
Ok(ViewerUpdate::Event(PipelineRunEvent::info(
"Flow 'ingest' has COMPLETED and wrote 500 rows.",
))),
Err("stream broke".to_string()),
Ok(ViewerUpdate::Event(PipelineRunEvent::info(
"Flow 'clean' is STARTING.",
))),
];
drive_updates(&mut sink, futures_util::stream::iter(updates)).await;
// run_source always sends a terminal done frame after the live loop.
let _ = sink.on_finish();
drop(sink); // close the channel so the body ends
let text = String::from_utf8(body.collect().await.unwrap().to_bytes().to_vec()).unwrap();
let frames = sse_frames(&text);
// 2 frames (one per applied Ok) + 1 error frame + 1 done frame; the
// post-error Ok is dropped.
assert_eq!(frames.len(), 4, "frames: {frames:?}");
// The first two frames are models on the wire, not control frames.
for f in &frames[..2] {
assert_wire_shape(f);
}
// The COMPLETED event was folded into the shared ViewerState: the target
// dataset flips to `done` and its row count rides the stage metrics.
let snap = &frames[1];
assert_eq!(stage(snap, "bronze")["status"], "done");
assert_eq!(stage(snap, "bronze")["metrics"]["rows"], 500);
// Terminal frames: the error (from the failed item), then the done marker.
assert_eq!(frames[2]["error"], "stream broke");
assert_eq!(frames[3], serde_json::json!({ "done": true }));
}
#[test]
fn synthetic_run_covers_every_flow_start_and_finish() {
let evs = synthetic_run(&demo());
// 3 flows × (start + complete).
assert_eq!(evs.len(), 6);
assert!(
evs.iter()
.any(|e| e.message.contains("'rollup'") && e.message.contains("COMPLETED"))
);
}
}