diff --git a/Cargo.lock b/Cargo.lock index 38e825e1e9..89e3765957 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3989,6 +3989,7 @@ dependencies = [ "http", "http-body-util", "httparse", + "if-addrs", "iroh", "iroh-relay", "json5", @@ -4042,6 +4043,7 @@ dependencies = [ "skippy-runtime", "skippy-server", "skippy-topology", + "socket2", "tabwriter", "tar", "tempfile", diff --git a/crates/mesh-llm-host-runtime/Cargo.toml b/crates/mesh-llm-host-runtime/Cargo.toml index b4d5a5807c..a6951766a2 100644 --- a/crates/mesh-llm-host-runtime/Cargo.toml +++ b/crates/mesh-llm-host-runtime/Cargo.toml @@ -65,6 +65,8 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } serde = { version = "1", features = ["derive"] } serde_json = "1" serde_yaml = "0.9" +socket2 = { version = "0.6", features = ["all"] } +if-addrs = "0.15" anyhow = "1" async-trait = "0.1" rand = "0.10" diff --git a/crates/mesh-llm-host-runtime/src/mesh/lan_bootstrap.rs b/crates/mesh-llm-host-runtime/src/mesh/lan_bootstrap.rs new file mode 100644 index 0000000000..f3fd334c90 --- /dev/null +++ b/crates/mesh-llm-host-runtime/src/mesh/lan_bootstrap.rs @@ -0,0 +1,123 @@ +use super::*; + +/// Detect the most likely LAN IPv4 address for mDNS-only meshes. +pub fn detect_primary_lan_ipv4() -> Option { + if let Some(ip) = default_route_source_ipv4().filter(is_private_lan_interface_ipv4) { + return Some(IpAddr::V4(ip)); + } + first_private_lan_interface_ipv4().map(IpAddr::V4) +} + +pub(super) fn lan_ipv4_candidates(addr: &EndpointAddr) -> Vec { + addr.addrs + .iter() + .filter_map(|addr| match addr { + TransportAddr::Ip(SocketAddr::V4(v4)) if is_private_lan_ipv4(v4.ip()) => Some(*v4), + _ => None, + }) + .collect() +} + +fn is_private_lan_ipv4(ip: &Ipv4Addr) -> bool { + ip.is_private() +} + +fn is_private_lan_interface_ipv4(ip: &Ipv4Addr) -> bool { + is_private_lan_ipv4(ip) && !is_container_bridge_ipv4(ip) +} + +fn is_container_bridge_ipv4(ip: &Ipv4Addr) -> bool { + matches!( + ip.octets(), + [10, 88 | 89, _, _] | [10, 96..=111, _, _] | [10, 244, _, _] | [172, 17, _, _] + ) +} + +/// Source IPv4 the kernel would use for the default route, via a connect-trick. +/// +/// 192.88.99.1 is a routable, globally-assigned target; connecting a UDP socket +/// to it only drives route/source selection — it sends nothing. Returns `None` +/// when there is no default route or the source is unspecified/loopback. +fn default_route_source_ipv4() -> Option { + let socket = std::net::UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).ok()?; + socket.connect((Ipv4Addr::new(192, 88, 99, 1), 9)).ok()?; + match socket.local_addr().ok()?.ip() { + IpAddr::V4(v4) if !v4.is_unspecified() && !v4.is_loopback() => Some(v4), + _ => None, + } +} + +/// First operational private-LAN IPv4 from the local interface table. +/// +/// Skips loopback, link-local, point-to-point, and common container bridge +/// addresses so the result is a host LAN interface peers can directly reach. +fn first_private_lan_interface_ipv4() -> Option { + let interfaces = if_addrs::get_if_addrs().ok()?; + interfaces + .into_iter() + .filter(|iface| !iface.is_loopback() && !iface.is_link_local() && !iface.is_p2p()) + .filter_map(|iface| match iface.addr { + if_addrs::IfAddr::V4(v4) => Some(v4.ip), + if_addrs::IfAddr::V6(_) => None, + }) + .find(is_private_lan_interface_ipv4) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn private_rfc1918_ranges_are_lan() { + for ip in [ + Ipv4Addr::new(10, 0, 0, 5), + Ipv4Addr::new(10, 96, 0, 5), + Ipv4Addr::new(172, 16, 4, 9), + Ipv4Addr::new(172, 17, 0, 5), + Ipv4Addr::new(172, 31, 255, 1), + Ipv4Addr::new(192, 168, 86, 60), + ] { + assert!(is_private_lan_ipv4(&ip), "{ip} should be treated as LAN"); + } + } + + #[test] + fn public_cgnat_link_local_and_loopback_are_not_lan() { + for ip in [ + Ipv4Addr::new(8, 8, 8, 8), + Ipv4Addr::new(100, 64, 0, 1), + Ipv4Addr::new(169, 254, 10, 10), + Ipv4Addr::new(127, 0, 0, 1), + Ipv4Addr::new(172, 32, 0, 1), + ] { + assert!(!is_private_lan_ipv4(&ip), "{ip} must not be treated as LAN"); + } + } + + #[test] + fn common_container_bridge_ranges_are_not_selected_as_local_interfaces() { + for ip in [ + Ipv4Addr::new(172, 17, 0, 1), + Ipv4Addr::new(10, 88, 0, 1), + Ipv4Addr::new(10, 96, 0, 1), + Ipv4Addr::new(10, 244, 0, 1), + ] { + assert!( + !is_private_lan_interface_ipv4(&ip), + "{ip} must not be selected as a local LAN interface" + ); + } + } + + #[test] + fn public_candidate_classifier_excludes_private_and_cgnat() { + let public = SocketAddr::from(([203, 0, 113, 0], 9)); + assert!(!is_public_ipv4_candidate(&public)); + let real_public = SocketAddr::from(([9, 9, 9, 9], 9)); + assert!(is_public_ipv4_candidate(&real_public)); + let lan = SocketAddr::from(([192, 168, 1, 50], 9)); + assert!(!is_public_ipv4_candidate(&lan)); + let cgnat = SocketAddr::from(([100, 100, 1, 1], 9)); + assert!(!is_public_ipv4_candidate(&cgnat)); + } +} diff --git a/crates/mesh-llm-host-runtime/src/mesh/mod.rs b/crates/mesh-llm-host-runtime/src/mesh/mod.rs index a3ad5fced9..5bc3660b2e 100644 --- a/crates/mesh-llm-host-runtime/src/mesh/mod.rs +++ b/crates/mesh-llm-host-runtime/src/mesh/mod.rs @@ -444,6 +444,38 @@ fn default_control_bind_addr() -> std::net::SocketAddr { std::net::SocketAddr::from(([127, 0, 0, 1], 0)) } +/// Detect this host's primary **private LAN** IPv4 without sending any packets. +/// +/// Returns a genuine RFC1918 LAN address (`10/8`, `172.16/12`, `192.168/16`) +/// or `None`. It deliberately never returns a public, CGNAT (`100.64/10`), or +/// VPN/tunnel address, so the caller can safely pin QUIC's bind to it. +/// +/// Detection has two phases: +/// +/// 1. **Default-route source probe.** Open an unconnected UDP socket and +/// `connect()` it to a routable target so the kernel fills in the source IP +/// it would use to reach that target. No datagrams are sent. This is the +/// fast, accurate answer on a normal single-LAN host — but on a full-tunnel +/// VPN host the default route points at the tunnel, so the source is a +/// VPN/utun address. We therefore accept this result **only if it is a +/// private LAN IPv4**. +/// 2. **Interface scan fallback.** If the probe yields a non-private address +/// (VPN default route) or fails (no default route on an isolated LAN), scan +/// local interfaces and pick the first private, operational, non-loopback, +/// non-link-local, non-point-to-point IPv4. Point-to-point interfaces are +/// skipped because VPN/tunnel interfaces present as p2p. +/// +/// Used to auto-pin QUIC's bind address to the real LAN interface on +/// multi-homed hosts (e.g. macOS with several `utun`/VPN interfaces). Binding +/// `0.0.0.0` on such hosts lets the kernel pick a wrong source for an +/// unconnected QUIC `sendmsg` (yielding `EHOSTUNREACH` or a slow WAN-hairpin +/// path) and breaks/degrades direct LAN connectivity in either dial direction. +/// Returning only a private LAN IPv4 (or `None`) means a wrong default route +/// can never hard-pin relay-less QUIC off-LAN; we fall back to `0.0.0.0` +/// instead. Public-relay (Nostr) mode keeps its IPv6/relay paths regardless, so +/// long-haul reachability to a remote mesh is never sacrificed for the LAN hint. +pub use lan_bootstrap::detect_primary_lan_ipv4; + fn is_public_ipv4_candidate(socket: &SocketAddr) -> bool { match socket.ip() { IpAddr::V4(ip) => is_global_ipv4_candidate(ip), @@ -2086,6 +2118,19 @@ async fn bind_mesh_endpoint( if let Some(addr) = quic_bind_addr(quic_bind) { tracing::info!("Binding QUIC to {addr}"); + if !relay.policy.uses_relay() && addr.is_ipv4() { + // LAN-only (relay-disabled) mode with a specific IPv4 bind: clear the + // pre-configured default sockets first. `bind_addr` only replaces the + // default for the *same* address family, so binding a specific IPv4 + // would otherwise leave the default IPv6 `[::]` socket in place. That + // extra local IPv6 path becomes a second candidate, and with no relay + // iroh's multipath negotiation across the IPv4+IPv6 locals fails with + // `MultipathNotNegotiated`, stalling the connection with no fallback. + // Pinning a single IPv4 socket keeps one local path family so the LAN + // direct path establishes cleanly. In relay (public) mode we keep the + // defaults so relay/IPv6 reachability is unaffected. + builder = builder.clear_ip_transports(); + } builder = builder.bind_addr(addr)?; } @@ -2266,6 +2311,10 @@ pub struct Node { genesis_policy: Arc>>, signed_genesis_policy: Arc>>, bootstrap_token: Arc>>, + /// Addresses we have been asked to join (from invite tokens), retained so + /// the LAN beacon can unicast a dial-back hint to them even before a direct + /// connection forms (relay-less multi-homed-initiator case). + join_targets: Arc>>, first_joined_mesh_ts: Arc>>, accepting: Arc<(tokio::sync::Notify, std::sync::atomic::AtomicBool)>, vram_bytes: u64, @@ -3760,6 +3809,7 @@ impl Node { genesis_policy: Arc::new(Mutex::new(None)), signed_genesis_policy: Arc::new(Mutex::new(None)), bootstrap_token: Arc::new(Mutex::new(None)), + join_targets: Arc::new(Mutex::new(Vec::new())), first_joined_mesh_ts: Arc::new(Mutex::new(None)), accepting: Arc::new(( tokio::sync::Notify::new(), @@ -3922,6 +3972,7 @@ impl Node { genesis_policy: Arc::new(Mutex::new(None)), signed_genesis_policy: Arc::new(Mutex::new(None)), bootstrap_token: Arc::new(Mutex::new(None)), + join_targets: Arc::new(Mutex::new(Vec::new())), first_joined_mesh_ts: Arc::new(Mutex::new(None)), accepting: Arc::new(( tokio::sync::Notify::new(), @@ -4676,6 +4727,49 @@ impl Node { addr } + /// The local node's reachable [`EndpointAddr`], filtered to the bound LAN + /// interface in the same way the invite token is. Used by mDNS reverse-dial + /// so a host can advertise (and peers can learn) a direct address to dial + /// back on the working direction. + pub fn advertised_endpoint_addr(&self) -> EndpointAddr { + self.endpoint_addr_for_advertisement() + } + + /// Dial a peer by its [`EndpointAddr`] directly (no token decode). + /// + /// Used by mDNS reverse-dial: when a relay-less direct connection cannot be + /// established in one direction (multi-homed initiator), the other side + /// dials back on the direction that works. + pub async fn dial_peer_addr(&self, addr: EndpointAddr) -> Result<()> { + self.state.lock().await.dead_peers.remove(&addr.id); + self.connect_to_peer(addr).await + } + + /// The set of peer endpoint IDs we currently hold a connection to. + /// + /// Used by mDNS reverse-dial to avoid redialing already-connected peers. + pub async fn connected_peer_ids(&self) -> std::collections::HashSet { + self.state + .lock() + .await + .connections + .keys() + .copied() + .collect() + } + + /// LAN IPv4 socket addresses of all known peers (from gossip/tokens), + /// regardless of connection state. Used by the LAN beacon to unicast a + /// dial-back hint directly to peers when multicast is unavailable. + pub async fn known_peer_lan_ipv4(&self) -> Vec { + let state = self.state.lock().await; + let mut out = Vec::new(); + for peer in state.peers.values() { + out.extend(lan_bootstrap::lan_ipv4_candidates(&peer.addr)); + } + out + } + /// Decode an invite token into an [`EndpointAddr`] without connecting. /// Returns `Err` if the token is not valid base64 or not valid JSON. pub fn decode_invite_token(invite_token: &str) -> Result { @@ -4797,9 +4891,37 @@ impl Node { }; // Clear dead status — explicit join should always attempt connection self.state.lock().await.dead_peers.remove(&addr.id); + self.remember_join_target(addr.clone()).await; self.connect_to_peer(addr).await } + /// Record a join target address so the LAN beacon can unicast a dial-back + /// hint to it even before a direct connection forms. + /// + /// If a target with the same endpoint id is already recorded, its address + /// is replaced with the newer one. A peer that restarts or rebinds to a new + /// QUIC port advertises a fresh `EndpointAddr` under the same id, and the + /// beacon must dial that rather than keep unicasting to the stale socket. + async fn remember_join_target(&self, addr: EndpointAddr) { + let mut targets = self.join_targets.lock().await; + if let Some(existing) = targets.iter_mut().find(|t| t.id == addr.id) { + *existing = addr; + } else { + targets.push(addr); + } + } + + /// LAN IPv4 socket addresses of recorded join targets (from invite tokens), + /// used by the LAN beacon for dial-back unicast before peers are connected. + pub async fn join_target_lan_ipv4(&self) -> Vec { + let targets = self.join_targets.lock().await; + let mut out = Vec::new(); + for addr in targets.iter() { + out.extend(lan_bootstrap::lan_ipv4_candidates(addr)); + } + out + } + /// Like [`join`], but retries once after a delay on transient (connect/timeout) /// errors. Decode errors (invalid base64/JSON) fail immediately. pub async fn join_with_retry(&self, invite_token: &str) -> Result<()> { @@ -4840,6 +4962,7 @@ impl Node { // total budget which covers all but the worst relay conditions. let backoffs = [5, 10]; self.state.lock().await.dead_peers.remove(&addr.id); + self.remember_join_target(addr.clone()).await; let mut last_err = match self.connect_to_peer(addr.clone()).await { Ok(()) => return Ok(()), Err(e) => e, @@ -10363,6 +10486,7 @@ mod artifact_transfer_io; mod direct_path; mod gossip; mod heartbeat; +mod lan_bootstrap; mod owner_control_response; mod plugin_streams; pub(crate) mod requirements; diff --git a/crates/mesh-llm-host-runtime/src/mesh/tests.rs b/crates/mesh-llm-host-runtime/src/mesh/tests.rs index 8a5720410c..145c22c3be 100644 --- a/crates/mesh-llm-host-runtime/src/mesh/tests.rs +++ b/crates/mesh-llm-host-runtime/src/mesh/tests.rs @@ -395,6 +395,7 @@ async fn make_test_node_with_requirements( genesis_policy: Arc::new(Mutex::new(None)), signed_genesis_policy: Arc::new(Mutex::new(None)), bootstrap_token: Arc::new(Mutex::new(None)), + join_targets: Arc::new(Mutex::new(Vec::new())), first_joined_mesh_ts: Arc::new(Mutex::new(None)), accepting: Arc::new(( tokio::sync::Notify::new(), @@ -3152,6 +3153,142 @@ fn relay_reconnect_controller_applies_cooldown_after_attempt_and_prunes_gone_pee ); } +mod lan_join_target_tracking_tests { + use super::*; + + #[tokio::test] + async fn remember_join_target_updates_address_on_peer_rebind() { + let node = make_test_node(super::super::NodeRole::Worker) + .await + .unwrap(); + let peer_id = make_test_endpoint_id(34); + + let mut first = EndpointAddr { + id: peer_id, + addrs: Default::default(), + }; + first + .addrs + .insert(TransportAddr::Ip("192.168.1.50:47916".parse().unwrap())); + node.remember_join_target(first).await; + + assert_eq!( + node.join_target_lan_ipv4().await, + vec!["192.168.1.50:47916".parse().unwrap()], + "the first advertised LAN address should be recorded" + ); + + let mut rebound = EndpointAddr { + id: peer_id, + addrs: Default::default(), + }; + rebound + .addrs + .insert(TransportAddr::Ip("192.168.1.50:51000".parse().unwrap())); + node.remember_join_target(rebound).await; + + assert_eq!( + node.join_target_lan_ipv4().await, + vec!["192.168.1.50:51000".parse().unwrap()], + "a rebind under the same peer id must replace the stale dial-back address" + ); + } + + #[tokio::test] + async fn join_target_lan_ipv4_keeps_only_lan_addresses() { + let node = make_test_node(super::super::NodeRole::Worker) + .await + .unwrap(); + let peer_id = make_test_endpoint_id(35); + let mut target = EndpointAddr { + id: peer_id, + addrs: Default::default(), + }; + for addr in [ + "192.168.1.50:47916", + "8.8.8.8:47916", + "100.64.0.1:47916", + "127.0.0.1:47916", + "172.17.0.1:47916", + ] { + target + .addrs + .insert(TransportAddr::Ip(addr.parse().unwrap())); + } + node.remember_join_target(target).await; + + let lan_addrs: HashSet<_> = node + .join_target_lan_ipv4() + .await + .into_iter() + .map(|addr| addr.to_string()) + .collect(); + assert_eq!( + lan_addrs, + ["192.168.1.50:47916", "172.17.0.1:47916"] + .into_iter() + .map(str::to_owned) + .collect() + ); + } + + #[tokio::test] + async fn known_peer_lan_ipv4_keeps_only_lan_addresses() { + let node = make_test_node(super::super::NodeRole::Worker) + .await + .unwrap(); + let peer_id = make_test_endpoint_id(36); + let mut peer = make_test_peer_info(peer_id); + for addr in [ + "10.0.0.5:47916", + "203.0.113.5:47916", + "100.64.0.1:47916", + "172.17.0.1:47916", + ] { + peer.addr + .addrs + .insert(TransportAddr::Ip(addr.parse().unwrap())); + } + node.state.lock().await.peers.insert(peer_id, peer); + + let lan_addrs: HashSet<_> = node + .known_peer_lan_ipv4() + .await + .into_iter() + .map(|addr| addr.to_string()) + .collect(); + assert_eq!( + lan_addrs, + ["10.0.0.5:47916", "172.17.0.1:47916"] + .into_iter() + .map(str::to_owned) + .collect() + ); + } + + #[tokio::test] + async fn dial_peer_addr_clears_dead_peer_gate_before_connect() { + let node = make_test_node(super::super::NodeRole::Worker) + .await + .unwrap(); + let peer_id = make_test_endpoint_id(37); + node.state + .lock() + .await + .dead_peers + .insert(peer_id, std::time::Instant::now()); + + let _ = node + .dial_peer_addr(EndpointAddr { + id: peer_id, + addrs: Default::default(), + }) + .await; + + assert!(!node.state.lock().await.dead_peers.contains_key(&peer_id)); + } +} + #[test] fn stale_dispatcher_cannot_remove_replacement_connection() { assert!( diff --git a/crates/mesh-llm-host-runtime/src/network/discovery.rs b/crates/mesh-llm-host-runtime/src/network/discovery.rs index d7672f6505..797a05270b 100644 --- a/crates/mesh-llm-host-runtime/src/network/discovery.rs +++ b/crates/mesh-llm-host-runtime/src/network/discovery.rs @@ -42,6 +42,12 @@ pub(crate) struct LanMeshAdvertisement { pub(crate) proof_challenge: Option, pub(crate) app_version: Option, pub join_material: LanJoinMaterial, + /// Base64url-encoded JSON of the publisher's own [`iroh::EndpointAddr`], + /// filtered to its bound LAN interface. Additive (TXT key `ep_addr`): + /// older nodes ignore it. Lets a peer dial the publisher back directly, + /// which is the working direction when a multi-homed node cannot initiate + /// a relay-less direct connection itself. + pub(crate) endpoint_addr_b64: Option, } impl LanMeshAdvertisement { @@ -88,9 +94,25 @@ impl LanMeshAdvertisement { proof_challenge, app_version: app_version.map(str::to_owned), join_material: LanJoinMaterial::RequiresSuppliedToken, + endpoint_addr_b64: None, } } + /// Attach the publisher's own reachable [`EndpointAddr`] so peers can dial + /// it back directly (mDNS reverse-dial). Encoded as base64url JSON under the + /// additive `ep_addr` TXT key. + pub(crate) fn with_endpoint_addr(mut self, addr: &iroh::EndpointAddr) -> Self { + self.endpoint_addr_b64 = encode_endpoint_addr_b64(addr); + self + } + + /// Decode the publisher's advertised [`EndpointAddr`], if present and valid. + pub(crate) fn endpoint_addr(&self) -> Option { + self.endpoint_addr_b64 + .as_deref() + .and_then(decode_endpoint_addr_b64) + } + pub(crate) fn matches_supplied_token(&self, supplied_invite_token: Option<&str>) -> bool { let Some(expected) = self.token_fingerprint.as_deref() else { return false; @@ -122,6 +144,7 @@ impl LanMeshAdvertisement { push_optional_txt(&mut txt, "details", self.details_path.as_deref()); push_optional_txt(&mut txt, "proof_challenge", self.proof_challenge.as_deref()); push_optional_txt(&mut txt, "version", self.app_version.as_deref()); + push_optional_txt(&mut txt, "ep_addr", self.endpoint_addr_b64.as_deref()); for (key, value) in &txt { anyhow::ensure!( @@ -161,6 +184,7 @@ impl LanMeshAdvertisement { "details", "proof_challenge", "version", + "ep_addr", ] .into_iter() .filter_map(|key| service.get_property_val_str(key).map(|value| (key, value))) @@ -207,6 +231,10 @@ pub struct LanDiscoveredMesh { pub discovered_at: u64, #[serde(skip)] join_token: Option, + /// Publisher's own dial-back [`EndpointAddr`] (from the additive `ep_addr` + /// TXT key), if advertised. Used by mDNS reverse-dial. + #[serde(skip)] + endpoint_addr: Option, } impl LanDiscoveredMesh { @@ -214,6 +242,11 @@ impl LanDiscoveredMesh { self.join_token.as_deref() } + /// The publisher's advertised dial-back address, if present. + pub fn endpoint_addr(&self) -> Option<&iroh::EndpointAddr> { + self.endpoint_addr.as_ref() + } + pub(crate) fn to_join_candidate(&self) -> Option<(String, nostr::DiscoveredMesh)> { let token = self.join_token.clone()?; let mut listing = self.listing.clone(); @@ -287,7 +320,17 @@ impl LanDetailsResponse { } pub(crate) async fn publish_lan_loop(node: crate::mesh::Node, config: LanPublishConfig) { - let Some(daemon) = create_lan_publish_daemon(&config.status_tx) else { + // Restrict the mDNS daemon to the bound LAN interface when known. On + // multi-homed hosts (e.g. many utun/VPN interfaces) advertising on every + // interface can prevent the advertisement from reaching the LAN peers + // listen on. Pinning to the LAN address keeps mDNS on the same interface + // QUIC is bound to. + let lan_ip = node + .advertised_endpoint_addr() + .ip_addrs() + .map(|addr| addr.ip()) + .find(|ip| ip.is_ipv4()); + let Some(daemon) = create_lan_publish_daemon(&config.status_tx, lan_ip) else { return; }; @@ -317,9 +360,15 @@ pub(crate) async fn publish_lan_loop(node: crate::mesh::Node, config: LanPublish fn create_lan_publish_daemon( status_tx: &Option>>, + lan_ip: Option, ) -> Option { match ServiceDaemon::new() { - Ok(daemon) => Some(daemon), + Ok(daemon) => { + // When bound to a specific LAN interface, advertise only there so + // the advertisement reaches LAN peers on multi-homed hosts. + restrict_daemon_to_interface(&daemon, lan_ip); + Some(daemon) + } Err(err) => { tracing::warn!("Failed to create mDNS daemon: {err}"); let _ = send_publish_state(status_tx, nostr::PublishStateUpdate::PublishFailed); @@ -362,7 +411,8 @@ async fn publish_lan_advertisement(attempt: LanPublishAttempt<'_>) { Some(&listing.invite_token), Some(crate::VERSION), details_reachable, - ); + ) + .with_endpoint_addr(&node.advertised_endpoint_addr()); let Some(service_info) = encode_lan_service_info( &advert, instance_name, @@ -419,12 +469,56 @@ fn register_lan_service( } } +/// Restrict an mDNS daemon to only the interface owning `lan_ip`. +/// +/// `enable_interface` is additive on top of the default (all interfaces +/// enabled), so to actually pin to one interface we must first disable all, +/// then enable the LAN one. On multi-homed hosts (many utun/VPN interfaces) +/// this keeps mDNS traffic on the same interface QUIC is bound to, so +/// advertisements and queries reach LAN peers instead of being flooded onto +/// interfaces the peers cannot see. +fn restrict_daemon_to_interface(daemon: &ServiceDaemon, lan_ip: Option) { + // On multi-homed hosts (many utun/VPN interfaces) mdns-sd's default of + // advertising on every interface can mean the advertisement is multicast on + // an interface LAN peers cannot see, while the real LAN interface is starved + // or never picked. A raw `IP_MULTICAST_IF`-pinned socket on the LAN address + // reaches LAN peers reliably, so we pin the mDNS daemon to the LAN interface + // the same way: disable all interfaces, then re-enable just the LAN address. + // + // Selections apply in order with last-match-wins (see mdns-sd's + // `apply_intf_selections`), so the LAN `enable` after `disable(All)` keeps + // exactly that interface active. + let Some(ip) = lan_ip else { + return; + }; + if let Err(err) = daemon.disable_interface(mdns_sd::IfKind::All) { + tracing::debug!("mDNS: could not disable interfaces before pinning to {ip}: {err}"); + return; + } + if let Err(err) = daemon.enable_interface(mdns_sd::IfKind::Addr(ip)) { + tracing::debug!("mDNS: could not pin daemon to {ip}: {err}"); + } +} + pub async fn discover_lan( filter: &nostr::MeshFilter, supplied_invite_token: Option<&str>, timeout: Duration, +) -> Result> { + discover_lan_on_interface(filter, supplied_invite_token, timeout, None).await +} + +/// Like [`discover_lan`] but, when `lan_ip` is set, restricts the browse to the +/// matching interface. On multi-homed hosts this keeps mDNS on the same +/// interface QUIC is bound to so LAN advertisements are seen. +pub async fn discover_lan_on_interface( + filter: &nostr::MeshFilter, + supplied_invite_token: Option<&str>, + timeout: Duration, + lan_ip: Option, ) -> Result> { let daemon = ServiceDaemon::new().context("create mDNS daemon")?; + restrict_daemon_to_interface(&daemon, lan_ip); let receiver = match daemon.browse(LAN_SERVICE_TYPE) { Ok(receiver) => receiver, Err(err) => { @@ -696,6 +790,7 @@ fn lan_discovered_mesh( supplied_invite_token: Option<&str>, joinable: bool, ) -> LanDiscoveredMesh { + let endpoint_addr = advert.endpoint_addr(); LanDiscoveredMesh { mode: MeshDiscoveryMode::Mdns.as_str(), scope: MeshDiscoveryMode::Mdns.scope(), @@ -718,6 +813,7 @@ fn lan_discovered_mesh( published_version: advert.app_version, discovered_at: current_unix_secs(), join_token: joinable.then(|| supplied_invite_token.unwrap_or_default().to_string()), + endpoint_addr, } } @@ -800,11 +896,11 @@ fn lan_serving_node_count(peers: &[crate::mesh::PeerInfo]) -> usize { } async fn lan_instance_name(node: &crate::mesh::Node) -> String { - let identity = node - .mesh_id() - .await - .unwrap_or_else(|| node.id().fmt_short().to_string()); - let suffix = sanitize_dns_label(&identity); + // The mDNS instance name must be unique per node, not per mesh: every node + // in a mesh advertises its own record (carrying its own `ep_addr`), and two + // nodes sharing an instance name would clobber each other in mDNS, hiding + // peers from reverse-dial. Use the node's endpoint id, which is unique. + let suffix = sanitize_dns_label(&node.id().fmt_short().to_string()); format!("mesh-llm-{suffix}") } @@ -861,9 +957,29 @@ fn parse_txt_properties(props: &HashMap<&str, &str>) -> Result Option { + use base64::Engine; + let json = serde_json::to_vec(addr).ok()?; + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json); + // Key "ep_addr" (7) + value must stay under the DNS-SD 255 limit; keep margin. + (encoded.len() < TXT_VALUE_LIMIT).then_some(encoded) +} + +/// Decode a base64url-JSON [`iroh::EndpointAddr`] from an mDNS TXT value. +fn decode_endpoint_addr_b64(value: &str) -> Option { + use base64::Engine; + let raw = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(value) + .ok()?; + serde_json::from_slice(&raw).ok() +} + fn parse_txt_number(props: &HashMap<&str, &str>, key: &str) -> Result where T: std::str::FromStr, @@ -1063,6 +1179,42 @@ mod tests { ); } + #[test] + fn lan_advertisement_endpoint_addr_txt_round_trips() { + use iroh::{EndpointAddr, SecretKey}; + let secret = SecretKey::from_bytes(&[7u8; 32]); + let mut addr = EndpointAddr::from(secret.public()); + addr = addr.with_ip_addr("192.168.1.50:9555".parse().unwrap()); + + let listing = sample_listing("tok"); + let advert = + LanMeshAdvertisement::from_listing(&listing, Some("tok"), Some(crate::VERSION), false) + .with_endpoint_addr(&addr); + + let txt = advert.to_txt_properties().expect("txt should encode"); + assert!(txt.iter().any(|(k, _)| k == "ep_addr")); + + let decoded = LanMeshAdvertisement::from_txt_properties(&txt).expect("txt should decode"); + let decoded_addr = decoded.endpoint_addr().expect("ep_addr should decode"); + assert_eq!(decoded_addr.id, addr.id); + assert!( + decoded_addr + .ip_addrs() + .any(|a| a.to_string() == "192.168.1.50:9555") + ); + } + + #[test] + fn lan_advertisement_without_endpoint_addr_decodes_none() { + let listing = sample_listing("tok"); + let advert = + LanMeshAdvertisement::from_listing(&listing, Some("tok"), Some(crate::VERSION), false); + let txt = advert.to_txt_properties().expect("txt should encode"); + assert!(!txt.iter().any(|(k, _)| k == "ep_addr")); + let decoded = LanMeshAdvertisement::from_txt_properties(&txt).expect("txt should decode"); + assert!(decoded.endpoint_addr().is_none()); + } + #[test] fn lan_advertisement_exposes_token_gated_details_without_raw_invite_token() { let invite_token = "invite-token-for-details-proof"; diff --git a/crates/mesh-llm-host-runtime/src/network/lan_beacon.rs b/crates/mesh-llm-host-runtime/src/network/lan_beacon.rs new file mode 100644 index 0000000000..ac56d1a024 --- /dev/null +++ b/crates/mesh-llm-host-runtime/src/network/lan_beacon.rs @@ -0,0 +1,246 @@ +//! Raw-multicast LAN beacon for relay-less direct-path bootstrapping. +//! +//! On multi-homed hosts (many utun/VPN interfaces) both the mDNS service +//! daemon and iroh's relay-less initial handshake can fail to traverse the LAN, +//! because they rely on per-packet source selection that the macOS kernel +//! routes onto the wrong interface. A plain UDP socket *bound to the LAN IP* +//! with `IP_MULTICAST_IF` pinned to that interface reaches LAN peers reliably. +//! +//! This beacon uses exactly that reliable mechanism, independent of mDNS: +//! every node in mDNS mode periodically multicasts its own reachable +//! `EndpointAddr` (plus mesh id) on a dedicated group/port, and listens for +//! peers' beacons. On hearing a peer it is not connected to, it dials that +//! peer's advertised address — the single-homed → multi-homed direction that +//! works. `connect_to_peer` is idempotent, so whichever side connects first +//! wins and duplicates are harmless. +//! +//! The beacon carries no trust-bearing material: only an endpoint id, LAN +//! addresses, and a mesh-id fingerprint. Admission is still enforced by the +//! mesh handshake. A node only dials peers advertising the same mesh id (or an +//! unknown mesh id, to allow first contact before mesh ids converge). + +use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use socket2::{Domain, Protocol, Socket, Type}; +use tokio::net::UdpSocket; + +use crate::mesh; + +/// Dedicated multicast group + port for the LAN direct-path beacon. +/// +/// `224.0.0.251` is the IANA-assigned mDNS link-local multicast group. Reusing +/// that group on the distinct mesh-llm beacon port keeps packets on the local +/// segment without interacting with mDNS responders on 5353. +const BEACON_GROUP: Ipv4Addr = Ipv4Addr::new(224, 0, 0, 251); +const BEACON_PORT: u16 = 47654; +/// How often to emit our beacon. +const BEACON_INTERVAL: Duration = Duration::from_secs(5); +/// Beacon wire-format version, for forward compatibility. +const BEACON_VERSION: u8 = 1; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct BeaconMessage { + v: u8, + /// Publisher endpoint id (canonical string). + id: String, + /// Publisher mesh id, if known. + mesh_id: Option, + /// Base64url-JSON of the publisher's `EndpointAddr` (LAN-filtered). + addr: String, +} + +/// Spawn the LAN beacon (sender + listener) for a node in mDNS mode. +/// +/// Returns the spawned task's [`JoinHandle`](tokio::task::JoinHandle) so the +/// runtime can abort the beacon (and release its UDP socket) during shutdown. +pub(crate) fn spawn(node: mesh::Node) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + if let Err(err) = run(node).await { + tracing::debug!("LAN beacon stopped: {err:#}"); + } + }) +} + +async fn run(node: mesh::Node) -> Result<()> { + let recv_sock = bind_beacon_listener().context("bind LAN beacon listener")?; + let mut buf = vec![0u8; 4096]; + let mut send_tick = tokio::time::interval(BEACON_INTERVAL); + + tracing::debug!( + "LAN beacon active on {}:{} (self={})", + BEACON_GROUP, + BEACON_PORT, + node.id().fmt_short() + ); + + loop { + tokio::select! { + _ = send_tick.tick() => on_send_tick(&node).await, + res = recv_sock.recv_from(&mut buf) => on_recv(&node, res, &buf).await, + } + } +} + +async fn on_send_tick(node: &mesh::Node) { + if let Err(err) = emit_beacon(node).await { + tracing::trace!("LAN beacon emit failed: {err:#}"); + } +} + +async fn on_recv(node: &mesh::Node, res: std::io::Result<(usize, SocketAddr)>, buf: &[u8]) { + match res { + Ok((n, _from)) => handle_beacon(node, &buf[..n]).await, + Err(err) => tracing::trace!("LAN beacon recv error: {err}"), + } +} + +/// Bind a multicast listener socket joined on all relevant interfaces. +fn bind_beacon_listener() -> Result { + let sock = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?; + sock.set_reuse_address(true)?; + #[cfg(unix)] + sock.set_reuse_port(true)?; + sock.bind(&SocketAddr::from((Ipv4Addr::UNSPECIFIED, BEACON_PORT)).into())?; + // Join the group on the unspecified interface; the kernel joins on the + // default interface, which is sufficient for receiving on the LAN. + sock.join_multicast_v4(&BEACON_GROUP, &Ipv4Addr::UNSPECIFIED)?; + sock.set_nonblocking(true)?; + let std_sock: std::net::UdpSocket = sock.into(); + Ok(UdpSocket::from_std(std_sock)?) +} + +/// Emit our beacon: multicast (best effort) plus a direct unicast to every +/// known peer's LAN address. +/// +/// On multi-homed macOS hosts an in-process multicast send can fail with +/// EHOSTUNREACH even though the route table is correct, while a plain unicast to +/// a known LAN address routes fine. So the unicast path is the reliable carrier: +/// a joiner already knows the host's address (from the invite token / gossip), +/// so it can unicast its own `EndpointAddr` straight to the host, which then +/// dials back on the working direction. +async fn emit_beacon(node: &mesh::Node) -> Result<()> { + let addr = node.advertised_endpoint_addr(); + let has_v4 = addr + .ip_addrs() + .any(|a| matches!(a.ip(), IpAddr::V4(v4) if !v4.is_loopback() && !v4.is_unspecified())); + if !has_v4 { + return Ok(()); + } + + let msg = BeaconMessage { + v: BEACON_VERSION, + id: node.id().to_string(), + mesh_id: node.mesh_id().await, + addr: encode_endpoint_addr(&addr).context("encode endpoint addr")?, + }; + let payload = serde_json::to_vec(&msg)?; + let mcast = SocketAddrV4::new(BEACON_GROUP, BEACON_PORT); + // Unicast to known peers and to join targets (invite-token addresses we may + // not have connected to yet — the key case for a multi-homed joiner that + // cannot complete its own outbound QUIC handshake). + let mut peers = node.known_peer_lan_ipv4().await; + peers.extend(node.join_target_lan_ipv4().await); + peers.sort(); + peers.dedup(); + // Beacon to the peer's beacon port, not its QUIC port. + for p in peers.iter_mut() { + p.set_port(BEACON_PORT); + } + + if let Err(err) = + tokio::task::spawn_blocking(move || emit_blocking(mcast, &peers, &payload)).await + { + tracing::warn!(%err, "LAN beacon emit task failed"); + } + Ok(()) +} + +/// Send the beacon synchronously: best-effort multicast plus unicast to each +/// known peer LAN address, all on plain unbound sockets (no interface pins, +/// which trigger in-process EHOSTUNREACH on multi-homed macOS hosts). +fn emit_blocking(mcast: SocketAddrV4, peers: &[SocketAddrV4], payload: &[u8]) { + if let Err(err) = send_multicast(mcast, payload) { + tracing::trace!("LAN beacon multicast failed: {err}"); + } + // Each send opens a short-lived socket: beacon traffic is tiny, and avoiding + // shared multicast socket state keeps interface pins out of unicast sends on + // multi-homed hosts. + for peer in peers { + let res = send_unicast(*peer, payload); + tracing::trace!("LAN beacon unicast to {peer}: {res:?}"); + } +} + +fn send_multicast(dst: SocketAddrV4, payload: &[u8]) -> Result<()> { + let sock = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?; + sock.set_multicast_ttl_v4(1)?; + sock.send_to(payload, &SocketAddr::V4(dst).into())?; + Ok(()) +} + +fn send_unicast(dst: SocketAddrV4, payload: &[u8]) -> Result<()> { + let sock = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?; + sock.send_to(payload, &SocketAddr::V4(dst).into())?; + Ok(()) +} + +/// Handle a received beacon: dial the peer back if appropriate. +async fn handle_beacon(node: &mesh::Node, payload: &[u8]) { + let Some((peer_id, addr)) = parse_beacon(node, payload).await else { + return; + }; + if node.connected_peer_ids().await.contains(&peer_id) { + return; + } + tracing::info!( + "LAN beacon: dialing peer {} on advertised LAN address", + peer_id.fmt_short() + ); + if let Err(err) = node.dial_peer_addr(addr).await { + tracing::debug!( + "LAN beacon dial to {} failed (will retry): {err}", + peer_id.fmt_short() + ); + } +} + +/// Validate and decode a beacon into a dialable peer, applying mesh-id and +/// self filtering. Returns `None` if the beacon should be ignored. +async fn parse_beacon( + node: &mesh::Node, + payload: &[u8], +) -> Option<(iroh::EndpointId, iroh::EndpointAddr)> { + let msg: BeaconMessage = serde_json::from_slice(payload).ok()?; + if msg.v != BEACON_VERSION { + return None; + } + let addr = decode_endpoint_addr(&msg.addr)?; + if addr.id == node.id() { + return None; + } + // Only dial peers in our mesh. Allow unknown/absent mesh ids so the first + // contact can happen before mesh ids are exchanged. + if let (Some(ours), Some(theirs)) = (node.mesh_id().await, msg.mesh_id.as_ref()) + && &ours != theirs + { + return None; + } + Some((addr.id, addr)) +} + +fn encode_endpoint_addr(addr: &iroh::EndpointAddr) -> Option { + use base64::Engine; + let json = serde_json::to_vec(addr).ok()?; + Some(base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json)) +} + +fn decode_endpoint_addr(value: &str) -> Option { + use base64::Engine; + let raw = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(value) + .ok()?; + serde_json::from_slice(&raw).ok() +} diff --git a/crates/mesh-llm-host-runtime/src/network/lan_bootstrap.rs b/crates/mesh-llm-host-runtime/src/network/lan_bootstrap.rs new file mode 100644 index 0000000000..8bb3aa0033 --- /dev/null +++ b/crates/mesh-llm-host-runtime/src/network/lan_bootstrap.rs @@ -0,0 +1,82 @@ +use crate::mesh; +use crate::network::discovery as mesh_discovery; +use crate::runtime::RuntimeOptions; +use std::net::IpAddr; +use tokio::task::JoinHandle; + +pub(crate) fn effective_quic_bind_ip(options: &RuntimeOptions) -> Option { + if let Some(ip) = options.bind_ip { + return Some(ip); + } + + let detected = mesh::detect_primary_lan_ipv4(); + if let Some(ip) = detected { + tracing::info!( + "Auto-binding QUIC endpoint to detected LAN address {ip}; override with --bind-ip" + ); + Some(ip) + } else { + tracing::debug!( + "Unable to detect a LAN IPv4 address for QUIC bind; using wildcard socket bind" + ); + None + } +} + +/// Background tasks spawned by [`spawn_mdns_reverse_dial`] for relay-less LAN +/// direct-path bootstrap. Dropping this guard aborts those loops. +#[derive(Default)] +pub(crate) struct LanBootstrapTasks { + handles: Vec>, +} + +impl LanBootstrapTasks { + pub(crate) fn abort(&self) { + for handle in &self.handles { + handle.abort(); + } + } +} + +impl Drop for LanBootstrapTasks { + fn drop(&mut self) { + self.abort(); + } +} + +pub(crate) fn spawn_mdns_reverse_dial( + options: &RuntimeOptions, + node: &mesh::Node, +) -> LanBootstrapTasks { + if options.mesh_discovery_mode != mesh_discovery::MeshDiscoveryMode::Mdns { + return LanBootstrapTasks::default(); + } + + let mut handles = Vec::new(); + + if !options.publish { + handles.push(tokio::spawn(Box::pin(mesh_discovery::publish_lan_loop( + node.clone(), + mesh_discovery::LanPublishConfig { + name: options.mesh_name.clone(), + region: options.region.clone(), + max_clients: options.max_clients, + api_port: options.console, + details_reachable: options.listen_all, + interval_secs: 30, + status_tx: None, + }, + )))); + } + + handles.push(tokio::spawn(Box::pin( + crate::network::mdns_reverse_dial::run_loop( + node.clone(), + options.mesh_name.clone(), + options.region.clone(), + ), + ))); + handles.push(crate::network::lan_beacon::spawn(node.clone())); + + LanBootstrapTasks { handles } +} diff --git a/crates/mesh-llm-host-runtime/src/network/mdns_reverse_dial.rs b/crates/mesh-llm-host-runtime/src/network/mdns_reverse_dial.rs new file mode 100644 index 0000000000..eb12cdca07 --- /dev/null +++ b/crates/mesh-llm-host-runtime/src/network/mdns_reverse_dial.rs @@ -0,0 +1,119 @@ +//! mDNS reverse-dial: bounded LAN dial-back for relay-less direct paths. +//! +//! In relay-less (mDNS) mode a direct connection is established by the joiner +//! dialing the host. On a multi-homed host (many interfaces, e.g. VPN/utun) the +//! joiner's own QUIC initiator path can fail to traverse even though the OS +//! network path and addresses are correct, leaving the connection stuck. +//! +//! The opposite direction works reliably: a single-homed peer dialing the +//! multi-homed peer establishes a clean LAN direct path. This loop exploits +//! that: every node in mDNS mode publishes its own reachable `EndpointAddr` in +//! its mDNS advert (additive `ep_addr` TXT key), and every node periodically +//! browses the LAN and dials back any advertised peer it is not already +//! connected to. Whichever direction succeeds first wins; `connect_to_peer` +//! is idempotent and skips peers that are already connected. + +use std::collections::HashSet; +use std::time::Duration; + +use crate::mesh; +use crate::network::discovery as mesh_discovery; +use crate::network::nostr; + +/// How often to browse the LAN and attempt reverse-dials. +const REVERSE_DIAL_INTERVAL: Duration = Duration::from_secs(10); +/// How long each browse is allowed to collect advertisements. +const BROWSE_TIMEOUT: Duration = Duration::from_secs(3); + +/// Runs the mDNS reverse-dial loop until the node shuts down. +/// +/// Bounded: one browse per tick, at most one dial attempt per discovered peer +/// per tick, and only for peers not already connected. Safe to run on both the +/// host and the joiner — the idempotent connect makes double-dialing harmless. +pub(crate) async fn run_loop(node: mesh::Node, mesh_name: Option, region: Option) { + let self_id = node.id(); + tracing::debug!( + "mDNS reverse-dial loop started (self={})", + self_id.fmt_short() + ); + loop { + tokio::time::sleep(REVERSE_DIAL_INTERVAL).await; + reverse_dial_tick(&node, self_id, mesh_name.as_deref(), region.as_deref()).await; + } +} + +async fn reverse_dial_tick( + node: &mesh::Node, + self_id: iroh::EndpointId, + mesh_name: Option<&str>, + region: Option<&str>, +) { + let discovered = browse_lan(node, mesh_name, region).await; + let connected: HashSet = node.connected_peer_ids().await; + + for mesh_advert in &discovered { + if let Some(addr) = dial_target(mesh_advert, self_id, &connected) { + dial_back(node, addr).await; + } + } +} + +/// Browse the LAN for mesh advertisements, pinned to the node's bound LAN +/// interface. Returns an empty list on error. +async fn browse_lan( + node: &mesh::Node, + mesh_name: Option<&str>, + region: Option<&str>, +) -> Vec { + let filter = nostr::MeshFilter { + name: mesh_name.map(str::to_string), + region: region.map(str::to_string), + ..Default::default() + }; + let lan_ip = node + .advertised_endpoint_addr() + .ip_addrs() + .map(|addr| addr.ip()) + .find(|ip| ip.is_ipv4()); + + match mesh_discovery::discover_lan_on_interface(&filter, None, BROWSE_TIMEOUT, lan_ip).await { + Ok(meshes) => { + tracing::debug!( + "mDNS reverse-dial browse (lan_ip={lan_ip:?}) found {} advert(s)", + meshes.len() + ); + meshes + } + Err(err) => { + tracing::debug!("mDNS reverse-dial browse failed: {err}"); + Vec::new() + } + } +} + +/// Returns the peer's advertised dial-back address if it is a new peer worth +/// dialing (not ourselves, not already connected, and carrying an `ep_addr`). +fn dial_target( + mesh_advert: &mesh_discovery::LanDiscoveredMesh, + self_id: iroh::EndpointId, + connected: &HashSet, +) -> Option { + let addr = mesh_advert.endpoint_addr()?; + if addr.id == self_id || connected.contains(&addr.id) { + return None; + } + Some(addr.clone()) +} + +async fn dial_back(node: &mesh::Node, addr: iroh::EndpointAddr) { + tracing::info!( + "mDNS reverse-dial: dialing peer {} on advertised LAN address", + addr.id.fmt_short() + ); + if let Err(err) = node.dial_peer_addr(addr.clone()).await { + tracing::debug!( + "mDNS reverse-dial to {} failed (will retry): {err}", + addr.id.fmt_short() + ); + } +} diff --git a/crates/mesh-llm-host-runtime/src/network/mod.rs b/crates/mesh-llm-host-runtime/src/network/mod.rs index 107402aa40..a8e8b0c4ea 100644 --- a/crates/mesh-llm-host-runtime/src/network/mod.rs +++ b/crates/mesh-llm-host-runtime/src/network/mod.rs @@ -1,5 +1,8 @@ pub(crate) mod affinity; pub(crate) mod discovery; +pub(crate) mod lan_beacon; +pub(crate) mod lan_bootstrap; +pub(crate) mod mdns_reverse_dial; pub(crate) mod metrics; pub(crate) mod nostr; pub(crate) mod openai; diff --git a/crates/mesh-llm-host-runtime/src/runtime/mod.rs b/crates/mesh-llm-host-runtime/src/runtime/mod.rs index 4d3bb19843..b95fee0faa 100644 --- a/crates/mesh-llm-host-runtime/src/runtime/mod.rs +++ b/crates/mesh-llm-host-runtime/src/runtime/mod.rs @@ -52,7 +52,11 @@ use crate::inference::{election, skippy}; use crate::mesh; use crate::mesh::NodeRole; use crate::models; -use crate::network::{affinity, discovery as mesh_discovery, nostr, tunnel}; +use crate::network::{ + affinity, discovery as mesh_discovery, + lan_bootstrap::{LanBootstrapTasks, effective_quic_bind_ip, spawn_mdns_reverse_dial}, + nostr, tunnel, +}; use crate::plugin; use crate::system::{autoupdate, backend, benchmark, hardware}; use anyhow::{Context, Result}; @@ -134,6 +138,7 @@ struct AutoRuntimeNodeSetup { channels: mesh::TunnelChannels, plugin_manager: plugin::PluginManager, survey_telemetry: survey::SurveyTelemetry, + lan_bootstrap_tasks: LanBootstrapTasks, } #[derive(Default)] @@ -5778,7 +5783,7 @@ pub(crate) async fn run_plugin_mcp(options: &RuntimeOptions) -> Result<()> { policy: relay_policy_for_runtime_options(options), }, mesh::QuicBindSelection { - ip: options.bind_ip, + ip: effective_quic_bind_ip(options), port: options.bind_port, }, Some(0.0), @@ -6023,7 +6028,7 @@ async fn start_run_auto_node_and_plugins( policy: relay_policy_for_runtime_options(options), }, mesh::QuicBindSelection { - ip: options.bind_ip, + ip: effective_quic_bind_ip(options), port: options.bind_port, }, max_vram, @@ -6165,6 +6170,7 @@ async fn build_run_auto_node_setup( node.start_rtt_refresh(); node.start_direct_path_maintenance(); start_relay_health_monitor_for_discovery_mode(&node, options.mesh_discovery_mode); + let lan_bootstrap_tasks = spawn_mdns_reverse_dial(options, &node); if !is_client { spawn_node_benchmark_task(&node, bin_dir); @@ -6181,6 +6187,7 @@ async fn build_run_auto_node_setup( channels, plugin_manager, survey_telemetry, + lan_bootstrap_tasks, }) } @@ -6624,6 +6631,7 @@ struct RunAutoShutdownContext<'a> { api_proxy_handle: tokio::task::JoinHandle<()>, console_server_handle: Option>, discovery_publisher: Option>, + lan_bootstrap_tasks: LanBootstrapTasks, runtime_models: &'a mut HashMap, runtime_survey_models: &'a mut HashMap, managed_models: &'a mut HashMap, @@ -6656,6 +6664,7 @@ struct RunAutoRuntimeLifecycleContext<'a> { api_proxy_handle: tokio::task::JoinHandle<()>, console_server_handle: Option>, discovery_publisher: Option>, + lan_bootstrap_tasks: LanBootstrapTasks, runtime: Option>, } @@ -6972,6 +6981,7 @@ async fn run_auto_runtime_loop_and_shutdown(ctx: RunAutoRuntimeLifecycleContext< api_proxy_handle, console_server_handle, discovery_publisher, + lan_bootstrap_tasks, runtime, } = ctx; let mut loop_ctx = RunAutoRuntimeLoopContext { @@ -7007,6 +7017,7 @@ async fn run_auto_runtime_loop_and_shutdown(ctx: RunAutoRuntimeLifecycleContext< api_proxy_handle, console_server_handle, discovery_publisher, + lan_bootstrap_tasks, runtime_models: &mut runtime_state.runtime_models, runtime_survey_models: &mut runtime_state.runtime_survey_models, managed_models: &mut runtime_state.managed_models, @@ -7030,6 +7041,7 @@ async fn shutdown_run_auto_runtime(ctx: RunAutoShutdownContext<'_>) { api_proxy_handle, console_server_handle, discovery_publisher, + lan_bootstrap_tasks, runtime_models, runtime_survey_models, managed_models, @@ -7048,6 +7060,9 @@ async fn shutdown_run_auto_runtime(ctx: RunAutoShutdownContext<'_>) { if let Some(handle) = discovery_publisher { handle.abort(); } + // Stop the relay-less LAN bootstrap loops (mDNS publisher, reverse-dial, + // and beacon) so they release their sockets and stop dialing on shutdown. + lan_bootstrap_tasks.abort(); shutdown_run_auto_services( node, @@ -8335,6 +8350,7 @@ async fn run_auto(ctx: RunAutoContext) -> Result<()> { channels, plugin_manager, survey_telemetry, + lan_bootstrap_tasks, } = build_run_auto_node_setup( &options, &config, @@ -8516,6 +8532,7 @@ async fn run_auto(ctx: RunAutoContext) -> Result<()> { api_proxy_handle, console_server_handle, discovery_publisher, + lan_bootstrap_tasks, runtime, }) .await;