beamr 0.16.2

//! TCP connection table and lifecycle management for distribution links.

use std::fmt;
use std::io;
use std::net::SocketAddr;
#[cfg(unix)]
use std::os::fd::OwnedFd;
#[cfg(unix)]
use std::sync::Mutex as StdMutex;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, OnceLock, RwLock, Weak};
use std::time::{Duration, Instant};

use dashmap::DashMap;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio::runtime::Handle;
use tokio::sync::{Mutex, Notify};
use tokio::task::JoinHandle;

use crate::atom::{Atom, AtomTable};
use crate::distribution::handshake::{
    HandshakeError, HandshakeNode, SimultaneousDecision, initiate_handshake_async,
    respond_handshake_async_with,
};
use crate::distribution::resolver::NodeResolver;

const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);

/// Default whole-handshake deadline. Mirrors [`DEFAULT_CONNECT_TIMEOUT`]: any
/// finite value removes the deadlock; 5s tolerates a loaded peer without wedging
/// a cluster (DISTRIBUTION-HANDSHAKE-DESIGN.md D3).
const DEFAULT_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(5);

/// Idle-link liveness keepalive frame: the 8-byte all-zero header the data-frame
/// read loop already accepts as `control_len = 0, payload_len = 0` — a no-op that
/// invokes no control handler. Each peer's periodic net-tick writes this so the
/// other side's read loop refreshes its last-inbound timestamp, letting a
/// silently-partitioned (black-holed) peer be detected by a missed deadline
/// rather than only by a TCP FIN/RST.
const KEEPALIVE_FRAME: [u8; 8] = [0_u8; 8];

/// Default proactive net-tick interval: how often an idle link emits a keepalive
/// and checks the inbound-liveness deadline.
const DEFAULT_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);

/// Default inbound-liveness deadline: a link with no inbound bytes (data frame
/// OR keepalive) for this long is marked down. Comfortably larger than the
/// interval so a healthy peer's keepalives always refresh liveness in time and
/// no healthy idle link is ever spuriously downed.
const DEFAULT_HEARTBEAT_DEADLINE: Duration = Duration::from_secs(45);

/// Error returned while creating an outbound distribution TCP connection.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ConnectError {
    /// The node resolver could not turn the node name into a socket address.
    ResolveFailure,
    /// The remote address refused the TCP connection.
    ConnectionRefused,
    /// Resolution succeeded but the TCP connect did not finish before the configured timeout.
    Timeout,
    /// The responder answered the simultaneous-connect tie-break with `nok`: the
    /// peer is keeping the reciprocal (its own outbound) link, so this outbound is
    /// a benign abort, NOT a failure. The caller should treat the pair as
    /// connected via the reciprocal link and must not retry-storm (HS-3, §3.2).
    SimultaneousAbort,
    /// TCP connection failed for an I/O reason other than refusal.
    Io(String),
}

impl fmt::Display for ConnectError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::ResolveFailure => formatter.write_str("distribution node resolution failed"),
            Self::ConnectionRefused => formatter.write_str("distribution TCP connection refused"),
            Self::Timeout => formatter.write_str("distribution TCP connection timed out"),
            Self::SimultaneousAbort => formatter
                .write_str("distribution outbound aborted by simultaneous-connect tie-break"),
            Self::Io(error) => write!(formatter, "distribution TCP connection failed: {error}"),
        }
    }
}

impl std::error::Error for ConnectError {}

pub use super::connection_events::{ConnectionDownEvent, ConnectionDownHook, ConnectionDownReason};

use super::connection_events::{
    ConnectionEvent, ConnectionEventHub, ConnectionGeneration, NodeUp, SubscriberId,
};

type ControlFrameHandler = dyn Fn(Atom, &[u8], &[u8]) + Send + Sync + 'static;

/// Proactive net-tick (heartbeat) configuration for idle distribution links.
///
/// When enabled, each connection runs a periodic task that (1) writes a
/// `KEEPALIVE_FRAME` so the peer's read loop refreshes its liveness clock, and
/// (2) marks the link down if no inbound bytes have arrived within `deadline`.
/// `deadline` MUST exceed `interval` (by a comfortable margin) so a healthy
/// peer's own keepalives always refresh liveness before the deadline and no
/// healthy idle link is spuriously downed. Disabled by default.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct HeartbeatConfig {
    /// How often an idle link emits a keepalive and checks the deadline.
    pub interval: Duration,
    /// Inbound-idle duration after which the link is marked down.
    pub deadline: Duration,
}

impl HeartbeatConfig {
    /// Sane production defaults: 15s tick, 45s deadline.
    #[must_use]
    pub fn with_defaults() -> Self {
        Self {
            interval: DEFAULT_HEARTBEAT_INTERVAL,
            deadline: DEFAULT_HEARTBEAT_DEADLINE,
        }
    }
}

/// Active distribution TCP connection shared by distribution subsystems.
pub struct DistConnection {
    node: Atom,
    peer_addr: SocketAddr,
    /// `Some` while the write direction is open; taken (sending FIN — dropping
    /// an `OwnedWriteHalf` shuts down the write direction of the shared stream)
    /// by `mark_down`/`write_raw` once the connection is down, so a retained
    /// `Arc<DistConnection>` cannot keep the socket's write half alive after
    /// teardown (spec §3.6 connection-complete shutdown).
    writer: Mutex<Option<OwnedWriteHalf>>,
    /// CLOEXEC duplicate of the socket's fd, owned by this connection, so
    /// `mark_down` can `shutdown(2)` the socket WITHOUT the writer mutex:
    /// `shutdown` acts on the socket (shared by every descriptor), the wire
    /// closes (FIN) and any write blocked on it errors immediately, even while
    /// a blocked or aborted-mid-poll write holds the writer mutex. Owning a
    /// dup (not the raw fd) makes this immune to fd reuse; taken (descriptor
    /// RELEASED) by the first `mark_down`, so a retained
    /// `Arc<DistConnection>` holds no dead descriptor after teardown. Costs
    /// one fd per live connection (§6 resource-budget line, routed to the
    /// pair). Duplication failure at construction REFUSES the connection — a
    /// connection that cannot guarantee teardown is not installed. The mutex
    /// is uncontended in practice (construction and one mark_down).
    #[cfg(unix)]
    socket_fd: StdMutex<Option<OwnedFd>>,
    down: AtomicBool,
    manager: Weak<ConnectionManagerInner>,
    /// Monotonic base for this connection's inbound-liveness clock.
    created_at: Instant,
    /// Milliseconds since `created_at` at which inbound bytes were last observed
    /// by the read loop (any data frame OR keepalive). Read by the net-tick to
    /// detect a silently-partitioned peer via a missed deadline.
    last_inbound_millis: AtomicU64,
    /// Fired by `mark_down` so the read loop wakes and exits promptly instead of
    /// blocking on `read_exact` until the peer happens to close. This drops the
    /// read half (closing the socket) the moment the link is retired — e.g. when
    /// a simultaneous-connect canonical winner displaces this non-canonical link,
    /// so its socket cannot linger as an orphaned half-link.
    shutdown: Arc<Notify>,
    /// Which side opened this link. Read by the install-time dedup so a
    /// canonical-direction newcomer can displace a non-canonical incumbent during
    /// a simultaneous connect, while a lone re-dial (any direction, no canonical
    /// incumbent) still installs normally.
    direction: LinkDirection,
    /// Session generation this socket serves (immutable; inherited across
    /// simultaneous-connect displacement).
    generation: ConnectionGeneration,
    /// Peer incarnation from the authenticated handshake (0 = handshake-less
    /// test helper sentinel).
    peer_creation: u32,
    /// Reason recorded by `mark_down` BEFORE the down flag flips, so any
    /// observer of `is_down() == true` reads `Some` (set → swap(AcqRel) gives
    /// happens-before). First-set-wins on a `mark_down` race: both reasons are
    /// genuine.
    down_reason: OnceLock<ConnectionDownReason>,
}

/// A stream paired with its pre-created teardown dup: the fallible
/// duplication happens BEFORE the connection-table entry lock, so the install
/// arms stay infallible and a dup failure refuses the connection outright
/// (spec §3.6 — a connection that cannot guarantee mutex-independent closure
/// is not installed).
struct PreparedSocket {
    stream: TcpStream,
    /// Atomically-CLOEXEC dup of the socket fd — the authenticated
    /// distribution socket must not leak into spawned child processes.
    #[cfg(unix)]
    teardown_fd: OwnedFd,
}

impl PreparedSocket {
    fn prepare(stream: TcpStream) -> io::Result<Self> {
        #[cfg(all(test, unix))]
        if FAIL_TEARDOWN_DUP_FOR_TEST.with(std::cell::Cell::get) {
            return Err(io::Error::other("teardown dup failure injected"));
        }
        #[cfg(unix)]
        let teardown_fd = rustix::io::fcntl_dupfd_cloexec(&stream, 0).map_err(io::Error::from)?;
        Ok(Self {
            stream,
            #[cfg(unix)]
            teardown_fd,
        })
    }
}

// Test-only fault injection for `PreparedSocket::prepare`: thread-local so a
// parallel test run cannot poison unrelated registrations.
#[cfg(all(test, unix))]
thread_local! {
    pub(crate) static FAIL_TEARDOWN_DUP_FOR_TEST: std::cell::Cell<bool> =
        const { std::cell::Cell::new(false) };
}

impl DistConnection {
    /// Split a [`PreparedSocket`] into a connection (owning the write half
    /// plus the pre-created teardown dup) and its read half.
    fn new(
        node: Atom,
        peer_addr: SocketAddr,
        socket: PreparedSocket,
        manager: Weak<ConnectionManagerInner>,
        direction: LinkDirection,
        generation: ConnectionGeneration,
        peer_creation: u32,
    ) -> (Self, OwnedReadHalf) {
        #[cfg(unix)]
        let teardown_fd = socket.teardown_fd;
        let (read_half, writer) = socket.stream.into_split();
        let connection = Self {
            node,
            peer_addr,
            writer: Mutex::new(Some(writer)),
            #[cfg(unix)]
            socket_fd: StdMutex::new(Some(teardown_fd)),
            down: AtomicBool::new(false),
            manager,
            created_at: Instant::now(),
            last_inbound_millis: AtomicU64::new(0),
            shutdown: Arc::new(Notify::new()),
            direction,
            generation,
            peer_creation,
            down_reason: OnceLock::new(),
        };
        (connection, read_half)
    }

    /// Test-only visibility for the teardown dup: `Some(is_cloexec)` while the
    /// dup is held, `None` once released by `mark_down`.
    #[cfg(all(test, unix))]
    pub(crate) fn teardown_fd_cloexec(&self) -> Option<bool> {
        self.socket_fd
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .as_ref()
            .map(|fd| {
                rustix::io::fcntl_getfd(fd)
                    .map(|flags| flags.contains(rustix::io::FdFlags::CLOEXEC))
                    .unwrap_or(false)
            })
    }

    /// Record that inbound bytes were just observed, refreshing liveness.
    fn note_inbound_activity(&self) {
        let elapsed = u64::try_from(self.created_at.elapsed().as_millis()).unwrap_or(u64::MAX);
        self.last_inbound_millis.store(elapsed, Ordering::Release);
    }

    /// Whether no inbound bytes have been observed for at least `deadline`.
    fn inbound_idle_for(&self, deadline: Duration) -> bool {
        let last = self.last_inbound_millis.load(Ordering::Acquire);
        let now = u64::try_from(self.created_at.elapsed().as_millis()).unwrap_or(u64::MAX);
        let deadline_millis = u64::try_from(deadline.as_millis()).unwrap_or(u64::MAX);
        now.saturating_sub(last) >= deadline_millis
    }

    /// Mark this connection down because the net-tick observed no inbound
    /// liveness within the deadline (silent partition). Drives the same
    /// connection-down path a read error would.
    fn mark_down_heartbeat_timeout(self: &Arc<Self>) {
        self.mark_down(ConnectionDownReason::HeartbeatTimeout);
    }

    /// Node-name atom used as this connection's table key.
    #[must_use]
    pub fn node(&self) -> Atom {
        self.node
    }

    /// TCP peer address for diagnostics and tests.
    #[must_use]
    pub fn peer_addr(&self) -> SocketAddr {
        self.peer_addr
    }

    /// Session generation this socket serves (immutable; inherited across
    /// simultaneous-connect displacement).
    #[must_use]
    pub fn generation(&self) -> ConnectionGeneration {
        self.generation
    }

    /// Peer incarnation from the authenticated handshake (0 = handshake-less
    /// test helper sentinel).
    #[must_use]
    pub fn peer_creation(&self) -> u32 {
        self.peer_creation
    }

    /// Test-only: flip the down flag WITHOUT running the reap path, holding a
    /// down-but-unreaped entry in the table (the HS-4 re-dial race window that
    /// normally lasts only between `mark_down` and `connection_down`).
    #[cfg(test)]
    pub(crate) fn force_down_without_reap(&self) {
        self.down.store(true, Ordering::Release);
    }

    /// Return true after this connection has observed a terminal read/write failure.
    #[must_use]
    pub fn is_down(&self) -> bool {
        self.down.load(Ordering::Acquire)
    }

    /// Write raw bytes to the connection and report write-side failures to the manager.
    ///
    /// This is a transport lifecycle seam only; message encoding/framing remains owned by B-117.
    pub async fn write_raw(self: &Arc<Self>, bytes: &[u8]) -> io::Result<()> {
        let result = {
            let mut writer = self.writer.lock().await;
            let result = match writer.as_mut() {
                Some(writer) => writer.write_all(bytes).await,
                // Write half already taken by teardown: the connection is down.
                None => Err(io::Error::from(io::ErrorKind::NotConnected)),
            };
            // Close-under-the-held-lock when the connection went down while this
            // write held it: `mark_down`'s own `try_lock` close is skipped in
            // that interleaving, and this check runs strictly after the down
            // store (Acquire/Release), so one of the two paths always takes the
            // half. (A write future aborted at runtime-join drops its lock
            // without reaching here — that residual half closes when the last
            // `Arc<DistConnection>` drops, the pre-teardown behavior.)
            if self.is_down() {
                writer.take();
            }
            result
        };
        if result.is_err() {
            self.mark_down(ConnectionDownReason::WriteError);
        }
        result
    }

    /// Mark this connection down because a write exceeded its deadline.
    ///
    /// The outbound sender's drain bounds each `write_raw` with a timeout so a
    /// wedged peer cannot stall propagation for the whole cluster. On timeout the
    /// write future is dropped without `write_raw` observing a failure, so the
    /// drain calls this to drive the same connection-down path (hook + remote
    /// purge) a genuine write error would. Idempotent via the inner `mark_down`.
    pub fn mark_down_write_timeout(self: &Arc<Self>) {
        self.mark_down(ConnectionDownReason::WriteTimeout);
    }

    /// Mark this connection down because the must-deliver control lane
    /// overflowed against it.
    ///
    /// Called by `DistSender::enqueue_control` when the bounded control lane is
    /// full: a peer that cannot absorb that many pending LINK/EXIT controls is
    /// effectively down (DC-1), so instead of dropping the control silently the
    /// pinned connection is torn down and the connection-down hook's
    /// noconnection backstop supplies the coarsened signals. Also the sink for
    /// control encode failures (DC-1 has no silent arm). Idempotent via the
    /// inner `mark_down`.
    pub(crate) fn mark_down_control_overflow(self: &Arc<Self>) {
        self.mark_down(ConnectionDownReason::ControlOverflow);
    }

    fn mark_down(self: &Arc<Self>, reason: ConnectionDownReason) {
        let _ = self.down_reason.set(reason);
        if self.down.swap(true, Ordering::AcqRel) {
            return;
        }
        // Wake the read loop so it exits and drops its read half (closing the
        // socket) without waiting for the peer to close first.
        self.shutdown.notify_waiters();
        // Close the WIRE now, independent of the writer mutex: `shutdown(2)` on
        // our owned dup acts on the socket itself — FIN to the peer, and any
        // write blocked on this socket (a wedged peer holding the mutex, or a
        // write about to be aborted mid-poll) errors immediately instead of
        // holding the connection open. This is the §3.6 closure guarantee; the
        // half-takes below are resource release, not the correctness mechanism.
        // The dup is TAKEN here (descriptor released at once) — a retained
        // `Arc<DistConnection>` must not hold a dead fd until it drops.
        #[cfg(unix)]
        if let Some(fd) = self
            .socket_fd
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take()
        {
            // ENOTCONN (already reset by the peer) is fine — the wire is down.
            let _ = rustix::net::shutdown(&fd, rustix::net::Shutdown::Both);
        }
        // Release the write half when uncontended; a write holding the lock has
        // just been errored out by the socket shutdown and its own post-write
        // `is_down` check (ordered after our `down.swap` per Acquire/Release)
        // takes the half on its way out.
        if let Ok(mut writer) = self.writer.try_lock() {
            writer.take();
        }
        if let Some(manager) = self.manager.upgrade() {
            manager.connection_down(self.node, self, reason);
        }
    }
}

/// Handle for a running inbound accept loop.
pub struct AcceptHandle {
    local_addr: SocketAddr,
    shutdown: Arc<Notify>,
    task: JoinHandle<()>,
}

impl AcceptHandle {
    /// The address actually bound by the TCP listener.
    #[must_use]
    pub fn local_addr(&self) -> SocketAddr {
        self.local_addr
    }

    /// Ask the accept loop to stop. The task exits asynchronously.
    pub fn shutdown(&self) {
        self.shutdown.notify_waiters();
    }

    /// Return true if the accept task has completed.
    #[must_use]
    pub fn is_finished(&self) -> bool {
        self.task.is_finished()
    }
}

impl Drop for AcceptHandle {
    fn drop(&mut self) {
        self.shutdown.notify_waiters();
        self.task.abort();
    }
}

struct ConnectionManagerInner {
    connections: DashMap<Atom, Arc<DistConnection>>,
    /// Peer-name atoms with an in-flight OUTBOUND dial (the `Connecting` state,
    /// DISTRIBUTION-HANDSHAKE-DESIGN.md §3.1). Recorded before the outbound
    /// handshake awaits so a concurrent inbound responder can detect the
    /// simultaneous case and apply the name-comparison tie-break (HS-3, §3.2).
    ///
    /// The value is an abort flag for that outbound. When this node is the
    /// lower-named peer it keeps the reciprocal INBOUND (the responder decides
    /// `ContinueSimultaneous`) and must retire its own competing outbound rather
    /// than letting both install and collide in the HS-2 dedup — a collision
    /// whose loser-socket drop can tear down the peer's surviving link, leaving
    /// the pair with zero links and no re-dial. The decider sets this flag; the
    /// outbound `connect` checks it after the handshake and bows out cleanly
    /// (`SimultaneousAbort`) if set (§3.2, point 2: "mark the local outbound to
    /// abort").
    connecting: DashMap<Atom, Arc<AtomicBool>>,
    atom_table: Arc<AtomTable>,
    resolver: Arc<dyn NodeResolver + Send + Sync>,
    connect_timeout: Duration,
    /// Whole-handshake deadline applied around the OTP exchange on both the
    /// outbound `connect` and the inbound accept-side responder. Bounds a stalled
    /// or malicious peer so `connect` always returns and no responder task parks
    /// forever (DISTRIBUTION-HANDSHAKE-DESIGN.md HS-1, D3).
    handshake_timeout: Duration,
    /// Connection lifecycle event hub: multi-subscriber Up/Down delivery,
    /// per-peer session generations, and the legacy single-slot down callback
    /// (which fires LAST, Down only).
    events: ConnectionEventHub,
    control_frame_handler: RwLock<Option<Arc<ControlFrameHandler>>>,
    /// Shared handshake secret. Both peers must agree on this value or the OTP
    /// challenge/response is rejected and the connection is dropped.
    cookie: String,
    /// This node's advertised distribution name, sent in the handshake name
    /// packet so the peer keys its connection table by our identity.
    local_node_name: String,
    /// This node's creation value, sent alongside the name in the handshake.
    local_creation: u32,
    /// Runtime handle that drives the read/accept tasks. In production the
    /// scheduler binds the [`DistSender`](crate::distribution::sender::DistSender)
    /// runtime here so the receive side is driven even though no ambient runtime
    /// exists. When unset (e.g. `#[tokio::test]`), the tasks fall back to the
    /// ambient runtime via bare `tokio::spawn`.
    runtime_handle: RwLock<Option<Handle>>,
    /// Proactive net-tick configuration. When `Some`, every established link runs
    /// a heartbeat task (keepalive + inbound-liveness deadline). `None` disables
    /// the net-tick (links are then only marked down on read EOF/error or a write
    /// timeout, the pre-net-tick behaviour).
    heartbeat: Option<HeartbeatConfig>,
    /// Count of proactive net-tick (heartbeat) tasks spawned since construction,
    /// one per established link when the net-tick is enabled. Reported as the
    /// distribution bundle's heartbeat task-class policy line (spec §3.7/§5) —
    /// heartbeats are async tasks with no OS thread, so they are inventoried as a
    /// counter, never a thread line.
    heartbeat_tasks_spawned: AtomicU64,
}

impl ConnectionManagerInner {
    /// Spawn `future` on the bound runtime handle when one is set, else on the
    /// ambient runtime. Used for the read/accept lifecycle tasks.
    fn spawn_lifecycle<F>(&self, future: F) -> JoinHandle<()>
    where
        F: std::future::Future<Output = ()> + Send + 'static,
    {
        let handle = self
            .runtime_handle
            .read()
            .unwrap_or_else(|error| error.into_inner())
            .clone();
        match handle {
            Some(handle) => handle.spawn(future),
            None => tokio::spawn(future),
        }
    }

    /// Build the local handshake descriptor advertised to peers.
    fn handshake_node(&self) -> Result<HandshakeNode, ConnectError> {
        HandshakeNode::with_default_flags(self.local_node_name.clone(), self.local_creation)
            .map_err(|error| ConnectError::Io(error.to_string()))
    }

    /// Produce a per-handshake challenge value. The challenge is drawn from a
    /// cryptographically secure random source, so it is unpredictable per
    /// session. This is the canonical OTP behavior: the shared cookie still
    /// provides authentication, while an unpredictable challenge adds
    /// defense-in-depth against replay (an attacker cannot precompute the
    /// digest for a challenge they cannot guess).
    fn gen_challenge(&self) -> u32 {
        rand::random::<u32>()
    }

    /// Decide the OTP status an inbound responder should emit for a peer whose
    /// advertised name is `peer_name` (HS-3, D1 — OTP verbatim).
    ///
    /// With no competing local outbound to that peer, continue normally. If a
    /// local outbound dial to the same peer name is in flight, break the tie by
    /// literal name comparison: the higher-named node's OUTBOUND survives, so the
    /// responder on the lower-named node continues this inbound
    /// (`ContinueSimultaneous`, when `peer_name > local_name`) and the responder
    /// on the higher-named node rejects it (`Reject`, when `local_name > peer_name`)
    /// to keep its own outbound. Distinct cluster members have unique names, so
    /// equality cannot occur; if it ever did, `Continue` plus the install-time
    /// dedup (HS-2) is the backstop.
    fn decide_inbound_status(&self, peer_name: &str) -> SimultaneousDecision {
        let peer_atom = self.atom_table.intern(peer_name);
        let Some(abort) = self
            .connecting
            .get(&peer_atom)
            .map(|entry| Arc::clone(&entry))
        else {
            return SimultaneousDecision::Continue;
        };
        match peer_name.cmp(self.local_node_name.as_str()) {
            std::cmp::Ordering::Greater => {
                // This (lower-named) node keeps the inbound. Retire its own
                // competing outbound so it does not also install and collide in
                // the HS-2 dedup (§3.2: "mark the local outbound to abort").
                abort.store(true, Ordering::SeqCst);
                SimultaneousDecision::ContinueSimultaneous
            }
            std::cmp::Ordering::Less => SimultaneousDecision::Reject,
            std::cmp::Ordering::Equal => SimultaneousDecision::Continue,
        }
    }

    /// The connection direction that survives a simultaneous connect for `peer`,
    /// computed identically on both nodes by literal name comparison: the
    /// higher-named node's OUTBOUND wins (equivalently, the lower-named node's
    /// INBOUND). This is the install-time backstop that makes
    /// [`ConnectionManager::register_connection`] timing-independent even when the
    /// in-handshake HS-3 tie-break window is missed. The pathological equal-name
    /// case (distinct cluster members never collide) falls to `Inbound`, an
    /// arbitrary-but-consistent local choice; the per-pair agreement that matters
    /// is preserved because names are unique.
    fn canonical_direction(&self, peer: Atom) -> LinkDirection {
        let Some(peer_name) = self.atom_table.resolve(peer) else {
            // Unknown peer name: no competing direction can be reasoned about, so
            // treat the incoming link as the canonical one (install it).
            return LinkDirection::Inbound;
        };
        if self.local_node_name.as_str() > peer_name {
            LinkDirection::Outbound
        } else {
            LinkDirection::Inbound
        }
    }
}

impl ConnectionManagerInner {
    fn connection_down(
        &self,
        node: Atom,
        connection: &Arc<DistConnection>,
        reason: ConnectionDownReason,
    ) {
        use dashmap::mapref::entry::Entry;
        if let Entry::Occupied(occupied) = self.connections.entry(node)
            && Arc::ptr_eq(occupied.get(), connection)
        {
            // Enqueue UNDER the entry guard: a racing register_connection
            // for this node blocks on the entry until we release, so its
            // Up(g+1) can never be queued ahead of this Down(g). The
            // enqueue precedes `remove` only because dashmap's
            // `OccupiedEntry::remove(self)` consumes the guard;
            // INV-DOWN-VISIBILITY still holds because a concurrent
            // dispatcher's `get_connection` blocks on the shard lock this
            // entry holds until the removal completes.
            self.events
                .enqueue(ConnectionEvent::down(node, connection.generation(), reason));
            occupied.remove();
        }
        // Guard released. Deliver with no locks held (same discipline the old
        // hook.invoke had, now ORDERED against concurrent installs and
        // SYNCHRONOUS: when this returns, every subscriber has run). Dispatch
        // even when the ptr-eq LOST: the winner (an HS-4 re-dial that replaced
        // this socket) enqueued this session's Down under the entry guard we
        // just contended on but may not have DELIVERED it yet — returning
        // without draining would let our caller (e.g. `disconnect_node`)
        // return before the Down its own `mark_down` caused was delivered,
        // breaking INV-SYNC. When the queue is already empty this is a cheap
        // bounce off the dispatch gate.
        self.events.dispatch();
    }
}

/// Which side opened a TCP connection, used by the install-time canonical
/// dedup ([`ConnectionManagerInner::canonical_direction`]) to resolve a
/// simultaneous connect deterministically on both nodes.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum LinkDirection {
    /// This node dialed the peer (`connect`).
    Outbound,
    /// This node accepted the peer's dial (`handle_accepted`).
    Inbound,
}

/// RAII marker that records an in-flight outbound dial in the manager's
/// `connecting` set and clears it on drop, on every `connect` exit path (HS-3).
struct ConnectingGuard {
    inner: Arc<ConnectionManagerInner>,
    peer: Atom,
    /// Set by a concurrent inbound responder (the tie-break) to tell this
    /// outbound to bow out so the reciprocal inbound is the sole survivor.
    abort: Arc<AtomicBool>,
}

impl ConnectingGuard {
    fn new(inner: &Arc<ConnectionManagerInner>, peer_name: &str) -> Self {
        let peer = inner.atom_table.intern(peer_name);
        let abort = Arc::new(AtomicBool::new(false));
        inner.connecting.insert(peer, Arc::clone(&abort));
        Self {
            inner: Arc::clone(inner),
            peer,
            abort,
        }
    }

    /// Whether a concurrent inbound responder has claimed the reciprocal link,
    /// asking this outbound to abort (HS-3 tie-break, §3.2).
    fn is_aborted(&self) -> bool {
        self.abort.load(Ordering::SeqCst)
    }
}

impl Drop for ConnectingGuard {
    fn drop(&mut self) {
        self.inner.connecting.remove(&self.peer);
    }
}

/// Distribution TCP connection manager and active connection table.
#[derive(Clone)]
pub struct ConnectionManager {
    inner: Arc<ConnectionManagerInner>,
}

impl ConnectionManager {
    /// Create a connection manager with the default five-second connect timeout.
    ///
    /// `cookie`, `local_node_name`, and `local_creation` are the local node's
    /// OTP handshake identity: the cookie authenticates peers, while the name and
    /// creation are advertised so a peer keys its connection table by this node.
    #[must_use]
    pub fn new(
        atom_table: Arc<AtomTable>,
        resolver: Arc<dyn NodeResolver + Send + Sync>,
        cookie: impl Into<String>,
        local_node_name: impl Into<String>,
        local_creation: u32,
    ) -> Self {
        Self::with_connect_timeout(
            atom_table,
            resolver,
            cookie,
            local_node_name,
            local_creation,
            DEFAULT_CONNECT_TIMEOUT,
        )
    }

    /// Create a connection manager with a caller-specified connect timeout.
    #[must_use]
    pub fn with_connect_timeout(
        atom_table: Arc<AtomTable>,
        resolver: Arc<dyn NodeResolver + Send + Sync>,
        cookie: impl Into<String>,
        local_node_name: impl Into<String>,
        local_creation: u32,
        connect_timeout: Duration,
    ) -> Self {
        Self {
            inner: Arc::new(ConnectionManagerInner {
                connections: DashMap::new(),
                connecting: DashMap::new(),
                atom_table,
                resolver,
                connect_timeout,
                handshake_timeout: DEFAULT_HANDSHAKE_TIMEOUT,
                events: ConnectionEventHub::new(),
                control_frame_handler: RwLock::new(None),
                cookie: cookie.into(),
                local_node_name: local_node_name.into(),
                local_creation,
                runtime_handle: RwLock::new(None),
                heartbeat: None,
                heartbeat_tasks_spawned: AtomicU64::new(0),
            }),
        }
    }

    /// Enable the proactive net-tick (heartbeat) on a freshly-built manager.
    ///
    /// Builder-style: must be called before the manager is cloned or any
    /// connection is started, while `inner` is still uniquely owned (the config
    /// is read by per-connection lifecycle tasks at spawn time). Returns `self`
    /// unchanged if the manager has already been shared. `config.deadline` should
    /// exceed `config.interval` so healthy idle links are never spuriously downed.
    #[must_use]
    pub fn with_heartbeat(mut self, config: HeartbeatConfig) -> Self {
        if let Some(inner) = Arc::get_mut(&mut self.inner) {
            inner.heartbeat = Some(config);
        }
        self
    }

    /// Bind a tokio runtime handle for the read/accept lifecycle tasks.
    ///
    /// The scheduler calls this with the owned `DistSender` runtime handle so the
    /// receive side is driven in production (where no ambient runtime exists).
    /// Must be called before any connection is established; existing tasks keep
    /// the runtime they were spawned on.
    pub fn set_runtime_handle(&self, handle: Handle) {
        *self
            .inner
            .runtime_handle
            .write()
            .unwrap_or_else(|error| error.into_inner()) = Some(handle);
    }

    /// Return the configured outbound TCP connection timeout.
    #[must_use]
    pub fn connect_timeout(&self) -> Duration {
        self.inner.connect_timeout
    }

    /// Return the configured whole-handshake deadline.
    #[must_use]
    pub fn handshake_timeout(&self) -> Duration {
        self.inner.handshake_timeout
    }

    /// Override the whole-handshake deadline on a freshly-built manager.
    ///
    /// Builder-style: must be called before the manager is cloned or any
    /// connection is started, while its `inner` is still uniquely owned. Returns
    /// `self` unchanged if the manager has already been shared (a clone exists),
    /// since the deadline is read by in-flight handshakes and cannot be mutated
    /// race-free afterward.
    #[must_use]
    pub fn with_handshake_timeout(mut self, handshake_timeout: Duration) -> Self {
        if let Some(inner) = Arc::get_mut(&mut self.inner) {
            inner.handshake_timeout = handshake_timeout;
        }
        self
    }

    /// Return a clone of the legacy connection-down callback slot.
    ///
    /// 0.11-compat surface: the slot is replace-on-register, fires LAST
    /// (after every hub subscriber), and only for Down. New consumers should
    /// prefer [`subscribe_connection_events`](Self::subscribe_connection_events).
    #[must_use]
    pub fn connection_down_hook(&self) -> ConnectionDownHook {
        self.inner.events.legacy_down_hook()
    }

    /// Register or replace the legacy connection-down callback.
    ///
    /// 0.11-compat surface: replace-on-register, fires LAST (after every hub
    /// subscriber), Down only. New consumers should prefer
    /// [`subscribe_connection_events`](Self::subscribe_connection_events).
    pub fn register_connection_down<F>(&self, callback: F)
    where
        F: Fn(ConnectionDownEvent) + Send + Sync + 'static,
    {
        self.inner.events.legacy_down_hook().register(callback);
    }

    /// Subscribe to connection lifecycle events (Up + Down). Unlimited
    /// subscribers, invoked in registration order; see the module-level
    /// "Delivery and ordering contract" in
    /// [`connection_events`](super::connection_events). Callbacks must not
    /// block, must not perform socket I/O, and must capture `Weak` (never
    /// `Arc`) handles to anything owning this manager.
    pub fn subscribe_connection_events<F>(&self, callback: F) -> SubscriberId
    where
        F: Fn(ConnectionEvent) + Send + Sync + 'static,
    {
        self.inner.events.subscribe(callback)
    }

    /// Remove a subscription. `false` if the id was not (or no longer)
    /// registered.
    pub fn unsubscribe_connection_events(&self, id: SubscriberId) -> bool {
        self.inner.events.unsubscribe(id)
    }

    /// Subscribe to connection lifecycle events with synthetic catch-up: the
    /// blessed late-subscriber path (INV-NO-REPLAY). Before this method
    /// returns, `callback` is invoked on the calling thread with a synthetic
    /// [`ConnectionEvent::Up`]`(node, generation, peer_creation)` for every
    /// currently live peer (down links excluded), then registered. Snapshot,
    /// synthetic delivery, and registration all happen while holding the
    /// event-dispatch gate, so no real event interleaves between them: the
    /// subscriber observes a per-node stream satisfying INV-ALTERNATION from
    /// its first synthetic Up, missing no session and seeing none twice.
    ///
    /// The synthetic Ups are subscriber-local catch-up: they are delivered to
    /// THIS callback only and are NOT part of the global event order other
    /// subscribers saw (nothing is replayed to, or duplicated for, anyone
    /// else). See the "Delivery and ordering contract" in
    /// [`connection_events`](super::connection_events).
    ///
    /// Do NOT call this from inside a subscriber callback on the same
    /// manager: the reentrancy check then registers and returns the
    /// subscription WITHOUT any synthetic events (no race-free snapshot
    /// exists mid-drain, and blocking would self-deadlock). Callback
    /// discipline is as
    /// [`subscribe_connection_events`](Self::subscribe_connection_events):
    /// must not block, must not perform socket I/O, and must capture `Weak`
    /// (never `Arc`) handles to anything owning this manager.
    pub fn subscribe_connection_events_with_snapshot<F>(&self, callback: F) -> SubscriberId
    where
        F: Fn(ConnectionEvent) + Send + Sync + 'static,
    {
        self.inner
            .events
            .subscribe_with_snapshot(callback, || self.connected_peers())
    }

    /// Snapshot of live connections as their in-force [`NodeUp`] rows (down
    /// links excluded). Per-peer-consistent; no cross-peer atomicity.
    /// Late-subscriber recipe: subscribe FIRST, then snapshot, then per peer
    /// keep the row/event with the highest generation — generation is the
    /// dedupe key. No replay.
    #[must_use]
    pub fn connected_peers(&self) -> Vec<NodeUp> {
        self.inner
            .connections
            .iter()
            .filter(|entry| !entry.value().is_down())
            .map(|entry| NodeUp {
                node: *entry.key(),
                generation: entry.value().generation(),
                peer_creation: entry.value().peer_creation(),
            })
            .collect()
    }

    /// Last generation ever assigned for `node` (even if currently down);
    /// `None` if this manager never installed a connection to `node`.
    #[must_use]
    pub fn last_peer_generation(&self, node: Atom) -> Option<ConnectionGeneration> {
        self.inner.events.last_generation(node)
    }

    /// Register a handler for framed distribution control messages read from active links.
    ///
    /// Legacy shape without the frame's origin; the body wraps
    /// [`Self::register_control_frame_handler_with_origin`] and drops the
    /// origin argument. Kept byte-identical for 0.11 embedders.
    pub fn register_control_frame_handler<F>(&self, handler: F)
    where
        F: Fn(&[u8], &[u8]) + Send + Sync + 'static,
    {
        self.register_control_frame_handler_with_origin(move |_origin, control, payload| {
            handler(control, payload);
        });
    }

    /// Register a handler for framed distribution control messages read from
    /// active links, receiving the authenticated origin with each frame.
    ///
    /// The origin is the connection's node atom — the authenticated handshake
    /// name that keys the connection table — passed by the read loop so the
    /// handler can reject link controls whose `from` field forges another
    /// peer's identity.
    pub fn register_control_frame_handler_with_origin<F>(&self, handler: F)
    where
        F: Fn(Atom, &[u8], &[u8]) + Send + Sync + 'static,
    {
        let mut slot = self
            .inner
            .control_frame_handler
            .write()
            .unwrap_or_else(|error| error.into_inner());
        *slot = Some(Arc::new(handler));
    }

    /// Number of active, identified distribution connections.
    #[must_use]
    pub fn connection_count(&self) -> usize {
        self.inner.connections.len()
    }

    /// Whether the proactive net-tick (heartbeat) is enabled on this manager.
    #[must_use]
    pub fn heartbeat_enabled(&self) -> bool {
        self.inner.heartbeat.is_some()
    }

    /// Count of proactive net-tick (heartbeat) tasks spawned since construction
    /// (spec §3.7/§5). One per established link when the net-tick is enabled;
    /// zero when it is disabled or no link has yet been established.
    #[must_use]
    pub fn heartbeat_tasks_spawned(&self) -> u64 {
        self.inner.heartbeat_tasks_spawned.load(Ordering::Relaxed)
    }

    /// The atom table this manager keys its connection table by.
    ///
    /// Connections are keyed by the peer's authenticated handshake name interned
    /// into this table, so callers that look up a connection by name (e.g.
    /// `get_connection(atom_table().intern(peer_name))`) must intern through the
    /// same table the manager used. Exposed for integration tests and callers that
    /// drive the manager directly rather than through the scheduler.
    #[must_use]
    pub fn atom_table(&self) -> Arc<AtomTable> {
        Arc::clone(&self.inner.atom_table)
    }

    /// Look up an active distribution connection by node-name atom.
    #[must_use]
    pub fn get_connection(&self, node: Atom) -> Option<Arc<DistConnection>> {
        self.inner
            .connections
            .get(&node)
            .map(|entry| Arc::clone(entry.value()))
    }

    /// Return the node-name atoms for all active distribution connections.
    #[must_use]
    pub fn connected_nodes(&self) -> Vec<Atom> {
        let mut nodes: Vec<_> = self
            .inner
            .connections
            .iter()
            .map(|entry| *entry.key())
            .collect();
        nodes.sort_unstable_by_key(|node| node.index());
        nodes
    }

    /// Idempotently connect to a node-name atom, returning `false` for transport failures.
    ///
    /// A simultaneous-connect `nok` abort ([`ConnectError::SimultaneousAbort`]) is
    /// treated as success, not a failure: the peer is keeping the reciprocal link,
    /// so the pair is (or is about to be) connected and the caller must not
    /// retry-storm (HS-3).
    ///
    /// The "already connected" early-return only fires for a LIVE link. A link
    /// that has gone down but not yet been reaped from the table (the window
    /// between `mark_down` flipping the flag and `connection_down` removing the
    /// entry) must NOT be reported as connected, or a caller's reconnect attempt
    /// would be told the peer is up and never re-dial. Skipping a down entry here
    /// makes re-dial deterministic: `connect` runs the handshake and
    /// `register_connection` replaces the stale entry (HS-4, §3.4).
    pub async fn connect_node(&self, node: Atom) -> bool {
        if self
            .get_connection(node)
            .is_some_and(|connection| !connection.is_down())
        {
            return true;
        }
        let Some(node_name) = self.inner.atom_table.resolve(node).map(str::to_owned) else {
            return false;
        };
        matches!(
            self.connect(&node_name).await,
            Ok(_) | Err(ConnectError::SimultaneousAbort)
        )
    }

    /// Manually disconnect an active node and emit the connection-down hook once.
    pub fn disconnect_node(&self, node: Atom) -> bool {
        let Some(connection) = self.get_connection(node) else {
            return true;
        };
        connection.mark_down(ConnectionDownReason::ManualDisconnect);
        true
    }

    /// Tear down every active connection and abort every in-flight outbound
    /// dial — the scheduler-shutdown half of §3.6's connection-complete
    /// teardown. Each active connection goes through the ordinary `mark_down`
    /// path (`ManualDisconnect`: the local node is explicitly closing), so its
    /// write half closes immediately (FIN), its read loop is woken to exit, the
    /// table entry is removed, and the Down event is DELIVERED before this
    /// returns (INV-SYNC). In-flight dials get their HS-3 abort flag set; the
    /// dial's own exit paths retire it.
    pub fn disconnect_all(&self) {
        for entry in &self.inner.connecting {
            entry.value().store(true, Ordering::Release);
        }
        let nodes: Vec<Atom> = self
            .inner
            .connections
            .iter()
            .map(|entry| *entry.key())
            .collect();
        for node in nodes {
            self.disconnect_node(node);
        }
    }

    /// Create a manager and start a dedicated asynchronous TCP accept loop.
    pub async fn start(
        listen_addr: SocketAddr,
        resolver: Arc<dyn NodeResolver + Send + Sync>,
        cookie: impl Into<String>,
        local_node_name: impl Into<String>,
        local_creation: u32,
    ) -> io::Result<(Self, AcceptHandle)> {
        let manager = Self::new(
            Arc::new(AtomTable::with_common_atoms()),
            resolver,
            cookie,
            local_node_name,
            local_creation,
        );
        let handle = manager.listen(listen_addr).await?;
        Ok((manager, handle))
    }

    /// Start a dedicated asynchronous TCP accept loop for this manager.
    pub async fn listen(&self, listen_addr: SocketAddr) -> io::Result<AcceptHandle> {
        let listener = TcpListener::bind(listen_addr).await?;
        Ok(self.listen_with(listener))
    }

    /// Start a dedicated asynchronous TCP accept loop on a pre-bound listener.
    ///
    /// Separated from [`listen`](Self::listen) so callers that must bind the
    /// listener before the manager exists (e.g. to publish the chosen port into a
    /// resolver) can reuse the same accept-loop spawn. The accept loop runs on the
    /// bound runtime handle via `ConnectionManagerInner::spawn_lifecycle`.
    #[must_use]
    pub fn listen_with(&self, listener: TcpListener) -> AcceptHandle {
        let local_addr = listener
            .local_addr()
            .unwrap_or_else(|_| SocketAddr::from(([0, 0, 0, 0], 0)));
        let shutdown = Arc::new(Notify::new());
        let task_shutdown = Arc::clone(&shutdown);
        let manager = self.clone();
        let task = self.inner.spawn_lifecycle(async move {
            manager.accept_loop(listener, task_shutdown).await;
        });
        AcceptHandle {
            local_addr,
            shutdown,
            task,
        }
    }

    /// Resolve `node_name`, open a TCP connection, run the OTP distribution
    /// handshake, and add the authenticated link to the active table.
    ///
    /// The connection is keyed by the name the peer advertises in the handshake
    /// — not by `node_name`/the resolver key — so identity is established by the
    /// authenticated handshake rather than by trusting the dialed address. On any
    /// handshake failure the stream is dropped (closing the TCP connection) and a
    /// [`ConnectError::Io`] is returned.
    pub async fn connect(&self, node_name: &str) -> Result<Arc<DistConnection>, ConnectError> {
        let addr = self
            .inner
            .resolver
            .resolve(node_name)
            .await
            .map_err(|_| ConnectError::ResolveFailure)?;
        let mut stream = match tokio::time::timeout(
            self.inner.connect_timeout,
            TcpStream::connect(addr),
        )
        .await
        {
            Ok(Ok(stream)) => stream,
            Ok(Err(error)) if error.kind() == io::ErrorKind::ConnectionRefused => {
                return Err(ConnectError::ConnectionRefused);
            }
            Ok(Err(error)) => return Err(ConnectError::Io(error.to_string())),
            Err(_) => return Err(ConnectError::Timeout),
        };
        let peer_addr = stream.peer_addr().unwrap_or(addr);

        let local = self.inner.handshake_node()?;
        // Mark this peer name as having an in-flight outbound BEFORE the handshake
        // awaits, so a concurrent inbound responder can detect the simultaneous
        // case and apply the tie-break (HS-3). The guard clears the mark on every
        // exit path. The dialed `node_name` is the peer's authenticated name in a
        // by-name cluster mesh (haematite FullMesh), which is what the peer
        // advertises and what its responder compares against.
        let _connecting = ConnectingGuard::new(&self.inner, node_name);
        // Bound the whole handshake so a stalled or malicious peer can never park
        // this call forever; `connect` is now guaranteed to return within
        // handshake_timeout (HS-1). On elapse the stream is dropped, closing the
        // TCP connection.
        let result = match tokio::time::timeout(
            self.inner.handshake_timeout,
            initiate_handshake_async(
                &mut stream,
                &local,
                &self.inner.cookie,
                self.inner.gen_challenge(),
            ),
        )
        .await
        {
            Ok(Ok(result)) => result,
            Ok(Err(HandshakeError::BadStatus(status))) if status == "nok" => {
                // The peer kept the reciprocal link via the tie-break. Benign:
                // drop our stream and report a non-failure abort so the caller
                // does not retry-storm.
                return Err(ConnectError::SimultaneousAbort);
            }
            Ok(Err(error)) => return Err(ConnectError::Io(error.to_string())),
            Err(_) => return Err(ConnectError::Io(HandshakeError::Timeout.to_string())),
        };
        // Dropping the stream on the error paths above closes the TCP connection;
        // on success the authenticated remote name becomes the connection-table
        // key.
        //
        // Tie-break, install side: if our concurrent inbound responder already
        // decided to keep the reciprocal inbound link for this peer (we are the
        // lower-named node, HS-3 §3.2), retire this outbound instead of also
        // installing it. Two installs for one peer would otherwise collide in the
        // HS-2 dedup, and the loser-socket drop can tear down the peer's surviving
        // link, leaving the pair with zero links and no re-dial. Dropping the
        // stream closes this TCP connection; the reciprocal inbound is the
        // survivor, so this is a benign `SimultaneousAbort`, not a failure.
        if _connecting.is_aborted() {
            drop(stream);
            return Err(ConnectError::SimultaneousAbort);
        }
        let node = self.inner.atom_table.intern(result.remote_name());
        self.register_connection(
            node,
            peer_addr,
            stream,
            LinkDirection::Outbound,
            result.remote_creation(),
        )
        .map_err(|error| ConnectError::Io(error.to_string()))
    }

    /// Install an authenticated link, deduplicating against an existing `Up`
    /// connection for the same peer name (HS-2) by the deterministic
    /// canonical-direction rule.
    ///
    /// Two simultaneous handshakes (one inbound, one outbound) for the same pair
    /// can both reach this point — and on a busy host BOTH outbounds can finish
    /// and register before either inbound responder is scheduled, so the HS-3
    /// in-handshake tie-break window is missed entirely. A first-registered-wins
    /// dedup then resolves the collision differently on the two nodes (whichever
    /// direction happened to register first locally), so each node can drop the
    /// very socket its peer is keeping — leaving the pair with zero live links and
    /// no re-dial.
    ///
    /// The fix is timing-independent: for any pair both nodes agree, by literal
    /// name comparison, that the survivor is the HIGHER-named node's OUTBOUND
    /// connection (equivalently the lower-named node's inbound) — the same single
    /// TCP socket on both ends. The dedup keyed on the incumbent's stored
    /// [`LinkDirection`]: a newcomer loses ONLY to a LIVE incumbent of the
    /// canonical direction carrying the same peer incarnation; otherwise it
    /// installs, displacing a down, non-canonical, or stale-incarnation
    /// incumbent (a nonzero `peer_creation` mismatch proves the peer
    /// restarted — a session boundary the tie-break must not shield, so the
    /// old session closes with Down(g) and the newcomer opens Up(g+1)). So
    /// during a simultaneous connect each node keeps
    /// only its canonical-direction link (the canonical socket is never torn down
    /// by either side), while a LONE re-dial — which only ever meets a stale,
    /// non-canonical, or absent incumbent — always re-establishes the link.
    fn register_connection(
        &self,
        node: Atom,
        peer_addr: SocketAddr,
        stream: TcpStream,
        direction: LinkDirection,
        peer_creation: u32,
    ) -> io::Result<Arc<DistConnection>> {
        use dashmap::mapref::entry::Entry;

        // The fallible half FIRST, before the entry lock: a connection whose
        // teardown dup cannot be created (fd exhaustion) is REFUSED here —
        // never installed with a degraded closure guarantee (spec §3.6).
        let socket = PreparedSocket::prepare(stream)?;
        let canonical = self.inner.canonical_direction(node);
        // The connection installed (if any) plus the displaced link to retire
        // AFTER the entry guard is released — `mark_down` re-enters this same
        // `connections` map, so calling it while holding the entry lock would
        // deadlock on the shard.
        let (installed, read_half, displaced) = match self.inner.connections.entry(node) {
            Entry::Occupied(mut occupied) => {
                let incumbent = occupied.get();
                // A nonzero-creation mismatch proves the incumbent serves a
                // DEAD peer incarnation: the peer restarted (its old
                // incarnation died without a FIN/RST reaching us — silent
                // partition, power loss, kill-9 + fast restart) and this
                // newcomer is the restarted VM's dial. The canonical
                // tie-break below exists to resolve SAME-incarnation
                // simultaneous connects; it must never shield a stale
                // incarnation, so a creation-mismatch newcomer always
                // installs. 0 is the handshake-less test-helper sentinel,
                // never a discriminator.
                let creation_mismatch = incumbent.peer_creation() != 0
                    && peer_creation != 0
                    && incumbent.peer_creation() != peer_creation;
                if !incumbent.is_down() && incumbent.direction == canonical && !creation_mismatch {
                    // A LIVE incumbent of the canonical direction, serving the
                    // same peer incarnation as far as the handshake can tell,
                    // already holds this pair — it is the rightful survivor on
                    // both nodes, so this newcomer loses regardless of its own
                    // direction. Drop its stream (closing the TCP connection)
                    // and do NOT spawn a reader. (A lone re-dial never hits
                    // this: the only live incumbent it could meet is a stale
                    // non-canonical link or a stale incarnation, both handled
                    // below.)
                    drop(socket);
                    return Ok(Arc::clone(incumbent));
                }
                // The incumbent is down (reap/reconnect), a non-canonical link
                // this newcomer is entitled to replace (a simultaneous-connect
                // canonical winner, or a lone re-dial superseding a stale
                // link), OR a stale incarnation losing to the restarted peer's
                // dial. Install this one and retire the old.
                let previous = Arc::clone(incumbent);
                // Sample the down flag ONCE: it can flip concurrently, and the
                // generation choice and the Down+Up emission must agree.
                let previous_down = previous.is_down();
                // A LIVE incumbent displaced by a NEW peer incarnation —
                // canonical or not — is a peer bounce, not a socket swap.
                // That is a session boundary: inheriting the generation here
                // would swallow the bounce forever (no pg purge, no
                // noconnection delivery, no peer_creation change on any Up).
                let peer_bounced = !previous_down && creation_mismatch;
                let generation = if previous_down {
                    // HS-4 re-dial window: the incumbent went down but its own
                    // `connection_down` has not (or will not, having lost the
                    // ptr-eq race after this replacement) removed the entry.
                    // Close the old session HERE, under the same entry guard
                    // any competing emission site needs, so its Down is never
                    // lost and always precedes the new session's Up.
                    self.inner.events.enqueue(ConnectionEvent::down(
                        node,
                        previous.generation(),
                        // No unwrap: the fallback is reachable only when a test
                        // flips `down` directly without `mark_down` recording a
                        // reason.
                        previous
                            .down_reason
                            .get()
                            .copied()
                            .unwrap_or(ConnectionDownReason::ReadError),
                    ));
                    self.inner.events.next_generation(node)
                } else if peer_bounced {
                    // Close the old incarnation's session under the same entry
                    // guard the HS-4 arm uses, so the Down is never lost and
                    // always precedes the new session's Up. The stale link
                    // never reported a failure (no reason recorded), so the
                    // reason is ReadError — matching the retirement reason the
                    // displaced socket itself is marked down with below.
                    self.inner.events.enqueue(ConnectionEvent::down(
                        node,
                        previous.generation(),
                        ConnectionDownReason::ReadError,
                    ));
                    self.inner.events.next_generation(node)
                } else {
                    // Live same-incarnation displacement (simultaneous
                    // connect): same logical session, so the newcomer inherits
                    // the generation and no event fires.
                    previous.generation()
                };
                let (connection, read_half) = self.build_connection(
                    node,
                    peer_addr,
                    socket,
                    direction,
                    generation,
                    peer_creation,
                );
                occupied.insert(Arc::clone(&connection));
                if previous_down || peer_bounced {
                    self.inner
                        .events
                        .enqueue(ConnectionEvent::up(node, generation, peer_creation));
                }
                (connection, read_half, Some(previous))
            }
            Entry::Vacant(vacant) => {
                let generation = self.inner.events.next_generation(node);
                let (connection, read_half) = self.build_connection(
                    node,
                    peer_addr,
                    socket,
                    direction,
                    generation,
                    peer_creation,
                );
                let entry_ref = vacant.insert(Arc::clone(&connection));
                // Enqueue AFTER the table mutation, still under the entry
                // guard: no event can be delivered while the table does not
                // yet reflect it (INV-UP-VISIBILITY).
                self.inner
                    .events
                    .enqueue(ConnectionEvent::up(node, generation, peer_creation));
                drop(entry_ref);
                (connection, read_half, None)
            }
        };
        // Entry guard dropped: safe to re-enter the map. The displaced link's
        // `connection_down` ptr-eq guard sees the freshly inserted entry (not the
        // displaced one), so it does not evict the survivor (and enqueues no
        // event); it wakes the old link's read loop to drop its socket, and its
        // unconditional dispatch may deliver the events this install queued —
        // harmless: same thread, same order, still before the read lifecycle
        // spawns below.
        if let Some(previous) = displaced {
            previous.mark_down(ConnectionDownReason::ReadError);
        }
        // Deliver BEFORE the read lifecycle spawns: no generation-g inbound
        // frame can reach the control-frame handler before Up(g) delivery
        // completes (INV-FRAME-ORDER), and a queued prior Down's cleanup has
        // run before the new generation's read loop exists.
        self.inner.events.dispatch();
        self.spawn_read_lifecycle(Arc::clone(&installed), read_half);
        Ok(installed)
    }

    /// Split a stream into a [`DistConnection`] and its read half, without
    /// touching the connection table. Shared by both `register_connection` arms.
    fn build_connection(
        &self,
        node: Atom,
        peer_addr: SocketAddr,
        socket: PreparedSocket,
        direction: LinkDirection,
        generation: ConnectionGeneration,
        peer_creation: u32,
    ) -> (Arc<DistConnection>, OwnedReadHalf) {
        let (connection, read_half) = DistConnection::new(
            node,
            peer_addr,
            socket,
            Arc::downgrade(&self.inner),
            direction,
            generation,
            peer_creation,
        );
        (Arc::new(connection), read_half)
    }

    /// Register a pre-connected standard stream for native BIF unit tests.
    ///
    /// `peer_creation` is 0, the documented "no handshake" sentinel: this
    /// helper skips the handshake, so there is no peer incarnation to surface
    /// (and the peer-bounce discriminator never fires on the sentinel).
    #[cfg(test)]
    pub(crate) fn register_test_connection(
        &self,
        node: Atom,
        peer_addr: SocketAddr,
        stream: std::net::TcpStream,
    ) -> io::Result<Arc<DistConnection>> {
        self.register_test_connection_with_creation(node, peer_addr, stream, 0)
    }

    /// [`Self::register_test_connection`] with an explicit `peer_creation`,
    /// for tests exercising the peer-bounce (creation-mismatch) install arm.
    #[cfg(test)]
    pub(crate) fn register_test_connection_with_creation(
        &self,
        node: Atom,
        peer_addr: SocketAddr,
        stream: std::net::TcpStream,
        peer_creation: u32,
    ) -> io::Result<Arc<DistConnection>> {
        stream.set_nonblocking(true)?;
        let stream = TcpStream::from_std(stream)?;
        // Test helper: a pre-connected stream, no handshake, so the direction
        // only selects the install arm; `Inbound` here.
        self.register_connection(
            node,
            peer_addr,
            stream,
            LinkDirection::Inbound,
            peer_creation,
        )
    }

    fn spawn_read_lifecycle(&self, connection: Arc<DistConnection>, mut read_half: OwnedReadHalf) {
        // A fresh link is live now; seed its inbound clock and start its net-tick.
        connection.note_inbound_activity();
        self.spawn_heartbeat(Arc::clone(&connection));
        let manager = Arc::clone(&self.inner);
        let shutdown = Arc::clone(&connection.shutdown);
        self.inner.spawn_lifecycle(async move {
            // A single long-lived `Notified` future, re-polled via `&mut` each
            // iteration so `notify_waiters` (which wakes only already-registered
            // waiters) is never missed mid-loop. `enable()` registers the waiter
            // NOW rather than on first poll inside the select below — otherwise a
            // `notify_waiters` racing the first iteration (after the `is_down`
            // check, before the first poll) would be lost and the read loop would
            // park until peer EOF instead of dropping a displaced link promptly.
            let notified = shutdown.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            loop {
                let mut header = [0_u8; 8];
                // Race the header read against a shutdown so a retired link (e.g.
                // displaced by a simultaneous-connect canonical winner) drops its
                // read half promptly instead of parking until the peer closes.
                if connection.is_down() {
                    break;
                }
                let read_header = tokio::select! {
                    biased;
                    () = &mut notified => break,
                    result = read_half.read_exact(&mut header) => result,
                };
                match read_header {
                    Ok(_) => {
                        // Any inbound bytes (data frame OR keepalive) refresh the
                        // net-tick liveness clock for this link.
                        connection.note_inbound_activity();
                        let control_len =
                            u32::from_be_bytes([header[0], header[1], header[2], header[3]])
                                as usize;
                        let payload_len =
                            u32::from_be_bytes([header[4], header[5], header[6], header[7]])
                                as usize;
                        let Some(total_len) = control_len.checked_add(payload_len) else {
                            connection.mark_down(ConnectionDownReason::ReadError);
                            break;
                        };
                        let mut frame = vec![0_u8; total_len];
                        if read_half.read_exact(&mut frame).await.is_err() {
                            connection.mark_down(ConnectionDownReason::ReadError);
                            break;
                        }
                        let handler = manager
                            .control_frame_handler
                            .read()
                            .unwrap_or_else(|error| error.into_inner())
                            .clone();
                        if let Some(handler) = handler {
                            let (control, payload) = frame.split_at(control_len);
                            handler(connection.node, control, payload);
                        }
                    }
                    // `read_exact` never returns `Ok(0)`: EOF surfaces as an
                    // `UnexpectedEof` error. At the header read — the frame
                    // boundary — that is the peer closing its side (FIN), not
                    // a read fault, so it maps to `PeerClosed`, keeping that
                    // variant's documented meaning reachable. (EOF mid-header
                    // is indistinguishable here and also maps to `PeerClosed`;
                    // either way the peer's side of the socket is gone.)
                    Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => {
                        connection.mark_down(ConnectionDownReason::PeerClosed);
                        break;
                    }
                    Err(_) => {
                        connection.mark_down(ConnectionDownReason::ReadError);
                        break;
                    }
                }
            }
        });
    }

    /// Spawn the proactive net-tick for `connection` when heartbeats are enabled.
    ///
    /// Every `interval` the task: (1) writes a [`KEEPALIVE_FRAME`] (via
    /// `write_raw`, which itself marks the link down on a write error), and (2)
    /// marks the link down via the existing connection-down path if no inbound
    /// bytes have arrived within `deadline` — catching a silently-partitioned
    /// peer that never sends a FIN/RST. The task exits once the connection is
    /// down (whether from the heartbeat, a read error, or a manual disconnect),
    /// so it does not outlive the link. No-op when heartbeats are disabled.
    fn spawn_heartbeat(&self, connection: Arc<DistConnection>) {
        let Some(config) = self.inner.heartbeat else {
            return;
        };
        self.inner
            .heartbeat_tasks_spawned
            .fetch_add(1, Ordering::Relaxed);
        self.inner.spawn_lifecycle(async move {
            let mut ticker = tokio::time::interval(config.interval);
            // The first tick fires immediately; skip it so the seeded inbound
            // clock is never compared against a zero-elapsed deadline.
            ticker.tick().await;
            loop {
                ticker.tick().await;
                if connection.is_down() {
                    break;
                }
                if connection.inbound_idle_for(config.deadline) {
                    connection.mark_down_heartbeat_timeout();
                    break;
                }
                // Best-effort keepalive: a write error already drives mark_down
                // inside write_raw, so a failure here simply ends the task on the
                // next is_down() check.
                let _ = connection.write_raw(&KEEPALIVE_FRAME).await;
            }
        });
    }

    async fn accept_loop(&self, listener: TcpListener, shutdown: Arc<Notify>) {
        loop {
            tokio::select! {
                _ = shutdown.notified() => {
                    break;
                }
                accepted = listener.accept() => {
                    let Ok((stream, peer_addr)) = accepted else {
                        continue;
                    };
                    self.handle_accepted(stream, peer_addr);
                }
            }
        }
    }

    /// Run the inbound OTP handshake on an accepted stream, then register it.
    ///
    /// The handshake is asynchronous, so it is spawned onto the bound runtime via
    /// [`ConnectionManagerInner::spawn_lifecycle`] — the same mechanism the
    /// read/accept lifecycle uses — so it is driven even in production where no
    /// ambient tokio runtime exists on worker threads. The handshake completes on
    /// the raw stream (2-byte length-prefixed packets) before the connection is
    /// registered and its data-frame read loop starts. On success the connection
    /// is keyed by the peer's authenticated handshake name; on failure the stream
    /// is dropped, closing the TCP connection.
    fn handle_accepted(&self, mut stream: TcpStream, peer_addr: SocketAddr) {
        let manager = self.clone();
        self.inner.spawn_lifecycle(async move {
            let local = match manager.inner.handshake_node() {
                Ok(local) => local,
                Err(_) => return,
            };
            // Bound the responder so a stalled or malicious peer can never park
            // this spawned task forever; on elapse the stream is dropped, closing
            // the TCP connection (HS-1). The decider resolves a simultaneous
            // connect by the name-comparison tie-break against the local outbound
            // state (HS-3); on `nok` the responder aborts and the reciprocal
            // outbound is the survivor.
            let decider = |peer_name: &str| manager.inner.decide_inbound_status(peer_name);
            let outcome = tokio::time::timeout(
                manager.inner.handshake_timeout,
                respond_handshake_async_with(
                    &mut stream,
                    &local,
                    &manager.inner.cookie,
                    manager.inner.gen_challenge(),
                    decider,
                ),
            )
            .await;
            match outcome {
                Ok(Ok(result)) => {
                    let node = manager.inner.atom_table.intern(result.remote_name());
                    // A refused install (teardown-dup failure under fd
                    // exhaustion) drops the stream: the peer sees EOF and may
                    // redial once descriptors free up.
                    let _ = manager.register_connection(
                        node,
                        peer_addr,
                        stream,
                        LinkDirection::Inbound,
                        result.remote_creation(),
                    );
                }
                Ok(Err(_)) | Err(_) => {
                    drop(stream);
                }
            }
        });
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Barrier, mpsc};
    use std::thread;
    use std::time::Instant;

    use tokio::net::TcpListener;
    use tokio::runtime::Builder;
    use tokio::task::JoinHandle;

    use super::*;
    use crate::distribution::handshake::HandshakeNode;
    use crate::distribution::resolver::StaticResolver;

    const TEST_COOKIE: &str = "test-cookie";

    fn manager_with_resolver(resolver: Arc<StaticResolver>) -> ConnectionManager {
        ConnectionManager::new(
            Arc::new(AtomTable::with_common_atoms()),
            resolver,
            TEST_COOKIE,
            "local@127.0.0.1",
            1,
        )
    }

    /// A connection manager with the proactive net-tick enabled at test-scale
    /// timings: a short interval and a deadline a few intervals long so a
    /// silently-partitioned peer is detected within a bounded test window while a
    /// healthy peer's keepalives still refresh liveness in time.
    fn manager_with_heartbeat(
        resolver: Arc<StaticResolver>,
        interval: Duration,
        deadline: Duration,
    ) -> ConnectionManager {
        ConnectionManager::new(
            Arc::new(AtomTable::with_common_atoms()),
            resolver,
            TEST_COOKIE,
            "local@127.0.0.1",
            1,
        )
        .with_heartbeat(HeartbeatConfig { interval, deadline })
    }

    /// Accept a single inbound stream on `listener` and respond to the OTP
    /// handshake advertising `name`, mirroring a real peer's accept side so the
    /// outbound `connect` under test can complete its handshake.
    fn spawn_responder(
        listener: TcpListener,
        name: &'static str,
        cookie: &'static str,
    ) -> JoinHandle<()> {
        tokio::spawn(async move {
            let Ok((mut stream, _peer)) = listener.accept().await else {
                return;
            };
            let local = HandshakeNode::with_default_flags(name, 7)
                .expect("responder node name should be valid");
            let _ = crate::distribution::handshake::respond_handshake_async(
                &mut stream,
                &local,
                cookie,
                99,
            )
            .await;
            // Keep the accepted stream alive so the connection is not torn down
            // while the test inspects the outbound side.
            tokio::time::sleep(Duration::from_millis(200)).await;
        })
    }

    /// Accept one inbound stream, complete the handshake advertising `name`, and
    /// hand the accepted (still-open) stream back to the caller so a test can
    /// later drop it to simulate the peer going away after a successful link.
    fn spawn_responder_handoff(
        listener: TcpListener,
        name: &'static str,
    ) -> tokio::sync::oneshot::Receiver<TcpStream> {
        let (sender, receiver) = tokio::sync::oneshot::channel();
        tokio::spawn(async move {
            let Ok((mut stream, _peer)) = listener.accept().await else {
                return;
            };
            let local = HandshakeNode::with_default_flags(name, 7)
                .expect("responder node name should be valid");
            if crate::distribution::handshake::respond_handshake_async(
                &mut stream,
                &local,
                TEST_COOKIE,
                99,
            )
            .await
            .is_ok()
            {
                let _ = sender.send(stream);
            }
        });
        receiver
    }

    #[tokio::test]
    async fn empty_manager_has_no_connections() {
        let manager = manager_with_resolver(Arc::new(StaticResolver::new(
            std::collections::HashMap::new(),
        )));
        let node = manager.inner.atom_table.intern("missing@127.0.0.1");

        assert_eq!(manager.connection_count(), 0);
        assert!(manager.get_connection(node).is_none());
    }

    #[tokio::test]
    async fn outbound_connect_inserts_table_entry() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap_or_else(|error| {
                panic!("failed to bind local listener: {error}");
            });
        let addr = listener.local_addr().unwrap_or_else(|error| {
            panic!("failed to inspect local listener: {error}");
        });
        let _responder = spawn_responder(listener, "remote@127.0.0.1", TEST_COOKIE);

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let connection = manager
            .connect("remote@127.0.0.1")
            .await
            .unwrap_or_else(|error| panic!("connect failed: {error}"));
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");

        assert!(Arc::ptr_eq(
            &connection,
            &manager
                .get_connection(node)
                .expect("connection should be present"),
        ));
    }

    #[tokio::test]
    async fn connect_keys_table_by_remote_handshake_name_not_resolver_key() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap_or_else(|error| panic!("failed to bind local listener: {error}"));
        let addr = listener
            .local_addr()
            .unwrap_or_else(|error| panic!("failed to inspect local listener: {error}"));
        // The peer advertises a DIFFERENT name than the resolver key the dialer
        // used, proving identity comes from the authenticated handshake.
        let _responder = spawn_responder(listener, "advertised@127.0.0.1", TEST_COOKIE);

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "dialed@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let connection = manager
            .connect("dialed@127.0.0.1")
            .await
            .unwrap_or_else(|error| panic!("connect failed: {error}"));

        let advertised = manager.inner.atom_table.intern("advertised@127.0.0.1");
        let dialed = manager.inner.atom_table.intern("dialed@127.0.0.1");
        assert_eq!(connection.node(), advertised);
        assert!(manager.get_connection(advertised).is_some());
        assert!(
            manager.get_connection(dialed).is_none(),
            "connection must not be keyed by the resolver key"
        );
    }

    #[tokio::test]
    async fn connect_rejects_wrong_cookie_and_records_no_entry() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap_or_else(|error| panic!("failed to bind local listener: {error}"));
        let addr = listener
            .local_addr()
            .unwrap_or_else(|error| panic!("failed to inspect local listener: {error}"));
        // Responder uses a different cookie, so the handshake digest mismatches.
        let _responder = spawn_responder(listener, "remote@127.0.0.1", "other-cookie");

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let result = manager.connect("remote@127.0.0.1").await;

        assert!(
            matches!(result, Err(ConnectError::Io(_))),
            "connect must fail with Io on cookie mismatch"
        );
        assert_eq!(manager.connection_count(), 0);
        let remote = manager.inner.atom_table.intern("remote@127.0.0.1");
        assert!(manager.get_connection(remote).is_none());
    }

    #[tokio::test]
    async fn inbound_wrong_cookie_registers_no_entry() {
        // A listening manager authenticates with TEST_COOKIE. A peer that
        // initiates the handshake with a DIFFERENT cookie must be rejected by
        // the register-side accept loop (the `handle_accepted` Err -> drop arm)
        // and must NOT receive a connection-table entry.
        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::new()));
        let manager = manager_with_resolver(resolver);
        let accept = manager
            .listen("127.0.0.1:0".parse().unwrap_or_else(|error| {
                panic!("failed to parse listen address: {error}");
            }))
            .await
            .unwrap_or_else(|error| panic!("failed to start accept loop: {error}"));

        let mut client = TcpStream::connect(accept.local_addr())
            .await
            .unwrap_or_else(|error| panic!("failed to open inbound stream: {error}"));
        let client_node = HandshakeNode::with_default_flags("client@127.0.0.1", 5)
            .expect("client node name should be valid");
        // The client uses the WRONG cookie, so the digest mismatches and the
        // listening manager's responder rejects the handshake.
        let result = crate::distribution::handshake::initiate_handshake_async(
            &mut client,
            &client_node,
            "wrong-cookie",
            42,
        )
        .await;
        assert!(
            result.is_err(),
            "inbound handshake with wrong cookie must fail"
        );

        // The inbound handshake runs on a spawned task, so poll (rather than a
        // fixed sleep) to confirm the rejection never produces a table entry.
        let node = manager.inner.atom_table.intern("client@127.0.0.1");
        for _ in 0..40 {
            assert_eq!(
                manager.connection_count(),
                0,
                "wrong-cookie peer must never register a connection"
            );
            assert!(
                manager.get_connection(node).is_none(),
                "wrong-cookie peer must not appear in the connection table"
            );
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        drop(client);
    }

    /// HS-1: an outbound `connect` to a peer that accepts the TCP connection but
    /// never speaks the handshake must return a handshake-timeout error within the
    /// configured handshake deadline, not hang. This is the bounded-return
    /// contract that lets the haematite-side retry above the seam make progress.
    #[tokio::test]
    async fn connect_returns_timeout_when_peer_never_handshakes() {
        // A bare listener that accepts then stays silent (no responder).
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap_or_else(|error| panic!("failed to bind local listener: {error}"));
        let addr = listener
            .local_addr()
            .unwrap_or_else(|error| panic!("failed to inspect local listener: {error}"));
        let _silent_accept = tokio::spawn(async move {
            // Accept and hold the stream open without ever writing a handshake byte.
            if let Ok((stream, _peer)) = listener.accept().await {
                tokio::time::sleep(Duration::from_secs(30)).await;
                drop(stream);
            }
        });

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "silent@127.0.0.1".to_string(),
            addr,
        )])));
        let manager =
            manager_with_resolver(resolver).with_handshake_timeout(Duration::from_secs(1));

        let started = std::time::Instant::now();
        let result =
            tokio::time::timeout(Duration::from_secs(15), manager.connect("silent@127.0.0.1"))
                .await;

        let outcome = result
            .expect("connect must return within the handshake deadline, not hang")
            .map(|_connection| ());
        assert!(
            matches!(outcome, Err(ConnectError::Io(_))),
            "a non-speaking peer must surface as a connect error, got {outcome:?}"
        );
        assert!(
            started.elapsed() < Duration::from_secs(10),
            "connect should return near the 1s handshake deadline, took {:?}",
            started.elapsed()
        );
        assert_eq!(manager.connection_count(), 0);
    }

    #[tokio::test]
    async fn connect_node_is_idempotent_and_lists_node() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap_or_else(|error| panic!("failed to bind local listener: {error}"));
        let addr = listener
            .local_addr()
            .unwrap_or_else(|error| panic!("failed to inspect local listener: {error}"));
        let _responder = spawn_responder(listener, "remote@127.0.0.1", TEST_COOKIE);

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");

        assert!(manager.connect_node(node).await);
        assert!(manager.connect_node(node).await);
        assert_eq!(manager.connected_nodes(), vec![node]);
        assert_eq!(manager.connection_count(), 1);
    }

    #[tokio::test]
    async fn connect_node_returns_false_for_unresolved_node() {
        let manager = manager_with_resolver(Arc::new(StaticResolver::new(
            std::collections::HashMap::new(),
        )));
        let node = manager.inner.atom_table.intern("missing@127.0.0.1");

        assert!(!manager.connect_node(node).await);
        assert!(manager.connected_nodes().is_empty());
    }

    #[tokio::test]
    async fn inbound_peer_registers_under_its_handshake_name() {
        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::new()));
        let manager = manager_with_resolver(resolver);
        let accept = manager
            .listen("127.0.0.1:0".parse().unwrap_or_else(|error| {
                panic!("failed to parse listen address: {error}");
            }))
            .await
            .unwrap_or_else(|error| panic!("failed to start accept loop: {error}"));

        // The inbound peer initiates the handshake advertising "client@127.0.0.1".
        // The manager must register it under that authenticated name with NO
        // address-identity seam.
        let mut client = TcpStream::connect(accept.local_addr())
            .await
            .unwrap_or_else(|error| panic!("failed to open inbound stream: {error}"));
        let client_node = HandshakeNode::with_default_flags("client@127.0.0.1", 5)
            .expect("client node name should be valid");
        crate::distribution::handshake::initiate_handshake_async(
            &mut client,
            &client_node,
            TEST_COOKIE,
            42,
        )
        .await
        .expect("inbound peer handshake should succeed");

        let node = manager.inner.atom_table.intern("client@127.0.0.1");
        let mut connected = false;
        for _ in 0..40 {
            if manager.get_connection(node).is_some() {
                connected = true;
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert!(
            connected,
            "inbound peer should register under its handshake name"
        );
        assert_eq!(manager.connected_nodes(), vec![node]);
        drop(client);
    }

    #[tokio::test]
    async fn dropping_peer_removes_connection_and_notifies_once() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap_or_else(|error| {
                panic!("failed to bind local listener: {error}");
            });
        let addr = listener.local_addr().unwrap_or_else(|error| {
            panic!("failed to inspect local listener: {error}");
        });
        let remote_stream = spawn_responder_handoff(listener, "remote@127.0.0.1");

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let callback_count = Arc::new(AtomicUsize::new(0));
        let callback_count_for_hook = Arc::clone(&callback_count);
        manager.register_connection_down(move |_| {
            callback_count_for_hook.fetch_add(1, Ordering::SeqCst);
        });
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");
        let _connection = manager
            .connect("remote@127.0.0.1")
            .await
            .unwrap_or_else(|error| panic!("connect failed: {error}"));

        let remote_stream = remote_stream
            .await
            .expect("responder did not complete handshake");
        drop(remote_stream);
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert!(manager.get_connection(node).is_none());
        assert_eq!(callback_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn manual_disconnect_removes_connection_and_notifies_once() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap_or_else(|error| panic!("failed to bind local listener: {error}"));
        let addr = listener
            .local_addr()
            .unwrap_or_else(|error| panic!("failed to inspect local listener: {error}"));
        let _responder = spawn_responder(listener, "remote@127.0.0.1", TEST_COOKIE);

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let callback_count = Arc::new(AtomicUsize::new(0));
        let callback_count_for_hook = Arc::clone(&callback_count);
        manager.register_connection_down(move |event| {
            assert_eq!(event.reason, ConnectionDownReason::ManualDisconnect);
            callback_count_for_hook.fetch_add(1, Ordering::SeqCst);
        });
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");

        assert!(manager.connect_node(node).await);
        assert!(manager.disconnect_node(node));
        assert!(manager.disconnect_node(node));

        assert!(manager.get_connection(node).is_none());
        assert!(manager.connected_nodes().is_empty());
        assert_eq!(callback_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn write_error_removes_connection_and_notifies_once() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap_or_else(|error| {
                panic!("failed to bind local listener: {error}");
            });
        let addr = listener.local_addr().unwrap_or_else(|error| {
            panic!("failed to inspect local listener: {error}");
        });
        let remote_stream = spawn_responder_handoff(listener, "remote@127.0.0.1");

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let callback_count = Arc::new(AtomicUsize::new(0));
        let callback_count_for_hook = Arc::clone(&callback_count);
        manager.register_connection_down(move |_| {
            callback_count_for_hook.fetch_add(1, Ordering::SeqCst);
        });
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");
        let connection = manager
            .connect("remote@127.0.0.1")
            .await
            .unwrap_or_else(|error| panic!("connect failed: {error}"));

        let remote_stream = remote_stream
            .await
            .expect("responder did not complete handshake");
        drop(remote_stream);

        for _ in 0..8 {
            if connection.write_raw(b"probe").await.is_err() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;

        assert!(manager.get_connection(node).is_none());
        assert_eq!(callback_count.load(Ordering::SeqCst), 1);
    }

    /// HS-3 (tie-break direction, D1): with a competing local outbound recorded,
    /// the responder emits `nok` when the local name is greater than the peer's,
    /// and `ok_simultaneous` when the peer's name is greater — matching OTP's
    /// literal name comparison. Drives the real accept loop so it exercises the
    /// production decider (`decide_inbound_status`), forcing the `connecting`
    /// marker that signals an in-flight outbound.
    #[tokio::test]
    async fn hs3_responder_rejects_when_local_name_is_greater() {
        // local = "zeta..." (greater); inbound peer = "alpha..." (lesser).
        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::new()));
        let manager = ConnectionManager::new(
            Arc::new(AtomTable::with_common_atoms()),
            resolver,
            TEST_COOKIE,
            "zeta@127.0.0.1",
            1,
        );
        let accept = manager
            .listen("127.0.0.1:0".parse().expect("parse listen addr"))
            .await
            .expect("start accept loop");

        // Simulate an in-flight local outbound to the inbound peer's name so the
        // decider sees the simultaneous case.
        let peer_atom = manager.inner.atom_table.intern("alpha@127.0.0.1");
        manager
            .inner
            .connecting
            .insert(peer_atom, Arc::new(AtomicBool::new(false)));

        let mut client = TcpStream::connect(accept.local_addr())
            .await
            .expect("inbound peer connects");
        let client_node = HandshakeNode::with_default_flags("alpha@127.0.0.1", 5)
            .expect("client node name valid");
        let result = crate::distribution::handshake::initiate_handshake_async(
            &mut client,
            &client_node,
            TEST_COOKIE,
            42,
        )
        .await;

        // local(zeta) > peer(alpha) => responder sends `nok`, the initiator sees
        // a BadStatus("nok") abort, and no inbound link is registered.
        assert_eq!(
            result.expect_err("initiator must see the nok rejection"),
            HandshakeError::BadStatus("nok".into())
        );
        assert!(
            manager.get_connection(peer_atom).is_none(),
            "a rejected inbound must not register a connection"
        );
        drop(accept);
    }

    /// Spawn a one-shot responder on `listener` that always answers the status
    /// step with `nok`, modelling a peer that keeps its reciprocal outbound.
    fn spawn_nok_responder(listener: TcpListener) -> JoinHandle<()> {
        tokio::spawn(async move {
            let Ok((mut stream, _peer)) = listener.accept().await else {
                return;
            };
            let local = HandshakeNode::with_default_flags("peer@127.0.0.1", 9)
                .expect("responder node valid");
            let _ = crate::distribution::handshake::respond_handshake_async_with(
                &mut stream,
                &local,
                TEST_COOKIE,
                3,
                |_peer_name| SimultaneousDecision::Reject,
            )
            .await;
        })
    }

    /// HS-3 (benign abort): an outbound `connect` that receives `nok` returns
    /// `ConnectError::SimultaneousAbort` (not an Io failure), and `connect_node`
    /// folds that into success so the caller does not retry-storm.
    #[tokio::test]
    async fn hs3_outbound_nok_is_a_benign_simultaneous_abort() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind nok responder");
        let addr = listener.local_addr().expect("inspect listener");
        let _responder = spawn_nok_responder(listener);

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "peer@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);

        let outcome = manager.connect("peer@127.0.0.1").await.map(|_| ());
        assert!(
            matches!(outcome, Err(ConnectError::SimultaneousAbort)),
            "nok must surface as a benign SimultaneousAbort, got {outcome:?}"
        );
        assert_eq!(manager.connection_count(), 0);
    }

    /// HS-3: `connect_node` treats a `nok` simultaneous abort as success.
    #[tokio::test]
    async fn hs3_connect_node_treats_nok_abort_as_success() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind nok responder");
        let addr = listener.local_addr().expect("inspect listener");
        let _responder = spawn_nok_responder(listener);

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "peer@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let node = manager.inner.atom_table.intern("peer@127.0.0.1");

        assert!(
            manager.connect_node(node).await,
            "connect_node must treat a nok abort as success (no retry-storm)"
        );
        // The abort registers no connection; the reciprocal inbound is the link.
        assert_eq!(manager.connection_count(), 0);
    }

    /// HS-2: two simultaneous installs for the same peer name must leave exactly
    /// one live link — the CANONICAL-direction one — regardless of arrival order,
    /// and the loser's socket must be closed (no orphan reader on a half-link).
    ///
    /// `local@` < `peer@`, so the canonical survivor is this node's INBOUND
    /// (the higher-named peer's outbound). The non-canonical OUTBOUND half is
    /// installed FIRST here; the canonical inbound must then displace it — proving
    /// the survivor is chosen by name comparison, not by who registered first.
    /// Both nodes apply the same rule, so the same single TCP socket survives on
    /// both ends and a pair can never lose both links.
    #[tokio::test]
    async fn hs2_two_simultaneous_installs_keep_exactly_one_no_orphan_reader() {
        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::new()));
        let manager = manager_with_resolver(resolver);
        let node = manager.inner.atom_table.intern("peer@127.0.0.1");

        // Two independent connected socket pairs standing in for the inbound and
        // outbound halves of a simultaneous connect. The client ends let us
        // observe whether each server end stays open or is closed.
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind helper listener");
        let addr = listener.local_addr().expect("inspect helper listener");

        // First (non-canonical) install: the OUTBOUND half.
        let mut client_outbound = TcpStream::connect(addr)
            .await
            .expect("client_outbound connects");
        let (server_outbound, _) = listener.accept().await.expect("accept server_outbound");
        // Second (canonical) install: the INBOUND half, which must win.
        let mut client_inbound = TcpStream::connect(addr)
            .await
            .expect("client_inbound connects");
        let (server_inbound, _) = listener.accept().await.expect("accept server_inbound");

        let displaced = manager
            .register_connection(node, addr, server_outbound, LinkDirection::Outbound, 0)
            .expect("outbound installs");
        let winner = manager
            .register_connection(node, addr, server_inbound, LinkDirection::Inbound, 0)
            .expect("inbound installs");

        // Exactly one table entry, and it is the canonical (inbound) winner, not
        // the first-installed outbound.
        assert_eq!(manager.connection_count(), 1);
        assert!(
            !Arc::ptr_eq(&winner, &displaced),
            "the canonical inbound must displace the non-canonical outbound"
        );
        assert!(Arc::ptr_eq(
            &winner,
            &manager
                .get_connection(node)
                .expect("survivor must be in the table"),
        ));

        // The winner's socket stays open: a write reaches its peer.
        winner
            .write_raw(&[0_u8; 8])
            .await
            .expect("winner link must remain writable");
        let mut header = [0_u8; 8];
        client_inbound
            .read_exact(&mut header)
            .await
            .expect("winner's peer must receive the keepalive frame");

        // The displaced link's read half was torn down (no orphan reader). Drop
        // the last `DistConnection` Arc so its write half also closes, then the
        // peer observes EOF rather than a live, orphaned half-link.
        drop(displaced);
        let mut byte = [0_u8; 1];
        let eof = tokio::time::timeout(Duration::from_secs(5), client_outbound.read(&mut byte))
            .await
            .expect("displaced socket should close promptly, not hang")
            .expect("reading the closed displaced socket should not error");
        assert_eq!(eof, 0, "the displaced link's socket must be closed (EOF)");
    }

    /// Accept inbound streams on `listener` in a loop, completing the OTP
    /// handshake advertising `name` for each, and hand every accepted (still-open)
    /// stream back over the returned channel. Unlike [`spawn_responder_handoff`]
    /// (single accept), this models a real peer that stays up across a re-dial: the
    /// test can drop the first handed-back stream to simulate the link dropping,
    /// then receive the second stream produced by the reconnect's fresh inbound.
    fn spawn_multi_responder_handoff(
        listener: TcpListener,
        name: &'static str,
    ) -> tokio::sync::mpsc::UnboundedReceiver<TcpStream> {
        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
        tokio::spawn(async move {
            loop {
                let Ok((mut stream, _peer)) = listener.accept().await else {
                    return;
                };
                let local = HandshakeNode::with_default_flags(name, 7)
                    .expect("responder node name should be valid");
                if crate::distribution::handshake::respond_handshake_async(
                    &mut stream,
                    &local,
                    TEST_COOKIE,
                    99,
                )
                .await
                .is_ok()
                {
                    if sender.send(stream).is_err() {
                        return;
                    }
                } else {
                    return;
                }
            }
        });
        receiver
    }

    /// HS-4: after a distribution link drops (peer closed), a fresh `connect`
    /// re-establishes the link, the stale table entry is replaced by a NEW
    /// connection (not the dead one), and the new link is writable end-to-end. This
    /// is the core reconnection-hardening contract: a dropped link can be re-dialed
    /// deterministically and the result is a whole, usable link.
    #[tokio::test]
    async fn hs4_redial_after_drop_reestablishes_writable_link() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind responder listener");
        let addr = listener.local_addr().expect("inspect listener");
        let mut streams = spawn_multi_responder_handoff(listener, "remote@127.0.0.1");

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");

        // First link.
        let first = manager
            .connect("remote@127.0.0.1")
            .await
            .expect("first connect should succeed");
        let first_remote = streams.recv().await.expect("first inbound handed back");

        // Drop the peer's side; our read loop observes EOF and reaps the entry.
        drop(first_remote);
        let deadline = Instant::now() + Duration::from_secs(5);
        while manager.get_connection(node).is_some() {
            assert!(Instant::now() < deadline, "dropped link was never reaped");
            tokio::time::sleep(Duration::from_millis(10)).await;
        }

        // Re-dial: a clean re-establish, NOT a return of the dead connection.
        let second = manager
            .connect("remote@127.0.0.1")
            .await
            .expect("re-dial after drop should succeed");
        let mut second_remote = streams.recv().await.expect("second inbound handed back");

        assert!(
            !Arc::ptr_eq(&first, &second),
            "re-dial must install a NEW connection, not resurrect the dead one"
        );
        assert!(first.is_down(), "the first link must be marked down");
        assert!(!second.is_down(), "the re-dialed link must be live");
        assert_eq!(manager.connection_count(), 1, "exactly one live link");
        assert!(Arc::ptr_eq(
            &second,
            &manager
                .get_connection(node)
                .expect("re-dialed link must be in the table"),
        ));

        // The new link is writable end-to-end: an 8-byte zero header reaches the
        // peer's (re-dialed) inbound socket.
        second
            .write_raw(&[0_u8; 8])
            .await
            .expect("re-dialed link must be writable");
        let mut header = [0_u8; 8];
        tokio::time::timeout(
            Duration::from_secs(5),
            second_remote.read_exact(&mut header),
        )
        .await
        .expect("re-dialed peer must receive the frame, not hang")
        .expect("re-dialed peer read must not error");

        // The `connecting` guard cleared on every dial: no stuck in-flight marker.
        assert_eq!(
            manager.inner.connecting.len(),
            0,
            "re-dial must not leak the connecting guard"
        );
    }

    /// HS-4: `connect_node` must not report a DOWN-but-not-yet-reaped link as
    /// connected. If it did, a caller's reconnect would be told the peer is up and
    /// would never re-dial. With a stale down entry still in the table,
    /// `connect_node` must run a fresh handshake and replace it.
    #[tokio::test]
    async fn hs4_connect_node_redials_a_down_but_unreaped_entry() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind responder listener");
        let addr = listener.local_addr().expect("inspect listener");
        let mut streams = spawn_multi_responder_handoff(listener, "remote@127.0.0.1");

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");

        let first = manager
            .connect("remote@127.0.0.1")
            .await
            .expect("first connect should succeed");
        let _first_remote = streams.recv().await.expect("first inbound handed back");

        // Flip the link to down WITHOUT removing it from the table: this is the
        // narrow race window between `mark_down` and `connection_down`'s reap. We
        // reproduce it deterministically by holding a down entry in place.
        first.down.store(true, Ordering::Release);
        assert!(
            manager.get_connection(node).is_some(),
            "the stale down entry is still in the table"
        );

        // connect_node must NOT short-circuit on the down entry; it must re-dial.
        assert!(
            manager.connect_node(node).await,
            "connect_node must re-dial a down-but-unreaped entry"
        );
        let _second_remote = streams.recv().await.expect("re-dial inbound handed back");

        assert_eq!(manager.connection_count(), 1, "exactly one live link");
        let live = manager
            .get_connection(node)
            .expect("re-dialed link present");
        assert!(!live.is_down(), "the table now holds a live re-dialed link");
        assert!(
            !Arc::ptr_eq(&first, &live),
            "the dead entry must have been replaced, not reused"
        );
        assert_eq!(manager.inner.connecting.len(), 0);
    }

    /// HS-4: every `connect` exit path clears the `connecting` guard, so a series
    /// of dials (success, hard failure, and benign `nok` abort) never leaves a
    /// stuck in-flight marker that would corrupt the simultaneous-connect decider
    /// or block a future re-dial.
    #[tokio::test]
    async fn hs4_connecting_guard_clears_on_every_exit_path() {
        // Success path.
        let ok_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind ok");
        let ok_addr = ok_listener.local_addr().expect("ok addr");
        let _ok = spawn_responder(ok_listener, "ok@127.0.0.1", TEST_COOKIE);

        // nok (benign abort) path.
        let nok_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind nok");
        let nok_addr = nok_listener.local_addr().expect("nok addr");
        let _nok = spawn_nok_responder(nok_listener);

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([
            ("ok@127.0.0.1".to_string(), ok_addr),
            ("peer@127.0.0.1".to_string(), nok_addr),
            // refused@ has no listener bound -> connection refused / io error path.
        ])));
        let manager = manager_with_resolver(resolver);

        // Success.
        manager
            .connect("ok@127.0.0.1")
            .await
            .expect("ok connect should succeed");
        assert_eq!(
            manager.inner.connecting.len(),
            0,
            "success path must clear the connecting guard"
        );

        // Benign nok abort.
        assert!(matches!(
            manager.connect("peer@127.0.0.1").await,
            Err(ConnectError::SimultaneousAbort)
        ));
        assert_eq!(
            manager.inner.connecting.len(),
            0,
            "nok abort path must clear the connecting guard"
        );

        // Hard failure: unresolved name never reaches the guard, but a resolvable
        // name with no listener exercises the TCP-connect failure exit with the
        // guard already armed. Bind then immediately drop a listener to free a
        // port that now refuses.
        let dead = TcpListener::bind("127.0.0.1:0").await.expect("bind dead");
        let dead_addr = dead.local_addr().expect("dead addr");
        drop(dead);
        let resolver2 = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "dead@127.0.0.1".to_string(),
            dead_addr,
        )])));
        let manager2 = manager_with_resolver(resolver2);
        let failed = manager2.connect("dead@127.0.0.1").await;
        assert!(failed.is_err(), "connect to a refused port must fail");
        assert_eq!(
            manager2.inner.connecting.len(),
            0,
            "TCP-failure path must clear the connecting guard"
        );
    }

    type Resolver = Arc<dyn NodeResolver + Send + Sync>;

    /// HS-0 (deterministic root-cause oracle): an inbound peer completes the TCP
    /// connect then sends nothing, so the accept-side responder's first read sits
    /// on an untimed `read_exact`. Pre-HS-1 that responder task never resolves and
    /// the silent peer's socket stays open forever — the canonical handshake hang
    /// that, multiplied across a `>=3`-node mesh of blocking dials, wedges a
    /// cluster. After HS-1 the responder hits the whole-handshake deadline, the
    /// server drops the stream, and the silent peer observes EOF.
    ///
    /// The oracle drives the REAL `ConnectionManager` accept loop (so it exercises
    /// the production timeout path, not a test-local wrapper) with a short
    /// handshake deadline, then reads the silent peer's socket under an inner
    /// bound. Pre-HS-1 the read never returns and the bound fires → failure,
    /// demonstrating the hang. Post-HS-1 the read returns EOF promptly → pass. A
    /// whole-test wall-clock watchdog guards against any hang escaping the bound.
    #[test]
    fn hs0_silent_peer_handshake_terminates_and_does_not_hang() {
        let (done_tx, done_rx) = mpsc::channel();
        let worker = thread::spawn(move || {
            run_silent_peer_scenario();
            let _ = done_tx.send(());
        });
        match done_rx.recv_timeout(Duration::from_secs(45)) {
            Ok(()) => worker.join().expect("HS-0 worker thread should not panic"),
            Err(_) => panic!(
                "HS-0 DEADLOCK: a silent peer's inbound handshake never terminated \
                 (untimed read parked the responder forever)"
            ),
        }
    }

    fn run_silent_peer_scenario() {
        let runtime = Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .expect("build handshake runtime");
        runtime.block_on(async {
            let resolver: Resolver = Arc::new(StaticResolver::new(HashMap::new()));
            // Short handshake deadline so the post-fix path resolves quickly; the
            // pre-fix path has no deadline at all and hangs regardless.
            let manager = ConnectionManager::new(
                Arc::new(AtomTable::with_common_atoms()),
                resolver,
                TEST_COOKIE,
                "server@127.0.0.1",
                1,
            )
            .with_handshake_timeout(Duration::from_secs(2));
            let accept = manager
                .listen("127.0.0.1:0".parse().expect("parse listen addr"))
                .await
                .expect("start accept loop");

            // Silent peer: connect, then never send a single byte. The accept loop
            // spawns a responder that blocks on the first handshake read.
            let mut silent = TcpStream::connect(accept.local_addr())
                .await
                .expect("silent peer connects");

            // Pre-HS-1 the responder never times out, so the server never closes
            // the socket and this read blocks forever (caught by the inner bound).
            // Post-HS-1 the responder hits the deadline, the server drops the
            // stream, and this read returns EOF (Ok(0)).
            let mut byte = [0_u8; 1];
            let read = tokio::time::timeout(Duration::from_secs(15), silent.read(&mut byte)).await;

            let read = read.expect(
                "silent peer's socket was never closed: the inbound responder \
                 parked on an untimed handshake read (HS-1 not in effect)",
            );
            assert_eq!(
                read.expect("reading the closed socket should not error"),
                0,
                "expected EOF after the responder timed out and dropped the stream"
            );

            // No connection should have been registered for the silent peer.
            assert_eq!(manager.connection_count(), 0);
            drop(accept);
        });
    }

    /// HS-0 (convergence): a 3-node full mesh, every node dialing its two peers
    /// simultaneously (barrier-released) from synchronous threads via
    /// `runtime.block_on` — the haematite seam. Each node's accept/responder
    /// tasks share its single worker. After HS-3 exactly one link survives per
    /// pair (no last-writer-wins clobber) and that link is usable in both
    /// directions. Pre-fix this can deadlock or leave mismatched half-links;
    /// run under a hard watchdog so a hang fails the test.
    #[test]
    fn hs0_three_node_simultaneous_dial_mesh_forms_without_deadlock() {
        let (done_tx, done_rx) = mpsc::channel();
        let worker = thread::spawn(move || {
            run_three_node_mesh();
            let _ = done_tx.send(());
        });
        match done_rx.recv_timeout(Duration::from_secs(30)) {
            Ok(()) => worker.join().expect("mesh worker thread should not panic"),
            Err(_) => panic!(
                "HS-0 DEADLOCK: 3-node simultaneous-dial mesh did not converge \
                 within the watchdog window (connect never returned)"
            ),
        }
    }

    fn run_three_node_mesh() {
        let names = ["alpha@127.0.0.1", "bravo@127.0.0.1", "charlie@127.0.0.1"];
        // Bind every listener first so the shared resolver maps all names.
        let mut prepared = Vec::new();
        let mut address_map = HashMap::new();
        for name in names {
            let runtime = Arc::new(
                Builder::new_multi_thread()
                    .worker_threads(1)
                    .enable_all()
                    .build()
                    .expect("build single-worker node runtime"),
            );
            let listener = runtime
                .block_on(TcpListener::bind("127.0.0.1:0"))
                .expect("bind node listener");
            address_map.insert(name.to_string(), listener.local_addr().expect("addr"));
            prepared.push((name, runtime, listener));
        }
        let resolver: Resolver = Arc::new(StaticResolver::new(address_map));

        let mut nodes = Vec::new();
        for (name, runtime, listener) in prepared {
            let manager = ConnectionManager::new(
                Arc::new(AtomTable::with_common_atoms()),
                Arc::clone(&resolver),
                TEST_COOKIE,
                name,
                1,
            );
            manager.set_runtime_handle(runtime.handle().clone());
            // Count control frames this node's read loops actually deliver. A
            // delivered frame proves the link is whole: the socket this node holds
            // for the peer is the same one the peer reads from. The pre-HS-2/3
            // last-writer-wins clobber can orphan one socket's reader, so a frame
            // written to the surviving write half is never observed here.
            let received = Arc::new(AtomicUsize::new(0));
            let received_for_handler = Arc::clone(&received);
            manager.register_control_frame_handler(move |_control, _payload| {
                received_for_handler.fetch_add(1, Ordering::SeqCst);
            });
            let accept = runtime.block_on(async { manager.listen_with(listener) });
            nodes.push((name, manager, runtime, accept, received));
        }

        // 3 nodes x 2 peers = 6 dialing threads, released together.
        let barrier = Arc::new(Barrier::new(6));
        let mut dialers = Vec::new();
        for (name, manager, runtime, _accept, _received) in &nodes {
            for peer in names {
                if peer == *name {
                    continue;
                }
                let manager = manager.clone();
                let runtime = Arc::clone(runtime);
                let barrier = Arc::clone(&barrier);
                let peer_name = peer.to_string();
                dialers.push(thread::spawn(move || {
                    barrier.wait();
                    let _ = runtime.block_on(manager.connect(&peer_name));
                }));
            }
        }
        for dialer in dialers {
            dialer
                .join()
                .expect("dialer thread should not panic (connect must return)");
        }

        // Exactly one link per pair on every node. Poll: the losing inbound may
        // still be tearing down when the winning `connect` returns.
        let deadline = Instant::now() + Duration::from_secs(10);
        loop {
            if nodes
                .iter()
                .all(|(_, manager, _, _, _)| manager.connection_count() == 2)
            {
                break;
            }
            assert!(
                Instant::now() < deadline,
                "mesh did not converge to one link per pair: counts = {:?}",
                nodes
                    .iter()
                    .map(|(_, manager, _, _, _)| manager.connection_count())
                    .collect::<Vec<_>>()
            );
            thread::sleep(Duration::from_millis(25));
        }

        // Every directed edge must carry a frame end-to-end. Each node writes one
        // 8-byte zero header (a zero-length control+payload frame) to each peer
        // link; each node must then OBSERVE the two frames its peers sent it. A
        // clobbered half-link silently drops the frame, so the receiver's count
        // stays below 2 and this fails — the deterministic pre-fix symptom.
        for (name, manager, runtime, _accept, _received) in &nodes {
            for peer in names {
                if peer == *name {
                    continue;
                }
                let peer_atom = manager.inner.atom_table.intern(peer);
                let connection = manager
                    .get_connection(peer_atom)
                    .unwrap_or_else(|| panic!("{name} has no link to {peer}"));
                runtime
                    .block_on(connection.write_raw(&[0_u8; 8]))
                    .unwrap_or_else(|error| {
                        panic!("{name} -> {peer} surviving link not writable: {error}")
                    });
            }
        }

        let deadline = Instant::now() + Duration::from_secs(10);
        loop {
            if nodes
                .iter()
                .all(|(_, _, _, _, received)| received.load(Ordering::SeqCst) >= 2)
            {
                break;
            }
            assert!(
                Instant::now() < deadline,
                "mesh links are not whole bidirectionally: per-node received \
                 frame counts = {:?} (expected >= 2 each)",
                nodes
                    .iter()
                    .map(|(_, _, _, _, received)| received.load(Ordering::SeqCst))
                    .collect::<Vec<_>>()
            );
            thread::sleep(Duration::from_millis(25));
        }
    }

    /// Part B CONTRACT (the bug): WITHOUT the proactive net-tick, a link to a
    /// silently-partitioned (black-holed) peer — one whose socket stays open but
    /// sends nothing and no TCP FIN/RST arrives — is NEVER marked down. The read
    /// loop blocks in `read_exact` forever, so `connected_nodes()` keeps listing
    /// a dead peer and the connection-down hook (pg-purge / monitor-DOWN) never
    /// fires. This pins the gap the net-tick closes.
    #[tokio::test]
    async fn without_net_tick_black_holed_peer_is_never_marked_down() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind responder listener");
        let addr = listener.local_addr().expect("inspect listener");
        let handoff = spawn_responder_handoff(listener, "remote@127.0.0.1");

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        let manager = manager_with_resolver(resolver);
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");

        let connection = manager
            .connect("remote@127.0.0.1")
            .await
            .expect("connect should succeed");
        // Hold the peer's accepted stream open WITHOUT ever reading or writing it:
        // a silent partition (no FIN/RST). The peer task has already returned.
        let _black_holed_peer = handoff.await.expect("peer hands back its open stream");

        // Over a window many times longer than any plausible net-tick deadline,
        // the link stays up: no heartbeat means no proactive liveness check.
        tokio::time::sleep(Duration::from_millis(600)).await;
        assert!(
            !connection.is_down(),
            "without a net-tick, a black-holed link is never detected as down"
        );
        assert!(
            manager.connected_nodes().contains(&node),
            "without a net-tick, connected_nodes keeps listing the dead peer"
        );
    }

    /// Part B FIX: WITH the proactive net-tick enabled, a link to a
    /// silently-partitioned peer is marked down within a bounded deadline via the
    /// EXISTING `mark_down` path — so `connected_nodes()` drops it and the
    /// connection-down hook fires, exactly as a real read EOF would. The black-
    /// holed peer never sends a keepalive, so the inbound-liveness deadline lapses.
    #[tokio::test]
    async fn net_tick_marks_black_holed_peer_down_within_deadline() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind responder listener");
        let addr = listener.local_addr().expect("inspect listener");
        let handoff = spawn_responder_handoff(listener, "remote@127.0.0.1");

        let resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            addr,
        )])));
        // Test-scale net-tick: 50ms interval, 200ms deadline.
        let manager = manager_with_heartbeat(
            resolver,
            Duration::from_millis(50),
            Duration::from_millis(200),
        );
        let node = manager.inner.atom_table.intern("remote@127.0.0.1");

        // Observe the connection-down hook firing for the black-holed peer.
        let down_fired = Arc::new(AtomicBool::new(false));
        let observed = Arc::clone(&down_fired);
        manager.register_connection_down(move |event| {
            if event.reason == ConnectionDownReason::HeartbeatTimeout {
                observed.store(true, Ordering::SeqCst);
            }
        });

        let connection = manager
            .connect("remote@127.0.0.1")
            .await
            .expect("connect should succeed");
        // Hold the peer's stream open but silent — never read, never write.
        let _black_holed_peer = handoff.await.expect("peer hands back its open stream");

        // Within a bounded window (a few deadlines) the net-tick must mark the
        // link down and reap it from the table.
        let deadline = Instant::now() + Duration::from_secs(5);
        while !connection.is_down() {
            assert!(
                Instant::now() < deadline,
                "net-tick must mark a black-holed link down within the deadline"
            );
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        // The down path reaped the table entry and fired the hook.
        while manager.connected_nodes().contains(&node) {
            assert!(
                Instant::now() < deadline,
                "the downed link must be removed from connected_nodes"
            );
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        assert!(
            down_fired.load(Ordering::SeqCst),
            "the connection-down hook must fire with HeartbeatTimeout"
        );
    }

    /// No false nodedowns: WITH the net-tick enabled on BOTH peers, a healthy but
    /// otherwise idle link (no application traffic) stays up indefinitely, because
    /// each side's periodic keepalive refreshes the other's inbound-liveness clock
    /// well within the deadline. This guards against the net-tick spuriously
    /// downing quiet-but-live links.
    #[tokio::test]
    async fn net_tick_keeps_healthy_idle_link_up() {
        // Build a REAL peer manager (also heartbeat-enabled) so both sides emit
        // keepalives, modelling a healthy bidirectional idle link. The remote
        // binds its own listener on an ephemeral port via `listen`.
        let remote = ConnectionManager::new(
            Arc::new(AtomTable::with_common_atoms()),
            Arc::new(StaticResolver::new(std::collections::HashMap::new())),
            TEST_COOKIE,
            "remote@127.0.0.1",
            7,
        )
        .with_heartbeat(HeartbeatConfig {
            interval: Duration::from_millis(50),
            deadline: Duration::from_millis(200),
        });
        let accept = remote
            .listen("127.0.0.1:0".parse().expect("listen address parses"))
            .await
            .expect("remote node listens");
        let remote_addr = accept.local_addr();

        let local_resolver = Arc::new(StaticResolver::new(std::collections::HashMap::from([(
            "remote@127.0.0.1".to_string(),
            remote_addr,
        )])));
        let local = manager_with_heartbeat(
            local_resolver,
            Duration::from_millis(50),
            Duration::from_millis(200),
        );

        let node = local.inner.atom_table.intern("remote@127.0.0.1");
        let connection = local
            .connect("remote@127.0.0.1")
            .await
            .expect("connect should succeed");

        // Over a window many deadlines long, both keepalives keep the link live.
        tokio::time::sleep(Duration::from_millis(800)).await;
        assert!(
            !connection.is_down(),
            "a healthy idle link with bidirectional keepalives must stay up"
        );
        assert!(
            local.connected_nodes().contains(&node),
            "a healthy idle link must remain in connected_nodes"
        );

        accept.shutdown();
    }
}