第 9 章：安全与性能：s2n-quic——AWS 的 Rust 工程标杆

// s2n-quic 的核心思想：一切皆 Provider

// Server 通过组合多个 Provider 来构建
let server = Server::builder()
    .with_tls(tls_provider)?          // TLS 提供者
    .with_io(io_provider)?            // I/O 提供者
    .with_event(event_provider)?      // 事件监听者
    .with_limits(limits_provider)?    // 限制策略
    .with_congestion_controller(cc)?  // 拥塞控制
    .start()?;

// 每个 Provider 都是一个 trait，可以替换

// TLS Provider（简化）
pub trait TlsProvider: 'static + Send {
    type Server: TlsSession;
    type Client: TlsSession;
    type Error: core::fmt::Display;
    
    fn start_server_session(&self, params: &TlsParams) 
        -> Result<Self::Server, Self::Error>;
    fn start_client_session(&self, params: &TlsParams) 
        -> Result<Self::Client, Self::Error>;
}

// I/O Provider（简化）
pub trait IoProvider: 'static + Send {
    type PathHandle: Copy;
    
    fn create_socket(&self, addr: SocketAddr) -> io::Result<UdpSocket>;
    fn send(&self, socket: &UdpSocket, transmit: &Transmit) -> io::Result<()>;
    fn recv(&self, socket: &UdpSocket, buf: &mut [u8]) -> io::Result<RecvMeta>;
}

// Congestion Controller Provider（简化）
pub trait CongestionController: 'static + Send {
    fn on_packet_sent(&mut self, bytes: usize, time: Timestamp);
    fn on_ack(&mut self, bytes: usize, rtt: Duration, time: Timestamp);
    fn on_loss(&mut self, bytes: usize, time: Timestamp);
    fn congestion_window(&self) -> usize;
    fn is_congestion_limited(&self) -> bool;
}

// Event Provider（简化）
pub trait EventProvider: 'static + Send {
    type Subscriber: EventSubscriber;
    
    fn create_subscriber(&self) -> Self::Subscriber;
}

// 生产环境的配置
let production_server = Server::builder()
    .with_tls(s2n_tls::Server::new(cert, key))?  // 真正的 TLS
    .with_io(tokio::Io::new(addr))?               // 真正的网络
    .with_event(tracing::Subscriber::new())?      // 真正的日志
    .start()?;

// 测试环境的配置
let test_server = Server::builder()
    .with_tls(testing::MockTls::new())?           // 模拟 TLS（立即完成）
    .with_io(testing::SimulatedIo::new())?        // 模拟网络（可控延迟/丢包）
    .with_event(testing::EventRecorder::new())?   // 记录所有事件（用于断言）
    .start()?;

// 同样的核心逻辑，不同的"外围设备"
// 测试可以精确控制网络条件：
let mut sim = testing::SimulatedIo::new();
sim.set_loss_rate(0.10);           // 10% 丢包
sim.set_delay(Duration::from_millis(100));  // 100ms 延迟
sim.inject_reorder(0.05);          // 5% 乱序
// 然后验证拥塞控制在这种条件下的行为

// 事件通过 proc macro 自动生成
// 定义一个事件结构
#[derive(Event)]
pub struct PacketSent {
    pub packet_header: PacketHeader,
    pub packet_len: usize,
    pub path: Path,
}

#[derive(Event)]  
pub struct PacketLost {
    pub packet_header: PacketHeader,
    pub bytes_lost: usize,
    pub is_congestion_event: bool,
}

// 使用时
self.event_subscriber.on_packet_sent(PacketSent {
    packet_header: header,
    packet_len: len,
    path: self.path,
});

// 实现 EventSubscriber 来消费事件
pub struct MetricsSubscriber {
    packets_sent: AtomicU64,
    packets_lost: AtomicU64,
    bytes_sent: AtomicU64,
}

impl EventSubscriber for MetricsSubscriber {
    fn on_packet_sent(&self, event: &PacketSent) {
        self.packets_sent.fetch_add(1, Ordering::Relaxed);
        self.bytes_sent.fetch_add(event.packet_len as u64, Ordering::Relaxed);
    }
    
    fn on_packet_lost(&self, event: &PacketLost) {
        self.packets_lost.fetch_add(1, Ordering::Relaxed);
    }
    
    // ... 其他事件处理
}

// 在生产中可以接入 Prometheus、CloudWatch 等

// 如果不需要事件，编译器会完全优化掉事件代码

// 空事件订阅者
pub struct NoopSubscriber;
impl EventSubscriber for NoopSubscriber {
    fn on_packet_sent(&self, _: &PacketSent) {} // 空函数
    fn on_packet_lost(&self, _: &PacketLost) {} // 空函数
    // 所有方法都是空的
}

// 编译器的单态化 + 内联 = 零开销
// 当使用 NoopSubscriber 时，所有事件调用都被优化掉
// 不会有任何运行时开销

// 这就是 Rust 的"零成本抽象"：
// 你不用的东西，不会有任何运行时代价

// s2n-quic 的 Path 抽象
pub struct Path {
    /// 本地地址
    local_address: SocketAddr,
    /// 远端地址
    remote_address: SocketAddr,
    /// 路径 MTU
    mtu: u16,
    /// 路径 RTT 估计
    rtt_estimator: RttEstimator,
    /// 路径的拥塞控制器（每路径独立）
    congestion_controller: Box<dyn CongestionController>,
    /// 路径验证状态
    validation_state: PathValidationState,
}

// 路径验证（Connection Migration 的基础）
enum PathValidationState {
    /// 未验证（新路径）
    Unknown,
    /// 正在验证（已发 PATH_CHALLENGE）
    Validating { challenge: [u8; 8] },
    /// 已验证（收到正确的 PATH_RESPONSE）
    Validated,
    /// 验证失败
    Failed,
}

// s2n-quic 的 CUBIC 实现（简化）
pub struct CubicCongestionController {
    /// 拥塞窗口（字节）
    congestion_window: u32,
    /// 慢启动阈值
    ssthresh: u32,
    /// 上次丢包时的窗口
    w_max: f64,
    /// CUBIC 的 C 常数
    cubic_c: f64,
    /// 丢包时的时间
    epoch_start: Option<Timestamp>,
    /// 慢启动中
    in_slow_start: bool,
}

impl CongestionController for CubicCongestionController {
    fn on_ack(&mut self, bytes_acked: usize, rtt: Duration, now: Timestamp) {
        if self.in_slow_start {
            // 慢启动：每 ACK 一个包，窗口增加一个 MSS
            self.congestion_window += bytes_acked as u32;
            if self.congestion_window >= self.ssthresh {
                self.in_slow_start = false;
                self.epoch_start = Some(now);
            }
        } else {
            // CUBIC 拥塞避免
            let t = now.duration_since(self.epoch_start.unwrap());
            let w_cubic = self.cubic_c * (t.as_secs_f64() - self.k()).powi(3) 
                         + self.w_max;
            // ... 窗口调整逻辑
        }
    }
    
    fn on_loss(&mut self, bytes_lost: usize, now: Timestamp) {
        // 丢包：记录当前窗口，减小窗口
        self.w_max = self.congestion_window as f64;
        self.congestion_window = (self.congestion_window as f64 * 0.7) as u32;
        self.ssthresh = self.congestion_window;
        self.epoch_start = Some(now);
    }
    
    fn congestion_window(&self) -> usize {
        self.congestion_window as usize
    }
}

// s2n-quic 的 BBR 状态机
pub struct BbrCongestionController {
    state: BbrState,
    /// 估计的瓶颈带宽
    btl_bw: Bandwidth,
    /// 估计的最小 RTT
    rt_prop: Duration,
    /// pacing 速率
    pacing_rate: Bandwidth,
    /// 拥塞窗口
    congestion_window: u32,
    /// 增益参数
    pacing_gain: f64,
    cwnd_gain: f64,
}

enum BbrState {
    /// 启动阶段：指数增长探测带宽
    Startup,
    /// 排空阶段：消耗启动积累的队列
    Drain,
    /// 稳态阶段：周期性探测带宽
    ProbeBW { cycle_index: u8 },
    /// 探测 RTT 阶段：降低窗口测量最小 RTT
    ProbeRTT,
}

// 属性测试：验证"不变量"始终成立
use proptest::prelude::*;

proptest! {
    // 性质：无论什么输入，拥塞窗口永远 >= 最小窗口
    #[test]
    fn congestion_window_never_below_minimum(
        events in prop::collection::vec(arbitrary_cc_event(), 0..1000)
    ) {
        let mut cc = CubicCongestionController::new();
        for event in events {
            match event {
                CcEvent::Ack { bytes, rtt } => cc.on_ack(bytes, rtt, now()),
                CcEvent::Loss { bytes } => cc.on_loss(bytes, now()),
            }
            // 不变量：窗口永远 >= 2 * MSS
            prop_assert!(cc.congestion_window() >= 2 * 1200);
        }
    }
    
    // 性质：ACK 永远不会减小窗口（在非丢包情况下）
    #[test]
    fn ack_never_decreases_window(
        acks in prop::collection::vec((1..10000usize, 1..500u64), 1..100)
    ) {
        let mut cc = CubicCongestionController::new();
        let mut prev_cwnd = cc.congestion_window();
        for (bytes, rtt_ms) in acks {
            cc.on_ack(bytes, Duration::from_millis(rtt_ms), now());
            prop_assert!(cc.congestion_window() >= prev_cwnd);
            prev_cwnd = cc.congestion_window();
        }
    }
}

// 最简单的 s2n-quic 服务端
use s2n_quic::Server;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut server = Server::builder()
        .with_tls((
            "cert.pem",   // 证书路径
            "key.pem",    // 密钥路径
        ))?
        .with_io("0.0.0.0:4433")?   // 监听地址
        .start()?;
    
    while let Some(mut connection) = server.accept().await {
        // 每个连接起一个任务
        tokio::spawn(async move {
            while let Ok(Some(mut stream)) = connection.accept_bidirectional_stream().await {
                tokio::spawn(async move {
                    // Echo：读什么写什么
                    while let Ok(Some(data)) = stream.receive().await {
                        stream.send(data).await.unwrap();
                    }
                });
            }
        });
    }
    
    Ok(())
}

// 最简单的 s2n-quic 客户端
use s2n_quic::Client;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::builder()
        .with_tls("ca-cert.pem")?   // CA 证书（验证服务端）
        .with_io("0.0.0.0:0")?     // 随机端口
        .start()?;
    
    let addr = "127.0.0.1:4433".parse()?;
    let mut connection = client.connect(addr).await?;
    
    let stream = connection.open_bidirectional_stream().await?;
    let (mut recv, mut send) = stream.split();
    
    send.send(bytes::Bytes::from("Hello QUIC!")).await?;
    send.finish()?;
    
    let response = recv.receive().await?;
    println!("Got: {:?}", response);
    
    Ok(())
}