freenet 0.2.67

//! # Connect Protocol
//!
//! This module implements the connect protocol for establishing peer connections in the Freenet
//! network. Understanding this protocol is essential for working on topology and routing.
//!
//! ## Key Concepts
//!
//! - **Joiner**: The peer initiating the connection request
//! - **Relay**: Any peer that receives and processes the request (first hop is often a gateway,
//!   but can be any known peer)
//! - **Acceptor**: A peer that agrees to connect with the joiner
//!
//! ## Critical Behavior: Accept Only at Terminus
//!
//! **Relays only ACCEPT when they can't forward to a closer peer** (terminus). This means:
//!
//! ```text
//! Joiner sends ConnectRequest targeting location 0.3
//!     |
//!     v
//! Gateway (loc=0.7) ──> FORWARDS toward 0.3 (not at terminus, doesn't accept)
//!     |
//!     v
//! Relay A (loc=0.5) ──> FORWARDS toward 0.3 (not at terminus, doesn't accept)
//!     |
//!     v
//! Relay B (loc=0.35) ──> FORWARDS toward 0.3 (not at terminus, doesn't accept)
//!     |
//!     v
//! Relay C (loc=0.31) ──> ACCEPTS (at terminus - no closer peer to forward to)
//! ```
//!
//! The joiner typically receives ONE ConnectResponse from the terminus peer. To get multiple
//! connections, the joiner sends multiple ConnectRequests (potentially targeting different
//! locations). This ensures connections are naturally local (short ring distance).
//!
//! ## Message Flow
//!
//! 1. **ConnectRequest**: Joiner → Relay₁ → Relay₂ → ... (routed toward `desired_location`)
//! 2. **ConnectResponse**: Each acceptor → back through relay chain → Joiner
//! 3. **ObservedAddress**: First relay → Joiner (tells joiner their external IP for NAT traversal)
//!
//! ## Address Discovery (NAT Traversal)
//!
//! The joiner typically doesn't know its external address (behind NAT). The protocol handles this:
//!
//! 1. Joiner creates ConnectRequest with `joiner.peer_addr = Unknown`
//! 2. First relay observes joiner's address from UDP packet source
//! 3. First relay fills in `joiner.peer_addr` and sends `ObservedAddress` back to joiner
//! 4. Subsequent relays see the already-filled address
//!
//! ## Acceptance Criteria: Accept Only at Terminus
//!
//! Relays only accept connect requests when they're at the routing terminus - meaning they
//! can't forward to a peer closer to the target location. This naturally creates local
//! connections without arbitrary distance thresholds.
//!
//! **Algorithm:**
//! 1. First, check if we can forward to a closer peer via `select_next_hop()`
//! 2. If we CAN forward: forward only (don't accept)
//! 3. If we CAN'T forward (terminus): accept if `should_accept()` allows
//!
//! `should_accept()` still uses capacity-based evaluation:
//! - Below min_connections → **accept**
//! - At max_connections → **reject**
//! - Between min and max → use density-based evaluation
//!
//! **Why this works for small-world topology:**
//! - Requests route toward the target location via greedy routing
//! - Only peers that can't forward further (near the target) accept
//! - Gateway and early relays forward without accepting
//! - Result: connections are naturally local (short ring distance)
//!
//! ## Routing
//!
//! The request is routed toward `desired_location` using greedy routing:
//! - Each relay forwards to its neighbor closest to `desired_location`
//! - The `visited` list prevents routing loops
//! - TTL decrements at each hop
//!
//! ## Why Target Location Matters
//!
//! If the joiner targets their own location, the request routes toward peers near them on the
//! ring. With accept-only-at-terminus, only the peer closest to the target (who can't forward
//! further) accepts. This naturally creates local connections.
//!
//! To build multiple connections, the joiner sends multiple ConnectRequests. Early in bootstrap
//! (0-4 connections), peers target their own location to build local neighborhoods. Later,
//! density-based targeting is used to optimize for request patterns.

use std::collections::{HashMap, HashSet};
use std::fmt;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::Instant;

use futures::{StreamExt, stream::FuturesUnordered};
#[cfg(test)]
use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;

use crate::config::{GlobalExecutor, GlobalRng};
use crate::dev_tool::Location;
use crate::message::{InnerMessage, NodeEvent, Transaction};
use crate::node::OpManager;
use crate::operations::OpError;
use crate::ring::{KnownPeerKeyLocation, PeerAddr, PeerKeyLocation};
use crate::router::{EstimatorType, IsotonicEstimator, IsotonicEvent};
use crate::transport::TransportKeypair;
use crate::util::{Contains, IterExt};

use super::VisitedPeers;

pub(crate) mod op_ctx_task;

#[cfg(test)]
const FORWARD_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(20);
const RECENCY_COOLDOWN: Duration = Duration::from_secs(30);

/// Minimum forward estimator score for a peer to be considered for CONNECT forwarding.
/// Peers scoring below this threshold are filtered out when alternatives exist,
/// preventing routing through known-bad paths.
const MIN_FORWARD_SCORE: f64 = 0.15;

/// Ring distance threshold below which a relay may probabilistically accept a
/// connection even when it can still forward. This implements "near-terminus
/// acceptance" — relays that are close to the target both accept AND forward,
/// giving the joiner more diverse connection options without waiting to reach
/// the strict terminus.
///
/// Value 0.05 = within 5% of the ring from the target. At this distance the
/// relay is topologically relevant to the joiner.
const NEAR_TERMINUS_DISTANCE: f64 = 0.05;

/// Base probability that a near-terminus relay accepts while forwarding.
/// Actual probability scales linearly with proximity: closer = higher chance.
/// At distance 0 this is the full probability; at `NEAR_TERMINUS_DISTANCE` it's 0.
const NEAR_TERMINUS_ACCEPT_PROB: f64 = 0.6;

/// Tolerance band for reliability-based uphill peer selection.
/// Candidates within this delta of the best reliability score are
/// treated as equivalent, allowing distance-based partitioning to
/// differentiate among similarly-reliable peers.
const UPHILL_RELIABILITY_TOLERANCE: f64 = 0.1;

/// Filter out scored peers below `MIN_FORWARD_SCORE` when alternatives exist.
///
/// If filtering would leave zero scored peers AND zero eligible (unscored) peers,
/// the scored list is left unchanged so we still have a fallback.
fn filter_low_score_peers<T>(scored: &mut Vec<(f64, T)>, eligible_count: usize) {
    if scored.len() + eligible_count <= 1 {
        return;
    }
    let survivors = scored
        .iter()
        .filter(|(s, _)| *s >= MIN_FORWARD_SCORE)
        .count();
    // Only filter if at least one scored peer survives OR unscored fallbacks exist.
    if survivors > 0 || eligible_count > 0 {
        scored.retain(|(score, _)| *score >= MIN_FORWARD_SCORE);
    }
}

/// Top-level message envelope used by the new connect handshake.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) enum ConnectMsg {
    /// Join request that travels *towards* the target location.
    /// Routing is determined by `payload.desired_location`, not an embedded target.
    /// The sender is determined from the transport layer's source address.
    Request {
        id: Transaction,
        payload: ConnectRequest,
    },
    /// Join acceptance that travels back along the discovered path.
    /// Routing uses hop-by-hop via `upstream_addr` stored in operation state.
    /// The sender is determined from the transport layer's source address.
    Response {
        id: Transaction,
        payload: ConnectResponse,
    },
    /// Informational packet letting the joiner know the address a peer observed.
    /// Routing uses hop-by-hop via `upstream_addr` stored in operation state.
    ObservedAddress {
        id: Transaction,
        address: SocketAddr,
    },
    /// Explicit rejection sent back when a relay at terminus can't accept and has
    /// no routing options. Travels hop-by-hop back to the joiner like Response.
    Rejected {
        id: Transaction,
        desired_location: Location,
    },
    /// Sent by the joiner upstream to its first hop (gateway) when hole-punch to an
    /// acceptor fails. Each relay that receives it may forward it downstream (toward
    /// the acceptor-facing relay for re-routing) or try re-routing locally. If no
    /// relay can re-route, propagates back upstream to the joiner.
    ConnectFailed {
        id: Transaction,
        failed_acceptor_addr: SocketAddr,
    },
}

impl InnerMessage for ConnectMsg {
    fn id(&self) -> &Transaction {
        match self {
            ConnectMsg::Request { id, .. }
            | ConnectMsg::Response { id, .. }
            | ConnectMsg::ObservedAddress { id, .. }
            | ConnectMsg::Rejected { id, .. }
            | ConnectMsg::ConnectFailed { id, .. } => id,
        }
    }

    fn requested_location(&self) -> Option<Location> {
        match self {
            ConnectMsg::Request { payload, .. } => Some(payload.desired_location),
            ConnectMsg::Rejected {
                desired_location, ..
            } => Some(*desired_location),
            ConnectMsg::Response { .. }
            | ConnectMsg::ObservedAddress { .. }
            | ConnectMsg::ConnectFailed { .. } => None,
        }
    }
}

impl fmt::Display for ConnectMsg {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ConnectMsg::Request { payload, .. } => write!(
                f,
                "ConnectRequest {{ desired: {}, ttl: {}, uphill: {}, joiner: {} }}",
                payload.desired_location, payload.ttl, payload.uphill_budget, payload.joiner
            ),
            ConnectMsg::Response { payload, .. } => {
                write!(f, "ConnectResponse {{ acceptor: {} }}", payload.acceptor,)
            }
            ConnectMsg::ObservedAddress { address, .. } => {
                write!(f, "ObservedAddress {{ address: {address} }}")
            }
            ConnectMsg::Rejected {
                desired_location, ..
            } => {
                write!(f, "Rejected {{ desired: {desired_location} }}")
            }
            ConnectMsg::ConnectFailed {
                failed_acceptor_addr,
                ..
            } => {
                write!(
                    f,
                    "ConnectFailed {{ failed_acceptor: {failed_acceptor_addr} }}"
                )
            }
        }
    }
}

/// Maximum number of uphill routing hops per CONNECT transaction.
///
/// Uphill routing occurs when a terminus peer cannot accept a connection
/// (already connected, at capacity, NAT failure) and forwards the request
/// away from the target to give other peers a chance.
///
/// This budget bounds amplification to O(UPHILL_BUDGET) messages per transaction,
/// independent of TTL. With ~84% NAT hole-punch failure rate in production,
/// 8 candidates give P(at least one success) ≈ 1 - 0.84^8 ≈ 72%.
pub(crate) const DEFAULT_UPHILL_BUDGET: u8 = 8;

/// Two-message request payload.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct ConnectRequest {
    /// Joiner's advertised location (fallbacks to the joiner's socket address).
    pub desired_location: Location,
    /// Joiner's identity and address. When the joiner creates this request,
    /// `joiner.peer_addr` is set to `PeerAddr::Unknown` because the joiner
    /// doesn't know its own external address (especially behind NAT).
    /// The first recipient (gateway) fills this in from the packet source address.
    pub joiner: PeerKeyLocation,
    /// Remaining hops before the request stops travelling.
    pub ttl: u8,
    /// Bloom filter tracking visited peers to prevent routing loops.
    /// Uses transaction-specific hashing for privacy (same peer hashes differently
    /// in different transactions, preventing topology inference).
    pub visited: VisitedPeers,
    /// Remaining uphill routing budget. Decremented only when routing away
    /// from the target (uphill hops). Normal toward-target hops use only TTL.
    /// Bounds amplification to O(uphill_budget) per transaction regardless of TTL.
    #[serde(default = "default_uphill_budget")]
    pub uphill_budget: u8,
}

fn default_uphill_budget() -> u8 {
    DEFAULT_UPHILL_BUDGET
}

/// Acceptance payload returned by candidates.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) struct ConnectResponse {
    /// The peer that accepted the join request.
    pub acceptor: PeerKeyLocation,
}

/// State machine retained for in-file relay/joiner unit tests
/// (`tests::*` in this file). Production CONNECT runs entirely on the
/// drivers in `op_ctx_task.rs` and rebuilds `RelayState`
/// in task locals; nothing constructs `ConnectState::Relaying` or
/// `Completed` outside the test module.
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub(crate) enum ConnectState {
    WaitingForResponses(JoinerState),
    Relaying(Box<RelayState>),
    Completed,
}

#[derive(Debug, Clone)]
#[allow(dead_code)] // Fields only read by in-file unit tests.
pub(crate) struct JoinerState {
    pub target_connections: usize,
    pub observed_address: Option<SocketAddr>,
    pub accepted: HashSet<PeerKeyLocation>,
    pub last_progress: Instant,
    pub started_without_address: bool,
}

#[derive(Debug, Clone)]
pub(crate) struct RelayState {
    /// Address of the peer that sent us this request (for response routing).
    /// This is determined from the transport layer's source address.
    pub upstream_addr: SocketAddr,
    pub request: ConnectRequest,
    pub forwarded_to: Option<PeerKeyLocation>,
    /// When the request was forwarded to `forwarded_to` peer.
    /// Used for GC timeout logging and diagnostics.
    pub forwarded_at: Option<Instant>,
    pub observed_sent: bool,
    pub accepted_locally: bool,
    /// Set to true after successfully forwarding a ConnectResponse upstream.
    /// Prevents false "uphill route timed out" logging in GC, and preserves
    /// `forwarded_to` for ConnectFailed downstream propagation.
    pub response_forwarded: bool,
}

/// Abstractions required to evaluate an inbound connect request at an
/// intermediate peer.
pub(crate) trait RelayContext {
    /// Location of the current peer.
    fn self_location(&self) -> &PeerKeyLocation;

    /// Determine whether we should accept the joiner immediately.
    ///
    /// Takes a `KnownPeerKeyLocation` to enforce at compile time that the joiner's
    /// address must be known before acceptance decisions can be made.
    fn should_accept(&self, joiner: &KnownPeerKeyLocation) -> bool;

    /// Choose the next hop for the request, avoiding peers already visited.
    fn select_next_hop(
        &self,
        desired_location: Location,
        visited: &VisitedPeers,
        recency: &HashMap<PeerKeyLocation, Instant>,
        estimator: &ConnectForwardEstimator,
    ) -> Option<PeerKeyLocation>;

    /// Choose an uphill hop for the request when at terminus but cannot accept.
    /// "Uphill" means routing AWAY from the target (farther from desired_location),
    /// which gives peers farther from the target a chance to accept.
    fn select_uphill_hop(
        &self,
        desired_location: Location,
        visited: &VisitedPeers,
        recency: &HashMap<PeerKeyLocation, Instant>,
    ) -> Option<PeerKeyLocation>;
}

/// Bundle of the two side-effects a relay must emit atomically when it
/// decides to accept a CONNECT request: the `ConnectResponse` message to send
/// upstream AND the `PeerKeyLocation` of the joiner whose transient transport
/// connection must be promoted to a ring connection.
///
/// These MUST be handled together. Accepting without promoting leaves the
/// joiner stuck in `transient_connections` on the acceptor, so downstream
/// lookups like `get_peer_by_addr` (used by subscribe interest registration
/// and broadcast fan-out) cannot find the peer. This was the root cause of
/// issue #3838 — the `Rejected`/`ConnectFailed` retry paths in
/// `process_message` had drifted apart and sent the response without
/// promoting the joiner. Combining the two fields into one struct makes that
/// class of bug unrepresentable.
#[derive(Debug)]
pub(crate) struct AcceptOutcome {
    pub response: ConnectResponse,
    pub joiner: PeerKeyLocation,
}

/// Result of processing a request at a relay.
#[derive(Debug, Default)]
pub(crate) struct RelayActions {
    /// Set when this relay decided to accept the joiner itself. The joiner
    /// and the response travel together so callers cannot forget to promote
    /// the transient connection to a ring connection. See [`AcceptOutcome`].
    pub accept: Option<AcceptOutcome>,
    pub forward: Option<(PeerKeyLocation, ConnectRequest)>,
    pub observed_address: Option<(PeerKeyLocation, SocketAddr)>,
    /// True when at terminus, cannot accept, and has no routing options.
    /// Signals that an explicit Rejected message should be sent back.
    pub rejected: bool,
}

#[derive(Debug, Clone)]
pub(crate) struct ForwardAttempt {
    #[cfg_attr(not(test), allow(dead_code))]
    peer: PeerKeyLocation,
    #[cfg_attr(not(test), allow(dead_code))]
    desired: Location,
    #[cfg_attr(not(test), allow(dead_code))]
    sent_at: Instant,
}

#[derive(Debug, Clone)]
pub(crate) struct ConnectForwardEstimator {
    estimator: IsotonicEstimator,
}

impl ConnectForwardEstimator {
    pub(crate) fn new() -> Self {
        // Seed with neutral points to allow construction without history. This estimator
        // learns, per-node, how often downstream peers accept/complete forwarded Connect
        // requests so we can bias forwarding toward peers likely to have capacity.
        let key = TransportKeypair::new();
        let dummy_peer = PeerKeyLocation::new(key.public().clone(), "127.0.0.1:0".parse().unwrap());
        let seed_events = [
            IsotonicEvent {
                peer: dummy_peer.clone(),
                contract_location: Location::new(0.0),
                result: 0.5,
            },
            IsotonicEvent {
                peer: dummy_peer,
                contract_location: Location::new(0.5),
                result: 0.5,
            },
        ];

        Self {
            estimator: IsotonicEstimator::new(seed_events, EstimatorType::Negative),
        }
    }

    pub(super) fn record(&mut self, peer: &PeerKeyLocation, desired: Location, success: bool) {
        if peer.location().is_none() {
            return;
        }
        // Treat each downstream forward attempt as a Bernoulli outcome for this peer and
        // desired ring location; these accumulate to bias future routing choices.
        let event = IsotonicEvent {
            peer: peer.clone(),
            contract_location: desired,
            result: if success { 1.0 } else { 0.0 },
        };
        self.estimator.add_event(event);
    }

    fn estimate(&self, peer: &PeerKeyLocation, desired: Location) -> Option<f64> {
        peer.location()?;
        self.estimator
            .estimate_retrieval_time(peer, desired)
            .ok()
            .map(|p| p.clamp(0.0, 1.0))
    }

    /// Return the sampled PAV curve, data range, event count, and peer adjustment count.
    #[allow(clippy::type_complexity)]
    pub(crate) fn snapshot(&self) -> (Vec<(f64, f64)>, (f64, f64), usize, usize) {
        let curve = self.estimator.sampled_curve(0.0, 1.0, 50);
        let data_range = self.estimator.data_x_range();
        (
            curve,
            data_range,
            self.estimator.len(),
            self.estimator.peer_adjustments.len(),
        )
    }
}
impl RelayState {
    /// Helper to prepare and execute forwarding to a peer.
    /// Returns the forward action to add to RelayActions.
    ///
    /// `now` is passed in (not derived from `Instant::now()`) so callers can
    /// provide a consistent timestamp shared with other time-sensitive operations
    /// in the same routing decision.
    fn forward_to_peer<C: RelayContext>(
        &mut self,
        _ctx: &C,
        peer: PeerKeyLocation,
        forward_attempts: &mut HashMap<PeerKeyLocation, ForwardAttempt>,
        now: Instant,
    ) -> (PeerKeyLocation, ConnectRequest) {
        let mut forward_req = self.request.clone();
        forward_req.ttl = forward_req.ttl.saturating_sub(1);
        // Note: self_addr is already marked as visited in handle_request() before
        // forward_to_peer is called, so no need to mark it again here.
        let forward_snapshot = forward_req.clone();
        self.forwarded_to = Some(peer.clone());
        self.forwarded_at = Some(now);
        self.request = forward_req;
        forward_attempts.insert(
            peer.clone(),
            ForwardAttempt {
                peer: peer.clone(),
                desired: self.request.desired_location,
                sent_at: now,
            },
        );
        (peer, forward_snapshot)
    }

    pub(crate) fn handle_request<C: RelayContext>(
        &mut self,
        ctx: &C,
        recency: &HashMap<PeerKeyLocation, Instant>,
        forward_attempts: &mut HashMap<PeerKeyLocation, ForwardAttempt>,
        estimator: &ConnectForwardEstimator,
        now: Instant,
    ) -> RelayActions {
        let mut actions = RelayActions::default();
        tracing::debug!(
            joiner_addr = ?self.request.joiner.peer_addr,
            upstream_addr = %self.upstream_addr,
            observed_sent = self.observed_sent,
            "RelayState::step() start"
        );
        // Add upstream's address (determined from transport layer) to visited list.
        // Mark upstream first since their observation of our address is more reliable for NAT.
        self.request.visited.mark_visited(self.upstream_addr);
        // Add our own address to visited list
        if let Some(self_addr) = ctx.self_location().socket_addr() {
            self.request.visited.mark_visited(self_addr);
        }

        // Fill in joiner's external address from transport layer if unknown.
        // This is the key step where the first recipient (gateway) determines the joiner's
        // external address from the actual packet source address.
        let discovered_joiner_addr = if self.request.joiner.peer_addr.is_unknown() {
            self.request.joiner.set_addr(self.upstream_addr);
            true
        } else {
            false
        };

        // Only send ObservedAddress if WE discovered the joiner's address (it was unknown
        // and we just filled it in from the packet source). If the address was already known
        // in the incoming request, the upstream relay already sent ObservedAddress.
        // NOTE: Always emit ObservedAddress from the first hop (gateway) since it observes
        // the joiner's external address from the packet source. The `discovered_joiner_addr`
        // flag indicates we just filled in the address from the transport layer.
        if discovered_joiner_addr {
            if let PeerAddr::Known(joiner_addr) = &self.request.joiner.peer_addr {
                if !self.observed_sent {
                    self.observed_sent = true;
                    let expected_location = crate::ring::Location::from_address(joiner_addr);
                    tracing::debug!(
                        joiner_addr = %joiner_addr,
                        expected_location = %expected_location,
                        upstream_addr = %self.upstream_addr,
                        "discovered joiner addr, emitting ObservedAddress"
                    );
                    actions.observed_address = Some((self.request.joiner.clone(), *joiner_addr));
                }
            }
        }

        // ACCEPT ONLY AT TERMINUS: Relays only accept when they can't forward to a closer peer.
        // This naturally creates local connections because only peers near the target accept.
        //
        // Algorithm:
        // 1. First, check if we can forward to a closer peer
        // 2. If we can forward: forward only (don't accept)
        // 3. If we can't forward (terminus): accept if should_accept() allows
        //
        // This prevents early relays (gateway, first hops) from accepting connections that
        // would be non-local on the ring. Only peers that are actually close to the target
        // (and thus can't forward further) will accept.

        let can_forward = self.forwarded_to.is_none() && self.request.ttl > 0;
        let next_hop = if can_forward {
            ctx.select_next_hop(
                self.request.desired_location,
                &self.request.visited,
                recency,
                estimator,
            )
        } else {
            None
        };

        let is_terminus = next_hop.is_none();

        // Forward if we have a next hop
        if let Some(next) = next_hop {
            let dist = ring_distance(next.location(), Some(self.request.desired_location));
            tracing::debug!(
                target = %self.request.desired_location,
                ttl = self.request.ttl,
                next_peer = %next.pub_key(),
                next_loc = ?next.location(),
                ring_distance_to_target = ?dist,
                "connect: forwarding join request to next hop (not accepting - not at terminus)"
            );
            actions.forward = Some(self.forward_to_peer(ctx, next, forward_attempts, now));

            // Near-terminus acceptance: if this relay is close to the target,
            // probabilistically also accept while forwarding. This gives the
            // joiner more diverse connections without waiting for strict terminus.
            if !self.accepted_locally {
                let my_dist = ring_distance(
                    ctx.self_location().location(),
                    Some(self.request.desired_location),
                );
                if let Some(d) = my_dist {
                    if d < NEAR_TERMINUS_DISTANCE {
                        // Probability scales linearly: closer → higher chance
                        let accept_prob =
                            NEAR_TERMINUS_ACCEPT_PROB * (1.0 - d / NEAR_TERMINUS_DISTANCE);
                        let roll: f64 = GlobalRng::random_range(0.0..1.0);
                        if roll < accept_prob {
                            if let Ok(joiner_known) =
                                KnownPeerKeyLocation::try_from(&self.request.joiner)
                            {
                                if ctx.should_accept(&joiner_known) {
                                    self.accepted_locally = true;
                                    let self_loc = ctx.self_location();
                                    let acceptor = PeerKeyLocation::with_unknown_addr(
                                        self_loc.pub_key().clone(),
                                    );
                                    actions.accept = Some(AcceptOutcome {
                                        response: ConnectResponse { acceptor },
                                        joiner: self.request.joiner.clone(),
                                    });
                                    tracing::info!(
                                        acceptor_loc = ?self_loc.location(),
                                        joiner_loc = ?self.request.joiner.location(),
                                        ring_distance = d,
                                        accept_prob,
                                        "connect: near-terminus acceptance (also forwarding)"
                                    );
                                }
                            }
                        }
                    }
                }
            }
        }

        // Only accept at terminus (can't forward to a closer peer)
        // IMPORTANT: Also check that we haven't already forwarded - once we forward,
        // we're committed to that path and should not also accept. This prevents
        // the "double-accept" bug where a retry call could trigger acceptance
        // after we've already forwarded the request elsewhere.
        //
        // The joiner must have a known address by this point (filled in by first relay).
        // If conversion fails, we cannot accept - the joiner's address is still unknown.
        let joiner_known = match KnownPeerKeyLocation::try_from(&self.request.joiner) {
            Ok(known) => Some(known),
            Err(e) => {
                // This is an internal consistency error - by this point in the connect flow,
                // the joiner's address should have been filled in by the first relay.
                // Log as error to aid debugging of NAT traversal or protocol issues.
                tracing::error!(
                    joiner_pub_key = %e.pub_key,
                    upstream_addr = %self.upstream_addr,
                    "INTERNAL ERROR: joiner has unknown address at acceptance check - first relay should have filled this in"
                );
                None
            }
        };
        if is_terminus
            && !self.accepted_locally
            && self.forwarded_to.is_none()
            && joiner_known.as_ref().is_some_and(|j| ctx.should_accept(j))
        {
            self.accepted_locally = true;
            // Use unknown address for the acceptor - the acceptor doesn't know their own
            // external address (especially behind NAT). The first relay that receives this
            // response will fill in the address from the packet source, similar to how
            // joiner addresses are filled in for ConnectRequest.
            let self_loc = ctx.self_location();
            let acceptor = PeerKeyLocation::with_unknown_addr(self_loc.pub_key().clone());
            let dist = ring_distance(self_loc.location(), self.request.joiner.location());
            actions.accept = Some(AcceptOutcome {
                response: ConnectResponse { acceptor },
                joiner: self.request.joiner.clone(),
            });
            // Response is routed hop-by-hop via upstream_addr, no target embedded in message
            tracing::info!(
                acceptor_pub_key = %self_loc.pub_key(),
                joiner_pub_key = %self.request.joiner.pub_key(),
                acceptor_loc = ?self_loc.location(),
                joiner_loc = ?self.request.joiner.location(),
                ring_distance = ?dist,
                "connect: acceptance issued at terminus (acceptor addr will be filled by relay)"
            );
        } else if is_terminus && !self.accepted_locally && self.forwarded_to.is_none() {
            // At terminus but should_accept() returned false (e.g., already connected
            // or at capacity). Route uphill to give other peers a chance to accept.
            //
            // Amplification is bounded by uphill_budget (separate from TTL).
            // Each uphill hop decrements uphill_budget by 1 but only decrements
            // TTL by the normal -1 (no extra burn). This separates the two concerns:
            // - TTL controls network reach (how far a CONNECT can travel)
            // - uphill_budget controls amplification (how many uphill retries)
            //
            // With DEFAULT_UPHILL_BUDGET=8 and ~84% NAT failure rate, peers get
            // enough candidates for ~72% chance of finding a reachable acceptor.
            if self.request.uphill_budget > 0 && self.request.ttl >= 2 {
                let uphill_hop = ctx.select_uphill_hop(
                    self.request.desired_location,
                    &self.request.visited,
                    recency,
                );

                if let Some(uphill_peer) = uphill_hop {
                    let dist =
                        ring_distance(uphill_peer.location(), Some(self.request.desired_location));
                    tracing::debug!(
                        target = %self.request.desired_location,
                        ttl = self.request.ttl,
                        uphill_budget = self.request.uphill_budget,
                        uphill_peer = %uphill_peer.pub_key(),
                        uphill_loc = ?uphill_peer.location(),
                        ring_distance_to_target = ?dist,
                        "connect: at terminus but cannot accept, routing uphill"
                    );
                    // Decrement uphill_budget on both the forwarded request AND
                    // self.request, so retries from the same relay also consume budget.
                    // TTL is decremented normally (-1) by forward_to_peer — no extra burn.
                    let (peer, mut fwd_req) =
                        self.forward_to_peer(ctx, uphill_peer, forward_attempts, now);
                    fwd_req.uphill_budget = fwd_req.uphill_budget.saturating_sub(1);
                    self.request.uphill_budget = self.request.uphill_budget.saturating_sub(1);
                    actions.forward = Some((peer, fwd_req));
                } else {
                    tracing::info!(
                        target = %self.request.desired_location,
                        ttl = self.request.ttl,
                        uphill_budget = self.request.uphill_budget,
                        visited = ?self.request.visited,
                        "connect: at terminus, cannot accept, no uphill peers available — rejecting"
                    );
                    actions.rejected = true;
                }
            } else {
                tracing::info!(
                    target = %self.request.desired_location,
                    ttl = self.request.ttl,
                    uphill_budget = self.request.uphill_budget,
                    visited = ?self.request.visited,
                    "connect: at terminus, cannot accept, uphill budget or TTL exhausted — rejecting"
                );
                actions.rejected = true;
            }
        } else if is_terminus && self.forwarded_to.is_some() {
            tracing::debug!(
                target = %self.request.desired_location,
                forwarded_to = ?self.forwarded_to,
                "connect: at terminus but already forwarded - waiting for response"
            );
        }

        actions
    }
}

pub(crate) struct RelayEnv<'a> {
    pub op_manager: &'a OpManager,
    self_location: PeerKeyLocation,
}

impl<'a> RelayEnv<'a> {
    pub fn new(op_manager: &'a OpManager) -> Self {
        let self_location = op_manager.ring.connection_manager.own_location();
        Self {
            op_manager,
            self_location,
        }
    }
}

impl RelayContext for RelayEnv<'_> {
    fn self_location(&self) -> &PeerKeyLocation {
        &self.self_location
    }

    fn should_accept(&self, joiner: &KnownPeerKeyLocation) -> bool {
        let addr = joiner.socket_addr();
        let location = joiner.location();

        // Check blocklist before capacity. If we block this peer, return false
        // so the uphill hop mechanism can route to an alternate acceptor.
        if let Some(blocked) = &self.op_manager.blocked_addresses {
            if blocked.contains(&addr) {
                tracing::info!(
                    joiner_addr = %addr,
                    "connect: rejecting join from blocked peer at routing level"
                );
                return false;
            }
        }

        self.op_manager
            .ring
            .connection_manager
            .should_accept(location, addr)
    }

    fn select_next_hop(
        &self,
        desired_location: Location,
        visited: &VisitedPeers,
        recency: &HashMap<PeerKeyLocation, Instant>,
        estimator: &ConnectForwardEstimator,
    ) -> Option<PeerKeyLocation> {
        // Capture now once so that the reliability lookup and the recency cooldown
        // check both see the same timestamp for this routing decision.
        let now = Instant::now();

        let skip = SkipListWithSelf {
            visited,
            self_addr: self.self_location.socket_addr(),
        };
        let router = self.op_manager.ring.router.read();

        let candidates = self.op_manager.ring.connection_manager.routing_candidates(
            desired_location,
            None,
            skip,
            false, // CONNECT bypasses readiness filter to allow routing to not-yet-ready peers
        );

        // Calculate our own distance to the target for greedy routing check
        let my_distance = self
            .self_location
            .location()
            .map(|loc| loc.distance(desired_location));
        let mut scored: Vec<(f64, PeerKeyLocation)> = Vec::with_capacity(candidates.len());
        let mut fallbacks: Vec<PeerKeyLocation> = Vec::with_capacity(candidates.len());

        let conn_mgr = &self.op_manager.ring.connection_manager;

        for cand in candidates {
            if let Some(ts) = recency.get(&cand) {
                if now.duration_since(*ts) < RECENCY_COOLDOWN {
                    continue;
                }
            }

            // GREEDY ROUTING: Only consider neighbors that are CLOSER to the target
            // than we are. We use `>` (not `>=`) to allow equally-distant peers,
            // giving more flexibility near ring boundary 0.0/1.0. Routing loops
            // are prevented by the `visited` list, not by this distance check.
            if let (Some(cand_loc), Some(my_dist)) = (cand.location(), my_distance) {
                if cand_loc.distance(desired_location) > my_dist {
                    continue;
                }
            }

            // Acceptor reliability: unknown peers get 0.5, observed failures lower.
            let reliability = cand
                .socket_addr()
                .map(|addr| conn_mgr.peer_acceptor_reliability(addr, now))
                .unwrap_or(0.5);

            if cand.location().is_some() {
                if let Some(estimator_score) = estimator.estimate(&cand, desired_location) {
                    // Combine path quality (estimator) with endpoint reliability.
                    // These signals partially overlap (the estimator sees end-to-end
                    // outcomes including hole-punch failures), but telemetry shows
                    // the estimator's global curve is too optimistic at close distances
                    // for specific bad peers (scores 0.4-0.8 where true rate is ~0.05).
                    // Multiplication is the correct combination for P(path_success) ×
                    // P(holepunch_success) — the partial overlap means we slightly
                    // over-penalize, but this is preferable to under-penalizing.
                    let combined_score = estimator_score * reliability;
                    scored.push((combined_score, cand.clone()));
                    continue;
                }
                // No estimator data — keep as fallback to avoid ranking untrained
                // peers above routes with learned history (score 0.5 from raw
                // reliability could beat a trained route at 0.4 * 1.0 = 0.4).
                fallbacks.push(cand.clone());
                continue;
            }
            // Candidates without a location go to fallback pool.
            fallbacks.push(cand.clone());
        }

        // Filter out low-score peers when alternatives exist.
        filter_low_score_peers(&mut scored, fallbacks.len());

        // If all remaining scored candidates have very low acceptor reliability
        // (from actual observed failures, not just sparse estimator data) and
        // there are no fallbacks, return None so the caller escalates to uphill
        // routing. Without this, a single unreliable closer peer would prevent
        // the node from ever reaching terminus — it would keep forwarding to
        // the bad peer instead of trying uphill alternatives.
        //
        // We check reliability directly (not the combined score) because a low
        // combined score from a fresh estimator × 0.5 prior should NOT trigger
        // uphill escalation — those are untested peers, not known-bad ones.
        if !scored.is_empty() && fallbacks.is_empty() {
            let all_unreliable = scored.iter().all(|(_, cand)| {
                cand.socket_addr()
                    .map(|addr| conn_mgr.peer_acceptor_reliability(addr, now))
                    .unwrap_or(0.5)
                    < MIN_FORWARD_SCORE
            });
            if all_unreliable {
                tracing::debug!(
                    candidates = scored.len(),
                    "connect: all closer peers have low acceptor reliability, escalating to uphill"
                );
                return None;
            }
        }

        if !scored.is_empty() {
            let best_score = scored
                .iter()
                .map(|(s, _)| *s)
                .max_by(|a, b| a.partial_cmp(b).unwrap())
                .unwrap_or(0.0);
            let best: Vec<_> = scored
                .into_iter()
                .filter(|(s, _)| (*s - best_score).abs() < f64::EPSILON)
                .map(|(_, c)| c)
                .collect();
            if !best.is_empty() {
                return router.select_peer(best.iter(), desired_location).cloned();
            }
        }

        if fallbacks.is_empty() {
            None
        } else {
            router
                .select_peer(fallbacks.iter(), desired_location)
                .cloned()
        }
    }

    fn select_uphill_hop(
        &self,
        desired_location: Location,
        visited: &VisitedPeers,
        recency: &HashMap<PeerKeyLocation, Instant>,
    ) -> Option<PeerKeyLocation> {
        // SMARTER UPHILL ROUTING: When at terminus but can't accept, prefer
        // uphill peers that are still relatively close to the target.
        //
        // This method is only called when select_next_hop returned None (terminus).
        // All remaining candidates are "uphill" (farther from target than us).
        // We prefer peers that are still within a reasonable distance of the
        // target, as they are more likely to have neighbors near the target
        // that we don't know about.
        //
        let now = Instant::now();
        let skip = SkipListWithSelf {
            visited,
            self_addr: self.self_location.socket_addr(),
        };
        let router = self.op_manager.ring.router.read();
        let conn_mgr = &self.op_manager.ring.connection_manager;

        let candidates = conn_mgr.routing_candidates(
            desired_location,
            None,
            skip,
            false, // CONNECT bypasses readiness filter to allow routing to not-yet-ready peers
        );

        // Score candidates by acceptor reliability, filtering out recency cooldown.
        // Single pass: score candidates, track the best reliability,
        // discard peers in recency cooldown.
        let mut best_reliability = 0.0_f64;
        let scored: Vec<(f64, PeerKeyLocation)> = candidates
            .into_iter()
            .filter_map(|cand| {
                // Skip peers we recently forwarded to
                if let Some(ts) = recency.get(&cand) {
                    if now.duration_since(*ts) < RECENCY_COOLDOWN {
                        return None;
                    }
                }
                let reliability = cand
                    .socket_addr()
                    .map(|addr| conn_mgr.peer_acceptor_reliability(addr, now))
                    .unwrap_or(0.5);
                best_reliability = best_reliability.max(reliability);
                Some((reliability, cand))
            })
            .collect();

        if scored.is_empty() {
            tracing::debug!(
                target = %desired_location,
                "connect: no uphill candidates available"
            );
            return None;
        }

        // Keep candidates whose reliability is within 0.1 of the best and
        // immediately partition into "close uphill" (within 2× NEAR_TERMINUS_DISTANCE of target)
        // and "far uphill". This avoids an intermediate Vec allocation.
        let close_threshold = NEAR_TERMINUS_DISTANCE * 2.0;
        let (close, far): (Vec<_>, Vec<_>) = scored
            .into_iter()
            .filter(|(r, _)| best_reliability - *r < UPHILL_RELIABILITY_TOLERANCE)
            .map(|(_, c)| c)
            .partition(|cand: &PeerKeyLocation| {
                cand.location()
                    .map(|loc| loc.distance(desired_location).as_f64() < close_threshold)
                    .unwrap_or(false)
            });

        if !close.is_empty() {
            tracing::debug!(
                target = %desired_location,
                close_count = close.len(),
                far_count = far.len(),
                "connect: preferring close uphill peer"
            );
            router.select_peer(close.iter(), desired_location).cloned()
        } else {
            router.select_peer(far.iter(), desired_location).cloned()
        }
    }
}

#[derive(Debug)]
#[allow(dead_code)] // Constructed only by in-file unit tests.
pub struct AcceptedPeer {
    pub peer: PeerKeyLocation,
}

#[derive(Debug, Default)]
#[allow(dead_code)] // Constructed only by in-file unit tests.
pub struct JoinerAcceptance {
    pub new_acceptor: Option<AcceptedPeer>,
    pub satisfied: bool,
    pub assigned_location: bool,
}

#[allow(dead_code)] // Methods used only by in-file unit tests.
impl JoinerState {
    pub(crate) fn register_acceptance(
        &mut self,
        response: &ConnectResponse,
        now: Instant,
    ) -> JoinerAcceptance {
        let mut acceptance = JoinerAcceptance::default();
        if self.accepted.insert(response.acceptor.clone()) {
            self.last_progress = now;
            acceptance.new_acceptor = Some(AcceptedPeer {
                peer: response.acceptor.clone(),
            });
            acceptance.assigned_location = self.accepted.len() == 1;
        }
        acceptance.satisfied = self.accepted.len() >= self.target_connections;
        acceptance
    }

    pub(crate) fn update_observed_address(&mut self, address: SocketAddr, now: Instant) {
        self.observed_address = Some(address);
        self.last_progress = now;
    }
}

/// Test-only fixture wrapping the relay/joiner state machine.
#[cfg(test)]
#[derive(Debug, Clone)]
pub(crate) struct ConnectOp {
    pub(crate) id: Transaction,
    pub(crate) state: Option<ConnectState>,
    recency: HashMap<PeerKeyLocation, Instant>,
    forward_attempts: HashMap<PeerKeyLocation, ForwardAttempt>,
    connect_forward_estimator: Arc<RwLock<ConnectForwardEstimator>>,
}

#[cfg(test)]
impl ConnectOp {
    fn record_forward_outcome(&mut self, peer: &PeerKeyLocation, desired: Location, success: bool) {
        self.forward_attempts.remove(peer);
        if !success {
            tracing::debug!(
                peer = %peer.pub_key,
                desired = %desired,
                "connect forward failed, recording failure in estimator"
            );
        }
        self.connect_forward_estimator
            .write()
            .record(peer, desired, success);
    }

    pub(crate) fn expire_forward_attempts(&mut self, now: Instant) {
        let mut expired = Vec::with_capacity(self.forward_attempts.len());
        for (peer, attempt) in self.forward_attempts.iter() {
            if now.duration_since(attempt.sent_at) >= FORWARD_ATTEMPT_TIMEOUT {
                expired.push((peer.clone(), attempt.desired));
            }
        }
        // Sort by public key for deterministic processing order
        expired.sort_by(|a, b| a.0.pub_key.cmp(&b.0.pub_key));
        for (peer, desired) in expired {
            if let Some(attempt) = self.forward_attempts.remove(&peer) {
                self.record_forward_outcome(&attempt.peer, desired, false);
            }
        }
    }

    pub(crate) fn new_joiner(
        id: Transaction,
        target_connections: usize,
        observed_address: Option<SocketAddr>,
        connect_forward_estimator: Arc<RwLock<ConnectForwardEstimator>>,
    ) -> Self {
        let started_without_address = observed_address.is_none();
        let state = ConnectState::WaitingForResponses(JoinerState {
            target_connections,
            observed_address,
            accepted: HashSet::new(),
            last_progress: Instant::now(),
            started_without_address,
        });
        Self {
            id,
            state: Some(state),
            recency: HashMap::new(),
            forward_attempts: HashMap::new(),
            connect_forward_estimator,
        }
    }

    /// Construct a relay-side `ConnectOp` from an inbound `Request`.
    pub(crate) fn new_relay(
        id: Transaction,
        upstream_addr: SocketAddr,
        mut request: ConnectRequest,
        connect_forward_estimator: Arc<RwLock<ConnectForwardEstimator>>,
    ) -> Self {
        // Restore bloom filter hash keys from transaction ID after deserialization.
        // The bloom filter uses transaction-specific hash keys for privacy, and these
        // must be restored since hash keys aren't serialized.
        request.visited = request.visited.with_transaction(&id);
        // Clamp uphill_budget to prevent amplification from malicious peers
        // sending inflated values. Without halving, the budget directly controls
        // the number of uphill hops, so we must enforce the cap.
        request.uphill_budget = request.uphill_budget.min(DEFAULT_UPHILL_BUDGET);
        let state = ConnectState::Relaying(Box::new(RelayState {
            upstream_addr,
            request,
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        }));
        Self {
            id,
            state: Some(state),
            recency: HashMap::new(),
            forward_attempts: HashMap::new(),
            connect_forward_estimator,
        }
    }

    pub(crate) fn initiate_join_request(
        own: PeerKeyLocation,
        target: PeerKeyLocation,
        desired_location: Location,
        ttl: u8,
        target_connections: usize,
        connect_forward_estimator: Arc<RwLock<ConnectForwardEstimator>>,
        exclude_addrs: &[SocketAddr],
    ) -> (Transaction, Self, ConnectMsg) {
        let tx = Transaction::new::<ConnectMsg>();
        let msg = prepare_join_request(tx, &own, &target, desired_location, ttl, exclude_addrs);

        let op = ConnectOp::new_joiner(
            tx,
            target_connections,
            own.socket_addr(),
            connect_forward_estimator,
        );

        (tx, op, msg)
    }

    pub(crate) fn handle_request<C: RelayContext>(
        &mut self,
        ctx: &C,
        upstream_addr: SocketAddr,
        request: ConnectRequest,
        estimator: &ConnectForwardEstimator,
        now: Instant,
    ) -> RelayActions {
        self.expire_forward_attempts(now);
        if !matches!(self.state, Some(ConnectState::Relaying(_))) {
            self.state = Some(ConnectState::Relaying(Box::new(RelayState {
                upstream_addr,
                request: request.clone(),
                forwarded_to: None,
                forwarded_at: None,
                observed_sent: false,
                accepted_locally: false,
                response_forwarded: false,
            })));
        }

        match self.state.as_mut() {
            Some(ConnectState::Relaying(state)) => {
                state.upstream_addr = upstream_addr;
                // Restore bloom filter hash keys from the transaction ID before
                // installing the new request on the relay state. The incoming
                // `request` has just been deserialized off the wire, so its
                // `VisitedPeers::hash_keys` are zeroed (`#[serde(skip)]`). Only
                // `ConnectOp::new_relay` re-keys on first receipt; this re-entry
                // path previously assigned the zeroed bloom directly, which
                // corrupted subsequent `mark_visited`/`probably_visited` calls
                // and produced CONNECT forwarding loops at scale (observed in
                // 50-node CI sims as non-monotonic visited set-bit oscillation).
                let mut request = request;
                request.visited = request.visited.with_transaction(&self.id);
                state.request = request;
                state.handle_request(
                    ctx,
                    &self.recency,
                    &mut self.forward_attempts,
                    estimator,
                    now,
                )
            }
            _ => RelayActions::default(),
        }
    }
}

/// Skip list that combines visited peers and own address.
/// Ensures we never select ourselves as a forwarding target and avoids
/// peers already visited in this CONNECT transaction.
struct SkipListWithSelf<'a> {
    visited: &'a VisitedPeers,
    self_addr: Option<SocketAddr>,
}

impl Contains<SocketAddr> for SkipListWithSelf<'_> {
    fn has_element(&self, target: SocketAddr) -> bool {
        if let Some(self_addr) = self.self_addr {
            if target == self_addr {
                return true;
            }
        }
        self.visited.probably_visited(target)
    }
}

impl Contains<&SocketAddr> for SkipListWithSelf<'_> {
    fn has_element(&self, target: &SocketAddr) -> bool {
        self.has_element(*target)
    }
}

fn ring_distance(a: Option<Location>, b: Option<Location>) -> Option<f64> {
    match (a, b) {
        (Some(a), Some(b)) => Some(a.distance(b).as_f64()),
        _ => None,
    }
}

/// Fire the acceptor-side NodeEvents required to promote the joiner's
/// transient transport connection into a ring connection. Must be invoked at
/// every site where a relay decides to accept a CONNECT — the happy-path
/// `handle_request` branch AND the retry branches that reach acceptance after
/// receiving `Rejected` / `ConnectFailed` from an uphill peer. Skipping this
/// call leaves the joiner stuck in `transient_connections` on the acceptor,
/// so downstream lookups like `get_peer_by_addr` (used by subscribe interest
/// registration, broadcast fan-out, etc.) never find the peer.
pub(super) async fn dispatch_expect_connection_from(
    op_manager: &OpManager,
    tx: Transaction,
    peer: PeerKeyLocation,
) -> Result<(), OpError> {
    let Some(addr) = peer.socket_addr() else {
        // Protocol invariant violation: `handle_request` only sets
        // `AcceptOutcome` when the joiner's address has been filled in (the
        // `KnownPeerKeyLocation::try_from` check upstream). Reaching this
        // branch means something broke that invariant. Log loudly rather than
        // silently no-op so we don't reintroduce #3838 in a new form.
        tracing::error!(
            tx = %tx,
            joiner_pub_key = %peer.pub_key(),
            "INTERNAL ERROR: acceptor reached dispatch_expect_connection_from with unknown joiner address — handle_request acceptance invariant violated"
        );
        return Ok(());
    };

    tracing::info!(
        joiner_addr = %addr,
        tx = %tx,
        "connect: acceptor accepted joiner, initiating hole punch"
    );

    op_manager
        .notify_node_event(NodeEvent::ExpectPeerConnection { addr })
        .await?;

    let (callback, mut rx) = mpsc::channel(1);
    op_manager
        .notify_node_event(NodeEvent::ConnectPeer {
            peer: peer.clone(),
            tx,
            callback,
            is_gw: false,
        })
        .await?;

    GlobalExecutor::spawn(async move {
        if let Some(result) = rx.recv().await {
            match result {
                Ok((connected_peer, _)) => {
                    tracing::info!(
                        %connected_peer,
                        tx=%tx,
                        "connect: acceptor hole-punch connection succeeded"
                    );
                }
                Err(_) => {
                    tracing::debug!(
                        %peer,
                        tx=%tx,
                        "connect: acceptor hole-punch connection failed (joiner may connect to us instead)"
                    );
                }
            }
        }
    });

    Ok(())
}

/// Build a `ConnectMsg::Request` for a joiner-initiated CONNECT.
///
/// Pure helper so `op_ctx_task::start_client_connect` and
/// `ring::Ring::acquire_new` can construct the wire message without
/// allocating a legacy `ConnectOp` carrier. Caller owns the
/// `Transaction`.
///
/// Bloom filter saturation note: `VisitedPeers` uses a 512-bit filter
/// with k=4 hash functions. Pre-loading N entries raises the false-positive
/// rate; at typical network sizes (≤ 20 connections in current deployments)
/// the FP rate is negligible (~0.01%). At DEFAULT_MAX_CONNECTIONS (200),
/// the rate reaches ~39%, which would cause routing to occasionally skip
/// reachable peers. This is an accepted trade-off for current network sizes.
pub(crate) fn prepare_join_request(
    tx: Transaction,
    own: &PeerKeyLocation,
    target: &PeerKeyLocation,
    desired_location: Location,
    ttl: u8,
    exclude_addrs: &[SocketAddr],
) -> ConnectMsg {
    let mut visited = VisitedPeers::new(&tx);
    if let Some(own_addr) = own.socket_addr() {
        visited.mark_visited(own_addr);
    }
    if let Some(target_addr) = target.socket_addr() {
        visited.mark_visited(target_addr);
    }
    for addr in exclude_addrs {
        visited.mark_visited(*addr);
    }
    if !exclude_addrs.is_empty() {
        tracing::debug!(
            excluded_addrs = exclude_addrs.len(),
            "connect: pre-populated visited filter with excluded peer addresses (failed + connected)"
        );
    }

    // Create joiner with PeerAddr::Unknown - the joiner doesn't know their own
    // external address (especially behind NAT). The first recipient (gateway)
    // will fill this in from the packet source address.
    let joiner = PeerKeyLocation::with_unknown_addr(own.pub_key.clone());
    let request = ConnectRequest {
        desired_location,
        joiner,
        ttl,
        visited,
        uphill_budget: DEFAULT_UPHILL_BUDGET,
    };

    ConnectMsg::Request {
        id: tx,
        payload: request,
    }
}

#[tracing::instrument(fields(peer = %op_manager.ring.connection_manager.pub_key), skip_all)]
pub(crate) async fn join_ring_request(
    gateway: &PeerKeyLocation,
    op_manager: &OpManager,
    overall_timeout: Option<std::time::Duration>,
) -> Result<(), OpError> {
    use crate::node::ConnectionError;
    let gateway_location = gateway.location().ok_or_else(|| {
        tracing::error!(
            phase = "error",
            "Gateway location not found, this should not be possible, report an error"
        );
        OpError::ConnError(ConnectionError::LocationUnknown)
    })?;

    let gateway_addr = gateway.socket_addr().ok_or_else(|| {
        tracing::error!(phase = "error", "Gateway address not found");
        OpError::ConnError(ConnectionError::LocationUnknown)
    })?;

    if !op_manager
        .ring
        .connection_manager
        .should_accept(gateway_location, gateway_addr)
    {
        return Err(OpError::ConnError(ConnectionError::UnwantedConnection));
    }

    let own = op_manager.ring.connection_manager.own_location();
    let base_location = own.location().unwrap_or_else(Location::random);

    // Capture now once so the jitter failure counter and any other time-sensitive
    // calls in this function share a consistent timestamp.
    let now = Instant::now();

    let current_connections = op_manager.ring.connection_manager.connection_count();

    // Helper: apply jitter to base_location for NAT-compatible acceptor exploration.
    // After consecutive hole-punch failures, the same location routes to the same
    // acceptors. Jittering explores different ring regions.
    let apply_bootstrap_jitter = |base: Location| -> Location {
        let failures = op_manager
            .ring
            .connection_manager
            .increment_connect_jitter_failures(now);
        if failures > 1 {
            // Jitter magnitude: 0.05 * 2^(failures-2), capped at 0.25
            let magnitude = (0.05 * (1u32 << (failures - 2).min(3)) as f64).min(0.25);
            let uniform_01 = (GlobalRng::random_u64() as f64) / (u64::MAX as f64);
            let offset = magnitude * (2.0 * uniform_01 - 1.0);
            let jittered = (base.as_f64() + offset).rem_euclid(1.0);
            tracing::info!(
                failures,
                magnitude,
                base = %base,
                jittered,
                "connect: applying location jitter after consecutive failures"
            );
            Location::new(jittered)
        } else {
            base
        }
    };

    // Choose desired_location based on whether we have enough connections for
    // gap-based targeting (Kleinberg-optimal) or need bootstrap targeting.
    //
    // With connections: use gap_target to find the largest gap in our connection
    // distribution in log-distance space and target its midpoint. This produces
    // Kleinberg-optimal 1/d-distributed connections.
    //
    // Without connections (true bootstrap): target own location with jitter.
    // Own-location targeting spreads requests across gateway neighbors (each
    // joiner has a unique location). Jitter after failures explores different
    // ring regions to find NAT-compatible acceptors.
    // Minimum connections needed for meaningful gap analysis. Independent of
    // min_connections because this is about statistical sample size for the 1/d
    // distribution, not connection targets. 3 points are sufficient to identify
    // the largest gap in log-distance space.
    const GAP_TARGET_THRESHOLD: usize = 3;
    let desired_location = if current_connections >= GAP_TARGET_THRESHOLD {
        // Use gap_target: find the largest gap in our current connections
        // and target its midpoint for Kleinberg-optimal placement.
        if let Some(my_loc) = own.location() {
            let distances: Vec<f64> = op_manager
                .ring
                .connection_manager
                .location_for_all_peers()
                .into_iter()
                .map(|peer_loc| my_loc.distance(peer_loc).as_f64())
                .collect();
            if distances.len() >= GAP_TARGET_THRESHOLD {
                let target =
                    crate::topology::small_world_rand::gap_target(my_loc, distances.into_iter());
                tracing::info!(
                    current_connections,
                    own_location = %my_loc,
                    target = %target,
                    distance = my_loc.distance(target).as_f64(),
                    "connect: using gap_target for Kleinberg-optimal placement"
                );
                target
            } else {
                // Fewer unique neighbor locations than connections (e.g. sybil
                // masking collapses peers to same ring location). Fall back to
                // jitter-based exploration to avoid retrying the same acceptors.
                apply_bootstrap_jitter(base_location)
            }
        } else {
            apply_bootstrap_jitter(base_location)
        }
    } else {
        // Bootstrap mode: target own location with jitter after failures.
        apply_bootstrap_jitter(base_location)
    };
    let tx = Transaction::new::<ConnectMsg>();
    op_ctx_task::start_client_connect(
        tx,
        gateway.clone(),
        gateway_addr,
        op_manager,
        own,
        desired_location,
        overall_timeout,
    )
    .await
}

/// Resolve the socket address of a gateway selected for a version probe.
///
/// Extracted from [`gateway_version_probe`] so the failure mode (gateway with
/// no known address) is unit-testable without an `OpManager`. Caller logs the
/// failure at debug level and retries on the next maintenance cycle.
pub(crate) fn resolve_probe_gateway_addr(
    gateway: &PeerKeyLocation,
) -> Result<std::net::SocketAddr, OpError> {
    use crate::node::ConnectionError;

    gateway.socket_addr().ok_or_else(|| {
        tracing::error!(phase = "error", "Gateway address not found");
        OpError::ConnError(ConnectionError::LocationUnknown)
    })
}

/// Initiate a CONNECT to a gateway for version discovery (#3677).
///
/// Skips `should_accept()` so the probe runs even when the ring is full.
/// The transport handshake exchanges version info via `report_peer_version()`
/// / `signal_urgent_update()`. Called periodically from `connection_maintenance`.
#[tracing::instrument(fields(peer = %op_manager.ring.connection_manager.pub_key), skip_all)]
pub(crate) async fn gateway_version_probe(
    gateway: &PeerKeyLocation,
    op_manager: &OpManager,
) -> Result<(), OpError> {
    let gateway_addr = resolve_probe_gateway_addr(gateway)?;

    let own = op_manager.ring.connection_manager.own_location();
    let desired_location = own.location().unwrap_or_else(Location::random);

    let tx = Transaction::new::<ConnectMsg>();
    op_ctx_task::start_client_connect(
        tx,
        gateway.clone(),
        gateway_addr,
        op_manager,
        own,
        desired_location,
        None,
    )
    .await
}

/// When a gateway is in backoff but the node already has ring connections,
/// we cap the per-iteration gateway-backoff sleep to this value so the loop
/// re-checks `open_conns` frequently.  The concurrent `connection_maintenance`
/// task adds connections via ring-based acquisition every 5s; polling at ~30s
/// intervals lets `initial_join_procedure` notice when `min_connections`
/// is reached without waiting for the full gateway backoff to expire.
///
/// ±20% jitter is applied at each call site to prevent thundering herd.
/// See issue #3304.
pub(crate) const GATEWAY_BACKOFF_POLL_CAP: Duration = Duration::from_secs(30);

pub(crate) async fn initial_join_procedure(
    op_manager: Arc<OpManager>,
    gateways: &[PeerKeyLocation],
) -> Result<JoinHandle<()>, OpError> {
    let number_of_parallel_connections = {
        let max_potential_conns_per_gw = op_manager.ring.max_hops_to_live;
        let needed_to_cover_max =
            op_manager.ring.connection_manager.max_connections / max_potential_conns_per_gw;
        gateways.iter().take(needed_to_cover_max).count().max(2)
    };
    let gateways = gateways.to_vec();
    let handle = GlobalExecutor::spawn(async move {
        if gateways.is_empty() {
            tracing::warn!(
                phase = "error",
                "No gateways available, aborting join procedure"
            );
            return;
        }

        const BASE_WAIT_SECS: u64 = 1;
        const LONG_WAIT_SECS: u64 = 30;
        // Per-peer timeout for cached peer reconnection attempts.
        // Kept short to avoid delaying gateway bootstrap if NAT holes
        // have expired or cached peers are unreachable.
        const CACHED_PEER_TIMEOUT: Duration = Duration::from_secs(5);
        // Use min_connections as the bootstrap threshold so the gateway-directed
        // join loop keeps driving connections until the node has enough peers for
        // ring-based acquisition to take over reliably. Previously hardcoded to 4,
        // which caused nodes to plateau far below min_connections when ring-based
        // acquisition was slow or had insufficient routing candidates.
        //
        // Gateway load note: higher min_connections increases gateway traffic during
        // bootstrap. This is throttled by per-gateway exponential backoff, the
        // parallel connection cap (number_of_parallel_connections), and the
        // LONG_WAIT_SECS sleep once the threshold is reached.
        let bootstrap_threshold = op_manager.ring.connection_manager.min_connections;

        tracing::info!(
            bootstrap_threshold,
            "Starting initial join procedure with {} gateways",
            gateways.len()
        );

        // Attempt fast reconnection to cached peers before gateway bootstrap.
        // After a brief restart, NAT holes to previously-connected peers may
        // still be open. Connecting directly is faster than routing through
        // gateways and avoids gateway load.
        if let Some(ref cache_dir) = op_manager.ring.peer_cache_dir {
            let cache = crate::ring::peer_cache::PeerCache::load(
                cache_dir,
                op_manager.ring.time_source.as_ref(),
            );
            if !cache.peers.is_empty() {
                tracing::info!(
                    cached_peers = cache.peers.len(),
                    "Attempting fast reconnection to cached peers"
                );
                let cached_peer_locs: Vec<PeerKeyLocation> = cache
                    .peers
                    .iter()
                    .take(number_of_parallel_connections)
                    .map(|cp| PeerKeyLocation::new(cp.pub_key.clone(), cp.addr))
                    .collect();

                // Spawn each cached peer reconnection as a separate task for true
                // parallelism (not just cooperative concurrency). Each task has its
                // own timeout so we don't need to track abort handles.
                let mut cached_futs: FuturesUnordered<_> = cached_peer_locs
                    .iter()
                    .map(|peer| {
                        let op_mgr = op_manager.clone();
                        let peer = peer.clone();
                        GlobalExecutor::spawn(async move {
                            // Pass timeout into the driver instead of
                            // wrapping in `tokio::time::timeout`. Outer
                            // wrappers cancel the future mid-flight,
                            // leaving `pending_op_results` slots until
                            // the 60s sweep (#3100). The internal
                            // timeout exits gracefully through the
                            // normal `release_pending_op_slot` cleanup.
                            // Driver returns Ok(()) on internal timeout
                            // — same outcome as the legacy "timed out"
                            // branch.
                            match join_ring_request(&peer, &op_mgr, Some(CACHED_PEER_TIMEOUT)).await
                            {
                                Ok(()) => {
                                    tracing::info!(peer = %peer, "Reconnected to cached peer");
                                    true
                                }
                                Err(e) => {
                                    tracing::debug!(
                                        peer = %peer, error = %e,
                                        "Cached peer reconnection failed"
                                    );
                                    false
                                }
                            }
                        })
                    })
                    .collect();

                let mut successes = 0usize;
                while let Some(result) = cached_futs.next().await {
                    if matches!(result, Ok(true)) {
                        successes += 1;
                    }
                }

                tracing::info!(
                    successes,
                    attempted = cached_peer_locs.len(),
                    "Cached peer reconnection complete"
                );
            }
        }

        loop {
            let open_conns = op_manager.ring.open_connections();

            let unconnected_gateways: Vec<_> =
                op_manager.ring.is_not_connected(gateways.iter()).collect();

            tracing::debug!(
                "Connection status: open_connections = {}, unconnected_gateways = {}",
                open_conns,
                unconnected_gateways.len()
            );

            let unconnected_count = unconnected_gateways.len();

            // When fully isolated (0 connections) but all gateways appear "connected/pending",
            // force-clear stale pending reservations so gateways become retryable immediately.
            // Without this, the peer is stuck for up to PENDING_RESERVATION_TTL (60s) per
            // failed attempt, compounding with gateway backoff to cause 20+ minute recovery
            // delays (#3319). The reservations were created by previous failed join_ring_request()
            // calls and no longer represent real in-progress connections.
            if open_conns == 0 && unconnected_count == 0 {
                let gateway_addrs: Vec<_> =
                    gateways.iter().filter_map(|gw| gw.socket_addr()).collect();
                op_manager
                    .ring
                    .connection_manager
                    .clear_pending_reservations_for(&gateway_addrs);
                tracing::warn!(
                    total_gateways = gateways.len(),
                    "Fully isolated: cleared stale gateway reservations to unblock recovery"
                );
                // Don't clear gateway backoff here — let the normal exponential backoff
                // control retry timing to avoid thundering herd when multiple peers lose
                // connectivity simultaneously. The 120-second isolation recovery in
                // connection_maintenance resets both backoff and reservations periodically.
            } else if open_conns < bootstrap_threshold && unconnected_count == 0 {
                tracing::info!(
                    open_conns,
                    total_gateways = gateways.len(),
                    "Below bootstrap threshold but all gateways appear connected/pending — \
                     waiting for handshakes to complete or pending reservations to expire"
                );
            }

            let use_connected_as_routers = open_conns < bootstrap_threshold
                && unconnected_count == 0
                && bootstrap_threshold.saturating_sub(open_conns) > gateways.len();

            if open_conns < bootstrap_threshold && unconnected_count > 0 {
                // Filter out gateways that are in backoff due to previous failures.
                // This prevents hammering acceptors that consistently fail (e.g., NAT issues).
                //
                // Design note: We track backoff per-gateway because join_ring_request uses
                // the joiner's own desired_location each time. Backing off the gateway gives the
                // network time to stabilize before we retry through that gateway again.
                let (eligible_gateways, min_backoff) = {
                    let backoff = op_manager.gateway_backoff.lock();
                    let mut min_remaining: Option<Duration> = None;
                    let eligible: Vec<_> = unconnected_gateways
                        .into_iter()
                        .filter(|gw| {
                            if let Some(addr) = gw.socket_addr() {
                                if backoff.is_in_backoff(addr) {
                                    // Track the minimum remaining backoff for logging/waiting
                                    if let Some(remaining) = backoff.remaining_backoff(addr) {
                                        min_remaining = Some(
                                            min_remaining
                                                .map(|m| m.min(remaining))
                                                .unwrap_or(remaining),
                                        );
                                    }
                                    tracing::debug!(
                                        gateway = %gw,
                                        "Skipping gateway due to backoff from previous failures"
                                    );
                                    return false;
                                }
                            }
                            true
                        })
                        .collect();
                    (eligible, min_remaining)
                };

                let eligible_count = eligible_gateways.len();

                if eligible_count == 0 {
                    // All gateways are in backoff - wait for the shortest one,
                    // but also wake up if connection_maintenance clears the backoff
                    // (e.g. during isolation recovery or suspend detection).
                    if let Some(min_wait) = min_backoff {
                        // When we already have some ring connections, cap the wait at
                        // GATEWAY_BACKOFF_POLL_CAP so we re-check open_conns frequently.
                        // Apply ±20% jitter (24–36s) to prevent thundering herd.
                        let effective_wait = if open_conns > 0 {
                            let jitter_ms = GlobalRng::random_range(
                                0u64..(GATEWAY_BACKOFF_POLL_CAP.as_millis() / 5) as u64,
                            );
                            let cap = GATEWAY_BACKOFF_POLL_CAP.mul_f64(0.8)
                                + Duration::from_millis(jitter_ms);
                            min_wait.min(cap)
                        } else {
                            min_wait
                        };
                        tracing::info!(
                            wait_secs = effective_wait.as_secs(),
                            gateway_backoff_secs = min_wait.as_secs(),
                            total_gateways = unconnected_count,
                            open_connections = open_conns,
                            "All gateways in backoff, waiting before retry"
                        );
                        tokio::select! {
                            _ = tokio::time::sleep(effective_wait) => {},
                            _ = op_manager.gateway_backoff_cleared.notified() => {
                                tracing::info!("Gateway backoff cleared externally, retrying immediately");
                            },
                        }
                        continue;
                    }
                }

                tracing::info!(
                    "Below bootstrap threshold ({} < {}), attempting to connect to {} gateways (skipped {} in backoff)",
                    open_conns,
                    bootstrap_threshold,
                    number_of_parallel_connections.min(eligible_count),
                    unconnected_count - eligible_count
                );
                let select_all = FuturesUnordered::new();
                for gateway in eligible_gateways
                    .into_iter()
                    .shuffle()
                    .take(number_of_parallel_connections)
                {
                    tracing::info!(%gateway, "Attempting connection to gateway");
                    let op_manager = op_manager.clone();
                    select_all.push(async move {
                        (join_ring_request(gateway, &op_manager, None).await, gateway)
                    });
                }
                select_all
                    .for_each(|(res, gateway)| async move {
                        if let Err(error) = res {
                            if !matches!(
                                error,
                                OpError::ConnError(
                                    crate::node::ConnectionError::UnwantedConnection
                                )
                            ) {
                                tracing::error!(
                                    %gateway,
                                    %error,
                                    "Failed while attempting connection to gateway"
                                );
                            }
                        }
                    })
                    .await;
            } else if use_connected_as_routers {
                // All gateways connected but still need many more peers.
                // Route CONNECTs through connected gateways toward gap locations.
                let eligible: Vec<_> = {
                    let backoff = op_manager.gateway_backoff.lock();
                    gateways
                        .iter()
                        .filter(|gw| {
                            gw.socket_addr()
                                .map(|addr| !backoff.is_in_backoff(addr))
                                .unwrap_or(false)
                        })
                        .collect()
                };

                if !eligible.is_empty() {
                    tracing::info!(
                        eligible = eligible.len(),
                        total_gateways = gateways.len(),
                        "Routing CONNECTs through connected gateways"
                    );
                    let select_all = FuturesUnordered::new();
                    for gateway in eligible
                        .into_iter()
                        .shuffle()
                        .take(number_of_parallel_connections)
                    {
                        let gateway = gateway.clone();
                        let op_manager = op_manager.clone();
                        select_all.push(async move {
                            (
                                join_ring_request(&gateway, &op_manager, None).await,
                                gateway,
                            )
                        });
                    }
                    select_all
                        .for_each(|(res, gateway)| async move {
                            if let Err(error) = res {
                                if !matches!(
                                    error,
                                    OpError::ConnError(
                                        crate::node::ConnectionError::UnwantedConnection
                                    )
                                ) {
                                    tracing::error!(
                                        %gateway,
                                        %error,
                                        "Failed while routing CONNECT through gateway"
                                    );
                                }
                            }
                        })
                        .await;
                }
            } else if open_conns >= bootstrap_threshold {
                tracing::trace!(
                    "Have {} connections (>= threshold of {}), not attempting gateway connections",
                    open_conns,
                    bootstrap_threshold
                );
            }

            // Add random jitter to prevent thundering herd after gateway restart.
            // Without jitter, all peers that lose their gateway connection retry
            // at the same interval, causing synchronized reconnection storms.
            let jitter_ms = GlobalRng::random_u64() % 2000; // 0-2s jitter
            let base_wait_ms = if open_conns == 0 {
                tracing::debug!(
                    "No connections yet, waiting ~{}s before retry",
                    BASE_WAIT_SECS
                );
                BASE_WAIT_SECS * 1000
            } else if open_conns < bootstrap_threshold {
                let wait = BASE_WAIT_SECS * 3 * 1000;
                tracing::debug!(
                    "Have {} connections (below threshold of {}), waiting ~{}ms",
                    open_conns,
                    bootstrap_threshold,
                    wait + jitter_ms
                );
                wait
            } else {
                tracing::trace!(
                    "Connection pool healthy ({} connections), waiting ~{}s",
                    open_conns,
                    LONG_WAIT_SECS
                );
                LONG_WAIT_SECS * 1000
            };

            tokio::time::sleep(Duration::from_millis(base_wait_ms + jitter_ms)).await;
        }
    });
    Ok(handle)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::transport::TransportKeypair;
    use std::net::{IpAddr, Ipv4Addr, SocketAddr};

    #[test]
    fn resolve_probe_gateway_addr_returns_known_socket() {
        let pub_key = TransportKeypair::new().public().clone();
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 12345);
        let gateway = PeerKeyLocation::new(pub_key, addr);

        let resolved = resolve_probe_gateway_addr(&gateway).expect("known address resolves");
        assert_eq!(resolved, addr);
    }

    #[test]
    fn resolve_probe_gateway_addr_errors_on_unknown_address() {
        // The probe loop in connection_maintenance treats this Err as a debug
        // log + continue. The structural guarantee here is that we never call
        // send_gateway_connect with a placeholder address — we surface the
        // error before any network state is mutated.
        let pub_key = TransportKeypair::new().public().clone();
        let gateway = PeerKeyLocation::with_unknown_addr(pub_key);

        let err =
            resolve_probe_gateway_addr(&gateway).expect_err("unknown address must fail to resolve");
        assert!(
            matches!(
                err,
                OpError::ConnError(crate::node::ConnectionError::LocationUnknown)
            ),
            "expected LocationUnknown, got {err:?}"
        );
    }

    struct TestRelayContext {
        self_loc: PeerKeyLocation,
        accept: bool,
        next_hop: Option<PeerKeyLocation>,
        uphill_hop: Option<PeerKeyLocation>,
    }

    impl TestRelayContext {
        fn new(self_loc: PeerKeyLocation) -> Self {
            Self {
                self_loc,
                accept: true,
                next_hop: None,
                uphill_hop: None,
            }
        }

        fn accept(mut self, accept: bool) -> Self {
            self.accept = accept;
            self
        }

        fn next_hop(mut self, hop: Option<PeerKeyLocation>) -> Self {
            self.next_hop = hop;
            self
        }

        fn uphill_hop(mut self, hop: Option<PeerKeyLocation>) -> Self {
            self.uphill_hop = hop;
            self
        }
    }

    impl RelayContext for TestRelayContext {
        fn self_location(&self) -> &PeerKeyLocation {
            &self.self_loc
        }

        fn should_accept(&self, _joiner: &KnownPeerKeyLocation) -> bool {
            self.accept
        }

        fn select_next_hop(
            &self,
            _desired_location: Location,
            _visited: &VisitedPeers,
            _recency: &HashMap<PeerKeyLocation, Instant>,
            _estimator: &ConnectForwardEstimator,
        ) -> Option<PeerKeyLocation> {
            self.next_hop.clone()
        }

        fn select_uphill_hop(
            &self,
            _desired_location: Location,
            _visited: &VisitedPeers,
            _recency: &HashMap<PeerKeyLocation, Instant>,
        ) -> Option<PeerKeyLocation> {
            self.uphill_hop.clone()
        }
    }

    fn make_peer(port: u16) -> PeerKeyLocation {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), port);
        let keypair = TransportKeypair::new();
        PeerKeyLocation::new(keypair.public().clone(), addr)
    }

    #[test]
    fn forward_estimator_handles_missing_location() {
        let mut estimator = ConnectForwardEstimator::new();
        let key = TransportKeypair::new();
        let peer = PeerKeyLocation::new(key.public().clone(), "127.0.0.1:1111".parse().unwrap());
        estimator.record(&peer, Location::new(0.25), true);
    }

    #[test]
    fn expired_forward_attempts_are_cleared() {
        let mut op = ConnectOp::new_joiner(
            Transaction::new::<ConnectMsg>(),
            1,
            None,
            Arc::new(RwLock::new(ConnectForwardEstimator::new())),
        );
        let peer = make_peer(2000);
        op.forward_attempts.insert(
            peer.clone(),
            ForwardAttempt {
                peer: peer.clone(),
                desired: Location::new(0.2),
                sent_at: Instant::now() - FORWARD_ATTEMPT_TIMEOUT - Duration::from_secs(1),
            },
        );
        op.expire_forward_attempts(Instant::now());
        assert!(op.forward_attempts.is_empty());
    }

    #[test]
    fn expired_forward_attempts_record_failures_in_estimator() {
        let estimator = Arc::new(RwLock::new(ConnectForwardEstimator::new()));
        let mut op =
            ConnectOp::new_joiner(Transaction::new::<ConnectMsg>(), 1, None, estimator.clone());
        let peer = make_peer(2000);
        op.forward_attempts.insert(
            peer.clone(),
            ForwardAttempt {
                peer: peer.clone(),
                desired: Location::new(0.2),
                sent_at: Instant::now() - FORWARD_ATTEMPT_TIMEOUT - Duration::from_secs(1),
            },
        );

        let (_, _, events_before, _) = estimator.read().snapshot();
        op.expire_forward_attempts(Instant::now());
        let (_, _, events_after, _) = estimator.read().snapshot();

        assert!(op.forward_attempts.is_empty());
        assert!(
            events_after > events_before,
            "estimator should have recorded failure; before={events_before}, after={events_after}"
        );
    }

    #[test]
    fn relay_accepts_when_policy_allows() {
        let self_loc = make_peer(4000);
        let joiner = make_peer(5000);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc.clone());
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        let accept = actions.accept.expect("expected acceptance");
        // Verify acceptor has correct identity
        assert_eq!(accept.response.acceptor.pub_key(), self_loc.pub_key());
        // Acceptor address should be Unknown - relay fills it in from packet source
        // This is critical for NAT traversal: acceptor doesn't know its external address
        assert!(
            accept.response.acceptor.peer_addr.is_unknown(),
            "ConnectResponse acceptor should have Unknown address for NAT traversal"
        );
        assert_eq!(accept.joiner.pub_key(), joiner.pub_key());
        assert!(actions.forward.is_none());
    }

    #[test]
    fn connect_response_acceptor_starts_with_unknown_address() {
        // Updated from #2207: ConnectResponse.acceptor now starts with Unknown address.
        // The first relay that receives the response fills in the address from the
        // packet source. This is critical for NAT traversal - the acceptor (behind NAT)
        // doesn't know its own external address.
        let self_loc = make_peer(4001);
        let joiner = make_peer(5001);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc.clone());
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        let accept = actions.accept.expect("expected acceptance");

        // Critical invariant: acceptor address must be Unknown initially
        // Relay will fill it in from the packet source address
        assert!(
            accept.response.acceptor.peer_addr.is_unknown(),
            "ConnectResponse.acceptor must have Unknown address for NAT traversal"
        );
        // pub_key should still match
        assert_eq!(
            accept.response.acceptor.pub_key(),
            self_loc.pub_key(),
            "acceptor pub_key should come from self_location"
        );
    }

    #[test]
    fn relay_forwards_when_not_accepting() {
        let self_loc = make_peer(4100);
        let joiner = make_peer(5100);
        let next_hop = make_peer(6100);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 2,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc)
            .accept(false)
            .next_hop(Some(next_hop.clone()));
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        assert!(actions.accept.is_none());
        let (forward_to, request) = actions.forward.expect("expected forward");
        assert_eq!(forward_to.pub_key(), next_hop.pub_key());
        assert_eq!(request.ttl, 1);
        // visited now contains SocketAddr (bloom filter check)
        assert!(
            request
                .visited
                .probably_visited(joiner.socket_addr().expect("test peer must have address"))
        );
    }

    /// Test the terminus-only acceptance behavior: relays should NOT accept when they can
    /// forward to a closer peer, even if should_accept() returns true.
    #[test]
    fn relay_does_not_accept_when_not_at_terminus() {
        let self_loc = make_peer(4150);
        let joiner = make_peer(5150);
        let next_hop = make_peer(6150);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                // Use next_hop's location so it's guaranteed closer to the target
                // than self_loc, ensuring the relay is NOT at terminus.
                // Location::random() was flaky — could land near self_loc,
                // triggering near-terminus acceptance. See #3657.
                desired_location: next_hop.location().expect("test peer has location"),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // should_accept() returns true, but we have a next_hop - so we're NOT at terminus
        let ctx = TestRelayContext::new(self_loc)
            .accept(true) // Relay WANTS to accept
            .next_hop(Some(next_hop.clone())); // But there's a closer peer
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Should NOT accept (not at terminus)
        assert!(
            actions.accept.is_none(),
            "relay should NOT accept when not at terminus, even if should_accept() is true"
        );

        // Should forward to next hop
        let (forward_to, _) = actions
            .forward
            .expect("relay should forward when not at terminus");
        assert_eq!(forward_to.pub_key(), next_hop.pub_key());
    }

    #[test]
    fn relay_emits_observed_address_when_discovering_joiner_addr() {
        // Test the gateway/relay discovering the joiner's external address.
        // The joiner sends ConnectRequest with Unknown address (doesn't know their NAT address).
        // The gateway/relay observes the actual packet source address and sends ObservedAddress back.
        let self_loc = make_peer(4050);
        let joiner_base = make_peer(5050);

        // The joiner's EXTERNAL address as seen by the gateway (from the UDP packet source)
        // This is what the gateway will discover and send back via ObservedAddress
        let external_nat_addr = SocketAddr::new(
            IpAddr::V4(Ipv4Addr::new(203, 0, 113, 10)),
            joiner_base
                .socket_addr()
                .expect("test peer must have address")
                .port(),
        );

        // Create joiner with UNKNOWN address - the joiner doesn't know their external NAT address
        let joiner_with_unknown_addr =
            PeerKeyLocation::with_unknown_addr(joiner_base.pub_key().clone());

        let mut state = RelayState {
            // upstream_addr is the ACTUAL source address from the transport layer
            // This is the joiner's external NAT address as seen by the gateway
            upstream_addr: external_nat_addr,
            request: ConnectRequest {
                desired_location: Location::random(),
                // Joiner has Unknown address in the request
                joiner: joiner_with_unknown_addr.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc);
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Gateway should emit ObservedAddress since it discovered the joiner's external address
        let (target, addr) = actions.observed_address.expect(
            "gateway should emit ObservedAddress when discovering joiner's external address",
        );

        // The address in ObservedAddress should be the external NAT address
        assert_eq!(addr, external_nat_addr);

        // The target should have the discovered address
        assert_eq!(
            target.socket_addr().expect("target must have address"),
            external_nat_addr
        );

        // The request's joiner should be updated with the discovered address
        assert_eq!(
            state
                .request
                .joiner
                .socket_addr()
                .expect("joiner must have address after discovery"),
            external_nat_addr
        );
    }

    #[test]
    fn relay_does_not_emit_observed_address_when_joiner_addr_already_known() {
        // Test that ObservedAddress is NOT emitted when the joiner already has a known address.
        // This happens for non-first hops in the routing path.
        let self_loc = make_peer(4051);
        let joiner_base = make_peer(5051);
        let known_addr = SocketAddr::new(
            IpAddr::V4(Ipv4Addr::new(203, 0, 113, 11)),
            joiner_base
                .socket_addr()
                .expect("test peer must have address")
                .port(),
        );

        // Joiner already has a Known address (filled in by the gateway/first hop)
        let joiner_with_known_addr =
            PeerKeyLocation::new(joiner_base.pub_key().clone(), known_addr);

        let mut state = RelayState {
            upstream_addr: known_addr,
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner_with_known_addr.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc);
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Should NOT emit ObservedAddress since the address was already known
        assert!(
            actions.observed_address.is_none(),
            "non-first hop should NOT emit ObservedAddress when joiner addr already known"
        );
    }

    #[test]
    fn joiner_tracks_acceptance() {
        let acceptor = make_peer(7000);
        let mut state = JoinerState {
            target_connections: 1,
            observed_address: None,
            accepted: HashSet::new(),
            last_progress: Instant::now(),
            started_without_address: true,
        };

        let response = ConnectResponse {
            acceptor: acceptor.clone(),
        };
        let result = state.register_acceptance(&response, Instant::now());
        assert!(result.satisfied);
        let new = result.new_acceptor.expect("expected new acceptor");
        assert_eq!(new.peer.pub_key(), acceptor.pub_key());
    }

    #[test]
    fn init_join_request_initializes_state() {
        let target = make_peer(7200);
        let desired = Location::random();
        let ttl = 5;
        let own = make_peer(7300);
        let (tx, op, msg) = ConnectOp::initiate_join_request(
            own.clone(),
            target.clone(),
            desired,
            ttl,
            2,
            Arc::new(RwLock::new(ConnectForwardEstimator::new())),
            &[],
        );

        match msg {
            ConnectMsg::Request { payload, .. } => {
                assert_eq!(payload.desired_location, desired);
                assert_eq!(payload.ttl, ttl);
                // visited is now a bloom filter, not PeerKeyLocation
                // Restore hash keys to check (simulating what relay does)
                let visited = payload.visited.with_transaction(&tx);
                if let Some(own_addr) = own.socket_addr() {
                    assert!(visited.probably_visited(own_addr));
                }
                if let Some(target_addr) = target.socket_addr() {
                    assert!(visited.probably_visited(target_addr));
                }
            }
            other @ ConnectMsg::Response { .. }
            | other @ ConnectMsg::ObservedAddress { .. }
            | other @ ConnectMsg::Rejected { .. }
            | other @ ConnectMsg::ConnectFailed { .. } => panic!("unexpected message: {other:?}"),
        }

        assert!(matches!(
            op.state,
            Some(ConnectState::WaitingForResponses(_))
        ));
    }

    #[test]
    fn multi_hop_forward_path_tracks_ttl_and_visited_peers() {
        let joiner = make_peer(8100);
        let relay_a = make_peer(8200);
        let relay_b = make_peer(8300);

        let joiner_addr = joiner.socket_addr().expect("test peer must have address");
        let tx = Transaction::new::<ConnectMsg>();

        // Create bloom filter with joiner's address marked
        let mut visited = VisitedPeers::new(&tx);
        visited.mark_visited(joiner_addr);

        let request = ConnectRequest {
            desired_location: Location::random(),
            joiner: joiner.clone(),
            ttl: 3,
            visited,
            uphill_budget: DEFAULT_UPHILL_BUDGET,
        };

        let mut relay_op = ConnectOp::new_relay(
            tx,
            joiner_addr,
            request.clone(),
            Arc::new(RwLock::new(ConnectForwardEstimator::new())),
        );
        let ctx = TestRelayContext::new(relay_a.clone())
            .accept(false)
            .next_hop(Some(relay_b.clone()));
        let estimator = ConnectForwardEstimator::new();
        let actions = relay_op.handle_request(
            &ctx,
            joiner_addr,
            request.clone(),
            &estimator,
            Instant::now(),
        );

        let (forward_target, forward_request) = actions
            .forward
            .expect("relay should forward when it declines to accept");
        assert_eq!(forward_target.pub_key(), relay_b.pub_key());
        assert_eq!(forward_request.ttl, 2);
        let relay_a_addr = relay_a.socket_addr().expect("test peer must have address");
        assert!(
            forward_request.visited.probably_visited(relay_a_addr),
            "forwarded request should record intermediate relay's address"
        );

        // Second hop should accept and notify the joiner.
        let mut accepting_relay = ConnectOp::new_relay(
            tx,
            relay_a_addr,
            forward_request.clone(),
            Arc::new(RwLock::new(ConnectForwardEstimator::new())),
        );
        let ctx_accept = TestRelayContext::new(relay_b.clone());
        let estimator = ConnectForwardEstimator::new();
        let accept_actions = accepting_relay.handle_request(
            &ctx_accept,
            relay_a_addr,
            forward_request,
            &estimator,
            Instant::now(),
        );

        let accept = accept_actions
            .accept
            .expect("second relay should accept when policy allows");
        // Compare pub_key since acceptor's address is intentionally Unknown (NAT scenario)
        assert_eq!(accept.response.acceptor.pub_key(), relay_b.pub_key());
        assert_eq!(accept.joiner.pub_key(), joiner.pub_key());
    }

    /// Regression test for issue #2141: expect_connection_from must have the joiner's
    /// observed external address for NAT hole-punching to work.
    #[test]
    fn connect_expect_connection_uses_observed_address() {
        // Joiner behind NAT: original creation used private address, but the network bridge
        // fills in the observed public address from the packet source.
        let private_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 9000);
        let keypair = TransportKeypair::new();
        let joiner_original = PeerKeyLocation::new(keypair.public().clone(), private_addr);

        // Gateway observes joiner's public/external address and fills it into joiner.peer_addr
        let observed_public_addr =
            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 50)), 9000);
        let joiner_with_observed_addr =
            PeerKeyLocation::new(keypair.public().clone(), observed_public_addr);

        let relay = make_peer(5000);

        let mut state = RelayState {
            upstream_addr: private_addr, // The address we received the request from
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner_with_observed_addr.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(relay.clone());
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Verify acceptance was issued and joiner carries the observed public
        // address (so the acceptor can hole-punch to the correct address).
        let accept = actions
            .accept
            .expect("expect_connection_from should be set when accepting");
        assert_eq!(
            accept
                .joiner
                .socket_addr()
                .expect("expect_connection_from must have address"),
            observed_public_addr,
            "expect_connection_from must use observed external address ({}) not private address ({})",
            observed_public_addr,
            private_addr
        );

        // Double-check: the original joiner had the private address
        assert_eq!(
            joiner_original
                .socket_addr()
                .expect("original joiner must have address"),
            private_addr,
            "original joiner should have private address"
        );
    }

    // Note: The SkipListWithSelf test has been removed as it now requires a ConnectionManager
    // to look up peers by address. The skip list behavior is tested via integration tests
    // and the self-exclusion logic is straightforward.

    /// Regression test: ConnectResponse acceptor must have Unknown address initially.
    /// The relay that first receives the response fills in the acceptor's address from
    /// the packet source. This is critical for NAT traversal - the acceptor (behind NAT)
    /// doesn't know its own external address.
    ///
    /// Bug scenario this prevents:
    /// 1. Peer1 (behind NAT at 192.168.1.x) accepts joiner and creates ConnectResponse
    /// 2. If acceptor uses self_location(), it has 127.0.0.1 or private IP
    /// 3. Joiner receives response with wrong address, can't connect back
    /// 4. NAT hole-punching fails - joiner sends to wrong address
    ///
    /// Correct behavior:
    /// 1. Peer1 creates ConnectResponse with acceptor.peer_addr = Unknown
    /// 2. First relay receives response, fills in acceptor address from UDP source
    /// 3. Joiner receives response with correct external address
    /// 4. NAT hole-punching works - both peers have each other's external addresses
    #[test]
    fn connect_response_acceptor_has_unknown_address() {
        let joiner = make_peer(9100);
        let acceptor_peer = make_peer(9200);

        let joiner_addr = joiner.socket_addr().expect("test peer must have address");
        let tx = Transaction::new::<ConnectMsg>();

        // Create bloom filter with joiner's address marked
        let mut visited = VisitedPeers::new(&tx);
        visited.mark_visited(joiner_addr);

        let request = ConnectRequest {
            desired_location: Location::random(),
            joiner: joiner.clone(),
            ttl: 3,
            visited,
            uphill_budget: DEFAULT_UPHILL_BUDGET,
        };

        let mut relay_op = ConnectOp::new_relay(
            tx,
            joiner_addr,
            request.clone(),
            Arc::new(RwLock::new(ConnectForwardEstimator::new())),
        );

        // Acceptor context - this peer will accept the joiner
        let ctx = TestRelayContext::new(acceptor_peer.clone());
        let estimator = ConnectForwardEstimator::new();
        let actions = relay_op.handle_request(
            &ctx,
            joiner_addr,
            request.clone(),
            &estimator,
            Instant::now(),
        );

        // Verify acceptance was issued
        let accept = actions
            .accept
            .expect("acceptor should issue ConnectResponse");

        // CRITICAL: acceptor's address must be Unknown, not the acceptor's local address
        assert!(
            accept.response.acceptor.peer_addr.is_unknown(),
            "acceptor address should be Unknown for NAT traversal. Got: {:?}",
            accept.response.acceptor.peer_addr
        );

        // The pub_key should be set correctly
        assert_eq!(
            accept.response.acceptor.pub_key(),
            acceptor_peer.pub_key(),
            "acceptor pub_key should match"
        );
    }

    /// Test that a relay does NOT accept after it has already forwarded, even on retry.
    ///
    /// This tests the "double-accept vulnerability" fix: once we commit to forwarding,
    /// we should NOT accept on subsequent handle_request() calls, even if is_terminus
    /// becomes true (because we already forwarded, so select_next_hop returns None).
    ///
    /// Bug scenario this prevents:
    /// 1. First handle_request(): We forward to next_hop, set forwarded_to = Some(...)
    /// 2. Second handle_request(): can_forward = false (already forwarded), so next_hop = None
    /// 3. is_terminus = true (because next_hop.is_none())
    /// 4. BUG: Without the fix, we would accept (is_terminus && !accepted_locally && should_accept)
    /// 5. This violates the "accept OR forward, not both" invariant
    #[test]
    fn relay_does_not_accept_after_forwarding_on_retry() {
        let self_loc = make_peer(4200);
        let joiner = make_peer(5200);
        let next_hop = make_peer(6200);

        // Pin the invariant the fix relies on: `desired_location` must be
        // far enough from `self_loc` that the near-terminus probabilistic
        // accept branch (line ~624: `d < NEAR_TERMINUS_DISTANCE`) is
        // statically unreachable. Using `next_hop`'s location makes this
        // deterministic — but only because `make_peer` derives location
        // from the loopback port. If port choices, `Location::from_address`,
        // or `NEAR_TERMINUS_DISTANCE` ever change such that the gap shrinks,
        // fail loudly here rather than silently re-flake. See #3944.
        let target_loc = next_hop.location().expect("test peer has location");
        let self_to_target = self_loc
            .location()
            .expect("test peer has location")
            .distance(target_loc)
            .as_f64();
        assert!(
            self_to_target > NEAR_TERMINUS_DISTANCE,
            "test invariant: self_loc must be > NEAR_TERMINUS_DISTANCE ({NEAR_TERMINUS_DISTANCE}) from desired_location, got {self_to_target}"
        );

        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: target_loc,
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // First call: should forward (next_hop available)
        let ctx = TestRelayContext::new(self_loc.clone())
            .accept(true) // would accept if at terminus
            .next_hop(Some(next_hop.clone()));
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();

        let actions1 = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Verify we forwarded, not accepted
        assert!(
            actions1.accept.is_none(),
            "first call should forward, not accept"
        );
        assert!(
            actions1.forward.is_some(),
            "first call should produce a forward action"
        );
        assert!(
            state.forwarded_to.is_some(),
            "forwarded_to should be set after forwarding"
        );

        // Second call: simulate retry after forwarding
        // Now select_next_hop returns None (no next hop), so is_terminus = true
        // But we've already forwarded, so we should NOT accept
        let ctx_retry = TestRelayContext::new(self_loc)
            .accept(true) // would accept if allowed
            .next_hop(None); // no next hop available anymore

        let actions2 = state.handle_request(
            &ctx_retry,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // CRITICAL: Should NOT accept (we already forwarded)
        assert!(
            actions2.accept.is_none(),
            "second call should NOT accept even though is_terminus=true (already forwarded)"
        );

        // Should NOT forward again either (already forwarded)
        assert!(
            actions2.forward.is_none(),
            "second call should not forward again"
        );
    }

    /// Regression for #3944: directly exercises the "already forwarded → don't
    /// accept" guard at the terminus-acceptance check without going through the
    /// forwarding path first. The companion test
    /// `relay_does_not_accept_after_forwarding_on_retry` was previously flaky
    /// because its first call exercised near-terminus probabilistic acceptance
    /// (driven by `GlobalRng`) before `forwarded_to` was set, sometimes
    /// triggering acceptance regardless of the post-forward guard.
    ///
    /// This test sets `forwarded_to` and `forwarded_at` in the initial
    /// `RelayState` so the only path under test is the terminus-acceptance
    /// gate, which is fully deterministic. It would have failed before the
    /// fix at `is_terminus && !self.accepted_locally && self.forwarded_to.is_none()`
    /// was introduced, and is unaffected by RNG state.
    #[test]
    fn relay_with_forwarded_state_rejects_acceptance_at_terminus() {
        let self_loc = make_peer(4250);
        let joiner = make_peer(5250);
        let prior_next_hop = make_peer(6250);

        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: prior_next_hop.location().expect("test peer has location"),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            // Simulate a prior forward that completed: the operation has
            // already committed to a path and must not also accept locally.
            forwarded_to: Some(prior_next_hop),
            forwarded_at: Some(Instant::now()),
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // Terminus (next_hop=None) with should_accept=true — every condition
        // that would normally green-light acceptance is in place EXCEPT the
        // forwarded_to guard.
        let ctx = TestRelayContext::new(self_loc).accept(true).next_hop(None);
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();

        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        assert!(
            actions.accept.is_none(),
            "relay must not accept once forwarded_to is set, even at terminus"
        );
        assert!(
            actions.forward.is_none(),
            "relay must not forward again when forwarded_to is already set"
        );
    }

    /// Regression: when a relay op is re-entered for the same transaction with a
    /// freshly deserialized `ConnectRequest`, `ConnectOp::handle_request` must restore
    /// the `VisitedPeers` bloom filter's transaction-derived hash keys before using
    /// the new request as relay state. `VisitedPeers::hash_keys` is `#[serde(skip)]`,
    /// so an incoming request has zeroed hash keys. `ConnectOp::new_relay` re-keys
    /// correctly, but the `handle_request` re-entry path at connect.rs used to assign
    /// the incoming request directly to `state.request` without re-keying. Subsequent
    /// `mark_visited`/`probably_visited` calls then ran against zero hash keys and
    /// silently corrupted the bloom, producing a CONNECT forwarding loop at scale
    /// (observed in 50-node CI simulations as non-monotonic visited set-bit
    /// oscillation on a single transaction).
    #[test]
    fn regression_visited_hash_keys_preserved_across_reentry() {
        let self_loc = make_peer(4800);
        let upstream = make_peer(4801);
        let joiner = make_peer(4802);
        let marked_peer = make_peer(4803);

        let tx = Transaction::new::<ConnectMsg>();

        // Build an initial ConnectRequest as it would look when first arriving over
        // the wire: bloom has been keyed (with_transaction) and a peer has been marked.
        let mut visited = VisitedPeers::new(&tx);
        let marked_addr = marked_peer
            .socket_addr()
            .expect("test peer must have address");
        visited.mark_visited(marked_addr);
        assert!(
            visited.probably_visited(marked_addr),
            "precondition: marked peer must be detectable before the wire round-trip"
        );

        let request = ConnectRequest {
            desired_location: Location::random(),
            joiner: joiner.clone(),
            ttl: 3,
            visited,
            uphill_budget: DEFAULT_UPHILL_BUDGET,
        };

        // Create the relay op as `new_relay` would on first receipt.
        let estimator_arc = Arc::new(RwLock::new(ConnectForwardEstimator::new()));
        let mut op = ConnectOp::new_relay(
            tx,
            upstream.socket_addr().expect("test peer must have address"),
            request.clone(),
            estimator_arc,
        );

        // Simulate the request making another hop and re-entering this relay: it
        // crosses the wire, so serde round-trips through `bincode`, which zeros out
        // `VisitedPeers::hash_keys` (marked `#[serde(skip)]`).
        let wire_bytes = bincode::serialize(&request).expect("serialize ConnectRequest");
        let reentered: ConnectRequest =
            bincode::deserialize(&wire_bytes).expect("deserialize ConnectRequest");

        // Sanity: the freshly deserialized bloom has zeroed hash keys, so the
        // previously-marked peer is NOT detectable through it directly.
        assert!(
            !reentered.visited.probably_visited(marked_addr),
            "sanity: deserialized bloom must fail to detect marked peer without re-keying (hash_keys are #[serde(skip)])"
        );

        // Re-enter `ConnectOp::handle_request` with the deserialized request. We do
        // NOT want the relay to take any meaningful action (forward/accept); all we
        // care about is that `state.request.visited` ends up with the correct hash
        // keys restored so subsequent visited-set operations behave correctly.
        let ctx = TestRelayContext::new(self_loc.clone())
            .accept(false)
            .next_hop(None);
        let estimator = ConnectForwardEstimator::new();
        let _actions = op.handle_request(
            &ctx,
            upstream.socket_addr().expect("test peer must have address"),
            reentered,
            &estimator,
            Instant::now(),
        );

        // After the re-entry, the relay state's bloom must still correctly detect the
        // originally-marked peer. Without the fix, hash_keys are (0, 0) and the bits
        // set during the initial mark hash to entirely different indices, so this
        // check fails — evidencing bloom corruption.
        match op.state.as_ref() {
            Some(ConnectState::Relaying(state)) => {
                assert!(
                    state.request.visited.probably_visited(marked_addr),
                    "bloom corruption after re-entry: previously-marked peer no longer detected. \
                     VisitedPeers::hash_keys were not restored via with_transaction() when \
                     assigning the deserialized request to state.request."
                );
            }
            other => panic!("expected Relaying state after handle_request, got {other:?}"),
        }
    }

    /// Test that greedy routing excludes peers farther from target than ourselves.
    /// This is tested indirectly through the terminus behavior - if we have no closer
    /// peers, we should be at terminus.
    #[test]
    fn terminus_reached_when_no_closer_peers() {
        let self_loc = make_peer(4300);
        let joiner = make_peer(5300);

        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // No next hop available (we're at terminus)
        let ctx = TestRelayContext::new(self_loc).accept(true).next_hop(None); // No closer peers

        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();

        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Should accept because we're at terminus
        assert!(
            actions.accept.is_some(),
            "should accept at terminus (no closer peers)"
        );

        // Should not forward
        assert!(actions.forward.is_none(), "should not forward at terminus");
    }

    /// Test TTL exhaustion: when TTL reaches 0, we can't forward even if next_hop exists.
    /// This should result in terminus behavior.
    #[test]
    fn ttl_zero_prevents_forwarding() {
        let self_loc = make_peer(4400);
        let joiner = make_peer(5400);
        let next_hop = make_peer(6400);

        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 0, // TTL exhausted
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // Next hop exists, but TTL is 0
        let ctx = TestRelayContext::new(self_loc)
            .accept(true)
            .next_hop(Some(next_hop)); // Would forward if TTL > 0

        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();

        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Should NOT forward (TTL = 0)
        assert!(
            actions.forward.is_none(),
            "should not forward when TTL is 0"
        );

        // Should accept (at terminus due to TTL exhaustion)
        assert!(
            actions.accept.is_some(),
            "should accept at terminus (TTL exhausted)"
        );
    }

    /// Regression test for issue #2420: When at terminus but should_accept() returns false,
    /// the request should route uphill (away from target) instead of being silently dropped.
    ///
    /// Bug scenario this prevents:
    /// 1. Peer 20 sends ConnectRequest targeting location 0.226
    /// 2. Gateway is at terminus (closest to target)
    /// 3. Gateway at capacity, should_accept() returns false
    /// 4. BUG: Request dropped silently, Peer 20 remains isolated for ~10 minutes
    ///
    /// Correct behavior:
    /// 1-3. Same as above
    /// 4. Gateway routes uphill (farther from target) to give other peers a chance to accept
    #[test]
    fn uphill_routing_when_at_terminus_but_cannot_accept() {
        let self_loc = make_peer(4500);
        let joiner = make_peer(5500);
        let uphill_peer = make_peer(6500);

        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // At terminus (no next_hop), cannot accept, but uphill_hop available
        let ctx = TestRelayContext::new(self_loc)
            .accept(false) // Cannot accept (at capacity)
            .next_hop(None) // At terminus - no closer peer
            .uphill_hop(Some(uphill_peer.clone())); // But there's an uphill peer

        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();

        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Should NOT accept (should_accept returned false)
        assert!(
            actions.accept.is_none(),
            "should not accept when should_accept() returns false"
        );

        // CRITICAL: Should forward uphill instead of dropping
        let (forward_to, request) = actions
            .forward
            .expect("should route uphill when at terminus but cannot accept");
        assert_eq!(forward_to.pub_key(), uphill_peer.pub_key());
        assert_eq!(
            request.ttl, 2,
            "TTL should be decremented by 1 only (no extra burn): 3 - 1 = 2"
        );
        assert_eq!(
            request.uphill_budget,
            DEFAULT_UPHILL_BUDGET - 1,
            "uphill_budget should be decremented by 1 on uphill forward"
        );

        // forwarded_to should be set
        assert!(
            state.forwarded_to.is_some(),
            "forwarded_to should be set after uphill routing"
        );
    }

    /// Test that no forwarding happens when at terminus, cannot accept, and no uphill peers available.
    #[test]
    fn no_forward_when_at_terminus_cannot_accept_and_no_uphill_peers() {
        let self_loc = make_peer(4600);
        let joiner = make_peer(5600);

        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // At terminus, cannot accept, no uphill peers
        let ctx = TestRelayContext::new(self_loc)
            .accept(false) // Cannot accept
            .next_hop(None) // At terminus
            .uphill_hop(None); // No uphill peers either

        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();

        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Should NOT accept
        assert!(
            actions.accept.is_none(),
            "should not accept when should_accept() returns false"
        );

        // Should NOT forward (no uphill peers available)
        assert!(
            actions.forward.is_none(),
            "should not forward when no uphill peers available"
        );
    }

    /// Test that uphill routing doesn't happen when TTL is exhausted.
    #[test]
    fn no_uphill_routing_when_ttl_exhausted() {
        let self_loc = make_peer(4700);
        let joiner = make_peer(5700);
        let uphill_peer = make_peer(6700);

        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 0, // TTL exhausted
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // At terminus, cannot accept, uphill peer exists but TTL is 0
        let ctx = TestRelayContext::new(self_loc)
            .accept(false) // Cannot accept
            .next_hop(None) // At terminus
            .uphill_hop(Some(uphill_peer)); // Uphill peer exists

        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();

        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        // Should NOT accept
        assert!(
            actions.accept.is_none(),
            "should not accept when should_accept() returns false"
        );

        // Should NOT forward (TTL exhausted)
        assert!(
            actions.forward.is_none(),
            "should not forward uphill when TTL is exhausted"
        );
    }

    #[test]
    fn no_uphill_routing_when_budget_exhausted() {
        let self_loc = make_peer(4750);
        let joiner = make_peer(5750);
        let uphill_peer = make_peer(6750);

        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 5, // TTL is fine
                visited: VisitedPeers::default(),
                uphill_budget: 0, // Budget exhausted
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc)
            .accept(false)
            .next_hop(None)
            .uphill_hop(Some(uphill_peer));

        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();

        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        assert!(
            actions.accept.is_none(),
            "should not accept when should_accept() returns false"
        );
        assert!(
            actions.forward.is_none(),
            "should not forward uphill when budget is exhausted"
        );
        assert!(actions.rejected, "should reject when budget is exhausted");
    }

    // NOTE: A test for joiner_known = None was considered, but this edge case cannot
    // actually occur in normal operation. The handle_request function automatically fills
    // in the joiner's address from upstream_addr if it's unknown (lines 385-390):
    //
    //   if self.request.joiner.peer_addr.is_unknown() {
    //       self.request.joiner.set_addr(self.upstream_addr);
    //   }
    //
    // The joiner_known = None check (line 462-475) is purely defensive - it handles the
    // theoretical case where the address somehow doesn't get filled in, but this cannot
    // be triggered in practice since the first relay always fills it in.

    // =============================================================================
    // Bloom Filter (VisitedPeers) Regression Tests
    // These tests ensure the bloom filter implementation correctly tracks visited
    // peers during the connect operation, preventing routing loops.
    // =============================================================================

    /// Test that VisitedPeers correctly tracks visited addresses.
    /// Regression test for issue #2421: Connect operation should use bloom filter.
    #[test]
    fn test_visited_peers_tracks_addresses() {
        let tx = Transaction::new::<ConnectMsg>();
        let mut visited = VisitedPeers::new(&tx);

        let addr1: SocketAddr = "127.0.0.1:8001".parse().unwrap();
        let addr2: SocketAddr = "127.0.0.1:8002".parse().unwrap();
        let addr3: SocketAddr = "127.0.0.1:8003".parse().unwrap();

        // Initially no addresses are visited
        assert!(!visited.probably_visited(addr1));
        assert!(!visited.probably_visited(addr2));
        assert!(!visited.probably_visited(addr3));

        // Mark addresses as visited
        visited.mark_visited(addr1);
        visited.mark_visited(addr2);

        // Visited addresses should be detected
        assert!(visited.probably_visited(addr1));
        assert!(visited.probably_visited(addr2));
        // Unvisited address should not be detected (with high probability)
        assert!(!visited.probably_visited(addr3));
    }

    /// Test that VisitedPeers hash keys are correctly restored after deserialization.
    /// Regression test for issue #2421: Hash keys must be restored using with_transaction().
    #[test]
    fn test_visited_peers_hash_key_restoration() {
        let tx = Transaction::new::<ConnectMsg>();
        let mut original = VisitedPeers::new(&tx);

        let addr: SocketAddr = "192.168.1.1:9000".parse().unwrap();
        original.mark_visited(addr);

        // Simulate serialization/deserialization using bincode (same as network)
        let serialized = bincode::serialize(&original).unwrap();
        let mut restored: VisitedPeers = bincode::deserialize(&serialized).unwrap();

        // Before restoring hash keys, the check may fail due to zeroed hash keys
        // (This is implementation-specific - hash_keys are skipped during serialization)

        // After restoration with transaction, the address should be detected
        restored = restored.with_transaction(&tx);
        assert!(
            restored.probably_visited(addr),
            "visited address should be detected after hash key restoration"
        );
    }

    /// Test that bloom filter is properly initialized in initiate_join_request.
    /// Regression test for issue #2421: Joiner should initialize bloom filter with tx.
    #[test]
    fn test_initiate_join_request_uses_bloom_filter() {
        let own = make_peer(8001);
        let target = make_peer(8002);
        let desired_location = Location::try_from(0.5).unwrap();
        let estimator = Arc::new(RwLock::new(ConnectForwardEstimator::new()));

        let (tx, _op, msg) = ConnectOp::initiate_join_request(
            own.clone(),
            target.clone(),
            desired_location,
            10,
            3,
            estimator,
            &[],
        );

        // Extract the request from the message
        let request = match msg {
            ConnectMsg::Request { payload, .. } => payload,
            ConnectMsg::Response { .. }
            | ConnectMsg::ObservedAddress { .. }
            | ConnectMsg::Rejected { .. }
            | ConnectMsg::ConnectFailed { .. } => panic!("Expected ConnectMsg::Request"),
        };

        // Restore hash keys to check (simulating what relay does)
        let visited = request.visited.with_transaction(&tx);

        // Own address and target address should be marked as visited
        if let Some(own_addr) = own.socket_addr() {
            assert!(
                visited.probably_visited(own_addr),
                "own address should be marked visited in initial request"
            );
        }
        if let Some(target_addr) = target.socket_addr() {
            assert!(
                visited.probably_visited(target_addr),
                "target address should be marked visited in initial request"
            );
        }
    }

    /// Test that SkipListWithSelf correctly uses bloom filter for visited detection.
    /// Regression test for issue #2421: Skip list should use probably_visited.
    #[test]
    fn test_skip_list_with_bloom_filter() {
        let tx = Transaction::new::<ConnectMsg>();
        let mut visited = VisitedPeers::new(&tx);

        let visited_addr: SocketAddr = "10.0.0.1:5000".parse().unwrap();
        let self_addr: SocketAddr = "10.0.0.2:5000".parse().unwrap();
        let other_addr: SocketAddr = "10.0.0.3:5000".parse().unwrap();

        visited.mark_visited(visited_addr);

        let skip = SkipListWithSelf {
            visited: &visited,
            self_addr: Some(self_addr),
        };

        // Should skip visited addresses
        assert!(
            skip.has_element(visited_addr),
            "visited address should be in skip list"
        );
        // Should skip self address
        assert!(
            skip.has_element(self_addr),
            "self address should be in skip list"
        );
        // Should not skip other addresses
        assert!(
            !skip.has_element(other_addr),
            "unvisited address should not be in skip list"
        );
    }

    // ============ Rejection flag tests ============

    #[test]
    fn relay_rejects_when_no_uphill_peers_available() {
        // At terminus, cannot accept, no uphill peers → rejected = true
        let self_loc = make_peer(4000);
        let joiner = make_peer(5000);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc)
            .accept(false)
            .uphill_hop(None);
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        assert!(actions.rejected, "should reject when no uphill peers");
        assert!(actions.accept.is_none());
        assert!(actions.forward.is_none());
    }

    #[test]
    fn relay_rejects_when_ttl_exhausted() {
        // At terminus, cannot accept, TTL exhausted → rejected = true
        let self_loc = make_peer(4000);
        let joiner = make_peer(5000);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 0, // TTL exhausted
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc).accept(false);
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        assert!(actions.rejected, "should reject when TTL exhausted");
        assert!(actions.accept.is_none());
        assert!(actions.forward.is_none());
    }

    #[test]
    fn relay_routes_uphill_when_available() {
        // At terminus, cannot accept, uphill peer available → rejected = false, forward.is_some()
        let self_loc = make_peer(4000);
        let joiner = make_peer(5000);
        let uphill = make_peer(6000);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        let ctx = TestRelayContext::new(self_loc)
            .accept(false)
            .uphill_hop(Some(uphill));
        let recency = HashMap::new();
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let actions = state.handle_request(
            &ctx,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        assert!(!actions.rejected, "should not reject when uphill available");
        assert!(actions.forward.is_some(), "should forward uphill");
        assert!(actions.accept.is_none());
        assert!(
            state.forwarded_at.is_some(),
            "forwarded_at should be set after forwarding"
        );
    }

    #[test]
    fn relay_retries_different_uphill_peer_after_rejection() {
        // Simulate the retry sequence that process_message performs on Rejected:
        // 1. Forward to peer_a (uphill)
        // 2. Rejection arrives → reset forwarded_to, call handle_request again
        // 3. Context provides peer_b → should forward to peer_b
        let self_loc = make_peer(4000);
        let joiner = make_peer(5000);
        let peer_a = make_peer(6000);
        let peer_b = make_peer(7000);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // Step 1: Initial forward to peer_a
        let ctx_a = TestRelayContext::new(self_loc.clone())
            .accept(false)
            .uphill_hop(Some(peer_a.clone()));
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let recency = HashMap::new();
        let actions = state.handle_request(
            &ctx_a,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );
        assert!(actions.forward.is_some(), "should forward to peer_a");
        assert_eq!(
            state.forwarded_to.as_ref().unwrap().pub_key(),
            peer_a.pub_key()
        );

        // Step 2: Simulate rejection — reset forwarded_to (as process_message does)
        state.forwarded_to = None;
        state.forwarded_at = None;

        // Step 3: Retry with peer_b available
        let ctx_b = TestRelayContext::new(self_loc)
            .accept(false)
            .uphill_hop(Some(peer_b.clone()));
        let retry_actions = state.handle_request(
            &ctx_b,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );
        assert!(
            retry_actions.forward.is_some(),
            "should forward to peer_b on retry"
        );
        let (fwd_peer, _) = retry_actions.forward.unwrap();
        assert_eq!(fwd_peer.pub_key(), peer_b.pub_key());
    }

    #[test]
    fn relay_rejects_when_no_uphill_peers_on_retry() {
        // After rejection from uphill peer, if no more uphill peers available → rejected
        let self_loc = make_peer(4000);
        let joiner = make_peer(5000);
        let peer_a = make_peer(6000);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // Step 1: Initial forward to peer_a
        let ctx_a = TestRelayContext::new(self_loc.clone())
            .accept(false)
            .uphill_hop(Some(peer_a));
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let recency = HashMap::new();
        let actions = state.handle_request(
            &ctx_a,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );
        assert!(actions.forward.is_some());

        // Step 2: Reset forwarded_to (simulating rejection receipt)
        state.forwarded_to = None;
        state.forwarded_at = None;

        // Step 3: No uphill peers available on retry → should reject
        let ctx_none = TestRelayContext::new(self_loc).accept(false);
        let retry_actions = state.handle_request(
            &ctx_none,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );
        assert!(
            retry_actions.rejected,
            "should reject when no uphill peers on retry"
        );
        assert!(retry_actions.forward.is_none());
        assert!(retry_actions.accept.is_none());
    }

    /// Regression test for #3838: when the `Rejected` handler in
    /// `process_message` invokes `handle_request` again after an uphill
    /// rejection and the retry decides to accept locally, the returned
    /// `AcceptOutcome` MUST carry the joiner alongside the response so that
    /// the caller cannot forget to promote the transient connection into a
    /// ring connection. Prior to this fix the two fields (`accept_response`,
    /// `expect_connection_from`) were separate, and the retry path silently
    /// dropped `expect_connection_from` — leaving the joiner stuck as a
    /// transient on the acceptor's side. `get_peer_by_addr` (used by
    /// subscribe interest registration and broadcast fan-out) only finds
    /// ring connections, so downstream UPDATE broadcasts never reached the
    /// joiner. The fields are now bundled in `AcceptOutcome`, making this
    /// bug structurally unrepresentable.
    #[test]
    fn relay_retry_acceptance_bundles_joiner_with_response() {
        let self_loc = make_peer(4000);
        let joiner = make_peer(5000);
        let peer_a = make_peer(6000);
        let mut state = RelayState {
            upstream_addr: joiner.socket_addr().expect("test peer must have address"),
            request: ConnectRequest {
                desired_location: Location::random(),
                joiner: joiner.clone(),
                ttl: 3,
                visited: VisitedPeers::default(),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: None,
            forwarded_at: None,
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: false,
        };

        // Step 1: Initial forward to peer_a (matches what process_message does
        // on receipt of the original ConnectMsg::Request).
        let ctx_a = TestRelayContext::new(self_loc.clone())
            .accept(false)
            .uphill_hop(Some(peer_a));
        let mut forward_attempts = HashMap::new();
        let estimator = ConnectForwardEstimator::new();
        let recency = HashMap::new();
        let first = state.handle_request(
            &ctx_a,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );
        assert!(
            first.forward.is_some(),
            "initial pass should forward uphill"
        );
        assert!(
            first.accept.is_none(),
            "initial pass should not accept (forwarding upstream)"
        );

        // Step 2: Simulate peer_a's Rejected message arriving. This mirrors
        // the mutation `process_message::Rejected` applies before re-calling
        // handle_request (see connect.rs Rejected handler).
        state.forwarded_to = None;
        state.forwarded_at = None;

        // Step 3: Retry with no more uphill peers available AND the relay
        // willing to accept — this is the exact path that produced the
        // #3838 regression (accept_response set, expect_connection_from
        // silently dropped by the caller).
        let ctx_retry = TestRelayContext::new(self_loc).accept(true);
        let retry = state.handle_request(
            &ctx_retry,
            &recency,
            &mut forward_attempts,
            &estimator,
            Instant::now(),
        );

        let accept = retry
            .accept
            .expect("retry must accept locally when no uphill peers remain");
        assert_eq!(
            accept.joiner.pub_key(),
            joiner.pub_key(),
            "AcceptOutcome must carry the joiner alongside the response so the caller \
             cannot forget ring promotion (#3838)"
        );
        assert!(
            accept.joiner.socket_addr().is_some(),
            "joiner address must be known so the acceptor can emit ConnectPeer / \
             ExpectPeerConnection against the observed address"
        );
    }

    // ============ filter_low_score_peers tests ============

    #[test]
    fn test_filter_low_score_peers_removes_bad_when_alternatives_exist() {
        let mut scored: Vec<(f64, &str)> = vec![(0.05, "bad"), (0.8, "good")];
        super::filter_low_score_peers(&mut scored, 0);
        assert_eq!(scored.len(), 1);
        assert_eq!(scored[0].1, "good");
    }

    #[test]
    fn test_filter_low_score_peers_keeps_sole_option() {
        let mut scored: Vec<(f64, &str)> = vec![(0.05, "only")];
        super::filter_low_score_peers(&mut scored, 0);
        assert_eq!(
            scored.len(),
            1,
            "Single peer must be kept even with low score"
        );
    }

    #[test]
    fn test_filter_low_score_peers_removes_when_unscored_fallback_exists() {
        let mut scored: Vec<(f64, &str)> = vec![(0.05, "low")];
        super::filter_low_score_peers(&mut scored, 1);
        assert!(
            scored.is_empty(),
            "Low-score peer should be filtered when unscored fallback exists"
        );
    }

    #[test]
    fn test_filter_low_score_peers_keeps_all_bad_when_no_alternatives() {
        // All scored peers below threshold AND no eligible fallback → keep them all.
        let mut scored: Vec<(f64, &str)> = vec![(0.05, "bad1"), (0.10, "bad2")];
        super::filter_low_score_peers(&mut scored, 0);
        assert_eq!(
            scored.len(),
            2,
            "Must keep all low-score peers when no alternatives exist"
        );
    }

    #[test]
    fn test_filter_low_score_preserves_all_when_no_alternatives() {
        // When all scored candidates are below MIN_FORWARD_SCORE and there are
        // no fallback candidates, filter_low_score_peers must keep them all
        // to prevent a routing dead-end.
        let mut scored = vec![
            (0.08, make_peer(7001)), // below MIN_FORWARD_SCORE
            (0.10, make_peer(7002)), // below MIN_FORWARD_SCORE
            (0.12, make_peer(7003)), // below MIN_FORWARD_SCORE
        ];
        let fallback_count = 0; // no fallbacks available
        filter_low_score_peers(&mut scored, fallback_count);
        assert_eq!(
            scored.len(),
            3,
            "all peers should be preserved when no alternatives exist"
        );
    }

    #[test]
    fn test_filter_low_score_removes_bad_when_alternatives_exist() {
        let good_peer = make_peer(7002);
        let good_key = good_peer.pub_key().clone();
        let mut scored = vec![
            (0.08, make_peer(7001)), // below MIN_FORWARD_SCORE
            (0.30, good_peer),       // above MIN_FORWARD_SCORE
        ];
        let fallback_count = 1; // fallbacks available
        filter_low_score_peers(&mut scored, fallback_count);
        assert_eq!(
            scored.len(),
            1,
            "low-score peer should be filtered when alternatives exist"
        );
        assert_eq!(scored[0].1.pub_key(), &good_key);
    }

    // ==================== ConnectFailed + Jitter + Exclusion Tests ====================

    #[test]
    fn test_connect_failed_serialization_roundtrip() {
        use bincode::{deserialize, serialize};
        let tx = Transaction::new::<ConnectMsg>();
        let addr: SocketAddr = "10.0.0.5:9000".parse().unwrap();
        let msg = ConnectMsg::ConnectFailed {
            id: tx,
            failed_acceptor_addr: addr,
        };

        let encoded = serialize(&msg).expect("serialize ConnectFailed");
        let decoded: ConnectMsg = deserialize(&encoded).expect("deserialize ConnectFailed");

        match decoded {
            ConnectMsg::ConnectFailed {
                id,
                failed_acceptor_addr,
            } => {
                assert_eq!(id, tx);
                assert_eq!(failed_acceptor_addr, addr);
            }
            ConnectMsg::Request { .. }
            | ConnectMsg::Response { .. }
            | ConnectMsg::ObservedAddress { .. }
            | ConnectMsg::Rejected { .. } => {
                panic!("expected ConnectFailed, got: {decoded}")
            }
        }
    }

    #[test]
    fn test_connect_failed_inner_message_traits() {
        let tx = Transaction::new::<ConnectMsg>();
        let addr: SocketAddr = "10.0.0.5:9000".parse().unwrap();
        let msg = ConnectMsg::ConnectFailed {
            id: tx,
            failed_acceptor_addr: addr,
        };

        assert_eq!(
            *msg.id(),
            tx,
            "ConnectFailed should return its transaction id"
        );
        assert_eq!(
            msg.requested_location(),
            None,
            "ConnectFailed should return None for requested_location"
        );
        let display = format!("{msg}");
        assert!(
            display.contains("ConnectFailed"),
            "Display should mention ConnectFailed: {display}"
        );
    }

    #[test]
    fn test_jitter_magnitude_at_various_failure_counts() {
        // Verify jitter magnitude formula: 0.05 * 2^(failures-2), capped at 0.25
        let cases = vec![
            (1u32, 0.0), // failures=1 → no jitter
            (2, 0.05),   // 0.05 * 2^0 = 0.05
            (3, 0.10),   // 0.05 * 2^1 = 0.10
            (4, 0.20),   // 0.05 * 2^2 = 0.20
            (5, 0.25),   // 0.05 * 2^3 = 0.40 → capped at 0.25
            (10, 0.25),  // Still capped at 0.25
        ];

        for (failures, expected_magnitude) in cases {
            let magnitude = if failures > 1 {
                (0.05 * (1u32 << (failures - 2).min(3)) as f64).min(0.25)
            } else {
                0.0
            };
            assert!(
                (magnitude - expected_magnitude).abs() < 1e-10,
                "failures={failures}: expected magnitude {expected_magnitude}, got {magnitude}"
            );
        }
    }

    #[test]
    fn test_connect_jitter_counter() {
        let cm = crate::ring::ConnectionManager::test_default();
        let now = Instant::now();

        assert_eq!(cm.connect_jitter_failure_count(), 0);

        let count = cm.increment_connect_jitter_failures(now);
        assert_eq!(count, 1);
        assert_eq!(cm.connect_jitter_failure_count(), 1);

        let count = cm.increment_connect_jitter_failures(now);
        assert_eq!(count, 2);

        cm.reset_connect_jitter_failures();
        assert_eq!(cm.connect_jitter_failure_count(), 0);
    }

    #[test]
    fn test_connect_jitter_decay() {
        let cm = crate::ring::ConnectionManager::test_default();
        let now = Instant::now();

        // Accumulate 6 failures
        for _ in 0..6 {
            cm.increment_connect_jitter_failures(now);
        }
        assert_eq!(cm.connect_jitter_failure_count(), 6);

        // Simulate 15 minutes of idle time (3 decay intervals of 5 minutes each)
        let later = now + Duration::from_secs(15 * 60);
        let count = cm.increment_connect_jitter_failures(later);
        // 6 - 3 decay steps = 3, then +1 = 4
        assert_eq!(count, 4, "jitter should decay proportionally to idle time");

        // Simulate enough idle time to decay to zero
        let much_later = later + Duration::from_secs(60 * 60);
        let count = cm.increment_connect_jitter_failures(much_later);
        // 4 - 12 decay steps = 0 (saturating), then +1 = 1
        assert_eq!(
            count, 1,
            "jitter should decay to zero after long idle period"
        );
    }

    #[test]
    fn test_acceptor_reliability_tracking() {
        let cm = crate::ring::ConnectionManager::test_default();
        let addr: SocketAddr = "10.0.0.5:9000".parse().unwrap();
        let now = Instant::now();

        // Unknown peer starts at 0.5 prior
        let score = cm.peer_acceptor_reliability(addr, now);
        assert!(
            (score - 0.5).abs() < f64::EPSILON,
            "unknown peer should have 0.5 reliability"
        );

        // Record failures — reliability should decrease
        cm.record_acceptor_outcome(addr, false, now);
        cm.record_acceptor_outcome(addr, false, now);
        cm.record_acceptor_outcome(addr, false, now);
        let score = cm.peer_acceptor_reliability(addr, now);
        // (0+1)/(3+2) = 0.2
        assert!(
            (score - 0.2).abs() < 0.01,
            "3 failures should give ~0.2 reliability, got {}",
            score
        );

        // Record a success — reliability should increase
        cm.record_acceptor_outcome(addr, true, now);
        let score = cm.peer_acceptor_reliability(addr, now);
        // (1+1)/(4+2) ≈ 0.333
        assert!(
            score > 0.2,
            "success should increase reliability from 0.2, got {}",
            score
        );
    }

    #[test]
    fn test_skip_list_skips_self_and_visited() {
        let tx = Transaction::new::<ConnectMsg>();
        let mut visited = VisitedPeers::new(&tx);
        let self_addr: SocketAddr = "10.0.0.1:5000".parse().unwrap();
        let visited_addr: SocketAddr = "10.0.0.2:5000".parse().unwrap();
        let normal_addr: SocketAddr = "10.0.0.3:5000".parse().unwrap();

        visited.mark_visited(visited_addr);

        let skip = SkipListWithSelf {
            visited: &visited,
            self_addr: Some(self_addr),
        };

        assert!(skip.has_element(self_addr), "self should be skipped");
        assert!(
            skip.has_element(visited_addr),
            "visited peer should be skipped"
        );
        assert!(
            !skip.has_element(normal_addr),
            "normal peer should not be skipped"
        );
    }

    #[test]
    fn test_bloom_filter_excludes_failed_acceptor() {
        // After receiving ConnectFailed, the relay marks the failed acceptor
        // in the bloom filter so routing avoids it.
        let tx = Transaction::new::<ConnectMsg>();
        let mut visited = VisitedPeers::new(&tx);
        let failed_addr: SocketAddr = "10.0.0.99:9000".parse().unwrap();

        assert!(
            !visited.probably_visited(failed_addr),
            "addr should not be visited initially"
        );

        visited.mark_visited(failed_addr);

        assert!(
            visited.probably_visited(failed_addr),
            "failed addr should be marked as visited after ConnectFailed processing"
        );
    }

    #[test]
    fn test_response_forwarded_prevents_false_timeout() {
        // Verify that response_forwarded=true prevents the GC from logging
        // a false "uphill route timed out" warning.
        let relay_peer = make_peer(5001);
        let state = RelayState {
            upstream_addr: "10.0.0.1:5000".parse().unwrap(),
            request: ConnectRequest {
                desired_location: Location::new(0.3),
                joiner: make_peer(5002),
                ttl: 5,
                visited: VisitedPeers::new(&Transaction::new::<ConnectMsg>()),
                uphill_budget: DEFAULT_UPHILL_BUDGET,
            },
            forwarded_to: Some(relay_peer),
            forwarded_at: Some(Instant::now()),
            observed_sent: false,
            accepted_locally: false,
            response_forwarded: true,
        };

        // With response_forwarded=true, even though forwarded_to is Some,
        // this should NOT be considered a timeout.
        assert!(state.response_forwarded, "response_forwarded should be set");
        assert!(
            state.forwarded_to.is_some(),
            "forwarded_to should still be preserved for ConnectFailed propagation"
        );
    }

    /// Regression test for #3303: already-connected peers must be excluded from the visited
    /// filter so routing doesn't forward connect requests back to them as acceptors.
    ///
    /// Before the fix, only recently-failed NAT addresses were excluded. This caused nodes to
    /// get permanently stuck: every ConnectResponse resolved to the same already-connected peer,
    /// `connect_peer` logged "reusing existing transport", and ring_connections never grew.
    #[test]
    fn test_initiate_join_request_excludes_connected_peers() {
        let own = make_peer(8001);
        let target = make_peer(8002);
        let desired_location = Location::try_from(0.5).unwrap();
        let estimator = Arc::new(RwLock::new(ConnectForwardEstimator::new()));

        let connected_peer1: SocketAddr = "10.0.0.10:9000".parse().unwrap();
        let connected_peer2: SocketAddr = "10.0.0.11:9001".parse().unwrap();
        let unconnected_peer: SocketAddr = "10.0.0.99:9999".parse().unwrap();

        let exclude_addrs = vec![connected_peer1, connected_peer2];

        let (tx, _op, msg) = ConnectOp::initiate_join_request(
            own.clone(),
            target.clone(),
            desired_location,
            10,
            3,
            estimator,
            &exclude_addrs,
        );

        let request = match msg {
            ConnectMsg::Request { payload, .. } => payload,
            ConnectMsg::Response { .. }
            | ConnectMsg::ObservedAddress { .. }
            | ConnectMsg::Rejected { .. }
            | ConnectMsg::ConnectFailed { .. } => panic!("Expected ConnectMsg::Request"),
        };

        let visited = request.visited.with_transaction(&tx);

        // Already-connected peers must be in the visited filter so routing skips them.
        assert!(
            visited.probably_visited(connected_peer1),
            "connected peer 1 should be marked visited to prevent routing back to it"
        );
        assert!(
            visited.probably_visited(connected_peer2),
            "connected peer 2 should be marked visited to prevent routing back to it"
        );
        // An unrelated peer should not be in the filter.
        assert!(
            !visited.probably_visited(unconnected_peer),
            "unconnected peer should not be in the visited filter"
        );
    }

    /// Verify that gap_target produces shorter connections than own-location+jitter.
    ///
    /// This is a regression test for the bootstrap topology bug: before the fix,
    /// join_ring_request always targeted the joiner's own location with up to ±0.25
    /// jitter, producing median connection distance ~0.12 instead of the Kleinberg-
    /// optimal ~0.02. Now, with ≥3 existing connections, gap_target is used.
    #[test]
    fn test_gap_target_produces_shorter_connections_than_jitter() {
        use crate::topology::small_world_rand::gap_target;

        let _guard = GlobalRng::seed_guard(42);

        let my_location = Location::new(0.5);

        // Simulate a peer with 5 existing connections at various distances
        let existing_connections = [
            Location::new(0.501), // very close CW
            Location::new(0.51),  // close CW
            Location::new(0.55),  // nearby CW
            Location::new(0.7),   // medium CW
            Location::new(0.9),   // far CW
        ];

        let distances: Vec<f64> = existing_connections
            .iter()
            .map(|loc| my_location.distance(*loc).as_f64())
            .collect();

        // Generate 100 gap_target samples and 100 jitter samples
        let mut gap_target_distances = Vec::new();
        let mut jitter_distances = Vec::new();

        for _ in 0..100 {
            let target = gap_target(my_location, distances.iter().copied());
            gap_target_distances.push(my_location.distance(target).as_f64());

            // Simulate jitter with failures=3 (magnitude=0.10)
            let magnitude = 0.10;
            let uniform_01 = (GlobalRng::random_u64() as f64) / (u64::MAX as f64);
            let offset = magnitude * (2.0 * uniform_01 - 1.0);
            let jittered = (my_location.as_f64() + offset).rem_euclid(1.0);
            jitter_distances.push(my_location.distance(Location::new(jittered)).as_f64());
        }

        gap_target_distances.sort_by(|a, b| a.partial_cmp(b).unwrap());
        jitter_distances.sort_by(|a, b| a.partial_cmp(b).unwrap());

        let gap_median = gap_target_distances[50];
        let jitter_median = jitter_distances[50];

        // gap_target should produce significantly shorter median distances
        // because it targets gaps in the 1/d distribution (Kleinberg-optimal),
        // while jitter produces roughly uniform distances around the peer.
        assert!(
            gap_median < jitter_median,
            "gap_target median ({gap_median:.4}) should be shorter than jitter median ({jitter_median:.4})"
        );

        // gap_target median should be well under 0.05 (typically ~0.001-0.01)
        assert!(
            gap_median < 0.05,
            "gap_target median ({gap_median:.4}) should be < 0.05 (Kleinberg-optimal produces short connections)"
        );
    }
}