diff --git a/Cargo.lock b/Cargo.lock index 1f600fb..1395c3f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9918,7 +9918,6 @@ dependencies = [ "gaol", "ipc-channel", "iroh", - "iroh-mdns-address-lookup", "iroh-persist", "keyboard-types 0.8.3", "log", @@ -9967,7 +9966,6 @@ version = "0.6.0" dependencies = [ "base64 0.23.1", "beaver-hal", - "beaver-p2p", "content-security-policy", "encoding_rs", "euclid", diff --git a/crates/beaver_p2p/src/discovery/mdns.rs b/crates/beaver_p2p/src/discovery/mdns.rs new file mode 100644 index 0000000..0b83ef3 --- /dev/null +++ b/crates/beaver_p2p/src/discovery/mdns.rs @@ -0,0 +1,44 @@ +/* SPDX Id: AGPL-3.0-or-later */ + +//! The mDNS discovery source: an adapter over `iroh-mdns-address-lookup`. +//! +//! This is the only place in Beaver that names that crate's types. Everything +//! above it sees [`DiscoveryEvent`] instead. + +use iroh_mdns_address_lookup::{DiscoveryEvent as MdnsEvent, MdnsAddressLookup}; +use n0_future::StreamExt; +use n0_future::boxed::BoxStream; + +use crate::discovery::{DiscoveredPeer, DiscoveryEvent, DiscoveryTransport}; + +/// Subscribe to mDNS. Infallible: the lookup is already bound by the time we get +/// here, so there is nothing left to fail. +pub(crate) async fn subscribe(lookup: MdnsAddressLookup) -> BoxStream { + // `subscribe` is an inherent method on MdnsAddressLookup, not part of iroh's + // AddressLookup trait -- the trait only has publish/resolve, with no way to + // enumerate. + Box::pin(lookup.subscribe().await.filter_map(convert)) +} + +/// `MdnsEvent` is `#[non_exhaustive]`, so unknown variants are dropped rather than +/// warned about on every tick. +fn convert(event: MdnsEvent) -> Option { + match event { + MdnsEvent::Discovered { endpoint_info, .. } => { + Some(DiscoveryEvent::Found(DiscoveredPeer { + id: endpoint_info.endpoint_id, + name: endpoint_info + .data + .user_data() + .map(|data| data.as_ref().to_owned()), + addr: endpoint_info.to_endpoint_addr(), + transport: DiscoveryTransport::Mdns, + })) + }, + MdnsEvent::Expired { endpoint_id } => Some(DiscoveryEvent::Lost { + id: endpoint_id, + transport: DiscoveryTransport::Mdns, + }), + _ => None, + } +} diff --git a/crates/beaver_p2p/src/discovery/mod.rs b/crates/beaver_p2p/src/discovery/mod.rs new file mode 100644 index 0000000..0088087 --- /dev/null +++ b/crates/beaver_p2p/src/discovery/mod.rs @@ -0,0 +1,66 @@ +/* SPDX Id: AGPL-3.0-or-later */ + +//! Peer discovery, abstracted away from any one transport. +//! +//! Discovery used to be mDNS and nothing else, and +//! `iroh_mdns_address_lookup::DiscoveryEvent` leaked all the way up into the +//! constellation. The types here are Beaver's own and `pub(crate)`, so a second +//! source (Bluetooth) can be added without either crate's type reaching past +//! `State`, which resolves raw events into the peer transitions the rest of the +//! system cares about. + +use iroh::{EndpointAddr, EndpointId}; +use n0_future::IterExt; +use n0_future::boxed::BoxStream; + +pub(crate) mod mdns; + +/// Which transport surfaced a peer. +/// +/// A peer can be visible on more than one at a time, so this is tracked as a set +/// per peer rather than a single value. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub(crate) enum DiscoveryTransport { + /// Found on the local network. + Mdns, + /// Found over Bluetooth LE. + #[expect(dead_code, reason = "constructed once the BLE source lands")] + Ble, +} + +/// A peer as reported by one discovery source. +#[derive(Clone, Debug)] +pub(crate) struct DiscoveredPeer { + pub id: EndpointId, + /// The peer's self-reported display name. + /// + /// `None` when the source cannot carry one, which must not be confused with + /// "the peer is called nothing": a source that reports `None` should never + /// overwrite a name another source already supplied. + pub name: Option, + pub addr: EndpointAddr, + pub transport: DiscoveryTransport, +} + +/// A change in what a discovery source can see. +#[derive(Clone, Debug)] +pub(crate) enum DiscoveryEvent { + /// A peer appeared, or its advertised data changed. + Found(DiscoveredPeer), + /// A peer is no longer visible *on this transport*. It may still be visible on + /// another, which is why the transport is part of the event. + Lost { + id: EndpointId, + transport: DiscoveryTransport, + }, +} + +/// Fan several sources into a single stream, so there is one consumer task +/// regardless of how many transports are enabled. +/// +/// Each source is a free function returning its own stream rather than an impl of +/// a shared trait: they have genuinely different setup (mDNS cannot fail, a radio +/// can), and nothing holds a source after it has been subscribed. +pub(crate) fn merge(streams: Vec>) -> BoxStream { + Box::pin(streams.merge()) +} diff --git a/crates/beaver_p2p/src/lib.rs b/crates/beaver_p2p/src/lib.rs index 83b37f1..edbcc52 100644 --- a/crates/beaver_p2p/src/lib.rs +++ b/crates/beaver_p2p/src/lib.rs @@ -1,5 +1,6 @@ /* SPDX Id: AGPL-3.0-or-later */ +mod discovery; mod message_protocol; mod packet; mod pairing_hook; @@ -8,6 +9,7 @@ mod state; use std::sync::Arc; use std::sync::mpsc::Sender; +use std::time::Duration; use iroh::address_lookup::UserData; use iroh::endpoint::{ @@ -16,7 +18,7 @@ use iroh::endpoint::{ use iroh::protocol::Router; use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey}; use iroh_mdns_address_lookup::MdnsAddressLookup; -use log::{error, info}; +use log::{debug, error, info}; use n0_future::StreamExt; pub use packet::PacketError; use packet::PostcardPacket; @@ -25,11 +27,59 @@ use thiserror::Error; use tokio::sync::Mutex; use tokio::task::AbortHandle; +use crate::discovery::mdns; use crate::packet::BasePacket; use crate::pairing_hook::{MESSAGE_ALPN, PAIRING_ALPN}; use crate::state::{EndpointDescription, PairingCommand, SharedState, State}; pub use crate::state::{EndpointStatus, PeerEvent}; +/// How long to wait for [`Endpoint::connect`] before giving up. +/// +/// iroh only resolves a pending connect once a path is found or the address-lookup +/// stream finishes. mDNS's stream does finish, so today an unreachable peer fails on +/// its own and this bound is belt-and-braces. A BLE address lookup's stream never +/// finishes by design, so once a second transport is wired in an unreachable peer +/// would park the caller forever - and `broadcast_message` walks peers serially, so a +/// single dead peer would stall every other one behind it. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + +/// Why a bounded connect did not produce a connection. Flattened into whichever +/// error enum the caller uses, so call sites stay a plain `?`. +enum ConnectFailed { + Timeout, + Connect(ConnectError), +} + +impl From for PairingError { + fn from(value: ConnectFailed) -> Self { + match value { + ConnectFailed::Timeout => PairingError::ConnectTimeout, + ConnectFailed::Connect(err) => PairingError::Connect(err), + } + } +} + +impl From for MessageError { + fn from(value: ConnectFailed) -> Self { + match value { + ConnectFailed::Timeout => MessageError::ConnectTimeout, + ConnectFailed::Connect(err) => MessageError::Connect(err), + } + } +} + +/// [`Endpoint::connect`], bounded by [`CONNECT_TIMEOUT`]. +async fn connect_bounded( + endpoint: &Endpoint, + addr: EndpointAddr, + alpn: &[u8], +) -> Result { + match n0_future::time::timeout(CONNECT_TIMEOUT, endpoint.connect(addr, alpn)).await { + Err(_) => Err(ConnectFailed::Timeout), + Ok(result) => result.map_err(ConnectFailed::Connect), + } +} + #[derive(Debug, Error)] pub enum PairingError { #[error("Pairing manager not initialized")] @@ -44,6 +94,8 @@ pub enum PairingError { FailedAck, #[error("Failed to connect")] Connect(#[from] ConnectError), + #[error("Timed out connecting to the remote endpoint")] + ConnectTimeout, #[error("Connection error")] Connection(#[from] ConnectionError), #[error("Write error")] @@ -77,6 +129,8 @@ pub enum MessageError { Packet(#[from] packet::PacketError), #[error("Failed to connect")] Connect(#[from] ConnectError), + #[error("Timed out connecting to the remote endpoint")] + ConnectTimeout, #[error("Connection error")] Connection(#[from] ConnectionError), #[error("Stream closed")] @@ -90,7 +144,7 @@ pub struct PairingManagerInner { state: SharedState, router: Router, - mdns_handle: AbortHandle, + discovery_handle: AbortHandle, } #[derive(Clone)] @@ -114,12 +168,15 @@ impl PairingManager { let mdns = MdnsAddressLookup::builder().build(endpoint.id()).unwrap(); endpoint.address_lookup().unwrap().add(mdns.clone()); + // One task drains a merged stream of every discovery source, so adding a + // second transport does not add a second consumer of `State`. + let streams = vec![mdns::subscribe(mdns).await]; let disco_state = Arc::clone(&state); - let mdns_handle = tokio::spawn(async move { - let mut events = mdns.subscribe().await; + let discovery_handle = tokio::spawn(async move { + let mut events = discovery::merge(streams); while let Some(event) = events.next().await { // Update the state. - disco_state.lock().await.on_discovery(&event); + disco_state.lock().await.on_discovery(event); } }); @@ -137,7 +194,7 @@ impl PairingManager { let inner = PairingManagerInner { state, router, - mdns_handle: mdns_handle.abort_handle(), + discovery_handle: discovery_handle.abort_handle(), }; Self { inner: Some(inner) } } @@ -212,7 +269,7 @@ impl PairingManager { remote.addr() }; - let connection = inner.router.endpoint().connect(addr, PAIRING_ALPN).await?; + let connection = connect_bounded(inner.router.endpoint(), addr, PAIRING_ALPN).await?; let (mut sender, mut receiver) = connection.open_bi().await?; PostcardPacket::send(PairingCommand::GuestRequest(pin.to_string()), &mut sender).await?; @@ -252,7 +309,7 @@ impl PairingManager { inner: &PairingManagerInner, addr: EndpointAddr, ) -> Result { - let connection = inner.router.endpoint().connect(addr, PAIRING_ALPN).await?; + let connection = connect_bounded(inner.router.endpoint(), addr, PAIRING_ALPN).await?; // Send the request. let (mut sender, mut receiver) = connection.open_bi().await?; @@ -414,11 +471,7 @@ impl PairingManager { } } - let conn = inner - .router - .endpoint() - .connect(addr.clone(), MESSAGE_ALPN) - .await?; + let conn = connect_bounded(inner.router.endpoint(), addr.clone(), MESSAGE_ALPN).await?; if let Some(r) = inner.state.lock().await.by_id_mut(to) { r.set_connection(conn.clone()) @@ -441,7 +494,7 @@ impl PairingManager { }; if !inner.router.is_shutdown() { - inner.mdns_handle.abort(); + inner.discovery_handle.abort(); let _ = inner.router.shutdown().await; } @@ -531,4 +584,35 @@ impl PairingManager { ); Ok(()) } + + /// Send `bytes` to each of `peers`, concurrently. + /// + /// Fans out rather than looping: `send_message` can block for up to + /// [`CONNECT_TIMEOUT`] (twice that, given its one retry) on a peer that has gone + /// away, and a serial loop would make every peer behind it wait. + pub async fn send_message_to_many(&self, peers: &[EndpointId], bytes: &[u8]) { + n0_future::join_all(peers.iter().map(|id| async move { + if let Err(e) = self.send_message(id, bytes).await { + error!("Failed to send message to {id}: {e}"); + } + })) + .await; + } + + /// Send `bytes` to every paired, connected peer. + pub async fn broadcast_message(&self, bytes: &[u8]) { + let peers = self.peers().await; + let total = peers.len(); + let connected: Vec = peers + .into_iter() + .filter(|peer| peer.status == EndpointStatus::PairedConnected) + .map(|peer| peer.id) + .collect(); + debug!( + "[P2P] broadcast_message: {total} total peers, {} connected", + connected.len() + ); + + self.send_message_to_many(&connected, bytes).await; + } } diff --git a/crates/beaver_p2p/src/main.rs b/crates/beaver_p2p/src/main.rs index f23f520..544721e 100644 --- a/crates/beaver_p2p/src/main.rs +++ b/crates/beaver_p2p/src/main.rs @@ -1,9 +1,7 @@ use std::sync::mpsc::channel; use beaver_p2p::{PairingManager, PairingResult, PeerEvent}; -use iroh::address_lookup::DiscoveryEvent; use log::info; -use serde::{Deserialize, Serialize}; #[tokio::main] async fn main() { @@ -23,22 +21,18 @@ async fn main() { loop { match receiver.recv() { Ok(event) => match event { - PeerEvent::Discovery(DiscoveryEvent::Discovered { endpoint_info, .. }) => { - info!( - "MDNS discovered: {:?} {:?}", - endpoint_info.endpoint_id, - endpoint_info.user_data() - ); + PeerEvent::PeerDiscovered(id, name) => { + info!("Discovered {name} ({id})"); if start_pairing { - match manager.request_pairing(&endpoint_info.endpoint_id).await { + match manager.request_pairing(&id).await { Ok(PairingResult::Accepted) => info!("Pairing accepted!"), Ok(PairingResult::Rejected) => info!("Pairing rejected."), Err(e) => info!("Pairing failed: {e}"), } } }, - PeerEvent::Discovery(DiscoveryEvent::Expired { endpoint_id }) => { - info!("MDNS expired: {endpoint_id}"); + PeerEvent::PeerLost(id) => { + info!("Lost {id}"); }, PeerEvent::PairingRequest(endpoint_id, name) => { info!("Accepting pairing request from {name} ({endpoint_id})"); @@ -56,14 +50,16 @@ async fn main() { PeerEvent::PairingFailed(endpoint_id) => { info!("Pairing failed with {endpoint_id}"); }, + PeerEvent::GuestAccepted(endpoint_id) => { + info!("Guest accepted: {endpoint_id}"); + }, + PeerEvent::GuestExpired(endpoint_id) => { + info!("Guest expired: {endpoint_id}"); + }, PeerEvent::Message(endpoint_id, payload) => { - match postcard::from_bytes::(&payload) { - Ok(msg) => info!("Message from {endpoint_id}: {msg:?}"), - Err(e) => info!( - "Raw message of {} bytes from {endpoint_id} (decode error: {e})", - payload.len() - ), - } + // The `P2pMessage` envelope is defined in the constellation, which + // this crate does not depend on, so just report the raw size here. + info!("Message of {} bytes from {endpoint_id}", payload.len()); }, }, Err(err) => { diff --git a/crates/beaver_p2p/src/state.rs b/crates/beaver_p2p/src/state.rs index 87e821a..12445c0 100644 --- a/crates/beaver_p2p/src/state.rs +++ b/crates/beaver_p2p/src/state.rs @@ -7,16 +7,17 @@ use std::time::Instant; use iroh::endpoint::Connection; use iroh::{EndpointAddr, EndpointId}; -use iroh_mdns_address_lookup::DiscoveryEvent; use log::warn; use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; use tokio::sync::mpsc::{Receiver as TokioReceiver, Sender as TokioSender}; +use crate::discovery::{DiscoveredPeer, DiscoveryEvent, DiscoveryTransport}; + pub(crate) type SharedState = Arc>; /// The status of a remote peer. -#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[derive(Clone, Debug, PartialEq)] pub enum EndpointStatus { /// Discovered but not paired. Discovered, @@ -34,6 +35,11 @@ pub(crate) struct EndpointProxy { id: EndpointId, addr: EndpointAddr, status: EndpointStatus, + /// Which discovery transports currently see this peer. A peer is only really + /// gone once this empties, so one transport expiring cannot evict a peer that + /// another still sees. Empty for peers we learned about some other way -- a + /// restored pairing, or an inbound connection -- until a source reports them. + seen_on: HashSet, /// Cached QUIC connection for sending messages. connection: Option, } @@ -50,10 +56,24 @@ impl EndpointProxy { id: id.to_owned(), addr, status, + seen_on: HashSet::new(), connection: None, } } + /// Build a freshly discovered peer, seeding `seen_on` with the source that + /// reported it. + pub(crate) fn from_discovered(peer: &DiscoveredPeer) -> Self { + let mut proxy = Self::new( + peer.name.as_deref().unwrap_or(""), + peer.id, + peer.addr.clone(), + EndpointStatus::Discovered, + ); + proxy.seen_on.insert(peer.transport); + proxy + } + pub(crate) fn addr(&self) -> EndpointAddr { self.addr.clone() } @@ -62,6 +82,31 @@ impl EndpointProxy { &self.name } + /// Fold in what a discovery source just reported. + /// + /// Addresses are unioned rather than replaced: two transports can each know a + /// different way to reach the same peer, and dropping one because the other + /// reported first would lose a working path. The name is only taken when the + /// source actually supplied one -- Bluetooth advertisements may not carry it, + /// and an absent name must not clobber a good one. + pub(crate) fn merge_discovered(&mut self, peer: &DiscoveredPeer) { + self.addr.addrs.extend(peer.addr.addrs.iter().cloned()); + // Re-announcements carry the same name every time; skip the reallocation. + if let Some(name) = &peer.name && + self.name != *name + { + self.name = name.clone(); + } + self.seen_on.insert(peer.transport); + } + + /// Record that `transport` no longer sees this peer. Returns whether any + /// transport still does. + pub(crate) fn unsee(&mut self, transport: DiscoveryTransport) -> bool { + self.seen_on.remove(&transport); + !self.seen_on.is_empty() + } + pub(crate) fn is_paired(&self) -> bool { matches!( self.status, @@ -88,7 +133,11 @@ impl EndpointProxy { // failure to complete the Ack handshake. #[derive(Debug)] pub enum PeerEvent { - Discovery(DiscoveryEvent), + /// A peer we had not seen before. The name is the merged one held by + /// `State`, not whatever the reporting source happened to carry. + PeerDiscovered(EndpointId, String), + /// A peer is gone: no discovery source can see it any more. + PeerLost(EndpointId), PairingRequest(EndpointId, String), PairingAccepted(EndpointId), PairingRejected(EndpointId), @@ -217,12 +266,6 @@ impl State { self.close_incoming_connection(id); } - fn discovered(&self, id: &EndpointId) -> bool { - self.endpoints - .get(id) - .is_some_and(|desc| desc.status != EndpointStatus::PairedDisconnected) - } - pub(crate) fn has_requested(&self, id: &EndpointId) -> bool { self.pairing_requested.contains(id) } @@ -300,68 +343,64 @@ impl State { } } - pub(crate) fn on_discovery(&mut self, event: &DiscoveryEvent) { + /// Fold a raw source event into the peer table, emitting a [`PeerEvent`] only + /// for a real transition: a peer we had never seen, a paired peer coming back, + /// or a peer no source can see any more. Repeat announcements are absorbed. + pub(crate) fn on_discovery(&mut self, event: DiscoveryEvent) { match event { - DiscoveryEvent::Discovered { endpoint_info, .. } => { - // Ignore if we already know about this endpoint (except PairedDisconnected). - if self.discovered(&endpoint_info.endpoint_id) { + DiscoveryEvent::Found(peer) => { + if let Some(existing) = self.endpoints.get_mut(&peer.id) { + let reconnecting = existing.status == EndpointStatus::PairedDisconnected; + if reconnecting { + existing.status = EndpointStatus::PairedConnected; + } + existing.merge_discovered(&peer); + if reconnecting { + // Fire PairingAccepted so JS gets "peerjoined", not "peerdiscovered". + self.notify(PeerEvent::PairingAccepted(peer.id)); + } return; } - let name = endpoint_info - .data - .user_data() - .map(|d| d.as_ref()) - .unwrap_or_else(|| ""); - let addr = endpoint_info.to_endpoint_addr(); - - // If this peer was previously paired and disconnected, reconnect it. - if let Some(existing) = self.endpoints.get_mut(&endpoint_info.endpoint_id) && - existing.status == EndpointStatus::PairedDisconnected - { - existing.status = EndpointStatus::PairedConnected; - existing.addr = addr; - existing.name = name.to_owned(); - // Fire PairingAccepted so JS gets "peerjoined" (not "peerdiscovered"). - self.notify(PeerEvent::PairingAccepted(endpoint_info.endpoint_id)); + let description = EndpointProxy::from_discovered(&peer); + let name = description.name.clone(); + self.endpoints.insert(peer.id, description); + self.notify(PeerEvent::PeerDiscovered(peer.id, name)); + }, + DiscoveryEvent::Lost { id, transport } => { + let Some(existing) = self.endpoints.get_mut(&id) else { + return; + }; + + // Still visible on another transport: nothing to report. + if existing.unsee(transport) { return; } - // Add it as Discovered and notify the listener. - let description = EndpointProxy { - name: name.into(), - id: endpoint_info.endpoint_id, - addr, - status: EndpointStatus::Discovered, - connection: None, - }; - self.endpoints - .insert(endpoint_info.endpoint_id, description); - self.notify(PeerEvent::Discovery(event.clone())); - }, - DiscoveryEvent::Expired { endpoint_id } => { - // PairedConnected -> PairedDisconnected + // PairedConnected -> PairedDisconnected (kept, so it can reconnect) // GuestConnected -> removed (no persistence) // Discovered -> removed - if let Some(mut old_desc) = self.endpoints.remove(endpoint_id) { - if old_desc.status == EndpointStatus::PairedConnected { - old_desc.status = EndpointStatus::PairedDisconnected; - self.endpoints.insert(*endpoint_id, old_desc); - } else if old_desc.status == EndpointStatus::GuestConnected { + if existing.status == EndpointStatus::PairedConnected { + existing.status = EndpointStatus::PairedDisconnected; + } else { + let old_desc = self + .endpoints + .remove(&id) + .expect("just looked the endpoint up"); + if old_desc.status == EndpointStatus::GuestConnected { if let Some(ref mut guest_state) = self.guest_mode { - guest_state.guest_peers.remove(endpoint_id); + guest_state.guest_peers.remove(&id); } - self.close_incoming_connection(endpoint_id); + self.close_incoming_connection(&id); } else if old_desc.status != EndpointStatus::Discovered { warn!( "Unexpected status for expired endpoint: {:?}", old_desc.status ); } - self.notify(PeerEvent::Discovery(event.clone())); } + self.notify(PeerEvent::PeerLost(id)); }, - _ => warn!("Unexpected discovery event: {event:?}"), } } } diff --git a/crates/beaver_p2p/tests/pairing.rs b/crates/beaver_p2p/tests/pairing.rs index ab0b811..37bcc81 100644 --- a/crates/beaver_p2p/tests/pairing.rs +++ b/crates/beaver_p2p/tests/pairing.rs @@ -1,9 +1,8 @@ use std::sync::Arc; use std::sync::mpsc::{Receiver, channel}; +use beaver_p2p::{EndpointStatus, MessageError, PairingManager, PairingResult, PeerEvent}; use iroh::EndpointId; -use iroh::address_lookup::DiscoveryEvent; -use p2p_beaver::{EndpointStatus, MessageError, PairingError, PairingManager, PeerEvent}; use parking_lot::Mutex; type PairingReceiver = Receiver; @@ -29,8 +28,8 @@ fn wait_for_discovery( loop { match receiver1.lock().recv() { Ok(event) => match event { - PeerEvent::Discovery(DiscoveryEvent::Discovered { endpoint_info, .. }) => { - return endpoint_info.endpoint_id; + PeerEvent::PeerDiscovered(id, _) => { + return id; }, _ => {}, }, @@ -45,8 +44,8 @@ fn wait_for_discovery( loop { match receiver2.lock().recv() { Ok(event) => match event { - PeerEvent::Discovery(DiscoveryEvent::Discovered { endpoint_info, .. }) => { - return endpoint_info.endpoint_id; + PeerEvent::PeerDiscovered(id, _) => { + return id; }, _ => {}, }, @@ -63,16 +62,6 @@ fn wait_for_discovery( (endpoint1, endpoint2) } -// Easier assert for PairingError -macro_rules! assert_pairing { - ($observed:expr, $expected:pat) => { - match $observed.err().unwrap() { - $expected => {}, - _ => panic!("expected: {{$expected}} but got {{$observed}}"), - } - }; -} - // Discover and stop. #[tokio::test(flavor = "multi_thread")] async fn discover_and_shutdown() { @@ -83,9 +72,8 @@ async fn discover_and_shutdown() { loop { match receiver1.recv() { Ok(event) => match event { - PeerEvent::Discovery(DiscoveryEvent::Discovered { endpoint_info, .. }) => { - let endpoint_name = format!("{}", endpoint_info.user_data().unwrap()); - assert_eq!(endpoint_name, "test-2"); + PeerEvent::PeerDiscovered(_, name) => { + assert_eq!(name, "test-2"); break; }, _ => {}, @@ -101,9 +89,8 @@ async fn discover_and_shutdown() { loop { match receiver2.recv() { Ok(event) => match event { - PeerEvent::Discovery(DiscoveryEvent::Discovered { endpoint_info, .. }) => { - let endpoint_name = format!("{}", endpoint_info.user_data().unwrap()); - assert_eq!(endpoint_name, "test-1"); + PeerEvent::PeerDiscovered(_, name) => { + assert_eq!(name, "test-1"); break; }, _ => {}, @@ -142,14 +129,11 @@ async fn discover_and_expire() { Ok(event) => { println!("mgr1 event: {event:?}"); match event { - PeerEvent::Discovery(DiscoveryEvent::Discovered { - endpoint_info, .. - }) => { - let endpoint_name = format!("{}", endpoint_info.user_data().unwrap()); - assert_eq!(endpoint_name, "test-2"); + PeerEvent::PeerDiscovered(_, name) => { + assert_eq!(name, "test-2"); }, - PeerEvent::Discovery(DiscoveryEvent::Expired { endpoint_id }) => { - assert_eq!(endpoint_id, endpoint2); + PeerEvent::PeerLost(id) => { + assert_eq!(id, endpoint2); break; }, _ => {}, @@ -217,7 +201,7 @@ async fn reject_pairing() { println!("mgr2 event: {event:?}"); let mgr = mgr.clone(); match event { - PeerEvent::PairingRequest(endpoint) => { + PeerEvent::PairingRequest(endpoint, _) => { assert_eq!(endpoint, endpoint2); rt.block_on(async { mgr.reject_pairing(&endpoint) @@ -240,7 +224,11 @@ async fn reject_pairing() { }); let response = manager1.request_pairing(&endpoint1).await; - assert_pairing!(response, PairingError::Rejected); + // A rejection is a successful handshake with a negative answer, not an error. + assert_eq!( + response.expect("pairing handshake failed"), + PairingResult::Rejected + ); let _ = handle1.join(); let _ = handle2.join(); @@ -297,7 +285,7 @@ async fn accept_pairing() { println!("mgr2 event: {event:?}"); let mgr = mgr.clone(); match event { - PeerEvent::PairingRequest(endpoint) => { + PeerEvent::PairingRequest(endpoint, _) => { assert_eq!(endpoint, endpoint2); rt.block_on(async { mgr.accept_pairing(&endpoint) diff --git a/patches/components/constellation/Cargo.toml.patch b/patches/components/constellation/Cargo.toml.patch index f36e2c3..5a1d632 100644 --- a/patches/components/constellation/Cargo.toml.patch +++ b/patches/components/constellation/Cargo.toml.patch @@ -9,17 +9,16 @@ content-security-policy = { workspace = true } crossbeam-channel = { workspace = true } devtools_traits = { workspace = true } -@@ -37,6 +39,9 @@ +@@ -37,6 +39,8 @@ euclid = { workspace = true } fonts = { workspace = true } ipc-channel = { workspace = true } +iroh = { workspace = true } -+iroh-mdns-address-lookup = { workspace = true } +iroh-persist = { workspace = true } keyboard-types = { workspace = true } layout_api = { workspace = true } log = { workspace = true } -@@ -43,13 +48,17 @@ +@@ -43,13 +47,17 @@ media = { workspace = true } net = { workspace = true } net_traits = { workspace = true } @@ -37,7 +36,7 @@ servo-background-hang-monitor = { workspace = true } servo-background-hang-monitor-api = { workspace = true } servo-base = { workspace = true } -@@ -63,10 +72,16 @@ +@@ -63,10 +71,16 @@ storage_traits = { workspace = true } stylo = { workspace = true } stylo_traits = { workspace = true } diff --git a/patches/components/constellation/pairing.rs.patch b/patches/components/constellation/pairing.rs.patch index 932bbeb..6480068 100644 --- a/patches/components/constellation/pairing.rs.patch +++ b/patches/components/constellation/pairing.rs.patch @@ -1,6 +1,6 @@ --- original +++ modified -@@ -0,0 +1,1002 @@ +@@ -0,0 +1,982 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//! P2P pairing service integration with the constellation. @@ -15,7 +15,6 @@ + +use beaver_p2p::{PairingManager, PairingResult, PeerEvent}; +use iroh::EndpointId; -+use iroh_mdns_address_lookup::DiscoveryEvent; +use log::{debug, error, info, warn}; +use serde::{Deserialize, Serialize}; +use servo_base::generic_channel::GenericCallback; @@ -291,14 +290,7 @@ + .map(|ep| PeerInfo { + id: ep.id.to_string(), + name: ep.name, -+ status: match ep.status { -+ EndpointStatus::Discovered => EndpointStatus::Discovered, -+ EndpointStatus::PairedConnected => EndpointStatus::PairedConnected, -+ EndpointStatus::PairedDisconnected => { -+ EndpointStatus::PairedDisconnected -+ }, -+ EndpointStatus::GuestConnected => EndpointStatus::GuestConnected, -+ }, ++ status: to_ipc_status(ep.status), + }) + .collect(); + Ok(peers) @@ -551,23 +543,7 @@ + }, + } + }; -+ let peers = mgr.peers().await; -+ let connected: Vec<_> = peers -+ .iter() -+ .filter(|p| p.status == beaver_p2p::EndpointStatus::PairedConnected) -+ .collect(); -+ warn!( -+ "[P2P BC] broadcast_message: {} total peers, {} connected", -+ peers.len(), -+ connected.len() -+ ); -+ for peer in peers { -+ if peer.status == beaver_p2p::EndpointStatus::PairedConnected { -+ if let Err(e) = mgr.send_message(&peer.id, &bytes).await { -+ error!("Failed to send broadcast to {}: {e}", peer.id); -+ } -+ } -+ } ++ mgr.broadcast_message(&bytes).await; + }); + } + @@ -621,22 +597,28 @@ + None => return, + } + }; -+ let channels = remote_channels.lock().await; -+ let key = (origin.clone(), name.clone()); -+ warn!( -+ "[P2P BC] Forwarding broadcast: looking for ({origin}, {name}) in {} peers, remote_channels={:?}", -+ channels.len(), -+ channels -+ ); -+ for (peer_id, peer_channels) in channels.iter() { -+ if peer_channels.contains(&key) { -+ if let Ok(endpoint_id) = peer_id.parse() { -+ if let Err(e) = mgr.send_message(&endpoint_id, &bytes).await { -+ error!("Failed to forward broadcast to {peer_id}: {e}"); -+ } -+ } -+ } -+ } ++ ++ // Resolve the target peers while holding the lock, then release it before ++ // sending. `send_message` can block for up to the connect timeout per ++ // peer, and holding `remote_channels` across that would stall every other ++ // task that touches it. ++ let targets = { ++ let channels = remote_channels.lock().await; ++ let key = (origin.clone(), name.clone()); ++ let targets: Vec = channels ++ .iter() ++ .filter(|(_, peer_channels)| peer_channels.contains(&key)) ++ .filter_map(|(peer_id, _)| peer_id.parse().ok()) ++ .collect(); ++ debug!( ++ "[P2P BC] Forwarding ({origin}, {name}) to {} of {} peers with channels", ++ targets.len(), ++ channels.len() ++ ); ++ targets ++ }; ++ ++ mgr.send_message_to_many(&targets, &bytes).await; + }); + } + @@ -960,30 +942,28 @@ + } +} + ++/// Bridge `beaver_p2p`'s status enum to the IPC one. The two are deliberately ++/// separate types so that `servo_constellation_traits` does not depend on ++/// `beaver-p2p`, and therefore not on iroh. A free function rather than a `From` ++/// impl because both types are foreign to this crate. ++fn to_ipc_status(status: beaver_p2p::EndpointStatus) -> EndpointStatus { ++ match status { ++ beaver_p2p::EndpointStatus::Discovered => EndpointStatus::Discovered, ++ beaver_p2p::EndpointStatus::PairedConnected => EndpointStatus::PairedConnected, ++ beaver_p2p::EndpointStatus::PairedDisconnected => EndpointStatus::PairedDisconnected, ++ beaver_p2p::EndpointStatus::GuestConnected => EndpointStatus::GuestConnected, ++ } ++} ++ +/// Convert a beaver_p2p PeerEvent to a serializable PairingEvent. +/// Returns None for events that don't map (e.g. Message). +fn to_pairing_event(event: &PeerEvent) -> Option { + match event { -+ PeerEvent::Discovery(DiscoveryEvent::Discovered { endpoint_info, .. }) => { -+ let name = endpoint_info -+ .data -+ .user_data() -+ .map(|d| d.as_ref().to_owned()) -+ .unwrap_or_else(|| "".to_owned()); -+ Some(PairingEvent::PeerDiscovered { -+ id: endpoint_info.endpoint_id.to_string(), -+ name, -+ }) -+ }, -+ PeerEvent::Discovery(DiscoveryEvent::Expired { endpoint_id }) => { -+ Some(PairingEvent::PeerExpired { -+ id: endpoint_id.to_string(), -+ }) -+ }, -+ PeerEvent::Discovery(discovery_event) => { -+ warn!("Unexpected discovery event: {discovery_event:?}"); -+ None -+ }, ++ PeerEvent::PeerDiscovered(id, name) => Some(PairingEvent::PeerDiscovered { ++ id: id.to_string(), ++ name: name.clone(), ++ }), ++ PeerEvent::PeerLost(id) => Some(PairingEvent::PeerExpired { id: id.to_string() }), + PeerEvent::PairingRequest(id, name) => Some(PairingEvent::PairingRequest { + id: id.to_string(), + name: name.clone(), diff --git a/patches/components/shared/constellation/Cargo.toml.patch b/patches/components/shared/constellation/Cargo.toml.patch index 22ae1b4..aa245da 100644 --- a/patches/components/shared/constellation/Cargo.toml.patch +++ b/patches/components/shared/constellation/Cargo.toml.patch @@ -1,15 +1,14 @@ --- original +++ modified -@@ -21,6 +21,8 @@ +@@ -21,6 +21,7 @@ [dependencies] base64 = { workspace = true } +beaver-hal = { path = "../../../../crates/beaver_hal" } -+beaver-p2p = { path = "../../../../crates/beaver_p2p" } content-security-policy = { workspace = true } devtools_traits = { workspace = true } embedder_traits = { workspace = true } -@@ -38,6 +40,7 @@ +@@ -38,6 +39,7 @@ profile_traits = { workspace = true } rustc-hash = { workspace = true } serde = { workspace = true } diff --git a/patches/components/shared/constellation/lib.rs.patch b/patches/components/shared/constellation/lib.rs.patch index 4902d9d..a39ca0a 100644 --- a/patches/components/shared/constellation/lib.rs.patch +++ b/patches/components/shared/constellation/lib.rs.patch @@ -1,12 +1,11 @@ --- original +++ modified -@@ -13,14 +13,17 @@ +@@ -13,14 +13,16 @@ use std::collections::VecDeque; use std::fmt; +use std::time::Duration; -+pub use beaver_p2p::EndpointStatus; use embedder_traits::user_contents::{ UserContentManagerId, UserScript, UserScriptId, UserStyleSheet, UserStyleSheetId, }; @@ -21,7 +20,7 @@ }; pub use from_script_message::*; use malloc_size_of_derive::MallocSizeOf; -@@ -30,15 +33,242 @@ +@@ -30,15 +32,259 @@ use rustc_hash::FxHashMap; use serde::{Deserialize, Serialize}; use servo_base::cross_process_instant::CrossProcessInstant; @@ -206,6 +205,23 @@ + pub status: EndpointStatus, +} + ++/// The status of a remote P2P peer. ++/// ++/// Mirrors `beaver_p2p::EndpointStatus`. Defined natively here rather than ++/// re-exported so that this IPC crate does not depend on `beaver-p2p`, and ++/// therefore not on iroh. The constellation converts between the two. ++#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] ++pub enum EndpointStatus { ++ /// Discovered but not paired. ++ Discovered, ++ /// Paired and currently connected. ++ PairedConnected, ++ /// Paired but currently disconnected. ++ PairedDisconnected, ++ /// Temporary guest pairing, currently connected. ++ GuestConnected, ++} ++ +/// Events from the P2P pairing service, using simple serializable types. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub enum PairingEvent { @@ -266,7 +282,7 @@ /// Messages to the Constellation from the embedding layer, whether from `ServoRenderer` or /// from `libservo` itself. #[derive(IntoStaticStr)] -@@ -55,6 +285,15 @@ +@@ -55,6 +301,15 @@ ChangeViewportDetails(WebViewId, ViewportDetails, WindowSizeType), /// Inform the constellation of a theme change. ThemeChange(WebViewId, Theme), @@ -282,7 +298,7 @@ /// Requests that the constellation instruct script/layout to try to layout again and tick /// animations. TickAnimation(Vec), -@@ -79,6 +318,12 @@ +@@ -79,6 +334,12 @@ BlurWebView, /// Forward an input event to an appropriate ScriptTask. ForwardInputEvent(WebViewId, InputEventAndId, Option), @@ -295,7 +311,7 @@ /// Request that the given pipeline refresh the cursor by doing a hit test at the most /// recently hovered cursor position and resetting the cursor. This happens after a /// display list update is rendered. -@@ -114,6 +359,9 @@ +@@ -114,6 +375,9 @@ UpdatePinchZoomInfos(PipelineId, PinchZoomInfos), /// Activate or deactivate accessibility features for the given `WebView`. SetAccessibilityActive(WebViewId, bool),