//! The per-network runtime.
//!
//! One of these runs for every locally active network. It owns that network's
//! sessions, its dial loop and its counters. Everything it touches carries an
//! explicit [`NetworkId`], so deactivating or breaking one network cannot
//! disturb another and cannot stop the agent.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};

use iroh::EndpointId;
use iroh::endpoint::{Connection, RecvStream, SendStream};
use tokio::sync::{broadcast, mpsc, oneshot};
use tokio::task::JoinHandle;

use crate::config::{DiscoveryPolicy, Limits, ReconnectPolicy};
use crate::dataplane::transport::health::{HealthLink, ProbeRoute, Settings as HealthSettings};
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};
use crate::net::{EndpointAdapter, PathAddr, snapshot_connection};
use crate::proto::handshake::{self, HandshakeOutcome, Role};
use crate::proto::message::{Announcement, ControlMessage, Envelope, encode, kind};
use crate::state::allocator::allocate;
use crate::state::{
    Ipv4Range, MAX_RETIREMENT_LEN, Merged, RecordBody, SignedRecord, SignedRetirement, StateSet,
};
use crate::storage::Storage;

use super::events::Event;
use super::session::{self, Session, SessionEvent};
use super::shutdown::Shutdown;
use super::status::{
    CandidateStatus, DataPath, HostnameConflict, MemberStatus, NetworkMetrics, NetworkState,
    NetworkStatus, PeerStatus,
};

/// An inbound connection that already passed the handshake.
#[derive(Debug)]
pub(crate) struct InboundSession {
    pub(crate) conn: Connection,
    pub(crate) send: SendStream,
    pub(crate) recv: RecvStream,
    pub(crate) outcome: HandshakeOutcome,
}

/// Commands accepted by a network runtime.
pub(crate) enum NetCommand {
    SaveVip {
        config: crate::vip::VipConfig,
        expected: Option<String>,
        delete: bool,
        reply: oneshot::Sender<Result<Vec<crate::vip::VipStatus>>>,
    },
    ForgetNode {
        peer: EndpointId,
        expected_version: u64,
        reply: oneshot::Sender<Result<super::ForgetNodeOutcome>>,
    },
    Inbound(Box<InboundSession>),
    Send {
        peer: EndpointId,
        message: ControlMessage,
        reply: oneshot::Sender<Result<()>>,
    },
    Broadcast {
        message: ControlMessage,
        reply: oneshot::Sender<usize>,
    },
    /// A peer opened a data plane link towards us.
    InboundLink(Box<InboundLink>),
    HealthProbe {
        peer: EndpointId,
        packet: bytes::Bytes,
    },
    Status {
        reply: oneshot::Sender<Box<NetworkStatus>>,
    },
    Recheck,
    SetBroadcast {
        enabled: bool,
        reply: oneshot::Sender<()>,
    },
    SetProtocols {
        interface: Option<Arc<crate::overlay::Interface>>,
        protocols: Vec<String>,
        reply: oneshot::Sender<()>,
    },
    /// Offer this agent as an exit node in this network, or stop.
    SetExitOffer {
        enabled: bool,
        reply: oneshot::Sender<()>,
    },
    /// Send this device's internet traffic through a member, or stop.
    SetExitVia {
        via: Option<EndpointId>,
        reply: oneshot::Sender<std::result::Result<(), String>>,
    },
    /// Resend this agent's announcement to every peer of this network.
    Reannounce,
    /// Answer to a different name from now on.
    SetHostname(String),
    /// Give up everything this agent claimed here, and tell whoever is
    /// listening.
    ///
    /// The reply is the number of peers the tombstone was queued to. They
    /// pass it on, so a member that is away learns of it from them rather
    /// than from an agent that has already gone.
    Release {
        reply: oneshot::Sender<usize>,
    },
    /// An IP plugin reported an error from one of its own tasks.
    PluginError {
        /// Plugin protocol id.
        protocol: String,
        /// Human readable reason, free of secrets.
        reason: String,
    },
    /// An IP plugin reported that data plane connectivity or a tunnel for a peer failed.
    ProtocolFailure {
        peer: EndpointId,
        protocol: String,
        reason: String,
    },
}

impl std::fmt::Debug for NetCommand {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            NetCommand::SaveVip { .. } => f.write_str("SaveVip"),
            NetCommand::ForgetNode { peer, .. } => write!(f, "ForgetNode({peer})"),
            NetCommand::Inbound(_) => f.write_str("Inbound"),
            NetCommand::InboundLink(link) => write!(f, "InboundLink({})", link.protocol),
            NetCommand::Send { peer, message, .. } => {
                write!(f, "Send({}, {})", peer.fmt_short(), kind(message))
            }
            NetCommand::Broadcast { message, .. } => write!(f, "Broadcast({})", kind(message)),
            NetCommand::Status { .. } => f.write_str("Status"),
            NetCommand::Recheck => f.write_str("Recheck"),
            NetCommand::SetBroadcast { enabled, .. } => write!(f, "SetBroadcast({enabled})"),
            NetCommand::SetProtocols { protocols, .. } => write!(f, "SetProtocols({protocols:?})"),
            NetCommand::SetExitOffer { enabled, .. } => write!(f, "SetExitOffer({enabled})"),
            NetCommand::SetExitVia { via, .. } => write!(f, "SetExitVia({})", via.is_some()),
            NetCommand::Reannounce => f.write_str("Reannounce"),
            NetCommand::SetHostname(_) => f.write_str("SetHostname"),
            NetCommand::Release { .. } => f.write_str("Release"),
            NetCommand::PluginError { protocol, .. } => write!(f, "PluginError({protocol})"),
            NetCommand::HealthProbe { peer, .. } => write!(f, "HealthProbe({})", peer.fmt_short()),
            NetCommand::ProtocolFailure { peer, protocol, .. } => {
                write!(f, "ProtocolFailure({}, {protocol})", peer.fmt_short())
            }
        }
    }
}

/// Handle to a running network runtime.
#[derive(Debug)]
pub(crate) struct NetworkHandle {
    pub(crate) keys: NetworkKeys,
    pub(crate) commands: mpsc::Sender<NetCommand>,
    shutdown: Shutdown,
    task: JoinHandle<()>,
    pub(crate) range_blocked: Arc<std::sync::atomic::AtomicBool>,
}

impl NetworkHandle {
    /// Stops the runtime and waits for its task to finish.
    pub(crate) async fn stop(self) {
        self.shutdown.trigger();
        crate::task::wind_down(self.task, crate::task::TASK_GRACE, "network runtime").await;
    }
}

/// Everything a network runtime needs to run.
pub(crate) struct RuntimeParams {
    pub(crate) vips: crate::vip::VipSet,
    pub(crate) retirements: Vec<SignedRetirement>,
    pub(crate) broadcast: bool,
    /// This agent offers itself as an exit node here.
    pub(crate) exit_offer: bool,
    /// The member this device's internet traffic goes through.
    pub(crate) exit_via: Option<EndpointId>,
    pub(crate) keys: NetworkKeys,
    pub(crate) adapter: EndpointAdapter,
    pub(crate) storage: Storage,
    pub(crate) events: broadcast::Sender<Event>,
    pub(crate) limits: Arc<Limits>,
    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>,
    pub(crate) available_plugins: Vec<SharedPlugin>,
    /// Who holds which overlay address, shared with every other network.
    pub(crate) routes: Arc<crate::overlay::RoutingTable>,
    /// The one interface, when the agent has one.
    pub(crate) interface: Option<Arc<crate::overlay::Interface>>,
    pub(crate) hostname: String,
    /// Signing key for this agent's own records.
    pub(crate) device_secret: iroh::SecretKey,
    /// Peers with no direct data path, for tests only.
    #[cfg(feature = "testing")]
    pub(crate) unreachable_data_peers: Arc<std::sync::Mutex<std::collections::HashSet<EndpointId>>>,
    /// The IPv4 overlay range this agent would use, if the network has not
    /// already settled on another one.
    pub(crate) ipv4_range: Option<Ipv4Range>,
    /// Coalesced notifications to the agent's lifecycle manager.
    pub(crate) range_changed: tokio::sync::watch::Sender<()>,
    /// How data plane links are opened. `None` disables the data plane.
    pub(crate) transport: Option<Arc<dyn PacketTransport>>,
}

/// Outcome of one attempt to open a data plane link.
struct LinkOutcome {
    generation: u64,
    peer: EndpointId,
    protocol: String,
    result: Result<SharedLink, String>,
}

/// Outcome of one outbound dial.
enum DialOutcome {
    Established(Box<InboundSession>),
    Failed {
        peer: EndpointId,
        reason: String,
        during_handshake: bool,
    },
}

/// Backoff bookkeeping for one candidate.
#[derive(Debug)]
struct DialState {
    consecutive_failures: u32,
    next_attempt: Instant,
    in_flight: bool,
    source: CandidateSource,
}

impl DialState {
    fn new(source: CandidateSource) -> Self {
        Self {
            consecutive_failures: 0,
            next_attempt: Instant::now(),
            in_flight: false,
            source,
        }
    }
}

/// Starts a network runtime.
pub(crate) fn spawn(params: RuntimeParams) -> NetworkHandle {
    let keys = params.keys.clone();
    let shutdown = Shutdown::new();
    let (commands_tx, commands_rx) = mpsc::channel(64);

    let runtime_shutdown = shutdown.clone();
    let range_blocked = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let runtime_blocked = Arc::clone(&range_blocked);
    let task = tokio::spawn(async move {
        let mut runtime = Runtime::new(params, runtime_shutdown, runtime_blocked);
        runtime.run(commands_rx).await;
    });

    NetworkHandle {
        keys,
        commands: commands_tx,
        shutdown,
        task,
        range_blocked,
    }
}

/// How long a peer's account of what it can reach is believed.
///
/// Soft state: it is repeated while it holds, and simply stops being
/// repeated when the peer goes. Nothing is revoked, so nothing can be
/// revoked wrongly — a relay that disappears stops being chosen by
/// itself, a few announcements later.
const REACH_EXPIRY: Duration = Duration::from_secs(90);

#[derive(Debug, Clone)]
struct ProtocolDialState {
    failures: u32,
    cooldown_until: std::time::Instant,
}

const DATA_LINK_BASE_BACKOFF: Duration = Duration::from_secs(3);
const DATA_LINK_MAX_BACKOFF: Duration = Duration::from_secs(60);

fn next_protocol_index(supported: &[String], last_failed: Option<&str>) -> usize {
    last_failed
        .and_then(|failed| supported.iter().position(|p| p == failed))
        .map_or(0, |i| (i + 1) % supported.len().max(1))
}

fn data_link_backoff(failures: u32) -> Duration {
    if failures == 0 {
        return Duration::ZERO;
    }
    let shift = (failures - 1).min(5);
    let secs = (DATA_LINK_BASE_BACKOFF.as_secs() << shift).min(DATA_LINK_MAX_BACKOFF.as_secs());
    Duration::from_secs(secs)
}

struct Runtime {
    discovery_worker: Option<DiscoveryWorker>,
    discovery_candidates: mpsc::Receiver<Candidate>,
    params: RuntimeParams,
    range_blocked: Arc<std::sync::atomic::AtomicBool>,
    /// Last range successfully published to the agent's routing table.
    published_range: Option<Ipv4Range>,
    network_id: NetworkId,
    local_id: EndpointId,
    shutdown: Shutdown,
    sessions: HashMap<EndpointId, Session>,
    dial_states: HashMap<EndpointId, DialState>,
    candidate_addrs: HashMap<EndpointId, iroh::EndpointAddr>,
    metrics: NetworkMetrics,
    session_events_tx: mpsc::Sender<SessionEvent>,
    session_events_rx: mpsc::Receiver<SessionEvent>,
    dial_results_tx: mpsc::Sender<DialOutcome>,
    dial_results_rx: mpsc::Receiver<DialOutcome>,
    /// Live data plane links, keyed by peer and plugin protocol.
    ///
    /// The direct ones. What a protocol holds is the hub's link for that
    /// peer, which outlives any particular path.
    links: HashMap<(EndpointId, String), Arc<HealthLink>>,
    pending_health: HashMap<(EndpointId, String), HealthSettings>,
    last_failed_protocol: HashMap<EndpointId, String>,
    /// Every data link of this network, and what is reached through what.
    hub: Arc<crate::dataplane::relay::RelayHub>,
    /// Which peers each peer says it has a live link with, and when it
    /// last said so.
    ///
    /// First-hand only: a peer speaks for itself and never for anybody
    /// else, so there is no hearsay to weigh and nothing to poison. Soft
    /// state — an entry that stops being refreshed expires, which is how a
    /// relay that has gone stops being chosen without anybody revoking it.
    reachable: HashMap<
        EndpointId,
        (
            HashSet<crate::proto::message::ReachableLink>,
            std::time::Instant,
        ),
    >,
    /// The protocols this agent last announced, in the order it announced
    /// them. The order is its preference, and the lower of two ids settles
    /// which protocol a pair runs.
    local_capabilities: Vec<PluginCapability>,
    /// What this agent last told its peers it could reach, so it is only
    /// said again when it changed.
    announced_reach: HashSet<crate::proto::message::ReachableLink>,
    /// Links currently being opened, so we do not start two.
    opening: HashSet<(EndpointId, String)>,
    link_tasks: HashMap<(EndpointId, String), tokio::task::AbortHandle>,
    link_attempts: HashMap<(EndpointId, String), u64>,
    next_link_attempt: u64,
    link_results_tx: mpsc::Sender<LinkOutcome>,
    link_results_rx: mpsc::Receiver<LinkOutcome>,
    /// Protocol dial failure count and backoff cooldown per peer and protocol.
    protocol_dial_states: HashMap<(EndpointId, String), ProtocolDialState>,
    /// Peers already told about a protocol version that cannot match, so it
    /// is said once rather than on every announcement.
    reported_mismatch: HashSet<(EndpointId, String)>,
    /// Set once this agent has given everything up here.
    ///
    /// A release is a statement on the way out, and anything that would
    /// publish a claim afterwards — the periodic check, a peer's records
    /// arriving — would silently take it back.
    released: bool,
    /// The name this agent last said was taken, so it is said once per
    /// change rather than on every round.
    reported_name_taken: Option<HostnameConflict>,
    /// The address last reported as absent from the interface, so it is said
    /// once rather than for ever.
    reported_missing: Option<std::net::Ipv4Addr>,
    /// Signed records, merged from every replica we have talked to.
    state: StateSet,
    /// Snapshots received while dispatching, handled on the next loop pass.
    pending_state: Vec<(EndpointId, Vec<SignedRecord>)>,
    pending_retirements: Vec<(EndpointId, Vec<SignedRetirement>)>,
    pending_vips: Vec<Vec<crate::vip::VipEntry>>,
    vip_cursor: HashMap<EndpointId, usize>,
    /// Next cleanup batch per live session; prevents queue pressure starving a suffix.
    retirement_cursor: HashMap<EndpointId, usize>,
    /// The highest version this agent has ever published for this network.
    own_version: u64,
    /// Whether the exit node in use was reachable when last looked at, so
    /// that losing it is said once and getting it back is too.
    exit_via_was_online: bool,
    /// What the last announcement said about being an exit node.
    announced_exit: bool,
    /// The last hostname each peer announced, kept across restarts so a
    /// member that is away can still be called by name.
    known_hostnames: HashMap<EndpointId, String>,
}

impl Runtime {
    fn new(
        params: RuntimeParams,
        shutdown: Shutdown,
        range_blocked: Arc<std::sync::atomic::AtomicBool>,
    ) -> Self {
        let network_id = params.keys.network_id();
        let local_id = params.adapter.endpoint_id();
        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 hub = crate::dataplane::relay::RelayHub::new(network_id, local_id);
        // The order the protocols were configured in is also how links are
        // preferred when a neighbour can be reached over more than one.
        hub.set_priority(
            params
                .plugins
                .iter()
                .map(|plugin| plugin.protocol_id().to_string())
                .collect(),
        );
        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,
            range_blocked,
            published_range: None,
            network_id,
            local_id,
            shutdown,
            sessions: HashMap::new(),
            dial_states: HashMap::new(),
            candidate_addrs: HashMap::new(),
            metrics: NetworkMetrics::default(),
            session_events_tx,
            session_events_rx,
            dial_results_tx,
            dial_results_rx,
            links: HashMap::new(),
            pending_health: HashMap::new(),
            last_failed_protocol: HashMap::new(),
            hub,
            reachable: HashMap::new(),
            local_capabilities: Vec::new(),
            announced_reach: HashSet::new(),
            opening: HashSet::new(),
            link_tasks: HashMap::new(),
            link_attempts: HashMap::new(),
            next_link_attempt: 0,
            link_results_tx,
            link_results_rx,
            protocol_dial_states: HashMap::new(),
            reported_mismatch: HashSet::new(),
            released: false,
            reported_missing: None,
            reported_name_taken: None,
            state: StateSet::new(),
            pending_state: Vec::new(),
            pending_retirements: Vec::new(),
            pending_vips: Vec::new(),
            vip_cursor: HashMap::new(),
            retirement_cursor: HashMap::new(),
            own_version: 0,
            exit_via_was_online: false,
            announced_exit: false,
            known_hostnames: HashMap::new(),
        }
    }

    fn emit(&self, event: Event) {
        // A broadcast with no subscribers is not an error.
        let _ = self.params.events.send(event);
    }

    async fn run(&mut self, mut commands: mpsc::Receiver<NetCommand>) {
        // Everything this agent knew before it restarted, including the
        // address it holds.
        self.load_state().await;

        let mut ticker = tokio::time::interval(self.params.discovery_interval);
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

        loop {
            tokio::select! {
                biased;
                _ = self.shutdown.wait() => break,
                _ = self.hub.changed() => {
                    self.ensure_links();
                    self.update_paths();
                    self.announce_reach(false);
                }
                command = commands.recv() => match command {
                    Some(command) => self.handle_command(command).await,
                    None => break,
                },
                event = self.session_events_rx.recv() => {
                    if let Some(event) = event {
                        self.handle_session_event(event).await;
                        self.drain_pending_state().await;
                    }
                }
                result = self.dial_results_rx.recv() => {
                    if let Some(result) = result {
                        self.handle_dial_result(result).await;
                    }
                }
                result = self.link_results_rx.recv() => {
                    if let Some(result) = result {
                        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();
                    self.update_paths();
                    self.announce_reach(false);
                }
            }
        }

        self.teardown().await;
    }

    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();
        self.hub.close();
        for link in self.links.values() {
            link.close();
        }
        self.links.clear();
        for (_, task) in self.link_tasks.drain() {
            task.abort();
        }
        self.opening.clear();
        self.pending_health.clear();
        self.last_failed_protocol.clear();
        self.protocol_dial_states.clear();
        let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
        for peer in peers {
            if let Some(session) = self.sessions.remove(&peer) {
                session.stop().await;
            }
        }
        self.emit(Event::NetworkDeactivated {
            network: self.network_id,
        });
    }

    // ---------------------------------------------------------------- commands

    async fn handle_command(&mut self, command: NetCommand) {
        match command {
            NetCommand::SaveVip {
                config,
                expected,
                delete,
                reply,
            } => {
                let result = self.save_vip(config, expected, delete).await;
                let _ = reply.send(result);
            }
            NetCommand::ForgetNode {
                peer,
                expected_version,
                reply,
            } => {
                let result = self.forget_node(peer, expected_version).await;
                let _ = reply.send(result);
            }
            NetCommand::Inbound(inbound) => {
                self.install_session(*inbound).await;
            }
            NetCommand::InboundLink(inbound) => self.install_link(*inbound),
            NetCommand::Send {
                peer,
                message,
                reply,
            } => {
                if matches!(&message, ControlMessage::Transport { .. })
                    && let Some(interface) = self.params.interface.clone()
                {
                    // Locally bound pair sockets must be protected before their
                    // private offer/accept invites the first inbound datagram.
                    interface.sync_exit_rules().await;
                }
                let _ = reply.send(self.send_to(peer, message));
            }
            NetCommand::Broadcast { message, reply } => {
                let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
                let mut delivered = 0;
                for peer in peers {
                    if self.send_to(peer, message.clone()).is_ok() {
                        delivered += 1;
                    }
                }
                let _ = reply.send(delivered);
            }
            NetCommand::Status { reply } => {
                let _ = reply.send(Box::new(self.status()));
            }
            NetCommand::SetProtocols {
                protocols,
                interface,
                reply,
            } => {
                self.params.interface = interface;
                let current: Vec<_> = self
                    .params
                    .plugins
                    .iter()
                    .map(|p| p.protocol_id().to_owned())
                    .collect();
                if current != protocols {
                    let peers: Vec<_> = self.sessions.keys().copied().collect();
                    for peer in &peers {
                        self.reset_peer_data(*peer, "local protocol selection changed");
                    }
                    for plugin in &self.params.plugins {
                        if !protocols.iter().any(|name| name == plugin.protocol_id()) {
                            plugin.on_network_deactivated(self.network_id);
                        }
                    }
                    self.params.plugins = protocols
                        .iter()
                        .filter_map(|name| {
                            self.params
                                .available_plugins
                                .iter()
                                .find(|p| p.protocol_id() == name)
                                .cloned()
                        })
                        .collect();
                    for plugin in &self.params.plugins {
                        if !current.iter().any(|name| name == plugin.protocol_id()) {
                            plugin.on_network_activated(self.network_id);
                        }
                    }
                    self.hub.set_priority(protocols);
                    self.publish_allocations().await;
                    for peer in peers {
                        let caps = self
                            .sessions
                            .get(&peer)
                            .map(|s| s.capabilities.clone())
                            .unwrap_or_default();
                        self.dispatch_capabilities(peer, &caps);
                    }
                    self.reannounce();
                    self.ensure_links();
                    self.update_paths();
                    self.announce_reach(true);
                }
                let _ = reply.send(());
            }
            NetCommand::SetBroadcast { enabled, reply } => {
                self.params.broadcast = enabled;
                self.update_broadcast();
                if let Some(interface) = self.params.interface.clone() {
                    interface.sync_broadcast_rules().await;
                }
                self.reannounce();
                let _ = reply.send(());
            }
            NetCommand::SetExitOffer { enabled, reply } => {
                self.params.exit_offer = enabled;
                self.apply_exit().await;
                self.reannounce();
                let _ = reply.send(());
            }
            NetCommand::SetExitVia { via, reply } => {
                let outcome = match via {
                    Some(peer) if !self.exit_node_online(peer) => Err(
                        "that member is not connected or does not offer to be an exit node"
                            .to_string(),
                    ),
                    _ => Ok(()),
                };
                if outcome.is_ok() {
                    self.params.exit_via = via;
                    self.exit_via_was_online = via.is_some();
                    self.apply_exit().await;
                }
                let _ = reply.send(outcome);
            }
            NetCommand::Recheck => self.discovery_round().await,
            NetCommand::Reannounce => {
                // A newly enabled plugin may prepare its key asynchronously.
                // Replay allocations and peer identities once it is ready.
                self.publish_allocations().await;
                let peers: Vec<_> = self
                    .sessions
                    .iter()
                    .map(|(peer, s)| (*peer, s.capabilities.clone()))
                    .collect();
                for (peer, caps) in peers {
                    self.dispatch_capabilities(peer, &caps);
                }
                self.reannounce();
                self.ensure_links();
                self.update_paths();
            }
            NetCommand::Release { reply } => {
                // A tombstone, so the address and the name this agent held
                // are freed for somebody else instead of staying reserved
                // to a member that has gone.
                self.publish_record(RecordBody::Release).await;
                self.released = true;
                let _ = reply.send(self.sessions.len());
            }
            NetCommand::SetHostname(hostname) => {
                if self.params.hostname != hostname {
                    self.params.hostname = hostname;
                    // Publishing the claim is the revocation: one record per
                    // author, so the new version replaces the old name
                    // rather than sitting beside it.
                    self.ensure_own_claim().await;
                    self.broadcast_state();
                }
                // Told to peers regardless, so a session that missed the
                // earlier announcement is not left with a stale name.
                self.reannounce();
            }
            NetCommand::PluginError { protocol, reason } => {
                // Counted here so that the per-network metric and the event
                // always agree, wherever the error came from.
                self.metrics.plugin_errors += 1;
                self.emit(Event::PluginError {
                    network: self.network_id,
                    protocol,
                    reason,
                });
            }
            NetCommand::HealthProbe { peer, packet } => {
                for ((p, _), link) in &self.links {
                    if *p == peer {
                        link.receive_probe(packet.clone());
                    }
                }
            }
            NetCommand::ProtocolFailure {
                peer,
                protocol,
                reason,
            } => {
                if let Some(link) = self.links.remove(&(peer, protocol.clone())) {
                    link.close();
                    self.hub.clear_direct(peer, &protocol);
                }
                self.metrics.data_link_failures += 1;
                self.record_protocol_failure(peer, &protocol);
                self.emit(Event::DataLinkDown {
                    network: self.network_id,
                    peer,
                    protocol,
                    reason,
                });
                self.ensure_links();
                self.update_paths();
                self.announce_reach(false);
            }
        }
    }

    /// Rebuilds this agent's announcement and pushes it to every session.
    ///
    /// Used when a plugin's capability changed, so peers do not have to wait
    /// for a reconnect to learn about it.
    fn reannounce(&mut self) {
        let announcement = ControlMessage::Announce(self.local_announcement());
        let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
        for peer in peers {
            if let Err(err) = self.send_to(peer, announcement.clone()) {
                tracing::debug!(%err, "could not queue re-announcement");
            }
        }
        // What was announced decides which protocol each pair runs.
        self.update_paths();
    }

    /// Queues a message without blocking the runtime loop.
    ///
    /// The envelope is encoded here so that the exact number of control bytes
    /// handed to the transport is known and can be reported honestly.
    ///
    /// A full queue is backpressure: the send fails rather than stalling every
    /// other peer in this network.
    fn send_to(&mut self, peer: EndpointId, message: ControlMessage) -> Result<()> {
        let network = self.network_id;
        let envelope = Envelope {
            network_id: *network.as_bytes(),
            message,
        };
        let encoded = encode(&envelope)?;
        let bytes = encoded.len() as u64;

        let session = self.sessions.get_mut(&peer).ok_or(Error::NoSuchPeer {
            network,
            peer: peer.fmt_short().to_string(),
        })?;

        match session.outbound.try_send(zeroize::Zeroizing::new(encoded)) {
            Ok(()) => {
                session.messages_sent += 1;
                session.bytes_sent += bytes;
                self.metrics.control_messages_sent += 1;
                self.metrics.control_bytes_sent += bytes;
                Ok(())
            }
            Err(mpsc::error::TrySendError::Full(_)) => Err(Error::Storage(format!(
                "outbound queue for peer {} is full",
                peer.fmt_short()
            ))),
            Err(mpsc::error::TrySendError::Closed(_)) => Err(Error::NoSuchPeer {
                network,
                peer: peer.fmt_short().to_string(),
            }),
        }
    }

    // ------------------------------------------------------------- discovery

    async fn discovery_round(&mut self) {
        if self.shutdown.is_triggered() {
            return;
        }
        // The fallback range becomes available with the passage of time
        // alone, so something has to look again; this runs on a timer and
        // the check is a comparison when nothing has changed.
        self.ensure_own_claim().await;
        self.report_hostname_taken();
        self.broadcast_vips();
        self.broadcast_retirements();
        // The same timer repairs the relay picture: reachability is soft
        // state, repeated while it holds and forgotten when it stops.
        self.update_paths();
        self.announce_reach(true);

        let mut candidates: Vec<Candidate> = Vec::new();

        // 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
        // still dialable when the endpoint's own discovery can resolve
        // one. This is what makes a network a mesh rather than a star
        // around whoever was named on a command line.
        for record in self.state.active_records() {
            if let Ok(author) = record.author_id()
                && author != self.local_id
            {
                candidates.push(Candidate::new(
                    iroh::EndpointAddr::new(author),
                    CandidateSource::Member,
                ));
            }
        }

        // A stale or missing cache only changes which candidates we try first.
        // It never bypasses authentication.
        for hint in self.params.storage.hints_for_network(self.network_id).await {
            let Ok(endpoint_id) = EndpointId::from_bytes(&hint.endpoint_id) else {
                continue;
            };
            if self.state.is_retired(&endpoint_id) {
                continue;
            }
            let Some(addr) = decode_hint(endpoint_id, &hint.addr) else {
                continue;
            };
            candidates.push(Candidate::new(addr, CandidateSource::Cache));
        }

        for candidate in candidates {
            self.add_candidate(candidate);
        }

        self.start_dials();
        // Last, so it carries what this round learned.
        self.introduce_peers();
        self.refresh_underlay().await;
        if let Some(interface) = self.params.interface.clone() {
            interface.sync_exit_rules().await;
        }
    }

    /// Hands the interface the underlay destinations this network depends
    /// on: the verified direct IPv4 paths of every control connection, and
    /// the relays in use. Only hosts that cannot exempt the agent from an
    /// exit node's default route read it.
    async fn refresh_underlay(&self) {
        let Some(interface) = self.params.interface.clone() else {
            return;
        };
        let mut addresses = Vec::new();
        let mut hosts = Vec::new();
        for session in self.sessions.values() {
            for path in snapshot_connection(&session.conn).paths {
                match path.remote {
                    PathAddr::Ip(std::net::SocketAddr::V4(socket)) => addresses.push(*socket.ip()),
                    PathAddr::Relay(url) => hosts.extend(relay_host(&url)),
                    _ => {}
                }
            }
        }
        for url in self.params.adapter.snapshot().relay_urls {
            hosts.extend(relay_host(&url));
        }
        interface
            .set_underlay(self.network_id, addresses, hosts)
            .await;
    }

    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
            .dial_states
            .values()
            .filter(|state| state.in_flight)
            .count();
        let mut budget = self
            .params
            .limits
            .max_concurrent_dials
            .saturating_sub(in_flight);
        if budget == 0 || self.sessions.len() >= self.params.limits.max_sessions_per_network {
            return;
        }

        let ready: Vec<EndpointId> = self
            .dial_states
            .iter()
            .filter(|(peer, state)| {
                !state.in_flight && state.next_attempt <= now && !self.sessions.contains_key(*peer)
            })
            .map(|(peer, _)| *peer)
            .collect();

        for peer in ready {
            if budget == 0 {
                break;
            }
            let Some(addr) = self.candidate_addrs.get(&peer).cloned() else {
                continue;
            };
            if let Some(state) = self.dial_states.get_mut(&peer) {
                state.in_flight = true;
            }
            budget -= 1;
            self.metrics.dial_attempts += 1;

            let adapter = self.params.adapter.clone();
            let keys = self.params.keys.clone();
            let limits = Arc::clone(&self.params.limits);
            let results = self.dial_results_tx.clone();
            let local_id = self.local_id;
            let shutdown = self.shutdown.clone();

            tokio::spawn(async move {
                let outcome = tokio::select! {
                    biased;
                    _ = shutdown.wait() => DialOutcome::Failed {
                        peer,
                        reason: "network deactivated".into(),
                        during_handshake: false,
                    },
                    outcome = dial(adapter, addr, keys, limits, local_id, peer) => outcome,
                };
                let _ = results.send(outcome).await;
            });
        }
    }

    async fn handle_dial_result(&mut self, result: DialOutcome) {
        match result {
            DialOutcome::Established(inbound) => {
                let peer = inbound.outcome.peer;
                if let Some(state) = self.dial_states.get_mut(&peer) {
                    state.in_flight = false;
                    state.consecutive_failures = 0;
                    state.next_attempt = Instant::now();
                }
                self.install_session(*inbound).await;
            }
            DialOutcome::Failed {
                peer,
                reason,
                during_handshake,
            } => {
                self.metrics.dial_failures += 1;
                if during_handshake {
                    self.metrics.handshake_failures += 1;
                }
                let give_up = {
                    let policy = &self.params.reconnect;
                    let state = self
                        .dial_states
                        .entry(peer)
                        .or_insert_with(|| DialState::new(CandidateSource::Discovery));
                    state.in_flight = false;
                    state.consecutive_failures = state.consecutive_failures.saturating_add(1);
                    let delay = policy.delay_for(state.consecutive_failures);
                    state.next_attempt = Instant::now() + delay;
                    policy
                        .max_consecutive_failures
                        .is_some_and(|max| state.consecutive_failures >= max)
                };
                if give_up {
                    // Keep the backoff entry but push it far out; discovery
                    // seeing the peer again resets it.
                    if let Some(state) = self.dial_states.get_mut(&peer) {
                        state.next_attempt = Instant::now() + self.params.reconnect.max_delay;
                    }
                }
                self.emit(Event::DialFailed {
                    network: self.network_id,
                    peer,
                    reason,
                });
            }
        }
    }

    // ----------------------------------------------------------- agreed state

    /// Reads back what this agent already knew before it restarted.
    ///
    /// Records are verified again on load: the database is not a trust
    /// boundary, because a restored backup or a copied file could hold
    /// anything.
    async fn load_state(&mut self) {
        for marker in std::mem::take(&mut self.params.retirements) {
            // Storage verified these proofs and activation enforced capacity.
            if let Err(err) = self.state.merge_retirement(
                self.network_id,
                marker,
                self.params.limits.max_retirements,
            ) {
                self.emit(Event::PluginError {
                    network: self.network_id,
                    protocol: "state".into(),
                    reason: err.to_string(),
                });
                self.shutdown.trigger();
                return;
            }
        }
        let stored = self
            .params
            .storage
            .signed_records(self.network_id)
            .await
            .unwrap_or_default();
        let (_, errors) = self.state.merge_all(self.network_id, stored);
        for err in errors {
            tracing::warn!(%err, "discarding an unusable stored record");
        }
        self.own_version = self
            .params
            .storage
            .own_record_version(self.network_id, self.local_id)
            .await
            .unwrap_or(0)
            .max(self.state.authored_version(&self.local_id))
            .max(self.params.vips.authored_version(self.local_id));
        for (id, hostname) in self
            .params
            .storage
            .peer_hostnames(self.network_id)
            .await
            .unwrap_or_default()
        {
            if let Ok(id) = EndpointId::from_bytes(&id) {
                self.known_hostnames.insert(id, hostname);
            }
        }

        self.ensure_own_claim().await;
        self.publish_allocations().await;
    }

    /// The range this network uses: whatever it has already settled on, else
    /// what this agent was configured with.
    ///
    /// Adopting the agreed one is what lets a participant join without being
    /// told the range out of band.
    fn effective_range(&self) -> Option<Ipv4Range> {
        if let Some(agreed) = self.state.agreed_range() {
            return Some(agreed);
        }
        self.params.ipv4_range
    }

    /// A late signed range can conflict even though activation was safe.
    /// Fail closed before touching claims, plugins or the host interface.
    fn suspend_if_overlapping(&self) -> bool {
        if let Some(range) = self.effective_range()
            && self
                .params
                .routes
                .would_overlap(self.network_id, range)
                .is_some()
        {
            self.range_blocked
                .store(true, std::sync::atomic::Ordering::Release);
            self.params.routes.remove_network(self.network_id);
            self.shutdown.trigger();
            self.params.range_changed.send_replace(());
            return true;
        }
        false
    }

    /// Makes sure this agent holds an address, claiming one if it does not.
    ///
    /// Called after anything that could change the picture: startup, and
    /// every time another replica's records arrive.
    async fn ensure_own_claim(&mut self) {
        // On the way out. Claiming again here is how a goodbye becomes a
        // hello nobody asked for.
        if self.released || self.suspend_if_overlapping() {
            return;
        }
        let wanted_hostname = {
            let hostname = crate::state::sanitise_hostname(&self.params.hostname);
            (!hostname.is_empty()).then_some(hostname)
        };
        let range = self.effective_range();

        let wanted_address = match range {
            None => None,
            Some(range) => {
                let holders = self.state.address_holders();
                // An address we still hold is kept; this is what makes a
                // returning participant get its old address back.
                match self.state.address_of(&self.local_id) {
                    Some(mine) if range.contains(mine) => Some(mine),
                    _ => {
                        let mut taken: std::collections::HashSet<std::net::Ipv4Addr> = holders
                            .iter()
                            .filter(|(_, holder)| **holder != self.local_id)
                            .map(|(address, _)| *address)
                            .collect();
                        for vip in self.params.vips.views(
                            Some(range),
                            taken.clone(),
                            &std::collections::BTreeSet::new(),
                            &self.params.limits,
                        ) {
                            if let Some(address) = vip.address {
                                taken.insert(address);
                            }
                        }
                        for vip in self.params.routes.vips.statuses(self.network_id) {
                            if let Some(address) = vip.address {
                                taken.insert(address);
                            }
                        }
                        match allocate(
                            self.network_id,
                            self.local_id,
                            range,
                            &taken,
                            self.state
                                .get(&self.local_id)
                                .and_then(|record| record.body.claimed_address()),
                        ) {
                            Ok(address) => Some(address),
                            Err(err) => {
                                self.metrics.plugin_errors += 1;
                                self.emit(Event::PluginError {
                                    network: self.network_id,
                                    protocol: "overlay".into(),
                                    reason: err.to_string(),
                                });
                                None
                            }
                        }
                    }
                }
            }
        };

        // A name keeps the time it was first claimed for as long as this
        // member keeps the name, so a later claimant cannot overtake it by
        // the passage of time, and a new or changed name starts the clock
        // afresh.
        let hostname_since = wanted_hostname.as_ref().map(|name| {
            let kept = self
                .state
                .get(&self.local_id)
                .filter(|_| !self.state.is_retired(&self.local_id))
                .filter(|record| record.body.hostname() == Some(name.as_str()))
                .and_then(|record| record.body.hostname_since());
            kept.unwrap_or_else(unix_millis)
        });
        let body = RecordBody::Claim {
            address: wanted_address,
            // The range only travels with an address, so a member of an
            // IPv6-only network does not assert one.
            range: wanted_address.and(range),
            hostname: wanted_hostname,
            hostname_since,
        };

        // Nothing to say is not the same as saying nothing changed: a member
        // that claims neither an address nor a name has no reason to occupy a
        // record at all.
        if matches!(
            &body,
            RecordBody::Claim {
                address: None,
                hostname: None,
                ..
            }
        ) {
            return;
        }

        // Republishing an unchanged claim would bump the version for no
        // reason and make every replica store it again.
        if self
            .state
            .get(&self.local_id)
            .is_some_and(|record| record.body == body && !self.state.is_retired(&self.local_id))
        {
            return;
        }

        self.publish_record(body).await;
    }

    async fn forget_node(
        &mut self,
        peer: EndpointId,
        expected_version: u64,
    ) -> Result<super::ForgetNodeOutcome> {
        if peer == self.local_id {
            return Err(Error::ForgetNode(
                "use network leave for this device".into(),
            ));
        }
        if self.is_online(peer) {
            return Err(Error::ForgetNode(
                "the node is online; only offline history can be forgotten".into(),
            ));
        }
        if self.params.exit_via == Some(peer) {
            return Err(Error::ForgetNode(
                "stop using this exit node before forgetting it".into(),
            ));
        }
        let target = self
            .state
            .get(&peer)
            .filter(|_| !self.state.is_retired(&peer))
            .filter(|record| matches!(record.body, RecordBody::Claim { .. }))
            .cloned()
            .ok_or_else(|| Error::ForgetNode("no current signed claim for this node".into()))?;
        if target.version != expected_version {
            return Err(Error::ForgetNode(
                "the node's claim changed; refresh and select it again".into(),
            ));
        }
        let version = self
            .own_version
            .max(self.state.authored_version(&self.local_id))
            .checked_add(1)
            .filter(|v| *v <= i64::MAX as u64)
            .ok_or_else(|| Error::ForgetNode("author counter exhausted".into()))?;
        let marker =
            SignedRetirement::sign(&self.params.device_secret, self.network_id, version, target);
        let mut candidate = self.state.clone();
        candidate
            .merge_retirement(
                self.network_id,
                marker.clone(),
                self.params.limits.max_retirements,
            )
            .map_err(|err| Error::ForgetNode(err.to_string()))?;
        // Signed cleanup and our author counter commit together before visibility.
        self.params
            .storage
            .publish_own_retirement(marker.clone())
            .await?;
        self.own_version = version;
        self.state = candidate;
        self.known_hostnames.remove(&peer);
        self.dial_states.remove(&peer);
        self.candidate_addrs.remove(&peer);
        self.publish_allocations().await;
        self.report_hostname_taken();
        let message = ControlMessage::Retirements {
            markers: vec![marker],
        };
        let peers: Vec<_> = self.sessions.keys().copied().collect();
        let peers_told = peers
            .into_iter()
            .filter(|peer| self.send_to(*peer, message.clone()).is_ok())
            .count();
        Ok(super::ForgetNodeOutcome {
            target: peer,
            record_version: expected_version,
            peers_told,
        })
    }

    fn broadcast_retirements(&mut self) -> usize {
        let markers = self.state.retirements();
        if markers.is_empty() {
            return 0;
        }
        // Each proof has an encoded bound; reserve envelope/list overhead and
        // chunk every marker, so a low-id prefix cannot starve later cleanup.
        let batch = (self.params.limits.max_frame_len.saturating_sub(128) / MAX_RETIREMENT_LEN)
            .max(1)
            .min(self.params.limits.max_state_records.max(1));
        let mut delivered = HashSet::new();
        let peers: Vec<_> = self.sessions.keys().copied().collect();
        let chunks: Vec<_> = markers.chunks(batch).collect();
        for peer in peers {
            let start = self.retirement_cursor.get(&peer).copied().unwrap_or(0) % chunks.len();
            for offset in 0..chunks.len() {
                let index = (start + offset) % chunks.len();
                let message = ControlMessage::Retirements {
                    markers: chunks[index].to_vec(),
                };
                if self.send_to(peer, message).is_err() {
                    self.retirement_cursor.insert(peer, index);
                    break;
                }
                delivered.insert(peer);
                self.retirement_cursor
                    .insert(peer, (index + 1) % chunks.len());
            }
        }
        delivered.len()
    }

    async fn receive_retirements(&mut self, peer: EndpointId, markers: Vec<SignedRetirement>) {
        let mut changed = false;
        for marker in markers {
            let mut candidate = self.state.clone();
            match candidate.merge_retirement(
                self.network_id,
                marker.clone(),
                self.params.limits.max_retirements,
            ) {
                Ok(false) => continue,
                Ok(true) => {}
                Err(err) => {
                    self.metrics.protocol_violations += 1;
                    self.emit(Event::ProtocolViolation {
                        network: Some(self.network_id),
                        peer: Some(peer),
                        reason: format!("unusable cleanup proof: {err}"),
                    });
                    continue;
                }
            }
            let target = marker.target.author_id().ok();
            if let Err(err) = self.params.storage.put_retirement(marker).await {
                self.emit(Event::PluginError {
                    network: self.network_id,
                    protocol: "state".into(),
                    reason: format!("cannot persist cleanup proof: {err}"),
                });
                continue;
            }
            self.state = candidate;
            if let Some(target) = target
                && self.state.is_retired(&target)
                && !self.sessions.contains_key(&target)
            {
                self.known_hostnames.remove(&target);
                self.dial_states.remove(&target);
                self.candidate_addrs.remove(&target);
            }
            changed = true;
        }
        if changed {
            // A live target answers cleanup with a fresh claim; its identity
            // is still authenticated, and old hostname priority is not reused.
            self.ensure_own_claim().await;
            self.publish_allocations().await;
            self.broadcast_retirements();
        }
    }

    /// Signs, stores and announces one of this agent's own records.
    ///
    /// Stored before it is announced, in one transaction with the version
    /// counter, so a crash can never let us reuse a version we already put on
    /// the wire.
    async fn publish_record(&mut self, body: RecordBody) {
        let observed = self.state.authored_version(&self.local_id).max(
            self.state
                .retirement_of(&self.local_id)
                .map_or(0, |m| m.target.version),
        );
        let Some(version) = self
            .own_version
            .max(observed)
            .checked_add(1)
            .filter(|v| *v <= i64::MAX as u64)
        else {
            self.emit(Event::PluginError {
                network: self.network_id,
                protocol: "state".into(),
                reason: "author counter exhausted".into(),
            });
            return;
        };
        let record = SignedRecord::sign(&self.params.device_secret, self.network_id, version, body);

        if let Err(err) = self.params.storage.publish_own_record(record.clone()).await {
            self.emit(Event::PluginError {
                network: self.network_id,
                protocol: "overlay".into(),
                reason: format!("cannot store our own record: {err}"),
            });
            return;
        }
        self.own_version = version;

        match self.state.merge(self.network_id, record) {
            Ok(_) => {}
            Err(err) => {
                tracing::error!(%err, "our own record did not verify");
                return;
            }
        }
        self.broadcast_state();
    }

    fn update_vips(&self) {
        let taken = self.state.address_holders().into_keys().collect();
        let names = self
            .state
            .active_records()
            .into_iter()
            .filter_map(|r| {
                r.author_id()
                    .ok()
                    .and_then(|p| self.state.hostname_of(&p).map(str::to_owned))
            })
            .collect();
        let views =
            self.params
                .vips
                .views(self.effective_range(), taken, &names, &self.params.limits);
        let peers = self
            .state
            .address_holders()
            .into_iter()
            .map(|(a, p)| (p, a, self.hub.reachable(p)))
            .collect();
        self.params.routes.vips.configure(
            self.network_id,
            self.local_id,
            self.state.address_of(&self.local_id),
            views,
            peers,
            &self.params.limits,
        );
    }
    async fn save_vip(
        &mut self,
        config: crate::vip::VipConfig,
        expected: Option<String>,
        delete: bool,
    ) -> Result<Vec<crate::vip::VipStatus>> {
        if self.released || self.shutdown.is_triggered() {
            return Err(Error::Stopped);
        }
        let version = self
            .own_version
            .checked_add(1)
            .filter(|n| *n <= i64::MAX as u64)
            .ok_or_else(|| crate::vip::invalid("author counter exhausted"))?;
        let entries = if delete {
            self.params.vips.delete(
                self.network_id,
                &self.params.device_secret,
                version,
                &config.id,
                expected.as_deref().unwrap_or(""),
            )?
        } else {
            self.params.vips.edit(
                self.network_id,
                &self.params.device_secret,
                version,
                &config,
                expected.as_deref(),
                &self.params.limits,
            )?
        };
        let mut candidate = self.params.vips.clone();
        candidate.merge(self.network_id, entries, &self.params.limits)?;
        if !delete {
            let duplicate = candidate
                .configs()
                .iter()
                .any(|v| v.id != config.id && v.name == config.name)
                || self.state.active_records().iter().any(|r| {
                    r.author_id().ok().and_then(|p| self.state.hostname_of(&p))
                        == Some(config.name.as_str())
                });
            if duplicate {
                return Err(crate::vip::invalid(
                    "VIP name is already used by a node or another VIP",
                ));
            }
        }
        let snapshot = candidate.entries(&config.id);
        self.validate_vip_frame(&snapshot)?;
        self.params
            .storage
            .put_vip_fields(
                self.network_id,
                snapshot.clone(),
                Some((*self.local_id.as_bytes(), version)),
            )
            .await?;
        self.own_version = version;
        self.params.vips = candidate;
        self.update_vips();
        let peers: Vec<_> = self.sessions.keys().copied().collect();
        for peer in peers {
            let _ = self.send_to(
                peer,
                ControlMessage::Vips {
                    entries: snapshot.clone(),
                },
            );
        }
        Ok(self.params.routes.vips.statuses(self.network_id))
    }
    async fn receive_vips(&mut self, entries: Vec<crate::vip::VipEntry>) -> Result<()> {
        let Some(id) = entries.first().map(|e| e.vip.clone()) else {
            return Ok(());
        };
        if entries.iter().any(|e| e.vip != id) {
            return Err(crate::vip::invalid("VIP snapshot mixes service identities"));
        }
        let own_seen = entries
            .iter()
            .filter(|e| e.author == *self.local_id.as_bytes())
            .map(|e| e.version)
            .max()
            .unwrap_or(0);
        let mut candidate = self.params.vips.clone();
        let changed = candidate.merge(self.network_id, entries, &self.params.limits)?;
        if !changed && own_seen <= self.own_version {
            return Ok(());
        }
        let counter =
            (own_seen > self.own_version).then_some((*self.local_id.as_bytes(), own_seen));
        let snapshot = candidate.entries(&id);
        self.validate_vip_frame(&snapshot)?;
        self.params
            .storage
            .put_vip_fields(self.network_id, snapshot, counter)
            .await?;
        self.own_version = self.own_version.max(own_seen);
        self.params.vips = candidate;
        self.update_vips();
        self.broadcast_vips();
        Ok(())
    }
    fn validate_vip_frame(&self, entries: &[crate::vip::VipEntry]) -> Result<()> {
        let frame = Envelope {
            network_id: *self.network_id.as_bytes(),
            message: ControlMessage::Vips {
                entries: entries.to_vec(),
            },
        };
        let bytes = encode(&frame).map_err(|e| crate::vip::invalid(e.to_string()))?;
        if bytes.len() > self.params.limits.max_frame_len {
            return Err(crate::vip::invalid(
                "VIP snapshot exceeds the control frame limit",
            ));
        }
        Ok(())
    }
    fn broadcast_vips(&mut self) {
        let ids = self.params.vips.ids();
        if ids.is_empty() {
            return;
        }
        let peers: Vec<_> = self.sessions.keys().copied().collect();
        for peer in peers {
            let start = self.vip_cursor.get(&peer).copied().unwrap_or(0) % ids.len();
            for step in 0..ids.len() {
                let index = (start + step) % ids.len();
                let entries = self.params.vips.entries(&ids[index]);
                if self
                    .send_to(peer, ControlMessage::Vips { entries })
                    .is_err()
                {
                    self.vip_cursor.insert(peer, index);
                    break;
                }
                self.vip_cursor.insert(peer, (index + 1) % ids.len());
            }
        }
    }

    /// Handles snapshots collected while dispatching messages.
    async fn drain_pending_state(&mut self) {
        for entries in std::mem::take(&mut self.pending_vips) {
            if let Err(err) = self.receive_vips(entries).await {
                self.metrics.protocol_violations += 1;
                tracing::warn!(%err,"rejecting VIP snapshot");
            }
        }
        for (peer, markers) in std::mem::take(&mut self.pending_retirements) {
            self.receive_retirements(peer, markers).await;
        }
        for (peer, records) in std::mem::take(&mut self.pending_state) {
            self.receive_state(peer, records).await;
        }
        self.report_hostname_taken();
    }

    /// Sends everything we know to every peer.
    fn broadcast_state(&mut self) {
        self.broadcast_vips();
        self.broadcast_retirements();
        let mut records = self.state.records();
        records.truncate(self.params.limits.max_state_records);
        if records.is_empty() {
            return;
        }
        let message = ControlMessage::State { records };
        let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
        for peer in peers {
            if let Err(err) = self.send_to(peer, message.clone()) {
                tracing::debug!(%err, "could not queue a state snapshot");
            }
        }
    }

    /// Merges a snapshot from a peer.
    async fn receive_state(&mut self, peer: EndpointId, records: Vec<SignedRecord>) {
        let before = self.state.records();
        let (outcomes, errors) = self.state.merge_all(self.network_id, records);

        for err in errors {
            self.metrics.protocol_violations += 1;
            self.emit(Event::ProtocolViolation {
                network: Some(self.network_id),
                peer: Some(peer),
                reason: format!("unusable signed record: {err}"),
            });
        }
        for outcome in &outcomes {
            if *outcome == Merged::Conflicted {
                self.emit(Event::PluginError {
                    network: self.network_id,
                    protocol: "overlay".into(),
                    reason: "two different records from one author at the same version; \
                             a device key appears to be in use in two places"
                        .into(),
                });
            }
        }

        let changed = outcomes.iter().any(|outcome| {
            matches!(
                outcome,
                Merged::Added | Merged::Updated | Merged::Conflicted
            )
        });
        if !changed {
            return;
        }

        for record in self.state.records() {
            if record.author == *self.local_id.as_bytes() {
                continue;
            }
            if let Err(err) = self.params.storage.put_signed_record(record).await {
                tracing::debug!(%err, "cannot persist a record");
            }
        }

        // Somebody may have taken the address we were using.
        self.ensure_own_claim().await;
        self.publish_allocations().await;
        if self.state.records() != before {
            self.broadcast_state();
        }
    }

    fn broadcast_policy(&self) -> crate::overlay::broadcast::BroadcastPolicy {
        crate::overlay::broadcast::BroadcastPolicy {
            enabled: self.params.broadcast,
            peers: self
                .sessions
                .values()
                .filter(|session| session.broadcast)
                .map(|session| session.peer)
                .collect(),
        }
    }

    fn exit_policy(&self) -> crate::overlay::ExitPolicy {
        crate::overlay::ExitPolicy {
            offer: self.params.exit_offer,
            via: self.params.exit_via,
        }
    }

    /// Whether this agent offers to be an exit node and can really be one.
    fn exit_ready(&self) -> bool {
        self.params.exit_offer
            && self
                .params
                .interface
                .as_ref()
                .is_some_and(|interface| interface.exit_offer_ready(self.network_id))
    }

    /// Whether a member is connected and says it is an exit node.
    fn exit_node_online(&self, peer: EndpointId) -> bool {
        self.sessions
            .get(&peer)
            .is_some_and(|session| session.exit_node)
    }

    /// Puts the exit settings into the routing table and the host rules.
    ///
    /// The routes and the rules stay as they are when the exit node in use
    /// goes away: dropping them would send the traffic out the old way, in
    /// the clear, without anyone having asked.
    async fn apply_exit(&mut self) {
        self.params
            .routes
            .set_exit(self.network_id, self.exit_policy());
        if let Some(interface) = self.params.interface.clone() {
            interface.sync_exit_rules().await;
        }
        // Said only once it is true: the rules may not have taken, or may
        // only now be possible.
        if self.exit_ready() != self.announced_exit {
            self.reannounce();
        }
    }

    /// Says so, once, when the exit node in use is lost or comes back.
    fn watch_exit_via(&mut self) {
        let Some(via) = self.params.exit_via else {
            return;
        };
        let online = self.exit_node_online(via);
        if online == self.exit_via_was_online {
            return;
        }
        self.exit_via_was_online = online;
        if online {
            tracing::info!(
                network = %self.network_id,
                exit_node = %via.fmt_short(),
                "the exit node is back"
            );
        } else {
            tracing::warn!(
                network = %self.network_id,
                exit_node = %via.fmt_short(),
                "the exit node is gone or no longer offers to be one; internet traffic stays \
                 routed through it and is dropped until it returns or `tsng exit-node off`"
            );
        }
    }

    fn update_broadcast(&self) {
        self.params
            .routes
            .set_broadcast(self.network_id, self.broadcast_policy());
    }

    /// Tells the plugins who holds which overlay address.
    async fn publish_allocations(&mut self) {
        if self.shutdown.is_triggered() || self.suspend_if_overlapping() {
            return;
        }
        let Some(range) = self.effective_range() else {
            return;
        };
        let mut allocations: Vec<(EndpointId, std::net::Ipv4Addr)> = self
            .state
            .address_holders()
            .into_iter()
            .map(|(address, holder)| (holder, address))
            .collect();
        allocations.sort_by_key(|(holder, _)| *holder.as_bytes());

        // And the system level's own view, which is what decides whose
        // packet is whose. The local address is kept out of the peer list:
        // a packet for ourselves does not go over a tunnel.
        let local = self.state.address_of(&self.local_id);
        let routes = crate::overlay::NetworkRoutes {
            exit: self.exit_policy(),
            broadcast: self.broadcast_policy(),
            range: Some(range),
            local,
            peers: allocations
                .iter()
                .filter(|(holder, _)| *holder != self.local_id)
                .map(|(holder, address)| (*address, *holder))
                .collect(),
        };
        if self
            .params
            .routes
            .set_network(self.network_id, routes)
            .is_err()
        {
            self.range_blocked
                .store(true, std::sync::atomic::Ordering::Release);
            self.params.routes.remove_network(self.network_id);
            self.shutdown.trigger();
            self.params.range_changed.send_replace(());
            return;
        }
        self.update_vips();
        for plugin in &self.params.plugins {
            plugin.on_address_allocation(self.network_id, range, &allocations);
        }
        if self.published_range != Some(range) {
            self.published_range = Some(range);
            self.params.range_changed.send_replace(());
        }

        let Some(interface) = self.params.interface.clone() else {
            return;
        };
        if let Err(err) = interface.sync_addresses().await {
            self.emit(Event::PluginError {
                network: self.network_id,
                protocol: "overlay".into(),
                reason: err.to_string(),
            });
            return;
        }

        // The agent assigns this address itself, so finding it absent means
        // the assignment did not take — something outside removed it, or the
        // provisioner reported a success it did not achieve. Left unsaid it
        // looks like a broken network: packets would leave with the wrong
        // source and every peer would drop them. Checked rather than
        // assumed, because the assumption is exactly the kind that has been
        // wrong here before. Said once per address, not every round.
        // An in-memory interface has no addresses to be missing from: with
        // `--no-tun` this is the arrangement asked for, not a fault.
        let is_missing = if let Some(address) = local {
            if interface.on_host() {
                let mut local_found =
                    crate::overlay::address_is_local(std::net::IpAddr::V4(address));
                // On Windows, duplicate address detection (DAD) and netsh/IP helper updates
                // may take a short moment before a newly assigned address is ready for binding.
                if !local_found && cfg!(target_os = "windows") {
                    for _ in 0..10 {
                        tokio::time::sleep(std::time::Duration::from_millis(150)).await;
                        if crate::overlay::address_is_local(std::net::IpAddr::V4(address)) {
                            local_found = true;
                            break;
                        }
                    }
                }
                !local_found
            } else {
                false
            }
        } else {
            false
        };

        let missing = if is_missing {
            if let Some(address) = local {
                let already = self.reported_missing == Some(address);
                self.reported_missing = Some(address);
                (!already).then_some(address)
            } else {
                None
            }
        } else {
            self.reported_missing = None;
            None
        };
        if let Some(address) = missing {
            self.emit(Event::PluginError {
                network: self.network_id,
                protocol: "overlay".into(),
                reason: format!(
                    "this agent was allocated {address}/{} but the address is not on any \
                     interface, so the overlay cannot work: packets would leave with the \
                     wrong source and every peer would drop them. It should have been \
                     assigned to `{}` automatically; check whether something else removed \
                     it.",
                    range.prefix_len,
                    interface.name()
                ),
            });
        }

        // The broadcast route names this address as its source, so it goes in
        // only once the address is really on the interface.
        if !is_missing {
            interface.sync_broadcast_rules().await;
            self.apply_exit().await;
        }
    }

    // ------------------------------------------------------------ data plane

    /// The protocols this agent speaks, with the version it speaks them at.
    fn served_protocols(&self) -> Vec<(String, u16)> {
        self.params
            .plugins
            .iter()
            .map(|plugin| (plugin.protocol_id().to_string(), plugin.protocol_version()))
            .collect()
    }

    fn record_protocol_failure(&mut self, peer: EndpointId, protocol: &str) {
        self.last_failed_protocol.insert(peer, protocol.to_string());
        let key = (peer, protocol.to_string());
        let entry = self
            .protocol_dial_states
            .entry(key)
            .or_insert(ProtocolDialState {
                failures: 0,
                cooldown_until: std::time::Instant::now(),
            });
        entry.failures = entry.failures.saturating_add(1);
        let backoff = data_link_backoff(entry.failures);
        entry.cooldown_until = std::time::Instant::now() + backoff;
    }

    fn record_protocol_success(&mut self, peer: EndpointId, protocol: &str) {
        self.protocol_dial_states
            .remove(&(peer, protocol.to_string()));
    }

    /// Opens whatever data plane links are missing, and forgets dead ones.
    ///
    /// Only one side dials, chosen by a rule both sides compute the same way,
    /// so two agents never open two links for the same thing.
    fn ensure_links(&mut self) {
        let Some(transport) = self.params.transport.clone() else {
            return;
        };
        let served = self.served_protocols();
        if served.is_empty() {
            return;
        }

        let mut dead: Vec<(EndpointId, String)> = Vec::new();
        for (key, link) in &self.links {
            if link.is_closed() || self.no_direct_data(key.0) {
                dead.push(key.clone());
            }
        }
        for (peer, protocol) in dead {
            let reason = self
                .links
                .remove(&(peer, protocol.clone()))
                .map(|link| {
                    let reason = link
                        .failure_reason()
                        .unwrap_or_else(|| "link closed".into());
                    link.close();
                    reason
                })
                .unwrap_or_else(|| "link closed".into());
            // Only the direct path is gone. Whether the peer is still
            // reachable through somebody is decided below; the protocol's
            // link stays either way.
            self.hub.clear_direct(peer, &protocol);
            self.record_protocol_failure(peer, &protocol);
            self.emit(Event::DataLinkDown {
                network: self.network_id,
                peer,
                protocol,
                reason,
            });
        }

        // Choose the highest priority mutually supported protocol for each peer.
        // If a direct data link is already open or opening, do not dial another.
        // If a protocol recently failed, fall back to other mutually supported
        // protocols according to backoff cooldown.
        let now = std::time::Instant::now();
        let mut wanted: Vec<(EndpointId, String)> = Vec::new();
        for session in self.sessions.values() {
            let peer = session.peer;
            if self.links.keys().any(|(p, _)| *p == peer)
                || self.opening.iter().any(|(p, _)| *p == peer)
            {
                continue;
            }
            if self.no_direct_data(peer) {
                continue;
            }
            // The smaller endpoint id dials; the other side accepts. Both
            // compute this identically, so exactly one link is created.
            if self.local_id.as_bytes() >= peer.as_bytes() {
                continue;
            }

            let supported: Vec<String> = served
                .iter()
                .filter(|(name, ours)| {
                    session
                        .capabilities
                        .iter()
                        .any(|c| c.enabled && c.protocol == *name && c.version == *ours)
                })
                .map(|(name, _)| name.clone())
                .collect();

            if supported.is_empty() {
                continue;
            }

            let start = next_protocol_index(
                &supported,
                self.last_failed_protocol.get(&peer).map(String::as_str),
            );
            let chosen = supported
                .iter()
                .cycle()
                .skip(start)
                .take(supported.len())
                .find(
                    |proto| match self.protocol_dial_states.get(&(peer, (*proto).clone())) {
                        Some(state) => state.cooldown_until <= now,
                        None => true,
                    },
                );

            if let Some(proto) = chosen {
                wanted.push((peer, proto.clone()));
            }
        }

        for (peer, protocol) in wanted {
            let key = (peer, protocol.clone());
            if self.links.contains_key(&key) || self.opening.contains(&key) {
                continue;
            }
            self.opening.insert(key.clone());

            let results = self.link_results_tx.clone();
            let transport = Arc::clone(&transport);
            let network = self.network_id;
            let shutdown = self.shutdown.clone();
            self.next_link_attempt = self.next_link_attempt.wrapping_add(1);
            let generation = self.next_link_attempt;
            self.link_attempts.insert(key.clone(), generation);
            let task = tokio::spawn(async move {
                let result = tokio::select! {
                    biased;
                    _ = shutdown.wait() => Err("network deactivated".to_string()),
                    result = transport.open(network, peer, &protocol) => {
                        result.map_err(|err| err.to_string())
                    }
                };
                let _ = results
                    .send(LinkOutcome {
                        generation,
                        peer,
                        protocol,
                        result,
                    })
                    .await;
            });
            self.link_tasks.insert(key, task.abort_handle());
        }
    }

    /// Takes in what another member knows about the rest.
    ///
    /// Returns how many of them were new here, which is the only reason to
    /// start dialling again straight away.
    fn learn_peers(&mut self, hints: &[crate::proto::message::PeerHint]) -> usize {
        let mut learned = 0;
        for hint in hints {
            let Ok(peer) = EndpointId::from_bytes(&hint.endpoint) else {
                continue;
            };
            if peer == self.local_id || self.sessions.contains_key(&peer) {
                continue;
            }
            let mut addr = iroh::EndpointAddr::new(peer);
            for text in &hint.addrs {
                // The same decoder as the cache's hints: one spelling, one
                // place it is read, nothing to drift.
                if let Some(decoded) = decode_hint(peer, text) {
                    merge_addr(&mut addr, &decoded);
                }
            }

            let fresh = !self.candidate_addrs.contains_key(&peer);
            self.candidate_addrs
                .entry(peer)
                .and_modify(|existing| merge_addr(existing, &addr))
                .or_insert(addr);
            self.dial_states
                .entry(peer)
                .or_insert_with(|| DialState::new(CandidateSource::Introduced));
            if fresh {
                learned += 1;
            }
        }
        learned
    }

    /// Tells the peers it is talking to about the peers it knows.
    ///
    /// A device told about one member ends up talking to all of them: a
    /// network is a mesh, and which member happened to be named on a
    /// command line should not decide who anybody talks to. Sent on every
    /// round, because it is cheap, bounded and self-repairing.
    fn introduce_peers(&mut self) {
        let mut hints: Vec<crate::proto::message::PeerHint> = Vec::new();

        // What this agent can see for itself: where each peer it is
        // talking to is right now. An agent that only ever accepts has no
        // candidates of its own, and it is exactly the one everybody else
        // was pointed at — so without this, introductions would come from
        // nobody.
        let mut observed: HashMap<EndpointId, iroh::EndpointAddr> = HashMap::new();
        for session in self.sessions.values() {
            let snapshot = crate::net::snapshot_connection(&session.conn);
            let mut addr = iroh::EndpointAddr::new(session.peer);
            for path in &snapshot.paths {
                match &path.remote {
                    crate::net::PathAddr::Ip(socket) => addr = addr.with_ip_addr(*socket),
                    crate::net::PathAddr::Relay(url) => {
                        if let Ok(url) = url.parse() {
                            addr = addr.with_relay_url(url);
                        }
                    }
                    crate::net::PathAddr::Other(_) => {}
                }
            }
            observed.insert(session.peer, addr);
        }
        for (peer, addr) in &self.candidate_addrs {
            observed
                .entry(*peer)
                .and_modify(|existing| merge_addr(existing, addr))
                .or_insert_with(|| addr.clone());
        }

        for (peer, addr) in &observed {
            if *peer == self.local_id {
                continue;
            }
            // An id with no address at all is still worth passing on: the
            // other side's own discovery may be able to resolve it.
            hints.push(crate::proto::message::PeerHint {
                endpoint: *peer.as_bytes(),
                addrs: hint_addrs(addr),
            });
        }
        // This agent, too: it is the one member the others cannot be told
        // about by anybody else.
        hints.push(crate::proto::message::PeerHint {
            endpoint: *self.local_id.as_bytes(),
            addrs: hint_addrs(&self.params.adapter.addr()),
        });
        hints.truncate(self.params.limits.max_state_records);
        if hints.is_empty() {
            return;
        }

        let message = ControlMessage::Peers { peers: hints };
        let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
        for peer in peers {
            if let Err(err) = self.send_to(peer, message.clone()) {
                tracing::debug!(%err, "could not queue an introduction");
            }
        }
    }

    /// Publishes only locally observed, protocol-specific transport edges.
    fn announce_reach(&mut self, force: bool) {
        let live: HashSet<_> = self
            .links
            .iter()
            .filter(|(_, link)| !link.is_closed())
            .map(
                |((peer, protocol), _)| crate::proto::message::ReachableLink {
                    peer: *peer.as_bytes(),
                    protocol: protocol.clone(),
                },
            )
            .collect();
        if !force && live == self.announced_reach {
            return;
        }
        self.announced_reach = live.clone();
        let mut links: Vec<_> = live.into_iter().collect();
        links.sort_by(|a, b| (&a.protocol, a.peer).cmp(&(&b.protocol, b.peer)));
        links.truncate(self.params.limits.max_state_records);
        let message = ControlMessage::Reachable { links };
        let peers: Vec<_> = self.sessions.keys().copied().collect();
        for peer in peers {
            if let Err(err) = self.send_to(peer, message.clone()) {
                tracing::debug!(%err, "could not queue a reachability announcement");
            }
        }
    }

    /// The protocol this agent and a member run their tunnel in, if they have
    /// one in common.
    ///
    /// Both ends must name the same one without asking each other, and what
    /// each side was configured to prefer can differ. So the member with the
    /// lower id decides, by the order it announces its protocols in, and the
    /// other side reads that order from the announcement it already has.
    fn tunnel_protocol(&self, session: &Session) -> Option<String> {
        let usable = |ours: &[PluginCapability], theirs: &[PluginCapability], name: &str| {
            ours.iter().any(|capability| {
                capability.enabled
                    && capability.protocol == name
                    && theirs.iter().any(|other| {
                        other.enabled
                            && other.protocol == name
                            && other.version == capability.version
                    })
            })
        };
        let (decides, other) = if self.local_id.as_bytes() < session.peer.as_bytes() {
            (&self.local_capabilities, &session.capabilities)
        } else {
            (&session.capabilities, &self.local_capabilities)
        };
        decides
            .iter()
            .find(|capability| usable(decides, other, &capability.protocol))
            .map(|capability| capability.protocol.clone())
    }

    /// Builds the graph off the packet path. Every remote row comes from that
    /// member's authenticated control session, and a row says only who that
    /// member has a link to, never what kind of link: a member in the middle
    /// needs no protocol in common with either end. Stale rows never enter the
    /// graph. The hub replaces an immutable table only on changes.
    fn update_paths(&mut self) {
        self.reachable.retain(|peer, (_, heard)| {
            heard.elapsed() < REACH_EXPIRY && self.sessions.contains_key(peer)
        });
        let mut members: HashSet<_> = self.sessions.keys().map(|peer| *peer.as_bytes()).collect();
        members.insert(*self.local_id.as_bytes());
        let graph = self
            .reachable
            .iter()
            .filter(|(peer, _)| members.contains(peer.as_bytes()))
            .map(|(peer, (links, _))| {
                let mut neighbors: Vec<_> = links
                    .iter()
                    .filter(|link| members.contains(&link.peer) && link.peer != *peer.as_bytes())
                    .map(|link| link.peer)
                    .collect();
                neighbors.sort_unstable();
                neighbors.dedup();
                (*peer.as_bytes(), neighbors)
            })
            .collect();
        self.hub.set_topology(graph, members);

        let wanted: Vec<(EndpointId, String)> = self
            .sessions
            .values()
            .filter_map(|session| Some((session.peer, self.tunnel_protocol(session)?)))
            .collect();
        for (peer, protocol) in wanted {
            if self.hub.has_link(peer, &protocol) || !self.hub.reachable(peer) {
                continue;
            }
            let Some(plugin) = self
                .params
                .plugins
                .iter()
                .find(|plugin| plugin.protocol_id() == protocol)
                .cloned()
            else {
                continue;
            };
            let link = self.hub.link(peer, &protocol);
            let relayed = self.hub.route(peer).is_some_and(|route| route.hops > 1);
            let path = link.path_description();
            let max_datagram = link.max_datagram_size();
            plugin.on_peer_link(self.network_id, peer, link);
            // A direct link has already been reported as one.
            if relayed {
                self.emit(Event::DataLinkUp {
                    network: self.network_id,
                    peer,
                    protocol,
                    path,
                    max_datagram,
                });
            }
        }
        self.update_vips();
    }

    fn handle_link_result(&mut self, outcome: LinkOutcome) {
        let key = (outcome.peer, outcome.protocol.clone());
        if self.link_attempts.get(&key) != Some(&outcome.generation) {
            if let Ok(link) = outcome.result {
                link.close();
            }
            return;
        }
        self.link_attempts.remove(&key);
        self.opening.remove(&key);
        self.link_tasks.remove(&key);
        match outcome.result {
            Ok(link) => {
                self.record_protocol_success(outcome.peer, &outcome.protocol);
                self.adopt_link(outcome.peer, outcome.protocol, link);
                // A new link is both a path for us and one we can offer
                // others, so it changes what we route and what we say.
                self.update_paths();
                self.announce_reach(false);
            }
            Err(reason) => {
                // A data plane that cannot be set up is reported, never fatal.
                self.metrics.data_link_failures += 1;
                self.record_protocol_failure(outcome.peer, &outcome.protocol);
                self.emit(Event::DataLinkDown {
                    network: self.network_id,
                    peer: outcome.peer,
                    protocol: outcome.protocol,
                    reason,
                });
                // Fall back to other mutually supported protocols immediately.
                self.ensure_links();
            }
        }
    }

    /// Whether this agent refuses a direct data path to a peer.
    ///
    /// Always false outside a test build: see
    /// [`crate::config::AgentConfig::with_unreachable_data_peers`].
    fn no_direct_data(&self, peer: EndpointId) -> bool {
        #[cfg(feature = "testing")]
        {
            match self.params.unreachable_data_peers.lock() {
                Ok(guard) => guard.contains(&peer),
                Err(poisoned) => poisoned.into_inner().contains(&peer),
            }
        }
        #[cfg(not(feature = "testing"))]
        {
            let _ = peer;
            false
        }
    }

    fn install_link(&mut self, inbound: InboundLink) {
        if self.no_direct_data(inbound.peer)
            || inbound.link.is_closed()
            || !self.sessions.contains_key(&inbound.peer)
        {
            inbound.link.close();
            return;
        }
        self.record_protocol_success(inbound.peer, &inbound.protocol);
        self.adopt_link(inbound.peer, inbound.protocol, inbound.link);
        self.update_paths();
        self.announce_reach(false);
    }

    /// Hands a link to the plugin that owns its protocol.
    fn adopt_link(&mut self, peer: EndpointId, protocol: String, link: SharedLink) {
        if link.is_closed()
            || link.network() != self.network_id
            || link.peer() != peer
            || !self.sessions.contains_key(&peer)
        {
            link.close();
            return;
        }
        if !self.params.plugins.iter().any(|plugin| {
            plugin.protocol_id() == protocol
                && self.sessions.get(&peer).is_some_and(|s| {
                    s.capabilities.iter().any(|c| {
                        c.enabled
                            && c.protocol == protocol
                            && c.version == plugin.protocol_version()
                    })
                })
        }) {
            link.close();
            return;
        }

        let key = (peer, protocol.clone());
        let settings = if self.local_id.as_bytes() < peer.as_bytes() {
            let settings = HealthSettings::fresh();
            if let Ok(bytes) = postcard::to_stdvec(&(protocol.clone(), settings.clone())) {
                let _ = self.send_to(
                    peer,
                    ControlMessage::Transport {
                        protocol: "link-health".into(),
                        payload: crate::proto::message::TransportPayload::new(bytes),
                    },
                );
            }
            Some(settings)
        } else {
            self.pending_health.remove(&key)
        };
        let route = self
            .sessions
            .get(&peer)
            .and_then(|s| self.tunnel_protocol(s))
            .and_then(|name| self.params.plugins.iter().find(|p| p.protocol_id() == name))
            .map(|plugin| ProbeRoute {
                plugin: plugin.clone(),
                routes: self.params.routes.clone(),
                network: self.network_id,
                peer,
                max_probe_size: self.params.interface.as_ref().map_or(540, |interface| {
                    (interface.mtu() as usize).saturating_sub(36)
                }),
            });
        let link = HealthLink::new(link, self.local_id, settings, route);
        let path = link.path_description();
        let max_datagram = link.max_datagram_size();
        if let Some(old) = self.links.insert(key, Arc::clone(&link)) {
            old.close();
        }
        // What this is, is a way to reach the peer. Which protocol's
        // tunnel runs over it is decided in `update_paths`, and that tunnel
        // is handed the hub's link, not this one: the same object for as
        // long as the peer is a peer, whatever happens to the path under
        // it. A tunnel is not torn down because a direct link came or went.
        self.hub.set_direct(peer, &protocol, link);
        self.metrics.data_links_established += 1;
        self.emit(Event::DataLinkUp {
            network: self.network_id,
            peer,
            protocol,
            path,
            max_datagram,
        });
    }

    /// Drops every link to a peer.
    fn drop_links_for(&mut self, peer: EndpointId) {
        self.retirement_cursor.remove(&peer);
        self.vip_cursor.remove(&peer);
        self.reset_peer_data(peer, "peer session ended");
        self.hub.remove_peer(peer);
        self.reachable.remove(&peer);
    }

    /// Retire only the data plane; authenticated membership and topology remain.
    fn reset_peer_data(&mut self, peer: EndpointId, reason: &str) {
        self.link_attempts.retain(|(id, _), _| *id != peer);
        self.opening.retain(|(id, _)| *id != peer);
        self.link_tasks.retain(|(id, _), task| {
            if *id == peer {
                task.abort();
                false
            } else {
                true
            }
        });
        self.hub.clear_peer_links(peer);
        let keys: Vec<_> = self
            .links
            .keys()
            .filter(|(id, _)| *id == peer)
            .cloned()
            .collect();
        for key in keys {
            if let Some(link) = self.links.remove(&key) {
                link.close();
            }
            self.emit(Event::DataLinkDown {
                network: self.network_id,
                peer,
                protocol: key.1,
                reason: reason.into(),
            });
        }
        self.protocol_dial_states.retain(|(id, _), _| *id != peer);
        self.last_failed_protocol.remove(&peer);
        self.pending_health.retain(|(id, _), _| *id != peer);
        for plugin in &self.params.plugins {
            plugin.on_peer_gone(self.network_id, peer);
        }
    }

    // --------------------------------------------------------------- sessions

    async fn install_session(&mut self, inbound: InboundSession) {
        let InboundSession {
            conn,
            send,
            recv,
            outcome,
        } = inbound;
        let peer = outcome.peer;

        if self.sessions.len() >= self.params.limits.max_sessions_per_network
            && !self.sessions.contains_key(&peer)
        {
            conn.close(1u32.into(), b"session limit reached");
            self.emit(Event::ProtocolViolation {
                network: Some(self.network_id),
                peer: Some(peer),
                reason: "session limit for this network reached".into(),
            });
            return;
        }

        // Two agents may dial each other at the same time. Both sides apply the
        // same deterministic rule, so they converge on the same session.
        if let Some(existing) = self.sessions.get(&peer) {
            let existing_initiator = initiator_of(existing.role, self.local_id, peer);
            let new_initiator = initiator_of(outcome.role, self.local_id, peer);
            if existing_initiator.as_bytes() <= new_initiator.as_bytes() {
                conn.close(0u32.into(), b"duplicate session");
                return;
            }
            if let Some(old) = self.sessions.remove(&peer) {
                old.abort();
                old.conn
                    .close(0u32.into(), b"replaced by preferred session");
            }
        }

        self.record_hints(&conn).await;

        let session = session::spawn(
            self.network_id,
            peer,
            outcome.role,
            conn.clone(),
            send,
            recv,
            Arc::clone(&self.params.limits),
            self.session_events_tx.clone(),
            self.shutdown.clone(),
        );

        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
        // capabilities without another round of discovery.
        let announcement = ControlMessage::Announce(self.local_announcement());
        if let Err(err) = self.send_to(peer, announcement) {
            tracing::debug!(%err, "could not queue initial announcement");
        }

        self.broadcast_state();

        self.emit(Event::PeerConnected {
            network: self.network_id,
            peer,
            role: outcome.role,
            transport: snapshot.transport,
            rtt: snapshot.rtt,
        });
    }

    fn local_announcement(&mut self) -> Announcement {
        let mut capabilities = Vec::new();
        let mut errors = Vec::new();
        for plugin in &self.params.plugins {
            match plugin.local_capability(self.network_id) {
                Ok(Some(capability)) => capabilities.push(capability),
                Ok(None) => {}
                Err(err) => errors.push((plugin.protocol_id().to_string(), err.to_string())),
            }
        }
        for (protocol, reason) in errors {
            self.metrics.plugin_errors += 1;
            self.emit(Event::PluginError {
                network: self.network_id,
                protocol,
                reason,
            });
        }
        capabilities.truncate(self.params.limits.max_capabilities);
        self.local_capabilities = capabilities
            .iter()
            .map(|capability| PluginCapability {
                data: Vec::new(),
                ..capability.clone()
            })
            .collect();
        let exit_node = self.exit_ready();
        self.announced_exit = exit_node;
        Announcement {
            broadcast: self.params.broadcast,
            exit_node,
            hostname: self.params.hostname.clone(),
            capabilities,
        }
    }

    async fn record_hints(&self, conn: &Connection) {
        let snapshot = snapshot_connection(conn);
        let peer_bytes = *snapshot.remote_id.as_bytes();
        for path in &snapshot.paths {
            if let Some(encoded) = encode_hint(&path.remote) {
                self.params
                    .storage
                    .record_hint(
                        self.network_id,
                        peer_bytes,
                        encoded,
                        self.params.limits.max_hints_per_peer,
                    )
                    .await;
            }
        }
    }

    async fn handle_session_event(&mut self, event: SessionEvent) {
        match event {
            SessionEvent::Message {
                session_id,
                peer,
                message,
                bytes,
            } => {
                let current = self.sessions.get(&peer).map(|session| session.id);
                if current != Some(session_id) {
                    return;
                }
                self.metrics.control_messages_received += 1;
                self.metrics.control_bytes_received += bytes as u64;
                if let Some(session) = self.sessions.get_mut(&peer) {
                    session.messages_received += 1;
                    session.bytes_received += bytes as u64;
                }
                self.dispatch_message(peer, message);
            }
            SessionEvent::Violation {
                session_id,
                peer,
                error,
            } => {
                let current = self.sessions.get(&peer).map(|session| session.id);
                if current != Some(session_id) {
                    return;
                }
                self.metrics.protocol_violations += 1;
                self.emit(Event::ProtocolViolation {
                    network: Some(self.network_id),
                    peer: Some(peer),
                    reason: error.to_string(),
                });
            }
            SessionEvent::Closed {
                session_id,
                peer,
                reason,
            } => {
                let current = self.sessions.get(&peer).map(|session| session.id);
                if current != Some(session_id) {
                    return;
                }
                if let Some(session) = self.sessions.remove(&peer) {
                    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);
                self.update_broadcast();
                self.watch_exit_via();
                self.update_paths();
                self.announce_reach(false);
                // Retry promptly, then back off if it keeps failing.
                let policy = &self.params.reconnect;
                let state = self
                    .dial_states
                    .entry(peer)
                    .or_insert_with(|| DialState::new(CandidateSource::Discovery));
                state.in_flight = false;
                state.next_attempt = Instant::now() + policy.initial_delay;
                self.emit(Event::PeerDisconnected {
                    network: self.network_id,
                    peer,
                    reason,
                });
            }
        }
    }

    /// Keeps the name a peer announced, on disk too when it changed.
    ///
    /// Written off the loop: a name that is not stored this time is stored
    /// the next time the peer announces it.
    fn remember_hostname(&mut self, peer: EndpointId, hostname: &str) {
        if hostname.is_empty()
            || self
                .known_hostnames
                .get(&peer)
                .is_some_and(|h| h == hostname)
        {
            return;
        }
        self.known_hostnames.insert(peer, hostname.to_string());
        let storage = self.params.storage.clone();
        let network_id = self.network_id;
        let hostname = hostname.to_string();
        tokio::spawn(async move {
            if let Err(err) = storage
                .remember_peer_hostname(network_id, *peer.as_bytes(), hostname)
                .await
            {
                tracing::debug!(%err, "could not remember a peer's hostname");
            }
        });
    }

    fn dispatch_message(&mut self, peer: EndpointId, message: ControlMessage) {
        match &message {
            ControlMessage::Transport { protocol, payload } => {
                if protocol == "link-health" {
                    // Only the elected initiator supplies a profile, for an enabled,
                    // mutually supported physical protocol in this authenticated session.
                    if peer.as_bytes() >= self.local_id.as_bytes() {
                        return;
                    }
                    let Ok((name, settings)) =
                        postcard::from_bytes::<(String, HealthSettings)>(payload.as_bytes())
                    else {
                        return;
                    };
                    if name.len() > crate::dataplane::MAX_PROTOCOL_ID_LEN
                        || !self.served_protocols().iter().any(|(p, v)| {
                            p == &name
                                && self.sessions.get(&peer).is_some_and(|s| {
                                    s.capabilities
                                        .iter()
                                        .any(|c| c.enabled && c.protocol == name && c.version == *v)
                                })
                        })
                    {
                        return;
                    }
                    let key = (peer, name);
                    // Negotiation and inbound installation are separate queues.
                    // Keep the newest settings even if the previous physical
                    // link is still in the map when this control frame arrives.
                    if let Some(link) = self.links.get(&key)
                        && !link.is_closed()
                    {
                        link.configure(settings.clone());
                    }
                    self.pending_health.insert(key, settings);
                    return;
                }
                if let Some(transport) = &self.params.transport
                    && let Err(err) =
                        transport.on_control(self.network_id, peer, protocol, payload.as_bytes())
                {
                    tracing::debug!(%err, "transport negotiation rejected");
                }
                return;
            }
            ControlMessage::Announce(announcement) => {
                let capabilities = announcement.capabilities.clone();
                let changed = self.sessions.get(&peer).is_some_and(|session| {
                    let selection = |caps: &[PluginCapability]| {
                        caps.iter()
                            .filter(|c| c.enabled)
                            .map(|c| (c.protocol.clone(), c.version))
                            .collect::<Vec<_>>()
                    };
                    selection(&session.capabilities) != selection(&capabilities)
                });
                if changed {
                    self.reset_peer_data(peer, "peer protocol selection changed");
                }
                if let Some(session) = self.sessions.get_mut(&peer) {
                    session.hostname = Some(announcement.hostname.clone());
                    session.broadcast = announcement.broadcast;
                    session.exit_node = announcement.exit_node;
                    session.capabilities = capabilities.clone();
                }
                self.remember_hostname(peer, &announcement.hostname);
                self.update_broadcast();
                self.watch_exit_via();
                self.dispatch_capabilities(peer, &capabilities);
                self.ensure_links();
                self.update_paths();
                self.announce_reach(false);
            }
            ControlMessage::Ping { seq, payload } => {
                let pong = ControlMessage::Pong {
                    seq: *seq,
                    payload: payload.clone(),
                };
                if let Err(err) = self.send_to(peer, pong) {
                    tracing::debug!(%err, "could not queue pong");
                }
            }
            ControlMessage::Vips { entries } => self.pending_vips.push(entries.clone()),
            ControlMessage::State { records } => {
                self.pending_state.push((peer, records.clone()));
            }
            ControlMessage::Retirements { markers } => {
                self.pending_retirements.push((peer, markers.clone()));
            }
            // Members somebody else knows of. Candidates, nothing more:
            // each still has to pass the handshake, and each is tried the
            // same way as one from any other source.
            ControlMessage::Peers { peers } => {
                let learned = self.learn_peers(peers);
                if learned > 0 {
                    self.start_dials();
                }
            }
            // What that peer can reach, from that peer. Kept with the time
            // it was heard, so it can go stale on its own.
            ControlMessage::Reachable { links } => {
                let heard = links.iter().cloned().collect();
                self.reachable
                    .insert(peer, (heard, std::time::Instant::now()));
                self.update_paths();
            }
            ControlMessage::Pong { .. } | ControlMessage::Bye { .. } => {}
        }

        self.emit(Event::MessageReceived {
            network: self.network_id,
            peer,
            message,
        });
    }

    fn dispatch_capabilities(&mut self, peer: EndpointId, capabilities: &[PluginCapability]) {
        let mut errors = Vec::new();
        for capability in capabilities {
            for plugin in &self.params.plugins {
                if plugin.protocol_id() != capability.protocol {
                    continue;
                }
                if plugin.protocol_version() != capability.version {
                    // Said once per peer and protocol. A version that will
                    // not match this build will not match it on the next
                    // announcement either, and repeating it every round
                    // would bury everything else.
                    let key = (peer, capability.protocol.clone());
                    if self.reported_mismatch.insert(key) {
                        errors.push((
                            capability.protocol.clone(),
                            format!(
                                "peer speaks {} version {} and this build speaks {}, so there \
                                 is no data plane with it; the control plane is unaffected",
                                capability.protocol,
                                capability.version,
                                plugin.protocol_version()
                            ),
                        ));
                    }
                    continue;
                }
                // The core hands the opaque payload over without interpreting it.
                if let Err(err) = plugin.on_peer_capability(self.network_id, peer, capability) {
                    errors.push((plugin.protocol_id().to_string(), err.to_string()));
                }
            }
        }
        for (protocol, reason) in errors {
            self.metrics.plugin_errors += 1;
            self.emit(Event::PluginError {
                network: self.network_id,
                protocol,
                reason,
            });
        }
    }

    // ----------------------------------------------------------------- status

    /// Whether a member can be reached now, by any path.
    ///
    /// The single definition every client reports: a live control session,
    /// or a data path to it through other members. Being reached over a long
    /// path is being online; how well is `data_path`'s to say.
    fn is_online(&self, peer: EndpointId) -> bool {
        peer == self.local_id || self.sessions.contains_key(&peer) || self.hub.reachable(peer)
    }

    /// How packets reach a member, which says nothing about whether it is
    /// there.
    fn data_path(&self, peer: EndpointId) -> DataPath {
        match self.hub.route(peer) {
            None => DataPath::None,
            Some(route) if route.hops <= 1 => DataPath::Direct {
                transport: self
                    .hub
                    .direct_kinds(peer)
                    .into_iter()
                    .next()
                    .unwrap_or_default(),
            },
            Some(route) => DataPath::Relayed {
                hops: route.hops,
                via: route.via,
            },
        }
    }

    /// Says so when this agent's own name is held by someone else, and
    /// again if that changes. Never anything but a statement: the agent keeps
    /// working by address either way.
    fn report_hostname_taken(&mut self) {
        let taken = self.hostname_conflict_of(&self.local_id);
        if taken == self.reported_name_taken {
            return;
        }
        if let Some(conflict) = &taken {
            tracing::warn!(
                network = %self.network_id,
                name = %conflict.name,
                holder = %conflict.holder.fmt_short(),
                "this name was claimed first by another member; it stays reachable by \
                 address but is not found by this name until it is renamed"
            );
            self.emit(Event::HostnameTaken {
                network: self.network_id,
                name: conflict.name.clone(),
                holder: conflict.holder,
            });
        }
        self.reported_name_taken = taken;
    }

    fn hostname_conflict_of(&self, author: &EndpointId) -> Option<HostnameConflict> {
        self.state
            .hostname_taken(author)
            .map(|(name, holder)| HostnameConflict {
                name: name.to_string(),
                holder,
            })
    }

    fn status(&self) -> NetworkStatus {
        let served = self.served_protocols();
        let mut peers: Vec<PeerStatus> = self
            .sessions
            .values()
            .map(|session| {
                let snapshot = snapshot_connection(&session.conn);
                PeerStatus {
                    broadcast: session.broadcast,
                    exit_node: session.exit_node,
                    endpoint_id: session.peer,
                    role: session.role,
                    hostname: session
                        .hostname
                        .clone()
                        .or_else(|| self.known_hostnames.get(&session.peer).cloned()),
                    protocols: session
                        .capabilities
                        .iter()
                        .filter(|capability| capability.enabled)
                        .filter(|capability| {
                            served.iter().any(|(name, ours)| {
                                *name == capability.protocol && *ours == capability.version
                            })
                        })
                        .map(|capability| capability.protocol.clone())
                        .collect(),
                    capabilities: session.capabilities.clone(),
                    connected_for: session.established.elapsed(),
                    paths: snapshot.paths,
                    transport: snapshot.transport,
                    rtt: snapshot.rtt,
                    connection: snapshot.counters,
                    control_messages_sent: session.messages_sent,
                    control_messages_received: session.messages_received,
                    control_bytes_sent: session.bytes_sent,
                    control_bytes_received: session.bytes_received,
                    data_path: self.data_path(session.peer),
                }
            })
            .collect();
        peers.sort_by(|a, b| a.endpoint_id.as_bytes().cmp(b.endpoint_id.as_bytes()));

        let mut candidates: Vec<CandidateStatus> = self
            .dial_states
            .iter()
            .map(|(peer, state)| CandidateStatus {
                endpoint_id: *peer,
                source: state.source,
                consecutive_failures: state.consecutive_failures,
            })
            .collect();
        candidates.sort_by(|a, b| a.endpoint_id.as_bytes().cmp(b.endpoint_id.as_bytes()));

        // The durable roster. Every author of a signed record is a member,
        // including this agent and including members that are not here —
        // except one that has released everything. That record is a
        // tombstone kept so the release cannot be undone by a replica that
        // has not heard of it; the author is not a member any more, and
        // listing it as one turns leaving into a peer that looks broken.
        let mut members: Vec<MemberStatus> = self
            .state
            .active_records()
            .into_iter()
            .filter(|record| !matches!(record.body, RecordBody::Release))
            .filter_map(|record| {
                let endpoint_id = record.author_id().ok()?;
                Some(MemberStatus {
                    record_version: record.version,
                    endpoint_id,
                    // Only uncontested claims are reported as held: two
                    // members may have claimed the same thing, and saying
                    // both hold it would be untrue on one of them.
                    overlay_address_v4: self.state.address_of(&endpoint_id),
                    hostname: self.state.hostname_of(&endpoint_id).map(str::to_string),
                    last_hostname: self.known_hostnames.get(&endpoint_id).cloned(),
                    hostname_conflict: self.hostname_conflict_of(&endpoint_id),
                    online: self.is_online(endpoint_id),
                    data_path: self.data_path(endpoint_id),
                })
            })
            .collect();
        members.sort_by(|a, b| a.endpoint_id.as_bytes().cmp(b.endpoint_id.as_bytes()));
        members.dedup_by(|a, b| a.endpoint_id == b.endpoint_id);

        NetworkStatus {
            vips: self.params.routes.vips.statuses(self.network_id),
            protocols: self
                .params
                .plugins
                .iter()
                .map(|p| p.protocol_id().to_owned())
                .collect(),
            auto_start: true,
            broadcast: self.params.broadcast,
            exit_offer: self.params.exit_offer,
            exit_via: self.params.exit_via,
            exit_via_online: self
                .params
                .exit_via
                .is_some_and(|via| self.exit_node_online(via)),
            exit_rules: self
                .params
                .interface
                .as_ref()
                .map(|interface| interface.exit_rules(self.network_id))
                .unwrap_or_default(),
            descriptor: self.params.keys.descriptor(),
            name: self.params.keys.name().clone(),
            network_id: self.network_id,
            state: NetworkState::Active,
            peers,
            candidates,
            members,
            hostname_conflict: self.hostname_conflict_of(&self.local_id),
            range: self.effective_range(),
            range_conflict: None,
            range_conflicts: Vec::new(),
            metrics: self.metrics.clone(),
            relay: self.hub.counters(),
        }
    }
}

/// Milliseconds since the Unix epoch, for the time a name was first claimed.
fn unix_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(u64::MAX, |elapsed| {
            u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX)
        })
}

/// Which side dialled, given a role and the two identities.
fn initiator_of(role: Role, local: EndpointId, peer: EndpointId) -> EndpointId {
    match role {
        Role::Initiator => local,
        Role::Responder => peer,
    }
}

/// Performs one dial and handshake.
async fn dial(
    adapter: EndpointAdapter,
    addr: iroh::EndpointAddr,
    keys: NetworkKeys,
    limits: Arc<Limits>,
    local_id: EndpointId,
    peer: EndpointId,
) -> DialOutcome {
    let connect = tokio::time::timeout(limits.dial_timeout, adapter.connect(addr)).await;
    let (conn, mut send, mut recv) = match connect {
        Ok(Ok(parts)) => parts,
        Ok(Err(err)) => {
            return DialOutcome::Failed {
                peer,
                reason: err.to_string(),
                during_handshake: false,
            };
        }
        Err(_) => {
            return DialOutcome::Failed {
                peer,
                reason: "dial timed out".into(),
                during_handshake: false,
            };
        }
    };

    match handshake::initiate(&conn, &mut send, &mut recv, local_id, &keys, &limits).await {
        Ok(outcome) => DialOutcome::Established(Box::new(InboundSession {
            conn,
            send,
            recv,
            outcome,
        })),
        Err(err) => {
            conn.close(2u32.into(), b"handshake failed");
            DialOutcome::Failed {
                peer,
                reason: err.to_string(),
                during_handshake: true,
            }
        }
    }
}

/// Encodes a path address as a cache hint.
fn encode_hint(addr: &PathAddr) -> Option<String> {
    match addr {
        PathAddr::Ip(socket) => Some(format!("ip:{socket}")),
        PathAddr::Relay(url) => Some(format!("relay:{url}")),
        PathAddr::Other(_) => None,
    }
}

/// The addresses of a candidate, in the spelling hints use.
///
/// Bounded and sorted, so what goes on the wire is the same for the same
/// address whatever order the transport happened to report it in.
fn hint_addrs(addr: &iroh::EndpointAddr) -> Vec<String> {
    let mut out: Vec<String> = addr
        .addrs
        .iter()
        .map(|transport| transport.to_string())
        .filter(|text| text.len() <= crate::proto::message::MAX_HINT_ADDR_LEN)
        .collect();
    out.sort();
    out.dedup();
    out.truncate(crate::proto::message::MAX_HINT_ADDRS);
    out
}

/// Decodes a cache hint back into an address. Malformed hints are ignored.
fn decode_hint(endpoint_id: EndpointId, hint: &str) -> Option<iroh::EndpointAddr> {
    if let Some(rest) = hint.strip_prefix("ip:") {
        let socket: std::net::SocketAddr = rest.parse().ok()?;
        return Some(iroh::EndpointAddr::new(endpoint_id).with_ip_addr(socket));
    }
    if let Some(rest) = hint.strip_prefix("relay:") {
        let url: iroh::RelayUrl = rest.parse().ok()?;
        return Some(iroh::EndpointAddr::new(endpoint_id).with_relay_url(url));
    }
    None
}

/// Folds newly learned addresses into a known candidate address.
fn merge_addr(existing: &mut iroh::EndpointAddr, incoming: &iroh::EndpointAddr) {
    if existing.id != incoming.id {
        *existing = incoming.clone();
        return;
    }
    for addr in &incoming.addrs {
        existing.addrs.insert(addr.clone());
    }
}

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

    #[test]
    fn test_data_link_backoff_progression() {
        assert_eq!(data_link_backoff(0), Duration::ZERO);
        assert_eq!(data_link_backoff(1), Duration::from_secs(3));
        assert_eq!(data_link_backoff(2), Duration::from_secs(6));
        assert_eq!(data_link_backoff(3), Duration::from_secs(12));
        assert_eq!(data_link_backoff(4), Duration::from_secs(24));
        assert_eq!(data_link_backoff(5), Duration::from_secs(48));
        assert_eq!(data_link_backoff(6), Duration::from_secs(60));
        assert_eq!(data_link_backoff(10), Duration::from_secs(60));
    }

    #[test]
    fn test_protocol_dial_state_cooldown_and_fallback_selection() {
        let peer = crate::identity::DeviceIdentity::generate().endpoint_id();
        let supported = ["wg", "tcp-tls", "wg-quic"];
        let mut states: HashMap<(EndpointId, String), ProtocolDialState> = HashMap::new();

        let now = std::time::Instant::now();

        // 1. Without cooldown, the highest priority protocol "wg" is chosen
        let chosen =
            supported
                .iter()
                .find(|proto| match states.get(&(peer, (*proto).to_string())) {
                    Some(state) => state.cooldown_until <= now,
                    None => true,
                });
        assert_eq!(chosen.copied(), Some("wg"));

        // 2. Record failure for "wg" putting it on cooldown
        states.insert(
            (peer, "wg".to_string()),
            ProtocolDialState {
                failures: 1,
                cooldown_until: now + Duration::from_secs(3),
            },
        );

        // With "wg" on cooldown, selection must fall back to next priority "tcp-tls"
        let chosen_after_failure =
            supported
                .iter()
                .find(|proto| match states.get(&(peer, (*proto).to_string())) {
                    Some(state) => state.cooldown_until <= now,
                    None => true,
                });
        assert_eq!(chosen_after_failure.copied(), Some("tcp-tls"));

        // 3. If "tcp-tls" also fails, selection falls back to "wg-quic"
        states.insert(
            (peer, "tcp-tls".to_string()),
            ProtocolDialState {
                failures: 1,
                cooldown_until: now + Duration::from_secs(3),
            },
        );
        let chosen_quic =
            supported
                .iter()
                .find(|proto| match states.get(&(peer, (*proto).to_string())) {
                    Some(state) => state.cooldown_until <= now,
                    None => true,
                });
        assert_eq!(chosen_quic.copied(), Some("wg-quic"));

        // 4. Once success is recorded (entry removed), "wg" is preferred again
        states.remove(&(peer, "wg".to_string()));
        let chosen_restored =
            supported
                .iter()
                .find(|proto| match states.get(&(peer, (*proto).to_string())) {
                    Some(state) => state.cooldown_until <= now,
                    None => true,
                });
        assert_eq!(chosen_restored.copied(), Some("wg"));
    }
}

/// The host name of a relay URL such as `https://relay.example:443/`.
fn relay_host(url: &str) -> Option<String> {
    let rest = url.split_once("://").map_or(url, |(_, rest)| rest);
    let authority = rest.split(['/', '?', '#']).next()?;
    let host = authority.rsplit_once('@').map_or(authority, |(_, h)| h);
    let host = if host.starts_with('[') {
        host.split_once(']')?.0.trim_start_matches('[')
    } else {
        host.split(':').next()?
    };
    (!host.is_empty()
        && host.len() <= 253
        && host
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == ':'))
    .then(|| host.to_ascii_lowercase())
}

#[cfg(test)]
mod relay_host_tests {
    use super::relay_host;

    #[test]
    fn the_host_is_taken_out_of_a_relay_url() {
        assert_eq!(
            relay_host("https://relay.example.com./").as_deref(),
            Some("relay.example.com.")
        );
        assert_eq!(
            relay_host("https://Euw1-1.relay.iroh.network:443/x").as_deref(),
            Some("euw1-1.relay.iroh.network")
        );
        assert_eq!(
            relay_host("http://192.0.2.7:3340").as_deref(),
            Some("192.0.2.7")
        );
        assert_eq!(
            relay_host("https://[2001:db8::1]:443/").as_deref(),
            Some("2001:db8::1")
        );
        assert_eq!(relay_host("https:///").as_deref(), None);
        assert_eq!(relay_host("https://bad host/").as_deref(), None);
    }
}

#[cfg(test)]
mod fallback_order_tests {
    use super::next_protocol_index;
    #[test]
    fn degradation_advances_the_cycle_even_after_earlier_cooldowns_expire() {
        let protocols = ["wg", "awg", "wg-quic", "tcp-tls"].map(str::to_owned);
        assert_eq!(next_protocol_index(&protocols, None), 0);
        for (i, p) in protocols.iter().enumerate() {
            assert_eq!(next_protocol_index(&protocols, Some(p)), (i + 1) % 4);
        }
        assert_eq!(next_protocol_index(&protocols, Some("removed")), 0);
    }
}