diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index 21fcee7..2046cc4 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -19,7 +19,7 @@ use tsunagi::discovery::{ CompositeDiscovery, MainlineDiscovery, NetworkDiscovery, StaticBootstrap, }; use tsunagi::identity::{NetworkName, NetworkSecret}; -use tsunagi::iroh_types::EndpointAddr; +use tsunagi::iroh_types::{EndpointAddr, EndpointId}; use tsunagi::overlay::{MemoryTunFactory, TunFactory}; use tsunagi::state::Ipv4Range; use tsunagi::{Agent, NetworkId}; @@ -3864,125 +3864,137 @@ async fn build_report( let overlay = if has_any_plugin && overlay.is_some() { let mut peers: Vec = Vec::new(); - // 1. Peers known to the wg-quic plugin. - if let Some(ref view) = wg_view { - for peer in &view.peers { - let tcp_tls_match = - tcp_tls_reports.iter().find(|r| r.peer == peer.endpoint_id); - - if let Some(tcp_tls) = tcp_tls_match { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: tcp_tls.protocol.clone(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: Some(tcp_tls.uptime_secs), - tx_packets: tcp_tls.tx_packets, - rx_packets: tcp_tls.rx_packets, - dropped: tcp_tls.dropped, - protocol_errors: 0, - path: tcp_tls.path.clone(), - }); - } else if let Some(tunnel) = &peer.tunnel { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: "wg-quic".into(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: tunnel - .health - .since_handshake - .map(|s| s.as_secs()), - tx_packets: tunnel.stats.tx_packets, - rx_packets: tunnel.stats.rx_packets, - dropped: tunnel.stats.dropped_wrong_source - + tunnel.stats.dropped_oversize, - protocol_errors: tunnel.stats.protocol_errors, - path: tunnel.path.clone(), - }); - } else { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: "wg-quic".into(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: None, - tx_packets: 0, - rx_packets: 0, - dropped: 0, - protocol_errors: 0, - path: "no data link".into(), - }); - } + // Collect all unique peer IDs known in this network. + let mut all_peer_ids: Vec = Vec::new(); + for member in &network.members { + if member.endpoint_id != status.endpoint_id + && !all_peer_ids.contains(&member.endpoint_id) + { + all_peer_ids.push(member.endpoint_id); } } - - // 2. Peers known to the pure WG plugin but not yet covered. if let Some(ref view) = pure_wg_view { for peer in &view.peers { - if peers - .iter() - .any(|p| p.endpoint_id == peer.endpoint_id.to_string()) - { - continue; + if !all_peer_ids.contains(&peer.endpoint_id) { + all_peer_ids.push(peer.endpoint_id); } - if let Some(tunnel) = &peer.tunnel { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: "wg".into(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: tunnel - .health - .since_handshake - .map(|s| s.as_secs()), - tx_packets: tunnel.stats.tx_packets, - rx_packets: tunnel.stats.rx_packets, - dropped: tunnel.stats.dropped_wrong_source - + tunnel.stats.dropped_oversize, - protocol_errors: tunnel.stats.protocol_errors, - path: tunnel.path.clone(), - }); - } else { - peers.push(OverlayPeerReport { - endpoint_id: peer.endpoint_id.to_string(), - protocol: "wg".into(), - public_key: peer.public_key.to_string(), - address: peer.overlay_address_v4.map(|addr| addr.to_string()), - handshake_secs_ago: None, - tx_packets: 0, - rx_packets: 0, - dropped: 0, - protocol_errors: 0, - path: "no data link".into(), - }); + } + } + if let Some(ref view) = wg_view { + for peer in &view.peers { + if !all_peer_ids.contains(&peer.endpoint_id) { + all_peer_ids.push(peer.endpoint_id); } } } - - // 3. tcp-tls peers not yet covered by either WG view. for report in &tcp_tls_reports { - if !peers + if !all_peer_ids.contains(&report.peer) { + all_peer_ids.push(report.peer); + } + } + + for peer_id in all_peer_ids { + let pure_wg_match = pure_wg_view + .as_ref() + .and_then(|v| v.peers.iter().find(|p| p.endpoint_id == peer_id)); + let tcp_tls_match = tcp_tls_reports.iter().find(|r| r.peer == peer_id); + let wg_quic_match = wg_view + .as_ref() + .and_then(|v| v.peers.iter().find(|p| p.endpoint_id == peer_id)); + + let member_addr = network + .members .iter() - .any(|p| p.endpoint_id == report.peer.to_string()) + .find(|m| m.endpoint_id == peer_id) + .and_then(|m| m.overlay_address_v4) + .map(|a| a.to_string()); + + if let Some(pure_peer) = pure_wg_match + && let Some(tunnel) = &pure_peer.tunnel { - let address = network - .members - .iter() - .find(|m| m.endpoint_id == report.peer) - .and_then(|m| m.overlay_address_v4) - .map(|a| a.to_string()); peers.push(OverlayPeerReport { - endpoint_id: report.peer.to_string(), - protocol: report.protocol.clone(), - public_key: String::new(), - address, - handshake_secs_ago: Some(report.uptime_secs), - tx_packets: report.tx_packets, - rx_packets: report.rx_packets, - dropped: report.dropped, + endpoint_id: peer_id.to_string(), + protocol: "wg".into(), + public_key: pure_peer.public_key.to_string(), + address: pure_peer + .overlay_address_v4 + .map(|a| a.to_string()) + .or(member_addr), + handshake_secs_ago: tunnel.health.since_handshake.map(|s| s.as_secs()), + tx_packets: tunnel.stats.tx_packets, + rx_packets: tunnel.stats.rx_packets, + dropped: tunnel.stats.dropped_wrong_source + + tunnel.stats.dropped_oversize, + protocol_errors: tunnel.stats.protocol_errors, + path: tunnel.path.clone(), + }); + } else if let Some(tcp_tls) = tcp_tls_match { + let pubkey = pure_wg_match + .map(|p| p.public_key.to_string()) + .or_else(|| wg_quic_match.map(|p| p.public_key.to_string())) + .unwrap_or_default(); + peers.push(OverlayPeerReport { + endpoint_id: peer_id.to_string(), + protocol: tcp_tls.protocol.clone(), + public_key: pubkey, + address: member_addr, + handshake_secs_ago: Some(tcp_tls.uptime_secs), + tx_packets: tcp_tls.tx_packets, + rx_packets: tcp_tls.rx_packets, + dropped: tcp_tls.dropped, protocol_errors: 0, - path: report.path.clone(), + path: tcp_tls.path.clone(), + }); + } else if let Some(wg_peer) = wg_quic_match + && let Some(tunnel) = &wg_peer.tunnel + { + peers.push(OverlayPeerReport { + endpoint_id: peer_id.to_string(), + protocol: "wg-quic".into(), + public_key: wg_peer.public_key.to_string(), + address: wg_peer + .overlay_address_v4 + .map(|a| a.to_string()) + .or(member_addr), + handshake_secs_ago: tunnel.health.since_handshake.map(|s| s.as_secs()), + tx_packets: tunnel.stats.tx_packets, + rx_packets: tunnel.stats.rx_packets, + dropped: tunnel.stats.dropped_wrong_source + + tunnel.stats.dropped_oversize, + protocol_errors: tunnel.stats.protocol_errors, + path: tunnel.path.clone(), + }); + } else if let Some(pure_peer) = pure_wg_match { + peers.push(OverlayPeerReport { + endpoint_id: peer_id.to_string(), + protocol: "wg".into(), + public_key: pure_peer.public_key.to_string(), + address: pure_peer + .overlay_address_v4 + .map(|a| a.to_string()) + .or(member_addr), + handshake_secs_ago: None, + tx_packets: 0, + rx_packets: 0, + dropped: 0, + protocol_errors: 0, + path: "no data link".into(), + }); + } else if let Some(wg_peer) = wg_quic_match { + peers.push(OverlayPeerReport { + endpoint_id: peer_id.to_string(), + protocol: "wg-quic".into(), + public_key: wg_peer.public_key.to_string(), + address: wg_peer + .overlay_address_v4 + .map(|a| a.to_string()) + .or(member_addr), + handshake_secs_ago: None, + tx_packets: 0, + rx_packets: 0, + dropped: 0, + protocol_errors: 0, + path: "no data link".into(), }); } } diff --git a/crates/tsunagi-wireguard/src/plugin.rs b/crates/tsunagi-wireguard/src/plugin.rs index f9c68be..ae2c7cb 100644 --- a/crates/tsunagi-wireguard/src/plugin.rs +++ b/crates/tsunagi-wireguard/src/plugin.rs @@ -51,12 +51,6 @@ pub fn local_ip_candidates() -> Vec { ips.push(IpAddr::V4(addr)); } } - for ip in iface.ipv6 { - let addr = ip.addr(); - if !addr.is_loopback() && !addr.is_unspecified() { - ips.push(IpAddr::V6(addr)); - } - } } } ips diff --git a/crates/tsunagi-wireguard/src/transport.rs b/crates/tsunagi-wireguard/src/transport.rs index ca9c66a..9ad0b64 100644 --- a/crates/tsunagi-wireguard/src/transport.rs +++ b/crates/tsunagi-wireguard/src/transport.rs @@ -3,7 +3,7 @@ //! Provides [`WireguardTransport`] implementing [`PacketTransport`] //! and [`WireguardLink`] implementing [`PacketLink`]. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, RwLock}; @@ -31,6 +31,8 @@ pub struct WireguardLink { network: NetworkId, peer: EndpointId, target_addr: Arc>, + candidates: Arc>>, + confirmed: Arc, tx: mpsc::Sender, inbound_tx: mpsc::Sender, rx: tokio::sync::Mutex>, @@ -44,18 +46,23 @@ impl WireguardLink { network: NetworkId, peer: EndpointId, target_addr: SocketAddr, + candidates: Vec, socket: Arc, ) -> Self { let (tx_out, mut rx_out) = mpsc::channel::(1024); let (tx_in, rx_in) = mpsc::channel::(1024); let is_closed = Arc::new(AtomicBool::new(false)); + let confirmed = Arc::new(AtomicBool::new(false)); let closed_notify = Arc::new(Notify::new()); let target_holder = Arc::new(RwLock::new(target_addr)); + let candidates_holder = Arc::new(RwLock::new(candidates)); let socket_w = Arc::clone(&socket); let target_r = Arc::clone(&target_holder); + let candidates_r = Arc::clone(&candidates_holder); + let confirmed_r = Arc::clone(&confirmed); let is_closed_w = Arc::clone(&is_closed); let closed_notify_w = Arc::clone(&closed_notify); @@ -64,12 +71,30 @@ impl WireguardLink { if msg.len() > MAX_UDP_DATAGRAM_SIZE { continue; } - let target = match target_r.read() { - Ok(g) => *g, - Err(p) => *p.into_inner(), - }; - if let Err(err) = socket_w.send_to(&msg, target).await { - tracing::trace!(%err, %target, "failed to send WireGuard UDP datagram"); + if confirmed_r.load(Ordering::Acquire) { + let target = match target_r.read() { + Ok(g) => *g, + Err(p) => *p.into_inner(), + }; + if let Err(err) = socket_w.send_to(&msg, target).await { + tracing::trace!(%err, %target, "failed to send WireGuard UDP datagram"); + } + } else { + let targets = match candidates_r.read() { + Ok(g) => g.clone(), + Err(p) => p.into_inner().clone(), + }; + if targets.is_empty() { + let target = match target_r.read() { + Ok(g) => *g, + Err(p) => *p.into_inner(), + }; + let _ = socket_w.send_to(&msg, target).await; + } else { + for target in targets { + let _ = socket_w.send_to(&msg, target).await; + } + } } } is_closed_w.store(true, Ordering::Release); @@ -80,6 +105,8 @@ impl WireguardLink { network, peer, target_addr: target_holder, + candidates: candidates_holder, + confirmed, tx: tx_out, inbound_tx: tx_in, rx: tokio::sync::Mutex::new(rx_in), @@ -93,12 +120,24 @@ impl WireguardLink { if self.is_closed.load(Ordering::Acquire) { return false; } - // If queue is full, dropping is normal for an unreliable datagram link + self.confirmed.store(true, Ordering::Release); self.inbound_tx.try_send(payload).is_ok() } + /// Updates candidate addresses for this peer. + pub fn update_candidates(&self, addrs: impl IntoIterator) { + if let Ok(mut guard) = self.candidates.write() { + for addr in addrs { + if !guard.contains(&addr) { + guard.push(addr); + } + } + } + } + /// Updates the target socket address (roaming / NAT hole punch update). pub fn update_target(&self, new_target: SocketAddr) { + self.confirmed.store(true, Ordering::Release); if let Ok(mut guard) = self.target_addr.write() && *guard != new_target { @@ -177,7 +216,7 @@ pub struct WireguardTransport { listen_port: u16, socket: Arc, peer_addresses: RwLock>>, - peer_networks: RwLock>, + peer_networks: RwLock>>, links: RwLock>>, inbound_cb: Arc>>, } @@ -233,12 +272,15 @@ impl WireguardTransport { /// Associates a peer with a network ID. pub fn set_peer_network(&self, peer: EndpointId, network: NetworkId) { if let Ok(mut guard) = self.peer_networks.write() { - guard.insert(peer, network); + guard.entry(peer).or_default().insert(network); } } /// Records an observed UDP target address for a peer. pub fn set_peer_addr(&self, peer: EndpointId, addr: SocketAddr) { + if !addr.is_ipv4() { + return; + } if let Ok(mut guard) = self.peer_addresses.write() { let addrs = guard.entry(peer).or_default(); addrs.retain(|a| *a != addr); @@ -251,7 +293,7 @@ impl WireguardTransport { if let Ok(mut guard) = self.peer_addresses.write() { let entry = guard.entry(peer).or_default(); for addr in addrs { - if !entry.contains(&addr) { + if addr.is_ipv4() && !entry.contains(&addr) { entry.push(addr); } } @@ -278,12 +320,18 @@ impl WireguardTransport { addrs: &[SocketAddr], we_accept: bool, ) { - if let Ok(mut guard) = self.peer_networks.write() { - guard.insert(peer, network); - } - self.set_peer_addrs(peer, addrs.iter().copied()); + self.set_peer_network(peer, network); + let ipv4_addrs: Vec = addrs.iter().copied().filter(|a| a.is_ipv4()).collect(); + self.set_peer_addrs(peer, ipv4_addrs.iter().copied()); - if we_accept && !addrs.is_empty() { + // Update candidate addresses on any existing link for this peer + if let Ok(guard) = self.links.read() + && let Some(link) = guard.get(&(network, peer)) + { + link.update_candidates(ipv4_addrs.iter().copied()); + } + + if we_accept && !ipv4_addrs.is_empty() { let mut guard = match self.links.write() { Ok(g) => g, Err(p) => p.into_inner(), @@ -292,7 +340,8 @@ impl WireguardTransport { let link = Arc::new(WireguardLink::new( network, peer, - addrs[0], + ipv4_addrs[0], + ipv4_addrs, Arc::clone(&self.socket), )); entry.insert(Arc::clone(&link)); @@ -333,89 +382,109 @@ impl WireguardTransport { while let Ok((len, src_addr)) = socket.recv_from(&mut buf).await { let datagram = Bytes::copy_from_slice(&buf[..len]); - // 1. Check if an active link exists matching this peer / target - let matched_link = { + // 1. Identify which peer this packet comes from. + let matched_peer = { + let links_guard = match self.links.read() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + links_guard + .values() + .find(|link| { + link.target_addr() == src_addr || link.target_addr().ip() == src_addr.ip() + }) + .map(|link| link.peer()) + .or_else(|| { + let addrs_guard = match self.peer_addresses.read() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + addrs_guard + .iter() + .find(|(_, addrs)| { + addrs + .iter() + .any(|a| *a == src_addr || a.ip() == src_addr.ip()) + }) + .map(|(peer, _)| *peer) + }) + }; + + let Some(peer) = matched_peer else { + continue; + }; + + // 2. Deliver to ALL active links for this peer (across all networks this peer shares with us). + let peer_links: Vec> = { let guard = match self.links.read() { Ok(g) => g, Err(p) => p.into_inner(), }; guard .values() - .find(|link| { - link.target_addr() == src_addr || link.target_addr().ip() == src_addr.ip() - }) + .filter(|l| l.peer() == peer && !l.is_closed()) .cloned() + .collect() }; - if let Some(link) = matched_link { - link.update_target(src_addr); - link.deliver_inbound(datagram); + if !peer_links.is_empty() { + for link in peer_links { + link.update_target(src_addr); + link.deliver_inbound(datagram.clone()); + } continue; } - // 2. Check if a peer has candidate addresses matching src_addr - let matched_peer = { - let guard = match self.peer_addresses.read() { + // 3. If no active link exists for this peer yet, create inbound links for all networks known for this peer. + let networks: Vec = { + let guard = match self.peer_networks.read() { Ok(g) => g, Err(p) => p.into_inner(), }; guard - .iter() - .find(|(_, addrs)| { - addrs - .iter() - .any(|a| *a == src_addr || a.ip() == src_addr.ip()) - }) - .map(|(peer, _)| *peer) + .get(&peer) + .map(|s| s.iter().copied().collect()) + .unwrap_or_default() }; - if let Some(peer) = matched_peer { - let network = { - let guard = match self.peer_networks.read() { + for network in networks { + let link = { + let mut guard = match self.links.write() { Ok(g) => g, Err(p) => p.into_inner(), }; - guard.get(&peer).copied().or_else(|| { - if guard.len() == 1 { - guard.values().next().copied() - } else { - match self.links.read() { - Ok(g) => g.keys().next().map(|(net, _)| *net), - Err(p) => p.into_inner().keys().next().map(|(net, _)| *net), - } - } - }) - }; - - if let Some(network) = network { + if let Some(existing) = guard.get(&(network, peer)) + && !existing.is_closed() + { + existing.update_target(src_addr); + existing.deliver_inbound(datagram.clone()); + continue; + } + let candidates = self.peer_addrs(peer); let link = Arc::new(WireguardLink::new( network, peer, src_addr, + candidates, Arc::clone(&self.socket), )); - link.deliver_inbound(datagram); + link.update_target(src_addr); + link.deliver_inbound(datagram.clone()); + guard.insert((network, peer), Arc::clone(&link)); + link + }; - { - let mut guard = match self.links.write() { - Ok(g) => g, - Err(p) => p.into_inner(), - }; - guard.insert((network, peer), Arc::clone(&link)); - } - - let cb_guard = match self.inbound_cb.lock() { - Ok(g) => g, - Err(p) => p.into_inner(), - }; - if let Some(ref cb) = *cb_guard { - cb(InboundLink { - network, - peer, - protocol: crate::announcement::WG_PROTOCOL.to_string(), - link, - }); - } + let cb_guard = match self.inbound_cb.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + if let Some(ref cb) = *cb_guard { + cb(InboundLink { + network, + peer, + protocol: crate::announcement::WG_PROTOCOL.to_string(), + link, + }); } } } @@ -454,12 +523,14 @@ impl PacketTransport for WireguardTransport { if let Some(existing) = guard.get(&(network, peer)) && !existing.is_closed() { + existing.update_candidates(addrs); return Ok(Arc::clone(existing) as SharedLink); } let link = Arc::new(WireguardLink::new( network, peer, target, + addrs, Arc::clone(&self.socket), )); guard.insert((network, peer), Arc::clone(&link)); diff --git a/crates/tsunagi-wireguard/tests/plugin_tests.rs b/crates/tsunagi-wireguard/tests/plugin_tests.rs index b9acdd6..0a5cbc5 100644 --- a/crates/tsunagi-wireguard/tests/plugin_tests.rs +++ b/crates/tsunagi-wireguard/tests/plugin_tests.rs @@ -204,3 +204,175 @@ async fn test_wireguard_plugin_end_to_end() { assert_eq!(received_a.1, peer_b); assert_eq!(received_a.2, packet_b_to_a); } + +#[tokio::test] +async fn test_wireguard_two_networks_same_peers() { + let identity_a = DeviceIdentity::generate(); + let peer_a = identity_a.endpoint_id(); + + let identity_b = DeviceIdentity::generate(); + let peer_b = identity_b.endpoint_id(); + + let net1 = NetworkId::from_bytes([11u8; 32]); + let net2 = NetworkId::from_bytes([22u8; 32]); + + let tmp_a = tempfile::tempdir().expect("tmp A"); + let tmp_b = tempfile::tempdir().expect("tmp B"); + + let transport_b = Arc::new( + WireguardTransport::bind(&identity_b, Some(0)) + .await + .expect("bind B"), + ); + let port_b = transport_b.bound_port(); + + let transport_a = Arc::new( + WireguardTransport::bind(&identity_a, Some(0)) + .await + .expect("bind A"), + ); + let port_a = transport_a.bound_port(); + + let plugin_a = + WireguardPlugin::open(WireguardConfig::new(tmp_a.path()), Arc::clone(&transport_a)) + .await + .expect("open plugin A"); + + let plugin_b = + WireguardPlugin::open(WireguardConfig::new(tmp_b.path()), Arc::clone(&transport_b)) + .await + .expect("open plugin B"); + + let (sink_a_tx, mut sink_a_rx) = mpsc::channel(20); + plugin_a.attach(PluginContext::with_sink(Arc::new(TestSink { + delivered: sink_a_tx, + }))); + + let (sink_b_tx, mut sink_b_rx) = mpsc::channel(20); + plugin_b.attach(PluginContext::with_sink(Arc::new(TestSink { + delivered: sink_b_tx, + }))); + + // Activate both networks + for net in [net1, net2] { + plugin_a.on_network_activated(net); + plugin_b.on_network_activated(net); + } + + // Exchange capabilities for both networks + for net in [net1, net2] { + let cap_a = loop { + if let Some(cap) = plugin_a.local_capability(net).expect("cap A") { + break cap; + } + tokio::time::sleep(Duration::from_millis(20)).await; + }; + let cap_b = loop { + if let Some(cap) = plugin_b.local_capability(net).expect("cap B") { + break cap; + } + tokio::time::sleep(Duration::from_millis(20)).await; + }; + plugin_a + .on_peer_capability(net, peer_b, &cap_b) + .expect("A on cap B"); + plugin_b + .on_peer_capability(net, peer_a, &cap_a) + .expect("B on cap A"); + } + + let addr_b: SocketAddr = format!("127.0.0.1:{port_b}").parse().unwrap(); + transport_a.set_peer_addr(peer_b, addr_b); + + let addr_a: SocketAddr = format!("127.0.0.1:{port_a}").parse().unwrap(); + transport_b.set_peer_addr(peer_a, addr_a); + + // Accept loops + let transport_a_clone = Arc::clone(&transport_a); + tokio::spawn(async move { + transport_a_clone.accept_loop(|_| {}).await; + }); + + let (inbound_tx, mut inbound_rx) = mpsc::channel::(10); + let transport_b_clone = Arc::clone(&transport_b); + tokio::spawn(async move { + transport_b_clone + .accept_loop(move |inbound| { + let _ = inbound_tx.try_send(inbound); + }) + .await; + }); + + // Open links for both networks from A to B + for net in [net1, net2] { + let link_a = transport_a + .open(net, peer_b, "wg") + .await + .expect("open A link"); + plugin_a.on_peer_link(net, peer_b, link_a); + } + + // Deliver inbound links to B + for _ in 0..2 { + let inbound = tokio::time::timeout(Duration::from_secs(5), inbound_rx.recv()) + .await + .expect("inbound timeout") + .expect("inbound option"); + plugin_b.on_peer_link(inbound.network, peer_a, inbound.link); + } + + // Wait for both handshakes to complete on both networks + for net in [net1, net2] { + let mut up = false; + for _ in 0..50 { + if let Some(overview) = plugin_a.overview(net) + && overview.peers.iter().any(|p| p.is_up()) + { + up = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(up, "handshake on {net:?} should complete"); + } + + // Send on net1 + let pkt1 = sample_ipv4_packet(); + assert!(plugin_a.carry(net1, peer_b, pkt1.clone())); + let recv1 = tokio::time::timeout(Duration::from_secs(5), sink_b_rx.recv()) + .await + .expect("recv1 timeout") + .expect("recv1 option"); + assert_eq!(recv1.0, net1); + assert_eq!(recv1.2, pkt1); + + // Send on net2 from A to B + let mut pkt2_bytes = sample_ipv4_packet().to_vec(); + pkt2_bytes[15] = 99; // different byte + let pkt2 = Bytes::from(pkt2_bytes); + assert!(plugin_a.carry(net2, peer_b, pkt2.clone())); + let recv2 = tokio::time::timeout(Duration::from_secs(5), sink_b_rx.recv()) + .await + .expect("recv2 timeout") + .expect("recv2 option"); + assert_eq!(recv2.0, net2); + assert_eq!(recv2.2, pkt2); + + // Send reply on net1 from B to A + assert!(plugin_b.carry(net1, peer_a, pkt1.clone())); + let reply1 = tokio::time::timeout(Duration::from_secs(5), sink_a_rx.recv()) + .await + .expect("reply1 timeout") + .expect("reply1 option"); + assert_eq!(reply1.0, net1); + assert_eq!(reply1.2, pkt1); + + // Send reply on net2 from B to A + assert!(plugin_b.carry(net2, peer_a, pkt2.clone())); + let reply2 = tokio::time::timeout(Duration::from_secs(5), sink_a_rx.recv()) + .await + .expect("reply2 timeout") + .expect("reply2 option"); + assert_eq!(reply2.0, net2); + assert_eq!(reply2.2, pkt2); +}