nordisk-pairing 0.5.25

//! **How the installer reaches a VM during bootstrap** (0.3.0, owner
//! 2026-09-24). The protocol is always the same — the installer's mutual TLS
//! ([`crate::bootstrap`]) — and only the PATH differs, chosen by the seed's
//! tunnel kind ([`reach`], which every [`crate::tunnel::Tunnel`] driver
//! answers through [`crate::tunnel::Tunnel::installer_reach`]):
//!
//! | tunnel | path |
//! |---|---|
//! | `wireguard` | the installer is a TEMPORARY PEER of the tunnel: every VM raises a WireGuard link from its first boot ([`vm_link`]) with the installer as its one peer, and binds the bootstrap listener on the link address ONLY — nothing of it on the public interface. The installer (behind NAT) initiates ([`WgPath`]) with a keepalive. |
//! | `mullvad` | the same: the estate mesh stays WireGuard, Mullvad is only each VM's exit. |
//! | `tailscale` | the installer would join with a one-time key and leave at the end — refused by name like the driver itself (`[tunnel-tailscale-unimplemented]`). |
//! | `none` | **REFUSED** `[installer-link-none]` (owner 2026-09-25: WireGuard + mutual TLS ALWAYS). [`DirectPath`] is a loopback harness for tests only. |
//!
//! When the installation is complete the installer tells each VM to close:
//! the VM answers and the install is gone ([`crate::installer::Client::install_gone`]).
//! Since 0.5.4 the link and its installer peer are KEPT for the VM's life —
//! the same listener is the log door ([`crate::bootstrap::serve_for_life`]).
//! A kept peer that has left goes QUIET on the VM: the installer only ever
//! dials the VM (no endpoint, no keepalive on this end), so the dataplane
//! (edda wg-appliance-core, `dataplane::GONE_AFTER`, a0e100a) drops its
//! session once the VM has sent to it and heard nothing for 14 s — before
//! boringtun would begin handshakes nobody answers (`HANDSHAKE(REKEY_TIMEOUT)`
//! every 5 s, then `CONNECTION_EXPIRED`) — and the installer's next handshake
//! reopens it. Proven on the real link: tests/installer_link_goes_quiet.rs.
//!
//! **The link's addresses are the link's own**, the same on every VM:
//! [`LINK_INSTALLER`] and [`LINK_VM`] on a /30. That is not a clash — each VM
//! is its own WireGuard session on the installer's side (one session per VM,
//! in user space), so two VMs with the same link address are two different
//! wires.
//!
//! **The installer's side runs in user space** (feature `installer-wireguard`):
//! the operator's machine holds no TUN device and no root, so the WireGuard
//! session is boringtun's (the suite's one engine) and the TCP inside it is
//! smoltcp's. The VM's side is a real interface ([`vm_link`], feature
//! `wireguard`): the VM is PID 1 and owns its network.

use crate::seed::TunnelKind;
use crate::tunnel::TunnelError;
use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpStream};
use std::time::Duration;

/// The installer's address on the link.
pub const LINK_INSTALLER: Ipv4Addr = Ipv4Addr::new(10, 45, 0, 1);
/// Every VM's address on the link.
pub const LINK_VM: Ipv4Addr = Ipv4Addr::new(10, 45, 0, 2);
pub const LINK_PREFIX: u8 = 30;
/// The UDP port each VM's end of the link listens on (the mesh's relay on
/// the front is 51820).
pub const LINK_PORT: u16 = 51821;
/// The VM's link interface.
pub const LINK_IFACE: &str = "wg-installer";
/// The MTU inside the link.
pub const LINK_MTU: usize = 1380;

/// **The path the installer takes to a VM** — since 0.4.0 exactly one: the
/// installer's WireGuard link, under mutual TLS (owner 2026-09-25).
/// `tunnel = none` is refused by name (`[installer-link-none]`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reach {
    /// Through the installer's WireGuard link (`wireguard`, `mullvad`).
    Tunnel,
}

/// **The one decision**, shared by the VM (where to bind, whether to raise a
/// link) and the installer (how to dial).
pub fn reach(kind: TunnelKind) -> Result<Reach, TunnelError> {
    match kind {
        TunnelKind::None => Err(TunnelError::InstallerLinkNone),
        TunnelKind::Wireguard | TunnelKind::Mullvad => Ok(Reach::Tunnel),
        TunnelKind::Tailscale => Err(TunnelError::Unimplemented {
            kind: TunnelKind::Tailscale,
            missing: "a one-time installer node on the tailnet (a pure-Rust Tailscale coordination client does not exist in the suite)",
        }),
    }
}

/// Where a VM binds its bootstrap listener: the link address ONLY — never
/// the public interface.
pub fn vm_bind(r: Reach) -> Ipv4Addr {
    match r {
        Reach::Tunnel => LINK_VM,
    }
}

/// A byte stream the installer speaks TLS over.
pub trait Wire: Read + Write {}
impl<T: Read + Write> Wire for T {}

/// **The installer's path to ONE VM.**
pub trait Path: Send {
    /// One connection to the VM's bootstrap listener.
    fn connect(&mut self, timeout: Duration) -> Result<Box<dyn Wire + '_>, String>;
    /// After the close: does the VM still answer on this path? `Ok(true)` =
    /// nothing answers within `wait` (the listener is gone, the link dropped).
    fn gone(&mut self, wait: Duration) -> Result<bool, String>;
    fn describe(&self) -> String;
}

/// **A loopback harness for tests** (0.4.0): TCP straight to an address.
/// [`reach`] never chooses it — production is always the link.
pub struct DirectPath {
    pub at: SocketAddrV4,
}

impl Path for DirectPath {
    fn connect(&mut self, timeout: Duration) -> Result<Box<dyn Wire + '_>, String> {
        let s = TcpStream::connect_timeout(&SocketAddr::V4(self.at), timeout).map_err(|e| format!("could not be reached: {}: {e}", self.at))?;
        let _ = s.set_read_timeout(Some(timeout));
        let _ = s.set_write_timeout(Some(timeout));
        Ok(Box::new(s))
    }

    /// Asked until the port refuses (the VM answers its close first and
    /// drops the listener a moment after).
    fn gone(&mut self, wait: Duration) -> Result<bool, String> {
        let t0 = std::time::Instant::now();
        while t0.elapsed() < wait {
            if TcpStream::connect_timeout(&SocketAddr::V4(self.at), Duration::from_secs(2)).is_err() {
                return Ok(true);
            }
            std::thread::sleep(Duration::from_millis(250));
        }
        Ok(false)
    }

    fn describe(&self) -> String {
        format!("{} (direct mutual TLS)", self.at)
    }
}

/// The installer's keys for the link, kept on the operator's machine: its own
/// private key, and the public half of the medium's VM key.
#[derive(Clone, PartialEq, Eq)]
pub struct LinkKeys {
    pub installer_private: [u8; 32],
    pub vm_public: [u8; 32],
}

impl std::fmt::Debug for LinkKeys {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("LinkKeys { .. }")
    }
}

#[cfg(feature = "installer-wireguard")]
pub use wg::WgPath;

#[cfg(feature = "installer-wireguard")]
mod wg {
    use super::*;
    use boringtun::noise::{Tunn, TunnResult};
    use boringtun::x25519::{PublicKey, StaticSecret};
    use smoltcp::iface::{Config, Interface, SocketHandle, SocketSet};
    use smoltcp::phy::{Device, DeviceCapabilities, Medium};
    use smoltcp::socket::tcp;
    use smoltcp::wire::{HardwareAddress, IpAddress, IpCidr};
    use std::collections::VecDeque;
    use std::net::UdpSocket;
    use std::time::Instant;

    /// The in-memory "device" smoltcp reads and writes: IP packets out of the
    /// tunnel (rx) and into it (tx).
    #[derive(Default)]
    struct Queues {
        rx: VecDeque<Vec<u8>>,
        tx: VecDeque<Vec<u8>>,
    }

    struct Rx(Vec<u8>);
    struct Tx<'a>(&'a mut VecDeque<Vec<u8>>);

    impl smoltcp::phy::RxToken for Rx {
        fn consume<R, F: FnOnce(&[u8]) -> R>(self, f: F) -> R {
            f(&self.0)
        }
    }

    impl smoltcp::phy::TxToken for Tx<'_> {
        fn consume<R, F: FnOnce(&mut [u8]) -> R>(self, len: usize, f: F) -> R {
            let mut buf = vec![0u8; len];
            let r = f(&mut buf);
            self.0.push_back(buf);
            r
        }
    }

    impl Device for Queues {
        type RxToken<'a> = Rx;
        type TxToken<'a> = Tx<'a>;
        fn receive(&mut self, _t: smoltcp::time::Instant) -> Option<(Rx, Tx<'_>)> {
            let p = self.rx.pop_front()?;
            Some((Rx(p), Tx(&mut self.tx)))
        }
        fn transmit(&mut self, _t: smoltcp::time::Instant) -> Option<Tx<'_>> {
            Some(Tx(&mut self.tx))
        }
        fn capabilities(&self) -> DeviceCapabilities {
            let mut c = DeviceCapabilities::default();
            c.medium = Medium::Ip;
            c.max_transmission_unit = LINK_MTU;
            c
        }
    }

    /// **One VM's link, the engine** — a WireGuard session (boringtun) over
    /// one UDP socket, and a TCP/IP stack (smoltcp) inside it, all in user
    /// space. ONE per VM per process ([`LINKS`]): the VM keeps one endpoint
    /// for its one peer (the installer) and moves it to the newest datagram's
    /// source, so a second socket under the same key would steal the first
    /// one's answers (estate-box3 two-link-readers).
    struct Engine {
        endpoint: SocketAddrV4,
        udp: UdpSocket,
        tunn: Tunn,
        iface: Interface,
        dev: Queues,
        sockets: SocketSet<'static>,
        epoch: Instant,
        next_port: u16,
        net: Vec<u8>,
    }

    type Shared = std::sync::Arc<std::sync::Mutex<Engine>>;

    /// Every link this process holds, by (VM endpoint, installer key, VM key):
    /// its engine, and the turn at the VM's door its readers take.
    static LINKS: std::sync::Mutex<Vec<Link>> = std::sync::Mutex::new(Vec::new());

    /// One entry of [`LINKS`]: (VM endpoint, installer key, VM key), the engine, the turn.
    type Link = ((SocketAddrV4, [u8; 32], [u8; 32]), Shared, std::sync::Arc<Turn>);

    /// **The turn at the VM's door.** The box's bootstrap listener and log
    /// door (`bootstrap::serve_*`) serve ONE connection at a time, while its
    /// kernel establishes every connection at once: a second reader's
    /// connect succeeds, its request waits in the box's queue, and its read
    /// timeout runs the whole time another reader is being served — so a
    /// slow exchange of one reader cost the next its answer (estate-box W6,
    /// `W6-two-readers-one-link`: "reader A #0: the request: nothing from
    /// the VM over the link" while reader B got all four, release gate red 2
    /// of 3). The readers of one VM in this process therefore take turns
    /// HERE, in the order they asked: a connection is opened when its reader
    /// has the turn, and a reader's timeout is for its own exchange.
    #[derive(Default)]
    struct Turn {
        q: std::sync::Mutex<TurnQueue>,
        moved: std::sync::Condvar,
    }

    /// A ticket queue: the order readers asked in, the ticket served, and
    /// tickets whose readers stopped waiting.
    #[derive(Default)]
    struct TurnQueue {
        next: u64,
        serving: u64,
        given_up: std::collections::BTreeSet<u64>,
    }

    /// The turn, held: given back (to the next ticket still waiting) when dropped.
    struct Held(std::sync::Arc<Turn>);

    impl Drop for Held {
        fn drop(&mut self) {
            let mut q = self.0.q.lock().unwrap_or_else(|p| p.into_inner());
            q.serving += 1;
            loop {
                let at = q.serving;
                if !q.given_up.remove(&at) {
                    break;
                }
                q.serving += 1;
            }
            self.0.moved.notify_all();
        }
    }

    /// **How long a reader waits for the turn** before it goes ahead without
    /// it (as every reader did before turns). The box gives one connection at
    /// most `bootstrap::REQUEST_BUDGET` to ask; a turn held longer is a
    /// reader that stopped (a process that died mid-answer, a stream leaked),
    /// and the others must not wait on it for ever.
    const TURN_WAIT: Duration = crate::bootstrap::REQUEST_BUDGET;

    /// Take the turn: `Some` once it is this reader's, `None` after `within`.
    fn take_turn(turn: &std::sync::Arc<Turn>, within: Duration) -> Option<Held> {
        let mut q = turn.q.lock().unwrap_or_else(|p| p.into_inner());
        let mine = q.next;
        q.next += 1;
        let until = Instant::now() + within;
        while q.serving != mine {
            let left = until.saturating_duration_since(Instant::now());
            if left.is_zero() {
                q.given_up.insert(mine);
                return None;
            }
            q = turn.moved.wait_timeout(q, left).unwrap_or_else(|p| p.into_inner()).0;
        }
        Some(Held(turn.clone()))
    }

    fn lock(e: &Shared) -> std::sync::MutexGuard<'_, Engine> {
        e.lock().unwrap_or_else(|p| p.into_inner())
    }

    /// **The installer as a temporary peer of one VM's link** — a handle on
    /// the process's one [`Engine`] for that VM: every reader in this process
    /// (the install, the capture follower, a probe, a re-admission) shares
    /// its socket and session, each with its own TCP connection inside.
    pub struct WgPath {
        endpoint: SocketAddrV4,
        keys: LinkKeys,
        engine: Shared,
        turn: std::sync::Arc<Turn>,
    }

    fn now(epoch: Instant) -> smoltcp::time::Instant {
        smoltcp::time::Instant::from_millis(epoch.elapsed().as_millis() as i64)
    }

    fn tunn_of(keys: &LinkKeys, index: u32) -> Tunn {
        Tunn::new(StaticSecret::from(keys.installer_private), PublicKey::from(keys.vm_public), None, Some(15), index, None)
    }

    impl Engine {
        fn new(endpoint: SocketAddrV4, keys: &LinkKeys) -> Result<Engine, String> {
            let udp = UdpSocket::bind("0.0.0.0:0").map_err(|e| format!("the installer's link socket: {e}"))?;
            udp.connect(endpoint).map_err(|e| format!("the installer's link to {endpoint}: {e}"))?;
            udp.set_read_timeout(Some(Duration::from_millis(5))).map_err(|e| e.to_string())?;
            let epoch = Instant::now();
            let mut dev = Queues::default();
            let mut cfg = Config::new(HardwareAddress::Ip);
            cfg.random_seed = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos() as u64).unwrap_or(7);
            let mut iface = Interface::new(cfg, &mut dev, now(epoch));
            iface.update_ip_addrs(|a| {
                let _ = a.push(IpCidr::new(IpAddress::Ipv4(LINK_INSTALLER), LINK_PREFIX));
            });
            let seed_port = 40000 + (cfg_port_seed() % 20000) as u16;
            Ok(Engine { endpoint, udp, tunn: tunn_of(keys, 1), iface, dev, sockets: SocketSet::new(Vec::new()), epoch, next_port: seed_port, net: vec![0u8; 65536] })
        }

        /// One turn of the engine: timers, datagrams in, the stack, packets out.
        fn pump(&mut self) {
            if let TunnResult::WriteToNetwork(p) = self.tunn.update_timers(&mut self.net) {
                let _ = self.udp.send(p);
            }
            let mut dgram = [0u8; 2048];
            for _ in 0..64 {
                let n = match self.udp.recv(&mut dgram) {
                    Ok(n) => n,
                    Err(_) => break,
                };
                let mut first = true;
                loop {
                    let r = self.tunn.decapsulate(if first { Some(std::net::IpAddr::V4(*self.endpoint.ip())) } else { None }, if first { &dgram[..n] } else { &[] }, &mut self.net);
                    first = false;
                    match r {
                        TunnResult::WriteToNetwork(p) => {
                            let _ = self.udp.send(p);
                        }
                        TunnResult::WriteToTunnelV4(p, src) => {
                            if src == LINK_VM {
                                self.dev.rx.push_back(p.to_vec());
                            }
                            break;
                        }
                        _ => break,
                    }
                }
            }
            let t = now(self.epoch);
            self.iface.poll(t, &mut self.dev, &mut self.sockets);
            while let Some(p) = self.dev.tx.pop_front() {
                if let TunnResult::WriteToNetwork(out) = self.tunn.encapsulate(&p, &mut self.net) {
                    let _ = self.udp.send(out);
                }
            }
        }
    }

    impl WgPath {
        /// The link to the VM whose end listens at `endpoint` (its public
        /// address and [`LINK_PORT`]) — this process's one engine for it,
        /// made on first use.
        pub fn new(endpoint: SocketAddrV4, keys: LinkKeys) -> Result<WgPath, String> {
            let id = (endpoint, keys.installer_private, keys.vm_public);
            let mut links = LINKS.lock().unwrap_or_else(|p| p.into_inner());
            let (engine, turn) = match links.iter().find(|(k, _, _)| *k == id) {
                Some((_, e, t)) => (e.clone(), t.clone()),
                None => {
                    let e: Shared = std::sync::Arc::new(std::sync::Mutex::new(Engine::new(endpoint, &keys)?));
                    let t = std::sync::Arc::new(Turn::default());
                    links.push((id, e.clone(), t.clone()));
                    (e, t)
                }
            };
            Ok(WgPath { endpoint, keys, engine, turn })
        }

        /// One TCP connection to the VM's listener inside the link, within `timeout`.
        fn try_connect(&mut self, timeout: Duration) -> Result<SocketHandle, String> {
            let sock = tcp::Socket::new(tcp::SocketBuffer::new(vec![0u8; 65536]), tcp::SocketBuffer::new(vec![0u8; 65536]));
            let h = {
                let mut e = lock(&self.engine);
                let h = e.sockets.add(sock);
                e.next_port = if e.next_port >= 60000 { 40000 } else { e.next_port + 1 };
                let port = e.next_port;
                let Engine { iface, sockets, .. } = &mut *e;
                let cx = iface.context();
                let s = sockets.get_mut::<tcp::Socket>(h);
                if let Err(err) = s.connect(cx, (IpAddress::Ipv4(LINK_VM), crate::bootstrap::PORT), port) {
                    sockets.remove(h);
                    return Err(format!("the link's TCP to {LINK_VM}: {err:?}"));
                }
                h
            };
            let t0 = Instant::now();
            loop {
                {
                    let mut e = lock(&self.engine);
                    e.pump();
                    let s = e.sockets.get_mut::<tcp::Socket>(h);
                    match s.state() {
                        tcp::State::Established => return Ok(h),
                        tcp::State::Closed | tcp::State::TimeWait => {
                            e.sockets.remove(h);
                            return Err(format!("could not be reached: the VM at {} refused the bootstrap listener inside the link", self.endpoint));
                        }
                        _ => {}
                    }
                    if t0.elapsed() > timeout {
                        e.sockets.get_mut::<tcp::Socket>(h).abort();
                        e.pump();
                        e.sockets.remove(h);
                        let hs = match e.tunn.stats().0 {
                            Some(ago) => format!("the link handshook {} s ago", ago.as_secs()),
                            None => "no WireGuard handshake yet".to_string(),
                        };
                        return Err(format!("could not be reached: {} over the installer's link ({hs})", self.endpoint));
                    }
                }
                std::thread::yield_now();
            }
        }

        /// A connect that found no VM: the session starts over (a VM that
        /// restarted has forgotten it; waiting out boringtun's re-key would
        /// cost a reader its whole timeout) — as a NEW HANDSHAKE on the link's
        /// one `Tunn`, sent now, never a new `Tunn`.
        ///
        /// The session is every reader's ([`LINKS`]): a new `Tunn` threw away
        /// the one another reader's open connection was using, so whatever the
        /// VM had already sealed under it — that reader's answer, its TCP's
        /// retransmissions — could no longer be opened, and the reader waited
        /// out the box's TCP back-off on top of the stall that made this
        /// connect slow (estate-box W6, `W6-two-readers-one-link`: a box busy
        /// for a few seconds became a read lost past its 20 s). A handshake on
        /// the same `Tunn` keeps the old session in its ring until the new one
        /// takes over, and a VM that did forget it answers the handshake.
        fn fresh_session(&self) {
            let mut e = lock(&self.engine);
            let Engine { tunn, udp, net, .. } = &mut *e;
            if let TunnResult::WriteToNetwork(p) = tunn.format_handshake_initiation(net, true) {
                let _ = udp.send(p);
            }
        }
    }

    fn cfg_port_seed() -> u64 {
        std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.subsec_nanos() as u64).unwrap_or(1)
    }

    /// One TCP connection inside the link — and, while it lives, the turn at
    /// the VM's door ([`Turn`]; given back after the connection is closed).
    struct LinkStream {
        engine: Shared,
        h: SocketHandle,
        timeout: Duration,
        _turn: Option<Held>,
    }

    impl Read for LinkStream {
        fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
            let t0 = Instant::now();
            loop {
                {
                    let mut e = lock(&self.engine);
                    e.pump();
                    let s = e.sockets.get_mut::<tcp::Socket>(self.h);
                    if s.can_recv() {
                        let n = s.recv_slice(buf).map_err(|e| std::io::Error::other(format!("{e:?}")))?;
                        if n > 0 {
                            return Ok(n);
                        }
                    } else if !s.may_recv() {
                        return Ok(0);
                    }
                }
                if t0.elapsed() > self.timeout {
                    return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "nothing from the VM over the link"));
                }
                std::thread::yield_now();
            }
        }
    }

    impl Write for LinkStream {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            let t0 = Instant::now();
            loop {
                {
                    let mut e = lock(&self.engine);
                    e.pump();
                    let s = e.sockets.get_mut::<tcp::Socket>(self.h);
                    if s.can_send() {
                        let n = s.send_slice(buf).map_err(|e| std::io::Error::other(format!("{e:?}")))?;
                        if n > 0 {
                            return Ok(n);
                        }
                    } else if !s.may_send() {
                        return Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "the VM closed the connection"));
                    }
                }
                if t0.elapsed() > self.timeout {
                    return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "the link would not take the bytes"));
                }
                std::thread::yield_now();
            }
        }

        fn flush(&mut self) -> std::io::Result<()> {
            let t0 = Instant::now();
            loop {
                {
                    let mut e = lock(&self.engine);
                    e.pump();
                    let s = e.sockets.get_mut::<tcp::Socket>(self.h);
                    if s.send_queue() == 0 || !s.may_send() {
                        return Ok(());
                    }
                }
                if t0.elapsed() > self.timeout {
                    return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "the link would not drain"));
                }
                std::thread::yield_now();
            }
        }
    }

    impl Drop for LinkStream {
        fn drop(&mut self) {
            let mut e = lock(&self.engine);
            e.sockets.get_mut::<tcp::Socket>(self.h).close();
            for _ in 0..20 {
                e.pump();
                if !e.sockets.get_mut::<tcp::Socket>(self.h).is_open() {
                    break;
                }
            }
            e.sockets.remove(self.h);
        }
    }

    impl Path for WgPath {
        /// The turn at the VM's door first ([`Turn`]), then the connection;
        /// `timeout` counts from the turn, never from the wait for it.
        fn connect(&mut self, timeout: Duration) -> Result<Box<dyn Wire + '_>, String> {
            let turn = take_turn(&self.turn, TURN_WAIT);
            let t0 = Instant::now();
            let mut fresh = false;
            loop {
                match self.try_connect(timeout.saturating_sub(t0.elapsed()).min(if fresh { timeout } else { Duration::from_secs(4) })) {
                    Ok(h) => return Ok(Box::new(LinkStream { engine: self.engine.clone(), h, timeout, _turn: turn })),
                    Err(e) if fresh || t0.elapsed() >= timeout => return Err(e),
                    // No answer within a few seconds on the shared session: the
                    // VM may have restarted and forgotten it — start over once.
                    Err(e) if e.starts_with("could not be reached: the VM at") => return Err(e),
                    Err(_) => {
                        fresh = true;
                        self.fresh_session();
                    }
                }
            }
        }

        /// A FRESH handshake each round (a new session, so nothing cached
        /// answers): the link is gone once a round of three seconds brings no
        /// answer. The VM answers its close first and drops the link a moment
        /// after, so an answer early in `wait` is asked again.
        fn gone(&mut self, wait: Duration) -> Result<bool, String> {
            let t0 = Instant::now();
            let mut index = 2u32;
            let mut dgram = [0u8; 2048];
            let mut net = vec![0u8; 2048];
            // Its OWN socket: the probe must not move the VM's endpoint off
            // the shared engine every other reader in this process uses.
            let udp = UdpSocket::bind("0.0.0.0:0").map_err(|e| format!("the installer's probe socket: {e}"))?;
            udp.connect(self.endpoint).map_err(|e| format!("the installer's probe to {}: {e}", self.endpoint))?;
            udp.set_read_timeout(Some(Duration::from_millis(5))).map_err(|e| e.to_string())?;
            while t0.elapsed() < wait {
                index += 1;
                let mut t = tunn_of(&self.keys, index);
                let mut buf = [0u8; 256];
                let init = match t.format_handshake_initiation(&mut buf, true) {
                    TunnResult::WriteToNetwork(p) => p.to_vec(),
                    other => return Err(format!("boringtun: {other:?}")),
                };
                let _ = udp.send(&init);
                let round = Instant::now();
                let mut answered = false;
                while round.elapsed() < Duration::from_secs(3) {
                    if let Ok(n) = udp.recv(&mut dgram) {
                        if matches!(t.decapsulate(Some(std::net::IpAddr::V4(*self.endpoint.ip())), &dgram[..n], &mut net), TunnResult::WriteToNetwork(_) | TunnResult::Done) {
                            answered = true;
                            break;
                        }
                    }
                }
                if !answered {
                    return Ok(true);
                }
                std::thread::sleep(Duration::from_millis(500));
            }
            Ok(false)
        }

        fn describe(&self) -> String {
            format!("{} (the installer's WireGuard link, {LINK_INSTALLER} → {LINK_VM})", self.endpoint)
        }
    }

    /// ★ estate-box W6 (`W6-two-readers-one-link`, release gate red 2 of 3 on
    /// byte-identical binaries, oden 2026-10-04: "reader A #0: the request:
    /// nothing from the VM over the link", rounds of 8-38 s): every reader in
    /// the process shares THIS engine's one session ([`LINKS`]), so the
    /// session belongs to all of them. A connect that hears nothing for 4 s —
    /// a box busy or stalled for those seconds, while another reader's
    /// connection is open on the link — must not throw that session away:
    /// what the VM already sealed under it (an answer, its TCP's
    /// retransmissions) would be unreadable, and the other reader waits out
    /// the box's TCP back-off on top of the stall.
    #[cfg(test)]
    mod start_over_tests {
        use super::*;

        fn installer_public(k: &LinkKeys) -> [u8; 32] {
            PublicKey::from(&StaticSecret::from(k.installer_private)).to_bytes()
        }

        /// The smallest IPv4 packet the VM's end could seal for the installer.
        fn packet_for_the_installer() -> Vec<u8> {
            let mut p = vec![0u8; 20];
            p[0] = 0x45;
            p[3] = 20;
            p[8] = 64;
            p[9] = 6;
            p[12..16].copy_from_slice(&LINK_VM.octets());
            p[16..20].copy_from_slice(&LINK_INSTALLER.octets());
            p
        }

        /// The VM's end, by hand: answer every datagram the engine sent it.
        fn vm_answers(vm_udp: &UdpSocket, vm: &mut Tunn) -> usize {
            let (mut dgram, mut out) = ([0u8; 2048], vec![0u8; 2048]);
            let mut handled = 0;
            while let Ok((n, from)) = vm_udp.recv_from(&mut dgram) {
                handled += 1;
                let mut first = true;
                loop {
                    let r = vm.decapsulate(None, if first { &dgram[..n] } else { &[] }, &mut out);
                    first = false;
                    match r {
                        TunnResult::WriteToNetwork(p) => {
                            let _ = vm_udp.send_to(p, from);
                        }
                        _ => break,
                    }
                }
            }
            handled
        }

        /// The link up: the engine handshakes, the VM answers, the engine confirms.
        fn handshake(path: &WgPath, vm_udp: &UdpSocket, vm: &mut Tunn) {
            {
                let mut e = lock(&path.engine);
                let Engine { tunn, udp, net, .. } = &mut *e;
                if let TunnResult::WriteToNetwork(p) = tunn.format_handshake_initiation(net, false) {
                    udp.send(p).unwrap();
                }
            }
            let t0 = Instant::now();
            while lock(&path.engine).tunn.stats().0.is_none() && t0.elapsed() < Duration::from_secs(5) {
                vm_answers(vm_udp, vm);
                lock(&path.engine).pump();
            }
            assert!(lock(&path.engine).tunn.stats().0.is_some(), "the link never handshook");
            vm_answers(vm_udp, vm); // the engine's confirming keepalive
        }

        /// Does the engine open what the VM sealed under its current session?
        fn engine_opens(path: &WgPath, vm: &mut Tunn) -> bool {
            let mut sealed = vec![0u8; 2048];
            let sealed = match vm.encapsulate(&packet_for_the_installer(), &mut sealed) {
                TunnResult::WriteToNetwork(p) => p.to_vec(),
                other => panic!("the VM could not seal: {other:?}"),
            };
            let mut e = lock(&path.engine);
            let Engine { tunn, net, endpoint, .. } = &mut *e;
            matches!(tunn.decapsulate(Some(std::net::IpAddr::V4(*endpoint.ip())), &sealed, net), TunnResult::WriteToTunnelV4(..))
        }

        #[test]
        fn a_reader_that_starts_the_session_over_keeps_the_session_the_other_readers_use() {
            let (vm_private, k) = ([3u8; 32], LinkKeys { installer_private: [9u8; 32], vm_public: PublicKey::from(&StaticSecret::from([3u8; 32])).to_bytes() });
            let vm_udp = UdpSocket::bind("127.0.0.1:0").unwrap();
            vm_udp.set_read_timeout(Some(Duration::from_millis(50))).unwrap();
            let at = match vm_udp.local_addr().unwrap() {
                SocketAddr::V4(a) => a,
                _ => unreachable!(),
            };
            let mut vm = Tunn::new(StaticSecret::from(vm_private), PublicKey::from(installer_public(&k)), None, None, 7, None);
            let reader = WgPath::new(at, k.clone()).unwrap();
            handshake(&reader, &vm_udp, &mut vm);
            assert!(engine_opens(&reader, &mut vm), "the link is up: the engine opens what the VM seals");

            // Another reader's connect heard nothing for 4 s (the box was busy)
            // and starts the session over, while this reader's connection is open.
            let other = WgPath::new(at, k.clone()).unwrap();
            other.fresh_session();
            // The VM has not heard of any of that yet: it seals the open
            // connection's answer under the session it has.
            assert!(engine_opens(&reader, &mut vm), "a reader starting the session over threw away the session another reader's open connection uses: what the VM sealed under it is unreadable now");

            // …and a VM that DID forget it (it restarted) is reached again.
            vm = Tunn::new(StaticSecret::from(vm_private), PublicKey::from(installer_public(&k)), None, None, 11, None);
            other.fresh_session();
            let t0 = Instant::now();
            while t0.elapsed() < Duration::from_secs(5) {
                vm_answers(&vm_udp, &mut vm);
                lock(&other.engine).pump();
                if vm.stats().0.is_some() && vm_answers(&vm_udp, &mut vm) == 0 && engine_opens(&other, &mut vm) {
                    return;
                }
            }
            panic!("after starting over, the engine never reached the VM that had forgotten the session");
        }
    }
}

/// **The VM's end of the link** (feature `wireguard`): a real interface
/// [`LINK_IFACE`] at [`LINK_VM`], the medium's VM key, the installer as its
/// ONE peer (endpoint learned from the installer's handshake), listening on
/// the seed's port. `drop_installer` takes the whole link down.
#[cfg(feature = "wireguard")]
pub struct VmLink {
    handle: Option<wg_appliance_core::TunnelHandle>,
    pub bind: Ipv4Addr,
}

#[cfg(feature = "wireguard")]
impl VmLink {
    /// Whether the installer's link is up (and whether it has handshaken).
    pub fn up(&self) -> bool {
        self.handle.is_some()
    }

    pub fn handshaken(&self) -> bool {
        self.handle.as_ref().map(|h| h.handshaken()).unwrap_or(false)
    }

    /// **Drop the installer as a peer**: the link goes down whole.
    pub fn drop_installer(&mut self) {
        if let Some(mut h) = self.handle.take() {
            h.shutdown();
        }
    }
}

/// Raise the VM's end of the link for this seed. `tunnel = none` is refused
/// (`[installer-link-none]`), and the bootstrap listener then never opens.
#[cfg(feature = "wireguard")]
pub fn vm_link(seed: &crate::seed::Seed) -> Result<VmLink, TunnelError> {
    use wg_appliance_core::{addr, dataplane, decode_key, PeerSpec, TunDevice, TunnelSpec};
    match reach(seed.tunnel)? {
        Reach::Tunnel => {
            let l = seed.installer_link.as_ref().ok_or_else(|| TunnelError::NotConfigured { kind: seed.tunnel, what: "the seed carries no [installer_link]".into() })?;
            let private = decode_key(&l.vm_private).map_err(|e| TunnelError::Failed(format!("installer_link.vm_private: {}", e.0)))?;
            let installer = decode_key(&l.installer_pubkey).map_err(|e| TunnelError::Failed(format!("installer_link.installer_pubkey: {}", e.0)))?;
            let tun = TunDevice::create(LINK_IFACE, LINK_MTU as u32 + 40).map_err(|e| TunnelError::Failed(format!("{LINK_IFACE}: {e}")))?;
            addr::assign_ipv4(LINK_IFACE, LINK_VM, LINK_PREFIX).map_err(|e| TunnelError::Failed(format!("{LINK_IFACE} {LINK_VM}/{LINK_PREFIX}: {e}")))?;
            addr::link_up(LINK_IFACE).map_err(|e| TunnelError::Failed(format!("{LINK_IFACE} up: {e}")))?;
            let spec = TunnelSpec {
                private_key: private,
                listen_port: Some(l.port),
                peers: vec![PeerSpec { public_key: installer, preshared_key: None, endpoint: None, allowed_ips: vec![(LINK_INSTALLER, 32)], persistent_keepalive: None }],
            };
            // Named: its timer lines say `wg{peer=wg-installer:installer:<fp>}`.
            let handle = dataplane::spawn_named(tun, &spec, &["installer"]).map_err(|e| TunnelError::Failed(format!("{LINK_IFACE}: {e}")))?;
            Ok(VmLink { handle: Some(handle), bind: LINK_VM })
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn the_path_is_the_tunnel_kinds_and_tailscale_is_refused_by_name() {
        assert!(reach(TunnelKind::None).unwrap_err().to_string().starts_with("REFUSED [installer-link-none]"), "WireGuard + mutual TLS always");
        assert_eq!(reach(TunnelKind::Wireguard).unwrap(), Reach::Tunnel);
        assert_eq!(reach(TunnelKind::Mullvad).unwrap(), Reach::Tunnel, "mullvad is only the exit; the mesh is WireGuard");
        let e = reach(TunnelKind::Tailscale).unwrap_err().to_string();
        assert!(e.starts_with("REFUSED [tunnel-tailscale-unimplemented]"), "{e}");
        assert_eq!(vm_bind(Reach::Tunnel), LINK_VM, "never the public interface in tunnel mode");
        assert!(!crate::in_overlay(LINK_VM) && !crate::in_overlay(LINK_INSTALLER), "the link is not the mesh");
        assert!(format!("{:?}", LinkKeys { installer_private: [7; 32], vm_public: [1; 32] }).contains(".."), "no key in Debug");
    }
}

/// ★ estate-box3 (two-link-readers, MEASURED plain3/plain5: P5 "a hello
/// after the close: peer closed connection", T2 "the installer's own knock
/// failed", W1 "the VM closed the connection" — every one while the capture
/// follower polled the same box): a VM's link keeps ONE endpoint for its one
/// peer, the installer, and moves it to wherever the newest authenticated
/// datagram came from. Every `WgPath::new` bound a NEW UDP socket and a new
/// session under the one installer key, so two readers in one process (the
/// capture and a check, or a re-admission) stole each other's answers.
#[cfg(all(test, feature = "installer-wireguard"))]
mod shared_link_tests {
    use super::*;
    use boringtun::noise::{Tunn, TunnResult};
    use boringtun::x25519::{PublicKey, StaticSecret};
    use smoltcp::iface::{Config, Interface, SocketSet};
    use smoltcp::phy::{Device, DeviceCapabilities, Medium};
    use smoltcp::socket::tcp;
    use smoltcp::wire::{HardwareAddress, IpAddress, IpCidr};
    use std::collections::VecDeque;
    use std::net::UdpSocket;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use std::time::Instant;

    #[derive(Default)]
    struct Q {
        rx: VecDeque<Vec<u8>>,
        tx: VecDeque<Vec<u8>>,
    }
    struct Rx(Vec<u8>);
    struct Tx<'a>(&'a mut VecDeque<Vec<u8>>);
    impl smoltcp::phy::RxToken for Rx {
        fn consume<R, F: FnOnce(&[u8]) -> R>(self, f: F) -> R {
            f(&self.0)
        }
    }
    impl smoltcp::phy::TxToken for Tx<'_> {
        fn consume<R, F: FnOnce(&mut [u8]) -> R>(self, len: usize, f: F) -> R {
            let mut b = vec![0u8; len];
            let r = f(&mut b);
            self.0.push_back(b);
            r
        }
    }
    impl Device for Q {
        type RxToken<'a> = Rx;
        type TxToken<'a> = Tx<'a>;
        fn receive(&mut self, _t: smoltcp::time::Instant) -> Option<(Rx, Tx<'_>)> {
            let p = self.rx.pop_front()?;
            Some((Rx(p), Tx(&mut self.tx)))
        }
        fn transmit(&mut self, _t: smoltcp::time::Instant) -> Option<Tx<'_>> {
            Some(Tx(&mut self.tx))
        }
        fn capabilities(&self) -> DeviceCapabilities {
            let mut c = DeviceCapabilities::default();
            c.medium = Medium::Ip;
            c.max_transmission_unit = LINK_MTU;
            c
        }
    }

    /// A VM's end of the link, in user space, as WireGuard keeps it: ONE
    /// peer, whose endpoint is where its newest authenticated datagram came
    /// from. Its listener answers every line `x` with `ok x`.
    fn fake_vm(vm: [u8; 32], installer_public: [u8; 32], stop: Arc<AtomicBool>) -> SocketAddrV4 {
        let udp = UdpSocket::bind("127.0.0.1:0").unwrap();
        udp.set_read_timeout(Some(Duration::from_millis(2))).unwrap();
        let at = match udp.local_addr().unwrap() {
            SocketAddr::V4(a) => a,
            _ => unreachable!(),
        };
        std::thread::spawn(move || {
            let mut tunn = Tunn::new(StaticSecret::from(vm), PublicKey::from(installer_public), None, None, 7, None);
            let epoch = Instant::now();
            let now = || smoltcp::time::Instant::from_millis(epoch.elapsed().as_millis() as i64);
            let mut dev = Q::default();
            let mut iface = Interface::new(Config::new(HardwareAddress::Ip), &mut dev, now());
            iface.update_ip_addrs(|a| {
                let _ = a.push(IpCidr::new(IpAddress::Ipv4(LINK_VM), LINK_PREFIX));
            });
            let mut sockets = SocketSet::new(Vec::new());
            let mut listeners = Vec::new();
            for _ in 0..4 {
                let mut s = tcp::Socket::new(tcp::SocketBuffer::new(vec![0u8; 8192]), tcp::SocketBuffer::new(vec![0u8; 8192]));
                s.listen(crate::bootstrap::PORT).unwrap();
                listeners.push((sockets.add(s), Vec::<u8>::new()));
            }
            let mut endpoint: Option<SocketAddr> = None;
            let (mut net, mut dgram) = (vec![0u8; 65536], [0u8; 2048]);
            while !stop.load(Ordering::SeqCst) {
                if let TunnResult::WriteToNetwork(p) = tunn.update_timers(&mut net) {
                    if let Some(e) = endpoint {
                        let _ = udp.send_to(p, e);
                    }
                }
                while let Ok((n, from)) = udp.recv_from(&mut dgram) {
                    let mut first = true;
                    loop {
                        let r = tunn.decapsulate(None, if first { &dgram[..n] } else { &[] }, &mut net);
                        first = false;
                        match r {
                            TunnResult::WriteToNetwork(p) => {
                                endpoint = Some(from);
                                let _ = udp.send_to(p, from);
                            }
                            TunnResult::WriteToTunnelV4(p, _) => {
                                endpoint = Some(from);
                                dev.rx.push_back(p.to_vec());
                                break;
                            }
                            TunnResult::Done => {
                                endpoint = Some(from);
                                break;
                            }
                            _ => break,
                        }
                    }
                }
                iface.poll(now(), &mut dev, &mut sockets);
                for (h, got) in listeners.iter_mut() {
                    let s = sockets.get_mut::<tcp::Socket>(*h);
                    if s.can_recv() {
                        let _ = s.recv(|b| {
                            got.extend_from_slice(b);
                            (b.len(), ())
                        });
                    }
                    if let Some(i) = got.iter().position(|b| *b == b'\n') {
                        if s.can_send() {
                            let line: Vec<u8> = got.drain(..=i).collect();
                            let mut out = b"ok ".to_vec();
                            out.extend_from_slice(&line);
                            let _ = s.send_slice(&out);
                            s.close();
                        }
                    }
                    if !s.is_open() {
                        got.clear();
                        let _ = s.listen(crate::bootstrap::PORT);
                    }
                }
                iface.poll(now(), &mut dev, &mut sockets);
                while let Some(p) = dev.tx.pop_front() {
                    if let (TunnResult::WriteToNetwork(out), Some(e)) = (tunn.encapsulate(&p, &mut net), endpoint) {
                        let _ = udp.send_to(out, e);
                    }
                }
                std::thread::sleep(Duration::from_millis(1));
            }
        });
        at
    }

    /// The VM's end as the real box's door behaves (bootstrap::serve_*): its
    /// kernel ESTABLISHES every connection at once (here: four listeners, the
    /// backlog), but the door SERVES them one at a time, in order, and each
    /// answer takes `serve_for` (a box that is busy). `x\n` → `ok x\n`.
    fn fake_door_vm(vm: [u8; 32], installer_public: [u8; 32], stop: Arc<AtomicBool>, serve_for: Duration) -> SocketAddrV4 {
        let udp = UdpSocket::bind("127.0.0.1:0").unwrap();
        udp.set_read_timeout(Some(Duration::from_millis(2))).unwrap();
        let at = match udp.local_addr().unwrap() {
            SocketAddr::V4(a) => a,
            _ => unreachable!(),
        };
        std::thread::spawn(move || {
            let mut tunn = Tunn::new(StaticSecret::from(vm), PublicKey::from(installer_public), None, None, 9, None);
            let epoch = Instant::now();
            let now = || smoltcp::time::Instant::from_millis(epoch.elapsed().as_millis() as i64);
            let mut dev = Q::default();
            let mut iface = Interface::new(Config::new(HardwareAddress::Ip), &mut dev, now());
            iface.update_ip_addrs(|a| {
                let _ = a.push(IpCidr::new(IpAddress::Ipv4(LINK_VM), LINK_PREFIX));
            });
            let mut sockets = SocketSet::new(Vec::new());
            let mut listeners = Vec::new();
            for _ in 0..4 {
                let mut s = tcp::Socket::new(tcp::SocketBuffer::new(vec![0u8; 8192]), tcp::SocketBuffer::new(vec![0u8; 8192]));
                s.listen(crate::bootstrap::PORT).unwrap();
                listeners.push(sockets.add(s));
            }
            // The door's queue (the order the kernel established them) and
            // the one connection it is serving: what it read, when it answers.
            let mut queue: VecDeque<smoltcp::iface::SocketHandle> = VecDeque::new();
            let mut got: Vec<u8> = Vec::new();
            let mut answer_at: Option<Instant> = None;
            let mut endpoint: Option<SocketAddr> = None;
            let (mut net, mut dgram) = (vec![0u8; 65536], [0u8; 2048]);
            while !stop.load(Ordering::SeqCst) {
                if let TunnResult::WriteToNetwork(p) = tunn.update_timers(&mut net) {
                    if let Some(e) = endpoint {
                        let _ = udp.send_to(p, e);
                    }
                }
                while let Ok((n, from)) = udp.recv_from(&mut dgram) {
                    let mut first = true;
                    loop {
                        let r = tunn.decapsulate(None, if first { &dgram[..n] } else { &[] }, &mut net);
                        first = false;
                        match r {
                            TunnResult::WriteToNetwork(p) => {
                                endpoint = Some(from);
                                let _ = udp.send_to(p, from);
                            }
                            TunnResult::WriteToTunnelV4(p, _) => {
                                endpoint = Some(from);
                                dev.rx.push_back(p.to_vec());
                                break;
                            }
                            TunnResult::Done => {
                                endpoint = Some(from);
                                break;
                            }
                            _ => break,
                        }
                    }
                }
                iface.poll(now(), &mut dev, &mut sockets);
                for h in &listeners {
                    let s = sockets.get_mut::<tcp::Socket>(*h);
                    if s.state() == tcp::State::Established && !queue.contains(h) {
                        queue.push_back(*h);
                    }
                }
                if let Some(&h) = queue.front() {
                    let s = sockets.get_mut::<tcp::Socket>(h);
                    if s.can_recv() {
                        let _ = s.recv(|b| {
                            got.extend_from_slice(b);
                            (b.len(), ())
                        });
                    }
                    if answer_at.is_none() && got.contains(&b'\n') {
                        answer_at = Some(Instant::now() + serve_for);
                    }
                    if answer_at.is_some_and(|t| Instant::now() >= t) && s.can_send() {
                        let i = got.iter().position(|b| *b == b'\n').unwrap();
                        let mut out = b"ok ".to_vec();
                        out.extend(got.drain(..=i));
                        let _ = s.send_slice(&out);
                        s.close();
                        answer_at = None;
                    }
                }
                // A connection that ended (answered, or its reader went away)
                // leaves the queue, and its listener listens again.
                for h in &listeners {
                    let s = sockets.get_mut::<tcp::Socket>(*h);
                    if !s.is_open() {
                        if queue.front() == Some(h) {
                            got.clear();
                            answer_at = None;
                        }
                        queue.retain(|q| q != h);
                        let _ = s.listen(crate::bootstrap::PORT);
                    }
                }
                iface.poll(now(), &mut dev, &mut sockets);
                while let Some(p) = dev.tx.pop_front() {
                    if let (TunnResult::WriteToNetwork(out), Some(e)) = (tunn.encapsulate(&p, &mut net), endpoint) {
                        let _ = udp.send_to(out, e);
                    }
                }
                std::thread::sleep(Duration::from_millis(1));
            }
        });
        at
    }

    /// ★ estate-box W6 (`W6-two-readers-one-link`, the release gate red 2 of 3
    /// on byte-identical binaries): "reader A #0: the request: nothing from
    /// the VM over the link" while reader B got all four answers, both rounds
    /// ending together. The box's door serves ONE connection at a time; A's
    /// connection was established by the box's kernel at once and then waited
    /// in its queue while the door served B — and A's read timeout was
    /// running all that time. A reader's timeout is for ITS exchange: waiting
    /// for the door while another reader of this process uses it is not.
    #[test]
    fn a_reader_queued_at_the_one_door_is_not_charged_for_another_readers_exchange() {
        let (vm, k) = keys();
        let installer_public = PublicKey::from(&StaticSecret::from(k.installer_private)).to_bytes();
        let stop = Arc::new(AtomicBool::new(false));
        // Each answer takes 3 s; each reader may wait 4.5 s for its own.
        let at = fake_door_vm(vm, installer_public, stop.clone(), Duration::from_secs(3));
        let timeout = Duration::from_millis(4500);
        let exchange = |who: &str, s: &mut Box<dyn Wire + '_>| -> Result<String, String> {
            s.write_all(format!("{who}\n").as_bytes()).map_err(|e| format!("{who}'s request: {e}"))?;
            s.flush().map_err(|e| format!("{who}'s request: {e}"))?;
            let mut got = Vec::new();
            s.read_to_end(&mut got).map_err(|e| format!("{who}'s answer: {e}"))?;
            Ok(String::from_utf8_lossy(&got).to_string())
        };
        // The link up first: its handshake is not what is measured.
        {
            let mut path = WgPath::new(at, k.clone()).unwrap();
            let mut s = path.connect(timeout).expect("the first connect");
            assert_eq!(exchange("warm", &mut s).as_deref(), Ok("ok warm\n"));
        }
        // B is connected first, so the door serves B first; A connects while B
        // is still open and asks right after B (if it may connect at all
        // before B is done, B asks once it has).
        let (b, a) = std::thread::scope(|sc| {
            let (b_up, b_is_up) = std::sync::mpsc::channel();
            let (a_up, a_is_up) = std::sync::mpsc::channel::<()>();
            let (kb, ex) = (&k, &exchange);
            let hb = sc.spawn(move || {
                let mut path = WgPath::new(at, kb.clone())?;
                let mut s = path.connect(timeout)?;
                let _ = b_up.send(());
                let _ = a_is_up.recv_timeout(Duration::from_millis(500));
                ex("b", &mut s)
            });
            let _ = b_is_up.recv();
            let a = (|| {
                let mut path = WgPath::new(at, k.clone())?;
                let mut s = path.connect(timeout)?;
                let _ = a_up.send(());
                std::thread::sleep(Duration::from_millis(100));
                exchange("a", &mut s)
            })();
            (hb.join().unwrap(), a)
        });
        stop.store(true, Ordering::SeqCst);
        assert_eq!(b.as_deref(), Ok("ok b\n"), "reader B, served first");
        assert_eq!(a.as_deref(), Ok("ok a\n"), "reader A lost its answer: its timeout ran while the door served reader B");
    }

    fn keys() -> ([u8; 32], LinkKeys) {
        let (vm, installer) = ([3u8; 32], [5u8; 32]);
        let vm_public = PublicKey::from(&StaticSecret::from(vm)).to_bytes();
        let installer_public = PublicKey::from(&StaticSecret::from(installer)).to_bytes();
        let _ = installer_public;
        (vm, LinkKeys { installer_private: installer, vm_public })
    }

    #[test]
    fn two_readers_of_one_vm_in_one_process_both_get_their_answers() {
        let (vm, k) = keys();
        let installer_public = PublicKey::from(&StaticSecret::from(k.installer_private)).to_bytes();
        let stop = Arc::new(AtomicBool::new(false));
        let at = fake_vm(vm, installer_public, stop.clone());
        let mut a = WgPath::new(at, k.clone()).unwrap();
        let mut sa = a.connect(Duration::from_secs(10)).expect("reader A connects");
        sa.write_all(b"reader-").unwrap();
        sa.flush().unwrap();
        let (got_a, got_b) = std::thread::scope(|sc| {
            // Reader B — the capture's next read, a check, a re-admission — on
            // its own thread, while A is open: it waits for A's turn at the
            // door (the box serves one connection at a time), then gets ITS answer.
            let kb = &k;
            let hb = sc.spawn(move || {
                let mut b = WgPath::new(at, kb.clone()).unwrap();
                let mut sb = b.connect(Duration::from_secs(10)).expect("reader B connects once A is done");
                sb.write_all(b"b\n").unwrap();
                sb.flush().unwrap();
                let mut got = Vec::new();
                let _ = sb.read_to_end(&mut got);
                String::from_utf8_lossy(&got).to_string()
            });
            std::thread::sleep(Duration::from_millis(300));
            assert!(!hb.is_finished(), "reader B waits for A's turn at the door");
            sa.write_all(b"a\n").unwrap();
            sa.flush().unwrap();
            let mut got = Vec::new();
            let r = sa.read_to_end(&mut got);
            drop(sa);
            ((String::from_utf8_lossy(&got).to_string(), r), hb.join().unwrap())
        });
        stop.store(true, Ordering::SeqCst);
        assert_eq!(got_a.0, "ok reader-a\n", "reader A lost its answer to reader B's session: {:?}", got_a.1);
        assert_eq!(got_b, "ok b\n", "reader B's answer");
    }
}