feat(maestro): DaemonInbound::CatalogEvent + RegistryEvent::for_peer (NAP-246)

New control-link message wrapping a registry event for daemons that want a
local catalog mirror. `RegistryEvent::for_peer(self_id)` strips events that
concern the recipient itself — daemons mirror peer state, not self.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-26 19:18:08 +00:00
co-authored by Claude Opus 4.7
parent 464bea2fae
commit c4c9c8a24f
+162
View File
@@ -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<RegistryEvent> {
match self {
RegistryEvent::Snapshot { daemons, sessions } => {
let daemons = daemons
.into_iter()
.filter(|d| d.daemon_id != self_daemon_id)
.collect::<Vec<_>>();
let sessions = sessions
.into_iter()
.filter(|s| s.daemon_id != self_daemon_id)
.collect::<Vec<_>>();
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 {