From 2bf6f95506f9e7d623f6192d82cf1bf1607f9225 Mon Sep 17 00:00:00 2001 From: naps62-yolo Date: Tue, 26 May 2026 19:18:25 +0000 Subject: [PATCH] =?UTF-8?q?feat(maestro):=20daemon=20CatalogMirror=20+=20o?= =?UTF-8?q?wn=E2=88=AApeer=20HTTP=20responses=20(NAP-246)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New module crates/maestro-daemon/src/catalog_mirror.rs: in-memory mirror of the global catalog (peer daemons + their sessions), updated from the `DaemonInbound::CatalogEvent` frames the registry now fans out. `local_api` consumes it: `GET /api/sessions` returns the union of the daemon's own sessions and the mirrored peers; `GET /api/daemons` returns self + connected peers. The web UI sees the same catalog from any home daemon. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/maestro-daemon/src/catalog_mirror.rs | 172 +++++++++++++++++++ crates/maestro-daemon/src/lib.rs | 2 + crates/maestro-daemon/src/local_api.rs | 18 +- crates/maestro-daemon/src/main.rs | 29 ++-- crates/maestro-daemon/src/registry_client.rs | 16 ++ crates/maestro-daemon/tests/local_api.rs | 74 +++++++- 6 files changed, 290 insertions(+), 21 deletions(-) create mode 100644 crates/maestro-daemon/src/catalog_mirror.rs diff --git a/crates/maestro-daemon/src/catalog_mirror.rs b/crates/maestro-daemon/src/catalog_mirror.rs new file mode 100644 index 0000000..4d93b10 --- /dev/null +++ b/crates/maestro-daemon/src/catalog_mirror.rs @@ -0,0 +1,172 @@ +//! Daemon-local mirror of the registry catalog (NAP-246). +//! +//! The registry fans out a stream of `RegistryEvent`s to each connected +//! daemon, filtered so each one only sees *peer* state. This module owns the +//! resulting in-memory copy and exposes read snapshots to the daemon's HTTP +//! handlers (`GET /api/sessions`, `GET /api/daemons`). +//! +//! The mirror does *not* include the daemon's own sessions — those come from +//! `SessionManager`. Combining the two for HTTP responses is the caller's +//! job (see `local_api.rs`). + +use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; + +use maestro_protocol::{DaemonInfo, RegistryEvent, Session}; +use tokio::sync::Mutex; +use uuid::Uuid; + +#[derive(Debug, Default)] +struct State { + daemons: HashMap, + /// `(daemon_id, session_id) -> Session` — mirrors the registry's catalog + /// keying so removals (which arrive as `(daemon_id, id)`) are O(1). + sessions: BTreeMap<(String, Uuid), Session>, +} + +#[derive(Debug, Clone, Default)] +pub struct CatalogMirror { + inner: Arc>, +} + +impl CatalogMirror { + pub fn new() -> Self { + Self::default() + } + + /// 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); + } + 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); + } + 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"); + } + } + } + + /// Snapshot the peer daemons currently mirrored. + pub async fn daemons(&self) -> Vec { + self.inner.lock().await.daemons.values().cloned().collect() + } + + /// Snapshot every peer session currently mirrored. + pub async fn sessions(&self) -> Vec { + self.inner.lock().await.sessions.values().cloned().collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::Utc; + use maestro_protocol::SessionStatus; + use std::collections::HashMap as StdHashMap; + + fn peer_session(daemon: &str, id_lo: u128) -> Session { + Session { + id: Uuid::from_u128(id_lo), + daemon_id: daemon.into(), + cwd: "/tmp".into(), + command: "bash".into(), + env: StdHashMap::new(), + metadata: StdHashMap::new(), + status: SessionStatus::Running, + created_at: Utc::now(), + } + } + + fn peer_daemon(daemon_id: &str) -> DaemonInfo { + DaemonInfo { + daemon_id: daemon_id.into(), + hostname: format!("{daemon_id}-host"), + version: "0.1.0".into(), + daemon_url: format!("ws://{daemon_id}:7070"), + } + } + + #[tokio::test] + async fn snapshot_replaces_state() { + let m = CatalogMirror::new(); + m.apply(RegistryEvent::DaemonAdded { + daemon: peer_daemon("old"), + }) + .await; + m.apply(RegistryEvent::Snapshot { + daemons: vec![peer_daemon("new")], + sessions: vec![peer_session("new", 1)], + }) + .await; + let ds = m.daemons().await; + assert_eq!(ds.len(), 1); + assert_eq!(ds[0].daemon_id, "new"); + assert_eq!(m.sessions().await.len(), 1); + } + + #[tokio::test] + async fn daemon_removed_cascades_to_sessions() { + let m = CatalogMirror::new(); + m.apply(RegistryEvent::DaemonAdded { + daemon: peer_daemon("konishi"), + }) + .await; + m.apply(RegistryEvent::SessionAdded { + session: peer_session("konishi", 1), + }) + .await; + m.apply(RegistryEvent::SessionAdded { + session: peer_session("konishi", 2), + }) + .await; + assert_eq!(m.sessions().await.len(), 2); + + m.apply(RegistryEvent::DaemonRemoved { + daemon_id: "konishi".into(), + }) + .await; + assert!(m.daemons().await.is_empty()); + assert!(m.sessions().await.is_empty()); + } + + #[tokio::test] + async fn session_update_replaces_in_place() { + let m = CatalogMirror::new(); + m.apply(RegistryEvent::SessionAdded { + session: peer_session("konishi", 1), + }) + .await; + let mut updated = peer_session("konishi", 1); + updated.status = SessionStatus::Dead; + m.apply(RegistryEvent::SessionUpdated { session: updated }) + .await; + let after = m.sessions().await; + assert_eq!(after.len(), 1); + assert_eq!(after[0].status, SessionStatus::Dead); + } +} diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index d8d3a04..048ec4f 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -7,6 +7,7 @@ #![deny(missing_debug_implementations)] pub mod attach; +pub mod catalog_mirror; pub mod config; pub mod configure; pub mod jwt; @@ -18,6 +19,7 @@ pub mod sidecar; pub mod tmux; pub mod worktree; +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}; diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index 102d79d..fe1c10b 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -32,6 +32,7 @@ use maestro_protocol::{CreateSessionRequest, DaemonInfo, ErrorInfo, Session, PRO use serde::Serialize; use uuid::Uuid; +use crate::catalog_mirror::CatalogMirror; use crate::session::SessionManager; /// Daemon identity returned by `GET /api/web/info`. Browsers render this in the @@ -50,6 +51,9 @@ pub struct LocalApiState { pub manager: SessionManager, pub hostname: Arc, pub daemon_url: Arc, + /// Peer-only mirror of the global catalog. Empty until the daemon's + /// registry client connects and receives the snapshot. + pub catalog: CatalogMirror, } pub fn router(state: LocalApiState) -> Router { @@ -62,17 +66,21 @@ pub fn router(state: LocalApiState) -> Router { } async fn list_sessions(State(s): State) -> Json> { - Json(s.manager.snapshot().await) + let mut out = s.manager.snapshot().await; + out.extend(s.catalog.sessions().await); + Json(out) } async fn list_daemons(State(s): State) -> Json> { - // v1: a daemon only knows about itself. NAP-246 populates peers. - Json(vec![DaemonInfo { + 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 out = vec![self_info]; + out.extend(s.catalog.daemons().await); + Json(out) } async fn web_info(State(s): State) -> Json { @@ -157,11 +165,13 @@ pub fn state_for( manager: SessionManager, hostname: impl Into, daemon_url: impl Into, + catalog: CatalogMirror, ) -> LocalApiState { LocalApiState { manager, hostname: Arc::new(hostname.into()), daemon_url: Arc::new(daemon_url.into()), + catalog, } } diff --git a/crates/maestro-daemon/src/main.rs b/crates/maestro-daemon/src/main.rs index 705ea26..8f21c51 100644 --- a/crates/maestro-daemon/src/main.rs +++ b/crates/maestro-daemon/src/main.rs @@ -133,19 +133,9 @@ async fn run(config_override: Option) -> Result<(), Box> { manager.start_liveness_watcher(DEFAULT_LIVENESS_INTERVAL); let verifier = JwtVerifier::new(&config.token, &config.daemon_id); - let local_state = local_api::state_for( - manager.clone(), - config.hostname.clone(), - config.daemon_url.clone(), - ); - let router = attach::router(manager.clone(), verifier) - .merge(local_api::router(local_state)) - .merge(local_api::broker_router(manager.clone())); - - let listener = TcpListener::bind(config.bind_addr).await?; - let bound = listener.local_addr()?; - tracing::info!(addr = %bound, "attach WS listening"); + // 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( config.registry_url.clone(), config.daemon_id.clone(), @@ -155,6 +145,21 @@ async fn run(config_override: Option) -> Result<(), Box> { manager.clone(), ) .with_perimeter(config.perimeter_id.clone(), config.private_addr.clone()); + let catalog = client.catalog(); + + let local_state = local_api::state_for( + manager.clone(), + config.hostname.clone(), + config.daemon_url.clone(), + catalog, + ); + let router = attach::router(manager.clone(), verifier) + .merge(local_api::router(local_state)) + .merge(local_api::broker_router(manager.clone())); + + let listener = TcpListener::bind(config.bind_addr).await?; + let bound = listener.local_addr()?; + tracing::info!(addr = %bound, "attach WS listening"); tokio::spawn(async move { client.run().await }); diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index e6bd836..4f8f7a4 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -17,6 +17,7 @@ use maestro_protocol::{DaemonInbound, DaemonOutbound, RpcResult, PROTOCOL_VERSIO use tokio::sync::broadcast::error::RecvError; use tokio_tungstenite::tungstenite::Message; +use crate::catalog_mirror::CatalogMirror; use crate::session::{SessionEvent, SessionManager}; /// Per-message heartbeat interval. Less than the registry's 30s timeout, with @@ -49,6 +50,10 @@ pub struct RegistryClient { /// Daemon's reachable address on the declared perimeter. private_addr: Option, manager: SessionManager, + /// Local mirror of the global catalog (peer daemons + peer sessions), + /// populated from `DaemonInbound::CatalogEvent` frames the registry + /// fans out. + catalog: CatalogMirror, } impl RegistryClient { @@ -69,9 +74,16 @@ impl RegistryClient { perimeter_id: None, private_addr: None, manager, + catalog: CatalogMirror::new(), } } + /// Cheap handle to the daemon's peer-catalog mirror. Clone freely; + /// internal state is `Arc>`. + pub fn catalog(&self) -> CatalogMirror { + self.catalog.clone() + } + /// Set the perimeter advertised in heartbeats. Pass both or neither — /// a private address without a perimeter id is meaningless. pub fn with_perimeter( @@ -235,6 +247,10 @@ impl RegistryClient { // emitting these messages don't crash older daemons on parse. tracing::debug!("broker message received but unhandled (NAP-243 pending)"); } + DaemonInbound::CatalogEvent { event } => { + // NAP-246: peer-filtered catalog delta from the registry. + self.catalog.apply(event).await; + } } Ok(()) } diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 8c0c22c..90ef17d 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -11,8 +11,8 @@ use std::time::Duration; use common::DaemonFixture; use futures_util::{SinkExt, StreamExt}; -use maestro_daemon::local_api; -use maestro_protocol::{CreateSessionRequest, DaemonInfo, Session}; +use maestro_daemon::{local_api, CatalogMirror}; +use maestro_protocol::{CreateSessionRequest, DaemonInfo, RegistryEvent, Session}; use tokio::net::TcpListener; use tokio::sync::oneshot; use tokio_tungstenite::tungstenite::Message as WsMessage; @@ -24,6 +24,7 @@ type WsClient = WebSocketStream>; struct Harness { addr: SocketAddr, manager: maestro_daemon::SessionManager, + catalog: CatalogMirror, _shutdown: oneshot::Sender<()>, _fx: DaemonFixture, } @@ -31,7 +32,8 @@ struct Harness { async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness { let fx = DaemonFixture::new(daemon_id); let manager = fx.manager.clone(); - let state = local_api::state_for(manager.clone(), hostname, daemon_url); + let catalog = CatalogMirror::new(); + let state = local_api::state_for(manager.clone(), hostname, daemon_url, catalog.clone()); let router = local_api::router(state).merge(local_api::broker_router(manager.clone())); let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) .await @@ -48,6 +50,7 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness { Harness { addr, manager, + catalog, _shutdown: tx, _fx: fx, } @@ -73,17 +76,78 @@ async fn web_info_returns_daemon_identity() { } #[tokio::test] -async fn list_daemons_returns_self_only() { +async fn list_daemons_includes_self_and_mirrored_peers() { let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + + // Before any catalog events arrive, the daemon only knows about itself. let list: Vec = reqwest::get(url(h.addr, "/api/daemons")) .await .unwrap() .json() .await .unwrap(); - // v1: no peer mirror, so a daemon only knows about itself. assert_eq!(list.len(), 1); assert_eq!(list[0].daemon_id, "laptop"); + + // Simulate the registry fanning out a peer addition. + h.catalog + .apply(RegistryEvent::DaemonAdded { + daemon: DaemonInfo { + daemon_id: "konishi".into(), + hostname: "konishi-host".into(), + version: "0.1.0".into(), + daemon_url: "ws://10.0.0.2:7070".into(), + }, + }) + .await; + + let list: Vec = reqwest::get(url(h.addr, "/api/daemons")) + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(list.len(), 2); + let ids: Vec<&str> = list.iter().map(|d| d.daemon_id.as_str()).collect(); + assert!(ids.contains(&"laptop")); + assert!(ids.contains(&"konishi")); +} + +#[tokio::test] +async fn list_sessions_unions_own_and_peer() { + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + // Create one local session. + let own = h + .manager + .create(CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: "/tmp".into(), + command: "sleep 60".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .await + .unwrap(); + // Inject a peer session via the mirror. + let mut peer = own.clone(); + peer.id = Uuid::new_v4(); + peer.daemon_id = "konishi".into(); + h.catalog + .apply(RegistryEvent::SessionAdded { + session: peer.clone(), + }) + .await; + + let list: Vec = reqwest::get(url(h.addr, "/api/sessions")) + .await + .unwrap() + .json() + .await + .unwrap(); + let ids: Vec = list.iter().map(|s| s.id).collect(); + assert!(ids.contains(&own.id), "own session missing: {ids:?}"); + assert!(ids.contains(&peer.id), "peer session missing: {ids:?}"); } #[tokio::test]