# Running a pipeline end-to-end — design
> **Class:** `design` · **Function axis:** `f0` one run, three frontends · `f1` define→run→progress → the CLI/GUI/web viewers · `f2` the `sc://` session seam · `f3` the Airflow AIP-82 asset sink · `f4` the native CDC / SCD merge operator · `f5` the skade-sourced changelog cutover.
> Strict: this file states *only the shape we run*. The crate-local *why* lives in
> [`pipelines-reasoning.md`](pipelines-reasoning.md) and [`thund-reasoning.md`](thund-reasoning.md).
>
> **Scope:** the cross-crate story added on `autonom/knut-impl` — how a Spark
> Declarative Pipeline is **defined, run, and watched** (terminal / egui / browser),
> how a session authenticates over `sc://`, how completions become Airflow asset
> events, and how a change-data-capture merge is applied natively over the
> **single-sourced** `_change_type` vocabulary. Builds on
> [`pipelines-design.md`](pipelines-design.md), [`thund-design.md`](thund-design.md)
> and [`bifrost-design.md`](bifrost-design.md).
## f0 — one run, three frontends (the whole)
A pipeline run is one merged stream folded into one `ViewerState`
(`progress.rs:240`), and **every** frontend is a pure function of that state — the
CLI text/JSON dump, the egui/snarl node graph, and now a **browser** page over
Server-Sent Events. Nothing about the run changes when you change how you watch
it; the web viewer can show nothing the CLI can't.
```
define_graph ─► start_run ─► PipelineRunEvent + ExecutionProgress
(client.rs:354) (client.rs:373) │
▼
ViewerState::apply (progress.rs:240) ◄── one truth
│ │ │
dump_state│ DagViewer│ WebEventSink│
(viewer.rs) (gui.rs) (web.rs:56)
terminal egui/snarl browser / SSE
```
Two laws from [`pipelines-design.md`](pipelines-design.md) still hold — **CLI
parity** (`show`/`dump_json` is the floor) and **one merged stream** — and the
web viewer is added *under* them, never beside them: it folds events through the
same `PipelineEventSink` seam (`web.rs:96`) into the same `ViewerState`.
## f1 — the web viewer (`src/web.rs` [feature `web`])
A third frontend: an `axum` server + SSE, serving facett's **Rust** `facett-pipeline`
viewer compiled to wasm (the `facet-wrapped` bundle) and feeding it `PipelineModel`
frames. knut draws nothing in JavaScript: the page is facet-wrapped's own loader,
served as bytes from a run-time directory (`FacetBundle`; `--facet-dist` /
`FACET_WRAPPED_DIST`), and every frame is `to_pipeline_model` encoded by the ONE
codec, `facett_pipeline::wire::to_json` (asserted by
`index_serves_the_facet_wrapped_loader_not_a_drawing` and
`events_stream_the_pipeline_model_wire_to_completion`). Before lane SW row 6
(2026-09-05) this module hand-drew the DAG in SVG/JS; that is deleted, not flagged.
| endpoint | serves | where |
|---|---|---|
| `GET /` | `303` to the loader, `/index.html?component=pipeline&events=/events` (`LOADER_URL`); `503` + the missing-file list if there is no bundle | `web.rs` `index` |
| `GET /index.html`, `/facet_wrapped.js`, `/facet_wrapped_bg.wasm` | the bundle, verbatim, typed `text/html` / `text/javascript` / `application/wasm` | `web.rs` `bundle_file` |
| `GET /graph.json` | the DAG as JSON, **reusing** `crate::dump_json` | `web.rs` `graph_json` |
| `GET /pipeline.json` | one all-idle frame on the wire | `web.rs` `pipeline_json` |
| `GET /events` | an SSE stream of frames (`wire_frame`), one per run event, ended by `{"done":true}` / `{"error":…}` | `web.rs` `events` |
**The event source is pluggable, the sink is shared.** `run_source`
(`web.rs:195`) drives one connection through a `WebEventSink` (`web.rs:56`) — a
`PipelineEventSink` that folds each event into a `ViewerState` and broadcasts the
resulting `state_json` over an mpsc channel. Two sources feed it:
- **Synthetic replay** (`connect = None`): `synthetic_run` (`web.rs:268`) walks the
DAG in topological order emitting one STARTING + one COMPLETED event per flow, so
the demo page animates **with no Spark server**. `delay` paces it
(`DEFAULT_REPLAY_DELAY = 450 ms`, `web.rs:128`; `Duration::ZERO` in tests).
- **Live run** (`connect = Some(url)`, feature `grpc`): `drive_live` (`web.rs:220`)
opens a `PipelinesClient` and folds `run_updates` through `drive_updates`
(`web.rs:248`) — the pure translation core, unit-tested offline with a canned
update stream (`drive_live_translates_event_stream_to_snapshots`).
The page (`web.rs:292`): `fetch('graph.json')` → `layout()` does longest-path
layering over `read → target` edges and draws one `g.node` per dataset; an
`EventSource('events')` recolours each node by `node_status` (`st-idle` /
`st-running` / `st-done` / `st-failed`) and updates `#meta` / `#log`. A `{"done":true}`
frame closes the stream.
### The `serve_on` bound-listener seam
`serve` (`web.rs:131`) binds an address then delegates to `serve_on`
(`web.rs:152`), which takes an **already-bound** `TcpListener`. That seam lets a
caller `bind("127.0.0.1:0")`, recover the OS-assigned port via `local_addr()`, and
*then* hand the live listener over — **no bind race**. The browser robot test uses
exactly this to point `fantoccini` at the real server (`robot_web_viewer.rs:63`).
### The browser robot test (`tests/robot_web_viewer.rs` [feature `web`, `#[ignore]`])
The browser analogue of `robot_snarl_gui.rs`: serve the **real** page on an
ephemeral port with a fast synthetic replay (`Duration::from_millis(50)`), drive a
**real** browser through `fantoccini` (WebDriver), and gate on a red/green `Row`
matrix. `dag_animates_to_all_done` (`robot_web_viewer.rs:53`) asserts the page drew
one `#nodes g.node` per dataset, every node reaches class `st-done`, `#meta` shows
"N done", and `#log` names the terminal `rollup` flow — ≥ 8 steps, zero reds. It is
`#[ignore]` because it needs a WebDriver at `127.0.0.1:4444` (chromedriver /
geckodriver), so the default build never links it and CI runs it only when a driver
is present.
## f2 — the `sc://` session seam (`src/client.rs` [feature `grpc`])
`connect` (`client.rs:206`) now **threads the Spark Connect URL's session params**
instead of dropping them. The `sc://` grammar is
`sc://host:port/;user_id=<u>;session_id=<s>;token=<t>;use_ssl=<b>;…`; `parse_sc_url`
(`client.rs:552`) splits host:port from the `;k=v` tail into an `ScConnect`
(`client.rs:539`):
| param | effect | where |
|---|---|---|
| `user_id` | sets `UserContext.user_id` on **every** request | `user_context` (`client.rs:230`), stamped in `pipeline_request`/`execute_sql` |
| `session_id` | pins the request `session_id` (else a fresh `new_uuid`) | `client.rs:214` |
| `token` | attached as `authorization: Bearer <t>` metadata via a tonic interceptor | `AuthInterceptor` (`client.rs:592`) |
| others (`use_ssl`, `user_agent`, …) | recognised-but-ignored (not session-shaping here) | `parse_sc_url` match arm |
`AuthInterceptor` (`client.rs:592`) implements `tonic::service::Interceptor`; a
`None` token is a pure pass-through so unauthenticated servers are unaffected
(`auth_interceptor_sets_bearer_only_when_token_present`). A plain `http(s)://` URL
carries no params — generated session, no auth. The client's inner type is now an
`InterceptedService<Channel, AuthInterceptor>`.
### The execute_sql progress finding
`execute_sql` (`client.rs:491`) streams `ExecutionProgress` for *any* Spark SQL
query, making the viewer a live Spark **job** monitor. The correction pinned on
this branch: **progress is server-driven, not requested.** Spark Connect pushes
`ExecutePlanResponse.execution_progress` (base.proto **field 18**) unconditionally
while a plan runs, paced by the server-side `spark.connect.progress.reportInterval`
config — there is **no per-request toggle**. `ExecutePlanRequest.request_options`
(field 5) is documented "currently unused" and only carries `ReattachOptions` /
`ResultChunkingOptions` / an extension `Any`, none of which control progress. So
the request leaves `request_options` at its default (empty) and simply consumes the
frames the server volunteers. The live integration test
`knutpipe_live_path_streams_spark_job_progress`
(`tests/integration_spark_connect.rs:83`) asserts progress snapshots actually
arrive.
## f3 — the Airflow AIP-82 asset sink (`src/airflow.rs` [feature `airflow`])
`AirflowAssetSink` (`airflow.rs:39`) is a `PipelineEventSink` that turns each SDP
output *completing* into an Airflow **asset event**, so a downstream Airflow 3 DAG
can schedule on a pipeline output instead of a cron. It **reuses** bifröst's model
— `knut_bifrost::airflow::{AssetEvent, AssetEmitter}` — and this crate's
`ViewerState` for attribution + status + row parsing; it duplicates neither.
- **Collection is synchronous** (fits the sync sink seam): `on_event`
(`airflow.rs:113`) folds the event into a `ViewerState`, then `scan_completed`
(`airflow.rs:87`) appends one `AssetEvent { uri, extra }` the first time each
dataset reaches `NodeStatus::Done` — URI `{uri_prefix}{dataset}`, `extra` carrying
`dataset` / `rows` / `output_type` / `source: "spark-sdp"`. Each output emits **at
most once** (`emitted: BTreeSet`); a **failed** output advances no state and emits
nothing (its asset is not current).
- **Emission is a separate async step**: `emit` (`airflow.rs:75`) forwards the batch
through any `AssetEmitter` (e.g. bifröst's HTTP `AirflowEmitter`) — the same
"derive events / POST them" split as bifröst's own `emit_round`. The AIP-82 wire
body is bifröst's `asset_event_body`.
## f4 — the native CDC / SCD merge operator (`knut-thund/src/backend/native.rs` [feature `native`])
Previously a `FlowKind::Cdc` flow was reported honestly as *deferred*. This branch
implements it natively over DataFusion. Dispatch is at `native.rs:476`: a CDC flow
routes to `run_cdc_flow` (`native.rs:710`), which resolves the changelog from an
upstream `reads` source and folds it into the target as an **idempotent, key-matched
merge** ordered by `cdc.sequence_by`.
The IR (`knut-thund/src/ir/flow.rs`): `CdcSpec { keys, sequence_by, apply_as_deletes,
scd_type }` (`flow.rs:30`); `ScdType::{Type1, Type2}` (`flow.rs:18`); `FlowKind::Cdc
{ cdc }` (`flow.rs:66`).
**The changelog contract** — `apply_cdc_merge` (`native.rs:846`) — consumes skade /
knut-bifrost's shared `_change_type` vocabulary (`CHANGE_TYPE_COLUMN`,
`native.rs:699`):
| `_change_type` | meaning | merge effect |
|---|---|---|
| `INSERT` / `UPDATE_AFTER` | new / post-image | upsert (latest `sequence_by` per key wins) |
| `DELETE` | tombstone | removes the key (or the `apply_as_deletes` predicate) |
| `UPDATE_BEFORE` | pre-image of an in-place update | dropped (no net change) |
The merge is a pure SQL rewrite over a registered changelog `MemTable`, with
identifiers quoted via `quote_ident` (`native.rs:828`):
- **SCD type 1** — `ROW_NUMBER() OVER (PARTITION BY keys ORDER BY sequence_by DESC)`,
keep rank 1 that is not a delete: the target is the latest non-deleted row per key
(history overwritten). The `_change_type` column is stripped from the output so the
target renders like an ordinary table.
- **SCD type 2** — `LEAD(sequence_by) OVER (PARTITION BY keys ORDER BY sequence_by)`
derives `__start_at` / `__end_at` validity bounds (`__end_at = NULL` for the current
version); every version is kept, and a delete closes the open version without
opening a new one.
**Micro-batch vs bounded.** When the changelog source was seeded as a micro-batch
stream, `run_cdc_flow` accumulates the changelog seen so far and **re-merges each
trigger** — apply-changes is a pure function of the ordered changelog, so
recomputation is correct and deterministic. With no seeded stream it runs once over
the bounded changelog. Backend capability: only `NativeBackend` runs CDC honestly;
`SparkBackend` gates it behind `spark_4_2()` (see [`thund-design.md`](thund-design.md)).
Tests (`knut-thund/tests/streaming_microbatch.rs`):
`cdc_scd_type1_upserts_and_deletes` (`:375`), `cdc_scd_type2_keeps_version_history`
(`:405`).
## f5 — the skade-sourced changelog cutover (`knut-bifrost/src/source.rs` [feature `skade`])
Before this branch, `knut-bifrost`'s `IcebergSource` kept a **private** `DeltaPlan`
struct and re-derived the CDC changelog itself (its own INSERT/DELETE tagging,
delete-identity widening, full-reload fallback). The cutover makes **skade the
single source of truth**:
- The private `DeltaPlan` is deleted; bifröst now **re-exports** the canonical
`skade::DeltaPlan` (`source.rs:355`).
- `ingest_delta` (`source.rs:443`) calls `skade::Table::read_changelog(from, to)`
(`source.rs:473`) — skade walks the manifest lineage (`read_delta`), tags appended
rows `INSERT` and equality-deleted identities `DELETE` (widening them to the table
schema), and re-reads the whole `to` snapshot as an INSERT stream on an
overwrite/position-delete window. The `_change_type` sentinel is `skade::change_type::COLUMN`.
- **bifröst's contract is preserved** by post-processing skade's `ChangelogBatch`
(`source.rs:370`): a pure-append or full-reload window flows through **untagged**
(`strip_change_type`, `source.rs:545`) straight to the upsert pipeline; only a
window that carries equality deletes keeps skade's INSERT/DELETE tags for the
listener's changelog split. skade clears `delete_files` on a full reload, so
`plan.delete_files.is_empty()` is exactly "no CDC deletes to route" — the same
branch bifröst took before the cutover.
The upshot: the `_change_type` changelog vocabulary is now defined **once** in
skade and consumed identically by knut-bifrost (source side) and knut-thund (merge
side, f4) — no divergent private copies.
## Also on this branch
- **graphar-flight nested-Cypher Utf8 fallback** (`graphar-flight/src/convert.rs`):
a non-scalar Cypher `Value` (map / node / list) has no scalar Arrow type, so
`cell_type` falls back to `Utf8` and the cell is rendered via its `Debug` form —
stringified, never dropped. Pinned by `auto_nested_cypher_value_falls_back_to_utf8`
and `cell_type_maps_nested_values_to_utf8`.
- **xtask `build-wasm-guest`** (`xtask/src/main.rs`): a pure-Rust task that builds
the `wasm32-unknown-unknown` guest under `crates/knut-bifrost/wasm-guest/` (excluded
from the native workspace), leaving the artifact where the `#[ignore]`d
`knut-bifrost` `tests/wasm_guest.rs` expects it. Best-effort `rustup target add`,
then a `--target wasm32-unknown-unknown --release` build, then a non-empty-artifact
check. Closes TEST-GAP-2.
## Feature-flag matrix
| feature | crate | pulls in | gates |
|---|---|---|---|
| `web` | knut-pipelines | axum, tokio, tokio-stream, futures-util | the SSE web viewer (`web.rs`), `knutpipe web`, `serve_on` |
| `web` + `grpc` | knut-pipelines | + tonic 0.14 tree | `knutpipe web --connect` (live run into the page) |
| `grpc` | knut-pipelines | tonic 0.14, prost, vendored proto | `PipelinesClient`, `sc://` params, `AuthInterceptor`, `execute_sql` |
| `airflow` | knut-pipelines | knut-bifrost (default-features off) | `AirflowAssetSink` — reuses bifröst's `AssetEvent`/`AssetEmitter` |
| `gui` / `live` | knut-pipelines | egui 0.28 / + grpc | snarl node graph / the live desktop viewer |
| `native` | knut-thund | datafusion, tokio, futures | the CDC/SCD merge operator (`run_cdc_flow`, `apply_cdc_merge`) |
| `spark` | knut-thund | knut-pipelines/grpc | IR→SDP lowering; CDC only on `spark_4_2()` |
| `skade` | knut-bifrost | skade | `IcebergSource` + the re-exported `skade::DeltaPlan` / `read_changelog` |
Dev-only: `fantoccini` (browser robot, `web`), `async-trait` + a minimal `tokio`
(airflow-sink mock-emitter test). All default builds are unchanged — every surface
above is off by default.
## Architecture map
| Area | Where |
|---|---|
| Web viewer + SSE | `knut-pipelines/src/web.rs` [`web`] (`serve`, `serve_on`, `WebEventSink`, `synthetic_run`, `drive_updates`, `INDEX_HTML`) |
| Browser robot | `knut-pipelines/tests/robot_web_viewer.rs` [`web`, `#[ignore]`] (`dag_animates_to_all_done`) |
| `sc://` session + auth | `knut-pipelines/src/client.rs` [`grpc`] (`parse_sc_url`, `ScConnect`, `AuthInterceptor`, `user_context`) |
| Progress finding | `knut-pipelines/src/client.rs:491` (`execute_sql`) + `tests/integration_spark_connect.rs:83` |
| Airflow asset sink | `knut-pipelines/src/airflow.rs` [`airflow`] (`AirflowAssetSink`, `scan_completed`, `emit`) |
| Native CDC/SCD | `knut-thund/src/backend/native.rs` [`native`] (`run_cdc_flow`, `apply_cdc_merge`, `CHANGE_TYPE_COLUMN`) |
| CDC IR | `knut-thund/src/ir/flow.rs` (`CdcSpec`, `ScdType`, `FlowKind::Cdc`) |
| Changelog cutover | `knut-bifrost/src/source.rs` [`skade`] (`read_changelog`, re-exported `skade::DeltaPlan`, `strip_change_type`) |
| Cypher Utf8 fallback | `graphar-flight/src/convert.rs` (`cell_type`) |
| wasm-guest build | `xtask/src/main.rs` (`build_wasm_guest`) |