feat(discovery): platform-pushed peer queue

Add a transport-agnostic seam for an embedding platform (e.g. an Android
Wi-Fi Aware radio) to push "peer npub reachable at addr over transport T"
events into a running node — the generalization of the UDP-only LAN mDNS
drain. A process-global queue (fips::discovery::platform) is drained each
tick by poll_platform_discovery, which selects the transport family-aware
(an IPv6 target picks an IPv6 socket) and initiates a Noise IK handshake;
the pushed npub is only a routing hint, the handshake authenticates. An
event for an already-active peer starts an alternate-path handshake, and
a Lost event closes the pooled connection.

src/discovery.rs

@@ -8,6 +8,7 @@
 
 pub mod lan;
 pub mod nostr;
+pub mod platform;
 
 use crate::config::UdpConfig;
 use crate::{NodeAddr, TransportId};

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<VecDeque<PlatformPeerEvent>> = 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<PlatformPeerEvent> {
+    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:?}"),
+        }
+    }
+}

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;

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<TransportId> {
+        if transport_type == "udp"
+            && let Ok(remote_addr) = addr.parse::<SocketAddr>()
+        {
+            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(&current_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::<SocketAddr>()
-                {
-                    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) {