feat(peer)!: cut over to authenticated catalog sharing

Replace address-only trust and pushed peer state with installation identities,
SPKI-pinned QUIC, candidate-only discovery, and bounded responder-owned
protocol-8 pulls. The runtime now owns each network generation and all admitted
work through shutdown.

Add exact bundled content identities, reproducible manifest publishing,
capability-confined downloads, streaming BLAKE3 verification, quarantine and
retry, and crash-recoverable download and install transactions. Ship generated
fixture catalogs and fail closed when production manifests are absent.

The Tauri backend exposes durable sharing policy, redacted identity state, and
attempt-keyed transfer snapshots. Frontend consumption follows in the next
commit. Repository-wide test certificates and protocol-7 paths are removed.

BREAKING CHANGE: peers must use protocol 8 and exact catalog content artifacts;
protocol-7 frames and shared-certificate identities are no longer accepted.

Test Plan:
- `just test` -- passed on the completed stack (708 workspace tests)
- `just clippy` -- passed on the completed stack
- `just build` -- passed with fixture catalogs on the completed stack
- `just catalog-check-production` -- failed closed because the external
  production manifest corpus is absent
- `git diff --cached --check` -- passed

crates/lanspread-peer/src/services/discovery.rs

@@ -1,31 +1,175 @@
 //! mDNS peer discovery and discovery-time protocol negotiation.
 
-use std::time::Duration;
+use std::{
+    collections::{HashSet, VecDeque},
+    future::Future,
+    thread::JoinHandle,
+    time::Duration,
+};
 
+use eyre::WrapErr as _;
+use futures::{StreamExt as _, stream::FuturesUnordered};
 use lanspread_mdns::{LANSPREAD_SERVICE_TYPE, MdnsBrowser, MdnsService, MdnsServicePoll};
-use lanspread_proto::PROTOCOL_VERSION;
-use tokio::sync::mpsc::UnboundedSender;
+use lanspread_proto::{PROTOCOL_VERSION, PeerEndpoint, PeerId};
+use tokio::sync::{
+    mpsc::{self, UnboundedSender},
+    oneshot,
+};
+use tokio_util::sync::CancellationToken;
 
 use crate::{
     PeerEvent,
-    context::Ctx,
+    context::NetworkServiceCtx,
     events,
-    peer_db::PeerId,
-    services::handshake::{HandshakeCtx, perform_handshake_with_peer},
+    services::{
+        handshake::{HandshakeCtx, ReservedCandidateHandshake},
+        state_sync::run_state_sync,
+    },
 };
 
+const MAX_ACTIVE_DISCOVERY_CANDIDATES: usize = 64;
+const MAX_PENDING_MDNS_SERVICES: usize = 64;
+const DISCOVERY_CANDIDATE_COOLDOWN: Duration = Duration::from_secs(5);
+
+#[derive(Default)]
+struct RecentCandidates {
+    entries: VecDeque<(PeerEndpoint, tokio::time::Instant)>,
+}
+
+impl RecentCandidates {
+    fn try_record(&mut self, candidate: PeerEndpoint, now: tokio::time::Instant) -> bool {
+        self.expire(now);
+        if self.entries.iter().any(|(endpoint, _)| {
+            endpoint.peer_id == candidate.peer_id || endpoint.addr == candidate.addr
+        }) || self.entries.len() >= MAX_ACTIVE_DISCOVERY_CANDIDATES
+        {
+            return false;
+        }
+        self.entries
+            .push_back((candidate, now + DISCOVERY_CANDIDATE_COOLDOWN));
+        true
+    }
+
+    fn expire(&mut self, now: tokio::time::Instant) {
+        while self
+            .entries
+            .front()
+            .is_some_and(|(_, deadline)| *deadline <= now)
+        {
+            self.entries.pop_front();
+        }
+    }
+
+    #[cfg(test)]
+    fn len(&self) -> usize {
+        self.entries.len()
+    }
+}
+
 struct MdnsPeerInfo {
     addr: std::net::SocketAddr,
     peer_id: Option<PeerId>,
     proto_ver: Option<u32>,
-    library_rev: u64,
-    library_digest: u64,
+}
+
+struct ProtocolNegotiation {
+    endpoint: PeerEndpoint,
+    handshake: ReservedCandidateHandshake,
+}
+
+struct DiscoveryWorker {
+    shutdown: CancellationToken,
+    result_rx: Option<oneshot::Receiver<eyre::Result<()>>>,
+    thread: Option<JoinHandle<()>>,
+}
+
+impl DiscoveryWorker {
+    fn spawn(
+        service_type: String,
+        service_tx: mpsc::Sender<MdnsService>,
+        shutdown: CancellationToken,
+    ) -> eyre::Result<Self> {
+        Self::spawn_with(shutdown, move |shutdown| {
+            run_mdns_browser(&service_type, &service_tx, &shutdown)
+        })
+    }
+
+    fn spawn_with(
+        shutdown: CancellationToken,
+        worker: impl FnOnce(CancellationToken) -> eyre::Result<()> + Send + 'static,
+    ) -> eyre::Result<Self> {
+        let (result_tx, result_rx) = oneshot::channel();
+        let worker_shutdown = shutdown.clone();
+        let thread = std::thread::Builder::new()
+            .name("lanspread-mdns-browser".to_owned())
+            .spawn(move || {
+                let result = worker(worker_shutdown);
+                let _ = result_tx.send(result);
+            })
+            .wrap_err("failed to spawn mDNS discovery worker")?;
+
+        Ok(Self {
+            shutdown,
+            result_rx: Some(result_rx),
+            thread: Some(thread),
+        })
+    }
+
+    async fn wait_result(&mut self) -> eyre::Result<()> {
+        let result_rx = self
+            .result_rx
+            .as_mut()
+            .ok_or_else(|| eyre::eyre!("mDNS discovery result was already consumed"))?;
+        result_rx
+            .await
+            .map_err(|_| eyre::eyre!("mDNS discovery worker stopped without a result"))?
+    }
+
+    async fn shutdown_and_join(
+        mut self,
+        observed_result: Option<eyre::Result<()>>,
+    ) -> eyre::Result<()> {
+        self.shutdown.cancel();
+        let result = if let Some(result) = observed_result {
+            self.result_rx.take();
+            result
+        } else {
+            let result_rx = self
+                .result_rx
+                .take()
+                .ok_or_else(|| eyre::eyre!("mDNS discovery result was already consumed"))?;
+            result_rx
+                .await
+                .map_err(|_| eyre::eyre!("mDNS discovery worker stopped without a result"))?
+        };
+        self.join_thread()?;
+        result
+    }
+
+    fn join_thread(&mut self) -> eyre::Result<()> {
+        let Some(thread) = self.thread.take() else {
+            return Ok(());
+        };
+        thread
+            .join()
+            .map_err(|_| eyre::eyre!("mDNS discovery worker panicked"))
+    }
+}
+
+impl Drop for DiscoveryWorker {
+    fn drop(&mut self) {
+        self.shutdown.cancel();
+        if let Err(err) = self.join_thread() {
+            log::error!("Failed to join mDNS discovery worker during cleanup: {err}");
+        }
+    }
 }
 
 /// Runs the peer discovery service using mDNS.
+#[allow(clippy::too_many_lines)]
 pub async fn run_peer_discovery(
     tx_notify_ui: UnboundedSender<PeerEvent>,
-    ctx: Ctx,
+    ctx: NetworkServiceCtx,
 ) -> eyre::Result<()> {
     log::info!("Starting peer discovery task");
 
@@ -34,38 +178,36 @@ pub async fn run_peer_discovery(
     }
 
     let service_type = LANSPREAD_SERVICE_TYPE.to_string();
-    let (service_tx, mut service_rx) = tokio::sync::mpsc::unbounded_channel();
-    let worker_shutdown = ctx.shutdown.clone();
-    let service_type_clone = service_type.clone();
+    let (service_tx, mut service_rx) = tokio::sync::mpsc::channel(MAX_PENDING_MDNS_SERVICES);
+    let service_shutdown = ctx.shutdown.child_token();
+    let mut worker = DiscoveryWorker::spawn(service_type, service_tx, service_shutdown.clone())?;
+    let mut negotiations = FuturesUnordered::new();
+    let mut active_candidates = HashSet::new();
+    let mut recent_candidates = RecentCandidates::default();
+    let mut mismatch_emitted = false;
+    let mut state_sync = Box::pin(run_state_sync(
+        ctx.clone(),
+        tx_notify_ui.clone(),
+        service_shutdown.clone(),
+    ));
+    let mut observed_state_sync_result = None;
 
-    let worker_handle = ctx
-        .task_tracker
-        .spawn_blocking(move || -> eyre::Result<()> {
-            let browser = MdnsBrowser::new(&service_type_clone)?;
-            while !worker_shutdown.is_cancelled() {
-                match browser.next_service_timeout(None, Duration::from_millis(250))? {
-                    MdnsServicePoll::Service(service) => {
-                        if service_tx.send(service).is_err() {
-                            log::debug!("Peer discovery consumer dropped; stopping worker");
-                            break;
-                        }
-                    }
-                    MdnsServicePoll::Timeout => {}
-                    MdnsServicePoll::Closed => {
-                        log::warn!("mDNS browser closed; stopping peer discovery worker");
-                        break;
-                    }
+    let observed_worker_result = loop {
+        tokio::select! {
+            () = ctx.shutdown.cancelled() => break None,
+            result = worker.wait_result() => break Some(result),
+            result = &mut state_sync => {
+                observed_state_sync_result = Some(result);
+                break None;
+            }
+            completed = negotiations.next(), if !negotiations.is_empty() => {
+                if let Some(endpoint) = completed {
+                    active_candidates.remove(&endpoint);
                 }
             }
-            Ok(())
-        });
-
-    loop {
-        tokio::select! {
-            () = ctx.shutdown.cancelled() => break,
             service = service_rx.recv() => {
                 let Some(service) = service else {
-                    break;
+                    break None;
                 };
 
                 let info = parse_mdns_peer(&service);
@@ -74,30 +216,145 @@ pub async fn run_peer_discovery(
                     continue;
                 }
 
-                handle_discovered_peer(info, &ctx, &tx_notify_ui).await;
+                if info.proto_ver != Some(PROTOCOL_VERSION) {
+                    if !mismatch_emitted {
+                        events::send(
+                            &tx_notify_ui,
+                            PeerEvent::IncompatibleProtocolDetected {
+                                observed: info.proto_ver,
+                                expected: PROTOCOL_VERSION,
+                            },
+                        );
+                        mismatch_emitted = true;
+                    }
+                    continue;
+                }
+
+                if let Some(endpoint) = validated_candidate_endpoint(&info) {
+                    if !candidate_is_admissible(
+                        &active_candidates,
+                        &mut recent_candidates,
+                        endpoint,
+                        tokio::time::Instant::now(),
+                    ) {
+                        log::warn!(
+                            "Discovery candidate is cooling down or the recent-attempt limit is full; ignoring {}",
+                            endpoint.addr
+                        );
+                        continue;
+                    }
+                    let handshake_ctx = HandshakeCtx::from_network(&ctx, &tx_notify_ui)
+                        .with_cancellation(service_shutdown.clone());
+                    let handshake = match ReservedCandidateHandshake::reserve(handshake_ctx, endpoint).await {
+                        Ok(handshake) => handshake,
+                        Err(error) => {
+                            log::warn!("Failed to reserve discovery candidate {}: {error}", endpoint.addr);
+                            continue;
+                        }
+                    };
+                    active_candidates.insert(endpoint);
+                    negotiations.push(run_protocol_negotiation(ProtocolNegotiation {
+                        endpoint,
+                        handshake,
+                    }));
+                }
             }
         }
+    };
+
+    service_shutdown.cancel();
+    drain_service_children(negotiations).await;
+    let state_sync_exited_early = observed_state_sync_result.is_some();
+    let state_sync_result = match observed_state_sync_result {
+        Some(result) => result,
+        None => state_sync.await,
+    };
+    let worker_result = worker.shutdown_and_join(observed_worker_result).await;
+
+    if let Err(error) = state_sync_result {
+        return Err(error.wrap_err("peer state-sync service failed"));
+    }
+    if state_sync_exited_early && !ctx.shutdown.is_cancelled() {
+        eyre::bail!("peer state-sync service exited unexpectedly");
     }
 
-    match worker_handle.await {
-        Ok(Ok(())) if ctx.shutdown.is_cancelled() => Ok(()),
-        Ok(Ok(())) => {
+    match worker_result {
+        Ok(()) if ctx.shutdown.is_cancelled() => Ok(()),
+        Ok(()) => {
             eyre::bail!("mDNS discovery worker exited unexpectedly");
         }
-        Ok(Err(err)) if ctx.shutdown.is_cancelled() => {
+        Err(err) if ctx.shutdown.is_cancelled() => {
             log::debug!("Peer discovery worker stopped during shutdown: {err}");
             Ok(())
         }
-        Ok(Err(err)) => Err(err.wrap_err("peer discovery worker failed")),
-        Err(err) if ctx.shutdown.is_cancelled() => {
-            log::debug!("Peer discovery worker join ended during shutdown: {err}");
-            Ok(())
-        }
-        Err(err) => Err(eyre::eyre!("peer discovery worker join error: {err}")),
+        Err(err) => Err(err.wrap_err("peer discovery worker failed")),
     }
 }
 
-async fn wait_for_local_peer_addr(ctx: &Ctx) -> bool {
+fn candidate_conflicts(active: &HashSet<PeerEndpoint>, candidate: PeerEndpoint) -> bool {
+    active
+        .iter()
+        .any(|endpoint| endpoint.peer_id == candidate.peer_id || endpoint.addr == candidate.addr)
+}
+
+fn candidate_is_admissible(
+    active: &HashSet<PeerEndpoint>,
+    recent: &mut RecentCandidates,
+    candidate: PeerEndpoint,
+    now: tokio::time::Instant,
+) -> bool {
+    active.len() < MAX_ACTIVE_DISCOVERY_CANDIDATES
+        && !candidate_conflicts(active, candidate)
+        && recent.try_record(candidate, now)
+}
+
+fn run_mdns_browser(
+    service_type: &str,
+    service_tx: &mpsc::Sender<MdnsService>,
+    shutdown: &CancellationToken,
+) -> eyre::Result<()> {
+    let browser = MdnsBrowser::new(service_type)?;
+    let browse_result = (|| {
+        while !shutdown.is_cancelled() {
+            match browser.next_service_timeout(None, Duration::from_millis(250))? {
+                MdnsServicePoll::Service(service) => {
+                    match service_tx.try_send(service) {
+                        Ok(()) => {}
+                        Err(mpsc::error::TrySendError::Full(_)) => {
+                            // Repeated mDNS observations are hints only. Coalesce an
+                            // overflow by dropping it rather than letting the native
+                            // browser thread allocate without bound or block shutdown.
+                            log::trace!(
+                                "Coalescing mDNS observation while the bounded discovery queue is full"
+                            );
+                        }
+                        Err(mpsc::error::TrySendError::Closed(_)) => {
+                            log::debug!("Peer discovery consumer dropped; stopping worker");
+                            break;
+                        }
+                    }
+                }
+                MdnsServicePoll::Timeout => {}
+                MdnsServicePoll::Closed => {
+                    log::warn!("mDNS browser closed; stopping peer discovery worker");
+                    break;
+                }
+            }
+        }
+        Ok(())
+    })();
+    let close_result = browser.close();
+
+    match (browse_result, close_result) {
+        (Ok(()), Ok(())) => Ok(()),
+        (Err(err), Ok(())) | (Ok(()), Err(err)) => Err(err),
+        (Err(browse_err), Err(close_err)) => Err(eyre::eyre!(
+            "mDNS browse failed: {browse_err:#}; browser shutdown also failed: {close_err:#}"
+        )),
+    }
+}
+
+async fn wait_for_local_peer_addr(ctx: &NetworkServiceCtx) -> bool {
     loop {
         if ctx.local_peer_addr.read().await.is_some() {
             return true;
@@ -113,88 +370,244 @@ async fn wait_for_local_peer_addr(ctx: &Ctx) -> bool {
 fn parse_mdns_peer(service: &MdnsService) -> MdnsPeerInfo {
     MdnsPeerInfo {
         addr: service.addr,
-        peer_id: service.properties.get("peer_id").cloned(),
+        peer_id: service
+            .properties
+            .get("peer_id")
+            .and_then(|value| value.parse::<PeerId>().ok()),
         proto_ver: service
             .properties
             .get("proto_ver")
             .and_then(|value| value.parse::<u32>().ok()),
-        library_rev: service
-            .properties
-            .get("library_rev")
-            .and_then(|value| value.parse::<u64>().ok())
-            .unwrap_or(0),
-        library_digest: service
-            .properties
-            .get("library_digest")
-            .and_then(|value| value.parse::<u64>().ok())
-            .unwrap_or(0),
     }
 }
 
-async fn is_self_advertisement(info: &MdnsPeerInfo, ctx: &Ctx) -> bool {
+async fn is_self_advertisement(info: &MdnsPeerInfo, ctx: &NetworkServiceCtx) -> bool {
     let guard = ctx.local_peer_addr.read().await;
     guard.as_ref().is_some_and(|addr| *addr == info.addr)
         || info
             .peer_id
             .as_ref()
-            .is_some_and(|peer_id| peer_id == ctx.peer_id.as_ref())
+            .is_some_and(|peer_id| *peer_id == ctx.peer_id)
 }
 
-async fn handle_discovered_peer(
-    info: MdnsPeerInfo,
-    ctx: &Ctx,
-    tx_notify_ui: &UnboundedSender<PeerEvent>,
-) {
+fn validated_candidate_endpoint(info: &MdnsPeerInfo) -> Option<PeerEndpoint> {
     if info.proto_ver != Some(PROTOCOL_VERSION) {
         log::debug!(
             "Ignoring peer at {} with protocol {:?}; expected {PROTOCOL_VERSION}",
             info.addr,
             info.proto_ver
         );
-        return;
+        return None;
     }
 
-    let Some(peer_id) = info.peer_id.clone() else {
+    let Some(peer_id) = info.peer_id else {
         log::debug!(
             "Ignoring current-protocol peer at {} without a peer_id TXT record",
             info.addr
         );
-        return;
+        return None;
     };
 
-    let upsert = {
-        let mut db = ctx.peer_game_db.write().await;
-        let upsert = db.upsert_peer(peer_id.clone(), info.addr);
-        let features = db.peer_features(&peer_id);
-        if info.library_rev > 0 || info.library_digest > 0 {
-            db.update_peer_library(&peer_id, info.library_rev, info.library_digest, features);
-        }
-        upsert
+    Some(PeerEndpoint::new(peer_id, info.addr))
+}
+
+async fn run_protocol_negotiation(negotiation: ProtocolNegotiation) -> PeerEndpoint {
+    let endpoint = negotiation.endpoint;
+    let result = negotiation.handshake.run().await;
+    if let Err(err) = result {
+        log::warn!(
+            "Failed to negotiate protocol with peer {}: {err}",
+            endpoint.addr
+        );
+    }
+    endpoint
+}
+
+async fn drain_service_children<F>(mut children: FuturesUnordered<F>)
+where
+    F: Future,
+{
+    while children.next().await.is_some() {}
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{
+        collections::HashSet,
+        net::SocketAddr,
+        sync::{
+            Arc,
+            atomic::{AtomicBool, AtomicUsize, Ordering},
+            mpsc,
+        },
+        time::Duration,
     };
 
-    if upsert.is_new {
-        log::info!("Discovered peer at: {}", info.addr);
-        events::emit_peer_discovered(&ctx.peer_game_db, tx_notify_ui, info.addr).await;
+    use futures::stream::FuturesUnordered;
+    use lanspread_proto::{PeerEndpoint, PeerId};
+    use tokio_util::sync::CancellationToken;
+
+    use super::{
+        DISCOVERY_CANDIDATE_COOLDOWN,
+        DiscoveryWorker,
+        MAX_ACTIVE_DISCOVERY_CANDIDATES,
+        RecentCandidates,
+        candidate_conflicts,
+        candidate_is_admissible,
+        drain_service_children,
+    };
+
+    fn endpoint(seed: u8, port: u16) -> PeerEndpoint {
+        PeerEndpoint::new(
+            PeerId::from_bytes([seed; 32]),
+            SocketAddr::from(([127, 0, 0, 1], port)),
+        )
     }
 
-    if upsert.is_new || upsert.addr_changed {
-        spawn_protocol_negotiation(&info, ctx, tx_notify_ui, peer_id);
+    #[test]
+    fn active_candidate_keys_bound_both_claimed_identity_and_address() {
+        let active_endpoint = endpoint(1, 12001);
+        let mut active = HashSet::from([active_endpoint]);
+
+        assert!(candidate_conflicts(&active, endpoint(1, 12002)));
+        assert!(candidate_conflicts(&active, endpoint(2, 12001)));
+        assert!(!candidate_conflicts(&active, endpoint(2, 12002)));
+
+        active.remove(&active_endpoint);
+        assert!(!candidate_conflicts(&active, endpoint(1, 12002)));
+        assert!(!candidate_conflicts(&active, endpoint(2, 12001)));
     }
-}
 
-fn spawn_protocol_negotiation(
-    info: &MdnsPeerInfo,
-    ctx: &Ctx,
-    tx_notify_ui: &UnboundedSender<PeerEvent>,
-    peer_id: PeerId,
-) {
-    let peer_addr = info.addr;
-    let handshake_ctx = HandshakeCtx::from_ctx(ctx, tx_notify_ui);
+    #[test]
+    fn completed_candidate_attempts_are_bounded_and_rate_limited_by_identity_and_address() {
+        let now = tokio::time::Instant::now();
+        let first = endpoint(1, 12001);
+        let mut recent = RecentCandidates::default();
+        assert!(recent.try_record(first, now));
 
-    ctx.task_tracker.spawn(async move {
-        if let Err(err) = perform_handshake_with_peer(handshake_ctx, peer_addr, Some(peer_id)).await
-        {
-            log::warn!("Failed to negotiate protocol with peer {peer_addr}: {err}");
+        for port in 12002..12102 {
+            assert!(!recent.try_record(endpoint(1, port), now));
         }
-    });
+        for seed in 2..=101 {
+            assert!(!recent.try_record(endpoint(seed, 12001), now));
+        }
+        for seed in 2..=u8::try_from(MAX_ACTIVE_DISCOVERY_CANDIDATES).expect("test bound fits u8") {
+            assert!(recent.try_record(endpoint(seed, 13000 + u16::from(seed)), now));
+        }
+        assert_eq!(recent.len(), MAX_ACTIVE_DISCOVERY_CANDIDATES);
+        assert!(!recent.try_record(endpoint(200, 14000), now));
+
+        let after_cooldown = now + DISCOVERY_CANDIDATE_COOLDOWN;
+        assert!(recent.try_record(endpoint(200, 14000), after_cooldown));
+        assert_eq!(recent.len(), 1);
+    }
+
+    #[test]
+    fn expiring_recent_entries_never_bypasses_the_independent_active_cap() {
+        let now = tokio::time::Instant::now();
+        let mut active = (0..MAX_ACTIVE_DISCOVERY_CANDIDATES)
+            .map(|index| {
+                endpoint(
+                    u8::try_from(index + 1).expect("test index fits u8"),
+                    15000 + u16::try_from(index).expect("test index fits u16"),
+                )
+            })
+            .collect::<HashSet<_>>();
+        let mut recent = RecentCandidates::default();
+        for candidate in &active {
+            assert!(recent.try_record(*candidate, now));
+        }
+
+        let after_cooldown = now + DISCOVERY_CANDIDATE_COOLDOWN;
+        let next = endpoint(100, 16000);
+        assert!(!candidate_is_admissible(
+            &active,
+            &mut recent,
+            next,
+            after_cooldown,
+        ));
+        assert_eq!(active.len(), MAX_ACTIVE_DISCOVERY_CANDIDATES);
+
+        let completed = *active.iter().next().expect("active set should be nonempty");
+        active.remove(&completed);
+        assert!(candidate_is_admissible(
+            &active,
+            &mut recent,
+            next,
+            after_cooldown,
+        ));
+    }
+
+    async fn cancellation_aware_child(
+        started: tokio::sync::mpsc::UnboundedSender<()>,
+        shutdown: CancellationToken,
+        completed: Arc<AtomicUsize>,
+    ) {
+        started.send(()).expect("start receiver should remain open");
+        shutdown.cancelled().await;
+        tokio::task::yield_now().await;
+        completed.fetch_add(1, Ordering::SeqCst);
+    }
+
+    #[tokio::test]
+    async fn negotiation_batch_drains_started_children_on_shutdown() {
+        let shutdown = CancellationToken::new();
+        let (started_tx, mut started_rx) = tokio::sync::mpsc::unbounded_channel();
+        let completed = Arc::new(AtomicUsize::new(0));
+        let children = FuturesUnordered::new();
+        for _ in 0..2 {
+            children.push(cancellation_aware_child(
+                started_tx.clone(),
+                shutdown.clone(),
+                completed.clone(),
+            ));
+        }
+
+        let control_shutdown = shutdown.clone();
+        let control = async move {
+            for _ in 0..2 {
+                started_rx
+                    .recv()
+                    .await
+                    .expect("every negotiation should start");
+            }
+            control_shutdown.cancel();
+        };
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            tokio::join!(drain_service_children(children), control);
+        })
+        .await
+        .expect("shutdown should drain every negotiation");
+        assert_eq!(completed.load(Ordering::SeqCst), 2);
+    }
+
+    #[test]
+    fn dropping_discovery_worker_cancels_and_joins_its_thread() {
+        let shutdown = CancellationToken::new();
+        let (started_tx, started_rx) = mpsc::sync_channel(0);
+        let stopped = Arc::new(AtomicBool::new(false));
+        let worker_stopped = stopped.clone();
+        let worker = DiscoveryWorker::spawn_with(shutdown, move |shutdown| {
+            started_tx
+                .send(())
+                .expect("test should wait for worker startup");
+            while !shutdown.is_cancelled() {
+                std::thread::sleep(Duration::from_millis(1));
+            }
+            worker_stopped.store(true, Ordering::SeqCst);
+            Ok(())
+        })
+        .expect("discovery worker should spawn");
+
+        started_rx
+            .recv_timeout(Duration::from_secs(1))
+            .expect("discovery worker should start");
+        drop(worker);
+
+        assert!(
+            stopped.load(Ordering::SeqCst),
+            "worker Drop must not return before its thread stops"
+        );
+    }
 }