//! SSE event bus + state snapshot for the dashboard. //! //! One multiplexed `broadcast` channel; `/events` forwards every event and //! the UI routes by `type`. Emits `state` (supervisor + job finishes), //! `job` (registry transitions), and `log` (attached window lines). use crate::config::Config; use crate::web::probe::ContainerSnap; use serde::Serialize; use std::time::{SystemTime, UNIX_EPOCH}; /// One SSE event (serialized as `{"type":"state"|"job", ...}` on the wire). #[derive(Serialize, Clone, Debug)] #[serde(tag = "type", rename_all = "lowercase")] pub enum Event { State { ts: u64, snapshot: Snapshot, }, Job { id: String, /// "serve" | "stop" | "power" kind: &'static str, event: String, cluster: String, recipe: Option, message: String, }, Log { /// Attach id (matches `POST /api/logs` response). attach: String, cluster: String, node: String, recipe: String, rank: u32, line: String, }, } /// Per-node view for the LED grid. #[derive(Serialize, Clone, Debug)] pub struct NodeSnap { pub name: String, pub ssh_ip: String, pub reachable: bool, pub occupied: bool, pub containers: Vec, /// Node rank — log windows attach the rank's `docker logs` stream. pub rank: u32, } /// Per-cluster view. #[derive(Serialize, Clone, Debug)] pub struct ClusterSnap { pub name: String, pub nodes: Vec, /// All nodes reachable AND blocking HA switches on (cluster verdict). pub up: bool, /// A blocking `extra_entity` reports anything but "on". pub switch_off: bool, /// Non-blocking entities that are off (display only, e.g. a fan). pub soft_off: Vec, /// Free nodes; 0 when the cluster is down. pub free: u32, } /// One log window attach (for state snapshots — the UI restores open /// log windows from `/api/state` after a reload). #[derive(Serialize, Clone, Debug)] pub struct AttachSnap { pub id: String, pub cluster: String, pub node: String, pub recipe: String, pub rank: u32, /// Stream ended on its own (container stopped, SSH drop). pub done: bool, } /// The full dashboard snapshot (also the `GET /api/state` body). #[derive(Serialize, Clone, Debug)] pub struct Snapshot { pub ts: u64, pub clusters: Vec, pub jobs: Vec, pub attaches: Vec, } fn now_ts() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_secs()) .unwrap_or(0) } /// Pure snapshot builder — unit-testable without SSH. /// /// Cluster-level semantics via [`crate::cluster_state::verdict`]: one /// offline node or a switch off takes the whole cluster down, and a down /// cluster reports no containers (even on nodes that are alive). pub fn build_snapshot( config: &Config, fleet: &crate::web::probe::Fleet, jobs: &[crate::web::jobs::Job], attaches: &[AttachSnap], ) -> Snapshot { let nodes = &fleet.nodes; let switch_off = &fleet.switch_off; let soft_off = &fleet.soft_off; let mut clusters = Vec::with_capacity(config.clusters.len()); for c in &config.clusters { let mut node_snaps = Vec::with_capacity(c.nodes.len()); let mut reachables: Vec<(bool, bool)> = Vec::with_capacity(c.nodes.len()); for n in &c.nodes { let s = nodes.get(&n.name); let reachable = s.is_some_and(|x| x.reachable); let occupied = s.is_some_and(|x| x.occupied); reachables.push((reachable, occupied)); } let so = switch_off.get(&c.name).copied().unwrap_or(false); let v = crate::cluster_state::verdict(&reachables, so); for n in &c.nodes { let s = nodes.get(&n.name); let reachable = s.is_some_and(|x| x.reachable); let occupied = s.is_some_and(|x| x.occupied); // A down cluster reports no containers — the cluster is off as // a whole, whatever individual nodes still run. let containers = if v.up { s.map(|x| x.containers.clone()).unwrap_or_default() } else { Vec::new() }; node_snaps.push(NodeSnap { name: n.name.clone(), ssh_ip: n.ssh_ip.clone(), reachable, occupied, containers, rank: n.rank, }); } clusters.push(ClusterSnap { name: c.name.clone(), nodes: node_snaps, up: v.up, switch_off: so, soft_off: soft_off.get(&c.name).cloned().unwrap_or_default(), free: v.free, }); } Snapshot { ts: now_ts(), clusters, jobs: jobs.to_vec(), attaches: attaches.to_vec(), } } #[cfg(test)] mod tests { use super::*; use std::collections::HashMap; fn config() -> Config { Config { ssh_user: "u".into(), ssh_key_path: "/k".into(), clusters: vec![ crate::config::Cluster { name: "a".into(), master_ib_ip: None, nodes: vec![ crate::config::Node { name: "n0".into(), ssh_ip: "10.0.0.1".into(), ib_ip: None, rank: 0, host_key: None, entity: None, }, crate::config::Node { name: "n1".into(), ssh_ip: "10.0.0.2".into(), ib_ip: None, rank: 1, host_key: None, entity: None, }, ], extra_entities: vec![], }, crate::config::Cluster { name: "b".into(), master_ib_ip: None, nodes: vec![crate::config::Node { name: "n2".into(), ssh_ip: "10.0.0.3".into(), ib_ip: None, rank: 0, host_key: None, entity: None, }], extra_entities: vec![], }, ], nccl: std::collections::HashMap::new(), hf_cache_dir: None, init_port: 25000, before_serve: std::collections::HashMap::new(), exec_aliases: std::collections::HashMap::new(), home_assistant: None, global_mods_dir: None, drop_caches: true, mount_aliases: std::collections::HashMap::new(), recipe_sources: Vec::new(), cluster_sources: Vec::new(), } } #[test] fn snapshot_reflects_probe_state() { let config = config(); let mut nodes = HashMap::new(); nodes.insert( "n0".into(), crate::web::probe::WebNodeState { reachable: true, occupied: true, containers: vec![ContainerSnap { name: "sm121_r_0".into(), status: "Up 3 hours".into(), cluster: "a".into(), recipe: "r".into(), }], }, ); nodes.insert( "n1".into(), crate::web::probe::WebNodeState { reachable: true, occupied: false, containers: vec![], }, ); // n2 absent → unreachable. let fleet = crate::web::probe::Fleet { nodes, switch_off: std::collections::HashMap::new(), soft_off: std::collections::HashMap::new(), }; let snap = build_snapshot(&config, &fleet, &[], &[]); assert_eq!(snap.clusters.len(), 2); let a = &snap.clusters[0]; assert!(a.up); assert!(a.nodes[0].reachable && a.nodes[0].occupied); assert_eq!(a.nodes[0].containers[0].recipe, "r"); assert!(a.nodes[1].reachable && !a.nodes[1].occupied); assert_eq!(a.free, 1); // cluster b has one offline node → down as a whole let b = &snap.clusters[1]; assert!(!b.up); assert!(!b.nodes[0].reachable); assert_eq!(b.free, 0); assert!(snap.jobs.is_empty()); } #[test] fn down_cluster_hides_containers_on_live_nodes() { // Both nodes of cluster a are live; n1 runs a container but n0 is // offline → cluster a is down as a whole and the live node's // container must not be reported. let config = config(); let mut nodes = HashMap::new(); nodes.insert("n0".into(), crate::web::probe::WebNodeState::default()); nodes.insert( "n1".into(), crate::web::probe::WebNodeState { reachable: true, occupied: true, containers: vec![ContainerSnap { name: "c2".into(), status: "Up".into(), cluster: "a".into(), recipe: "r".into(), }], }, ); let fleet = crate::web::probe::Fleet { nodes, switch_off: std::collections::HashMap::new(), soft_off: std::collections::HashMap::new(), }; let snap = build_snapshot(&config, &fleet, &[], &[]); let a = &snap.clusters[0]; assert!(!a.up, "one offline node takes the cluster down"); assert_eq!(a.free, 0); assert!( a.nodes[1].containers.is_empty(), "containers hidden on a down cluster" ); assert!(a.nodes[1].reachable, "reachability is still shown"); } #[test] fn switch_off_takes_cluster_down() { let config = config(); let mut nodes = HashMap::new(); for n in ["n0", "n1", "n2"] { nodes.insert( n.into(), crate::web::probe::WebNodeState { reachable: true, occupied: false, containers: vec![], }, ); } let mut so = HashMap::new(); so.insert("a".into(), true); let fleet = crate::web::probe::Fleet { nodes, switch_off: so, soft_off: std::collections::HashMap::new(), }; let snap = build_snapshot(&config, &fleet, &[], &[]); assert!(!snap.clusters[0].up); assert!(snap.clusters[0].switch_off); assert_eq!(snap.clusters[0].free, 0); assert!(snap.clusters[1].up); assert_eq!(snap.clusters[1].free, 1); } #[test] fn soft_off_does_not_take_cluster_down() { // A non-blocking entity (fan) is off, all nodes live → cluster // stays up; the off entity is surfaced for display only. let config = config(); let mut nodes = HashMap::new(); for n in ["n0", "n1", "n2"] { nodes.insert( n.into(), crate::web::probe::WebNodeState { reachable: true, occupied: false, containers: vec![], }, ); } let mut soft = HashMap::new(); soft.insert("a".into(), vec!["switch.fan".to_string()]); let fleet = crate::web::probe::Fleet { nodes, switch_off: HashMap::new(), soft_off: soft, }; let snap = build_snapshot(&config, &fleet, &[], &[]); assert!( snap.clusters[0].up, "non-blocking off must not down the cluster" ); assert_eq!(snap.clusters[0].free, 2); assert_eq!(snap.clusters[0].soft_off, vec!["switch.fan".to_string()]); assert!(snap.clusters[1].soft_off.is_empty()); } #[test] fn job_event_serializes_with_type_tag() { let ev = Event::Job { id: "j1".into(), kind: "serve", event: "started".into(), cluster: "c".into(), recipe: Some("r".into()), message: "ok".into(), }; let json = serde_json::to_value(&ev).unwrap(); assert_eq!(json["type"], "job"); assert_eq!(json["kind"], "serve"); assert_eq!(json["id"], "j1"); assert_eq!(json["event"], "started"); } }