diff --git a/crates/maestro-daemon/src/cli.rs b/crates/maestro-daemon/src/cli.rs index 4c9e61c..6cefeb4 100644 --- a/crates/maestro-daemon/src/cli.rs +++ b/crates/maestro-daemon/src/cli.rs @@ -131,7 +131,7 @@ async fn run(config_override: Option) -> Result<(), Box> { manager.clone(), Some(events.clone()), ) - .with_perimeter(config.perimeter_id.clone(), config.private_addr.clone()); + .with_perimeter(config.perimeter()); let catalog = client.catalog(); let local_state = local_api::state_for( diff --git a/crates/maestro-daemon/src/config.rs b/crates/maestro-daemon/src/config.rs index 41e342f..31406a1 100644 --- a/crates/maestro-daemon/src/config.rs +++ b/crates/maestro-daemon/src/config.rs @@ -17,6 +17,7 @@ use std::net::SocketAddr; use std::path::{Path, PathBuf}; use maestro_config::{derive_daemon_ws_url, MaestroConfigFile}; +use maestro_protocol::PerimeterCandidate; pub use maestro_config::{ default_config_path, resolve_config_path, DaemonSection, SharedSection, CONFIG_PATH_ENV, @@ -107,6 +108,30 @@ impl DaemonConfig { pub fn fifo_dir(&self) -> PathBuf { self.state_dir.join("fifos") } + + /// Combine the two perimeter config keys into the typed advertisement the + /// daemon broadcasts. A perimeter is only advertised when BOTH the id and + /// the private address are set — this is the single place the "both or + /// neither" invariant lives (NAP-272). A half-configured pair is dropped + /// with a warning rather than silently advertising a useless value. + pub fn perimeter(&self) -> Option { + match (self.perimeter_id.clone(), self.private_addr.clone()) { + (Some(perimeter_id), Some(private_addr)) => Some(PerimeterCandidate { + perimeter_id, + private_addr, + }), + (None, None) => None, + (id, addr) => { + tracing::warn!( + perimeter_id = ?id, + private_addr = ?addr, + "ignoring half-configured perimeter — set both [daemon].perimeter_id and \ + [daemon].private_addr, or neither" + ); + None + } + } + } } /// Env-var overrides applied by [`load_config`]. Tests pass explicit diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index c878d51..6d36ac7 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -13,7 +13,9 @@ use std::time::Duration; use futures_util::{SinkExt, StreamExt}; -use maestro_protocol::{DaemonInbound, DaemonOutbound, RpcResult, PROTOCOL_VERSION}; +use maestro_protocol::{ + DaemonInbound, DaemonOutbound, PerimeterCandidate, RpcResult, PROTOCOL_VERSION, +}; use tokio::sync::broadcast::error::RecvError; use tokio_tungstenite::tungstenite::Message; @@ -47,11 +49,10 @@ pub struct RegistryClient { hostname: String, daemon_url: String, token: String, - /// Operator-set perimeter id advertised in heartbeats so peer daemons - /// can dial this one over T2 when they share the same perimeter. - perimeter_id: Option, - /// Daemon's reachable address on the declared perimeter. - private_addr: Option, + /// Operator-set perimeter (id + reachable address) advertised in heartbeats + /// so peer daemons can dial this one over T2 when they share the same + /// perimeter. Typed so "both or neither" holds by construction (NAP-272). + perimeter: Option, manager: SessionManager, /// Local mirror of the global catalog (peer daemons + peer sessions), /// populated from `DaemonInbound::CatalogEvent` frames the registry @@ -108,8 +109,7 @@ impl RegistryClient { hostname: hostname.into(), daemon_url: daemon_url.into(), token: token.into(), - perimeter_id: None, - private_addr: None, + perimeter: None, manager, catalog, broker, @@ -129,15 +129,11 @@ impl RegistryClient { 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( - mut self, - perimeter_id: Option, - private_addr: Option, - ) -> Self { - self.perimeter_id = perimeter_id; - self.private_addr = private_addr; + /// Set the perimeter advertised in heartbeats. The typed argument encodes + /// "both id and addr, or neither" — a private address without a perimeter + /// id is no longer representable (NAP-272). + pub fn with_perimeter(mut self, perimeter: Option) -> Self { + self.perimeter = perimeter; self } @@ -242,8 +238,7 @@ impl RegistryClient { _ = hb.tick() => { sink.send(Message::Text(serde_json::to_string( &DaemonOutbound::Heartbeat { - perimeter_id: self.perimeter_id.clone(), - private_addr: self.private_addr.clone(), + perimeter: self.perimeter.clone(), }, )?)).await?; } diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index 8191ea8..e937a6b 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/crates/maestro-protocol/src/lib.rs @@ -113,14 +113,13 @@ pub enum DaemonOutbound { SessionUpdated { session: Session, }, - /// Liveness ticker. The optional `perimeter_id` / `private_addr` advertise - /// the daemon's reachability for the T2 transport (NAP-242). Old daemons - /// that don't set them still produce the bare `{"type":"Heartbeat"}` shape. + /// Liveness ticker. The optional `perimeter` advertises the daemon's + /// reachability for the T2 transport (NAP-242); the typed object encodes the + /// "both id and addr, or neither" invariant once (NAP-272). Old daemons that + /// don't set it still produce the bare `{"type":"Heartbeat"}` shape. Heartbeat { #[serde(default, skip_serializing_if = "Option::is_none")] - perimeter_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - private_addr: Option, + perimeter: Option, }, CreateSessionResult { reply_to: ReplyTo, @@ -522,13 +521,12 @@ mod tests { DaemonOutbound::SessionUpdated { session: session.clone(), }, + DaemonOutbound::Heartbeat { perimeter: None }, DaemonOutbound::Heartbeat { - perimeter_id: None, - private_addr: None, - }, - DaemonOutbound::Heartbeat { - perimeter_id: Some("wg-naps62".into()), - private_addr: Some("10.0.0.7:7070".into()), + perimeter: Some(PerimeterCandidate { + perimeter_id: "wg-naps62".into(), + private_addr: "10.0.0.7:7070".into(), + }), }, DaemonOutbound::CreateSessionResult { reply_to, @@ -690,12 +688,8 @@ mod tests { let legacy = serde_json::json!({ "type": "Heartbeat" }); let parsed: DaemonOutbound = serde_json::from_value(legacy).unwrap(); match parsed { - DaemonOutbound::Heartbeat { - perimeter_id, - private_addr, - } => { - assert!(perimeter_id.is_none()); - assert!(private_addr.is_none()); + DaemonOutbound::Heartbeat { perimeter } => { + assert!(perimeter.is_none()); } other => panic!("expected Heartbeat, got {other:?}"), } @@ -703,13 +697,10 @@ mod tests { #[test] fn heartbeat_empty_serializes_to_legacy_shape() { - // A new daemon that doesn't set perimeter/addr must produce wire bytes + // A new daemon that doesn't set a perimeter must produce wire bytes // identical to an old daemon — registries doing equality checks or // logging payloads see no surprise. - let hb = DaemonOutbound::Heartbeat { - perimeter_id: None, - private_addr: None, - }; + let hb = DaemonOutbound::Heartbeat { perimeter: None }; let v = serde_json::to_value(&hb).unwrap(); assert_eq!(v, serde_json::json!({ "type": "Heartbeat" })); } @@ -717,13 +708,15 @@ mod tests { #[test] fn heartbeat_with_fields_round_trips() { let hb = DaemonOutbound::Heartbeat { - perimeter_id: Some("wg-naps62".into()), - private_addr: Some("10.0.0.7:7070".into()), + perimeter: Some(PerimeterCandidate { + perimeter_id: "wg-naps62".into(), + private_addr: "10.0.0.7:7070".into(), + }), }; let v = serde_json::to_value(&hb).unwrap(); assert_eq!(v["type"], "Heartbeat"); - assert_eq!(v["perimeter_id"], "wg-naps62"); - assert_eq!(v["private_addr"], "10.0.0.7:7070"); + assert_eq!(v["perimeter"]["perimeter_id"], "wg-naps62"); + assert_eq!(v["perimeter"]["private_addr"], "10.0.0.7:7070"); assert_eq!(hb, round_trip(&hb)); } diff --git a/crates/maestro-registry/src/handler.rs b/crates/maestro-registry/src/handler.rs index ac08eb7..796e64d 100644 --- a/crates/maestro-registry/src/handler.rs +++ b/crates/maestro-registry/src/handler.rs @@ -191,16 +191,11 @@ async fn dispatch_text( DaemonOutbound::SessionUpdated { session } => { registry.update_session(daemon_id, session).await; } - DaemonOutbound::Heartbeat { - perimeter_id, - private_addr, - } => { + DaemonOutbound::Heartbeat { perimeter } => { // Per-read timeout above is the heartbeat watchdog; capture the // perimeter advertisement so the broker can hand it back as a T2 // candidate (NAP-247). - registry - .update_perimeter(daemon_id, perimeter_id, private_addr) - .await; + registry.update_perimeter(daemon_id, perimeter).await; } DaemonOutbound::CreateSessionResult { reply_to, result } => { registry.complete_create(reply_to, result).await; diff --git a/crates/maestro-registry/src/state.rs b/crates/maestro-registry/src/state.rs index 8f88bda..829a275 100644 --- a/crates/maestro-registry/src/state.rs +++ b/crates/maestro-registry/src/state.rs @@ -121,11 +121,10 @@ struct DaemonEntry { version: String, daemon_url: String, outbound: mpsc::Sender, - /// Latest perimeter advertised by this daemon in its heartbeats. - /// `None` until the daemon advertises one (NAP-247 T2 dial target). - perimeter_id: Option, - /// Latest private address advertised on the declared perimeter. - private_addr: Option, + /// Latest perimeter advertised by this daemon in its heartbeats (id + addr + /// together, or absent). `None` until the daemon advertises one (NAP-247 T2 + /// dial target; typed to encode the both-or-neither invariant, NAP-272). + perimeter: Option, /// Fencing token minted per WS connection in [`Registry::register`]. Only /// the handler holding the matching token may [`Registry::deregister`] this /// entry, so a stale handler (old socket timing out after the daemon @@ -206,8 +205,7 @@ impl Registry { version: version.clone(), daemon_url: daemon_url.clone(), outbound, - perimeter_id: None, - private_addr: None, + perimeter: None, connection_token, }, ); @@ -334,30 +332,23 @@ impl Registry { /// Update a daemon's advertised perimeter from its heartbeat. Called by /// the WS handler on every `Heartbeat` frame (NAP-247). - pub async fn update_perimeter( - &self, - daemon_id: &str, - perimeter_id: Option, - private_addr: Option, - ) { + pub async fn update_perimeter(&self, daemon_id: &str, perimeter: Option) { let mut state = self.inner.lock().await; if let Some(entry) = state.daemons.get_mut(daemon_id) { - entry.perimeter_id = perimeter_id; - entry.private_addr = private_addr; + entry.perimeter = perimeter; } } - /// `(perimeter_id, private_addr)` last advertised by this daemon, when - /// both are set. The broker uses this to assemble `AttachCandidates` for - /// the T2 transport (NAP-247). - pub async fn perimeter_for(&self, daemon_id: &str) -> Option<(String, String)> { + /// The perimeter (`perimeter_id` + `private_addr`) last advertised by this + /// daemon, when set. The broker uses this to assemble `AttachCandidates` + /// for the T2 transport (NAP-247). The both-or-neither invariant is encoded + /// in the type, so no defensive AND-check is needed (NAP-272). + pub async fn perimeter_for(&self, daemon_id: &str) -> Option { let state = self.inner.lock().await; - state.daemons.get(daemon_id).and_then(|e| { - match (e.perimeter_id.clone(), e.private_addr.clone()) { - (Some(pid), Some(addr)) => Some((pid, addr)), - _ => None, - } - }) + state + .daemons + .get(daemon_id) + .and_then(|e| e.perimeter.clone()) } /// Handle a `ResolveAttach` from a home daemon (NAP-247). Look up which @@ -427,13 +418,7 @@ impl Registry { // T1 (direct WSS) stays None — NAP-235 was canceled. // T2: hand back the target's perimeter so the home daemon can match. // T3 (relay) stays None — NAP-238 spike pending. - let perimeter = self - .perimeter_for(&target) - .await - .map(|(pid, addr)| PerimeterCandidate { - perimeter_id: pid, - private_addr: addr, - }); + let perimeter = self.perimeter_for(&target).await; RpcResult::Ok { value: AttachCandidates { direct_wss: None, @@ -720,10 +705,12 @@ mod tests { } /// The `(perimeter_id, private_addr)` pair only yields a candidate when - /// BOTH halves are advertised — a half-configured daemon must not leak a - /// T2 dial target (NAP-270). + /// The perimeter is advertised as one typed object, so "both id and addr, + /// or neither" is enforced by construction — `perimeter_for` round-trips + /// `None` and a full advertisement, and a half-pair is unrepresentable + /// (NAP-270 / NAP-272). #[tokio::test] - async fn perimeter_for_requires_both_id_and_addr() { + async fn perimeter_for_round_trips_typed_advertisement() { let reg = Registry::new(); let (_t, _rx) = register(®, "d1").await; @@ -732,22 +719,16 @@ mod tests { "nothing advertised" ); - reg.update_perimeter("d1", Some("wg".into()), None).await; - assert!( - reg.perimeter_for("d1").await.is_none(), - "addr missing → None" - ); + let adv = PerimeterCandidate { + perimeter_id: "wg".into(), + private_addr: "10.0.0.1:7070".into(), + }; + reg.update_perimeter("d1", Some(adv.clone())).await; + assert_eq!(reg.perimeter_for("d1").await, Some(adv)); - reg.update_perimeter("d1", None, Some("10.0.0.1:7070".into())) - .await; - assert!(reg.perimeter_for("d1").await.is_none(), "id missing → None"); - - reg.update_perimeter("d1", Some("wg".into()), Some("10.0.0.1:7070".into())) - .await; - assert_eq!( - reg.perimeter_for("d1").await, - Some(("wg".into(), "10.0.0.1:7070".into())) - ); + // Clearing it goes back to None. + reg.update_perimeter("d1", None).await; + assert!(reg.perimeter_for("d1").await.is_none()); } /// A home daemon asking to attach to a session it already owns is refused