freenet 0.2.43

//! Comprehensive tests for BBRv3 congestion control.
//!
//! This module contains integration tests for the BBR controller,
//! including state machine tests and network condition simulations.

use std::time::Duration;

use rstest::rstest;

use crate::simulation::VirtualTime;

use super::config::BbrConfig;
use super::controller::BbrController;
use super::state::{BbrState, ProbeBwPhase};

/// Network condition presets for testing (matching LEDBAT presets).
#[derive(Debug, Clone, Copy)]
pub struct NetworkCondition {
    /// Base RTT (propagation delay).
    pub rtt: Duration,
    /// Bandwidth in bytes/sec.
    pub bandwidth: u64,
    /// Packet loss rate (0.0 - 1.0).
    pub loss_rate: f64,
    /// RTT jitter as fraction of RTT (0.0 - 1.0).
    pub jitter: f64,
}

impl NetworkCondition {
    /// LAN conditions: 1ms RTT, 100 MB/s, no loss.
    pub const LAN: Self = Self {
        rtt: Duration::from_millis(1),
        bandwidth: 100_000_000,
        loss_rate: 0.0,
        jitter: 0.05,
    };

    /// Datacenter conditions: 10ms RTT, 10 MB/s, minimal loss.
    pub const DATACENTER: Self = Self {
        rtt: Duration::from_millis(10),
        bandwidth: 10_000_000,
        loss_rate: 0.001,
        jitter: 0.1,
    };

    /// Continental conditions: 50ms RTT, 5 MB/s, some loss.
    pub const CONTINENTAL: Self = Self {
        rtt: Duration::from_millis(50),
        bandwidth: 5_000_000,
        loss_rate: 0.005,
        jitter: 0.15,
    };

    /// Intercontinental conditions: 135ms RTT, 2 MB/s, moderate loss.
    pub const INTERCONTINENTAL: Self = Self {
        rtt: Duration::from_millis(135),
        bandwidth: 2_000_000,
        loss_rate: 0.01,
        jitter: 0.2,
    };

    /// High latency conditions: 250ms RTT, 1 MB/s, higher loss.
    pub const HIGH_LATENCY: Self = Self {
        rtt: Duration::from_millis(250),
        bandwidth: 1_000_000,
        loss_rate: 0.02,
        jitter: 0.25,
    };
}

/// Test harness for deterministic BBR testing.
pub struct BbrTestHarness {
    /// Virtual time source.
    pub time: VirtualTime,
    /// BBR controller under test.
    pub controller: BbrController<VirtualTime>,
    /// Network conditions.
    pub condition: NetworkCondition,
    /// RNG seed for reproducibility.
    seed: u64,
    /// Simple LCG state for deterministic randomness.
    rng_state: u64,
}

impl BbrTestHarness {
    /// Create a new test harness.
    pub fn new(config: BbrConfig, condition: NetworkCondition, seed: u64) -> Self {
        let time = VirtualTime::new();
        let controller = BbrController::new_with_time_source(config, time.clone());

        Self {
            time,
            controller,
            condition,
            seed,
            rng_state: seed,
        }
    }

    /// Simple LCG random number generator for determinism.
    fn random(&mut self) -> f64 {
        // LCG parameters from Numerical Recipes
        self.rng_state = self
            .rng_state
            .wrapping_mul(6364136223846793005)
            .wrapping_add(1);
        (self.rng_state >> 33) as f64 / (1u64 << 31) as f64
    }

    /// Simulate sending and receiving a burst of data.
    ///
    /// This simulates a more realistic scenario where multiple packets are
    /// sent before ACKs arrive (pipelining).
    ///
    /// Returns the number of bytes successfully delivered.
    pub fn transfer_bytes(&mut self, bytes: usize) -> usize {
        let packet_size = 1400; // Typical MTU payload
        let mut delivered = 0;
        let mut pending_tokens: Vec<(super::delivery_rate::DeliveryRateToken, usize, Duration)> =
            Vec::new();

        // Send a burst of packets up to cwnd
        let mut bytes_to_send = bytes;
        while bytes_to_send > 0 {
            let chunk = bytes_to_send.min(packet_size);

            // Check if we can send (cwnd allows)
            if self.controller.flightsize() + chunk > self.controller.current_cwnd() {
                // Can't send more, wait for some ACKs
                if let Some((token, size, rtt)) = pending_tokens.pop() {
                    self.time.advance(rtt);
                    if self.random() < self.condition.loss_rate {
                        self.controller.on_loss(size);
                    } else {
                        self.controller.on_ack_with_token(rtt, size, Some(token));
                        delivered += size;
                    }
                } else {
                    break;
                }
                continue;
            }

            // Send packet
            let token = self.controller.on_send(chunk);
            bytes_to_send -= chunk;

            // Calculate RTT with jitter
            let jitter_factor = 1.0 + (self.random() - 0.5) * 2.0 * self.condition.jitter;
            let rtt =
                Duration::from_nanos((self.condition.rtt.as_nanos() as f64 * jitter_factor) as u64);

            pending_tokens.push((token, chunk, rtt));
        }

        // Receive remaining ACKs
        for (token, size, rtt) in pending_tokens {
            self.time.advance(rtt);
            if self.random() < self.condition.loss_rate {
                self.controller.on_loss(size);
            } else {
                self.controller.on_ack_with_token(rtt, size, Some(token));
                delivered += size;
            }
        }

        delivered
    }

    /// Run for a number of RTTs.
    pub fn run_rtts(&mut self, count: usize, bytes_per_rtt: usize) -> Vec<BbrSnapshot> {
        let mut snapshots = Vec::with_capacity(count);

        for _ in 0..count {
            self.transfer_bytes(bytes_per_rtt);
            snapshots.push(self.snapshot());
        }

        snapshots
    }

    /// Take a snapshot of current state.
    pub fn snapshot(&self) -> BbrSnapshot {
        BbrSnapshot {
            state: self.controller.state(),
            probe_bw_phase: self.controller.stats().probe_bw_phase,
            cwnd: self.controller.current_cwnd(),
            flightsize: self.controller.flightsize(),
            max_bw: self.controller.max_bw(),
            min_rtt: self.controller.min_rtt(),
            bdp: self.controller.bdp(),
            pacing_rate: self.controller.pacing_rate(),
        }
    }

    /// Inject a timeout event.
    pub fn inject_timeout(&mut self) {
        self.controller.on_timeout();
    }
}

/// Snapshot of BBR state at a point in time.
#[derive(Debug, Clone)]
pub struct BbrSnapshot {
    pub state: BbrState,
    pub probe_bw_phase: ProbeBwPhase,
    pub cwnd: usize,
    pub flightsize: usize,
    pub max_bw: u64,
    pub min_rtt: Option<Duration>,
    pub bdp: usize,
    pub pacing_rate: u64,
}

// =============================================================================
// State Machine Tests
// =============================================================================

#[test]
fn test_startup_initial_state() {
    let harness = BbrTestHarness::new(BbrConfig::default(), NetworkCondition::CONTINENTAL, 12345);

    assert_eq!(harness.controller.state(), BbrState::Startup);
}

#[test]
fn test_startup_to_drain_transition() {
    let mut harness = BbrTestHarness::new(BbrConfig::default(), NetworkCondition::LAN, 12345);

    // Run enough RTTs to plateau bandwidth and exit Startup.
    // Use LAN conditions (1ms RTT, 100 MB/s) for fast bandwidth discovery.
    // With MIN_BW_FOR_STARTUP_EXIT = 1 MB/s, need enough traffic to measure
    // bandwidth above this threshold.
    let snapshots = harness.run_rtts(30, 1_000_000);

    // Should eventually exit Startup
    let final_state = snapshots.last().unwrap().state;
    assert!(
        final_state == BbrState::Drain || final_state == BbrState::ProbeBW,
        "Expected Drain or ProbeBW, got {:?}. max_bw: {} bytes/sec",
        final_state,
        snapshots.last().unwrap().max_bw
    );
}

#[test]
fn test_drain_to_probe_bw_transition() {
    let mut harness =
        BbrTestHarness::new(BbrConfig::default(), NetworkCondition::DATACENTER, 12345);

    // Run until we reach ProbeBW
    for _ in 0..50 {
        harness.transfer_bytes(50_000);
        if harness.controller.state() == BbrState::ProbeBW {
            break;
        }
    }

    // Should eventually reach ProbeBW
    // (might still be in Startup or Drain depending on conditions)
    let state = harness.controller.state();
    assert!(
        state == BbrState::Startup || state == BbrState::Drain || state == BbrState::ProbeBW,
        "Unexpected state: {:?}",
        state
    );
}

// =============================================================================
// Network Condition Tests
// =============================================================================

#[test]
fn test_lan_conditions() {
    let mut harness = BbrTestHarness::new(BbrConfig::default(), NetworkCondition::LAN, 12345);

    let snapshots = harness.run_rtts(10, 100_000);

    // Should have low RTT
    let last = snapshots.last().unwrap();
    if let Some(rtt) = last.min_rtt {
        assert!(rtt < Duration::from_millis(10));
    }
}

#[test]
fn test_intercontinental_conditions() {
    let mut harness = BbrTestHarness::new(
        BbrConfig::default(),
        NetworkCondition::INTERCONTINENTAL,
        12345,
    );

    let snapshots = harness.run_rtts(20, 50_000);

    // Should have higher RTT
    let last = snapshots.last().unwrap();
    if let Some(rtt) = last.min_rtt {
        assert!(rtt > Duration::from_millis(50));
    }
}

#[test]
fn test_high_latency_no_death_spiral() {
    let mut harness =
        BbrTestHarness::new(BbrConfig::default(), NetworkCondition::HIGH_LATENCY, 12345);

    // Run for many RTTs
    let snapshots = harness.run_rtts(50, 20_000);

    // cwnd should never collapse to minimum
    let min_cwnd = snapshots.iter().map(|s| s.cwnd).min().unwrap();
    assert!(
        min_cwnd >= BbrConfig::default().min_cwnd,
        "cwnd collapsed to {} (min_cwnd={})",
        min_cwnd,
        BbrConfig::default().min_cwnd
    );
}

// =============================================================================
// Timeout Recovery Tests
// =============================================================================

#[test]
fn test_timeout_recovery() {
    let mut harness =
        BbrTestHarness::new(BbrConfig::default(), NetworkCondition::CONTINENTAL, 12345);

    // Run some RTTs to establish state
    harness.run_rtts(10, 50_000);

    // Inject timeout
    harness.inject_timeout();

    // Should reset to Startup
    assert_eq!(harness.controller.state(), BbrState::Startup);
    // cwnd should be reset to initial value
    assert_eq!(
        harness.controller.current_cwnd(),
        BbrConfig::default().initial_cwnd
    );

    // Should be able to recover - run more RTTs
    let snapshots = harness.run_rtts(20, 50_000);

    // After recovery, cwnd should be at least min_cwnd and reasonable for the network
    // (BBR will settle to a cwnd based on measured BDP, which may be lower than initial_cwnd)
    let final_cwnd = snapshots.last().unwrap().cwnd;
    assert!(
        final_cwnd >= BbrConfig::default().min_cwnd,
        "cwnd collapsed below min_cwnd after recovery: {}",
        final_cwnd
    );

    // Should have transitioned out of Startup after recovery
    let final_state = snapshots.last().unwrap().state;
    assert!(
        final_state != BbrState::Startup || final_cwnd >= BbrConfig::default().min_cwnd,
        "Should have recovered to a stable state"
    );
}

// =============================================================================
// Loss Response Tests
// =============================================================================

#[test]
fn test_loss_response_reduces_inflight_hi() {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    // Send some packets
    for _ in 0..10 {
        controller.on_send(1000);
    }

    // Simulate loss
    controller.on_loss(1000);

    // inflight_hi should be reduced
    // (exact value depends on current inflight)
    let stats = controller.stats();
    assert!(stats.lost > 0);
}

// =============================================================================
// BDP Calculation Tests
// =============================================================================

#[test]
fn test_bdp_scales_with_bandwidth() {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    // Simulate traffic at known rate
    for _i in 0..20 {
        let token = controller.on_send(10000);
        time.advance(Duration::from_millis(50));
        controller.on_ack_with_token(Duration::from_millis(50), 10000, Some(token));
    }

    // BDP should reflect the measured bandwidth * RTT
    let bdp = controller.bdp();
    let min_rtt = controller.min_rtt();
    let max_bw = controller.max_bw();

    if let Some(rtt) = min_rtt {
        if max_bw > 0 {
            let expected_bdp = (max_bw as u128 * rtt.as_nanos() / 1_000_000_000) as usize;
            // Allow some tolerance
            assert!(
                bdp >= expected_bdp / 2 && bdp <= expected_bdp * 2,
                "BDP {} not close to expected {}",
                bdp,
                expected_bdp
            );
        }
    }
}

// =============================================================================
// Regression: Timeout Storm (Issue from v0.1.92)
// =============================================================================

/// Regression test: BBR cannot recover between timeouts in high-latency conditions.
///
/// In production (v0.1.92), we observed 935 timeouts in 10 seconds when transferring
/// to high-latency peers. The root cause was:
///
/// 1. MIN_RTO was 200ms, but real RTT (144ms) + ACK_CHECK_INTERVAL (100ms) = 244ms
/// 2. This caused spurious timeouts even when packets were being delivered
/// 3. Each timeout resets ALL BBR state (cwnd, bandwidth estimates, min_rtt)
/// 4. BBR could never build up measurements before the next timeout
///
/// Fix:
/// 1. MIN_RTO was increased to 500ms to account for ACK batching delay
/// 2. BBR now uses adaptive timeout floor based on max BDP seen
///
/// This test verifies BBR's timeout behavior with minimal traffic (no high BDP).
#[test]
fn test_timeout_storm_prevents_recovery() {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let initial_cwnd = BbrConfig::default().initial_cwnd;
    println!("Initial cwnd: {}", initial_cwnd);

    // Simulate 10 rounds of: send packets -> partial recovery -> timeout
    // With minimal traffic, max_bdp_seen stays low, so adaptive floor = initial_cwnd
    let mut max_cwnd_achieved = initial_cwnd;

    println!("\nSimulating timeout storm with recovery attempts (minimal traffic):");
    for round in 0..10 {
        // Try to recover: send packets and get ACKs
        for _ in 0..5 {
            let token = controller.on_send(1400);
            time.advance(Duration::from_millis(50));
            controller.on_ack_with_token(Duration::from_millis(100), 1400, Some(token));
        }

        let cwnd_before_timeout = controller.current_cwnd();
        max_cwnd_achieved = max_cwnd_achieved.max(cwnd_before_timeout);

        // Timeout hits!
        controller.on_timeout();

        let cwnd_after_timeout = controller.current_cwnd();
        println!(
            "  Round {}: cwnd {} -> {} (max_bdp_seen={})",
            round + 1,
            cwnd_before_timeout,
            cwnd_after_timeout,
            controller.max_bdp_seen()
        );

        // With low BDP measurements, adaptive floor should be at least initial_cwnd
        assert!(
            cwnd_after_timeout >= initial_cwnd,
            "BBR cwnd ({}) should not drop below initial_cwnd ({}) after timeout",
            cwnd_after_timeout,
            initial_cwnd
        );
    }

    let final_cwnd = controller.current_cwnd();
    println!(
        "\nFinal cwnd: {} (initial was {}, max_bdp_seen={})",
        final_cwnd,
        initial_cwnd,
        controller.max_bdp_seen()
    );
    println!(
        "Max cwnd achieved during recovery attempts: {}",
        max_cwnd_achieved
    );
}

/// Test that BBR's adaptive timeout floor kicks in with high BDP.
///
/// When max_bdp_seen is high enough (>4x initial_cwnd), the adaptive floor
/// should prevent cwnd from collapsing to initial_cwnd on timeout.
#[test]
fn test_bbr_adaptive_timeout_floor_with_high_bdp() {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let initial_cwnd = BbrConfig::default().initial_cwnd;
    println!("Initial cwnd: {}", initial_cwnd);

    // Build up high BDP measurements by simulating high-bandwidth traffic
    // We need max_bdp_seen > 4 * initial_cwnd for adaptive floor to kick in
    println!("\nBuilding up high BDP measurements...");
    for _ in 0..100 {
        // Send a large burst
        for _ in 0..50 {
            controller.on_send(1400);
        }
        time.advance(Duration::from_millis(20)); // 20ms RTT = high bandwidth
        for _ in 0..50 {
            let token = controller.on_send(1400);
            controller.on_ack_with_token(Duration::from_millis(20), 1400, Some(token));
        }
    }

    let max_bdp = controller.max_bdp_seen();
    let cwnd_before = controller.current_cwnd();
    println!(
        "After warmup: cwnd={}, max_bdp_seen={}",
        cwnd_before, max_bdp
    );

    // Trigger timeout
    controller.on_timeout();

    let cwnd_after = controller.current_cwnd();
    println!(
        "After timeout: cwnd={} (initial={}, adaptive_floor=max_bdp/4={})",
        cwnd_after,
        initial_cwnd,
        max_bdp / 4
    );

    // If max_bdp > 4 * initial_cwnd, adaptive floor should kick in
    if max_bdp > 4 * initial_cwnd {
        let expected_floor = max_bdp / 4;
        assert!(
            cwnd_after >= expected_floor,
            "BBR cwnd ({}) should use adaptive floor ({}) when max_bdp ({}) > 4*initial ({})",
            cwnd_after,
            expected_floor,
            max_bdp,
            4 * initial_cwnd
        );
        println!(
            "SUCCESS: Adaptive floor kicked in - cwnd {} >= floor {}",
            cwnd_after, expected_floor
        );
    } else {
        // max_bdp not high enough, should use initial_cwnd
        assert_eq!(
            cwnd_after, initial_cwnd,
            "BBR should use initial_cwnd when max_bdp < 4*initial"
        );
        println!(
            "Note: max_bdp ({}) < 4*initial_cwnd ({}), using initial_cwnd",
            max_bdp,
            4 * initial_cwnd
        );
    }
}

// =============================================================================
// Regression: Issue #2682/#2697 - BBR Death Spiral on High-Latency Links
// =============================================================================

/// Regression test for issue #2697: BBR timeout should preserve bandwidth/RTT estimates.
///
/// On high-latency links (125-500ms RTT), timeouts were causing `on_timeout()` to reset
/// both `bw_filter` and `rtt_tracker`, which cleared all learned state. This caused:
///
/// 1. `max_bw()` returns 0 (bandwidth filter cleared)
/// 2. `min_rtt()` returns None (RTT tracker reset to u64::MAX)
/// 3. BDP calculation falls back to initial_cwnd (14KB)
/// 4. With high RTT and 14KB cwnd → severely limited throughput
/// 5. Slow recovery triggers more timeouts → death spiral
///
/// The fix (PR #2699) preserves bandwidth and RTT estimates across timeouts since they
/// represent physical path characteristics that don't change due to timeout.
///
/// Parameters:
/// - `rtt_ms`: RTT in milliseconds (tests intercontinental to satellite links)
/// - `warmup_rounds`: Number of RTTs to establish bandwidth measurements
/// - `packets_per_round`: Packets sent per RTT during warmup
#[rstest]
#[case::intercontinental_minimal(125, 30, 10)]
#[case::intercontinental_standard(135, 50, 20)]
#[case::high_latency_standard(250, 50, 20)]
#[case::high_latency_extended(250, 100, 30)]
#[case::satellite_link(500, 50, 20)]
fn test_issue_2697_timeout_preserves_bandwidth_and_rtt(
    #[case] rtt_ms: u64,
    #[case] warmup_rounds: usize,
    #[case] packets_per_round: usize,
) {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let rtt = Duration::from_millis(rtt_ms);
    let packet_size = 1400;

    // Phase 1: Establish good bandwidth/RTT measurements
    for _ in 0..warmup_rounds {
        let mut tokens = Vec::new();
        for _ in 0..packets_per_round {
            tokens.push(controller.on_send(packet_size));
        }
        time.advance(rtt);
        for token in tokens {
            controller.on_ack_with_token(rtt, packet_size, Some(token));
        }
    }

    let pre_timeout_max_bw = controller.max_bw();
    let pre_timeout_min_rtt = controller.min_rtt();

    // Verify we have good measurements before timeout
    assert!(
        pre_timeout_max_bw > 0,
        "[{}ms RTT] Should have bandwidth measurement before timeout",
        rtt_ms
    );
    assert!(
        pre_timeout_min_rtt.is_some(),
        "[{}ms RTT] Should have RTT measurement before timeout",
        rtt_ms
    );

    // Phase 2: Trigger timeout (simulates spurious timeout on high-latency link)
    controller.on_timeout();

    let post_timeout_max_bw = controller.max_bw();
    let post_timeout_min_rtt = controller.min_rtt();

    // CRITICAL: Bandwidth and RTT must be preserved across timeout
    // These represent physical path characteristics that don't change
    assert!(
        post_timeout_max_bw > 0,
        "[{}ms RTT] Bandwidth estimate must be preserved across timeout (was {}, now {})",
        rtt_ms,
        pre_timeout_max_bw,
        post_timeout_max_bw
    );
    assert!(
        post_timeout_min_rtt.is_some(),
        "[{}ms RTT] RTT estimate must be preserved across timeout (was {:?}, now {:?})",
        rtt_ms,
        pre_timeout_min_rtt,
        post_timeout_min_rtt
    );

    // Verify reasonable preservation (allow some reduction but not total reset)
    let bw_retention = post_timeout_max_bw as f64 / pre_timeout_max_bw as f64;
    assert!(
        bw_retention >= 0.25,
        "[{}ms RTT] Bandwidth should retain at least 25% after timeout (retained {:.1}%)",
        rtt_ms,
        bw_retention * 100.0
    );
}

/// Regression test for issue #2697: Repeated timeouts must not wipe bandwidth estimates.
///
/// This simulates a realistic scenario where:
/// 1. Connection operates on high-latency link
/// 2. ACK batching + jitter causes spurious timeouts
/// 3. Each timeout should NOT wipe bandwidth state to 0
/// 4. Bandwidth estimate should be preserved across each timeout
///
/// Parameters:
/// - `rtt_ms`: RTT in milliseconds
/// - `timeout_count`: Number of consecutive timeouts to simulate
#[rstest]
#[case::intercontinental_3_timeouts(135, 3)]
#[case::high_latency_5_timeouts(250, 5)]
#[case::high_latency_10_timeouts(250, 10)]
#[case::satellite_5_timeouts(500, 5)]
fn test_issue_2697_repeated_timeouts_preserve_bandwidth(
    #[case] rtt_ms: u64,
    #[case] timeout_count: usize,
) {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let rtt = Duration::from_millis(rtt_ms);
    let packet_size = 1400;

    // Phase 1: Build up good state with sufficient traffic
    for _ in 0..50 {
        let mut tokens = Vec::new();
        for _ in 0..20 {
            tokens.push(controller.on_send(packet_size));
        }
        time.advance(rtt);
        for token in tokens {
            controller.on_ack_with_token(rtt, packet_size, Some(token));
        }
    }

    let healthy_max_bw = controller.max_bw();
    assert!(
        healthy_max_bw > 0,
        "[{}ms RTT] Should have healthy bandwidth before storm",
        rtt_ms
    );

    // Phase 2: Simulate timeout storm - check bandwidth IMMEDIATELY after each timeout
    for i in 0..timeout_count {
        controller.on_timeout();

        // CRITICAL: Check bandwidth immediately after timeout (before any recovery)
        let post_timeout_bw = controller.max_bw();
        assert!(
            post_timeout_bw > 0,
            "[{}ms RTT] Timeout #{}: Bandwidth must NOT reset to 0 (was {}, now {})",
            rtt_ms,
            i + 1,
            healthy_max_bw,
            post_timeout_bw
        );

        // Brief recovery attempt (2 RTTs worth of traffic)
        for _ in 0..2 {
            let mut tokens = Vec::new();
            for _ in 0..5 {
                tokens.push(controller.on_send(packet_size));
            }
            time.advance(rtt);
            for token in tokens {
                controller.on_ack_with_token(rtt, packet_size, Some(token));
            }
        }
    }
}

/// Regression test for issue #2697: Large transfers must complete after timeout recovery.
///
/// Simulates a realistic large transfer (like the 2.9MB River web interface) on a
/// high-latency link, with timeouts occurring during the transfer. The transfer
/// should still complete at reasonable throughput, not degrade to ~1 KB/s.
///
/// Parameters:
/// - `rtt_ms`: RTT in milliseconds
/// - `transfer_kb`: Transfer size in KB
#[rstest]
#[case::river_ui_intercontinental(135, 2900)]
#[case::river_ui_high_latency(250, 2900)]
#[case::large_contract_satellite(500, 1024)]
fn test_issue_2697_large_transfer_with_timeouts(#[case] rtt_ms: u64, #[case] transfer_kb: usize) {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let rtt = Duration::from_millis(rtt_ms);
    let packet_size = 1400;
    let transfer_bytes = transfer_kb * 1024;

    // Phase 1: Establish connection with good bandwidth measurements
    for _ in 0..30 {
        let mut tokens = Vec::new();
        for _ in 0..20 {
            tokens.push(controller.on_send(packet_size));
        }
        time.advance(rtt);
        for token in tokens {
            controller.on_ack_with_token(rtt, packet_size, Some(token));
        }
    }

    let healthy_max_bw = controller.max_bw();
    assert!(
        healthy_max_bw > 0,
        "Should establish bandwidth before transfer"
    );

    // Phase 2: Simulate transfer with periodic timeouts (every ~500KB)
    let mut bytes_transferred = 0usize;
    let timeout_interval = 500 * 1024; // Timeout every 500KB
    let mut next_timeout_at = timeout_interval;

    while bytes_transferred < transfer_bytes {
        // Send a burst
        let burst_size = 20;
        let mut tokens = Vec::new();
        for _ in 0..burst_size {
            tokens.push(controller.on_send(packet_size));
        }

        time.advance(rtt);

        // Receive ACKs
        for token in tokens {
            controller.on_ack_with_token(rtt, packet_size, Some(token));
            bytes_transferred += packet_size;
        }

        // Inject timeout at intervals
        if bytes_transferred >= next_timeout_at {
            controller.on_timeout();

            // CRITICAL: Bandwidth must not collapse to 0
            let post_timeout_bw = controller.max_bw();
            assert!(
                post_timeout_bw > 0,
                "[{}ms RTT, {}KB transfer] Bandwidth collapsed to 0 at {}KB transferred",
                rtt_ms,
                transfer_kb,
                bytes_transferred / 1024
            );

            next_timeout_at += timeout_interval;
        }
    }

    // Phase 3: Verify final throughput is reasonable
    let final_bw = controller.max_bw();
    let final_bw_kbps = final_bw as f64 / 1024.0;

    // Should maintain at least 10 KB/s (much better than the ~1 KB/s bug)
    assert!(
        final_bw_kbps >= 10.0,
        "[{}ms RTT, {}KB transfer] Final bandwidth {:.1} KB/s is too low (bug caused ~1 KB/s)",
        rtt_ms,
        transfer_kb,
        final_bw_kbps
    );
}

// =============================================================================
// Realistic Throughput Tests - Respect pacing_rate like real transport does
// =============================================================================

/// Helper to simulate realistic transfer respecting pacing_rate.
///
/// Unlike the basic tests that blast packets, this simulates what the real
/// transport does: send only as many bytes as pacing_rate allows per time unit.
fn simulate_paced_transfer(
    controller: &BbrController<VirtualTime>,
    time: &VirtualTime,
    rtt: Duration,
    transfer_bytes: usize,
    packet_size: usize,
) -> (usize, Duration) {
    use crate::simulation::TimeSource;

    let mut bytes_sent = 0usize;
    let start_time = time.now_nanos();

    while bytes_sent < transfer_bytes {
        // Check pacing rate - how many bytes can we send this RTT?
        let pacing_rate = controller.pacing_rate();
        let bytes_allowed_by_pacing = (pacing_rate as f64 * rtt.as_secs_f64()) as usize;

        // Check cwnd - don't exceed congestion window
        let cwnd = controller.current_cwnd();
        let flightsize = controller.flightsize();
        let bytes_allowed_by_cwnd = cwnd.saturating_sub(flightsize);

        // Send the minimum of what pacing and cwnd allow
        let bytes_to_send = bytes_allowed_by_pacing
            .min(bytes_allowed_by_cwnd)
            .min(transfer_bytes - bytes_sent)
            .max(packet_size); // Send at least one packet

        let packets_to_send = (bytes_to_send / packet_size).max(1);

        let mut tokens = Vec::new();
        for _ in 0..packets_to_send {
            tokens.push(controller.on_send(packet_size));
        }

        time.advance(rtt);

        for token in tokens {
            controller.on_ack_with_token(rtt, packet_size, Some(token));
            bytes_sent += packet_size;
        }
    }

    let elapsed = Duration::from_nanos(time.now_nanos() - start_time);
    (bytes_sent, elapsed)
}

/// Test that BBR achieves reasonable throughput on a fresh connection.
///
/// This test documents the ACTUAL behavior when respecting pacing_rate.
/// The current BBR implementation is slow to ramp up, which explains
/// Ian's observation of 12 KB/s transfers.
///
/// Expected vs actual (with pacing respected):
/// - 50ms RTT: ~400 KB/s actual (want 1+ MB/s)
/// - 135ms RTT: ~16 KB/s actual (want 500+ KB/s) <- matches Ian's 12 KB/s!
/// - 250ms RTT: ~80 KB/s actual (want 250+ KB/s)
#[rstest]
#[case::domestic_broadband(50, 1_000_000)] // 50ms RTT should achieve 1+ MB/s
#[case::intercontinental(135, 500_000)] // 135ms RTT should achieve 500+ KB/s
#[case::high_latency(250, 250_000)] // 250ms RTT should achieve 250+ KB/s
fn test_fresh_connection_ramps_up_throughput(
    #[case] rtt_ms: u64,
    #[case] min_expected_throughput: usize,
) {
    use crate::simulation::TimeSource;

    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let rtt = Duration::from_millis(rtt_ms);
    let packet_size = 1400;
    let transfer_size = 500 * 1024; // 500KB transfer

    // Print initial state for debugging
    println!("\n[{}ms RTT] Initial state:", rtt_ms);
    println!(
        "  pacing_rate: {} B/s ({:.1} KB/s)",
        controller.pacing_rate(),
        controller.pacing_rate() as f64 / 1024.0
    );
    println!("  cwnd: {} bytes", controller.current_cwnd());

    // Simulate paced transfer
    let (bytes_sent, elapsed) =
        simulate_paced_transfer(&controller, &time, rtt, transfer_size, packet_size);

    let throughput = bytes_sent as f64 / elapsed.as_secs_f64();
    let throughput_kbps = throughput / 1024.0;

    // Print final state
    println!("[{}ms RTT] Final state:", rtt_ms);
    println!(
        "  pacing_rate: {} B/s ({:.1} KB/s)",
        controller.pacing_rate(),
        controller.pacing_rate() as f64 / 1024.0
    );
    println!("  cwnd: {} bytes", controller.current_cwnd());
    println!(
        "  max_bw: {} B/s ({:.1} KB/s)",
        controller.max_bw(),
        controller.max_bw() as f64 / 1024.0
    );
    println!("  throughput: {:.1} KB/s", throughput_kbps);

    // These thresholds represent expected correct behavior
    // Test is #[ignore]d until BBR bootstrap/ramp-up is fixed
    assert!(
        throughput >= min_expected_throughput as f64,
        "[{}ms RTT] Fresh connection throughput {:.1} KB/s is below expected {} KB/s - BBR bootstrap problem",
        rtt_ms,
        throughput_kbps,
        min_expected_throughput / 1024
    );
}

/// Test that timeout on fresh connection doesn't cause permanent slowdown.
///
/// This simulates the problematic scenario:
/// 1. Fresh connection starts transfer
/// 2. Timeout occurs before building up max_bdp_seen
/// 3. Connection should still recover to reasonable throughput
///
/// Current behavior: early timeout significantly hurts throughput because
/// max_bdp_seen is 0, so cwnd falls back to initial_cwnd.
#[rstest]
#[case::early_timeout_domestic(50, 500_000)] // 50ms RTT should recover to 500+ KB/s
#[case::early_timeout_intercontinental(135, 250_000)] // 135ms RTT should recover to 250+ KB/s
#[case::early_timeout_high_latency(250, 125_000)] // 250ms RTT should recover to 125+ KB/s
fn test_early_timeout_recovers_throughput(
    #[case] rtt_ms: u64,
    #[case] min_expected_throughput: usize,
) {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let rtt = Duration::from_millis(rtt_ms);
    let packet_size = 1400;

    println!("\n[{}ms RTT] Testing early timeout recovery", rtt_ms);
    println!("  Initial pacing_rate: {} B/s", controller.pacing_rate());
    println!("  Initial cwnd: {} bytes", controller.current_cwnd());

    // Phase 1: Send just a few packets (fresh connection)
    for _ in 0..5 {
        let token = controller.on_send(packet_size);
        time.advance(rtt);
        controller.on_ack_with_token(rtt, packet_size, Some(token));
    }

    println!("  After 5 packets:");
    println!("    pacing_rate: {} B/s", controller.pacing_rate());
    println!("    cwnd: {} bytes", controller.current_cwnd());
    println!("    max_bw: {} B/s", controller.max_bw());

    // Phase 2: EARLY TIMEOUT (before max_bdp_seen is built up)
    controller.on_timeout();

    let post_timeout_cwnd = controller.current_cwnd();
    let post_timeout_pacing = controller.pacing_rate();

    println!("  After early timeout:");
    println!("    pacing_rate: {} B/s", post_timeout_pacing);
    println!("    cwnd: {} bytes", post_timeout_cwnd);
    println!("    max_bw: {} B/s", controller.max_bw());

    // Phase 3: Continue transfer and measure throughput
    let transfer_size = 500 * 1024; // 500KB
    let (bytes_sent, elapsed) =
        simulate_paced_transfer(&controller, &time, rtt, transfer_size, packet_size);

    let throughput = bytes_sent as f64 / elapsed.as_secs_f64();
    let throughput_kbps = throughput / 1024.0;

    println!("  Final throughput: {:.1} KB/s", throughput_kbps);

    // Key assertion: even after early timeout, should have some throughput
    assert!(
        throughput >= min_expected_throughput as f64,
        "[{}ms RTT] Post-early-timeout throughput {:.1} KB/s is below {} KB/s. \
         Post-timeout state: cwnd={}B, pacing={}B/s",
        rtt_ms,
        throughput_kbps,
        min_expected_throughput / 1024,
        post_timeout_cwnd,
        post_timeout_pacing
    );
}

/// Test that pacing_rate is at least 1 MB/s for fresh connections.
///
/// The real transport uses pacing_rate to throttle sends. If pacing_rate
/// is too low, the connection will be artificially slow regardless of cwnd.
#[test]
fn test_initial_pacing_rate_is_reasonable() {
    let controller = BbrController::new(BbrConfig::default());

    let initial_pacing = controller.pacing_rate();

    // Initial pacing rate should be at least 1 MB/s
    // (set in controller.rs as INITIAL_PACING_RATE = 1_000_000)
    assert!(
        initial_pacing >= 1_000_000,
        "Initial pacing rate {} B/s is below 1 MB/s minimum",
        initial_pacing
    );
}

/// Test that pacing_rate grows during normal operation.
///
/// After successful transfers, pacing_rate should increase as BBR
/// discovers the available bandwidth.
#[test]
fn test_pacing_rate_grows_with_successful_transfers() {
    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let rtt = Duration::from_millis(50);
    let packet_size = 1400;

    let initial_pacing = controller.pacing_rate();

    // Send traffic for 20 RTTs
    for _ in 0..20 {
        let cwnd = controller.current_cwnd();
        let packets = (cwnd / packet_size).max(1);

        let mut tokens = Vec::new();
        for _ in 0..packets {
            tokens.push(controller.on_send(packet_size));
        }

        time.advance(rtt);

        for token in tokens {
            controller.on_ack_with_token(rtt, packet_size, Some(token));
        }
    }

    let final_pacing = controller.pacing_rate();

    // Pacing rate should have increased significantly
    assert!(
        final_pacing > initial_pacing * 10,
        "Pacing rate didn't grow enough: {} -> {} (expected 10x+ increase)",
        initial_pacing,
        final_pacing
    );
}

/// Test that longer transfers achieve higher throughput as BBR ramps up.
///
/// This validates that BBR's bandwidth discovery works correctly over time.
/// The first portion of a transfer may be slow while BBR probes, but
/// throughput should improve significantly as the connection matures.
#[rstest]
#[case::domestic_50ms(50, 5_000_000)] // 5 MB/s expected for 50ms RTT
#[case::intercontinental_135ms(135, 2_000_000)] // 2 MB/s expected for 135ms RTT
fn test_longer_transfer_achieves_higher_throughput(
    #[case] rtt_ms: u64,
    #[case] min_expected_throughput: usize,
) {
    use crate::simulation::TimeSource;

    let time = VirtualTime::new();
    let controller = BbrController::new_with_time_source(BbrConfig::default(), time.clone());

    let rtt = Duration::from_millis(rtt_ms);
    let packet_size = 1400;
    let total_transfer = 5 * 1024 * 1024; // 5 MB transfer

    println!(
        "\n[{}ms RTT] Testing 5MB transfer throughput ramp-up",
        rtt_ms
    );
    println!(
        "  Initial pacing_rate: {} B/s ({:.1} KB/s)",
        controller.pacing_rate(),
        controller.pacing_rate() as f64 / 1024.0
    );
    println!("  Initial cwnd: {} bytes", controller.current_cwnd());

    // Transfer in chunks, measuring throughput at each stage
    let chunk_size = total_transfer / 5;
    let mut total_bytes_sent = 0usize;
    let start_time = time.now_nanos();
    let mut stage_throughputs = Vec::new();

    for stage in 0..5 {
        println!(
            "  [Before Stage {}] cwnd: {}KB, pacing: {:.1} KB/s, max_bw: {:.1} KB/s, state: {:?}",
            stage + 1,
            controller.current_cwnd() / 1024,
            controller.pacing_rate() as f64 / 1024.0,
            controller.max_bw() as f64 / 1024.0,
            controller.state()
        );

        let stage_start = time.now_nanos();
        let (bytes, _) = simulate_paced_transfer(&controller, &time, rtt, chunk_size, packet_size);
        let stage_elapsed = Duration::from_nanos(time.now_nanos() - stage_start);
        let stage_throughput = bytes as f64 / stage_elapsed.as_secs_f64();
        total_bytes_sent += bytes;

        stage_throughputs.push(stage_throughput);
        println!(
            "  [After Stage {}] {:.1} KB/s (cwnd: {}KB, pacing: {:.1} KB/s, max_bw: {:.1} KB/s)",
            stage + 1,
            stage_throughput / 1024.0,
            controller.current_cwnd() / 1024,
            controller.pacing_rate() as f64 / 1024.0,
            controller.max_bw() as f64 / 1024.0
        );
    }

    let total_elapsed = Duration::from_nanos(time.now_nanos() - start_time);
    let overall_throughput = total_bytes_sent as f64 / total_elapsed.as_secs_f64();

    println!(
        "  Overall throughput: {:.1} KB/s ({:.2} MB/s)",
        overall_throughput / 1024.0,
        overall_throughput / (1024.0 * 1024.0)
    );
    println!(
        "  Final state: {:?}, max_bw: {:.1} KB/s",
        controller.state(),
        controller.max_bw() as f64 / 1024.0
    );

    // Throughput should be sustained at reasonable levels across all stages.
    // Note: Stage 1 is artificially boosted by startup_min_pacing_rate, so later
    // stages may have slightly lower throughput than Stage 1. The key metric is
    // that throughput doesn't COLLAPSE (which would indicate the bootstrap
    // death spiral bug).
    let last_stage = stage_throughputs[4];
    let min_sustained = min_expected_throughput as f64 * 0.5; // At least 50% of target per stage
    assert!(
        last_stage >= min_sustained,
        "[{}ms RTT] Throughput collapsed: last stage {:.1} KB/s is below {:.1} KB/s",
        rtt_ms,
        last_stage / 1024.0,
        min_sustained / 1024.0
    );

    // Overall throughput should meet minimum expectation
    assert!(
        overall_throughput >= min_expected_throughput as f64,
        "[{}ms RTT] Overall throughput {:.1} KB/s ({:.2} MB/s) is below expected {} KB/s",
        rtt_ms,
        overall_throughput / 1024.0,
        overall_throughput / (1024.0 * 1024.0),
        min_expected_throughput / 1024
    );
}