diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index 98215af..5e80032 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/crates/maestro-protocol/src/lib.rs @@ -174,6 +174,14 @@ pub enum DaemonInbound { home_daemon: String, session_id: Uuid, }, + /// Catalog delta forwarded from the registry so each connected daemon can + /// maintain a local mirror of the global catalog (peer daemons + their + /// sessions). The registry filters out events originating at the + /// recipient daemon — a daemon only sees *peer* deltas. The first event + /// a freshly-connected daemon receives is a `RegistryEvent::Snapshot`. + CatalogEvent { + event: RegistryEvent, + }, } // --------------------------------------------------------------------------- @@ -278,6 +286,46 @@ pub enum RegistryEvent { }, } +impl RegistryEvent { + /// Filter this event for a daemon recipient that should only see *peer* + /// state. Returns `None` if the entire event concerns the recipient, + /// otherwise a possibly-pruned event with self-references removed. + /// + /// Used by the registry's daemon fan-out (NAP-246) so each daemon's + /// local catalog mirror reflects every other daemon, not itself. + pub fn for_peer(self, self_daemon_id: &str) -> Option { + match self { + RegistryEvent::Snapshot { daemons, sessions } => { + let daemons = daemons + .into_iter() + .filter(|d| d.daemon_id != self_daemon_id) + .collect::>(); + let sessions = sessions + .into_iter() + .filter(|s| s.daemon_id != self_daemon_id) + .collect::>(); + Some(RegistryEvent::Snapshot { daemons, sessions }) + } + RegistryEvent::DaemonAdded { ref daemon } if daemon.daemon_id == self_daemon_id => None, + RegistryEvent::DaemonRemoved { ref daemon_id } if daemon_id == self_daemon_id => None, + RegistryEvent::SessionAdded { ref session } if session.daemon_id == self_daemon_id => { + None + } + RegistryEvent::SessionRemoved { ref daemon_id, .. } if daemon_id == self_daemon_id => { + None + } + RegistryEvent::SessionUpdated { ref session } + if session.daemon_id == self_daemon_id => + { + None + } + // Everything else (Lagged, peer adds/removes/updates) is forwarded + // as-is. + other => Some(other), + } + } +} + // --------------------------------------------------------------------------- // Client ↔ Daemon control messages (text frames on the attach WS) // --------------------------------------------------------------------------- @@ -711,6 +759,120 @@ mod tests { } } + #[test] + fn registry_event_for_peer_filters_self() { + let mine = sample_session(); // daemon_id = "laptop" + let mut peer_session = sample_session(); + peer_session.daemon_id = "konishi".into(); + peer_session.id = Uuid::from_u128(99); + + let snapshot = RegistryEvent::Snapshot { + daemons: vec![ + DaemonInfo { + daemon_id: "laptop".into(), + hostname: "naps-laptop".into(), + version: "0.1.0".into(), + daemon_url: "ws://10.0.0.1:7070".into(), + }, + DaemonInfo { + daemon_id: "konishi".into(), + hostname: "konishi".into(), + version: "0.1.0".into(), + daemon_url: "ws://10.0.0.2:7070".into(), + }, + ], + sessions: vec![mine.clone(), peer_session.clone()], + }; + let filtered = snapshot.clone().for_peer("laptop").unwrap(); + match filtered { + RegistryEvent::Snapshot { daemons, sessions } => { + assert_eq!(daemons.len(), 1); + assert_eq!(daemons[0].daemon_id, "konishi"); + assert_eq!(sessions.len(), 1); + assert_eq!(sessions[0].daemon_id, "konishi"); + } + other => panic!("unexpected {other:?}"), + } + } + + #[test] + fn registry_event_for_peer_drops_self_deltas() { + let mine = sample_session(); // daemon_id "laptop" + assert!(RegistryEvent::SessionAdded { + session: mine.clone() + } + .for_peer("laptop") + .is_none()); + assert!(RegistryEvent::SessionUpdated { + session: mine.clone() + } + .for_peer("laptop") + .is_none()); + assert!(RegistryEvent::SessionRemoved { + daemon_id: "laptop".into(), + id: mine.id, + } + .for_peer("laptop") + .is_none()); + assert!(RegistryEvent::DaemonAdded { + daemon: DaemonInfo { + daemon_id: "laptop".into(), + hostname: "x".into(), + version: "y".into(), + daemon_url: "z".into(), + } + } + .for_peer("laptop") + .is_none()); + assert!(RegistryEvent::DaemonRemoved { + daemon_id: "laptop".into() + } + .for_peer("laptop") + .is_none()); + } + + #[test] + fn registry_event_for_peer_forwards_peer_deltas_and_lagged() { + let mut peer = sample_session(); + peer.daemon_id = "konishi".into(); + assert!(matches!( + RegistryEvent::SessionAdded { + session: peer.clone() + } + .for_peer("laptop"), + Some(RegistryEvent::SessionAdded { .. }) + )); + assert!(matches!( + RegistryEvent::Lagged { skipped: 3 }.for_peer("laptop"), + Some(RegistryEvent::Lagged { skipped: 3 }) + )); + } + + #[test] + fn daemon_inbound_catalog_event_round_trip_and_tag() { + let snapshot = DaemonInbound::CatalogEvent { + event: RegistryEvent::Snapshot { + daemons: vec![DaemonInfo { + daemon_id: "peer".into(), + hostname: "peer-host".into(), + version: "0.1.0".into(), + daemon_url: "ws://peer:7070".into(), + }], + sessions: vec![sample_session()], + }, + }; + let v = serde_json::to_value(&snapshot).unwrap(); + assert_eq!(v["type"], "CatalogEvent"); + assert_eq!(snapshot, round_trip(&snapshot)); + + let delta = DaemonInbound::CatalogEvent { + event: RegistryEvent::SessionAdded { + session: sample_session(), + }, + }; + assert_eq!(delta, round_trip(&delta)); + } + #[test] fn daemon_inbound_relay_setup_round_trip_and_tag() { let msg = DaemonInbound::RelaySetup {