diff --git a/Cargo.lock b/Cargo.lock index 2afcb6f..cc95dcf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1126,10 +1126,19 @@ dependencies = [ name = "maestro-tui" version = "0.1.0" dependencies = [ + "axum", + "chrono", "clap", "crossterm", + "futures-util", "maestro-config", + "maestro-protocol", + "serde_json", "thiserror 1.0.69", + "tokio", + "tokio-tungstenite 0.21.0", + "tracing", + "uuid", ] [[package]] diff --git a/crates/maestro-tui/Cargo.toml b/crates/maestro-tui/Cargo.toml index 13e091c..ab8cc0f 100644 --- a/crates/maestro-tui/Cargo.toml +++ b/crates/maestro-tui/Cargo.toml @@ -10,5 +10,17 @@ description = "Terminal UI client over the local maestro daemon's `/api/*` surfa [dependencies] clap.workspace = true crossterm.workspace = true +futures-util.workspace = true maestro-config.workspace = true +maestro-protocol.workspace = true +serde_json.workspace = true thiserror.workspace = true +tokio = { workspace = true, features = ["sync", "time", "macros", "rt"] } +tokio-tungstenite.workspace = true +tracing.workspace = true +uuid.workspace = true + +[dev-dependencies] +axum.workspace = true +chrono.workspace = true +tokio = { workspace = true, features = ["sync", "time", "macros", "rt-multi-thread", "net"] } diff --git a/crates/maestro-tui/src/catalog.rs b/crates/maestro-tui/src/catalog.rs new file mode 100644 index 0000000..c8a17d1 --- /dev/null +++ b/crates/maestro-tui/src/catalog.rs @@ -0,0 +1,378 @@ +//! `CatalogStore` — the TUI's view of the daemons-and-sessions tree. +//! +//! Backed entirely by the daemon's `WS /api/events` stream: +//! +//! 1. The first frame after connect is a [`RegistryEvent::Snapshot`] — that's +//! the bootstrap. No separate REST round-trip needed. +//! 2. Subsequent frames are deltas (`DaemonAdded`, `SessionAdded`, …) that we +//! apply atomically and bump a monotonic version counter. +//! 3. On WS disconnect: exponential backoff (1s → 30s, ±25% jitter); the next +//! connect resends a fresh `Snapshot` so we converge without needing +//! server-side replay. +//! 4. `Lagged { skipped }` from the broadcast queue is logged at warn level +//! and ignored — the server will re-snapshot on its own bookkeeping; for +//! safety we trigger a self-reconnect so the next Snapshot reconciles us. +//! +//! Cloneable and `Send + Sync` — pass it to the render loop and to a Tokio +//! task that drives [`CatalogStore::run_events_loop`]. + +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; +use std::time::Duration; + +use futures_util::StreamExt; +use maestro_protocol::{DaemonInfo, RegistryEvent, Session}; +use tokio::sync::Notify; +use tokio_tungstenite::tungstenite::Message; +use uuid::Uuid; + +/// Snapshot returned by [`CatalogStore::snapshot`]. `version` is monotone — a +/// renderer can cheap-compare it across frames to know whether re-render is +/// needed. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct CatalogSnapshot { + pub daemons: Vec, + pub sessions: Vec, + pub version: u64, +} + +#[derive(Debug, Default, Clone)] +pub struct CatalogStore { + inner: Arc, +} + +#[derive(Debug, Default)] +struct Inner { + state: RwLock, + notify: Notify, +} + +#[derive(Debug, Default)] +struct State { + daemons: HashMap, + sessions: HashMap, + /// Bumped on every applied event. Wraps at u64::MAX, which is fine + /// because comparisons are equality, not ordering. + version: u64, + /// True once we've applied at least one Snapshot. Used by + /// [`CatalogStore::wait_for_initial_snapshot`]. + bootstrapped: bool, +} + +impl CatalogStore { + /// Apply one event atomically. Used by tests directly and by the WS loop. + pub fn apply_event(&self, ev: RegistryEvent) { + let mut s = self.inner.state.write().expect("catalog rwlock poisoned"); + match ev { + RegistryEvent::Snapshot { daemons, sessions } => { + s.daemons = daemons + .into_iter() + .map(|d| (d.daemon_id.clone(), d)) + .collect(); + s.sessions = sessions.into_iter().map(|x| (x.id, x)).collect(); + s.bootstrapped = true; + } + RegistryEvent::DaemonAdded { daemon } => { + s.daemons.insert(daemon.daemon_id.clone(), daemon); + } + RegistryEvent::DaemonRemoved { daemon_id } => { + s.daemons.remove(&daemon_id); + // Drop sessions owned by the gone daemon — server will + // re-add them if it reappears. + s.sessions.retain(|_, sess| sess.daemon_id != daemon_id); + } + RegistryEvent::SessionAdded { session } | RegistryEvent::SessionUpdated { session } => { + s.sessions.insert(session.id, session); + } + RegistryEvent::SessionRemoved { daemon_id: _, id } => { + s.sessions.remove(&id); + } + RegistryEvent::Lagged { skipped } => { + tracing::warn!( + skipped, + "catalog events stream lagged; resync via reconnect" + ); + // We don't reset state here — the WS loop notices the warn and + // can choose to drop+reconnect. Keep state until that happens. + } + } + s.version = s.version.wrapping_add(1); + drop(s); + self.inner.notify.notify_waiters(); + } + + /// Cheap O(daemons + sessions) read of current state. + pub fn snapshot(&self) -> CatalogSnapshot { + let s = self.inner.state.read().expect("catalog rwlock poisoned"); + let mut daemons: Vec = s.daemons.values().cloned().collect(); + daemons.sort_by(|a, b| a.daemon_id.cmp(&b.daemon_id)); + let mut sessions: Vec = s.sessions.values().cloned().collect(); + sessions.sort_by_key(|x| x.id); + CatalogSnapshot { + daemons, + sessions, + version: s.version, + } + } + + /// Returns true once we've applied at least one Snapshot frame. + pub fn is_bootstrapped(&self) -> bool { + self.inner + .state + .read() + .expect("catalog rwlock poisoned") + .bootstrapped + } + + /// Wait until the catalog has been changed by something — applied event + /// or reconnect Snapshot. The render loop awaits this between frames. + pub async fn changed(&self) { + self.inner.notify.notified().await; + } + + /// Wait until the first Snapshot has landed. Resolves immediately if + /// already bootstrapped. + pub async fn wait_for_initial_snapshot(&self) { + loop { + if self.is_bootstrapped() { + return; + } + self.inner.notify.notified().await; + } + } + + /// Long-running task: connect to `ws_url`, parse `RegistryEvent` frames, + /// apply them. Reconnects with exponential backoff on disconnect. Never + /// returns — spawn it on a Tokio task. + pub async fn run_events_loop(self, ws_url: String) { + let mut backoff = Duration::from_secs(1); + loop { + match tokio_tungstenite::connect_async(&ws_url).await { + Ok((ws, _resp)) => { + backoff = Duration::from_secs(1); + let (_sink, mut stream) = ws.split(); + while let Some(frame) = stream.next().await { + match frame { + Ok(Message::Text(text)) => { + match serde_json::from_str::(&text) { + Ok(ev) => self.apply_event(ev), + Err(e) => { + tracing::warn!(error = ?e, "catalog: bad event frame"); + } + } + } + Ok(Message::Close(_)) | Err(_) => break, + _ => {} + } + } + } + Err(e) => { + tracing::debug!(error = ?e, "catalog: connect failed"); + } + } + tracing::debug!(?backoff, "catalog: sleeping before reconnect"); + tokio::time::sleep(jitter(backoff)).await; + backoff = std::cmp::min(backoff.saturating_mul(2), Duration::from_secs(30)); + } + } +} + +/// ±25 % jitter on the backoff. Cheap LCG seeded from the duration itself — +/// avoids touching `rand` for one knob. +fn jitter(base: Duration) -> Duration { + let ms = base.as_millis() as u64; + // Tiny pseudo-random in [0, 1) from system time nanos. + let now_nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.subsec_nanos()) + .unwrap_or(0) as u64; + // ±25 %: scale by [-250, 250] of one quarter ms per ms. + let delta = (now_nanos % 500) as i64 - 250; + let scaled = (ms as i64).saturating_mul(delta).div_euclid(1000); + let total = ms.saturating_add_signed(scaled); + Duration::from_millis(total) +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::Utc; + use maestro_protocol::{DaemonInfo, Session, SessionStatus}; + + fn daemon(id: &str) -> DaemonInfo { + DaemonInfo { + daemon_id: id.into(), + hostname: format!("{id}.lan"), + version: "test".into(), + daemon_url: format!("ws://{id}.lan:7070"), + } + } + + fn session(id: u128, daemon_id: &str, command: &str) -> Session { + Session { + id: Uuid::from_u128(id), + daemon_id: daemon_id.into(), + command: command.into(), + cwd: std::path::PathBuf::from("/tmp"), + env: Default::default(), + metadata: Default::default(), + status: SessionStatus::Running, + created_at: Utc::now(), + } + } + + #[test] + fn snapshot_event_replaces_state() { + let store = CatalogStore::default(); + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![daemon("konishi")], + sessions: vec![session(1, "konishi", "a")], + }); + let snap = store.snapshot(); + assert_eq!(snap.daemons.len(), 1); + assert_eq!(snap.sessions.len(), 1); + assert_eq!(snap.version, 1); + assert!(store.is_bootstrapped()); + + // A second snapshot wipes out the old state. + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![daemon("yamato")], + sessions: vec![], + }); + let snap = store.snapshot(); + assert_eq!(snap.daemons.len(), 1); + assert_eq!(snap.daemons[0].daemon_id, "yamato"); + assert!(snap.sessions.is_empty()); + assert_eq!(snap.version, 2); + } + + #[test] + fn daemon_added_then_removed() { + let store = CatalogStore::default(); + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![], + sessions: vec![], + }); + store.apply_event(RegistryEvent::DaemonAdded { + daemon: daemon("konishi"), + }); + assert_eq!(store.snapshot().daemons.len(), 1); + store.apply_event(RegistryEvent::DaemonRemoved { + daemon_id: "konishi".into(), + }); + assert!(store.snapshot().daemons.is_empty()); + } + + #[test] + fn daemon_removed_evicts_its_sessions() { + let store = CatalogStore::default(); + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![daemon("konishi"), daemon("yamato")], + sessions: vec![session(1, "konishi", "a"), session(2, "yamato", "b")], + }); + store.apply_event(RegistryEvent::DaemonRemoved { + daemon_id: "konishi".into(), + }); + let snap = store.snapshot(); + assert_eq!(snap.sessions.len(), 1); + assert_eq!(snap.sessions[0].daemon_id, "yamato"); + } + + #[test] + fn session_added_updated_removed() { + let store = CatalogStore::default(); + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![daemon("konishi")], + sessions: vec![], + }); + store.apply_event(RegistryEvent::SessionAdded { + session: session(7, "konishi", "vim"), + }); + assert_eq!(store.snapshot().sessions.len(), 1); + assert_eq!(store.snapshot().sessions[0].command, "vim"); + + let mut updated = session(7, "konishi", "claude"); + updated.status = SessionStatus::Dead; + store.apply_event(RegistryEvent::SessionUpdated { session: updated }); + let snap = store.snapshot(); + assert_eq!(snap.sessions[0].command, "claude"); + assert!(matches!(snap.sessions[0].status, SessionStatus::Dead)); + + store.apply_event(RegistryEvent::SessionRemoved { + daemon_id: "konishi".into(), + id: Uuid::from_u128(7), + }); + assert!(store.snapshot().sessions.is_empty()); + } + + #[test] + fn lagged_event_doesnt_blow_up_state() { + let store = CatalogStore::default(); + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![daemon("konishi")], + sessions: vec![session(1, "konishi", "a")], + }); + let before = store.snapshot(); + store.apply_event(RegistryEvent::Lagged { skipped: 42 }); + let after = store.snapshot(); + assert_eq!(after.daemons, before.daemons); + assert_eq!(after.sessions, before.sessions); + assert_eq!(after.version, before.version + 1); + } + + #[test] + fn snapshot_is_sorted() { + let store = CatalogStore::default(); + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![daemon("yamato"), daemon("konishi"), daemon("aurora")], + sessions: vec![ + session(3, "aurora", "c"), + session(1, "konishi", "a"), + session(2, "yamato", "b"), + ], + }); + let snap = store.snapshot(); + let ids: Vec<_> = snap.daemons.iter().map(|d| d.daemon_id.as_str()).collect(); + assert_eq!(ids, vec!["aurora", "konishi", "yamato"]); + let session_ids: Vec<_> = snap.sessions.iter().map(|s| s.id.as_u128()).collect(); + assert_eq!(session_ids, vec![1, 2, 3]); + } + + #[tokio::test] + async fn wait_for_initial_snapshot_resolves_after_first_event() { + let store = CatalogStore::default(); + let task_store = store.clone(); + let handle = tokio::spawn(async move { + task_store.wait_for_initial_snapshot().await; + }); + tokio::time::sleep(Duration::from_millis(10)).await; + assert!(!handle.is_finished()); + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![], + sessions: vec![], + }); + // Give the waker a chance. + tokio::time::timeout(Duration::from_millis(200), handle) + .await + .expect("waker did not fire") + .unwrap(); + } + + #[tokio::test] + async fn changed_notifies_on_every_apply() { + let store = CatalogStore::default(); + let waiter = { + let s = store.clone(); + tokio::spawn(async move { s.changed().await }) + }; + tokio::time::sleep(Duration::from_millis(10)).await; + store.apply_event(RegistryEvent::Snapshot { + daemons: vec![], + sessions: vec![], + }); + tokio::time::timeout(Duration::from_millis(200), waiter) + .await + .expect("changed did not fire") + .unwrap(); + } +} diff --git a/crates/maestro-tui/src/lib.rs b/crates/maestro-tui/src/lib.rs index 36eb75f..fd23a38 100644 --- a/crates/maestro-tui/src/lib.rs +++ b/crates/maestro-tui/src/lib.rs @@ -9,6 +9,7 @@ #![deny(missing_debug_implementations)] pub mod action; +pub mod catalog; pub mod cli; pub mod config; pub mod input; diff --git a/crates/maestro-tui/tests/catalog_ws.rs b/crates/maestro-tui/tests/catalog_ws.rs new file mode 100644 index 0000000..7805f53 --- /dev/null +++ b/crates/maestro-tui/tests/catalog_ws.rs @@ -0,0 +1,94 @@ +//! WS round-trip test for `CatalogStore::run_events_loop`. Spins a tiny axum +//! server that emulates the daemon's `/api/events` contract: send a `Snapshot` +//! on connect, then a delta. + +use std::time::Duration; + +use axum::extract::ws::{Message, WebSocketUpgrade}; +use axum::response::Response; +use axum::routing::any; +use axum::Router; +use chrono::Utc; +use maestro_protocol::{DaemonInfo, RegistryEvent, Session, SessionStatus}; +use maestro_tui::catalog::CatalogStore; +use uuid::Uuid; + +fn daemon(id: &str) -> DaemonInfo { + DaemonInfo { + daemon_id: id.into(), + hostname: format!("{id}.lan"), + version: "test".into(), + daemon_url: format!("ws://{id}.lan:7070"), + } +} + +fn session(id: u128, daemon_id: &str) -> Session { + Session { + id: Uuid::from_u128(id), + daemon_id: daemon_id.into(), + command: "bash".into(), + cwd: std::path::PathBuf::from("/tmp"), + env: Default::default(), + metadata: Default::default(), + status: SessionStatus::Running, + created_at: Utc::now(), + } +} + +async fn events_handler(ws: WebSocketUpgrade) -> Response { + ws.on_upgrade(|mut socket| async move { + let snapshot = RegistryEvent::Snapshot { + daemons: vec![daemon("konishi")], + sessions: vec![], + }; + let text = serde_json::to_string(&snapshot).unwrap(); + if socket.send(Message::Text(text)).await.is_err() { + return; + } + // Small delay so the client has the snapshot applied before the delta. + tokio::time::sleep(Duration::from_millis(50)).await; + let added = RegistryEvent::SessionAdded { + session: session(1, "konishi"), + }; + let text = serde_json::to_string(&added).unwrap(); + let _ = socket.send(Message::Text(text)).await; + // Keep the connection alive so the client doesn't enter reconnect. + tokio::time::sleep(Duration::from_secs(2)).await; + }) +} + +#[tokio::test(flavor = "multi_thread")] +async fn run_events_loop_applies_snapshot_then_delta() { + let app = Router::new().route("/api/events", any(events_handler)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let store = CatalogStore::default(); + let bg = store.clone(); + let ws_url = format!("ws://{addr}/api/events"); + tokio::spawn(async move { bg.run_events_loop(ws_url).await }); + + // Bootstrap should land within ~250 ms. + tokio::time::timeout( + Duration::from_millis(500), + store.wait_for_initial_snapshot(), + ) + .await + .expect("initial snapshot never arrived"); + + let snap = store.snapshot(); + assert_eq!(snap.daemons.len(), 1); + assert_eq!(snap.daemons[0].daemon_id, "konishi"); + + // Delta arrives shortly after. Poll for up to 500 ms. + for _ in 0..50 { + if store.snapshot().sessions.len() == 1 { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("delta never propagated; got {:?}", store.snapshot()); +} diff --git a/web/dist/.gitkeep b/web/dist/.gitkeep deleted file mode 100644 index e69de29..0000000