irontide-engine 1.0.2

//! M137: Unified peer lifecycle state machine.
//!
//! Tracks every discovered peer address through a deterministic lifecycle:
//! `Queued → Connecting → Live → Dead`, with exponential backoff retry from
//! Dead back to Queued. Provides atomic pipeline counters for diagnostics.

use std::collections::{BTreeMap, VecDeque};
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::{Duration, Instant};

use dashmap::DashMap;
use parking_lot::Mutex;
use smallvec::SmallVec;
use tokio::sync::mpsc;
use tracing::debug;

use crate::peer_state::PeerSource;

/// Maximum backoff duration for a dead peer (1 hour).
const MAX_BACKOFF_SECS: u64 = 3600;

/// M148 + v0.187.3: legacy default ban duration (30 min). The runtime value
/// is configurable via `Settings::eviction_ban_duration_secs` (default
/// shipped at 600 / 10 min in v0.187.3 — see settings.rs). This constant is
/// retained as the fallback for `PeerStates::new` callers that do not
/// supply a config (production code goes through `new_with_config`).
#[allow(dead_code)]
const EVICTION_BAN_DURATION: Duration = Duration::from_mins(30);
/// v0.187.3 / OV4: legacy fallback for the FIFO cap on the banned-peer set
/// when `PeerStates::new` is used without a config. The runtime value is
/// `Settings::eviction_ban_set_cap`.
#[allow(dead_code)]
const EVICTION_BAN_SET_CAP_FALLBACK: usize = 1024;

/// Base backoff duration in seconds.
const BASE_BACKOFF_SECS: u64 = 10;

/// Backoff multiplier per attempt.
const BACKOFF_FACTOR: u64 = 6;

/// Total retry window — peers that have been failing for longer than this
/// are removed entirely.
const RETRY_WINDOW_SECS: u64 = 86400;

/// Lifecycle state for a discovered peer address.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum PeerLifecycle {
    /// Peer is queued for connection (waiting for a semaphore permit).
    Queued,
    /// Peer is currently connecting (semaphore permit acquired, handshake in progress).
    Connecting,
    /// Peer has completed handshake and is actively exchanging data.
    Live,
    /// Peer has disconnected or timed out, awaiting backoff before retry.
    Dead,
}

/// Per-peer tracking entry in the lifecycle [`DashMap`].
#[derive(Debug)]
pub(crate) struct PeerEntry {
    /// Current lifecycle state.
    pub state: PeerLifecycle,
    /// How this peer was discovered.
    pub source: PeerSource,
    /// Number of consecutive failure attempts (for backoff calculation).
    pub backoff_attempt: u32,
    /// When the first failure occurred in the current failure sequence.
    /// Used to enforce the 24-hour total retry window.
    pub first_failure_at: Option<Instant>,
    /// Earliest time this dead peer may be retried.
    pub retry_at: Option<Instant>,
    /// M147: When the connection attempt started (set by `mark_connecting`).
    pub connecting_since: Option<Instant>,
    /// M147: When TCP SYN-ACK was received (set directly by spawned peer task).
    /// Peers with this set get the full `peer_connect_timeout` for BT handshake.
    /// Peers without it are eligible for soft reap after `connect_soft_timeout`.
    pub tcp_connected_at: Option<Instant>,
}

/// Atomic counters for peer pipeline visibility.
///
/// All counters use [`Ordering::Relaxed`] — they are informational and do not
/// need happens-before guarantees.
#[derive(Debug, Default)]
pub(crate) struct PeerPipelineStats {
    /// Total number of known peer addresses (all states).
    pub known: AtomicU32,
    /// Number of peers in the `Queued` state.
    pub queued: AtomicU32,
    /// Number of peers in the `Connecting` state.
    pub connecting: AtomicU32,
    /// Number of peers in the `Live` state.
    pub live: AtomicU32,
    /// Number of peers in the `Dead` state.
    pub dead: AtomicU32,
}

impl PeerPipelineStats {
    /// Capture a point-in-time snapshot of all counters.
    pub fn snapshot(&self) -> PeerPipelineSnapshot {
        PeerPipelineSnapshot {
            known: self.known.load(Ordering::Relaxed),
            queued: self.queued.load(Ordering::Relaxed),
            connecting: self.connecting.load(Ordering::Relaxed),
            live: self.live.load(Ordering::Relaxed),
            dead: self.dead.load(Ordering::Relaxed),
        }
    }

    /// Increment the counter for the given lifecycle state.
    fn inc(&self, state: PeerLifecycle) {
        self.counter(state).fetch_add(1, Ordering::Relaxed);
    }

    /// Decrement the counter for the given lifecycle state.
    fn dec(&self, state: PeerLifecycle) {
        self.counter(state).fetch_sub(1, Ordering::Relaxed);
    }

    /// Return a reference to the atomic counter for the given state.
    fn counter(&self, state: PeerLifecycle) -> &AtomicU32 {
        match state {
            PeerLifecycle::Queued => &self.queued,
            PeerLifecycle::Connecting => &self.connecting,
            PeerLifecycle::Live => &self.live,
            PeerLifecycle::Dead => &self.dead,
        }
    }
}

/// Point-in-time snapshot of [`PeerPipelineStats`] — relocated to
/// `irontide-session-types` at M244a; re-exported here so
/// `crate::peer_states::PeerPipelineSnapshot` resolves unchanged.
pub use irontide_session_types::PeerPipelineSnapshot;

/// Unified peer lifecycle tracker.
///
/// Wraps a [`DashMap`] keyed on [`SocketAddr`] with atomic state counters
/// and a channel for re-queuing retried peers.
pub(crate) struct PeerStates {
    /// Per-address lifecycle entries.
    states: DashMap<SocketAddr, PeerEntry>,
    /// Atomic pipeline counters.
    pub stats: PeerPipelineStats,
    /// Channel to re-queue peers for retry after backoff expires.
    queue_tx: mpsc::UnboundedSender<SocketAddr>,
    /// Temporary bans for peers evicted with zero throughput (Pass 0).
    /// Cleanup is lazy: entries expire on next `is_eviction_banned()` lookup.
    /// Stale entries for non-retried peers persist (~56 bytes each) —
    /// bounded by [`Self::eviction_ban_cap`] via the FIFO `eviction_ban_order`.
    eviction_bans: DashMap<SocketAddr, Instant>,
    /// v0.187.3 / OV4: FIFO insertion order for eviction bans, used to enforce
    /// `eviction_ban_cap`. When `bans.len() >= cap`, the front entry is
    /// dropped on the next `add_eviction_ban` call. Coordinates with
    /// `eviction_bans` (lazy expiration) — a stale entry in this queue is
    /// harmless; the corresponding `bans` entry has already been reaped.
    eviction_ban_order: Mutex<VecDeque<SocketAddr>>,
    /// v0.187.3: configured FIFO cap on the banned set.
    eviction_ban_cap: usize,
    /// v0.187.3: configured ban duration.
    eviction_ban_duration: Duration,
    /// v0.173.4 prong 2: side index for O(expired + log n) soft-reap walk.
    ///
    /// Invariant (load-bearing): `addr ∈ connecting_since_index` ⇔
    /// `states[addr].state == Connecting AND states[addr].tcp_connected_at.is_none()`.
    ///
    /// Maintained by every transition method; soft-reap walks
    /// `range(..=cutoff)` instead of iterating the full `DashMap`.
    /// `SmallVec<[_; 2]>` absorbs same-`Instant` collisions from
    /// burst peer admission without spilling to the heap.
    connecting_since_index: Mutex<BTreeMap<Instant, SmallVec<[SocketAddr; 2]>>>,
}

impl PeerStates {
    /// Create a new `PeerStates` with the given re-queue channel and legacy
    /// defaults for the eviction-ban cap + duration (1024 entries / 30 min).
    /// Used by tests and the legacy construction path; production code
    /// (e.g. `TorrentActor::spawn`) uses [`Self::new_with_config`] so the
    /// runtime values track `Settings::eviction_ban_*`.
    #[allow(dead_code)]
    pub fn new(queue_tx: mpsc::UnboundedSender<SocketAddr>) -> Self {
        Self::new_with_config(
            queue_tx,
            EVICTION_BAN_SET_CAP_FALLBACK,
            EVICTION_BAN_DURATION,
        )
    }

    /// v0.187.3: full constructor — supply the runtime cap + duration from
    /// `Settings::eviction_ban_set_cap` and `Settings::eviction_ban_duration_secs`.
    pub fn new_with_config(
        queue_tx: mpsc::UnboundedSender<SocketAddr>,
        eviction_ban_cap: usize,
        eviction_ban_duration: Duration,
    ) -> Self {
        Self {
            states: DashMap::new(),
            stats: PeerPipelineStats::default(),
            queue_tx,
            eviction_bans: DashMap::new(),
            eviction_ban_order: Mutex::new(VecDeque::new()),
            // Floor of 1 so a misconfigured cap doesn't degenerate into "skip
            // every ban" (which would let the churn loop run unbounded again).
            eviction_ban_cap: eviction_ban_cap.max(1),
            eviction_ban_duration,
            connecting_since_index: Mutex::new(BTreeMap::new()),
        }
    }

    /// v0.173.4 prong 2: insert an entry into the `connecting_since` side index.
    ///
    /// Lock-ordering invariant: callers MUST drop any `DashMap` RefMut/Ref guard
    /// before invoking this method. Holding both the `DashMap` shard lock and
    /// the index mutex simultaneously is forbidden — risks deadlock.
    fn idx_insert(&self, ts: Instant, addr: SocketAddr) {
        self.connecting_since_index
            .lock()
            .entry(ts)
            .or_default()
            .push(addr);
    }

    /// v0.173.4 prong 2: remove an entry from the `connecting_since` side
    /// index. Empty buckets are pruned to bound memory.
    ///
    /// Lock-ordering invariant: callers MUST drop any `DashMap` RefMut/Ref guard
    /// before invoking this method. Holding both the `DashMap` shard lock and
    /// the index mutex simultaneously is forbidden — risks deadlock.
    fn idx_remove(&self, ts: Instant, addr: SocketAddr) {
        let mut idx = self.connecting_since_index.lock();
        if let Some(bucket) = idx.get_mut(&ts) {
            bucket.retain(|a| *a != addr);
            if bucket.is_empty() {
                idx.remove(&ts);
            }
        }
    }

    /// Add a newly discovered peer. Returns `true` if this is a new address.
    ///
    /// If the address is already tracked (in any lifecycle state), this is a
    /// no-op and returns `false`.
    pub fn add_if_not_seen(&self, addr: SocketAddr, source: PeerSource) -> bool {
        use dashmap::mapref::entry::Entry;

        match self.states.entry(addr) {
            Entry::Occupied(_) => false,
            Entry::Vacant(vacant) => {
                vacant.insert(PeerEntry {
                    state: PeerLifecycle::Queued,
                    source,
                    backoff_attempt: 0,
                    first_failure_at: None,
                    retry_at: None,
                    connecting_since: None,
                    tcp_connected_at: None,
                });
                self.stats.known.fetch_add(1, Ordering::Relaxed);
                self.stats.inc(PeerLifecycle::Queued);
                // Send to the adder task's queue for connection processing.
                let _ = self.queue_tx.send(addr);
                true
            }
        }
    }

    /// Transition `Queued → Connecting` (called when semaphore permit acquired).
    ///
    /// Returns `true` if the transition succeeded, `false` if the peer was not
    /// in the `Queued` state (or not tracked at all).
    pub fn mark_connecting(&self, addr: SocketAddr) -> bool {
        let idx_insert_ts = {
            if let Some(mut entry) = self.states.get_mut(&addr)
                && entry.state == PeerLifecycle::Queued
            {
                self.stats.dec(PeerLifecycle::Queued);
                entry.state = PeerLifecycle::Connecting;
                let now = Instant::now();
                entry.connecting_since = Some(now);
                entry.tcp_connected_at = None;
                self.stats.inc(PeerLifecycle::Connecting);
                Some(now)
            } else {
                None
            }
        };
        if let Some(ts) = idx_insert_ts {
            self.idx_insert(ts, addr);
            return true;
        }
        false
    }

    /// M147: Mark that TCP SYN-ACK was received for a connecting peer.
    ///
    /// Called directly from the spawned peer task (not via actor mailbox) after
    /// `factory.connect_tcp()` succeeds. This prevents the soft reap task from
    /// disconnecting this peer, giving it the full `peer_connect_timeout` for
    /// BT handshake.
    ///
    /// v0.173.4 prong 2: also removes the peer from
    /// `connecting_since_index`. Without this, peers that complete TCP
    /// SYN-ACK but stall in the `BitTorrent` handshake stay in the index
    /// forever — bounded growth under stall pathologies.
    pub fn set_tcp_connected(&self, addr: SocketAddr) {
        let idx_remove_ts = {
            if let Some(mut entry) = self.states.get_mut(&addr)
                && entry.state == PeerLifecycle::Connecting
            {
                let ts = entry.connecting_since;
                entry.tcp_connected_at = Some(Instant::now());
                ts
            } else {
                None
            }
        };
        if let Some(ts) = idx_remove_ts {
            self.idx_remove(ts, addr);
        }
    }

    /// Transition `Connecting → Live` (called on successful handshake).
    pub fn mark_live(&self, addr: SocketAddr) {
        let idx_remove_ts = {
            if let Some(mut entry) = self.states.get_mut(&addr)
                && entry.state == PeerLifecycle::Connecting
            {
                self.stats.dec(PeerLifecycle::Connecting);
                entry.state = PeerLifecycle::Live;
                // Reset backoff on successful connection.
                entry.backoff_attempt = 0;
                entry.first_failure_at = None;
                entry.retry_at = None;
                self.stats.inc(PeerLifecycle::Live);
                entry.connecting_since
            } else {
                None
            }
        };
        if let Some(ts) = idx_remove_ts {
            self.idx_remove(ts, addr);
        }
    }

    /// Transition `Live|Connecting → Dead` (called on disconnect/timeout).
    ///
    /// Returns the backoff duration the caller should sleep before retrying,
    /// or `None` if the 24-hour retry limit is exhausted (peer is removed
    /// entirely).
    pub fn mark_dead(&self, addr: SocketAddr) -> Option<Duration> {
        let mut entry = self.states.get_mut(&addr)?;
        let old_state = entry.state;
        if old_state != PeerLifecycle::Live && old_state != PeerLifecycle::Connecting {
            return None;
        }

        let now = Instant::now();
        let first_failure = entry.first_failure_at.unwrap_or(now);

        // v0.173.4 prong 2: capture connecting_since for index removal. Only
        // valid when transitioning FROM Connecting — Live entries were never
        // in the index.
        let idx_remove_ts = if old_state == PeerLifecycle::Connecting {
            entry.connecting_since
        } else {
            None
        };

        // Check 24-hour total retry window.
        if first_failure.elapsed() > Duration::from_secs(RETRY_WINDOW_SECS) {
            // Retry limit exhausted — remove this peer entirely.
            self.stats.dec(old_state);
            self.stats.known.fetch_sub(1, Ordering::Relaxed);
            let addr_to_remove = *entry.key();
            drop(entry);
            // Lock-ordering: RefMut dropped above, safe to acquire index mutex.
            if let Some(ts) = idx_remove_ts {
                self.idx_remove(ts, addr_to_remove);
            }
            self.states.remove(&addr_to_remove);
            debug!(%addr_to_remove, "peer removed: 24hr retry window exhausted");
            return None;
        }

        let attempt = entry.backoff_attempt;
        let backoff_secs = BASE_BACKOFF_SECS
            .saturating_mul(BACKOFF_FACTOR.saturating_pow(attempt))
            .min(MAX_BACKOFF_SECS);

        self.stats.dec(old_state);
        entry.state = PeerLifecycle::Dead;
        entry.backoff_attempt = attempt.saturating_add(1);
        entry.first_failure_at = Some(first_failure);
        entry.retry_at = Some(now + Duration::from_secs(backoff_secs));
        self.stats.inc(PeerLifecycle::Dead);

        drop(entry);
        // Lock-ordering: RefMut dropped above, safe to acquire index mutex.
        if let Some(ts) = idx_remove_ts {
            self.idx_remove(ts, addr);
        }

        Some(Duration::from_secs(backoff_secs))
    }

    /// Transition `Dead → Queued` (called after backoff sleep completes).
    ///
    /// Re-queues the peer by sending its address to `queue_tx`.
    /// Returns `true` if the transition succeeded.
    pub fn mark_queued_for_retry(&self, addr: SocketAddr) -> bool {
        if let Some(mut entry) = self.states.get_mut(&addr)
            && entry.state == PeerLifecycle::Dead
        {
            self.stats.dec(PeerLifecycle::Dead);
            entry.state = PeerLifecycle::Queued;
            entry.retry_at = None;
            self.stats.inc(PeerLifecycle::Queued);

            // Re-queue the peer for connection.
            let _ = self.queue_tx.send(addr);
            return true;
        }
        false
    }

    /// Returns `true` if the peer is in the `Live` state.
    pub fn is_live(&self, addr: &SocketAddr) -> bool {
        self.states
            .get(addr)
            .is_some_and(|e| e.state == PeerLifecycle::Live)
    }

    /// Returns `true` if the peer is in `Queued`, `Connecting`, or `Live` state
    /// (i.e., actively in the pipeline, not dead or unknown).
    #[allow(dead_code)] // Wired in later tasks (diagnose output).
    pub fn is_active(&self, addr: &SocketAddr) -> bool {
        self.states
            .get(addr)
            .is_some_and(|e| e.state != PeerLifecycle::Dead)
    }

    /// Get the source of a tracked peer.
    pub fn source(&self, addr: &SocketAddr) -> Option<PeerSource> {
        self.states.get(addr).map(|e| e.source)
    }

    /// M147: Returns addresses of connecting peers eligible for soft reap.
    ///
    /// A peer is eligible if:
    /// - State is `Connecting`
    /// - `tcp_connected_at` is `None` (no SYN-ACK received)
    /// - `connecting_since` elapsed > `soft_timeout`
    ///
    /// v0.173.3: thin wrapper over `soft_reap_candidates_into` for
    /// callers that don't hold a persistent buffer (tests, one-shot
    /// paths). The hot path on `TorrentActor` uses the `_into` variant.
    #[allow(dead_code)] // Used by tests; lib hot path uses `_into` variant.
    pub fn soft_reap_candidates(&self, soft_timeout: Duration) -> Vec<SocketAddr> {
        let mut out = Vec::new();
        self.soft_reap_candidates_into(soft_timeout, &mut out);
        out
    }

    /// v0.173.3: Buffer-fill variant of `soft_reap_candidates`.
    ///
    /// Fills `out` with eligible peer addresses, clearing any existing
    /// content first. Eligibility rules are identical to
    /// `soft_reap_candidates` (the Vec-returning version is now a thin
    /// wrapper over this method). Callers that reap on a tick should
    /// hold their `Vec<SocketAddr>` across calls to reclaim its heap
    /// allocation — the 2026-04-24 heaptrack attributed 1782 calls/s
    /// and 6.5 M total allocations to the Vec-returning variant.
    pub fn soft_reap_candidates_into(&self, soft_timeout: Duration, out: &mut Vec<SocketAddr>) {
        out.clear();
        let now = Instant::now();
        // Underflow on monotonic clocks early in process life (rare).
        // No peer can be older than `now`, so nothing is reapable.
        let Some(cutoff) = now.checked_sub(soft_timeout) else {
            return;
        };

        let idx = self.connecting_since_index.lock();
        for (_ts, bucket) in idx.range(..=cutoff) {
            for addr in bucket.iter().copied() {
                // The invariant is mutator-enforced. No concurrency path can
                // break it (every transition method updates index and DashMap
                // atomically, with the lock-ordering rule documented on
                // idx_insert/idx_remove). debug_assert! fails fast in tests
                // and dev builds rather than silently filtering in prod —
                // silent filtering would mask a real invariant bug.
                #[cfg(debug_assertions)]
                if let Some(entry) = self.states.get(&addr) {
                    debug_assert!(
                        entry.state == PeerLifecycle::Connecting
                            && entry.tcp_connected_at.is_none(),
                        "soft_reap index/DashMap invariant violated for {addr:?}: \
                         state={:?} tcp_connected_at={:?}",
                        entry.state,
                        entry.tcp_connected_at,
                    );
                }
                out.push(addr);
            }
        }
    }

    /// v0.173.4 prong 2: test-only helper that backdates `connecting_since`
    /// for a peer AND atomically updates the side index. Replaces the
    /// pre-prong-2 pattern of direct `iter_mut()` mutation, which left
    /// the index out of sync with the `DashMap` entry.
    #[cfg(test)]
    pub(crate) fn test_backdate_connecting_since(&self, addr: SocketAddr, new_ts: Instant) {
        let old_ts = {
            let Some(mut entry) = self.states.get_mut(&addr) else {
                return;
            };
            let old = entry.connecting_since;
            entry.connecting_since = Some(new_ts);
            old
        };
        if let Some(old) = old_ts {
            self.idx_remove(old, addr);
        }
        self.idx_insert(new_ts, addr);
    }

    /// Returns the number of tracked peer addresses (all states).
    #[allow(dead_code)] // Wired in later tasks (diagnose output).
    pub fn len(&self) -> usize {
        self.states.len()
    }

    /// M148: Record an eviction ban for a peer evicted via Pass 0 (zero-throughput).
    ///
    /// The ban prevents the peer from being reconnected for the configured
    /// `eviction_ban_duration`, breaking the cycle where deadweight peers
    /// immediately rejoin the `LivePool`. v0.187.3 / OV4: enforces a FIFO
    /// cap (`eviction_ban_cap`) — when the set is full the oldest entry is
    /// dropped to make room.
    pub fn add_eviction_ban(&self, addr: SocketAddr) {
        let mut order = self.eviction_ban_order.lock();
        while self.eviction_bans.len() >= self.eviction_ban_cap {
            let Some(victim) = order.pop_front() else {
                break;
            };
            self.eviction_bans.remove(&victim);
        }
        // If the address was already banned we leave the old `order` entry in
        // place — duplicate entries are harmless (a single `remove` clears
        // both rounds) and avoid an O(n) scan on a hot path.
        if self.eviction_bans.insert(addr, Instant::now()).is_none() {
            order.push_back(addr);
        }
    }

    /// M148: Check whether a peer is currently eviction-banned.
    ///
    /// Returns `true` if the peer was banned and the ban has not yet expired.
    /// Expired bans are lazily removed on check. v0.187.3: duration is the
    /// runtime `eviction_ban_duration` instead of the hardcoded 30 min.
    pub fn is_eviction_banned(&self, addr: &SocketAddr) -> bool {
        let Some(entry) = self.eviction_bans.get(addr) else {
            return false;
        };
        let banned_at = *entry;
        drop(entry); // Release DashMap ref before potential remove
        if banned_at.elapsed() >= self.eviction_ban_duration {
            // Ban expired — clean up lazily
            self.eviction_bans.remove(addr);
            false
        } else {
            true
        }
    }

    /// v0.187.3 tests: read-only accessor for the current banned-set size.
    #[cfg(test)]
    pub fn eviction_ban_count(&self) -> usize {
        self.eviction_bans.len()
    }
}

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

    fn test_addr(port: u16) -> SocketAddr {
        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1)), port)
    }

    fn test_addr_ip(last_octet: u8, port: u16) -> SocketAddr {
        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, last_octet)), port)
    }

    fn make_peer_states() -> (PeerStates, mpsc::UnboundedReceiver<SocketAddr>) {
        let (tx, rx) = mpsc::unbounded_channel();
        (PeerStates::new(tx), rx)
    }

    #[test]
    fn add_if_not_seen_returns_true_for_new_peer() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        assert!(ps.add_if_not_seen(addr, PeerSource::Tracker));
    }

    #[test]
    fn add_if_not_seen_returns_false_for_duplicate() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        assert!(ps.add_if_not_seen(addr, PeerSource::Tracker));
        assert!(!ps.add_if_not_seen(addr, PeerSource::Dht));
    }

    #[test]
    fn add_if_not_seen_sets_queued_state() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        let entry = ps.states.get(&addr).expect("entry should exist");
        assert_eq!(entry.state, PeerLifecycle::Queued);
        assert_eq!(entry.source, PeerSource::Tracker);
        assert_eq!(entry.backoff_attempt, 0);
        assert!(entry.first_failure_at.is_none());
        assert!(entry.retry_at.is_none());
    }

    #[test]
    fn counters_increment_on_add() {
        let (ps, _rx) = make_peer_states();
        let snap_before = ps.stats.snapshot();
        assert_eq!(snap_before.known, 0);
        assert_eq!(snap_before.queued, 0);

        ps.add_if_not_seen(test_addr(6881), PeerSource::Tracker);
        ps.add_if_not_seen(test_addr_ip(2, 6882), PeerSource::Dht);

        let snap = ps.stats.snapshot();
        assert_eq!(snap.known, 2);
        assert_eq!(snap.queued, 2);
        assert_eq!(snap.connecting, 0);
        assert_eq!(snap.live, 0);
        assert_eq!(snap.dead, 0);
    }

    #[test]
    fn mark_connecting_transitions_from_queued() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);

        assert!(ps.mark_connecting(addr));

        let entry = ps.states.get(&addr).expect("entry should exist");
        assert_eq!(entry.state, PeerLifecycle::Connecting);

        let snap = ps.stats.snapshot();
        assert_eq!(snap.queued, 0);
        assert_eq!(snap.connecting, 1);
    }

    #[test]
    fn mark_connecting_rejects_non_queued() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);

        // Already Connecting — second call should fail.
        assert!(!ps.mark_connecting(addr));
    }

    #[test]
    fn mark_connecting_returns_false_for_unknown_addr() {
        let (ps, _rx) = make_peer_states();
        assert!(!ps.mark_connecting(test_addr(9999)));
    }

    #[test]
    fn mark_live_transitions_from_connecting() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        ps.mark_live(addr);

        let entry = ps.states.get(&addr).expect("entry should exist");
        assert_eq!(entry.state, PeerLifecycle::Live);
        assert_eq!(entry.backoff_attempt, 0);
        assert!(entry.first_failure_at.is_none());

        let snap = ps.stats.snapshot();
        assert_eq!(snap.connecting, 0);
        assert_eq!(snap.live, 1);
    }

    #[test]
    fn mark_live_resets_backoff_state() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);

        // Artificially set backoff state before going live.
        if let Some(mut entry) = ps.states.get_mut(&addr) {
            entry.backoff_attempt = 3;
            entry.first_failure_at = Some(Instant::now());
        }

        ps.mark_live(addr);

        let entry = ps.states.get(&addr).expect("entry should exist");
        assert_eq!(entry.backoff_attempt, 0);
        assert!(entry.first_failure_at.is_none());
        assert!(entry.retry_at.is_none());
    }

    #[test]
    fn mark_live_no_op_if_not_connecting() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);

        // Still Queued — mark_live should be a no-op.
        ps.mark_live(addr);
        let entry = ps.states.get(&addr).expect("entry should exist");
        assert_eq!(entry.state, PeerLifecycle::Queued);
    }

    #[test]
    fn mark_dead_from_live_returns_backoff_duration() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        ps.mark_live(addr);

        let backoff = ps.mark_dead(addr);
        assert_eq!(backoff, Some(Duration::from_secs(10)));

        let entry = ps.states.get(&addr).expect("entry should exist");
        assert_eq!(entry.state, PeerLifecycle::Dead);
        assert_eq!(entry.backoff_attempt, 1);
        assert!(entry.first_failure_at.is_some());
        assert!(entry.retry_at.is_some());

        let snap = ps.stats.snapshot();
        assert_eq!(snap.live, 0);
        assert_eq!(snap.dead, 1);
    }

    #[test]
    fn mark_dead_from_connecting_returns_backoff_duration() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);

        let backoff = ps.mark_dead(addr);
        assert_eq!(backoff, Some(Duration::from_secs(10)));

        let entry = ps.states.get(&addr).expect("entry should exist");
        assert_eq!(entry.state, PeerLifecycle::Dead);

        let snap = ps.stats.snapshot();
        assert_eq!(snap.connecting, 0);
        assert_eq!(snap.dead, 1);
    }

    #[test]
    fn mark_dead_returns_none_for_queued() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);

        assert!(ps.mark_dead(addr).is_none());
    }

    #[test]
    fn mark_dead_returns_none_for_unknown() {
        let (ps, _rx) = make_peer_states();
        assert!(ps.mark_dead(test_addr(9999)).is_none());
    }

    #[test]
    fn backoff_increases_exponentially() {
        // Expected backoff sequence: 10, 60, 360, 2160, 3600 (cap)
        let expected_secs = [10u64, 60, 360, 2160, 3600];

        // Test the formula directly (matching peer_adder.rs pattern).
        for (attempt, &expected) in expected_secs.iter().enumerate() {
            #[allow(clippy::cast_possible_truncation)]
            let attempt = attempt as u32;
            let got = BASE_BACKOFF_SECS
                .saturating_mul(BACKOFF_FACTOR.saturating_pow(attempt))
                .min(MAX_BACKOFF_SECS);
            assert_eq!(
                got, expected,
                "attempt {attempt}: expected {expected}s, got {got}s"
            );
        }

        // Verify cap holds for higher attempts.
        for attempt in 5u32..=10 {
            let got = BASE_BACKOFF_SECS
                .saturating_mul(BACKOFF_FACTOR.saturating_pow(attempt))
                .min(MAX_BACKOFF_SECS);
            assert_eq!(
                got, MAX_BACKOFF_SECS,
                "attempt {attempt} should cap at {MAX_BACKOFF_SECS}s"
            );
        }
    }

    #[test]
    fn mark_dead_preserves_first_failure_at_across_retries() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        // First connection cycle: Queued → Connecting → Live → Dead.
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        ps.mark_live(addr);
        let _ = ps.mark_dead(addr);

        let first_failure = ps
            .states
            .get(&addr)
            .expect("entry should exist")
            .first_failure_at
            .expect("first_failure_at should be set");

        // Simulate retry: Dead → Queued → Connecting → Live → Dead.
        ps.mark_queued_for_retry(addr);
        ps.mark_connecting(addr);
        // Note: mark_live resets backoff. To test first_failure_at preservation,
        // we go Connecting → Dead directly (simulating a connection failure).
        let _ = ps.mark_dead(addr);

        let second_failure = ps
            .states
            .get(&addr)
            .expect("entry should exist")
            .first_failure_at
            .expect("first_failure_at should still be set");

        assert_eq!(
            first_failure, second_failure,
            "first_failure_at must not change across retries without a successful connection"
        );
    }

    #[test]
    fn mark_dead_removes_peer_after_24hr_window() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        ps.mark_live(addr);

        // Simulate a first_failure_at that is >24 hours in the past.
        if let Some(mut entry) = ps.states.get_mut(&addr) {
            // Instant::now() - 24hr - 1s. Use checked_sub to be safe.
            let too_old = Instant::now().checked_sub(Duration::from_secs(RETRY_WINDOW_SECS + 1));
            if let Some(old_instant) = too_old {
                entry.first_failure_at = Some(old_instant);
            } else {
                // If checked_sub returns None (system uptime < 24hr), skip this test.
                return;
            }
        }

        let result = ps.mark_dead(addr);
        assert!(result.is_none(), "peer past 24hr window should return None");
        assert!(
            ps.states.get(&addr).is_none(),
            "peer should be removed from the map"
        );

        let snap = ps.stats.snapshot();
        assert_eq!(snap.known, 0, "known counter should be decremented");
        assert_eq!(snap.live, 0);
        assert_eq!(snap.dead, 0);
    }

    #[test]
    fn mark_queued_for_retry_transitions_dead_to_queued() {
        let (ps, mut rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        ps.mark_live(addr);
        let _ = ps.mark_dead(addr);

        assert!(ps.mark_queued_for_retry(addr));

        let entry = ps.states.get(&addr).expect("entry should exist");
        assert_eq!(entry.state, PeerLifecycle::Queued);
        assert!(entry.retry_at.is_none(), "retry_at should be cleared");

        let snap = ps.stats.snapshot();
        assert_eq!(snap.dead, 0);
        assert_eq!(snap.queued, 1);

        // Verify the address was sent to the re-queue channel.
        let queued_addr = rx.try_recv().expect("should have received re-queued addr");
        assert_eq!(queued_addr, addr);
    }

    #[test]
    fn mark_queued_for_retry_rejects_non_dead() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        // Still Queued — should not transition.
        assert!(!ps.mark_queued_for_retry(addr));
    }

    #[test]
    fn mark_queued_for_retry_returns_false_for_unknown() {
        let (ps, _rx) = make_peer_states();
        assert!(!ps.mark_queued_for_retry(test_addr(9999)));
    }

    #[test]
    fn is_live_reflects_state() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        assert!(!ps.is_live(&addr));

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        assert!(!ps.is_live(&addr));

        ps.mark_connecting(addr);
        assert!(!ps.is_live(&addr));

        ps.mark_live(addr);
        assert!(ps.is_live(&addr));
    }

    #[test]
    fn is_active_reflects_state() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        // Unknown peer is not active.
        assert!(!ps.is_active(&addr));

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        assert!(ps.is_active(&addr), "Queued peer should be active");

        ps.mark_connecting(addr);
        assert!(ps.is_active(&addr), "Connecting peer should be active");

        ps.mark_live(addr);
        assert!(ps.is_active(&addr), "Live peer should be active");

        let _ = ps.mark_dead(addr);
        assert!(!ps.is_active(&addr), "Dead peer should not be active");
    }

    #[test]
    fn len_tracks_entry_count() {
        let (ps, _rx) = make_peer_states();
        assert_eq!(ps.len(), 0);

        ps.add_if_not_seen(test_addr(6881), PeerSource::Tracker);
        assert_eq!(ps.len(), 1);

        ps.add_if_not_seen(test_addr_ip(2, 6882), PeerSource::Dht);
        assert_eq!(ps.len(), 2);

        // Duplicate should not increase len.
        ps.add_if_not_seen(test_addr(6881), PeerSource::Pex);
        assert_eq!(ps.len(), 2);
    }

    #[test]
    fn full_lifecycle_counters_are_consistent() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        // Queued
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        let s = ps.stats.snapshot();
        assert_eq!(
            (s.known, s.queued, s.connecting, s.live, s.dead),
            (1, 1, 0, 0, 0)
        );

        // Connecting
        ps.mark_connecting(addr);
        let s = ps.stats.snapshot();
        assert_eq!(
            (s.known, s.queued, s.connecting, s.live, s.dead),
            (1, 0, 1, 0, 0)
        );

        // Live
        ps.mark_live(addr);
        let s = ps.stats.snapshot();
        assert_eq!(
            (s.known, s.queued, s.connecting, s.live, s.dead),
            (1, 0, 0, 1, 0)
        );

        // Dead
        let _ = ps.mark_dead(addr);
        let s = ps.stats.snapshot();
        assert_eq!(
            (s.known, s.queued, s.connecting, s.live, s.dead),
            (1, 0, 0, 0, 1)
        );

        // Re-queued
        ps.mark_queued_for_retry(addr);
        let s = ps.stats.snapshot();
        assert_eq!(
            (s.known, s.queued, s.connecting, s.live, s.dead),
            (1, 1, 0, 0, 0)
        );
    }

    #[test]
    fn multiple_peers_tracked_independently() {
        let (ps, _rx) = make_peer_states();
        let addr1 = test_addr(6881);
        let addr2 = test_addr_ip(2, 6882);
        let addr3 = test_addr_ip(3, 6883);

        ps.add_if_not_seen(addr1, PeerSource::Tracker);
        ps.add_if_not_seen(addr2, PeerSource::Dht);
        ps.add_if_not_seen(addr3, PeerSource::Pex);

        ps.mark_connecting(addr1);
        ps.mark_connecting(addr2);
        ps.mark_live(addr1);

        let s = ps.stats.snapshot();
        assert_eq!(s.known, 3);
        assert_eq!(s.queued, 1); // addr3
        assert_eq!(s.connecting, 1); // addr2
        assert_eq!(s.live, 1); // addr1
        assert_eq!(s.dead, 0);
    }

    #[test]
    fn mark_dead_increments_backoff_attempt() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        let backoff1 = ps.mark_dead(addr);
        assert_eq!(backoff1, Some(Duration::from_secs(10))); // attempt 0

        let attempt_after_first = ps.states.get(&addr).expect("entry exists").backoff_attempt;
        assert_eq!(attempt_after_first, 1);

        // Retry cycle: Dead → Queued → Connecting → Dead.
        ps.mark_queued_for_retry(addr);
        ps.mark_connecting(addr);
        let backoff2 = ps.mark_dead(addr);
        assert_eq!(backoff2, Some(Duration::from_mins(1))); // attempt 1

        let attempt_after_second = ps.states.get(&addr).expect("entry exists").backoff_attempt;
        assert_eq!(attempt_after_second, 2);
    }

    #[test]
    fn snapshot_is_serializable() {
        let snap = PeerPipelineSnapshot {
            known: 42,
            queued: 10,
            connecting: 5,
            live: 20,
            dead: 7,
        };
        let json = serde_json::to_string(&snap).expect("should serialize");
        assert!(json.contains("\"known\":42"));
        assert!(json.contains("\"live\":20"));
    }

    // ── M147: tcp_connected_at + connecting_since tests ─────────────────

    #[test]
    fn mark_connecting_sets_connecting_since() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);

        let before = Instant::now();
        ps.mark_connecting(addr);
        let after = Instant::now();

        let entry = ps.states.get(&addr).expect("entry should exist");
        let since = entry
            .connecting_since
            .expect("connecting_since should be set");
        assert!(since >= before && since <= after);
        assert!(entry.tcp_connected_at.is_none());
    }

    #[test]
    fn set_tcp_connected_marks_connecting_peer() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);

        let before = Instant::now();
        ps.set_tcp_connected(addr);
        let after = Instant::now();

        let entry = ps.states.get(&addr).expect("entry should exist");
        let connected_at = entry
            .tcp_connected_at
            .expect("tcp_connected_at should be set");
        assert!(connected_at >= before && connected_at <= after);
    }

    #[test]
    fn set_tcp_connected_no_op_for_non_connecting() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);

        // Peer is Queued, not Connecting — should be a no-op
        ps.set_tcp_connected(addr);
        let entry = ps.states.get(&addr).expect("entry should exist");
        assert!(
            entry.tcp_connected_at.is_none(),
            "should not set tcp_connected_at for non-Connecting peer"
        );
    }

    #[test]
    fn soft_reap_candidates_finds_unreachable_peers() {
        let (ps, _rx) = make_peer_states();
        let addr1 = test_addr(6881);
        let addr2 = test_addr_ip(2, 6882);
        let addr3 = test_addr_ip(3, 6883);

        ps.add_if_not_seen(addr1, PeerSource::Tracker);
        ps.add_if_not_seen(addr2, PeerSource::Tracker);
        ps.add_if_not_seen(addr3, PeerSource::Tracker);

        ps.mark_connecting(addr1);
        ps.mark_connecting(addr2);
        ps.mark_connecting(addr3);

        // Simulate: addr2 got TCP SYN-ACK
        ps.set_tcp_connected(addr2);

        // Backdate connecting_since for addr1 and addr3 to make them eligible.
        // v0.173.4 prong 2: must use `test_backdate_connecting_since` to keep
        // the BTreeMap side index in sync with the DashMap entry — direct
        // `iter_mut()` mutation would leave them out of sync and the
        // shadow-assert in `soft_reap_candidates_into` would panic.
        let backdated = Instant::now()
            .checked_sub(Duration::from_secs(5))
            .unwrap_or_else(Instant::now);
        ps.test_backdate_connecting_since(addr1, backdated);
        ps.test_backdate_connecting_since(addr3, backdated);

        let candidates = ps.soft_reap_candidates(Duration::from_secs(3));
        assert_eq!(
            candidates.len(),
            2,
            "should find 2 peers without TCP SYN-ACK past timeout"
        );
        assert!(candidates.contains(&addr1));
        assert!(candidates.contains(&addr3));
        assert!(
            !candidates.contains(&addr2),
            "peer with tcp_connected_at should be spared"
        );
    }

    #[test]
    fn soft_reap_spares_recently_connecting_peers() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);

        // connecting_since was just set — should NOT be a candidate with 3s timeout
        let candidates = ps.soft_reap_candidates(Duration::from_secs(3));
        assert!(
            candidates.is_empty(),
            "recently connecting peer should not be reaped"
        );
    }

    // ── v0.173.3: soft_reap_candidates_into buffer-fill variant ──

    #[test]
    fn soft_reap_candidates_into_matches_vec_variant() {
        let (ps, _rx) = make_peer_states();
        let addr1 = test_addr_ip(1, 1000);
        let addr2 = test_addr_ip(2, 1001);
        let addr3 = test_addr_ip(3, 1002);

        ps.add_if_not_seen(addr1, PeerSource::Tracker);
        ps.add_if_not_seen(addr2, PeerSource::Tracker);
        ps.add_if_not_seen(addr3, PeerSource::Tracker);

        ps.mark_connecting(addr1);
        ps.mark_connecting(addr2);
        ps.mark_connecting(addr3);

        std::thread::sleep(Duration::from_millis(50));

        let vec_result = ps.soft_reap_candidates(Duration::from_millis(10));
        let mut buf: Vec<SocketAddr> = Vec::new();
        ps.soft_reap_candidates_into(Duration::from_millis(10), &mut buf);

        let mut a = vec_result;
        let mut b = buf.clone();
        a.sort();
        b.sort();
        assert_eq!(a, b, "into variant must match Vec variant");
    }

    #[test]
    fn soft_reap_candidates_into_clears_caller_buffer() {
        let (ps, _rx) = make_peer_states();
        let mut buf = vec![test_addr(42), test_addr(43)]; // stale content

        ps.soft_reap_candidates_into(Duration::from_hours(1), &mut buf);

        // Nothing is eligible (no connecting peers, huge timeout) —
        // buffer must be cleared of stale content.
        assert!(buf.is_empty(), "into variant must clear caller buffer");
    }

    #[test]
    fn soft_reap_candidates_into_with_no_peers() {
        let (ps, _rx) = make_peer_states();
        let mut buf = vec![test_addr(99), test_addr(100)]; // stale content
        ps.soft_reap_candidates_into(Duration::from_millis(100), &mut buf);
        assert!(
            buf.is_empty(),
            "empty PeerStates must produce empty buffer (clearing stale content)"
        );
    }

    #[test]
    fn promotion_flow_increments_live_count() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        assert_eq!(ps.stats.snapshot().live, 0);

        ps.mark_connecting(addr);
        assert_eq!(ps.stats.snapshot().live, 0);
        assert_eq!(ps.stats.snapshot().connecting, 1);

        // Simulate TCP SYN-ACK
        ps.set_tcp_connected(addr);
        // Still Connecting — not yet Live
        assert_eq!(ps.stats.snapshot().live, 0);

        // HandshakeComplete → mark_live
        ps.mark_live(addr);
        assert_eq!(ps.stats.snapshot().live, 1);
        assert_eq!(ps.stats.snapshot().connecting, 0);
    }

    // ── M148: Eviction ban tests ──

    /// M148: `add_eviction_ban` + `is_eviction_banned` returns true for fresh ban.
    #[test]
    fn test_eviction_ban_blocks_peer() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(7001);

        assert!(
            !ps.is_eviction_banned(&addr),
            "should not be banned before add"
        );

        ps.add_eviction_ban(addr);

        assert!(
            ps.is_eviction_banned(&addr),
            "should be banned immediately after add"
        );
    }

    /// M148: Eviction ban contract — a fresh ban returns true, and the ban
    /// entry exists in the map. We cannot easily test the 30-minute expiry
    /// without mocking time, so we verify the contract: add → banned, and
    /// check the internal state for the timestamp being recent.
    #[test]
    fn test_eviction_ban_expires() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(7002);

        ps.add_eviction_ban(addr);
        assert!(ps.is_eviction_banned(&addr), "fresh ban should block");

        // Verify the ban is stored with a recent timestamp
        let entry = ps.eviction_bans.get(&addr).expect("ban entry should exist");
        assert!(
            entry.elapsed() < Duration::from_secs(5),
            "ban timestamp should be recent"
        );
    }

    /// v0.187.3 / OV4: FIFO cap drops the oldest entry when the banned set
    /// hits capacity. With cap=4, adding 5 distinct peers should leave 4
    /// entries, and the first one inserted should be evicted.
    #[test]
    fn banned_set_fifo_cap_drops_oldest_when_full() {
        let (tx, _rx) = mpsc::unbounded_channel();
        let ps = PeerStates::new_with_config(tx, 4, EVICTION_BAN_DURATION);

        let p1 = test_addr_ip(1, 8001);
        let p2 = test_addr_ip(2, 8002);
        let p3 = test_addr_ip(3, 8003);
        let p4 = test_addr_ip(4, 8004);
        let p5 = test_addr_ip(5, 8005);

        ps.add_eviction_ban(p1);
        ps.add_eviction_ban(p2);
        ps.add_eviction_ban(p3);
        ps.add_eviction_ban(p4);
        assert_eq!(ps.eviction_ban_count(), 4, "all 4 should fit at cap=4");

        // 5th entry must evict p1 (FIFO front).
        ps.add_eviction_ban(p5);
        assert_eq!(ps.eviction_ban_count(), 4, "cap should hold steady");
        assert!(
            !ps.is_eviction_banned(&p1),
            "oldest entry should be dropped"
        );
        assert!(ps.is_eviction_banned(&p2));
        assert!(ps.is_eviction_banned(&p3));
        assert!(ps.is_eviction_banned(&p4));
        assert!(ps.is_eviction_banned(&p5));
    }

    /// v0.187.3: configured ban duration is honoured by `is_eviction_banned`.
    /// A duration of `0` means the ban expires immediately on lookup, so
    /// `is_eviction_banned` returns false right after `add_eviction_ban`.
    #[test]
    fn ban_duration_zero_expires_immediately() {
        let (tx, _rx) = mpsc::unbounded_channel();
        let ps = PeerStates::new_with_config(tx, 1024, Duration::from_secs(0));
        let addr = test_addr(8100);
        ps.add_eviction_ban(addr);
        assert!(
            !ps.is_eviction_banned(&addr),
            "ban with zero duration must expire instantly"
        );
    }

    /// v0.187.3 / OV4: cap floor at 1 prevents misconfigured cap=0 from
    /// disabling the ban mechanism entirely. With cap=0 the set would
    /// otherwise drop every entry on insert; we clamp to 1.
    #[test]
    fn banned_set_cap_floor_is_one() {
        let (tx, _rx) = mpsc::unbounded_channel();
        let ps = PeerStates::new_with_config(tx, 0, EVICTION_BAN_DURATION);
        let only = test_addr_ip(1, 8101);
        ps.add_eviction_ban(only);
        assert!(
            ps.is_eviction_banned(&only),
            "cap-floor of 1 must hold the entry"
        );
    }

    /// M148: `is_eviction_banned` for an unknown address returns false.
    #[test]
    fn test_eviction_ban_not_present_returns_false() {
        let (ps, _rx) = make_peer_states();
        let unknown = test_addr(7003);
        assert!(
            !ps.is_eviction_banned(&unknown),
            "unknown address should not be banned"
        );
    }

    /// M148: Eviction ban blocks only the banned peer — other peers are
    /// unaffected. This mirrors the `peer_adder_task` check where
    /// `is_eviction_banned` is called per-address to filter reconnections.
    #[test]
    fn test_eviction_ban_blocks_specific_peer_only() {
        let (ps, _rx) = make_peer_states();

        let banned_addr = test_addr_ip(10, 7010);
        let innocent_addr = test_addr_ip(20, 7020);
        let another_addr = test_addr_ip(30, 7030);

        // Ban only the first peer
        ps.add_eviction_ban(banned_addr);

        // Banned peer is blocked
        assert!(
            ps.is_eviction_banned(&banned_addr),
            "banned peer should be blocked"
        );

        // Other peers are not affected
        assert!(
            !ps.is_eviction_banned(&innocent_addr),
            "innocent peer should not be blocked by another peer's ban"
        );
        assert!(
            !ps.is_eviction_banned(&another_addr),
            "unrelated peer should not be blocked"
        );

        // Ban a second peer — first is still banned, third is still free
        ps.add_eviction_ban(innocent_addr);
        assert!(
            ps.is_eviction_banned(&banned_addr),
            "first ban should still be active"
        );
        assert!(
            ps.is_eviction_banned(&innocent_addr),
            "second peer should now be banned"
        );
        assert!(
            !ps.is_eviction_banned(&another_addr),
            "third peer should remain unaffected"
        );
    }

    // ── v0.173.4 prong 2: connecting_since_index machinery ──────────────

    /// Returns the total number of (Instant → bucket) entries in the index.
    fn idx_buckets(ps: &PeerStates) -> usize {
        ps.connecting_since_index.lock().len()
    }

    /// Returns the total number of addrs across all buckets.
    fn idx_total_addrs(ps: &PeerStates) -> usize {
        ps.connecting_since_index
            .lock()
            .values()
            .map(smallvec::SmallVec::len)
            .sum()
    }

    fn idx_contains(ps: &PeerStates, addr: SocketAddr) -> bool {
        ps.connecting_since_index
            .lock()
            .values()
            .any(|bucket| bucket.contains(&addr))
    }

    #[test]
    fn index_inserted_on_mark_connecting() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        assert_eq!(idx_total_addrs(&ps), 0, "Queued state must not be indexed");

        ps.mark_connecting(addr);
        assert_eq!(idx_total_addrs(&ps), 1);
        assert!(idx_contains(&ps, addr));

        // The bucket key matches the entry's connecting_since.
        let entry_ts = ps
            .states
            .get(&addr)
            .expect("entry exists")
            .connecting_since
            .expect("connecting_since set");
        assert!(ps.connecting_since_index.lock().contains_key(&entry_ts));
    }

    #[test]
    fn index_removed_on_set_tcp_connected() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        assert_eq!(idx_total_addrs(&ps), 1);

        // Outside-voice review #2: SYN-ACK arrival must remove from index,
        // otherwise stalled BT-handshake peers leak forever.
        ps.set_tcp_connected(addr);
        assert_eq!(idx_total_addrs(&ps), 0);
        assert_eq!(idx_buckets(&ps), 0, "empty buckets must be pruned");
    }

    #[test]
    fn index_removed_on_mark_live() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        ps.mark_live(addr);

        assert_eq!(idx_total_addrs(&ps), 0);
        assert_eq!(idx_buckets(&ps), 0);
    }

    #[test]
    fn index_removed_on_mark_dead_normal() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        assert_eq!(idx_total_addrs(&ps), 1);

        // Connecting → Dead within retry window: index must be pruned.
        let backoff = ps.mark_dead(addr);
        assert!(backoff.is_some(), "should return backoff in retry window");
        assert_eq!(idx_total_addrs(&ps), 0);
        assert_eq!(idx_buckets(&ps), 0);
    }

    #[test]
    fn index_removed_on_mark_dead_exhausted() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        assert_eq!(idx_total_addrs(&ps), 1);

        // Backdate first_failure_at past the 24-hour retry window.
        if let Some(mut entry) = ps.states.get_mut(&addr) {
            let too_old = Instant::now().checked_sub(Duration::from_secs(RETRY_WINDOW_SECS + 1));
            if let Some(old) = too_old {
                entry.first_failure_at = Some(old);
            } else {
                // System uptime < 24h; cannot test exhaustion path.
                return;
            }
        }

        let result = ps.mark_dead(addr);
        assert!(result.is_none(), "exhausted path returns None");
        assert!(
            ps.states.get(&addr).is_none(),
            "DashMap entry must be removed in exhausted path"
        );
        assert_eq!(idx_total_addrs(&ps), 0, "index must also be empty");
        assert_eq!(idx_buckets(&ps), 0);
    }

    #[test]
    fn index_removed_on_mark_dead_from_live_no_op() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        // Live state has no index entry, so mark_dead from Live must be a
        // pure no-op on the index. Outside-voice review #4 derivative:
        // verifies branch 3 of the mark_dead state machine.
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        ps.mark_live(addr);
        assert_eq!(idx_total_addrs(&ps), 0, "Live state must not be indexed");

        let _ = ps.mark_dead(addr);
        assert_eq!(idx_total_addrs(&ps), 0);
        assert_eq!(idx_buckets(&ps), 0);
    }

    #[test]
    fn mark_queued_for_retry_leaves_index_empty() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        // The full Connecting→Dead→Queued path. mark_dead removes from
        // index; mark_queued_for_retry must NOT re-insert (the next
        // mark_connecting does that). Outside-voice review #4 derivative.
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        let _ = ps.mark_dead(addr);
        assert_eq!(idx_total_addrs(&ps), 0);

        ps.mark_queued_for_retry(addr);
        assert_eq!(
            idx_total_addrs(&ps),
            0,
            "mark_queued_for_retry must not touch the index"
        );
    }

    #[test]
    fn index_handles_relifecycle() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);

        // Cycle 1: Queued → Connecting → Live → Dead → Queued.
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);
        let first_ts = ps
            .states
            .get(&addr)
            .expect("entry exists")
            .connecting_since
            .expect("connecting_since set");
        ps.mark_live(addr);
        let _ = ps.mark_dead(addr);
        ps.mark_queued_for_retry(addr);
        assert_eq!(idx_total_addrs(&ps), 0);

        // Cycle 2: Queued → Connecting again. New connecting_since = new
        // index bucket. Sleep briefly so the second Instant differs.
        std::thread::sleep(Duration::from_millis(2));
        ps.mark_connecting(addr);
        let second_ts = ps
            .states
            .get(&addr)
            .expect("entry exists")
            .connecting_since
            .expect("connecting_since set");

        assert!(
            second_ts > first_ts,
            "second cycle must yield a later connecting_since"
        );
        assert_eq!(idx_total_addrs(&ps), 1);
        assert!(
            ps.connecting_since_index.lock().contains_key(&second_ts),
            "second cycle's Instant must be the index key"
        );
        assert!(
            !ps.connecting_since_index.lock().contains_key(&first_ts),
            "first cycle's Instant must not linger in the index"
        );
    }

    #[test]
    fn index_handles_collision() {
        let (ps, _rx) = make_peer_states();
        let addr1 = test_addr_ip(1, 6881);
        let addr2 = test_addr_ip(2, 6882);

        // Two peers backdated to identical Instant. Their index entries
        // must coexist in a single SmallVec bucket. Exercises the SmallVec
        // collision case (peer_adder_task admits multiple peers per tick).
        ps.add_if_not_seen(addr1, PeerSource::Tracker);
        ps.add_if_not_seen(addr2, PeerSource::Tracker);
        ps.mark_connecting(addr1);
        ps.mark_connecting(addr2);

        let shared_ts = Instant::now()
            .checked_sub(Duration::from_secs(5))
            .unwrap_or_else(Instant::now);
        ps.test_backdate_connecting_since(addr1, shared_ts);
        ps.test_backdate_connecting_since(addr2, shared_ts);

        let idx = ps.connecting_since_index.lock();
        let bucket = idx.get(&shared_ts).expect("collision bucket exists");
        assert_eq!(bucket.len(), 2, "both addrs collide into one bucket");
        assert!(bucket.contains(&addr1));
        assert!(bucket.contains(&addr2));
        drop(idx);

        // Both peers must reap together.
        let candidates = ps.soft_reap_candidates(Duration::from_secs(3));
        assert_eq!(candidates.len(), 2);
        assert!(candidates.contains(&addr1));
        assert!(candidates.contains(&addr2));
    }

    #[test]
    fn soft_reap_short_circuits_on_first_unexpired() {
        let (ps, _rx) = make_peer_states();
        let mut expired = Vec::with_capacity(50);
        let mut fresh = Vec::with_capacity(50);

        // 100 distinct addrs split 50/50 expired/fresh.
        for i in 0u8..100 {
            let addr = test_addr_ip(i, 6881);
            ps.add_if_not_seen(addr, PeerSource::Tracker);
            ps.mark_connecting(addr);
            if i < 50 {
                expired.push(addr);
            } else {
                fresh.push(addr);
            }
        }

        // Backdate the first 50 to make them eligible.
        let backdated = Instant::now()
            .checked_sub(Duration::from_secs(10))
            .unwrap_or_else(Instant::now);
        for &addr in &expired {
            ps.test_backdate_connecting_since(addr, backdated);
        }

        let candidates = ps.soft_reap_candidates(Duration::from_secs(5));
        assert_eq!(candidates.len(), 50, "exactly 50 expired peers");
        for &addr in &expired {
            assert!(candidates.contains(&addr), "expired addr {addr:?} missing");
        }
        for &addr in &fresh {
            assert!(
                !candidates.contains(&addr),
                "fresh addr {addr:?} must not be reaped"
            );
        }
    }

    #[test]
    fn soft_reap_handles_max_timeout() {
        let (ps, _rx) = make_peer_states();
        let addr = test_addr(6881);
        ps.add_if_not_seen(addr, PeerSource::Tracker);
        ps.mark_connecting(addr);

        // Duration::MAX → checked_sub returns None on most platforms (the
        // Instant is some monotonic point well below i64::MAX seconds since
        // process start). soft_reap_candidates_into must early-return with
        // an empty buffer, not panic.
        let candidates = ps.soft_reap_candidates(Duration::MAX);
        assert!(
            candidates.is_empty(),
            "Duration::MAX must not yield any reap candidates"
        );
    }

    #[test]
    fn soft_reap_stress_500_peers() {
        let (ps, _rx) = make_peer_states();
        let mut addrs = Vec::with_capacity(500);

        for i in 0u32..500 {
            #[allow(clippy::cast_possible_truncation)]
            let octet1 = ((i / 256) & 0xff) as u8;
            #[allow(clippy::cast_possible_truncation)]
            let octet2 = (i & 0xff) as u8;
            let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, octet1, octet2, 1)), 6881);
            ps.add_if_not_seen(addr, PeerSource::Tracker);
            ps.mark_connecting(addr);
            addrs.push(addr);
        }
        assert_eq!(idx_total_addrs(&ps), 500);

        // Backdate the first 250 — 250 reapable, 250 fresh.
        let backdated = Instant::now()
            .checked_sub(Duration::from_secs(10))
            .unwrap_or_else(Instant::now);
        for &addr in &addrs[..250] {
            ps.test_backdate_connecting_since(addr, backdated);
        }

        let candidates = ps.soft_reap_candidates(Duration::from_secs(5));
        assert_eq!(candidates.len(), 250, "250 backdated peers must reap");

        // Memory-leak canary (outside-voice review #9): after every peer
        // transitions to Live, the index must be empty. If this fails, an
        // index entry leaked through a transition path.
        for &addr in &addrs {
            ps.mark_live(addr);
        }
        assert_eq!(
            idx_total_addrs(&ps),
            0,
            "leak canary: index must drain to 0 after all peers go Live"
        );
        assert_eq!(
            idx_buckets(&ps),
            0,
            "leak canary: index buckets must all be pruned"
        );
    }

    // ── Property-based test: index/DashMap invariant under random ops ──

    /// Verifies the load-bearing invariant in BOTH directions:
    /// 1. Every addr in the index has `state == Connecting` AND
    ///    `tcp_connected_at.is_none()` in the `DashMap`.
    /// 2. Every `DashMap` entry with `state == Connecting` AND
    ///    `tcp_connected_at.is_none()` is in the index.
    fn assert_index_dashmap_invariant(ps: &PeerStates) {
        // Forward direction: index → DashMap.
        let idx_entries: Vec<(Instant, Vec<SocketAddr>)> = ps
            .connecting_since_index
            .lock()
            .iter()
            .map(|(ts, bucket)| (*ts, bucket.iter().copied().collect()))
            .collect();
        for (ts, bucket) in &idx_entries {
            for addr in bucket {
                let entry = ps
                    .states
                    .get(addr)
                    .unwrap_or_else(|| panic!("indexed addr {addr:?} missing from DashMap"));
                assert_eq!(
                    entry.state,
                    PeerLifecycle::Connecting,
                    "indexed addr {addr:?} has state {:?}, expected Connecting",
                    entry.state,
                );
                assert!(
                    entry.tcp_connected_at.is_none(),
                    "indexed addr {addr:?} has tcp_connected_at set",
                );
                assert_eq!(
                    entry.connecting_since,
                    Some(*ts),
                    "indexed addr {addr:?} bucket-ts {ts:?} mismatch with entry connecting_since",
                );
            }
        }

        // Reverse direction: DashMap → index.
        for entry in &ps.states {
            if entry.state == PeerLifecycle::Connecting && entry.tcp_connected_at.is_none() {
                let addr = *entry.key();
                let ts = entry
                    .connecting_since
                    .expect("Connecting must have connecting_since");
                drop(entry);
                let idx = ps.connecting_since_index.lock();
                let bucket = idx
                    .get(&ts)
                    .unwrap_or_else(|| panic!("Connecting addr {addr:?} missing from index"));
                assert!(
                    bucket.contains(&addr),
                    "Connecting addr {addr:?} not in bucket at ts {ts:?}",
                );
            }
        }
    }

    fn pool_addr(i: u8) -> SocketAddr {
        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, i)), 6881)
    }

    #[derive(Debug, Clone, Copy)]
    enum Op {
        Add(u8),
        MarkConnecting(u8),
        SetTcpConnected(u8),
        MarkLive(u8),
        MarkDead(u8),
        MarkQueuedForRetry(u8),
    }

    use proptest::prelude::*;

    proptest! {
        #![proptest_config(ProptestConfig {
            cases: 64,
            ..ProptestConfig::default()
        })]

        #[test]
        fn index_dashmap_invariant_under_random_transitions(
            ops in proptest::collection::vec(
                prop_oneof![
                    (0u8..8u8).prop_map(Op::Add),
                    (0u8..8u8).prop_map(Op::MarkConnecting),
                    (0u8..8u8).prop_map(Op::SetTcpConnected),
                    (0u8..8u8).prop_map(Op::MarkLive),
                    (0u8..8u8).prop_map(Op::MarkDead),
                    (0u8..8u8).prop_map(Op::MarkQueuedForRetry),
                ],
                1..200,
            )
        ) {
            let (ps, _rx) = make_peer_states();
            for op in ops {
                let i = match op {
                    Op::Add(i)
                    | Op::MarkConnecting(i)
                    | Op::SetTcpConnected(i)
                    | Op::MarkLive(i)
                    | Op::MarkDead(i)
                    | Op::MarkQueuedForRetry(i) => i,
                };
                let addr = pool_addr(i);
                match op {
                    Op::Add(_) => {
                        ps.add_if_not_seen(addr, PeerSource::Tracker);
                    }
                    Op::MarkConnecting(_) => {
                        ps.mark_connecting(addr);
                    }
                    Op::SetTcpConnected(_) => {
                        ps.set_tcp_connected(addr);
                    }
                    Op::MarkLive(_) => {
                        ps.mark_live(addr);
                    }
                    Op::MarkDead(_) => {
                        let _ = ps.mark_dead(addr);
                    }
                    Op::MarkQueuedForRetry(_) => {
                        ps.mark_queued_for_retry(addr);
                    }
                }
                assert_index_dashmap_invariant(&ps);
            }
        }
    }
}