Added DHT peer resolver, fixed MTU

crates/tsunagi-cli/src/main.rs

@@ -15,7 +15,9 @@ use clap::{Args, Parser, Subcommand, ValueEnum};
 use tsunagi::agent::Event;
 use tsunagi::config::{AgentConfig, StoragePaths, TransportPolicy};
 use tsunagi::dataplane::IpPlugin;
-use tsunagi::discovery::{CompositeDiscovery, NetworkDiscovery, StaticBootstrap};
+use tsunagi::discovery::{
+    CompositeDiscovery, MainlineDiscovery, NetworkDiscovery, StaticBootstrap,
+};
 use tsunagi::identity::{NetworkName, NetworkSecret};
 use tsunagi::iroh_types::EndpointAddr;
 use tsunagi::overlay::{MemoryTunFactory, TunFactory};
@@ -347,9 +349,17 @@ struct UpArgs {
     #[arg(long, value_enum, default_value_t = Reach::Relay, help_heading = "System")]
     reach: Reach,
 
+    /// Enable Mainline DHT rendezvous (already enabled by default).
+    #[arg(long, conflicts_with = "no_dht", help_heading = "System")]
+    dht: bool,
+
+    /// Disable Mainline DHT lookup and publication. --reach local also disables it.
+    #[arg(long, conflicts_with = "dht", help_heading = "System")]
+    no_dht: bool,
+
     /// A peer to contact, as `<endpoint-id>` or `<endpoint-id>@<ip:port>,...`.
     ///
-    /// One agent needs to know another to begin with. Repeat for several.
+    /// Optional alongside DHT discovery. Repeat for several.
     #[arg(long = "peer", value_name = "PEER", help_heading = "System")]
     peers: Vec<String>,
 
@@ -470,7 +480,11 @@ fn load_secret(
 /// the tests, can each take a port of their own); this only decides what a
 /// plain `--dns` picks. Elsewhere the default is a high port that needs no
 /// privilege, since systemd-resolved can be pointed at any port.
-const DEFAULT_DNS_PORT: u16 = if cfg!(target_os = "windows") { 53 } else { 5354 };
+const DEFAULT_DNS_PORT: u16 = if cfg!(target_os = "windows") {
+    53
+} else {
+    5354
+};
 
 /// Settings key: whether the local resolver is wanted.
 const DNS_ENABLED: &str = "dns.enabled";
@@ -1546,8 +1560,7 @@ async fn join_network(
     // next restart.
     if tsunagi::ipc::is_serving(socket).await {
         let report =
-            tsunagi::ipc::join_network(socket, name.as_str(), secret.encode().as_str())
-                .await?;
+            tsunagi::ipc::join_network(socket, name.as_str(), secret.encode().as_str()).await?;
         // The id in full either way: it is what every other command takes,
         // and the shortened form in a report is for reading, not copying.
         match standing {
@@ -1644,8 +1657,8 @@ async fn invite(socket: &std::path::Path, name: &NetworkName, secret: &NetworkSe
     );
     match endpoint {
         Some(endpoint) => println!(
-            "\nIts agent has to be running. If it is not:\n\n  \
-             tsunagi up --peer {endpoint}"
+            "\nStart its agent with `tsunagi up`; DHT discovery is enabled by default.\n\n  \
+             Optional manual bootstrap: tsunagi up --peer {endpoint}"
         ),
         None => println!("\nIts agent has to be running: `tsunagi up`."),
     }
@@ -2566,10 +2579,8 @@ fn network_section(
                 "members",
                 "none: this network has no range to allocate from",
             )),
-            // Nobody to contact and nowhere to look. An agent finds a peer
-            // by being told about one, or from what it remembers of an
-            // earlier session — with neither it waits for ever, and the
-            // report should say so rather than imply patience.
+            // No candidates yet. DHT may still be bootstrapping; manual
+            // bootstrap remains useful when public UDP is unavailable.
             (_, _, 0) => section.push(
                 Row::new(
                     Health::Degraded,
@@ -2577,9 +2588,9 @@ fn network_section(
                     "none, and nobody to contact: no candidates in this network",
                 )
                 .with_note(format!(
-                    "somebody has to make the introduction. Start this agent with \
-                     `--peer <their-endpoint-id>`, or have them start theirs with \
-                     `--peer {}`. Once they have met, each remembers the other.",
+                    "DHT lookup retries automatically when enabled. For manual \
+                     bootstrap, start with `--peer <their-endpoint-id>`, or have \
+                     the other device start with `--peer {}`.",
                     short(own_id, 12)
                 )),
             ),
@@ -3233,6 +3244,10 @@ async fn netwatch_addresses() -> Vec<std::net::IpAddr> {
     addresses
 }
 
+fn dht_enabled(args: &UpArgs) -> bool {
+    !matches!(args.reach, Reach::Local) && (args.dht || !args.no_dht)
+}
+
 async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
     let paths = args.paths.resolve()?;
 
@@ -3273,6 +3288,9 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
         .with_transport(args.reach.into())
         .with_discovery(discovery)
         .with_discovery_interval(Duration::from_secs(5));
+    if dht_enabled(&args) {
+        config = config.with_dht(MainlineDiscovery::default());
+    }
     if let Some(hostname) = &args.hostname {
         config = config.with_hostname(hostname.clone());
     }
@@ -4157,6 +4175,26 @@ mod status_tests {
 #[cfg(test)]
 mod network_tests {
     #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+    #[test]
+    fn dht_defaults_and_local_mode_do_not_require_manual_peers() {
+        use clap::Parser;
+        for (flags, expected) in [
+            (vec![], true),
+            (vec!["--dht"], true),
+            (vec!["--no-dht"], false),
+            (vec!["--reach", "local"], false),
+            (vec!["--reach", "local", "--dht"], false),
+            (vec!["--reach", "direct"], true),
+        ] {
+            let cli =
+                super::Cli::try_parse_from(["tsunagi", "up"].into_iter().chain(flags)).unwrap();
+            let super::Command::Up(args) = cli.command else {
+                panic!("expected up");
+            };
+            assert_eq!(super::dht_enabled(&args), expected);
+        }
+        assert!(super::Cli::try_parse_from(["tsunagi", "up", "--dht", "--no-dht"]).is_err());
+    }
 
     use super::*;
     use tsunagi::identity::{NetworkKeys, NetworkName, NetworkSecret};

crates/tsunagi-wg-quic/src/plugin.rs

@@ -58,23 +58,22 @@ pub const WIREGUARD_PROTOCOL: &str = "wg-quic";
 /// Smallest interface MTU the overlay accepts.
 ///
 /// 576 bytes is what IPv4 guarantees every host can reassemble (RFC 1122),
-/// so nothing below it is worth offering. The floor used to be 1280 because
-/// Linux tears IPv6 down on an interface below that; the overlay is IPv4
-/// now, so that constraint is gone and a path with small datagrams — a
-/// relay, typically — can be matched instead of warned about.
+/// so nothing below it is worth offering. The current overlay is IPv4.
+/// Lowering this is not needed to accommodate a narrow QUIC path: the default
+/// transport fragments opaque payloads below the plugin.
 pub const MIN_MTU: u32 = 576;
 
 /// Default interface MTU.
 ///
-/// Comfortably under what a direct path carries, and the same number the
-/// overlay used before, so an existing network does not have to change.
+/// Kept stable across path changes. The default transport splits encrypted
+/// packets when the current QUIC path cannot carry them in one datagram.
 pub const DEFAULT_MTU: u32 = 1280;
 
 /// Bytes WireGuard adds to a packet: type and reserved, receiver index,
 /// counter and the Poly1305 tag.
 ///
-/// A link therefore has to carry `mtu + WIREGUARD_OVERHEAD` bytes in one
-/// datagram for a full-size packet to get through.
+/// A link must carry `mtu + WIREGUARD_OVERHEAD` logical payload bytes.
+/// Transport fragmentation is independent of WireGuard and the inner IP flags.
 pub const WIREGUARD_OVERHEAD: u32 = 32;
 
 /// Configuration of the WireGuard plugin.
@@ -87,7 +86,8 @@ pub struct WireguardConfig {
     /// The largest packet a tunnel will carry.
     ///
     /// Not the interface MTU, which belongs to the agent: this is what this
-    /// protocol refuses to encrypt because it would not fit one datagram.
+    /// protocol is configured to carry. The transport may split ciphertext
+    /// into smaller datagrams without changing the original IP packet.
     pub mtu: u32,
     /// How long to coalesce changes before reconciling.
     pub reconcile_debounce: Duration,

crates/tsunagi-wg-quic/tests/wireguard.rs

@@ -23,12 +23,26 @@ use tsunagi::overlay::{
     MemoryTun, MemoryTunFactory, OverlayError, TunDevice, TunFactory, TunRequest,
 };
 use tsunagi::state::Ipv4Range;
-use tsunagi::testing::{config_with, network, settle, wait_event, wait_for_peers, wait_until};
+use tsunagi::testing::{network, settle, wait_event, wait_for_peers, wait_until};
 use tsunagi::{Agent, NetworkStatus};
 use tsunagi_wg_quic::{
     WIREGUARD_PROTOCOL, WgAnnouncement, WgSecretKey, WireguardConfig, WireguardPlugin,
 };
 
+/// Real QUIC at its minimum path MTU: loopback PMTU discovery must not hide
+/// failures that occur on ordinary Internet paths. The TUN MTU stays at 1280.
+fn config_with(root: &std::path::Path, discovery: &SharedMemoryDiscovery) -> tsunagi::AgentConfig {
+    let mut config = tsunagi::testing::config_with(root, discovery);
+    config.test_quic_transport = Some(
+        iroh::endpoint::QuicTransportConfig::builder()
+            .initial_mtu(1200)
+            .min_mtu(1200)
+            .mtu_discovery_config(None)
+            .build(),
+    );
+    config
+}
+
 /// A factory that claims the host and puts nothing on it.
 ///
 /// This is the case the missing-address report exists for: a provisioner
@@ -291,6 +305,84 @@ fn ipv4_packet(source: Ipv4Addr, destination: Ipv4Addr, payload: &[u8]) -> Bytes
     Bytes::from(packet)
 }
 
+#[tokio::test]
+async fn full_size_tcp_packets_cross_the_default_overlay() {
+    let discovery = SharedMemoryDiscovery::new();
+    let (name, secret) = network("wg-default-mtu");
+    let a = WgAgent::spawn(&discovery, "tmtua").await;
+    let b = WgAgent::spawn(&discovery, "tmtub").await;
+    let id = a.agent.join_network(&name, &secret).await.unwrap();
+    b.agent.join_network(&name, &secret).await.unwrap();
+    a.wait_for_tunnels(id, 1).await;
+    b.wait_for_tunnels(id, 1).await;
+    let addr_a = a.overlay(id).await;
+    let addr_b = b.overlay(id).await;
+    let tun_a = a.tun(id).await;
+    let tun_b = b.tun(id).await;
+    assert_eq!(tun_a.mtu(), tsunagi_wg_quic::DEFAULT_MTU);
+    for size in [1280, 1279, 1098, 1097, 64] {
+        for (source, dest, from, to) in [
+            (addr_a, addr_b, &tun_a, &tun_b),
+            (addr_b, addr_a, &tun_b, &tun_a),
+        ] {
+            let packet = tcp_packet(source, dest, size);
+            from.push_from_os(packet.clone());
+            let received = tokio::time::timeout(Duration::from_secs(5), to.pop_to_os())
+                .await
+                .expect("a full-size TCP packet must arrive at the default MTU")
+                .unwrap();
+            assert_eq!(received, packet, "including DF, TCP flags and checksums");
+        }
+    }
+    for peer in [&a, &b] {
+        let view = peer.plugin.overview(id).unwrap();
+        assert_eq!(
+            view.peers[0]
+                .tunnel
+                .as_ref()
+                .unwrap()
+                .stats
+                .dropped_oversize,
+            0
+        );
+    }
+    a.shutdown().await;
+    b.shutdown().await;
+}
+
+/// A checksummed IPv4/TCP segment with DF set, as a host's TCP stack sends it.
+fn tcp_packet(source: Ipv4Addr, destination: Ipv4Addr, size: usize) -> Bytes {
+    fn checksum(bytes: &[u8]) -> u16 {
+        let mut sum: u32 = bytes
+            .chunks(2)
+            .map(|c| u16::from_be_bytes([c[0], *c.get(1).unwrap_or(&0)]) as u32)
+            .sum();
+        while sum > 0xffff {
+            sum = (sum & 0xffff) + (sum >> 16);
+        }
+        !(sum as u16)
+    }
+    assert!(size >= 40);
+    let mut packet = ipv4_packet(source, destination, &vec![0x5a; size - 20]).to_vec();
+    packet[6..8].copy_from_slice(&0x4000u16.to_be_bytes()); // don't fragment
+    packet[9] = 6; // TCP
+    packet[20..40].fill(0);
+    packet[20..22].copy_from_slice(&40000u16.to_be_bytes());
+    packet[22..24].copy_from_slice(&22u16.to_be_bytes());
+    packet[24..28].copy_from_slice(&1u32.to_be_bytes());
+    packet[32] = 5 << 4;
+    packet[33] = 0x18; // PSH, ACK
+    packet[34..36].copy_from_slice(&65535u16.to_be_bytes());
+    let mut pseudo = Vec::from(&packet[12..20]);
+    pseudo.extend_from_slice(&[0, 6]);
+    pseudo.extend_from_slice(&((size - 20) as u16).to_be_bytes());
+    pseudo.extend_from_slice(&packet[20..]);
+    packet[36..38].copy_from_slice(&checksum(&pseudo).to_be_bytes());
+    let ip_checksum = checksum(&packet[..20]);
+    packet[10..12].copy_from_slice(&ip_checksum.to_be_bytes());
+    Bytes::from(packet)
+}
+
 #[tokio::test]
 async fn two_agents_carry_real_ip_packets_through_a_wireguard_tunnel() {
     let discovery = SharedMemoryDiscovery::new();
@@ -1404,7 +1496,7 @@ async fn a_peer_with_no_direct_path_is_reached_through_one_that_has_both() {
     // And a real packet crosses: A's interface to B's interface, through C.
     a.tun(network_id)
         .await
-        .push_from_os(ipv4_packet(a_addr, b_addr, b"through the middle"));
+        .push_from_os(tcp_packet(a_addr, b_addr, 1280));
     let seen = tokio::time::timeout(
         tsunagi::testing::DEADLINE,
         b.tun(network_id).await.pop_to_os(),
@@ -1412,7 +1504,7 @@ async fn a_peer_with_no_direct_path_is_reached_through_one_that_has_both() {
     .await
     .expect("the packet should arrive")
     .unwrap();
-    assert_eq!(&seen[20..], b"through the middle");
+    assert_eq!(seen, tcp_packet(a_addr, b_addr, 1280));
 
     a.shutdown().await;
     b.shutdown().await;

crates/tsunagi/Cargo.toml

@@ -65,6 +65,8 @@ directories = "6.0"
 # forbidden `unsafe` and will not make one.
 gethostname = "1.1"
 netwatch = "0.19.3"
+mainline = "8.0"
+futures-lite = "2.6"
 tun = { version = "0.8", features = ["async"], optional = true }
 # Only for the `testing` harness, which a protocol crate's tests use too.
 tempfile = { version = "3.24", optional = true }
@@ -94,7 +96,7 @@ netdev = { version = "0.45", optional = true }
 [dev-dependencies]
 # Its own tests use the harness it publishes.
 tsunagi = { path = ".", features = ["testing"] }
-tokio = { version = "1.53", features = ["rt", "rt-multi-thread", "sync", "time", "macros", "process"] }
+tokio = { version = "1.53", features = ["rt", "rt-multi-thread", "sync", "time", "macros", "process", "test-util"] }
 tempfile.workspace = true
 tracing-subscriber.workspace = true
 

crates/tsunagi/src/agent/mod.rs

@@ -552,6 +552,21 @@ impl Agent {
         }
 
         let range = self.reserve_range(network_id);
+        let mut backends = Vec::new();
+        if let Some(discovery) = &self.inner.config.discovery {
+            backends.push(discovery.clone());
+        }
+        if let Some(dht) = &self.inner.config.dht {
+            backends.push(dht.for_network(&keys));
+        }
+        let discovery = if backends.is_empty() {
+            None
+        } else {
+            Some(
+                Arc::new(crate::discovery::CompositeDiscovery::new(backends))
+                    as Arc<dyn crate::discovery::NetworkDiscovery>,
+            )
+        };
         let handle = network::spawn(RuntimeParams {
             keys,
             adapter: self.inner.adapter.clone(),
@@ -559,8 +574,9 @@ impl Agent {
             events: self.inner.events.clone(),
             limits: Arc::clone(&self.inner.limits),
             reconnect: self.inner.config.reconnect.clone(),
-            discovery: self.inner.config.discovery.clone(),
+            discovery,
             discovery_interval: self.inner.config.discovery_interval,
+            discovery_policy: self.inner.config.discovery_policy.clone(),
             plugins: self.inner.config.plugins.clone(),
             routes: Arc::clone(&self.inner.routes),
             interface: self.inner.interface.get().cloned(),
@@ -798,7 +814,8 @@ impl Agent {
         self.command(network_id, NetCommand::Reannounce).await
     }
 
-    /// Asks one network to re-run discovery and re-evaluate dials right now.
+    /// Asks one network to re-evaluate known peers and dials right now.
+    /// External lookup still follows its connected/isolation policy.
     ///
     /// Call this when the host's network environment changed. Platform wake-up
     /// notifications can be wired to it later.
@@ -806,7 +823,7 @@ impl Agent {
         self.command(network_id, NetCommand::Recheck).await
     }
 
-    /// Asks every running network to re-run discovery right now.
+    /// Asks every running network to re-evaluate known peers right now.
     pub async fn recheck(&self) {
         let senders: Vec<mpsc::Sender<NetCommand>> = self
             .inner
@@ -835,6 +852,10 @@ impl Agent {
             handle.stop().await;
         }
 
+        if let Some(dht) = &self.inner.config.dht {
+            dht.shutdown().await;
+        }
+
         // iroh's own close waits for peers to acknowledge; a peer that has
         // gone silent must not decide how long that takes.
         if tokio::time::timeout(TASK_GRACE, self.inner.adapter.close())

crates/tsunagi/src/agent/network.rs

@@ -14,9 +14,10 @@ use iroh::endpoint::{Connection, RecvStream, SendStream};
 use tokio::sync::{broadcast, mpsc, oneshot};
 use tokio::task::JoinHandle;
 
-use crate::config::{Limits, ReconnectPolicy};
+use crate::config::{DiscoveryPolicy, Limits, ReconnectPolicy};
 use crate::dataplane::transport::{InboundLink, PacketLink, PacketTransport, SharedLink};
 use crate::dataplane::{PluginCapability, SharedPlugin};
+use crate::discovery::worker::DiscoveryWorker;
 use crate::discovery::{Candidate, CandidateSource, NetworkDiscovery};
 use crate::error::{Error, Result};
 use crate::identity::{NetworkId, NetworkKeys};
@@ -129,6 +130,7 @@ pub(crate) struct RuntimeParams {
     pub(crate) reconnect: ReconnectPolicy,
     pub(crate) discovery: Option<Arc<dyn NetworkDiscovery>>,
     pub(crate) discovery_interval: Duration,
+    pub(crate) discovery_policy: DiscoveryPolicy,
     pub(crate) plugins: Vec<SharedPlugin>,
     /// Who holds which overlay address, shared with every other network.
     pub(crate) routes: Arc<crate::overlay::RoutingTable>,
@@ -232,6 +234,8 @@ const REACH_EXPIRY: Duration = Duration::from_secs(90);
 const RANGE_PROPOSAL_GRACE: Duration = Duration::from_secs(3);
 
 struct Runtime {
+    discovery_worker: Option<DiscoveryWorker>,
+    discovery_candidates: mpsc::Receiver<Candidate>,
     params: RuntimeParams,
     /// When this runtime started, for the fallback range's grace period.
     activated: std::time::Instant,
@@ -295,7 +299,21 @@ impl Runtime {
         let (session_events_tx, session_events_rx) = mpsc::channel(256);
         let (dial_results_tx, dial_results_rx) = mpsc::channel(64);
         let (link_results_tx, link_results_rx) = mpsc::channel(64);
+        let (candidate_tx, discovery_candidates) =
+            mpsc::channel(params.limits.max_discovery_candidates.max(1));
+        let discovery_worker = params.discovery.as_ref().map(|backend| {
+            DiscoveryWorker::spawn(
+                backend.clone(),
+                params.keys.discovery_key(),
+                params.adapter.clone(),
+                params.discovery_policy.clone(),
+                params.discovery_interval,
+                candidate_tx,
+            )
+        });
         Self {
+            discovery_worker,
+            discovery_candidates,
             params,
             activated: std::time::Instant::now(),
             network_id,
@@ -362,6 +380,10 @@ impl Runtime {
                         self.handle_link_result(result);
                     }
                 }
+                Some(candidate) = self.discovery_candidates.recv() => {
+                    self.add_candidate(candidate);
+                    self.start_dials();
+                }
                 _ = ticker.tick() => {
                     self.discovery_round().await;
                     self.ensure_links();
@@ -373,14 +395,12 @@ impl Runtime {
     }
 
     async fn teardown(&mut self) {
+        if let Some(worker) = self.discovery_worker.take() {
+            worker.stop().await;
+        }
         // Stop accepting session events first: nothing is going to act on them
         // any more, and a sender blocked on a full queue would stall shutdown.
         self.session_events_rx.close();
-        if let Some(discovery) = &self.params.discovery {
-            let _ = discovery
-                .unpublish(self.params.keys.discovery_key(), self.local_id)
-                .await;
-        }
         self.links.clear();
         let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
         for peer in peers {
@@ -528,25 +548,6 @@ impl Runtime {
 
         let mut candidates: Vec<Candidate> = Vec::new();
 
-        if let Some(discovery) = self.params.discovery.clone() {
-            let key = self.params.keys.discovery_key();
-            // Publishing every round keeps a restarted agent reachable at its
-            // new local port without any special case.
-            if let Err(err) = discovery.publish(key, self.params.adapter.addr()).await {
-                tracing::debug!(%err, "discovery publish failed");
-            }
-            if let Err(err) = discovery
-                .publish(key, self.params.adapter.loopback_addr())
-                .await
-            {
-                tracing::debug!(%err, "discovery publish of bound sockets failed");
-            }
-            match discovery.resolve(key).await {
-                Ok(found) => candidates.extend(found),
-                Err(err) => tracing::debug!(%err, "discovery resolve failed"),
-            }
-        }
-
         // Everybody the signed state says belongs here. Their records
         // reached us through somebody, so we know they exist and who they
         // are, even having never spoken to them; an id with no address is
@@ -577,17 +578,7 @@ impl Runtime {
         }
 
         for candidate in candidates {
-            let peer = candidate.endpoint_id();
-            if peer == self.local_id {
-                continue;
-            }
-            self.candidate_addrs
-                .entry(peer)
-                .and_modify(|existing| merge_addr(existing, &candidate.addr))
-                .or_insert_with(|| candidate.addr.clone());
-            self.dial_states
-                .entry(peer)
-                .or_insert_with(|| DialState::new(candidate.source));
+            self.add_candidate(candidate);
         }
 
         self.start_dials();
@@ -595,6 +586,50 @@ impl Runtime {
         self.introduce_peers();
     }
 
+    fn add_candidate(&mut self, candidate: Candidate) {
+        let peer = candidate.endpoint_id();
+        if peer == self.local_id {
+            return;
+        }
+        if candidate.source == CandidateSource::Discovery && !self.dial_states.contains_key(&peer) {
+            let count = self
+                .dial_states
+                .values()
+                .filter(|s| s.source == CandidateSource::Discovery)
+                .count();
+            if count >= self.params.limits.max_discovery_candidates {
+                let replace = self
+                    .dial_states
+                    .iter()
+                    .filter(|(id, s)| {
+                        s.source == CandidateSource::Discovery
+                            && !s.in_flight
+                            && !self.sessions.contains_key(*id)
+                    })
+                    .max_by_key(|(_, s)| s.consecutive_failures)
+                    .map(|(id, _)| *id);
+                let Some(replace) = replace else {
+                    return;
+                };
+                self.dial_states.remove(&replace);
+                self.candidate_addrs.remove(&replace);
+            }
+        }
+        self.candidate_addrs
+            .entry(peer)
+            .and_modify(|existing| {
+                if candidate.source == CandidateSource::Discovery {
+                    *existing = candidate.addr.clone();
+                } else {
+                    merge_addr(existing, &candidate.addr);
+                }
+            })
+            .or_insert_with(|| candidate.addr.clone());
+        self.dial_states
+            .entry(peer)
+            .or_insert_with(|| DialState::new(candidate.source));
+    }
+
     fn start_dials(&mut self) {
         let now = Instant::now();
         let in_flight = self
@@ -1563,6 +1598,9 @@ impl Runtime {
 
         let snapshot = snapshot_connection(&conn);
         self.sessions.insert(peer, session);
+        if let Some(worker) = &self.discovery_worker {
+            worker.set_connected(true);
+        }
         self.metrics.sessions_established += 1;
 
         // Announce ourselves straight away so the peer learns our hostname and
@@ -1675,6 +1713,9 @@ impl Runtime {
                     session.abort();
                     session.conn.close(0u32.into(), b"session ended");
                 }
+                if let Some(worker) = &self.discovery_worker {
+                    worker.set_connected(!self.sessions.is_empty());
+                }
                 self.metrics.disconnects += 1;
                 self.drop_links_for(peer);
                 for plugin in &self.params.plugins {

crates/tsunagi/src/config.rs

@@ -11,12 +11,38 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use crate::dataplane::SharedPlugin;
-use crate::discovery::NetworkDiscovery;
+use crate::discovery::{MainlineDiscovery, NetworkDiscovery};
 use crate::error::{Error, Result};
 
 /// Qualifier/organisation/application triple used for platform directories.
 const APP_NAME: &str = "tsunagi";
 
+/// Rendezvous slots per network; this bounds a discovery sample, not membership.
+pub const DHT_SLOTS: u8 = 16;
+/// BEP44 permits 1000 bencoded bytes; a 996-byte string has a four-byte prefix.
+pub const DHT_MAX_VALUE: usize = 996;
+/// Maximum direct addresses in one rendezvous record.
+pub const DHT_MAX_ADDRS: usize = 8;
+/// Maximum relay URL bytes in a rendezvous record.
+pub const DHT_MAX_RELAY_LEN: usize = 512;
+/// Application freshness, independent of storage nodes' own expiration.
+pub const DHT_RECORD_TTL: Duration = Duration::from_secs(900);
+/// Clock skew tolerated when reading a recently published record.
+pub const DHT_CLOCK_SKEW: Duration = Duration::from_secs(120);
+
+/// Maximum reassembled data-transport payload, independent of the path MTU.
+pub const MAX_DATA_DATAGRAM: usize = 64 * 1024;
+/// Maximum incomplete datagrams held by one authenticated data link.
+pub const MAX_DATA_ASSEMBLIES: usize = 64;
+/// Maximum allocated payload bytes for incomplete datagrams on one data link.
+pub const MAX_DATA_REASSEMBLY_BYTES: usize = 256 * 1024;
+/// Maximum fragments of one logical datagram, including after a path MTU change.
+pub const MAX_DATA_FRAGMENTS: usize = 128;
+/// Time allowed to assemble a datagram; loss never stalls later datagrams.
+pub const DATA_REASSEMBLY_TTL: Duration = Duration::from_secs(5);
+/// Completed or discarded packet IDs retained to ignore late duplicate fragments.
+pub const DATA_RECENT_IDS: usize = 128;
+
 /// Where the two stores live.
 ///
 /// The mandatory state and the disposable cache are separate both logically and
@@ -137,6 +163,8 @@ pub struct Limits {
     pub write_timeout: Duration,
     /// Maximum simultaneous outbound dials per network.
     pub max_concurrent_dials: usize,
+    /// Maximum unverified discovery candidates retained per network.
+    pub max_discovery_candidates: usize,
     /// Maximum simultaneous authenticated sessions per network.
     pub max_sessions_per_network: usize,
     /// Maximum simultaneous inbound connections being handshaken.
@@ -163,6 +191,7 @@ impl Default for Limits {
             dial_timeout: Duration::from_secs(10),
             write_timeout: Duration::from_secs(30),
             max_concurrent_dials: 8,
+            max_discovery_candidates: 16,
             max_sessions_per_network: 64,
             max_inbound_handshakes: 32,
             session_send_queue: 64,
@@ -212,6 +241,33 @@ impl ReconnectPolicy {
     }
 }
 
+/// Scheduling of candidate lookup and independent self-publication.
+#[derive(Debug, Clone)]
+pub struct DiscoveryPolicy {
+    /// Mean interval between successful self-publications (20% jitter).
+    pub publish_interval: Duration,
+    /// Initial delay between unsuccessful bootstrap lookups.
+    pub lookup_interval: Duration,
+    /// Maximum delay between bootstrap lookups.
+    pub max_lookup_interval: Duration,
+    /// Continuous isolation before a previously connected network searches again.
+    pub reconnect_delay: Duration,
+    /// Deadline for one backend operation.
+    pub request_timeout: Duration,
+}
+
+impl Default for DiscoveryPolicy {
+    fn default() -> Self {
+        Self {
+            publish_interval: Duration::from_secs(300),
+            lookup_interval: Duration::from_secs(5),
+            max_lookup_interval: Duration::from_secs(30),
+            reconnect_delay: Duration::from_secs(60),
+            request_timeout: Duration::from_secs(60),
+        }
+    }
+}
+
 /// Everything needed to start an [`crate::Agent`].
 #[derive(Clone)]
 pub struct AgentConfig {
@@ -223,6 +279,9 @@ pub struct AgentConfig {
     pub bind_addrs: Vec<SocketAddr>,
     /// How much external connectivity machinery the endpoint may use.
     pub transport: TransportPolicy,
+    /// Real QUIC transport settings for constrained-path integration tests.
+    #[cfg(feature = "testing")]
+    pub test_quic_transport: Option<iroh::endpoint::QuicTransportConfig>,
     /// Peers with no direct data path, for tests. See
     /// [`AgentConfig::with_unreachable_data_peers`].
     #[cfg(feature = "testing")]
@@ -230,10 +289,14 @@ pub struct AgentConfig {
     /// Hostname announced to peers. `None` keeps whatever the state store holds,
     /// falling back to the OS hostname and finally to a short endpoint id.
     pub hostname: Option<String>,
-    /// Discovery backend. `None` disables discovery-driven dialling; static
-    /// bootstrap candidates still work.
+    /// Additional discovery backend, composed with Mainline when it is enabled.
     pub discovery: Option<Arc<dyn NetworkDiscovery>>,
-    /// How often each active network re-runs discovery and re-evaluates dials.
+    /// Optional Mainline client, shared by this agent's networks. The CLI enables
+    /// it by default except in local-only mode; library callers opt in explicitly.
+    pub dht: Option<MainlineDiscovery>,
+    /// Lookup, publication and recovery scheduling.
+    pub discovery_policy: DiscoveryPolicy,
+    /// How often each active network maintains known peers and observes address changes.
     pub discovery_interval: Duration,
     /// Bounds applied to network input.
     pub limits: Limits,
@@ -269,11 +332,15 @@ impl AgentConfig {
             bind_addrs: Vec::new(),
             transport: TransportPolicy::default(),
             #[cfg(feature = "testing")]
+            test_quic_transport: None,
+            #[cfg(feature = "testing")]
             unreachable_data_peers: Arc::new(std::sync::Mutex::new(
                 std::collections::HashSet::new(),
             )),
             hostname: None,
             discovery: None,
+            dht: None,
+            discovery_policy: DiscoveryPolicy::default(),
             discovery_interval: Duration::from_secs(5),
             limits: Limits::default(),
             reconnect: ReconnectPolicy::default(),
@@ -327,7 +394,19 @@ impl AgentConfig {
         self
     }
 
-    /// Sets how often discovery runs.
+    /// Adds Mainline discovery alongside the configured candidate backend.
+    pub fn with_dht(mut self, dht: MainlineDiscovery) -> Self {
+        self.dht = Some(dht);
+        self
+    }
+
+    /// Sets lookup, publication and isolation recovery timing.
+    pub fn with_discovery_policy(mut self, policy: DiscoveryPolicy) -> Self {
+        self.discovery_policy = policy;
+        self
+    }
+
+    /// Sets the known-peer maintenance and endpoint-address observation interval.
     pub fn with_discovery_interval(mut self, interval: Duration) -> Self {
         self.discovery_interval = interval;
         self
@@ -385,6 +464,8 @@ impl std::fmt::Debug for AgentConfig {
             .field("transport", &self.transport)
             .field("hostname", &self.hostname)
             .field("discovery", &self.discovery.as_ref().map(|d| d.name()))
+            .field("dht", &self.dht.is_some())
+            .field("discovery_policy", &self.discovery_policy)
             .field("discovery_interval", &self.discovery_interval)
             .field("limits", &self.limits)
             .field("reconnect", &self.reconnect)

crates/tsunagi/src/dataplane/transport/fragments.rs

@@ -0,0 +1,234 @@
+//! Bounded fragmentation of opaque transport payloads, below peer relaying.
+//!
+//! Data ALPN v3: u64 packet ID, u32 total length, u32 byte offset (big endian),
+//! followed by payload. Each QUIC connection owns its IDs and reassembly state.
+//! There is no retransmission: one missing fragment loses one datagram only.
+
+use std::collections::{HashMap, VecDeque};
+
+use bytes::{BufMut, Bytes, BytesMut};
+use tokio::time::Instant;
+
+use crate::config::{
+    DATA_REASSEMBLY_TTL, DATA_RECENT_IDS, MAX_DATA_ASSEMBLIES, MAX_DATA_DATAGRAM,
+    MAX_DATA_FRAGMENTS, MAX_DATA_REASSEMBLY_BYTES,
+};
+
+pub(super) const HEADER: usize = 16;
+
+pub(super) fn encode(id: u64, total: usize, offset: usize, payload: &[u8]) -> Bytes {
+    let mut frame = BytesMut::with_capacity(HEADER + payload.len());
+    frame.put_u64(id);
+    frame.put_u32(total as u32);
+    frame.put_u32(offset as u32);
+    frame.extend_from_slice(payload);
+    frame.freeze()
+}
+
+#[derive(Debug)]
+struct Assembly {
+    started: Instant,
+    bytes: Vec<u8>,
+    ranges: Vec<(usize, usize)>,
+    received: usize,
+}
+
+#[derive(Debug, Default)]
+pub(super) struct Reassembler {
+    packets: HashMap<u64, Assembly>,
+    buffered: usize,
+    recent: VecDeque<u64>,
+}
+
+impl Reassembler {
+    fn remember(&mut self, id: u64) {
+        if self.recent.len() == DATA_RECENT_IDS {
+            self.recent.pop_front();
+        }
+        self.recent.push_back(id);
+    }
+
+    fn remove(&mut self, id: u64) -> Option<Assembly> {
+        let packet = self.packets.remove(&id)?;
+        self.buffered -= packet.bytes.len();
+        self.remember(id);
+        Some(packet)
+    }
+
+    pub(super) fn expire(&mut self, now: Instant) {
+        let expired: Vec<_> = self
+            .packets
+            .iter()
+            .filter(|(_, p)| now.duration_since(p.started) >= DATA_REASSEMBLY_TTL)
+            .map(|(id, _)| *id)
+            .collect();
+        for id in expired {
+            self.remove(id);
+        }
+    }
+
+    pub(super) fn push(&mut self, frame: Bytes, now: Instant) -> Option<Bytes> {
+        self.expire(now);
+        let id = u64::from_be_bytes(frame.get(..8)?.try_into().ok()?);
+        let total = u32::from_be_bytes(frame.get(8..12)?.try_into().ok()?) as usize;
+        let offset = u32::from_be_bytes(frame.get(12..HEADER)?.try_into().ok()?) as usize;
+        let payload = frame.get(HEADER..)?;
+        let end = offset.checked_add(payload.len())?;
+        if total > MAX_DATA_DATAGRAM
+            || end > total
+            || (payload.is_empty() && total != 0)
+            || self.recent.contains(&id)
+        {
+            return None;
+        }
+        if !self.packets.contains_key(&id) {
+            if offset == 0 && end == total {
+                self.remember(id);
+                return Some(frame.slice(HEADER..));
+            }
+            // The length is validated before allocating. Drop the oldest
+            // incomplete packet when the per-link byte or entry budget is full.
+            while self.packets.len() >= MAX_DATA_ASSEMBLIES
+                || self.buffered + total > MAX_DATA_REASSEMBLY_BYTES
+            {
+                let oldest = self
+                    .packets
+                    .iter()
+                    .min_by_key(|(_, p)| p.started)
+                    .map(|(id, _)| *id)?;
+                self.remove(oldest);
+            }
+            self.packets.insert(
+                id,
+                Assembly {
+                    started: now,
+                    bytes: vec![0; total],
+                    ranges: Vec::new(),
+                    received: 0,
+                },
+            );
+            self.buffered += total;
+        }
+        let packet = self.packets.get_mut(&id)?;
+        if packet.bytes.len() != total {
+            self.remove(id);
+            return None;
+        }
+        for &(start, stop) in &packet.ranges {
+            if start == offset && stop == end && packet.bytes[offset..end] == *payload {
+                return None; // a duplicate does not reset the expiry time
+            }
+            if offset < stop && end > start {
+                self.remove(id); // conflicting or overlapping fragments
+                return None;
+            }
+        }
+        if packet.ranges.len() == MAX_DATA_FRAGMENTS {
+            self.remove(id);
+            return None;
+        }
+        packet.bytes[offset..end].copy_from_slice(payload);
+        packet.ranges.push((offset, end));
+        packet.received += payload.len();
+        if packet.received == total {
+            return self.remove(id).map(|p| Bytes::from(p.bytes));
+        }
+        None
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+    use super::*;
+
+    #[test]
+    fn reordered_duplicates_and_changing_fragment_sizes_reassemble_once() {
+        let mut rx = Reassembler::default();
+        let now = Instant::now();
+        let packet: Vec<u8> = (0..9000).map(|i| (i % 251) as u8).collect();
+        let mut parts = Vec::new();
+        let mut offset = 0;
+        for size in [1100, 700, 1100, 1100, 700, 1100, 1100, 1100, 1000] {
+            parts.push(encode(
+                7,
+                packet.len(),
+                offset,
+                &packet[offset..offset + size],
+            ));
+            offset += size;
+        }
+        assert_eq!(offset, packet.len());
+        for part in parts[1..].iter().rev() {
+            assert!(rx.push(part.clone(), now).is_none());
+            assert!(rx.push(part.clone(), now).is_none());
+        }
+        assert_eq!(rx.push(parts[0].clone(), now).unwrap().as_ref(), packet);
+        assert!(rx.push(parts[0].clone(), now).is_none());
+        assert_eq!(rx.buffered, 0);
+        assert!(rx.packets.is_empty());
+    }
+
+    #[test]
+    fn a_missing_fragment_never_blocks_the_next_packet_and_expires() {
+        let mut rx = Reassembler::default();
+        let now = Instant::now();
+        assert!(rx.push(encode(1, 1280, 0, &[1; 800]), now).is_none());
+        assert_eq!(
+            rx.push(encode(2, 4, 0, b"next"), now).unwrap(),
+            &b"next"[..]
+        );
+        rx.expire(now + DATA_REASSEMBLY_TTL);
+        assert!(rx.packets.is_empty());
+        assert_eq!(rx.buffered, 0);
+        assert!(
+            rx.push(encode(1, 1280, 800, &[1; 480]), now + DATA_REASSEMBLY_TTL)
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn malformed_conflicting_and_oversized_fragments_are_bounded() {
+        let now = Instant::now();
+        let mut rx = Reassembler::default();
+        for len in 0..HEADER {
+            assert!(rx.push(Bytes::from(vec![0xff; len]), now).is_none());
+        }
+        assert!(
+            rx.push(encode(1, MAX_DATA_DATAGRAM + 1, 0, b"x"), now)
+                .is_none()
+        );
+        assert!(rx.push(encode(2, 100, 100, b"x"), now).is_none());
+        assert!(rx.push(encode(3, 100, 0, b""), now).is_none());
+        assert_eq!(rx.buffered, 0);
+        for (id, bad) in [
+            (4, encode(4, 100, 0, b"different")),
+            (5, encode(5, 100, 3, b"overlap")),
+            (6, encode(6, 101, 9, b"changed total")),
+        ] {
+            assert!(rx.push(encode(id, 100, 0, b"123456789"), now).is_none());
+            assert!(rx.push(bad, now).is_none());
+            assert_eq!(rx.buffered, 0);
+        }
+        for id in 100..400 {
+            assert!(
+                rx.push(encode(id, MAX_DATA_DATAGRAM, 0, b"x"), now)
+                    .is_none()
+            );
+            assert!(rx.buffered <= MAX_DATA_REASSEMBLY_BYTES);
+            assert!(rx.packets.len() <= MAX_DATA_ASSEMBLIES);
+            assert!(rx.recent.len() <= DATA_RECENT_IDS);
+        }
+    }
+
+    #[test]
+    fn excessive_fragment_counts_are_discarded() {
+        let mut rx = Reassembler::default();
+        let now = Instant::now();
+        for offset in 0..=MAX_DATA_FRAGMENTS {
+            assert!(rx.push(encode(1, 1000, offset, b"x"), now).is_none());
+        }
+        assert!(rx.packets.is_empty());
+        assert_eq!(rx.buffered, 0);
+    }
+}

crates/tsunagi/src/dataplane/transport/iroh_link.rs

@@ -20,13 +20,14 @@
 
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 
 use bytes::Bytes;
 use iroh::EndpointId;
 use iroh::endpoint::{Connection, RecvStream, SendStream};
 
 use crate::BoxFuture;
-use crate::config::Limits;
+use crate::config::{DATA_REASSEMBLY_TTL, Limits, MAX_DATA_DATAGRAM, MAX_DATA_FRAGMENTS};
 use crate::error::ProtocolError;
 use crate::identity::{NetworkId, NetworkKeys};
 use crate::net::EndpointAdapter;
@@ -36,6 +37,7 @@ use crate::proto::message::{
 };
 use crate::proto::{read_frame, write_frame};
 
+use super::fragments::{self, Reassembler};
 use super::{InboundLink, PacketLink, PacketTransport, SharedLink, TransportError};
 
 /// What the iroh transport needs from the agent.
@@ -61,6 +63,8 @@ pub struct IrohLink {
     peer: EndpointId,
     conn: Connection,
     max_datagram: usize,
+    next_packet: AtomicU64,
+    reassembly: tokio::sync::Mutex<Reassembler>,
     // Kept alive so the peer sees the channel as open; the connection closes
     // when the link is dropped.
     _send: tokio::sync::Mutex<SendStream>,
@@ -76,14 +80,16 @@ impl IrohLink {
         send: SendStream,
         recv: RecvStream,
     ) -> Self {
-        let local_limit = conn.max_datagram_size().unwrap_or(0);
-        // Both ends must agree, so the smaller limit wins.
-        let max_datagram = local_limit.min(peer_limit);
+        // The negotiated limit is for reassembled payloads, never a snapshot
+        // of the initial path (which may still be a relay or another VPN).
+        let max_datagram = MAX_DATA_DATAGRAM.min(peer_limit);
         Self {
             network,
             peer,
             conn,
             max_datagram,
+            next_packet: AtomicU64::new(0),
+            reassembly: tokio::sync::Mutex::new(Reassembler::default()),
             _send: tokio::sync::Mutex::new(send),
             _recv: tokio::sync::Mutex::new(recv),
         }
@@ -110,17 +116,65 @@ impl PacketLink for IrohLink {
                 limit: self.max_datagram,
             });
         }
-        self.conn.send_datagram(payload).map_err(|err| {
-            use iroh::endpoint::SendDatagramError;
-            match err {
-                SendDatagramError::ConnectionLost(_) => TransportError::Closed,
-                other => TransportError::Other(other.to_string()),
+        let id = self.next_packet.fetch_add(1, Ordering::Relaxed);
+        let mut offset: usize = 0;
+        let mut attempts = 0;
+        // Re-read the current path's capacity for every fragment. A migration
+        // can shrink it even between two send_datagram calls.
+        for _ in 0..MAX_DATA_FRAGMENTS {
+            loop {
+                let capacity = self
+                    .conn
+                    .max_datagram_size()
+                    .unwrap_or(0)
+                    .checked_sub(fragments::HEADER)
+                    .filter(|n| *n > 0)
+                    .ok_or_else(|| {
+                        TransportError::Other("QUIC path cannot carry data fragments".into())
+                    })?;
+                let end = offset.saturating_add(capacity).min(payload.len());
+                let frame = fragments::encode(id, payload.len(), offset, &payload[offset..end]);
+                match self.conn.send_datagram(frame) {
+                    Ok(()) => {
+                        offset = end;
+                        break;
+                    }
+                    Err(iroh::endpoint::SendDatagramError::TooLarge) if attempts < 3 => {
+                        attempts += 1;
+                    }
+                    Err(iroh::endpoint::SendDatagramError::ConnectionLost(_)) => {
+                        return Err(TransportError::Closed);
+                    }
+                    Err(err) => return Err(TransportError::Other(err.to_string())),
+                }
             }
-        })
+            if offset == payload.len() {
+                return Ok(());
+            }
+        }
+        Err(TransportError::Other(
+            "QUIC path needs too many fragments".into(),
+        ))
     }
 
     fn recv(&self) -> BoxFuture<'_, Option<Bytes>> {
-        Box::pin(async move { self.conn.read_datagram().await.ok() })
+        Box::pin(async move {
+            let mut reassembly = self.reassembly.lock().await;
+            loop {
+                match tokio::time::timeout(DATA_REASSEMBLY_TTL, self.conn.read_datagram()).await {
+                    Ok(Ok(frame)) => {
+                        if let Some(packet) = reassembly.push(frame, tokio::time::Instant::now()) {
+                            return Some(packet);
+                        }
+                    }
+                    Ok(Err(_)) => {
+                        *reassembly = Reassembler::default();
+                        return None;
+                    }
+                    Err(_) => reassembly.expire(tokio::time::Instant::now()),
+                }
+            }
+        })
     }
 
     fn closed(&self) -> BoxFuture<'_, ()> {
@@ -208,10 +262,14 @@ impl IrohTransport {
         }
 
         let serves = self.lookup.serves(outcome.network_id, &open.protocol).await;
-        let max_datagram = conn.max_datagram_size().unwrap_or(0);
+        if open.max_datagram == 0 || conn.max_datagram_size().is_none() {
+            return Err(TransportError::Other(
+                "data channel has no datagram support".into(),
+            ));
+        }
         let ack = DataOpenAck {
             accepted: serves,
-            max_datagram: max_datagram as u32,
+            max_datagram: MAX_DATA_DATAGRAM as u32,
         };
         write_frame(
             &mut send,
@@ -226,7 +284,14 @@ impl IrohTransport {
             return Err(TransportError::Declined(open.protocol));
         }
 
-        let link = IrohLink::new(outcome.network_id, peer, conn, usize::MAX, send, recv);
+        let link = IrohLink::new(
+            outcome.network_id,
+            peer,
+            conn,
+            open.max_datagram as usize,
+            send,
+            recv,
+        );
         Ok(InboundLink {
             network: outcome.network_id,
             peer,
@@ -289,6 +354,7 @@ impl PacketTransport for IrohTransport {
 
             let open = DataOpen {
                 protocol: protocol.to_string(),
+                max_datagram: MAX_DATA_DATAGRAM as u32,
             };
             write_frame(
                 &mut send,
@@ -310,6 +376,12 @@ impl PacketTransport for IrohTransport {
                 return Err(TransportError::Declined(protocol.to_string()));
             }
 
+            if ack.max_datagram == 0 || conn.max_datagram_size().is_none() {
+                return Err(TransportError::Other(
+                    "data channel has no datagram support".into(),
+                ));
+            }
+
             let link = IrohLink::new(network, peer, conn, ack.max_datagram as usize, send, recv);
             Ok(Arc::new(link) as SharedLink)
         })

crates/tsunagi/src/dataplane/transport/mod.rs

@@ -29,6 +29,7 @@
 //! encrypted by the transport, and scoped to exactly one network, one peer and
 //! one plugin protocol.
 
+mod fragments;
 pub mod iroh_link;
 
 use bytes::Bytes;
@@ -77,8 +78,8 @@ pub trait PacketLink: Send + Sync + std::fmt::Debug + 'static {
 
     /// The largest datagram this link can carry, in bytes.
     ///
-    /// A plugin must size its own packets to fit, because there is no
-    /// fragmentation here.
+    /// This is the logical payload limit. A transport can fragment underneath
+    /// it so path MTU changes do not force a plugin to resize the host interface.
     fn max_datagram_size(&self) -> usize;
 
     /// Sends one datagram.

crates/tsunagi/src/discovery.rs

@@ -12,8 +12,7 @@
 //!
 //! * *Finding members of a network* — [`NetworkDiscovery::resolve`], keyed by
 //!   the secret-derived [`DiscoveryKey`]. That is what lives here, and today
-//!   it is [`StaticBootstrap`] plus a test backend; a DHT backend is future
-//!   work.
+//!   it includes [`StaticBootstrap`], [`MainlineDiscovery`] and a test backend.
 //! * *Resolving the address of one iroh endpoint* — **iroh's job, not ours**.
 //!   With [`crate::config::TransportPolicy::N0Defaults`] or `DirectOnly`, iroh
 //!   publishes and resolves endpoint addresses through Number 0's public
@@ -23,14 +22,18 @@
 //! No empty result ever proves a network is empty. It only means "nobody found
 //! yet".
 //!
-//! Mainline DHT discovery is future work and is not implemented here.
+//! Mainline rendezvous is opt-in for library callers; the command line enables it.
+
+mod mainline;
+pub(crate) mod worker;
+pub use mainline::MainlineDiscovery;
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
 
 use iroh::{EndpointAddr, EndpointId};
 
-use crate::error::Result;
+use crate::error::{Error, Result};
 use crate::identity::DiscoveryKey;
 
 pub use crate::BoxFuture;
@@ -103,6 +106,23 @@ pub trait NetworkDiscovery: Send + Sync + std::fmt::Debug + 'static {
 
     /// Returns the candidates currently known for `key`.
     fn resolve<'a>(&'a self, key: DiscoveryKey) -> BoxFuture<'a, Result<Vec<Candidate>>>;
+
+    /// Delivers candidates as they arrive. The default adapts a batch backend;
+    /// remote backends can override this so the first dial need not wait for all lookups.
+    fn resolve_into<'a>(
+        &'a self,
+        key: DiscoveryKey,
+        candidates: tokio::sync::mpsc::Sender<Candidate>,
+    ) -> BoxFuture<'a, Result<()>> {
+        Box::pin(async move {
+            for candidate in self.resolve(key).await? {
+                if candidates.send(candidate).await.is_err() {
+                    break;
+                }
+            }
+            Ok(())
+        })
+    }
 }
 
 /// A statically configured list of bootstrap candidates.
@@ -273,13 +293,25 @@ impl NetworkDiscovery for CompositeDiscovery {
 
     fn publish<'a>(&'a self, key: DiscoveryKey, addr: EndpointAddr) -> BoxFuture<'a, Result<()>> {
         Box::pin(async move {
+            let mut tasks = tokio::task::JoinSet::new();
             for backend in &self.backends {
-                // One failing backend must not stop the others.
-                if let Err(err) = backend.publish(key, addr.clone()).await {
-                    tracing::debug!(backend = backend.name(), %err, "publish failed");
+                let backend = backend.clone();
+                let addr = addr.clone();
+                tasks.spawn(async move { backend.publish(key, addr).await });
+            }
+            let mut failed = false;
+            while let Some(result) = tasks.join_next().await {
+                if !matches!(result, Ok(Ok(()))) {
+                    failed = true;
                 }
             }
-            Ok(())
+            if failed {
+                Err(Error::Discovery(
+                    "a discovery publication failed; will retry".into(),
+                ))
+            } else {
+                Ok(())
+            }
         })
     }
 
@@ -289,11 +321,12 @@ impl NetworkDiscovery for CompositeDiscovery {
         endpoint: EndpointId,
     ) -> BoxFuture<'a, Result<()>> {
         Box::pin(async move {
+            let mut tasks = tokio::task::JoinSet::new();
             for backend in &self.backends {
-                if let Err(err) = backend.unpublish(key, endpoint).await {
-                    tracing::debug!(backend = backend.name(), %err, "unpublish failed");
-                }
+                let backend = backend.clone();
+                tasks.spawn(async move { backend.unpublish(key, endpoint).await });
             }
+            while tasks.join_next().await.is_some() {}
             Ok(())
         })
     }
@@ -312,4 +345,25 @@ impl NetworkDiscovery for CompositeDiscovery {
             Ok(out)
         })
     }
+
+    fn resolve_into<'a>(
+        &'a self,
+        key: DiscoveryKey,
+        candidates: tokio::sync::mpsc::Sender<Candidate>,
+    ) -> BoxFuture<'a, Result<()>> {
+        Box::pin(async move {
+            let mut tasks = tokio::task::JoinSet::new();
+            for backend in &self.backends {
+                let backend = backend.clone();
+                let candidates = candidates.clone();
+                tasks.spawn(async move {
+                    if let Err(err) = backend.resolve_into(key, candidates).await {
+                        tracing::debug!(backend = backend.name(), %err, "resolve failed");
+                    }
+                });
+            }
+            while tasks.join_next().await.is_some() {}
+            Ok(())
+        })
+    }
 }

crates/tsunagi/src/discovery/mainline.rs

@@ -0,0 +1,516 @@
+//! Mainline BEP44 rendezvous. Slots are a changing sample, not a membership list.
+
+use ::mainline::{Dht, MutableItem, SigningKey, async_dht::AsyncDht};
+use futures_lite::StreamExt;
+use iroh::{EndpointAddr, EndpointId};
+use sha2::{Digest, Sha256};
+use std::collections::{HashSet, VecDeque};
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use tokio::sync::{Mutex as AsyncMutex, mpsc};
+use tokio::task::JoinSet;
+use tokio::time::timeout;
+use zeroize::Zeroizing;
+
+use super::{BoxFuture, Candidate, CandidateSource, NetworkDiscovery};
+use crate::config::{
+    DHT_CLOCK_SKEW, DHT_MAX_ADDRS, DHT_MAX_RELAY_LEN, DHT_MAX_VALUE, DHT_RECORD_TTL, DHT_SLOTS,
+};
+use crate::error::{Error, Result};
+use crate::identity::{DiscoveryKey, NetworkKeys};
+
+const QUERY_TIMEOUT: Duration = Duration::from_secs(8);
+const MAGIC: &[u8; 5] = b"TSND\x01";
+
+/// One lazily started Mainline client, shared by all networks on one agent.
+/// No socket is opened until an active network publishes or looks up candidates.
+/// Agent shutdown closes this client and all its clones; a new agent needs a new client.
+#[derive(Debug, Clone, Default)]
+pub struct MainlineDiscovery {
+    client: Arc<AsyncMutex<ClientState>>,
+    allow_loopback: bool,
+}
+
+#[derive(Debug, Default)]
+enum ClientState {
+    #[default]
+    Pending,
+    Ready(AsyncDht),
+    Stopped,
+}
+
+impl MainlineDiscovery {
+    /// Uses a caller-supplied Mainline node instead of public bootstrap defaults.
+    pub fn from_dht(dht: Dht) -> Self {
+        Self {
+            client: Arc::new(AsyncMutex::new(ClientState::Ready(dht.as_async()))),
+            allow_loopback: false,
+        }
+    }
+
+    /// Creates a loopback-only client for a local Mainline Testnet.
+    #[cfg(feature = "testing")]
+    pub fn local_testnet(bootstrap: &[String]) -> Result<Self> {
+        // Reject anything that could resolve or send outside loopback.
+        if bootstrap.is_empty()
+            || bootstrap.iter().any(|s| {
+                s.parse::<SocketAddr>()
+                    .map_or(true, |a| !a.ip().is_loopback())
+            })
+        {
+            return Err(Error::Discovery(
+                "test bootstrap must contain loopback sockets".into(),
+            ));
+        }
+        let dht = Dht::builder()
+            .bootstrap(bootstrap)
+            .bind_address(Ipv4Addr::LOCALHOST)
+            .port(0)
+            .request_timeout(Duration::from_millis(200))
+            .build()
+            .map_err(|_| Error::Discovery("cannot bind local DHT".into()))?;
+        Ok(Self {
+            allow_loopback: true,
+            ..Self::from_dht(dht)
+        })
+    }
+
+    /// Binds this client's rendezvous backend to one network's independent keys.
+    pub fn for_network(&self, keys: &NetworkKeys) -> Arc<dyn NetworkDiscovery> {
+        Arc::new(NetworkDht {
+            client: self.clone(),
+            key: keys.discovery_key(),
+            seed: Zeroizing::new(*keys.dht_write_key()),
+            observed: Mutex::new(VecDeque::new()),
+        })
+    }
+
+    async fn client(&self) -> Result<AsyncDht> {
+        let mut client = self.client.lock().await;
+        match &*client {
+            ClientState::Ready(dht) => return Ok(dht.clone()),
+            ClientState::Stopped => return Err(Error::Discovery("DHT client is stopped".into())),
+            ClientState::Pending => {}
+        }
+        // Construction may resolve bootstrap hostnames. Keep it off Tokio's
+        // executor; a construction failure is retried on the next operation.
+        let dht = tokio::task::spawn_blocking(|| Dht::builder().port(0).build())
+            .await
+            .map_err(|_| Error::Discovery("DHT startup task failed".into()))?
+            .map(Dht::as_async)
+            .map_err(|_| Error::Discovery("cannot start DHT client".into()))?;
+        *client = ClientState::Ready(dht.clone());
+        Ok(dht)
+    }
+
+    pub(crate) async fn shutdown(&self) {
+        *self.client.lock().await = ClientState::Stopped;
+    }
+}
+
+struct NetworkDht {
+    client: MainlineDiscovery,
+    key: DiscoveryKey,
+    seed: Zeroizing<[u8; 32]>,
+    // Records encountered while publishing. Only an explicit lookup consumes
+    // these hints: publication never causes dials on a connected network.
+    observed: Mutex<VecDeque<Vec<u8>>>,
+}
+
+impl std::fmt::Debug for NetworkDht {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("NetworkDht(<redacted>)")
+    }
+}
+
+fn salt(slot: u8) -> Vec<u8> {
+    let mut salt = b"tsunagi-rendezvous-v1".to_vec();
+    salt.push(slot);
+    salt
+}
+
+fn slots_for(id: EndpointId) -> [u8; 2] {
+    let hash = Sha256::digest(id.as_bytes());
+    let first = hash[0] % DHT_SLOTS;
+    [first, (first + 1 + hash[1] % (DHT_SLOTS - 1)) % DHT_SLOTS]
+}
+
+fn now() -> Result<u64> {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .map_err(|_| Error::Discovery("system clock predates Unix epoch".into()))
+}
+
+impl NetworkDht {
+    async fn publish_slot(&self, dht: &AsyncDht, slot: u8, value: &[u8]) -> Result<()> {
+        let salt = salt(slot);
+        let signer = SigningKey::from_bytes(&self.seed);
+        let public = signer.verifying_key().to_bytes();
+        for attempt in 0..3 {
+            let recent = timeout(
+                QUERY_TIMEOUT,
+                dht.get_mutable_most_recent(&public, Some(&salt)),
+            )
+            .await
+            .map_err(|_| Error::Discovery("DHT read timed out".into()))?;
+            if let Some(item) = &recent
+                && item.value().len() <= DHT_MAX_VALUE
+                && let Ok(mut observed) = self.observed.lock()
+            {
+                if observed.len() == DHT_SLOTS as usize {
+                    observed.pop_front();
+                }
+                observed.push_back(item.value().to_vec());
+            }
+            let seq = recent
+                .as_ref()
+                .map_or(0, MutableItem::seq)
+                .checked_add(1)
+                .ok_or_else(|| Error::Discovery("DHT sequence exhausted".into()))?;
+            let item = MutableItem::new(signer.clone(), value, seq, Some(&salt));
+            match timeout(
+                QUERY_TIMEOUT,
+                dht.put_mutable(item, recent.as_ref().map(MutableItem::seq)),
+            )
+            .await
+            {
+                Ok(Ok(outcome)) if outcome.stored_at > 0 => return Ok(()),
+                _ if attempt < 2 => {
+                    // CAS is per storage node, not a global lock. Read again on
+                    // conflicts; another writer's entry is an acceptable sample.
+                    tokio::time::sleep(Duration::from_millis(rand::random_range(30..150))).await;
+                }
+                _ => break,
+            }
+        }
+        Err(Error::Discovery(
+            "DHT publication was not acknowledged".into(),
+        ))
+    }
+}
+
+impl NetworkDiscovery for NetworkDht {
+    fn name(&self) -> &str {
+        "mainline"
+    }
+
+    fn publish<'a>(&'a self, key: DiscoveryKey, addr: EndpointAddr) -> BoxFuture<'a, Result<()>> {
+        Box::pin(async move {
+            if key != self.key {
+                return Err(Error::Discovery("wrong rendezvous network".into()));
+            }
+            let value = encode_record(&addr, now()?, self.client.allow_loopback)?;
+            let dht = self.client.client().await?;
+            let [first, second] = slots_for(addr.id);
+            let (a, b) = tokio::join!(
+                self.publish_slot(&dht, first, &value),
+                self.publish_slot(&dht, second, &value)
+            );
+            a.and(b)
+        })
+    }
+
+    fn unpublish<'a>(&'a self, _: DiscoveryKey, _: EndpointId) -> BoxFuture<'a, Result<()>> {
+        // Stopping publication lets freshness expire. Deleting a shared slot
+        // could erase a different writer, and BEP44 has no delete operation.
+        Box::pin(async { Ok(()) })
+    }
+
+    fn resolve<'a>(&'a self, key: DiscoveryKey) -> BoxFuture<'a, Result<Vec<Candidate>>> {
+        Box::pin(async move {
+            let (tx, mut rx) = mpsc::channel(DHT_SLOTS as usize);
+            let collect = async move {
+                let mut found = Vec::new();
+                while let Some(candidate) = rx.recv().await {
+                    found.push(candidate);
+                }
+                found
+            };
+            let (result, found) = tokio::join!(self.resolve_into(key, tx), collect);
+            result?;
+            Ok(found)
+        })
+    }
+
+    fn resolve_into<'a>(
+        &'a self,
+        key: DiscoveryKey,
+        candidates: mpsc::Sender<Candidate>,
+    ) -> BoxFuture<'a, Result<()>> {
+        Box::pin(async move {
+            if key != self.key {
+                return Err(Error::Discovery("wrong rendezvous network".into()));
+            }
+            let dht = self.client.client().await?;
+            let public = SigningKey::from_bytes(&self.seed)
+                .verifying_key()
+                .to_bytes();
+            let observed = self
+                .observed
+                .lock()
+                .map(|mut values| std::mem::take(&mut *values))
+                .unwrap_or_default();
+            let mut seen = HashSet::new();
+            for value in observed {
+                if let Some(addr) = decode_record(&value, now()?, self.client.allow_loopback)
+                    && seen.insert(addr.id)
+                    && candidates
+                        .send(Candidate::new(addr, CandidateSource::Discovery))
+                        .await
+                        .is_err()
+                {
+                    return Ok(());
+                }
+            }
+            let mut slots: Vec<_> = (0..DHT_SLOTS).collect();
+            for i in (1..slots.len()).rev() {
+                slots.swap(i, rand::random_range(0..=i));
+            }
+            let mut queries = JoinSet::new();
+            loop {
+                if seen.len() >= DHT_SLOTS as usize {
+                    break;
+                }
+                while queries.len() < 4 {
+                    let Some(slot) = slots.pop() else {
+                        break;
+                    };
+                    let dht = dht.clone();
+                    let allow_loopback = self.client.allow_loopback;
+                    queries.spawn(async move {
+                        timeout(QUERY_TIMEOUT, async move {
+                            let salt = salt(slot);
+                            let target = MutableItem::target_from_key(&public, Some(&salt));
+                            let mut items = dht.get_mutable(&public, Some(&salt), None);
+                            while let Some(item) = items.next().await {
+                                // Mainline verifies the signature; also bind the
+                                // returned item to the key and slot we requested.
+                                if item.key() != &public || item.target() != &target {
+                                    continue;
+                                }
+                                if let Some(addr) =
+                                    decode_record(item.value(), now().ok()?, allow_loopback)
+                                {
+                                    return Some(addr);
+                                }
+                            }
+                            None
+                        })
+                        .await
+                        .ok()
+                        .flatten()
+                    });
+                }
+                let Some(result) = queries.join_next().await else {
+                    break;
+                };
+                if let Ok(Some(addr)) = result
+                    && seen.insert(addr.id)
+                    && candidates
+                        .send(Candidate::new(addr, CandidateSource::Discovery))
+                        .await
+                        .is_err()
+                {
+                    break;
+                }
+            }
+            Ok(())
+        })
+    }
+}
+
+fn usable(addr: &SocketAddr, allow_loopback: bool) -> bool {
+    addr.port() != 0
+        && !addr.ip().is_unspecified()
+        && !addr.ip().is_multicast()
+        && (allow_loopback || !addr.ip().is_loopback())
+        && match addr.ip() {
+            IpAddr::V4(ip) => !ip.is_broadcast() && !ip.is_link_local(),
+            IpAddr::V6(ip) => !ip.is_unicast_link_local(),
+        }
+}
+
+fn encode_record(addr: &EndpointAddr, timestamp: u64, allow_loopback: bool) -> Result<Vec<u8>> {
+    let ips: Vec<_> = addr
+        .ip_addrs()
+        .filter(|ip| usable(ip, allow_loopback))
+        .take(DHT_MAX_ADDRS)
+        .collect();
+    let relay = addr
+        .relay_urls()
+        .next()
+        .map(ToString::to_string)
+        .unwrap_or_default();
+    if relay.len() > DHT_MAX_RELAY_LEN || (ips.is_empty() && relay.is_empty()) {
+        return Err(Error::Discovery("no encodable endpoint address yet".into()));
+    }
+    let mut bytes = Vec::with_capacity(DHT_MAX_VALUE);
+    bytes.extend_from_slice(MAGIC);
+    bytes.extend_from_slice(&timestamp.to_be_bytes());
+    bytes.extend_from_slice(addr.id.as_bytes());
+    bytes.extend_from_slice(&(relay.len() as u16).to_be_bytes());
+    bytes.extend_from_slice(relay.as_bytes());
+    bytes.push(ips.len() as u8);
+    for ip in ips {
+        match ip.ip() {
+            IpAddr::V4(v) => {
+                bytes.push(4);
+                bytes.extend_from_slice(&v.octets());
+            }
+            IpAddr::V6(v) => {
+                bytes.push(6);
+                bytes.extend_from_slice(&v.octets());
+            }
+        }
+        bytes.extend_from_slice(&ip.port().to_be_bytes());
+    }
+    if bytes.len() > DHT_MAX_VALUE {
+        return Err(Error::Discovery("DHT record too large".into()));
+    }
+    Ok(bytes)
+}
+
+fn take<'a>(input: &mut &'a [u8], n: usize) -> Option<&'a [u8]> {
+    let (head, rest) = input.split_at_checked(n)?;
+    *input = rest;
+    Some(head)
+}
+
+fn decode_record(mut bytes: &[u8], now: u64, allow_loopback: bool) -> Option<EndpointAddr> {
+    if bytes.len() > DHT_MAX_VALUE || take(&mut bytes, MAGIC.len())? != MAGIC {
+        return None;
+    }
+    let timestamp = u64::from_be_bytes(take(&mut bytes, 8)?.try_into().ok()?);
+    if timestamp > now.saturating_add(DHT_CLOCK_SKEW.as_secs())
+        || now.saturating_sub(timestamp) >= DHT_RECORD_TTL.as_secs()
+    {
+        return None;
+    }
+    let id = EndpointId::from_bytes(take(&mut bytes, 32)?.try_into().ok()?).ok()?;
+    let mut addr = EndpointAddr::new(id);
+    let len = u16::from_be_bytes(take(&mut bytes, 2)?.try_into().ok()?) as usize;
+    if len > DHT_MAX_RELAY_LEN {
+        return None;
+    }
+    let relay = std::str::from_utf8(take(&mut bytes, len)?).ok()?;
+    if !relay.is_empty() {
+        addr = addr.with_relay_url(relay.parse().ok()?);
+    }
+    let count = *take(&mut bytes, 1)?.first()? as usize;
+    if count > DHT_MAX_ADDRS {
+        return None;
+    }
+    for _ in 0..count {
+        let ip = match *take(&mut bytes, 1)?.first()? {
+            4 => IpAddr::V4(Ipv4Addr::from(
+                <[u8; 4]>::try_from(take(&mut bytes, 4)?).ok()?,
+            )),
+            6 => IpAddr::V6(Ipv6Addr::from(
+                <[u8; 16]>::try_from(take(&mut bytes, 16)?).ok()?,
+            )),
+            _ => return None,
+        };
+        let port = u16::from_be_bytes(take(&mut bytes, 2)?.try_into().ok()?);
+        let socket = SocketAddr::new(ip, port);
+        if usable(&socket, allow_loopback) {
+            addr = addr.with_ip_addr(socket);
+        }
+    }
+    (bytes.is_empty() && !addr.addrs.is_empty()).then_some(addr)
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+    use super::*;
+    use crate::testing::network;
+
+    fn address() -> EndpointAddr {
+        EndpointAddr::new(iroh::SecretKey::generate().public())
+            .with_ip_addr("127.0.0.1:12345".parse().unwrap())
+    }
+
+    #[test]
+    fn records_reject_expiration_future_dates_truncation_and_oversize() {
+        let addr = address();
+        let bytes = encode_record(&addr, 1_000, true).unwrap();
+        assert_eq!(decode_record(&bytes, 1_100, true).unwrap(), addr);
+        assert!(decode_record(&bytes, 1_901, true).is_none());
+        assert!(decode_record(&bytes, 100, true).is_none());
+        for n in 0..bytes.len() {
+            assert!(decode_record(&bytes[..n], 1_100, true).is_none());
+        }
+        assert!(decode_record(&vec![0; DHT_MAX_VALUE + 1], 1_100, true).is_none());
+        assert!(encode_record(&addr, 1_000, false).is_err());
+        assert!(decode_record(&bytes, 1_100, false).is_none());
+    }
+
+    #[tokio::test]
+    async fn a_stopped_custom_client_never_falls_back_to_public_bootstrap() {
+        let net = mainline::Testnet::builder(3).build().unwrap();
+        let client = MainlineDiscovery::local_testnet(&net.bootstrap).unwrap();
+        let (name, secret) = network("mainline-shutdown");
+        let keys = NetworkKeys::derive(&name, &secret);
+        let backend = client.for_network(&keys);
+        client.shutdown().await;
+        assert!(backend.resolve(keys.discovery_key()).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn concurrent_writers_share_slots_and_other_secrets_find_nothing() {
+        let net = mainline::Testnet::builder(5).build().unwrap();
+        let client_a = MainlineDiscovery::local_testnet(&net.bootstrap).unwrap();
+        let client_b = MainlineDiscovery::local_testnet(&net.bootstrap).unwrap();
+        let (name, secret) = network("mainline-slots");
+        let keys = NetworkKeys::derive(&name, &secret);
+        let a = client_a.for_network(&keys);
+        let b = client_b.for_network(&keys);
+        let first = address();
+        let second = loop {
+            let next = address();
+            if slots_for(next.id) == slots_for(first.id) {
+                break next;
+            }
+        };
+        let key = keys.discovery_key();
+        let (ra, rb) = tokio::join!(
+            a.publish(key, first.clone()),
+            b.publish(key, second.clone())
+        );
+        assert!(ra.is_ok() || rb.is_ok());
+        let (found_a, found_b) = tokio::join!(a.resolve(key), b.resolve(key));
+        let found_a = found_a.unwrap();
+        let found_b = found_b.unwrap();
+        assert!(
+            found_a.iter().any(|c| c.addr == second) || found_b.iter().any(|c| c.addr == first),
+            "at least one writer must discover the other despite colliding in both slots"
+        );
+        assert!(
+            found_a
+                .iter()
+                .chain(&found_b)
+                .all(|c| c.addr == first || c.addr == second)
+        );
+        let other = NetworkKeys::derive(&name, &crate::identity::NetworkSecret::generate());
+        assert!(
+            client_a
+                .for_network(&other)
+                .resolve(other.discovery_key())
+                .await
+                .unwrap()
+                .is_empty()
+        );
+        // Restarting the publisher must read seq from DHT instead of starting at 1.
+        let restarted = client_a.for_network(&keys);
+        restarted.publish(key, first.clone()).await.unwrap();
+        assert!(
+            b.resolve(key)
+                .await
+                .unwrap()
+                .iter()
+                .any(|c| c.addr == first)
+        );
+    }
+}

crates/tsunagi/src/discovery/worker.rs

@@ -0,0 +1,251 @@
+//! Per-network discovery tasks. Dropping the owner cancels all backend futures.
+
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::{mpsc, watch};
+use tokio::task::JoinSet;
+use tokio::time::{Instant, timeout};
+
+use super::{Candidate, NetworkDiscovery};
+use crate::config::DiscoveryPolicy;
+use crate::identity::DiscoveryKey;
+use crate::net::EndpointAdapter;
+
+pub(crate) struct DiscoveryWorker {
+    connected: watch::Sender<Connectivity>,
+    tasks: JoinSet<()>,
+    backend: Arc<dyn NetworkDiscovery>,
+    key: DiscoveryKey,
+    endpoint: iroh::EndpointId,
+}
+
+#[derive(Clone, Copy)]
+enum Connectivity {
+    Initial,
+    Connected,
+    Isolated(Instant),
+}
+
+impl DiscoveryWorker {
+    pub(crate) fn spawn(
+        backend: Arc<dyn NetworkDiscovery>,
+        key: DiscoveryKey,
+        adapter: EndpointAdapter,
+        policy: DiscoveryPolicy,
+        address_check: Duration,
+        candidates: mpsc::Sender<Candidate>,
+    ) -> Self {
+        let (connected, receiver) = watch::channel(Connectivity::Initial);
+        let mut tasks = JoinSet::new();
+        tasks.spawn(publish_loop(
+            backend.clone(),
+            key,
+            adapter.clone(),
+            policy.clone(),
+            address_check,
+        ));
+        tasks.spawn(lookup_loop(
+            backend.clone(),
+            key,
+            policy,
+            receiver,
+            candidates,
+        ));
+        Self {
+            connected,
+            tasks,
+            backend,
+            key,
+            endpoint: adapter.endpoint_id(),
+        }
+    }
+
+    pub(crate) fn set_connected(&self, connected: bool) {
+        self.connected.send_if_modified(|current| {
+            if matches!(*current, Connectivity::Connected) == connected {
+                return false;
+            }
+            *current = if connected {
+                Connectivity::Connected
+            } else {
+                Connectivity::Isolated(Instant::now())
+            };
+            true
+        });
+    }
+
+    pub(crate) async fn stop(mut self) {
+        self.tasks.abort_all();
+        while self.tasks.join_next().await.is_some() {}
+        // BEP44 has no deletion. Memory/static backends can withdraw promptly;
+        // an unresponsive backend must not extend agent shutdown.
+        let _ = timeout(
+            Duration::from_millis(250),
+            self.backend.unpublish(self.key, self.endpoint),
+        )
+        .await;
+    }
+}
+
+fn jitter(delay: Duration) -> Duration {
+    delay
+        .mul_f64(0.8 + 0.4 * rand::random::<f64>())
+        .max(Duration::from_millis(10))
+}
+
+async fn publish_loop(
+    backend: Arc<dyn NetworkDiscovery>,
+    key: DiscoveryKey,
+    adapter: EndpointAdapter,
+    policy: DiscoveryPolicy,
+    address_check: Duration,
+) {
+    let mut previous = None;
+    let mut due = Instant::now();
+    let mut ticker = tokio::time::interval(address_check.max(Duration::from_millis(10)));
+    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    loop {
+        ticker.tick().await;
+        let mut addr = adapter.addr();
+        addr.addrs.extend(adapter.loopback_addr().addrs);
+        if previous.as_ref() != Some(&addr) {
+            previous = Some(addr.clone());
+            due = Instant::now();
+        }
+        if Instant::now() < due {
+            continue;
+        }
+        let result = timeout(policy.request_timeout, backend.publish(key, addr.clone())).await;
+        match result {
+            Ok(Ok(())) => {
+                due = Instant::now() + jitter(policy.publish_interval);
+            }
+            result => {
+                tracing::debug!(?result, "discovery publication failed; will retry");
+                due = Instant::now() + jitter(policy.lookup_interval);
+            }
+        }
+    }
+}
+
+async fn lookup_loop(
+    backend: Arc<dyn NetworkDiscovery>,
+    key: DiscoveryKey,
+    policy: DiscoveryPolicy,
+    mut connected: watch::Receiver<Connectivity>,
+    candidates: mpsc::Sender<Candidate>,
+) {
+    let mut last_isolation = None;
+    let mut due = Instant::now();
+    let mut delay = policy.lookup_interval;
+    loop {
+        let state = *connected.borrow_and_update();
+        if matches!(state, Connectivity::Connected) {
+            if connected.changed().await.is_err() {
+                break;
+            }
+            continue;
+        }
+        if let Connectivity::Isolated(since) = state
+            && last_isolation != Some(since)
+        {
+            // Preserve the transition time even if a rapid reconnect/disconnect
+            // overwrote a watch value before this task had a chance to run.
+            due = since + policy.reconnect_delay;
+            delay = policy.lookup_interval;
+            last_isolation = Some(since);
+        }
+        tokio::select! {
+            biased;
+            changed = connected.changed() => {
+                if changed.is_err() { break; }
+                continue;
+            }
+            _ = tokio::time::sleep_until(due) => {}
+        }
+        tokio::select! {
+            biased;
+            changed = connected.changed() => {
+                if changed.is_err() { break; }
+                continue;
+            }
+            result = timeout(policy.request_timeout, backend.resolve_into(key, candidates.clone())) => {
+                if !matches!(result, Ok(Ok(()))) {
+                    tracing::debug!(?result, "discovery lookup failed; will retry");
+                }
+            }
+        }
+        due = Instant::now() + jitter(delay);
+        delay = delay.saturating_mul(2).min(policy.max_lookup_interval);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+    use super::*;
+    use crate::Result;
+    use crate::discovery::BoxFuture;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    #[derive(Debug, Default)]
+    struct Reads(AtomicUsize);
+    impl NetworkDiscovery for Reads {
+        fn name(&self) -> &str {
+            "reads"
+        }
+        fn publish<'a>(
+            &'a self,
+            _: DiscoveryKey,
+            _: iroh::EndpointAddr,
+        ) -> BoxFuture<'a, Result<()>> {
+            Box::pin(async { Ok(()) })
+        }
+        fn unpublish<'a>(
+            &'a self,
+            _: DiscoveryKey,
+            _: iroh::EndpointId,
+        ) -> BoxFuture<'a, Result<()>> {
+            Box::pin(async { Ok(()) })
+        }
+        fn resolve<'a>(&'a self, _: DiscoveryKey) -> BoxFuture<'a, Result<Vec<Candidate>>> {
+            self.0.fetch_add(1, Ordering::SeqCst);
+            Box::pin(async { Ok(Vec::new()) })
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn recovery_requires_a_full_minute_and_reconnection_restarts_that_minute() {
+        let reads = Arc::new(Reads::default());
+        let (tx, rx) = watch::channel(Connectivity::Connected);
+        let (candidates, _receiver) = mpsc::channel(16);
+        let mut tasks = JoinSet::new();
+        tasks.spawn(lookup_loop(
+            reads.clone(),
+            DiscoveryKey::from_bytes([0; 32]),
+            DiscoveryPolicy::default(),
+            rx,
+            candidates,
+        ));
+        tokio::task::yield_now().await;
+        tx.send(Connectivity::Isolated(Instant::now())).unwrap();
+        tokio::task::yield_now().await;
+        tokio::time::advance(Duration::from_secs(59)).await;
+        tokio::task::yield_now().await;
+        assert_eq!(reads.0.load(Ordering::SeqCst), 0);
+        tx.send(Connectivity::Connected).unwrap();
+        tx.send(Connectivity::Isolated(Instant::now())).unwrap();
+        tokio::task::yield_now().await;
+        tokio::time::advance(Duration::from_secs(59)).await;
+        tokio::task::yield_now().await;
+        assert_eq!(reads.0.load(Ordering::SeqCst), 0);
+        tokio::time::advance(Duration::from_secs(1)).await;
+        tokio::task::yield_now().await;
+        assert_eq!(reads.0.load(Ordering::SeqCst), 1);
+        tx.send(Connectivity::Connected).unwrap();
+        tokio::task::yield_now().await;
+        tokio::time::advance(Duration::from_secs(600)).await;
+        tokio::task::yield_now().await;
+        assert_eq!(reads.0.load(Ordering::SeqCst), 1);
+    }
+}

crates/tsunagi/src/dns/publish/windows.rs

@@ -201,8 +201,7 @@ impl DnsPublisher for NrptPublisher {
                 )));
             }
 
-            let namespaces: Vec<String> =
-                published.domains.iter().map(|d| namespace(d)).collect();
+            let namespaces: Vec<String> = published.domains.iter().map(|d| namespace(d)).collect();
             let servers: Vec<String> = servers.iter().map(|ip| ip.to_string()).collect();
 
             powershell(apply_script(&namespaces, &servers)).await?;
@@ -257,7 +256,10 @@ mod tests {
         let script = apply_script(&[".lab".into()], &["10.13.37.69".into()]);
         let remove = script.find("Remove-DnsClientNrptRule").unwrap();
         let add = script.find("Add-DnsClientNrptRule").unwrap();
-        assert!(remove < add, "a stale rule must go before the new one:\n{script}");
+        assert!(
+            remove < add,
+            "a stale rule must go before the new one:\n{script}"
+        );
         assert!(script.contains("@('.lab')"));
         assert!(script.contains("@('10.13.37.69')"));
     }
@@ -272,7 +274,10 @@ mod tests {
             classify("The term 'Add-DnsClientNrptRule' is not recognized"),
             PublishError::Unavailable(_)
         ));
-        assert!(matches!(classify("something else"), PublishError::Failed(_)));
+        assert!(matches!(
+            classify("something else"),
+            PublishError::Failed(_)
+        ));
     }
 
     #[tokio::test]

crates/tsunagi/src/identity/network.rs

@@ -340,6 +340,8 @@ pub struct NetworkKeys {
     network_id: NetworkId,
     discovery_key: DiscoveryKey,
     auth_key: Zeroizing<[u8; 32]>,
+    // Separate from the semi-public DiscoveryKey. Existing derivations stay frozen.
+    dht_write_key: Zeroizing<[u8; 32]>,
     name: NetworkName,
 }
 
@@ -373,6 +375,7 @@ impl NetworkKeys {
             network_id: NetworkId(expand("network-id")),
             discovery_key: DiscoveryKey(expand("discovery-key")),
             auth_key: Zeroizing::new(expand("handshake-auth")),
+            dht_write_key: Zeroizing::new(expand("mainline-rendezvous-write-v1")),
             name: name.clone(),
         }
     }
@@ -405,6 +408,11 @@ impl NetworkKeys {
     pub(crate) fn auth_key(&self) -> &[u8; 32] {
         &self.auth_key
     }
+
+    /// Secret signing material for network rendezvous; never exported or logged.
+    pub(crate) fn dht_write_key(&self) -> &[u8; 32] {
+        &self.dht_write_key
+    }
 }
 
 impl std::fmt::Debug for NetworkKeys {

crates/tsunagi/src/ipc/windows.rs

@@ -112,7 +112,11 @@ impl Drop for ControlSocket {
 /// A named pipe server instance serves a single client, so a new instance is
 /// created as soon as one is taken — otherwise a second `tsunagi status` while
 /// the first is mid-flight would find nothing listening.
-async fn serve(name: String, first: tokio::net::windows::named_pipe::NamedPipeServer, source: Arc<dyn ReportSource>) {
+async fn serve(
+    name: String,
+    first: tokio::net::windows::named_pipe::NamedPipeServer,
+    source: Arc<dyn ReportSource>,
+) {
     let mut server = first;
     loop {
         if server.connect().await.is_err() {
@@ -149,7 +153,10 @@ async fn serve(name: String, first: tokio::net::windows::named_pipe::NamedPipeSe
 /// Creates the next pipe instance, or `None` if the name can no longer be
 /// served.
 fn next_instance(name: &str) -> Option<tokio::net::windows::named_pipe::NamedPipeServer> {
-    match ServerOptions::new().reject_remote_clients(true).create(name) {
+    match ServerOptions::new()
+        .reject_remote_clients(true)
+        .create(name)
+    {
         Ok(server) => Some(server),
         Err(err) => {
             tracing::debug!(%err, "cannot create the next control pipe instance");
@@ -272,7 +279,9 @@ mod tests {
             Box::pin(async { StatusReport::default() })
         });
 
-        let first = ControlSocket::bind(&path, Arc::clone(&source)).await.unwrap();
+        let first = ControlSocket::bind(&path, Arc::clone(&source))
+            .await
+            .unwrap();
         let second = ControlSocket::bind(&path, source).await;
         assert!(
             matches!(second, Err(Error::StateLocked { .. })),

crates/tsunagi/src/net.rs

@@ -240,6 +240,11 @@ impl EndpointAdapter {
             TransportPolicy::N0Defaults => builder.preset(presets::N0),
         };
 
+        #[cfg(feature = "testing")]
+        if let Some(transport) = &config.test_quic_transport {
+            builder = builder.transport_config(transport.clone());
+        }
+
         if !config.bind_addrs.is_empty() {
             builder = builder.clear_ip_transports();
             for addr in &config.bind_addrs {

crates/tsunagi/src/overlay/provision/windows.rs

@@ -415,7 +415,9 @@ async fn netsh(args: Vec<String>) -> Result<(), OverlayError> {
             text
         }
     };
-    Err(OverlayError::Unavailable(format!("{display} failed: {message}")))
+    Err(OverlayError::Unavailable(format!(
+        "{display} failed: {message}"
+    )))
 }
 
 /// The absolute path to a program in `System32`.
@@ -436,7 +438,11 @@ mod tests {
     }
 
     fn v6(last: u16) -> Cidr {
-        Cidr::new(IpAddr::V6(Ipv6Addr::new(0xfd00, 0, 0, 0, 0, 0, 0, last)), 64).unwrap()
+        Cidr::new(
+            IpAddr::V6(Ipv6Addr::new(0xfd00, 0, 0, 0, 0, 0, 0, last)),
+            64,
+        )
+        .unwrap()
     }
 
     #[test]

crates/tsunagi/src/proto/message.rs

@@ -29,11 +29,11 @@ pub const ALPN: &[u8] = b"tsunagi/ctrl/1";
 /// saturated or broken data plane cannot disturb control traffic, and the
 /// transport underneath can be replaced without touching the control protocol.
 ///
-/// Version 2 puts a tag on every datagram, so one can say "this is for
-/// somebody else" and be passed on by the peer in the middle. An agent
-/// speaking version 1 simply does not form a data link with one speaking
-/// version 2, which is what the version in an ALPN is for.
-pub const DATA_ALPN: &[u8] = b"tsunagi/data/2";
+/// Version 3 fragments logical datagrams below the peer-relay envelope, so
+/// the overlay's MTU is independent of the current QUIC path MTU. Older data
+/// versions cannot form a data link; control and persistent identities remain
+/// compatible. Both ends, including any intermediate peer, must be upgraded.
+pub const DATA_ALPN: &[u8] = b"tsunagi/data/3";
 
 /// Largest plugin protocol identifier accepted when opening a data channel.
 pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
@@ -86,6 +86,8 @@ pub struct Announcement {
 pub struct DataOpen {
     /// Which IP plugin's packets this channel will carry.
     pub protocol: String,
+    /// Maximum reassembled payload the initiator is willing to receive.
+    pub max_datagram: u32,
 }
 
 /// The responder's answer to [`DataOpen`].

crates/tsunagi/tests/discovery_lifecycle.rs

@@ -0,0 +1,117 @@
+//! Discovery is a cancellable bootstrap job, never the network's event loop.
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+use iroh::{EndpointAddr, EndpointId};
+use std::sync::{
+    Arc,
+    atomic::{AtomicUsize, Ordering},
+};
+use std::time::Duration;
+use tsunagi::config::DiscoveryPolicy;
+use tsunagi::discovery::{BoxFuture, Candidate, NetworkDiscovery, SharedMemoryDiscovery};
+use tsunagi::identity::DiscoveryKey;
+use tsunagi::testing::{local_config, network, wait_for_peers, wait_until};
+use tsunagi::{Agent, Result};
+
+#[derive(Debug, Default)]
+struct Counted {
+    table: SharedMemoryDiscovery,
+    reads: AtomicUsize,
+    writes: AtomicUsize,
+}
+
+impl NetworkDiscovery for Counted {
+    fn name(&self) -> &str {
+        "counted"
+    }
+    fn publish<'a>(&'a self, key: DiscoveryKey, addr: EndpointAddr) -> BoxFuture<'a, Result<()>> {
+        self.writes.fetch_add(1, Ordering::SeqCst);
+        self.table.publish(key, addr)
+    }
+    fn unpublish<'a>(&'a self, key: DiscoveryKey, id: EndpointId) -> BoxFuture<'a, Result<()>> {
+        self.table.unpublish(key, id)
+    }
+    fn resolve<'a>(&'a self, key: DiscoveryKey) -> BoxFuture<'a, Result<Vec<Candidate>>> {
+        self.reads.fetch_add(1, Ordering::SeqCst);
+        self.table.resolve(key)
+    }
+}
+
+fn policy() -> DiscoveryPolicy {
+    DiscoveryPolicy {
+        publish_interval: Duration::from_millis(100),
+        lookup_interval: Duration::from_millis(50),
+        max_lookup_interval: Duration::from_millis(100),
+        reconnect_delay: Duration::from_millis(300),
+        request_timeout: Duration::from_secs(2),
+    }
+}
+
+#[tokio::test]
+async fn connected_peers_only_publish_and_isolated_peers_resume_search() {
+    let discovery = Arc::new(Counted::default());
+    let first_dir = tempfile::tempdir().unwrap();
+    let second_dir = tempfile::tempdir().unwrap();
+    let config = |path: &std::path::Path| {
+        local_config(path)
+            .with_discovery(discovery.clone())
+            .with_discovery_policy(policy())
+    };
+    let first = Agent::spawn(config(first_dir.path())).await.unwrap();
+    let second = Agent::spawn(config(second_dir.path())).await.unwrap();
+    let (name, secret) = network("discovery-lifecycle");
+    let id = first.join_network(&name, &secret).await.unwrap();
+    second.join_network(&name, &secret).await.unwrap();
+    wait_for_peers(&first, id, 1).await;
+    wait_for_peers(&second, id, 1).await;
+    let reads = discovery.reads.load(Ordering::SeqCst);
+    let writes = discovery.writes.load(Ordering::SeqCst);
+    wait_until("publications continue while connected", || async {
+        (discovery.writes.load(Ordering::SeqCst) >= writes + 6).then_some(())
+    })
+    .await;
+    assert_eq!(discovery.reads.load(Ordering::SeqCst), reads);
+    second.shutdown().await;
+    wait_for_peers(&first, id, 0).await;
+    wait_until("isolation resumes discovery", || async {
+        (discovery.reads.load(Ordering::SeqCst) > reads).then_some(())
+    })
+    .await;
+    first.shutdown().await;
+}
+
+#[derive(Debug)]
+struct Stalled;
+impl NetworkDiscovery for Stalled {
+    fn name(&self) -> &str {
+        "stalled"
+    }
+    fn publish<'a>(&'a self, _: DiscoveryKey, _: EndpointAddr) -> BoxFuture<'a, Result<()>> {
+        Box::pin(std::future::pending())
+    }
+    fn unpublish<'a>(&'a self, _: DiscoveryKey, _: EndpointId) -> BoxFuture<'a, Result<()>> {
+        Box::pin(std::future::pending())
+    }
+    fn resolve<'a>(&'a self, _: DiscoveryKey) -> BoxFuture<'a, Result<Vec<Candidate>>> {
+        Box::pin(std::future::pending())
+    }
+}
+
+#[tokio::test]
+async fn stalled_discovery_does_not_block_status_or_shutdown() {
+    let dir = tempfile::tempdir().unwrap();
+    let agent = Agent::spawn(local_config(dir.path()).with_discovery(Arc::new(Stalled)))
+        .await
+        .unwrap();
+    let (name, secret) = network("stalled-discovery");
+    let id = agent.join_network(&name, &secret).await.unwrap();
+    // A recheck schedules work; it must not await the unreachable backend.
+    agent.recheck_network(id).await.unwrap();
+    tokio::time::timeout(Duration::from_secs(2), agent.network_status(id))
+        .await
+        .unwrap()
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(2), agent.shutdown())
+        .await
+        .unwrap();
+}

crates/tsunagi/tests/mainline.rs

@@ -0,0 +1,92 @@
+//! Local Mainline Testnet, real iroh authentication, and durable agent state.
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+use std::time::Duration;
+use tsunagi::testing::{local_config, network, wait_for_peers};
+use tsunagi::{Agent, config::DiscoveryPolicy, discovery::MainlineDiscovery};
+
+#[tokio::test]
+#[ignore = "uses public Mainline DHT and iroh relay services"]
+async fn public_dht_finds_and_authenticates_two_agents() {
+    tsunagi::testing::init_tracing();
+    let a_dir = tempfile::tempdir().unwrap();
+    let b_dir = tempfile::tempdir().unwrap();
+    let config = |path: &std::path::Path| {
+        tsunagi::config::AgentConfig::new(tsunagi::config::StoragePaths::under(path))
+            .with_transport(tsunagi::config::TransportPolicy::N0Defaults)
+            .with_dht(MainlineDiscovery::default())
+    };
+    let a = Agent::spawn(config(a_dir.path())).await.unwrap();
+    let b = Agent::spawn(config(b_dir.path())).await.unwrap();
+    let (name, secret) = network("mainline-public-smoke");
+    let id = a.join_network(&name, &secret).await.unwrap();
+    b.join_network(&name, &secret).await.unwrap();
+    let found = tokio::time::timeout(Duration::from_secs(180), async {
+        loop {
+            if !a.network_status(id).await.unwrap().peers.is_empty()
+                && !b.network_status(id).await.unwrap().peers.is_empty()
+            {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(250)).await;
+        }
+    })
+    .await;
+    a.shutdown().await;
+    b.shutdown().await;
+    assert!(
+        found.is_ok(),
+        "public DHT/relay discovery did not connect within three minutes"
+    );
+}
+#[tokio::test]
+async fn two_agents_restore_after_all_dht_records_and_addresses_are_gone() {
+    let first_dir = tempfile::tempdir().unwrap();
+    let second_dir = tempfile::tempdir().unwrap();
+    let (name, secret) = network("dht-restart");
+    let mut identities = None;
+    for _ in 0..2 {
+        if identities.is_some() {
+            for path in [first_dir.path(), second_dir.path()] {
+                let cache = tsunagi::config::StoragePaths::under(path).cache_dir;
+                if cache.exists() {
+                    std::fs::remove_dir_all(cache).unwrap();
+                }
+            }
+        }
+        // A fresh Testnet means no old rendezvous record survives. The second
+        // pass restores the networks from SQLite without another join command.
+        let net = mainline::Testnet::builder(5).build().unwrap();
+        let config = |path: &std::path::Path| {
+            local_config(path)
+                .with_dht(MainlineDiscovery::local_testnet(&net.bootstrap).unwrap())
+                .with_discovery_policy(DiscoveryPolicy {
+                    lookup_interval: Duration::from_millis(100),
+                    max_lookup_interval: Duration::from_millis(300),
+                    ..Default::default()
+                })
+        };
+        let (a, b) = tokio::join!(
+            Agent::spawn(config(first_dir.path())),
+            Agent::spawn(config(second_dir.path()))
+        );
+        let a = a.unwrap();
+        let b = b.unwrap();
+        let id = tsunagi::identity::NetworkKeys::derive(&name, &secret).network_id();
+        if let Some(old) = identities {
+            assert_eq!((a.endpoint_id(), b.endpoint_id()), old);
+        } else {
+            identities = Some((a.endpoint_id(), b.endpoint_id()));
+            let (ra, rb) = tokio::join!(
+                a.join_network(&name, &secret),
+                b.join_network(&name, &secret)
+            );
+            assert_eq!(ra.unwrap(), id);
+            assert_eq!(rb.unwrap(), id);
+        }
+        wait_for_peers(&a, id, 1).await;
+        wait_for_peers(&b, id, 1).await;
+        a.shutdown().await;
+        b.shutdown().await;
+    }
+}