udp_relay_core/

lib.rs

//! Lock-free, cache-line-aligned core types for building high-performance
//! UDP relay and tunnel servers.
//!
//! # Overview
//!
//! This crate provides the shared-state primitives you need when forwarding
//! UDP packets between many clients at high throughput:
//!
//! * [`TunnelClient`] — per-client atomic metrics (bandwidth, latency,
//!   packet loss, priority scoring) designed for `Arc<TunnelClient>` in a
//!   [`DashMap`].
//! * [`QualityAnalyzer`] — lightweight packet loss tracker with automatic
//!   counter halving so counters never overflow.
//! * [`validate_address`] — rejects loopback, unspecified, broadcast,
//!   multicast, and port-0 addresses.
//! * [`create_dashmap_with_capacity`] — creates a [`DashMap`] with a shard
//!   count tuned to the CPU count.
//!
//! # Quick start
//!
//! ```rust
//! use std::sync::Arc;
//! use udp_relay_core::{TunnelClient, create_dashmap_with_capacity};
//!
//! let clients = create_dashmap_with_capacity::<u32, Arc<TunnelClient>>(200);
//!
//! let addr = "1.2.3.4:5678".parse().unwrap();
//! let client = Arc::new(TunnelClient::new_with_endpoint(addr, 30));
//! clients.insert(42, client.clone());
//!
//! let now = TunnelClient::current_timestamp();
//! client.update_stats(512, 0, now);
//! client.set_last_receive_tick_at(now);
//!
//! if client.is_timed_out() {
//!     clients.remove(&42);
//! }
//! ```

use std::net::{IpAddr, SocketAddr};
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};

use dashmap::DashMap;

/// A UDP tunnel client with lock-free atomic metrics.
///
/// Designed to live behind `Arc` in a `DashMap`, so all fields use bare atomics
/// (no `Arc` wrappers). Not `Clone` — share via `Arc<TunnelClient>`.
///
/// Fields are ordered so the hot-path data (packet receive + stats) sits on the
/// first cache line, and cold/periodic data sits on the second.
#[repr(C, align(128))]
pub struct TunnelClient {
    // --- cache line 0: hot path (packet receive + bandwidth) ---
    last_receive_tick: AtomicU64,
    packet_count: AtomicU64,
    bytes_received: AtomicU64,
    bytes_sent: AtomicU64,
    bandwidth_estimate: AtomicU64,
    last_bandwidth_calc: AtomicU64,
    last_bandwidth_bytes: AtomicU64,
    timeout_seconds: u64,

    // --- cache line 1: cold / periodic ---
    priority_score: AtomicU64,
    connection_start: AtomicU64,
    priority_dirty: AtomicU32,
    latency_ms: AtomicU32,
    packet_loss_rate: AtomicU32,
    is_slow_connection: AtomicU32,
    pub remote_ep: Option<SocketAddr>,
}

impl TunnelClient {
    /// Create a new client without a remote endpoint.
    ///
    /// The client will time out after `timeout_seconds` of inactivity.
    #[must_use]
    pub fn new(timeout_seconds: u64) -> Self {
        let now = Self::current_timestamp();
        Self {
            last_receive_tick: AtomicU64::new(now),
            timeout_seconds,
            packet_count: AtomicU64::new(0),
            bytes_received: AtomicU64::new(0),
            bytes_sent: AtomicU64::new(0),
            bandwidth_estimate: AtomicU64::new(0),
            last_bandwidth_calc: AtomicU64::new(now),
            last_bandwidth_bytes: AtomicU64::new(0),
            priority_score: AtomicU64::new(1000),
            priority_dirty: AtomicU32::new(0),
            latency_ms: AtomicU32::new(0),
            packet_loss_rate: AtomicU32::new(0),
            connection_start: AtomicU64::new(now),
            is_slow_connection: AtomicU32::new(0),
            remote_ep: None,
        }
    }

    /// Create a new client bound to the given remote endpoint.
    ///
    /// The client will time out after `timeout_seconds` of inactivity.
    #[must_use]
    pub fn new_with_endpoint(addr: SocketAddr, timeout_seconds: u64) -> Self {
        let now = Self::current_timestamp();
        Self {
            last_receive_tick: AtomicU64::new(now),
            timeout_seconds,
            packet_count: AtomicU64::new(0),
            bytes_received: AtomicU64::new(0),
            bytes_sent: AtomicU64::new(0),
            bandwidth_estimate: AtomicU64::new(0),
            last_bandwidth_calc: AtomicU64::new(now),
            last_bandwidth_bytes: AtomicU64::new(0),
            priority_score: AtomicU64::new(1000),
            priority_dirty: AtomicU32::new(0),
            latency_ms: AtomicU32::new(0),
            packet_loss_rate: AtomicU32::new(0),
            connection_start: AtomicU64::new(now),
            is_slow_connection: AtomicU32::new(0),
            remote_ep: Some(addr),
        }
    }

    /// Mark the client as having received a packet at `now` (unix seconds).
    /// Call with a cached timestamp from the event loop — avoids a syscall.
    #[inline]
    pub fn set_last_receive_tick_at(&self, now: u64) {
        self.last_receive_tick.store(now, Ordering::Release);
    }

    /// Mark the client as having received a packet right now.
    /// Convenience wrapper that fetches the clock; prefer `set_last_receive_tick_at`
    /// on the hot path.
    #[inline]
    pub fn set_last_receive_tick(&self) {
        self.set_last_receive_tick_at(Self::current_timestamp());
    }

    /// Returns `true` if no packet has been received within the configured
    /// timeout window.
    #[inline]
    #[must_use]
    pub fn is_timed_out(&self) -> bool {
        let current = Self::current_timestamp();
        let last = self.last_receive_tick.load(Ordering::Acquire);
        current.saturating_sub(last) >= self.timeout_seconds
    }

    /// Record packet bytes. `now` should be a cached unix-seconds timestamp
    /// from the event loop — avoids a per-packet syscall.
    #[inline]
    pub fn update_stats(&self, bytes_in: usize, bytes_out: usize, now: u64) {
        self.packet_count.fetch_add(1, Ordering::Relaxed);
        self.bytes_received
            .fetch_add(bytes_in as u64, Ordering::Relaxed);
        self.bytes_sent
            .fetch_add(bytes_out as u64, Ordering::Relaxed);

        let last_calc = self.last_bandwidth_calc.load(Ordering::Relaxed);

        if now > last_calc {
            let time_delta = now - last_calc;
            if time_delta >= 1 {
                // CAS ensures only one thread wins; losers skip the recalculation.
                if self
                    .last_bandwidth_calc
                    .compare_exchange(last_calc, now, Ordering::AcqRel, Ordering::Relaxed)
                    .is_err()
                {
                    return;
                }

                let current_total = self.bytes_received.load(Ordering::Relaxed)
                    + self.bytes_sent.load(Ordering::Relaxed);
                let previous_total =
                    self.last_bandwidth_bytes
                        .swap(current_total, Ordering::Relaxed);
                let delta_bytes = current_total.saturating_sub(previous_total);
                let bandwidth = delta_bytes / time_delta;

                let _ = self.bandwidth_estimate.fetch_update(
                    Ordering::Relaxed,
                    Ordering::Relaxed,
                    |old_estimate| Some((old_estimate * 7 + bandwidth * 3) / 10),
                );

                self.mark_priority_dirty();
            }
        }
    }

    #[inline]
    fn mark_priority_dirty(&self) {
        self.priority_dirty.store(1, Ordering::Relaxed);
    }

    fn recompute_priority(&self) {
        let now = Self::current_timestamp();
        let bandwidth = self.bandwidth_estimate.load(Ordering::Relaxed);
        let latency = u64::from(self.latency_ms.load(Ordering::Relaxed));
        let loss_rate = u64::from(self.packet_loss_rate.load(Ordering::Relaxed));
        let connection_age = now.saturating_sub(self.connection_start.load(Ordering::Relaxed));

        // Approximate 1_000_000 / (bandwidth + 1000) using shifts:
        //   ~1_048_576 / (bandwidth + 1024) via right-shift by 20 of the numerator.
        // Avoids a hardware `div` on the hot path.
        let bw_denom = bandwidth.max(100) + 1024;
        let bandwidth_score = (1 << 20) / bw_denom;
        let latency_score = latency.min(500);
        let loss_score = loss_rate;
        let age_score = (connection_age / 10).min(100);

        let priority = bandwidth_score
            .saturating_add(latency_score)
            .saturating_add(loss_score)
            .saturating_add(age_score);

        self.priority_score.store(priority, Ordering::Relaxed);

        let is_slow = u32::from(bandwidth < 100_000);
        self.is_slow_connection.store(is_slow, Ordering::Relaxed);
    }

    /// Returns the priority score (lower is better), recomputing lazily only
    /// when metrics have changed since the last read.
    #[inline]
    #[must_use]
    pub fn get_priority(&self) -> u64 {
        if self.priority_dirty.swap(0, Ordering::Relaxed) != 0 {
            self.recompute_priority();
        }
        self.priority_score.load(Ordering::Relaxed)
    }

    /// Record the measured round-trip latency in milliseconds.
    ///
    /// Marks the priority score as dirty so it will be recomputed on the
    /// next call to [`Self::get_priority`].
    #[inline]
    pub fn set_latency(&self, latency_ms: u32) {
        self.latency_ms.store(latency_ms, Ordering::Relaxed);
        self.mark_priority_dirty();
    }

    /// Record the measured packet loss rate (0–1000, i.e. 0.0%–100.0%).
    ///
    /// Values above 1000 are clamped. Marks the priority score as dirty.
    #[inline]
    pub fn set_packet_loss_rate(&self, rate: u32) {
        self.packet_loss_rate
            .store(rate.min(1000), Ordering::Relaxed);
        self.mark_priority_dirty();
    }

    /// Returns `true` if the estimated bandwidth is below 100 KB/s.
    ///
    /// Updated as a side-effect of [`Self::get_priority`] (lazy recompute).
    #[inline]
    #[must_use]
    pub fn is_slow_connection(&self) -> bool {
        self.is_slow_connection.load(Ordering::Relaxed) == 1
    }

    /// Returns the current unix timestamp in seconds using `coarsetime`.
    ///
    /// Uses `CLOCK_MONOTONIC_COARSE` on Linux (~10-25x faster than `SystemTime::now()`).
    /// For the absolute fastest path, call [`Self::update_clock`] once per event-loop
    /// iteration and use [`Self::recent_timestamp`] everywhere else.
    #[inline]
    #[must_use]
    pub fn current_timestamp() -> u64 {
        coarsetime::Clock::now_since_epoch().as_secs()
    }

    /// Returns the most recently cached unix timestamp in seconds.
    ///
    /// Zero overhead — a single atomic load. Requires [`Self::update_clock`] to be
    /// called periodically (e.g. once per event-loop tick).
    #[inline]
    #[must_use]
    pub fn recent_timestamp() -> u64 {
        coarsetime::Clock::recent_since_epoch().as_secs()
    }

    /// Refresh the cached clock value.  Call once per event-loop iteration so
    /// that [`Self::recent_timestamp`] stays fresh.
    #[inline]
    pub fn update_clock() {
        coarsetime::Clock::update();
    }

    /// Reset all traffic counters and the bandwidth estimate to zero.
    ///
    /// Also resets the connection start time and marks priority as dirty.
    pub fn reset_stats(&self) {
        let now = Self::current_timestamp();
        self.packet_count.store(0, Ordering::Relaxed);
        self.bytes_received.store(0, Ordering::Relaxed);
        self.bytes_sent.store(0, Ordering::Relaxed);
        self.bandwidth_estimate.store(0, Ordering::Relaxed);
        self.last_bandwidth_calc.store(now, Ordering::Relaxed);
        self.last_bandwidth_bytes.store(0, Ordering::Relaxed);
        self.connection_start.store(now, Ordering::Relaxed);
        self.mark_priority_dirty();
    }
}

/// Lock-free packet loss tracker with periodic counter halving to avoid overflow.
///
/// Uses atomics so it can be shared across threads behind `Arc` without a mutex.
pub struct QualityAnalyzer {
    loss_count: AtomicU32,
    total_count: AtomicU32,
}

impl QualityAnalyzer {
    /// Create a new analyzer with zeroed counters.
    #[must_use]
    pub fn new() -> Self {
        Self {
            loss_count: AtomicU32::new(0),
            total_count: AtomicU32::new(0),
        }
    }

    /// Record a packet. Pass `true` if the packet was lost.
    ///
    /// Thread-safe: multiple threads may call this concurrently.
    /// The halving step uses a CAS loop so only one thread performs
    /// the reduction.
    #[inline]
    pub fn record_packet(&self, lost: bool) {
        if lost {
            self.loss_count.fetch_add(1, Ordering::Relaxed);
        }
        let prev = self.total_count.fetch_add(1, Ordering::Relaxed);
        if prev + 1 > 10_000 {
            let did_halve = self.total_count.fetch_update(
                Ordering::Relaxed,
                Ordering::Relaxed,
                |t| if t > 10_000 { Some(t / 2) } else { None },
            ).is_ok();

            if did_halve {
                let _ = self.loss_count.fetch_update(
                    Ordering::Relaxed,
                    Ordering::Relaxed,
                    |l| Some(l / 2),
                );
            }
        }
    }

    /// Returns packet loss rate as 0–1000 (i.e. 0.0%–100.0%).
    #[inline]
    #[must_use]
    pub fn get_packet_loss_rate(&self) -> u32 {
        // Read loss *before* total: if a halving interleaves between the two
        // loads we overestimate rather than underestimate — safer for QoS.
        let loss = self.loss_count.load(Ordering::Relaxed);
        let total = self.total_count.load(Ordering::Relaxed);
        if total == 0 {
            return 0;
        }
        ((u64::from(loss) * 1000) / u64::from(total)).min(1000) as u32
    }
}

impl Default for QualityAnalyzer {
    fn default() -> Self {
        Self::new()
    }
}

/// Validate that a socket address is routable (not loopback/unspecified/broadcast/multicast)
/// and has a non-zero port.
#[inline]
#[must_use]
pub fn validate_address(addr: &SocketAddr) -> bool {
    let ip_valid = match addr.ip() {
        IpAddr::V4(v4) => {
            !v4.is_loopback() && !v4.is_unspecified() && !v4.is_broadcast() && !v4.is_multicast()
        }
        IpAddr::V6(v6) => !v6.is_loopback() && !v6.is_unspecified() && !v6.is_multicast(),
    };
    ip_valid && addr.port() != 0
}

/// Create a [`DashMap`] with a shard count tuned to the CPU count.
///
/// Uses 4× the CPU count (rounded to next power-of-two, clamped 4..=256) so
/// `DashMap` can use bitmask-based shard selection.
pub fn create_dashmap_with_capacity<K, V>(capacity: usize) -> DashMap<K, V>
where
    K: Eq + std::hash::Hash,
{
    #[allow(clippy::redundant_closure_for_method_calls)]
    let detected_cpus = std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(1);

    let shard_amount = (detected_cpus * 4).next_power_of_two().clamp(4, 256);

    tracing::debug!(
        "Creating DashMap with {} shards for {} CPUs (capacity: {})",
        shard_amount,
        detected_cpus,
        capacity
    );

    DashMap::with_capacity_and_shard_amount(capacity, shard_amount)
}

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

    // ──────────────────────────────────────────────
    // TunnelClient — construction
    // ──────────────────────────────────────────────

    #[test]
    fn new_with_endpoint_stores_address() {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 9000);
        let client = TunnelClient::new_with_endpoint(addr, 30);
        assert_eq!(client.remote_ep, Some(addr));
    }

    #[test]
    fn new_without_endpoint_is_none() {
        let client = TunnelClient::new(30);
        assert_eq!(client.remote_ep, None);
    }

    // ──────────────────────────────────────────────
    // TunnelClient — timeout
    // ──────────────────────────────────────────────

    #[test]
    fn tunnel_client_timeout() {
        let client = TunnelClient::new(0);
        assert!(client.is_timed_out());
    }

    #[test]
    fn tunnel_client_not_timed_out() {
        let client = TunnelClient::new(60);
        assert!(!client.is_timed_out());
    }

    #[test]
    fn set_last_receive_tick_resets_timeout() {
        let client = TunnelClient::new(0);
        assert!(client.is_timed_out());
        client.set_last_receive_tick();
        let client2 = TunnelClient::new(60);
        client2.set_last_receive_tick_at(TunnelClient::current_timestamp());
        assert!(!client2.is_timed_out());
    }

    #[test]
    fn set_last_receive_tick_at_with_old_timestamp_times_out() {
        let client = TunnelClient::new(5);
        client.set_last_receive_tick_at(0);
        assert!(client.is_timed_out());
    }

    // ──────────────────────────────────────────────
    // TunnelClient — update_stats
    // ──────────────────────────────────────────────

    #[test]
    fn update_stats_increments_counters() {
        let client = TunnelClient::new(60);
        let now = TunnelClient::current_timestamp();
        client.update_stats(100, 50, now);
        assert_eq!(client.packet_count.load(Ordering::Relaxed), 1);
        assert_eq!(client.bytes_received.load(Ordering::Relaxed), 100);
        assert_eq!(client.bytes_sent.load(Ordering::Relaxed), 50);
    }

    #[test]
    fn update_stats_accumulates_multiple_packets() {
        let client = TunnelClient::new(60);
        let now = TunnelClient::current_timestamp();
        client.update_stats(100, 50, now);
        client.update_stats(200, 75, now);
        client.update_stats(300, 25, now);
        assert_eq!(client.packet_count.load(Ordering::Relaxed), 3);
        assert_eq!(client.bytes_received.load(Ordering::Relaxed), 600);
        assert_eq!(client.bytes_sent.load(Ordering::Relaxed), 150);
    }

    #[test]
    fn update_stats_computes_bandwidth_after_time_delta() {
        let client = TunnelClient::new(60);
        let t0 = TunnelClient::current_timestamp();
        client.update_stats(5000, 5000, t0);

        let t1 = t0 + 2;
        client.update_stats(5000, 5000, t1);

        let bw = client.bandwidth_estimate.load(Ordering::Relaxed);
        assert!(bw > 0, "bandwidth should be non-zero after time delta, got {bw}");
    }

    #[test]
    fn update_stats_skips_bandwidth_when_now_equals_last_calc() {
        let client = TunnelClient::new(60);
        let now = TunnelClient::current_timestamp();
        client.update_stats(1000, 1000, now);
        client.update_stats(1000, 1000, now);
        let bw = client.bandwidth_estimate.load(Ordering::Relaxed);
        assert_eq!(bw, 0, "bandwidth should not update when time_delta == 0");
    }

    #[test]
    fn update_stats_skips_bandwidth_when_now_is_in_the_past() {
        let client = TunnelClient::new(60);
        let now = TunnelClient::current_timestamp();
        client.update_stats(1000, 1000, now);
        client.update_stats(1000, 1000, now.saturating_sub(10));
        let bw = client.bandwidth_estimate.load(Ordering::Relaxed);
        assert_eq!(bw, 0, "bandwidth should not update when now < last_calc");
    }

    // ──────────────────────────────────────────────
    // TunnelClient — priority, latency, loss, slow
    // ──────────────────────────────────────────────

    #[test]
    fn priority_is_lazy() {
        let client = TunnelClient::new(60);
        let initial = client.get_priority();
        assert!(initial > 0);
        client.set_latency(100);
        let score_before_read = client.priority_score.load(Ordering::Relaxed);
        let score_after_read = client.get_priority();
        assert_eq!(score_before_read, initial);
        assert_ne!(score_after_read, initial);
    }

    #[test]
    fn set_latency_stores_value() {
        let client = TunnelClient::new(60);
        client.set_latency(42);
        assert_eq!(client.latency_ms.load(Ordering::Relaxed), 42);
    }

    #[test]
    fn set_packet_loss_rate_clamps_to_1000() {
        let client = TunnelClient::new(60);
        client.set_packet_loss_rate(9999);
        assert_eq!(client.packet_loss_rate.load(Ordering::Relaxed), 1000);
    }

    #[test]
    fn set_packet_loss_rate_stores_valid_value() {
        let client = TunnelClient::new(60);
        client.set_packet_loss_rate(250);
        assert_eq!(client.packet_loss_rate.load(Ordering::Relaxed), 250);
    }

    #[test]
    fn high_latency_increases_priority_score() {
        let client = TunnelClient::new(60);
        let low = {
            client.set_latency(0);
            client.get_priority()
        };
        let high = {
            client.set_latency(400);
            client.get_priority()
        };
        assert!(high > low, "higher latency should increase priority score (lower is better)");
    }

    #[test]
    fn is_slow_connection_reflects_bandwidth() {
        let client = TunnelClient::new(60);
        client.set_latency(0);
        let _ = client.get_priority();
        assert!(client.is_slow_connection());
    }

    // ──────────────────────────────────────────────
    // TunnelClient — reset_stats
    // ──────────────────────────────────────────────

    #[test]
    fn reset_stats_zeroes_counters() {
        let client = TunnelClient::new(60);
        let now = TunnelClient::current_timestamp();
        client.update_stats(1000, 500, now);
        client.update_stats(1000, 500, now + 2);
        client.reset_stats();

        assert_eq!(client.packet_count.load(Ordering::Relaxed), 0);
        assert_eq!(client.bytes_received.load(Ordering::Relaxed), 0);
        assert_eq!(client.bytes_sent.load(Ordering::Relaxed), 0);
        assert_eq!(client.bandwidth_estimate.load(Ordering::Relaxed), 0);
        assert_eq!(client.last_bandwidth_bytes.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn reset_stats_marks_priority_dirty() {
        let client = TunnelClient::new(60);
        let _ = client.get_priority(); // clear dirty flag
        client.reset_stats();
        assert_eq!(
            client.priority_dirty.load(Ordering::Relaxed),
            1,
            "reset_stats should mark priority as dirty"
        );
    }

    // ──────────────────────────────────────────────
    // TunnelClient — timestamps
    // ──────────────────────────────────────────────

    #[test]
    fn recent_timestamp_works() {
        TunnelClient::update_clock();
        let recent = TunnelClient::recent_timestamp();
        let now = TunnelClient::current_timestamp();
        assert!(now.abs_diff(recent) <= 1);
    }

    #[test]
    fn current_timestamp_is_plausible() {
        let ts = TunnelClient::current_timestamp();
        // Should be after 2024-01-01 and before 2100-01-01
        assert!(ts > 1_704_067_200, "timestamp too small: {ts}");
        assert!(ts < 4_102_444_800, "timestamp too large: {ts}");
    }

    // ──────────────────────────────────────────────
    // TunnelClient — thread safety
    // ──────────────────────────────────────────────

    #[test]
    fn tunnel_client_is_send_and_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<TunnelClient>();
    }

    #[test]
    fn tunnel_client_concurrent_update_stats() {
        let client = Arc::new(TunnelClient::new(60));
        let threads: Vec<_> = (0..4)
            .map(|i| {
                let c = Arc::clone(&client);
                std::thread::spawn(move || {
                    let base = TunnelClient::current_timestamp() + i;
                    for j in 0..1000 {
                        c.update_stats(100, 50, base + j);
                    }
                })
            })
            .collect();

        for t in threads {
            t.join().unwrap();
        }

        let total_packets = client.packet_count.load(Ordering::Relaxed);
        assert_eq!(total_packets, 4000, "all packets should be counted");

        let total_in = client.bytes_received.load(Ordering::Relaxed);
        assert_eq!(total_in, 400_000, "all bytes_in should be counted");
    }

    // ──────────────────────────────────────────────
    // QualityAnalyzer
    // ──────────────────────────────────────────────

    #[test]
    fn quality_analyzer_no_loss() {
        let qa = QualityAnalyzer::new();
        for _ in 0..100 {
            qa.record_packet(false);
        }
        assert_eq!(qa.get_packet_loss_rate(), 0);
    }

    #[test]
    fn quality_analyzer_50_percent_loss() {
        let qa = QualityAnalyzer::new();
        for i in 0..100 {
            qa.record_packet(i % 2 == 0);
        }
        assert_eq!(qa.get_packet_loss_rate(), 500);
    }

    #[test]
    fn quality_analyzer_100_percent_loss() {
        let qa = QualityAnalyzer::new();
        for _ in 0..100 {
            qa.record_packet(true);
        }
        assert_eq!(qa.get_packet_loss_rate(), 1000);
    }

    #[test]
    fn quality_analyzer_empty_returns_zero() {
        let qa = QualityAnalyzer::new();
        assert_eq!(qa.get_packet_loss_rate(), 0);
    }

    #[test]
    fn quality_analyzer_halving() {
        let qa = QualityAnalyzer::new();
        for _ in 0..10_001 {
            qa.record_packet(false);
        }
        let total = qa.total_count.load(Ordering::Relaxed);
        assert!(total <= 5_001, "expected halving, got total={total}");
    }

    #[test]
    fn quality_analyzer_halving_preserves_rate() {
        let qa = QualityAnalyzer::new();
        // ~25% loss: every 4th packet is lost
        for i in 0..10_001 {
            qa.record_packet(i % 4 == 0);
        }
        let rate = qa.get_packet_loss_rate();
        // After halving, the ratio should remain approximately 250 (25%)
        // Allow some tolerance since halving is integer division
        assert!(
            (200..=300).contains(&rate),
            "loss rate should be ~250 after halving, got {rate}"
        );
    }

    #[test]
    fn quality_analyzer_default_trait() {
        let qa = QualityAnalyzer::default();
        assert_eq!(qa.get_packet_loss_rate(), 0);
        assert_eq!(qa.total_count.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn quality_analyzer_is_send_and_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<QualityAnalyzer>();
    }

    #[test]
    fn quality_analyzer_concurrent_recording() {
        let qa = Arc::new(QualityAnalyzer::new());
        let threads: Vec<_> = (0..4)
            .map(|_| {
                let q = Arc::clone(&qa);
                std::thread::spawn(move || {
                    for i in 0..2500 {
                        q.record_packet(i % 2 == 0);
                    }
                })
            })
            .collect();

        for t in threads {
            t.join().unwrap();
        }

        let rate = qa.get_packet_loss_rate();
        // 50% loss across all threads, allow tolerance due to halving
        assert!(
            (400..=600).contains(&rate),
            "concurrent 50% loss rate should be ~500, got {rate}"
        );
    }

    // ──────────────────────────────────────────────
    // validate_address — IPv4
    // ──────────────────────────────────────────────

    #[test]
    fn validate_address_rejects_loopback() {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1234);
        assert!(!validate_address(&addr));
    }

    #[test]
    fn validate_address_rejects_zero_port() {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 0);
        assert!(!validate_address(&addr));
    }

    #[test]
    fn validate_address_rejects_unspecified() {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 1234);
        assert!(!validate_address(&addr));
    }

    #[test]
    fn validate_address_rejects_broadcast() {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::BROADCAST), 1234);
        assert!(!validate_address(&addr));
    }

    #[test]
    fn validate_address_rejects_multicast_v4() {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(224, 0, 0, 1)), 1234);
        assert!(!validate_address(&addr));
    }

    #[test]
    fn validate_address_accepts_valid() {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 1234);
        assert!(validate_address(&addr));
    }

    #[test]
    fn validate_address_accepts_private_v4() {
        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
        assert!(validate_address(&addr));
    }

    // ──────────────────────────────────────────────
    // validate_address — IPv6
    // ──────────────────────────────────────────────

    #[test]
    fn validate_address_rejects_ipv6_loopback() {
        let addr = SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 1234);
        assert!(!validate_address(&addr));
    }

    #[test]
    fn validate_address_rejects_ipv6_unspecified() {
        let addr = SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 1234);
        assert!(!validate_address(&addr));
    }

    #[test]
    fn validate_address_rejects_ipv6_multicast() {
        // ff02::1 is a well-known multicast address
        let addr = SocketAddr::new(
            IpAddr::V6(Ipv6Addr::new(0xff02, 0, 0, 0, 0, 0, 0, 1)),
            1234,
        );
        assert!(!validate_address(&addr));
    }

    #[test]
    fn validate_address_accepts_valid_ipv6() {
        let addr = SocketAddr::new(
            IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)),
            443,
        );
        assert!(validate_address(&addr));
    }

    #[test]
    fn validate_address_rejects_ipv6_zero_port() {
        let addr = SocketAddr::new(
            IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)),
            0,
        );
        assert!(!validate_address(&addr));
    }

    // ──────────────────────────────────────────────
    // DashMap creation
    // ──────────────────────────────────────────────

    #[test]
    fn dashmap_creation() {
        let map: DashMap<u32, String> = create_dashmap_with_capacity(100);
        map.insert(1, "test".to_string());
        assert_eq!(map.len(), 1);
    }

    #[test]
    fn dashmap_creation_zero_capacity() {
        let map: DashMap<u32, u32> = create_dashmap_with_capacity(0);
        map.insert(1, 42);
        assert_eq!(*map.get(&1).unwrap(), 42);
    }
}