diff --git a/crates/maestro-daemon/src/catalog_mirror.rs b/crates/maestro-daemon/src/catalog_mirror.rs index 4d93b10..9cd360a 100644 --- a/crates/maestro-daemon/src/catalog_mirror.rs +++ b/crates/maestro-daemon/src/catalog_mirror.rs @@ -27,6 +27,10 @@ struct State { #[derive(Debug, Clone, Default)] pub struct CatalogMirror { inner: Arc>, + /// Optional fan-out (NAP-248). When attached, every applied event is + /// re-published here so `WS /api/events` subscribers see peer state + /// changes the same way the registry's clients do. + bus: Option, } impl CatalogMirror { @@ -34,40 +38,59 @@ impl CatalogMirror { Self::default() } + /// Attach a [`LocalEventBus`](crate::local_events::LocalEventBus) so peer + /// catalog deltas reach the daemon's `/api/events` subscribers. + pub fn with_bus(mut self, bus: crate::local_events::LocalEventBus) -> Self { + self.bus = Some(bus); + self + } + /// Apply one delta. Snapshot events replace the mirror wholesale; /// `Lagged` resets it on the assumption that the next full snapshot will /// arrive later (today the registry doesn't re-send one — see follow-up /// notes in `handler.rs`). pub async fn apply(&self, event: RegistryEvent) { - let mut s = self.inner.lock().await; - match event { - RegistryEvent::Snapshot { daemons, sessions } => { - s.daemons.clear(); - s.sessions.clear(); - for d in daemons { - s.daemons.insert(d.daemon_id.clone(), d); + { + let mut s = self.inner.lock().await; + match &event { + RegistryEvent::Snapshot { daemons, sessions } => { + s.daemons.clear(); + s.sessions.clear(); + for d in daemons { + s.daemons.insert(d.daemon_id.clone(), d.clone()); + } + for sess in sessions { + s.sessions + .insert((sess.daemon_id.clone(), sess.id), sess.clone()); + } } - for sess in sessions { - s.sessions.insert((sess.daemon_id.clone(), sess.id), sess); + RegistryEvent::DaemonAdded { daemon } => { + s.daemons.insert(daemon.daemon_id.clone(), daemon.clone()); + } + RegistryEvent::DaemonRemoved { daemon_id } => { + s.daemons.remove(daemon_id); + s.sessions.retain(|(d, _), _| d != daemon_id); + } + RegistryEvent::SessionAdded { session } + | RegistryEvent::SessionUpdated { session } => { + s.sessions + .insert((session.daemon_id.clone(), session.id), session.clone()); + } + RegistryEvent::SessionRemoved { daemon_id, id } => { + s.sessions.remove(&(daemon_id.clone(), *id)); + } + RegistryEvent::Lagged { skipped } => { + tracing::warn!( + skipped = *skipped, + "catalog mirror lagged — local view may be stale" + ); } } - RegistryEvent::DaemonAdded { daemon } => { - s.daemons.insert(daemon.daemon_id.clone(), daemon); - } - RegistryEvent::DaemonRemoved { daemon_id } => { - s.daemons.remove(&daemon_id); - s.sessions.retain(|(d, _), _| d != &daemon_id); - } - RegistryEvent::SessionAdded { session } | RegistryEvent::SessionUpdated { session } => { - s.sessions - .insert((session.daemon_id.clone(), session.id), session); - } - RegistryEvent::SessionRemoved { daemon_id, id } => { - s.sessions.remove(&(daemon_id, id)); - } - RegistryEvent::Lagged { skipped } => { - tracing::warn!(skipped, "catalog mirror lagged — local view may be stale"); - } + } + // Re-publish AFTER releasing the lock so subscribers reading `daemons()` + // / `sessions()` in response observe the new state. + if let Some(bus) = &self.bus { + bus.publish(event); } } diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index 319c48e..1fe4f11 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -13,6 +13,7 @@ pub mod config; pub mod configure; pub mod jwt; pub mod local_api; +pub mod local_events; pub mod pty; pub mod registry_client; pub mod session; @@ -25,5 +26,6 @@ pub use catalog_mirror::CatalogMirror; pub use config::{default_config_path, default_state_dir, load_config, DaemonConfig, EnvOverrides}; pub use configure::{run as run_configure, ConfigureError, ConfigureInputs}; pub use jwt::{issue as issue_jwt, JwtError, JwtVerifier, PairAttachVerifier}; +pub use local_events::{spawn_session_publisher, LocalEventBus}; pub use registry_client::{ClientError, RegistryClient, HEARTBEAT_INTERVAL}; pub use session::{SessionError, SessionEvent, SessionManager}; diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index df0a6e9..efb2459 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -33,6 +33,7 @@ use serde::Serialize; use uuid::Uuid; use crate::catalog_mirror::CatalogMirror; +use crate::local_events::LocalEventBus; use crate::session::SessionManager; /// Daemon identity returned by `GET /api/web/info`. Browsers render this in the @@ -54,6 +55,9 @@ pub struct LocalApiState { /// Peer-only mirror of the global catalog. Empty until the daemon's /// registry client connects and receives the snapshot. pub catalog: CatalogMirror, + /// Daemon-local fan-out of own + peer catalog deltas (NAP-248). The + /// `/api/events` WS subscribes here; everything else just reads REST. + pub events: LocalEventBus, } pub fn router(state: LocalApiState) -> Router { @@ -62,6 +66,7 @@ pub fn router(state: LocalApiState) -> Router { .route("/api/sessions/:id", delete(kill_session)) .route("/api/daemons", get(list_daemons)) .route("/api/web/info", get(web_info)) + .route("/api/events", any(events_ws)) .with_state(state) } @@ -166,15 +171,90 @@ pub fn state_for( hostname: impl Into, daemon_url: impl Into, catalog: CatalogMirror, + events: LocalEventBus, ) -> LocalApiState { LocalApiState { manager, hostname: Arc::new(hostname.into()), daemon_url: Arc::new(daemon_url.into()), catalog, + events, } } +// --------------------------------------------------------------------------- +// WS /api/events — live catalog deltas for browser/CLI subscribers (NAP-248) +// --------------------------------------------------------------------------- + +async fn events_ws( + State(s): State, + ws: axum::extract::WebSocketUpgrade, +) -> Response { + ws.on_upgrade(move |socket| async move { + let _ = pump_events(socket, s).await; + }) +} + +async fn pump_events( + socket: axum::extract::ws::WebSocket, + s: LocalApiState, +) -> Result<(), axum::Error> { + use axum::extract::ws::Message; + use futures_util::{SinkExt, StreamExt}; + use maestro_protocol::RegistryEvent; + use tokio::sync::broadcast::error::RecvError; + + let (mut sink, mut stream) = socket.split(); + + // Subscribe BEFORE building the snapshot so deltas applied while we're + // marshaling can't slip between the two and disappear. The UI is + // idempotent for repeated *Added events, so any overlap is harmless. + let mut rx = s.events.subscribe(); + + let self_info = DaemonInfo { + daemon_id: s.manager.daemon_id().to_string(), + hostname: (*s.hostname).clone(), + version: PROTOCOL_VERSION.into(), + daemon_url: (*s.daemon_url).clone(), + }; + let mut daemons = vec![self_info]; + daemons.extend(s.catalog.daemons().await); + let mut sessions = s.manager.snapshot().await; + sessions.extend(s.catalog.sessions().await); + let snapshot = RegistryEvent::Snapshot { daemons, sessions }; + let text = serde_json::to_string(&snapshot).expect("snapshot encode"); + if sink.send(Message::Text(text)).await.is_err() { + return Ok(()); + } + + loop { + tokio::select! { + ev = rx.recv() => { + let event = match ev { + Ok(e) => e, + Err(RecvError::Lagged(n)) => RegistryEvent::Lagged { skipped: n }, + Err(RecvError::Closed) => break, + }; + let text = serde_json::to_string(&event).expect("event encode"); + if sink.send(Message::Text(text)).await.is_err() { + break; + } + } + frame = stream.next() => { + // Server-push only. Anything from the client either closes or + // is ignored. + match frame { + None => break, + Some(Err(_)) => break, + Some(Ok(Message::Close(_))) => break, + Some(Ok(_)) => {} + } + } + } + } + Ok(()) +} + // --------------------------------------------------------------------------- // Broker WS — `WS /api/attach/:session_id` // --------------------------------------------------------------------------- diff --git a/crates/maestro-daemon/src/local_events.rs b/crates/maestro-daemon/src/local_events.rs new file mode 100644 index 0000000..538d552 --- /dev/null +++ b/crates/maestro-daemon/src/local_events.rs @@ -0,0 +1,124 @@ +//! Daemon-local fan-out of `RegistryEvent`s for the `WS /api/events` +//! endpoint (NAP-248). +//! +//! Two sources feed this bus: +//! +//! 1. **Own-session events** — the daemon's `SessionManager::events()` stream, +//! translated from `SessionEvent` into the matching `RegistryEvent` variant +//! by [`spawn_session_publisher`]. +//! 2. **Peer-catalog events** — applied to the daemon's [`CatalogMirror`] when +//! the registry fans them out (NAP-246). `CatalogMirror::apply` publishes +//! each event here after mutating its local view. +//! +//! Subscribers get the merged stream. The `WS /api/events` handler builds a +//! one-shot `Snapshot` of own ∪ peer state on connect, then forwards live +//! deltas until the client disconnects. `Lagged` follows the registry's +//! existing convention — clients resync via REST. + +use std::sync::Arc; + +use maestro_protocol::RegistryEvent; +use tokio::sync::broadcast; + +use crate::session::{SessionEvent, SessionManager}; + +/// Capacity of the per-bus broadcast channel. Matches the registry's +/// `EVENT_CHANNEL_CAPACITY` so a slow client behaves the same on either side. +const BUS_CAPACITY: usize = 256; + +/// Cheap-to-clone handle around a `broadcast::Sender`. +#[derive(Debug, Clone)] +pub struct LocalEventBus { + sender: Arc>, +} + +impl LocalEventBus { + pub fn new() -> Self { + let (sender, _) = broadcast::channel(BUS_CAPACITY); + Self { + sender: Arc::new(sender), + } + } + + /// Push an event onto the bus. No-op if there are no subscribers. + pub fn publish(&self, event: RegistryEvent) { + let _ = self.sender.send(event); + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.sender.subscribe() + } +} + +impl Default for LocalEventBus { + fn default() -> Self { + Self::new() + } +} + +/// Spawn a task that converts each `SessionEvent` from `manager` into the +/// matching `RegistryEvent` and publishes it on `bus`. Exits when the +/// `SessionManager` is dropped (the events channel closes). +pub fn spawn_session_publisher(manager: SessionManager, bus: LocalEventBus) { + let daemon_id = manager.daemon_id().to_string(); + tokio::spawn(async move { + let mut rx = manager.events(); + loop { + match rx.recv().await { + Ok(SessionEvent::Added(session)) => { + bus.publish(RegistryEvent::SessionAdded { session }); + } + Ok(SessionEvent::Updated(session)) => { + bus.publish(RegistryEvent::SessionUpdated { session }); + } + Ok(SessionEvent::Removed(id)) => { + bus.publish(RegistryEvent::SessionRemoved { + daemon_id: daemon_id.clone(), + id, + }); + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + bus.publish(RegistryEvent::Lagged { skipped }); + } + Err(broadcast::error::RecvError::Closed) => return, + } + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn publish_visible_to_subscriber() { + let bus = LocalEventBus::new(); + let mut rx = bus.subscribe(); + bus.publish(RegistryEvent::Lagged { skipped: 7 }); + match rx.recv().await { + Ok(RegistryEvent::Lagged { skipped }) => assert_eq!(skipped, 7), + other => panic!("unexpected: {other:?}"), + } + } + + #[tokio::test] + async fn no_subscribers_swallows_event() { + let bus = LocalEventBus::new(); + // No one's listening — `publish` must not panic and must not block. + bus.publish(RegistryEvent::DaemonRemoved { + daemon_id: "ghost".into(), + }); + } + + #[tokio::test] + async fn clones_share_state() { + let a = LocalEventBus::new(); + let b = a.clone(); + let mut rx = a.subscribe(); + b.publish(RegistryEvent::Lagged { skipped: 3 }); + match rx.recv().await { + Ok(RegistryEvent::Lagged { skipped }) => assert_eq!(skipped, 3), + other => panic!("unexpected: {other:?}"), + } + } +} diff --git a/crates/maestro-daemon/src/main.rs b/crates/maestro-daemon/src/main.rs index af5bcf4..9fe459a 100644 --- a/crates/maestro-daemon/src/main.rs +++ b/crates/maestro-daemon/src/main.rs @@ -15,7 +15,8 @@ use maestro_daemon::tmux::Tmux; use maestro_daemon::worktree::WorktreeManager; use maestro_daemon::{ attach, default_config_path, default_state_dir, load_config, local_api, run_configure, - ConfigureInputs, EnvOverrides, JwtVerifier, PairAttachVerifier, RegistryClient, SessionManager, + spawn_session_publisher, ConfigureInputs, EnvOverrides, JwtVerifier, LocalEventBus, + PairAttachVerifier, RegistryClient, SessionManager, }; use tokio::net::TcpListener; use tracing_subscriber::EnvFilter; @@ -139,15 +140,22 @@ async fn run(config_override: Option) -> Result<(), Box> { config.allow_inbound.iter().cloned(), ); + // Daemon-local event bus (NAP-248): own session events + peer catalog + // deltas funnel through here for `WS /api/events` subscribers. + let events = LocalEventBus::new(); + spawn_session_publisher(manager.clone(), events.clone()); + // Build the registry client first so we can hand its catalog mirror to - // the local-API state — both halves share the same Arc>. - let client = RegistryClient::new( + // the local-API state — both halves share the same Arc>. Hand + // the event bus to the catalog mirror so peer deltas fan out too. + let client = RegistryClient::new_with_bus( config.registry_url.clone(), config.daemon_id.clone(), config.hostname.clone(), config.daemon_url.clone(), config.token.clone(), manager.clone(), + Some(events.clone()), ) .with_perimeter(config.perimeter_id.clone(), config.private_addr.clone()); let catalog = client.catalog(); @@ -157,6 +165,7 @@ async fn run(config_override: Option) -> Result<(), Box> { config.hostname.clone(), config.daemon_url.clone(), catalog.clone(), + events, ); let broker_state = local_api::BrokerState { manager: manager.clone(), diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index c9ea366..6d3ab7a 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -74,8 +74,34 @@ impl RegistryClient { daemon_url: impl Into, token: impl Into, manager: SessionManager, + ) -> Self { + Self::new_with_bus( + registry_url, + daemon_id, + hostname, + daemon_url, + token, + manager, + None, + ) + } + + /// Variant that wires a [`LocalEventBus`](crate::LocalEventBus) into the + /// catalog mirror so peer deltas reach `WS /api/events` (NAP-248). + pub fn new_with_bus( + registry_url: impl Into, + daemon_id: impl Into, + hostname: impl Into, + daemon_url: impl Into, + token: impl Into, + manager: SessionManager, + bus: Option, ) -> Self { let (broker, broker_rx) = BrokerHandle::new(); + let catalog = match &bus { + Some(b) => CatalogMirror::new().with_bus(b.clone()), + None => CatalogMirror::new(), + }; Self { registry_url: registry_url.into(), daemon_id: daemon_id.into(), @@ -85,7 +111,7 @@ impl RegistryClient { perimeter_id: None, private_addr: None, manager, - catalog: CatalogMirror::new(), + catalog, broker, broker_rx, } diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 8e3167e..23d5447 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -11,7 +11,7 @@ use std::time::Duration; use common::DaemonFixture; use futures_util::{SinkExt, StreamExt}; -use maestro_daemon::{local_api, BrokerHandle, CatalogMirror}; +use maestro_daemon::{local_api, BrokerHandle, CatalogMirror, LocalEventBus}; use maestro_protocol::{CreateSessionRequest, DaemonInfo, RegistryEvent, Session}; use tokio::net::TcpListener; use tokio::sync::oneshot; @@ -33,7 +33,14 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness { let fx = DaemonFixture::new(daemon_id); let manager = fx.manager.clone(); let catalog = CatalogMirror::new(); - let state = local_api::state_for(manager.clone(), hostname, daemon_url, catalog.clone()); + let events = LocalEventBus::new(); + let state = local_api::state_for( + manager.clone(), + hostname, + daemon_url, + catalog.clone(), + events, + ); // NAP-247: tests don't exercise the cross-host broker path; hand the // router an isolated BrokerHandle whose receiver is dropped, so any // resolve_attach attempt fails fast with NoControlLink. diff --git a/crates/maestro-e2e/tests/e2e.rs b/crates/maestro-e2e/tests/e2e.rs index 984503b..217824a 100644 --- a/crates/maestro-e2e/tests/e2e.rs +++ b/crates/maestro-e2e/tests/e2e.rs @@ -642,3 +642,69 @@ async fn broker_rejects_when_target_doesnt_allow_inbound() { drop(beta); drop(registry); } + +#[tokio::test(flavor = "multi_thread")] +async fn daemon_events_ws_streams_snapshot_then_session_added() { + // NAP-248: subscribe to a daemon's WS /api/events, expect an initial + // Snapshot, then a SessionAdded after we spawn a session via REST. + let (_reg, daemons, client) = bring_up(&["alpha"]).await; + let cwd = TempDir::new().unwrap(); + + let alpha_addr = daemons[0].bind_addr; + let ws_url = format!("ws://{alpha_addr}/api/events"); + let (mut ws, _) = tokio_tungstenite::connect_async(ws_url) + .await + .expect("events WS handshake"); + + // First frame must be a Snapshot (the WS sends it eagerly on connect). + let snapshot = next_event(&mut ws).await; + match &snapshot { + maestro_protocol::RegistryEvent::Snapshot { daemons, sessions } => { + assert!(daemons.iter().any(|d| d.daemon_id == "alpha")); + assert!(sessions.is_empty()); + } + other => panic!("expected initial Snapshot, got {other:?}"), + } + + let session = client + .spawn_session(&create_req("alpha", cwd.path(), "sleep 30")) + .await + .expect("spawn alpha"); + + // SessionAdded must show up on the WS within a reasonable window. + let saw_added = tokio::time::timeout(Duration::from_secs(5), async { + loop { + match next_event(&mut ws).await { + maestro_protocol::RegistryEvent::SessionAdded { session: s } + if s.id == session.id => + { + return true; + } + _ => continue, + } + } + }) + .await + .unwrap_or(false); + assert!( + saw_added, + "events WS never delivered SessionAdded for own session" + ); + + client.kill_session(session.id).await.expect("kill"); +} + +async fn next_event(ws: &mut S) -> maestro_protocol::RegistryEvent +where + S: StreamExt> + Unpin, +{ + loop { + match ws.next().await { + Some(Ok(WsMessage::Text(t))) => { + return serde_json::from_str(&t).expect("RegistryEvent") + } + Some(Ok(_)) => continue, + other => panic!("WS closed waiting for event: {other:?}"), + } + } +} diff --git a/web/src/events.ts b/web/src/events.ts index 644175d..5aed7e5 100644 --- a/web/src/events.ts +++ b/web/src/events.ts @@ -16,11 +16,6 @@ export class CatalogStore { private backoffMs = 500; start(): void { - // Seed the catalog from REST so the UI renders even when the events WS - // hasn't connected yet (or isn't available — e.g. talking to a daemon's - // local /api surface, which has no /api/events). The WS, when it - // succeeds, takes over with live deltas. - void this.resync(); this.connect(); }