facett-demo/src/synthetic_sdp.rs
//! **Synthetic SDP / Spark Pipelines demo** — ~100 fun-named pipelines streaming
//! between each other in a multi-layer DAG. Push data in at the **sources** →
//! watch a wave of "data flowing" sweep layer-by-layer → out the **sinks**, a
//! beautiful animated pattern (PLAN §G synthetic SDP demo).
//!
//! This is a **driver around the already-implemented [`PipelineView`]** — we do
//! NOT reimplement any rendering. The module's whole job is:
//!
//! 1. **[`SyntheticDag`]** — a deterministic generator: ~100 stages with whimsical
//! Norse/mythology dataset names (`bifrost_raw`, `mimir_dedup`, `huginn_join`,
//! `valhalla_agg`, …), arranged as a layered DAG (sources → several transform
//! layers → sinks) with realistic fan-in / fan-out edges. Names + topology are
//! derived from indices — **no RNG, no `Date::now`** — so snapshots/tests are
//! reproducible.
//! 2. A **time-driven wave** ([`SyntheticDag::statuses_at`]): a clock value drives a
//! front that propagates `Running → Done` down the layers and loops, so it reads
//! as data pushed in at the sources flowing out the sinks.
//! 3. **[`SyntheticSdpView`]** — a [`Facet`] that, each frame, folds the current
//! statuses + [`JobProgress`] + synthetic per-edge throughput labels into a
//! fresh [`PipelineView`] (animation on) and delegates `ui`/`caps`. Its own
//! [`state_json`](Facet::state_json) surfaces the wave (tick, running/done/failed
//! counts, fraction, node/edge totals) so a headless robot test asserts the
//! picture without pixels (facett emit LAW).
use facett::core::FacetCaps;
use facett::pipeline::{JobProgress, PipelineView, Status};
use facett::Facet;
/// How many stages in the synthetic pipeline world. ~100 fun-named pipelines.
pub const STAGES: usize = 100;
/// How many layers the DAG is split into (sources … transforms … sinks). The
/// wave propagates one layer per beat, so this is also the wave's depth.
pub const LAYERS: usize = 7;
/// Wall-clock seconds for the wave to advance **one full layer**. The whole
/// pattern loops every `LAYERS * BEAT_SECONDS` seconds.
pub const BEAT_SECONDS: f64 = 1.1;
/// Whimsical dataset/stage **name stems** — Norse myth + data-pipeline verbs.
/// The stage name is `"{stem}_{suffix}"` chosen deterministically from the index,
/// so 100 stages all read as plausible (and distinct) pipeline names.
const STEMS: &[&str] = &[
"bifrost", "mimir", "huginn", "muninn", "valhalla", "yggdrasil", "asgard", "midgard", "ragnarok",
"fenrir", "sleipnir", "gungnir", "mjolnir", "freyja", "heimdall", "loki", "odin", "thor", "baldr",
"njord", "skadi", "tyr", "vidar", "hel", "jormungandr", "norns", "valkyrie", "draupnir", "gleipnir",
"andvari", "brokkr", "eitri", "kvasir", "surtr", "ymir", "buri", "vili", "ve", "aegir",
];
/// The dataflow-stage **role suffixes** — what a Spark/SDP stage tends to do.
const SUFFIXES: &[&str] = &[
"raw", "ingest", "dedup", "clean", "join", "enrich", "agg", "window", "filter", "pivot",
"rollup", "score", "rank", "sink", "mart", "stream", "cdc", "merge", "explode", "flatten",
];
/// A deterministic, layered dataflow DAG of fun-named pipeline stages — the
/// synthetic SDP world. Built once; the per-frame status wave is a pure function
/// of a clock value over this fixed topology.
#[derive(Clone)]
pub struct SyntheticDag {
/// Stage names, index = node id.
pub names: Vec<String>,
/// The layer (0 = sources … `LAYERS-1` = sinks) each stage lives in.
pub layer_of: Vec<usize>,
/// Edges in [`PipelineView`] convention: `(reads_idx, target_idx)` — stage
/// `reads_idx` reads from `target_idx`, dataflow paints `target → reads`.
pub edges: Vec<(usize, usize)>,
}
impl Default for SyntheticDag {
fn default() -> Self {
Self::new(STAGES, LAYERS)
}
}
impl SyntheticDag {
/// Generate `n` stages across `layers` layers, deriving names + a fan-in /
/// fan-out topology purely from indices (deterministic — no RNG). Each
/// non-source stage reads from a small, deterministic set of stages in the
/// previous layer, giving realistic fan-in; popular upstreams get fan-out.
pub fn new(n: usize, layers: usize) -> Self {
let n = n.max(2);
let layers = layers.clamp(2, n);
// ── names: "{stem}_{suffix}", de-collided by index so all `n` are unique ──
let names: Vec<String> = (0..n)
.map(|i| {
let stem = STEMS[i % STEMS.len()];
let suffix = SUFFIXES[(i / STEMS.len() + i * 7) % SUFFIXES.len()];
// The numeric tail guarantees uniqueness once we wrap the stems.
if i < STEMS.len() {
format!("{stem}_{suffix}")
} else {
format!("{stem}_{suffix}{}", i / STEMS.len())
}
})
.collect();
// ── layer assignment: contiguous bands, sources first, sinks last. We
// bias the band sizes so the middle (transform) layers are widest — a
// realistic "fan out in the middle" silhouette. ──────────────────────
let layer_of: Vec<usize> = (0..n).map(|i| (i * layers) / n).collect();
// Index ranges per layer (sorted because layer_of is monotone in i).
let mut layer_members: Vec<Vec<usize>> = vec![Vec::new(); layers];
for (i, &l) in layer_of.iter().enumerate() {
layer_members[l].push(i);
}
// ── edges: every non-source stage reads 1–3 stages from the previous
// layer, picked deterministically by index arithmetic. This yields
// fan-in (a stage with multiple upstreams) AND fan-out (a popular
// upstream feeding several downstreams). ─────────────────────────────
let mut edges: Vec<(usize, usize)> = Vec::new();
for l in 1..layers {
let prev = &layer_members[l - 1];
if prev.is_empty() {
continue;
}
for (k, &node) in layer_members[l].iter().enumerate() {
// 1..=3 upstreams, count derived from the node index (stable).
let fanin = 1 + (node % 3);
for f in 0..fanin {
// Spread picks across the previous layer so popular upstreams
// recur (fan-out) without ever pointing forward (acyclic).
let pick = prev[(k * 2 + f * 3 + node) % prev.len()];
if pick != node && !edges.contains(&(node, pick)) {
edges.push((node, pick));
}
}
}
}
SyntheticDag { names, layer_of, edges }
}
/// Number of stages.
pub fn len(&self) -> usize {
self.names.len()
}
pub fn is_empty(&self) -> bool {
self.names.is_empty()
}
/// The source stages (layer 0 — no incoming dataflow, the push-in points).
pub fn sources(&self) -> Vec<usize> {
(0..self.len()).filter(|&i| self.layer_of[i] == 0).collect()
}
/// The sink stages (the last populated layer — where data flows out).
pub fn sinks(&self) -> Vec<usize> {
let max_layer = self.layer_of.iter().copied().max().unwrap_or(0);
(0..self.len()).filter(|&i| self.layer_of[i] == max_layer).collect()
}
/// The deepest layer index that actually has members.
fn max_layer(&self) -> usize {
self.layer_of.iter().copied().max().unwrap_or(0)
}
/// **Validate the topology** — that it's an acyclic DAG (the edge convention
/// `(reads, target)` only ever points to a strictly-earlier layer, which is
/// acyclic by construction; we verify it). Returns `true` when valid.
pub fn is_acyclic(&self) -> bool {
self.edges.iter().all(|&(reads, target)| {
reads < self.len() && target < self.len() && self.layer_of[target] < self.layer_of[reads]
})
}
/// **The wave.** Status of every stage at clock `time` (seconds). A front
/// advances one layer per [`BEAT_SECONDS`]; layers strictly behind the front
/// are `Done`, the layer at the front is `Running`, layers ahead are `Idle`.
/// The front wraps (`mod (LAYERS+1)` beats) so the pattern loops — data keeps
/// being pushed in at the sources and flowing out the sinks. Pure +
/// deterministic: identical `time` ⇒ identical statuses (no RNG / no now()).
pub fn statuses_at(&self, time: f64) -> Vec<Status> {
let depth = self.max_layer() + 1; // number of populated layers
// The front position in [0, depth]; one extra beat ("drain") so the last
// layer gets to finish (all Done) before the wave restarts.
let cycle = (depth + 1) as f64;
let beat = (time / BEAT_SECONDS).rem_euclid(cycle);
let front = beat.floor() as usize; // 0..=depth
(0..self.len())
.map(|i| {
let l = self.layer_of[i];
if l < front {
Status::Done
} else if l == front {
Status::Running
} else {
Status::Idle
}
})
.collect()
}
/// The wave **front layer** at `time` (0..=max_layer; one beyond = drain).
pub fn front_at(&self, time: f64) -> usize {
let depth = self.max_layer() + 1;
let cycle = (depth + 1) as f64;
((time / BEAT_SECONDS).rem_euclid(cycle)).floor() as usize
}
/// Per-edge throughput labels at `time`: an edge is **hot** (ramping rows/s)
/// while data is flowing across it — i.e. its target is `Done`/`Running` and
/// its reader is at the front or just behind. Deterministic off the same
/// clock. Indexed like [`edges`](Self::edges) (the [`PipelineView`] contract).
pub fn throughput_at(&self, time: f64, statuses: &[Status]) -> Vec<Option<String>> {
let front = self.front_at(time);
// A sub-beat ramp 0..1 so hot edges "spin up" within a beat.
let frac = (time / BEAT_SECONDS).rem_euclid(1.0) as f32;
self.edges
.iter()
.map(|&(reads, target)| {
let reader_layer = self.layer_of[reads];
let target_live = matches!(statuses.get(target), Some(Status::Done | Status::Running));
// Edge is hot when its target has produced and its reader is the
// current front (consuming) — the moving band of activity.
if target_live && reader_layer == front {
// A stable per-edge base rate (rows/s), ramped by the sub-beat.
let base = 4_000 + (reads * 137 + target * 31) % 18_000;
let rate = (base as f32) * (0.35 + 0.65 * frac);
Some(format!("{:.1}k rows/s", rate / 1_000.0))
} else {
None
}
})
.collect()
}
/// Overall [`JobProgress`] at `time`: fraction = done / total over the current
/// wave, plus synthetic rows/elapsed derived from the same clock (so the
/// progress bar + rows/s readout animate in lock-step with the wave).
pub fn progress_at(&self, time: f64, statuses: &[Status]) -> JobProgress {
let total = self.len() as u64;
let done = statuses.iter().filter(|s| **s == Status::Done).count() as u64;
let running = statuses.iter().filter(|s| **s == Status::Running).count() as u64;
let fraction = if total == 0 { 0.0 } else { done as f32 / total as f32 };
// Synthetic, deterministic live metrics off the clock.
let elapsed = (time.rem_euclid(BEAT_SECONDS * (self.max_layer() + 2) as f64)) as f32;
let rows = done.saturating_mul(125_000) + running.saturating_mul(40_000);
JobProgress {
fraction,
tasks_done: done,
tasks_total: total,
rows,
bytes_read: rows.saturating_mul(96),
elapsed_s: elapsed,
inflight: running,
}
}
}
/// A [`Facet`] that animates the synthetic SDP world: each frame it folds the
/// wave's current statuses + progress + throughput into a fresh [`PipelineView`]
/// (animation on) and delegates the render. Reachable from the demo deck as the
/// `synthetic-sdp` tab.
pub struct SyntheticSdpView {
title: String,
dag: SyntheticDag,
/// Injected animation clock (seconds). `None` = follow egui's ambient time
/// (live UX); a test pins it via [`set_clock`](Self::set_clock) /
/// [`tick`](Self::tick) for determinism.
clock_s: Option<f64>,
/// Flow speed handed to the inner [`PipelineView`] (marching-ants cycles/sec).
speed: f32,
}
impl Default for SyntheticSdpView {
fn default() -> Self {
Self::new("synthetic-sdp")
}
}
impl SyntheticSdpView {
/// A synthetic-SDP view with the default ~100-stage world.
pub fn new(title: impl Into<String>) -> Self {
SyntheticSdpView { title: title.into(), dag: SyntheticDag::default(), clock_s: None, speed: 0.7 }
}
/// Build over a specific generated DAG (used by tests for smaller worlds).
pub fn with_dag(title: impl Into<String>, dag: SyntheticDag) -> Self {
SyntheticSdpView { title: title.into(), dag, clock_s: None, speed: 0.7 }
}
/// Read-only access to the generated DAG (for tests / hosts).
pub fn dag(&self) -> &SyntheticDag {
&self.dag
}
/// **Pin the animation clock** (seconds). Once set, the wave + the inner
/// view's flow are a pure function of this value (deterministic frame).
pub fn set_clock(&mut self, seconds: f64) {
self.clock_s = Some(seconds);
}
/// Builder form of [`set_clock`](Self::set_clock).
pub fn with_clock(mut self, seconds: f64) -> Self {
self.clock_s = Some(seconds);
self
}
/// Advance the injected clock by `dt` seconds (initialising from 0 if unset).
/// Steps the wave one frame for deterministic, headless animation.
pub fn tick(&mut self, dt: f64) {
self.clock_s = Some(self.clock_s.unwrap_or(0.0) + dt);
}
/// The clock the view will animate at this frame (injected, else 0 headless).
fn clock(&self) -> f64 {
self.clock_s.unwrap_or(0.0)
}
/// (idle, running, done, failed) at the given clock — the wave's status mix.
///
/// `Status` is `#[non_exhaustive]`, so the wildcard arm is required. The
/// synthetic wave only ever produces the four known states (it *is* the
/// ground truth — nothing here is observed over a wire, so nothing here can
/// be `Status::Unknown`); anything else is counted as neither, rather than
/// being folded into `idle` and quietly inflating the "not started" tally.
pub fn counts_at(&self, time: f64) -> (usize, usize, usize, usize) {
let st = self.dag.statuses_at(time);
let mut c = (0usize, 0usize, 0usize, 0usize);
for s in st {
match s {
Status::Idle => c.0 += 1,
Status::Running => c.1 += 1,
Status::Done => c.2 += 1,
Status::Failed => c.3 += 1,
_ => {}
}
}
c
}
/// Build the inner [`PipelineView`] for `time` — the single mapping from the
/// synthetic world to the reusable component (statuses + progress + per-edge
/// throughput + animation on). Shared by `ui` and the tests.
pub fn pipeline_at(&self, time: f64) -> PipelineView {
let statuses = self.dag.statuses_at(time);
let nodes: Vec<(String, Status)> =
self.dag.names.iter().cloned().zip(statuses.iter().copied()).collect();
let throughput = self.dag.throughput_at(time, &statuses);
let progress = self.dag.progress_at(time, &statuses);
PipelineView::new(self.title.clone(), nodes, self.dag.edges.clone())
.with_animation(true)
.with_speed(self.speed)
.with_throughput(throughput)
.with_progress(progress)
.with_clock(time)
}
}
impl Facet for SyntheticSdpView {
fn title(&self) -> &str {
&self.title
}
fn caps(&self) -> FacetCaps {
// It paints a themed, animated, navigable DAG (delegated to PipelineView)
// and tolerates host resizing — mirror the inner view's caps.
FacetCaps::NONE.themeable().resizable().selectable().navigable()
}
fn ui(&mut self, ui: &mut egui::Ui) {
facett::core::navgraph::declare_own_draw_order(ui); // atoms follow the data order (navgraph §9)
// Follow ambient wall-time for the live UX unless a clock is pinned, so the
// wave flows on its own in the binary; request_repaint keeps it animating.
let time = self.clock_s.unwrap_or_else(|| ui.input(|i| i.time));
let mut view = self.pipeline_at(time);
view.ui(ui);
if self.clock_s.is_none() {
ui.ctx().request_repaint();
}
// ── render-lane emit: this Facet::ui path RAN ─────────────────────────
#[cfg(feature = "testmatrix")]
{
let (idle, running, done, failed) = self.counts_at(time);
facett::core::testmatrix::emit(
"facett-demo::SyntheticSdpView::ui",
"ui_render",
// OK = every stage has exactly one status at this clock: the four
// status buckets partition the world's node count (rendered states
// == input nodes — no stage dropped or double-counted).
idle + running + done + failed == self.dag.len(),
&format!(
"nodes={} edges={} idle={} running={} done={} failed={}",
self.dag.len(),
self.dag.edges.len(),
idle,
running,
done,
failed,
),
);
}
}
fn state_json(&self) -> serde_json::Value {
// Emit what we render at the current clock (facett emit LAW): the wave's
// tick/front, the status mix, the fraction, and the world's totals — so a
// headless robot test asserts the animated picture without pixels.
let time = self.clock();
let (idle, running, done, failed) = self.counts_at(time);
let total = self.dag.len();
let fraction = if total == 0 { 0.0 } else { done as f32 / total as f32 };
let front = self.dag.front_at(time);
let tick = (time / BEAT_SECONDS) as u64;
// Fold in the inner PipelineView's own state for the full picture.
let inner = self.pipeline_at(time).state_json();
serde_json::json!({
"title": self.title,
"clock_s": self.clock_s,
"tick": tick,
"front_layer": front,
"layers": self.dag.max_layer() + 1,
"node_count": total,
"edge_count": self.dag.edges.len(),
"source_count": self.dag.sources().len(),
"sink_count": self.dag.sinks().len(),
"acyclic": self.dag.is_acyclic(),
"fraction": fraction,
"statuses": { "idle": idle, "running": running, "done": done, "failed": failed },
"animated": true,
"speed": self.speed,
// The reusable PipelineView's own readable state (statuses/progress/dag),
// so the matrix sees the component that actually paints the picture.
"pipeline": inner,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
/// LAW (DAG validity): the generated world has ~100 fun-named stages, is a
/// valid acyclic DAG, and has ≥1 source and ≥1 sink. Inject a size, assert the
/// real topology — names unique, every edge points strictly upstream.
#[test]
fn synthetic_dag_is_a_valid_layered_dag() {
let dag = SyntheticDag::default();
assert_eq!(dag.len(), STAGES, "the synthetic world has ~100 stages");
assert!(dag.is_acyclic(), "the topology is acyclic (edges point strictly upstream)");
assert!(!dag.sources().is_empty(), "there is at least one source");
assert!(!dag.sinks().is_empty(), "there is at least one sink");
// Sources and sinks are disjoint (a real multi-layer pipeline).
let srcs: std::collections::HashSet<_> = dag.sources().into_iter().collect();
let snks: std::collections::HashSet<_> = dag.sinks().into_iter().collect();
assert!(srcs.is_disjoint(&snks), "sources and sinks are different stages");
// Fun names, all distinct.
let uniq: std::collections::HashSet<_> = dag.names.iter().collect();
assert_eq!(uniq.len(), dag.names.len(), "every stage name is unique");
assert!(dag.names.iter().any(|n| n.starts_with("bifrost")), "names are fun/Norse");
// Real fan-in/fan-out: more edges than nodes (the middle layers branch).
assert!(dag.edges.len() > dag.len(), "the DAG has real fan-in/fan-out ({} edges)", dag.edges.len());
}
/// LAW (wave advances + monotone): stepping the clock makes the front advance,
/// `done` grows monotonically toward total across a single sweep, and the
/// running/done mix actually changes between an early and a late tick. Real
/// input (ticks) → real asserted output (the moving wave), not "didn't panic".
#[test]
fn wave_advances_and_done_grows_monotonically() {
let dag = SyntheticDag::default();
let total = dag.len();
let depth = *dag.layer_of.iter().max().unwrap() + 1;
// Sample the wave across one sweep (front 0..=depth), one beat apart.
let mut prev_done = 0usize;
let mut dones = Vec::new();
for beat in 0..=depth {
let time = beat as f64 * BEAT_SECONDS + 0.01; // just inside the beat
let st = dag.statuses_at(time);
let done = st.iter().filter(|s| **s == Status::Done).count();
assert!(done >= prev_done, "done is monotone non-decreasing across the sweep ({done} >= {prev_done})");
prev_done = done;
dones.push(done);
}
// At the start almost nothing is done; by the drain beat everything is.
assert_eq!(*dones.first().unwrap(), 0, "no stage is done at the very first beat");
assert_eq!(*dones.last().unwrap(), total, "every stage is done by the drain beat");
// The status MIX changes between an early and a late tick (the wave moves).
let early = dag.statuses_at(0.5 * BEAT_SECONDS);
let late = dag.statuses_at((depth as f64 - 0.5) * BEAT_SECONDS);
let early_done = early.iter().filter(|s| **s == Status::Done).count();
let late_done = late.iter().filter(|s| **s == Status::Done).count();
assert!(late_done > early_done, "more is done late than early ({late_done} > {early_done})");
let early_run = early.iter().filter(|s| **s == Status::Running).count();
assert!(early_run > 0, "the source layer is running early in the sweep");
}
/// LAW (determinism): identical tick ⇒ identical state (no RNG / no now() in
/// the hot path), and the wave loops (front wraps back to 0).
#[test]
fn wave_is_deterministic_and_loops() {
let dag = SyntheticDag::default();
let t = 3.3 * BEAT_SECONDS;
// Status has no Debug; compare via the stable label projection.
let labels = |st: &[Status]| st.iter().map(|s| s.label()).collect::<Vec<_>>();
assert_eq!(labels(&dag.statuses_at(t)), labels(&dag.statuses_at(t)), "same clock ⇒ same statuses");
assert_eq!(dag.throughput_at(t, &dag.statuses_at(t)), dag.throughput_at(t, &dag.statuses_at(t)));
// Loops: one full cycle later the front returns to the same place.
let cycle = (dag.layer_of.iter().max().unwrap() + 2) as f64 * BEAT_SECONDS;
assert_eq!(dag.front_at(0.4 * BEAT_SECONDS), dag.front_at(0.4 * BEAT_SECONDS + cycle), "the wave loops");
}
/// LAW (state_json reflects the picture): the view's `state_json` carries the
/// right node + edge counts, the live status mix, and an advancing tick — and
/// folds in the inner PipelineView's own state (the component that paints it).
#[test]
fn view_state_json_reflects_counts_and_wave() {
let mut view = SyntheticSdpView::new("synthetic-sdp");
view.set_clock(0.01); // just inside beat 0 — sources running, nothing done
let s0 = view.state_json();
assert_eq!(s0["node_count"], STAGES);
assert_eq!(s0["edge_count"].as_u64().unwrap(), view.dag().edges.len() as u64);
assert_eq!(s0["acyclic"], true);
assert!(s0["source_count"].as_u64().unwrap() >= 1);
assert!(s0["sink_count"].as_u64().unwrap() >= 1);
assert_eq!(s0["statuses"]["done"], 0, "nothing done at the first beat");
assert!(s0["statuses"]["running"].as_u64().unwrap() > 0, "sources running at the first beat");
// The inner PipelineView's own state is folded in with matching totals.
assert_eq!(s0["pipeline"]["node_count"], STAGES);
assert_eq!(s0["pipeline"]["edge_count"].as_u64().unwrap(), view.dag().edges.len() as u64);
assert_eq!(s0["pipeline"]["animated"], true);
// Step the clock to late in the sweep → the wave has moved (more done, the
// fraction climbed, the tick advanced).
let depth = *view.dag().layer_of.iter().max().unwrap() + 1;
view.set_clock((depth as f64 - 0.5) * BEAT_SECONDS);
let s1 = view.state_json();
assert!(
s1["statuses"]["done"].as_u64().unwrap() > s0["statuses"]["done"].as_u64().unwrap(),
"more stages are done later in the sweep"
);
assert!(s1["fraction"].as_f64().unwrap() > s0["fraction"].as_f64().unwrap(), "the fraction climbs");
assert!(s1["tick"].as_u64().unwrap() > s0["tick"].as_u64().unwrap(), "the tick advanced");
}
/// LAW (tick stepping): `tick(dt)` steps the animation deterministically and
/// the throughput labels light up while edges are hot (the flow is visible).
#[test]
fn tick_steps_animation_and_throughput_lights_up() {
let mut view = SyntheticSdpView::new("synthetic-sdp");
view.set_clock(0.0);
let mut hot_seen = false;
let mut last_done = 0u64;
for _ in 0..8 {
view.tick(BEAT_SECONDS); // step one beat
let s = view.state_json();
let done = s["statuses"]["done"].as_u64().unwrap();
assert!(done >= last_done || done == 0, "done advances or wraps at loop");
last_done = done;
// Some edge carries a throughput label while data is flowing.
let labels = s["pipeline"]["nodes"].as_array().unwrap();
assert_eq!(labels.len(), STAGES);
let t = view.dag().throughput_at(view.clock_s.unwrap(), &view.dag().statuses_at(view.clock_s.unwrap()));
if t.iter().any(|l| l.is_some()) {
hot_seen = true;
}
}
assert!(hot_seen, "at least one edge was hot (carried a rows/s label) during the sweep");
}
}