refactor(maestro): typed perimeter advertisement replaces Option pair (NAP-272)
ci / rust (push) Failing after 5m58s
ci / rust (push) Failing after 5m58s
`(perimeter_id, private_addr)` was carried as two `Option<String>`s with an
unwritten "both Some or both None" invariant, enforced ad-hoc at every reader
and silently violable (`with_perimeter(Some, None)`).
Collapse it to `Option<PerimeterCandidate>` (reusing the existing protocol
type, so it flows unconverted from Heartbeat → registry → AttachCandidates):
- protocol: `Heartbeat { perimeter: Option<PerimeterCandidate> }` (empty still
serializes to the legacy `{"type":"Heartbeat"}` shape).
- registry: `DaemonEntry.perimeter`, `update_perimeter`, `perimeter_for` lose
the defensive AND-check; `build_attach_candidates` uses it directly.
- daemon: `RegistryClient.with_perimeter(Option<PerimeterCandidate>)`; the
both-or-neither rule now lives once, in `DaemonConfig::perimeter()`, which
warns and drops a half-configured pair instead of advertising it.
Unblocked by NAP-270's coverage; broker T2 + perimeter-mismatch e2e green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -131,7 +131,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
|
||||
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(
|
||||
|
||||
@@ -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<PerimeterCandidate> {
|
||||
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
|
||||
|
||||
@@ -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<String>,
|
||||
/// Daemon's reachable address on the declared perimeter.
|
||||
private_addr: Option<String>,
|
||||
/// 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<PerimeterCandidate>,
|
||||
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<String>,
|
||||
private_addr: Option<String>,
|
||||
) -> 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<PerimeterCandidate>) -> 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?;
|
||||
}
|
||||
|
||||
@@ -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<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
private_addr: Option<String>,
|
||||
perimeter: Option<PerimeterCandidate>,
|
||||
},
|
||||
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));
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -121,11 +121,10 @@ struct DaemonEntry {
|
||||
version: String,
|
||||
daemon_url: String,
|
||||
outbound: mpsc::Sender<DaemonInbound>,
|
||||
/// Latest perimeter advertised by this daemon in its heartbeats.
|
||||
/// `None` until the daemon advertises one (NAP-247 T2 dial target).
|
||||
perimeter_id: Option<String>,
|
||||
/// Latest private address advertised on the declared perimeter.
|
||||
private_addr: Option<String>,
|
||||
/// 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<PerimeterCandidate>,
|
||||
/// 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<String>,
|
||||
private_addr: Option<String>,
|
||||
) {
|
||||
pub async fn update_perimeter(&self, daemon_id: &str, perimeter: Option<PerimeterCandidate>) {
|
||||
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<PerimeterCandidate> {
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user