diff --git a/src/discovery.rs b/src/discovery.rs index 3612fb5..c44d986 100644 --- a/src/discovery.rs +++ b/src/discovery.rs @@ -8,6 +8,7 @@ pub mod lan; pub mod nostr; +pub mod platform; use crate::config::UdpConfig; use crate::{NodeAddr, TransportId}; diff --git a/src/discovery/platform.rs b/src/discovery/platform.rs new file mode 100644 index 0000000..8f73719 --- /dev/null +++ b/src/discovery/platform.rs @@ -0,0 +1,141 @@ +//! Platform-pushed peer discovery. +//! +//! A generic seam for an embedding platform (e.g. an Android app layer that +//! runs its own radio discovery, such as Wi-Fi Aware) to push "peer `npub` is +//! reachable at `addr` over transport type `T`" events into a running node — +//! the transport-agnostic generalization of the LAN mDNS drain +//! (`poll_lan_discovery`), which delivers the same shape but is hardwired to +//! UDP transports. +//! +//! The queue is a process-global, like the Android BLE bridge injection seam +//! (`set_android_ble_bridge`): the embedder pushes without holding a `Node` +//! handle, and the node drains once per tick in `poll_platform_discovery`. +//! Events pushed while no node is running are retained up to [`QUEUE_CAP`] +//! (oldest dropped first) so a push racing a node rebuild is not lost. +//! With more than one node in a process, whichever drains first consumes +//! the events (same caveat as the BLE bridge) — intended for the +//! single-node embedding case. +//! +//! The pushed npub is only a routing hint: the Noise IK handshake is the +//! authentication, exactly as with mDNS adverts — a spoofed push fails the +//! IK exchange and is dropped. + +use std::collections::VecDeque; +use std::sync::Mutex; + +/// Maximum retained events while undrained. Beyond this the oldest event is +/// dropped: platform pushes are periodic (radio discovery re-fires), so a +/// dropped event is re-learned, while an unbounded queue would grow forever +/// if the node is stopped. +const QUEUE_CAP: usize = 256; + +/// A peer reachability event pushed by the embedding platform. +/// +/// Addresses and identities are strings at this seam (it is crossed from +/// JNI); they are parsed and validated at drain time, where a bad value is +/// logged and skipped rather than surfaced to the pusher. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum PlatformPeerEvent { + /// The platform established reachability: dial `addr` on an operational + /// transport whose type name matches `transport_type`. For `udp` the + /// selection is family-aware — an IPv6 target picks an IPv6-capable + /// socket, never a wildcard IPv4 one. For IPv6 link-local addresses the + /// scope must be a numeric ifindex (`"[fe80::x%3]:4870"`) — + /// interface-name scopes do not parse. + Available { + npub: String, + addr: String, + transport_type: String, + }, + /// The platform observed the link go away (e.g. the Wi-Fi Aware data + /// path was lost). The node closes any pooled connection it holds for + /// the peer's current address on that transport so a dead socket is + /// not re-used; reconnection is left to the ordinary machinery. + Lost { + npub: String, + transport_type: String, + }, +} + +static QUEUE: Mutex> = Mutex::new(VecDeque::new()); + +fn push(event: PlatformPeerEvent) { + let mut queue = QUEUE.lock().unwrap_or_else(|e| e.into_inner()); + if queue.len() >= QUEUE_CAP { + queue.pop_front(); + } + queue.push_back(event); +} + +/// Push "peer is reachable at `addr` over `transport_type`". +pub fn platform_peer_available(npub: &str, addr: &str, transport_type: &str) { + push(PlatformPeerEvent::Available { + npub: npub.to_string(), + addr: addr.to_string(), + transport_type: transport_type.to_string(), + }); +} + +/// Push "the platform-managed link to peer went away". +pub fn platform_peer_lost(npub: &str, transport_type: &str) { + push(PlatformPeerEvent::Lost { + npub: npub.to_string(), + transport_type: transport_type.to_string(), + }); +} + +/// Drain all queued events. Called by the node once per tick. +pub fn drain_platform_peer_events() -> Vec { + let mut queue = QUEUE.lock().unwrap_or_else(|e| e.into_inner()); + queue.drain(..).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The queue is a process-global, so tests touching it must not + /// interleave across test threads. + static TEST_LOCK: Mutex<()> = Mutex::new(()); + + #[test] + fn push_drain_roundtrip() { + let _guard = TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + drain_platform_peer_events(); + platform_peer_available("npub1abc", "[fe80::1%3]:4870", "tcp"); + platform_peer_lost("npub1abc", "tcp"); + let events = drain_platform_peer_events(); + assert_eq!(events.len(), 2); + assert_eq!( + events[0], + PlatformPeerEvent::Available { + npub: "npub1abc".into(), + addr: "[fe80::1%3]:4870".into(), + transport_type: "tcp".into(), + } + ); + assert_eq!( + events[1], + PlatformPeerEvent::Lost { + npub: "npub1abc".into(), + transport_type: "tcp".into(), + } + ); + assert!(drain_platform_peer_events().is_empty()); + } + + #[test] + fn queue_caps_by_dropping_oldest() { + let _guard = TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + drain_platform_peer_events(); + for i in 0..(QUEUE_CAP + 10) { + platform_peer_available(&format!("npub{i}"), "addr", "tcp"); + } + let events = drain_platform_peer_events(); + assert_eq!(events.len(), QUEUE_CAP); + match &events[0] { + PlatformPeerEvent::Available { npub, .. } => assert_eq!(npub, "npub10"), + other => panic!("unexpected event: {other:?}"), + } + } +} diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 525550f..bb8d243 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -264,6 +264,7 @@ impl Node { self.poll_pending_connects().await; self.poll_nostr_discovery().await; self.poll_lan_discovery().await; + self.poll_platform_discovery().await; self.resend_pending_handshakes(now_ms).await; self.resend_pending_rekeys(now_ms).await; self.resend_pending_session_handshakes(now_ms).await; diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index bd577a9..4bd5e62 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -367,6 +367,27 @@ impl Node { .min_by_key(|(id, _)| id.as_u32()) } + /// Resolve a discovered `(transport_type, addr)` pair to an operational + /// transport. For `udp` addresses the socket family matters — a wildcard + /// IPv4 socket cannot send to an IPv6 link-local target (e.g. a Wi-Fi + /// Aware data path) — so when the address parses as a socket address the + /// selection is family-aware and skips bootstrap-adopted sockets. All + /// other transport types match by name alone. + fn find_transport_for_discovered_addr( + &self, + transport_type: &str, + addr: &str, + ) -> Option { + if transport_type == "udp" + && let Ok(remote_addr) = addr.parse::() + { + return self + .find_udp_transport_for_remote_addr(remote_addr) + .map(|(id, _)| id); + } + self.find_transport_for_type(transport_type) + } + /// Initiate a connection to a peer on a specific transport and address. /// /// For connectionless transports (UDP, Ethernet): allocates a link, starts @@ -956,6 +977,139 @@ impl Node { } } + /// Drain platform-pushed peers and initiate Noise IK handshakes. + /// + /// The transport-agnostic sibling of `poll_lan_discovery`: an embedding + /// platform (e.g. the Android Wi-Fi Aware radio) pushes + /// `(npub, addr, transport type)` events into the process-global queue + /// (`crate::discovery::platform`) and this drains them once per tick. + /// As with mDNS, the pushed npub is only a hint — the IK handshake is + /// the authentication. + /// + /// Unlike the LAN drain, an event for an already-active peer starts an + /// alternate-path handshake (gated by the same freshness/in-flight + /// checks as `poll_transport_discovery`), so a platform push can move a + /// peer onto a faster transport — the BLE→Wi-Fi-Aware cutover. + pub(super) async fn poll_platform_discovery(&mut self) { + let events = crate::discovery::platform::drain_platform_peer_events(); + if events.is_empty() { + return; + } + let mut connect_budget = self.discovery_connect_budget(); + for event in events { + match event { + crate::discovery::platform::PlatformPeerEvent::Available { + npub, + addr, + transport_type, + } => { + let Some(transport_id) = + self.find_transport_for_discovered_addr(&transport_type, &addr) + else { + debug!( + npub = %npub, + transport_type = %transport_type, + addr = %addr, + "platform: skip pushed peer with no compatible operational transport" + ); + continue; + }; + let identity = match crate::PeerIdentity::from_npub(&npub) { + Ok(id) => id, + Err(err) => { + debug!(npub = %npub, error = %err, "platform: skip bad npub"); + continue; + } + }; + let peer_node_addr = *identity.node_addr(); + if peer_node_addr == *self.identity().node_addr() { + continue; + } + let remote_addr = crate::transport::TransportAddr::from_string(&addr); + + if self.peers.contains_key(&peer_node_addr) { + // Active peer: this is a path upgrade, not a first + // contact — apply the alternate-path gates. + let candidate = PeerAddress::new(&transport_type, addr.clone()); + if self.active_peer_candidate_is_fresh_enough_to_skip( + &peer_node_addr, + std::slice::from_ref(&candidate), + ) { + continue; + } + } + if self.is_connecting_to_peer_on_path( + &peer_node_addr, + transport_id, + &remote_addr, + ) { + continue; + } + if connect_budget == 0 { + debug!(npub = %npub, "platform: connect budget exhausted"); + continue; + } + connect_budget = connect_budget.saturating_sub(1); + info!( + peer = %self.peer_display_name(&peer_node_addr), + transport_id = %transport_id, + remote_addr = %remote_addr, + "platform: initiating handshake to pushed peer" + ); + if let Err(err) = self + .initiate_connection(transport_id, remote_addr, identity) + .await + { + debug!( + npub = %npub, + error = %err, + "platform: failed to initiate connection to pushed peer" + ); + } + } + crate::discovery::platform::PlatformPeerEvent::Lost { + npub, + transport_type, + } => { + let Ok(identity) = crate::PeerIdentity::from_npub(&npub) else { + continue; + }; + let peer_node_addr = *identity.node_addr(); + let Some(peer) = self.peers.get(&peer_node_addr) else { + continue; + }; + // Only act if the peer currently sits on the named + // transport type: close the pooled connection so the + // dead socket is not re-used. Reconnection (including + // falling back to another transport) is the ordinary + // machinery's job. + let (Some(transport_id), Some(current_addr)) = + (peer.transport_id(), peer.current_addr().cloned()) + else { + continue; + }; + let on_named_transport = self + .transports + .get(&transport_id) + .map(|t| t.transport_type().name == transport_type) + .unwrap_or(false); + if !on_named_transport { + continue; + } + info!( + peer = %self.peer_display_name(&peer_node_addr), + transport_id = %transport_id, + remote_addr = %current_addr, + "platform: closing connection for lost pushed peer" + ); + if let Some(transport) = self.transports.get(&transport_id) { + transport.close_connection(¤t_addr).await; + } + } + } + } + } + /// Poll pending transport connects and initiate handshakes for ready ones. /// /// Called from the tick handler. For each pending connect, queries the @@ -1769,34 +1923,17 @@ impl Node { continue; } } else { - let tid = if addr.transport == "udp" - && let Ok(remote_socket_addr) = addr.addr.parse::() - { - match self.find_udp_transport_for_remote_addr(remote_socket_addr) { - Some((id, _)) => id, - None => { - debug!( - transport = %addr.transport, - addr = %addr.addr, - "No compatible operational UDP transport for address" - ); - continue; - } + match self.find_transport_for_discovered_addr(&addr.transport, &addr.addr) { + Some(tid) => (tid, TransportAddr::from_string(&addr.addr)), + None => { + debug!( + transport = %addr.transport, + addr = %addr.addr, + "No compatible operational transport for address" + ); + continue; } - } else { - match self.find_transport_for_type(&addr.transport) { - Some(id) => id, - None => { - debug!( - transport = %addr.transport, - addr = %addr.addr, - "No operational transport for address type" - ); - continue; - } - } - }; - (tid, TransportAddr::from_string(&addr.addr)) + } }; if self.is_connecting_to_peer_on_path(&peer_node_addr, transport_id, &remote_addr) {