zenith-web 0.1.0

//! AF_XDP → HTTP/3 数据面桥接（仅 Linux）
//!
//! 打通 XDP 主数据面到 L7 的完整闭环：
//!
//! ```text
//! NIC → XDP → AF_XDP Worker（L2-L4 解析 + 来源准入）
//!   → UdpDatagram（载荷 + 回包上下文）
//!   → ProtocolServer::serve_udp_quic_datagram（QUIC/H3 全栈 + 安全管道 + App）
//!   → 响应载荷 → UdpDatagram::build_reply_frame（L2-L4 回包构造）
//!   → Worker::queue_tx_data → TX Ring → NIC
//! ```
//!
//! # 铁则 1 合规
//! 全程零内核 socket 旁路：QUIC 服务器须以
//! [`ProtocolServer::bind_quic_socketless`] 初始化（不绑定内核 UDP socket），
//! 收发均经 AF_XDP Worker 的 Ring。
//!
//! # 设计约束
//! - 单 Owner：桥独占 Worker 与 ProtocolServer 的可变访问，调用方单线程驱动
//! - 有界收集：单周期交付的数据报 ≤ Worker 批量上限（64），Vec 预分配
//! - 单包失败不中断整批：L7 错误 / 回包构造失败 / TX 注入失败均计数后继续
//!   （数据面语义：单个畸形包不得 DoS 整批；Worker 级错误为数据面致命，向上传播）

#![cfg(all(feature = "afxdp", target_os = "linux"))]

use std::fmt;

use zenith_net::NetError;
use zenith_net::worker::{UdpDatagram, Worker};

use crate::server::{ProtocolServer, ServerError};

/// 单周期数据报收集上限（与 Worker 的 MAX_BATCH_SIZE 一致）
const MAX_DATAGRAMS_PER_CYCLE: usize = 64;

/// 回包帧栈上缓冲大小（Eth14 + IPv4 20 + UDP8 + QUIC 载荷 ≤ 1350，留足余量）
const REPLY_FRAME_BUF_SIZE: usize = 2048;

/// 桥接统计（全部真实计数，saturating 加法防回绕）
#[derive(Debug, Clone, Copy, Default)]
pub struct BridgeStats {
    /// 已完成的数据面周期数
    pub cycles_completed: u64,
    /// 交付 L7 的 UDP 数据报总数
    pub datagrams_delivered: u64,
    /// L7 返回的响应数据报总数
    pub l7_responses: u64,
    /// 成功注入 TX Ring 的回包帧数
    pub tx_frames_queued: u64,
    /// 回包丢弃数（构造失败 / TX 注入失败 / 超 MTU）
    pub tx_dropped: u64,
    /// L7 处理错误数（单包失败，不中断整批）
    pub l7_errors: u64,
}

impl BridgeStats {
    /// saturating 自增（统计永不回绕）
    fn saturating_add(&mut self, other: &BridgeStats) {
        self.cycles_completed = self.cycles_completed.saturating_add(other.cycles_completed);
        self.datagrams_delivered = self.datagrams_delivered.saturating_add(other.datagrams_delivered);
        self.l7_responses = self.l7_responses.saturating_add(other.l7_responses);
        self.tx_frames_queued = self.tx_frames_queued.saturating_add(other.tx_frames_queued);
        self.tx_dropped = self.tx_dropped.saturating_add(other.tx_dropped);
        self.l7_errors = self.l7_errors.saturating_add(other.l7_errors);
    }
}

/// 桥接错误
#[derive(Debug)]
pub enum BridgeError {
    /// QUIC 服务器未以 socketless 模式初始化（须先 `bind_quic_socketless`）
    QuicNotBound,
    /// 数据面错误（Worker 周期失败，数据面致命）
    Net(NetError),
    /// L7 服务错误（构造期校验失败）
    Server(ServerError),
}

impl fmt::Display for BridgeError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            BridgeError::QuicNotBound => write!(
                f,
                "QUIC server not bound — call bind_quic_socketless() before bridging"
            ),
            BridgeError::Net(e) => write!(f, "datapath error: {e}"),
            BridgeError::Server(e) => write!(f, "L7 server error: {e}"),
        }
    }
}

impl std::error::Error for BridgeError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            BridgeError::QuicNotBound => None,
            BridgeError::Net(e) => Some(e),
            BridgeError::Server(e) => Some(e),
        }
    }
}

impl From<NetError> for BridgeError {
    fn from(e: NetError) -> Self {
        BridgeError::Net(e)
    }
}

/// AF_XDP → HTTP/3 数据面桥
///
/// 独占 Worker（数据面）与 ProtocolServer（L7）的所有权，
/// 由调用方在专用线程中以 [`Self::run_cycle`] 循环驱动。
///
/// # 构造前置条件（fail-closed 校验）
/// - `server` 必须已完成 `bind_quic_socketless`（否则 [`BridgeError::QuicNotBound`]）
/// - `worker` 状态由调用方管理（`start()` 后桥周期才有意义；未启动时 Worker 自身报错）
#[derive(Debug)]
pub struct AfxdpH3Bridge {
    /// AF_XDP 数据面 Worker（L2-L4 解析 + 准入 + Ring 管理）
    worker: Worker,
    /// L7 协议服务器（QUIC/H3 全栈 + 安全管道 + App）
    server: ProtocolServer,
    /// 桥接统计
    stats: BridgeStats,
    /// UDP 数据报收拢缓冲（构造时一次性预分配，跨周期 clear 复用，
    /// 稳态热路径零堆分配；容量有界 = 单周期 RX 批量上限）
    datagrams_buf: Vec<UdpDatagram>,
}

impl AfxdpH3Bridge {
    /// 创建桥（fail-closed：QUIC 未以 socketless 模式绑定时拒绝构造）
    ///
    /// 严格校验 `is_quic_socketless`：仅接受 `bind_quic_socketless` 初始化的
    /// 服务器（无内核 UDP socket）。`bind_quic`（内核 socket 旁路）或未绑定
    /// 均拒绝——桥的存在意义就是替代内核 socket 路径。
    pub fn new(worker: Worker, server: ProtocolServer) -> Result<Self, BridgeError> {
        if !server.is_quic_socketless() {
            return Err(BridgeError::QuicNotBound);
        }
        Ok(Self {
            worker,
            server,
            stats: BridgeStats::default(),
            datagrams_buf: Vec::with_capacity(MAX_DATAGRAMS_PER_CYCLE),
        })
    }

    /// 把桥内 Worker 的 XSK socket 注册进 XSKMAP（`queue_id → xsk_fd`）
    ///
    /// XDP 程序经 `xsk_map` 将命中包重定向到本 socket；未注册时真实数据面哑死。
    /// 真实部署接线顺序：Worker（`with_real_af_xdp`）→ `AfxdpH3Bridge::new`
    /// → **本方法** → `run_cycle` 循环。模拟 Worker / map 更新失败由
    /// [`Worker::register_xsk`] fail-closed 原样向上传播。
    #[cfg(target_os = "linux")]
    pub fn register_worker_xsk(
        &mut self,
        maps: &zenith_ebpf::BpfMaps<'_>,
    ) -> Result<(), BridgeError> {
        self.worker.register_xsk(maps).map_err(BridgeError::Net)
    }

    /// 执行一个桥接周期：RX 解析交付 → L7 处理 → 回包注入 TX
    ///
    /// # 阶段
    /// 1. 数据面周期：收包 / 解析 / 准入，放行的 UDP 数据报收集到本地有界 Vec
    /// 2. L7 处理：逐数据报喂入 QUIC/H3 全栈（含安全管道与 App）
    /// 3. 回包注入：逐响应构造 L2-L4 回包帧，注入 TX Ring
    ///
    /// # 返回
    /// `Ok(本周期 Worker 处理的包数)`；Worker 级错误向上传播（数据面致命）。
    /// 单包级失败（L7 错误 / 回包构造失败 / TX 满）计数后继续，绝不中断整批。
    pub fn run_cycle(&mut self) -> Result<u32, BridgeError> {
        // ── 阶段 1：数据面周期，收集准入放行的 UDP 数据报
        // （有界：单周期 RX 批量 ≤ 64；缓冲为桥持有，clear 复用，稳态零堆分配）
        self.datagrams_buf.clear();
        let processed = {
            let buf = &mut self.datagrams_buf;
            self.worker
                .process_cycle_with_udp_sink(&mut |d| buf.push(d))?
        };

        let mut cycle_stats = BridgeStats {
            cycles_completed: 1,
            datagrams_delivered: self.datagrams_buf.len() as u64,
            ..BridgeStats::default()
        };

        // ── 阶段 2 + 3：L7 处理与回包注入
        for dgram in &self.datagrams_buf {
            let responses = match self
                .server
                .serve_udp_quic_datagram(&dgram.payload, dgram.src_socket_addr())
            {
                Ok(r) => r,
                Err(e) => {
                    // 单包 L7 失败：计数后继续（畸形包不得 DoS 整批）
                    cycle_stats.l7_errors += 1;
                    tracing::debug!("afxdp bridge: L7 datagram error: {e}");
                    continue;
                }
            };
            cycle_stats.l7_responses += responses.len() as u64;

            for (payload, _to) in responses {
                // 回包帧构造（栈上缓冲，热路径零堆分配）
                let mut frame_buf = [0u8; REPLY_FRAME_BUF_SIZE];
                match dgram.build_reply_frame(&payload, &mut frame_buf) {
                    Some(len) => match self.worker.queue_tx_data(&frame_buf[..len]) {
                        Ok(_) => cycle_stats.tx_frames_queued += 1,
                        Err(e) => {
                            cycle_stats.tx_dropped += 1;
                            tracing::debug!("afxdp bridge: TX inject failed: {e}");
                        }
                    },
                    None => {
                        cycle_stats.tx_dropped += 1;
                        tracing::debug!("afxdp bridge: reply frame build failed (oversize/IPv6)");
                    }
                }
            }
        }

        self.stats.saturating_add(&cycle_stats);
        Ok(processed)
    }

    /// 获取桥接统计
    #[inline]
    pub fn stats(&self) -> BridgeStats {
        self.stats
    }

    /// 获取 Worker 引用（观测：WorkerStats / QUIC 统计）
    #[inline]
    pub fn worker(&self) -> &Worker {
        &self.worker
    }

    /// 获取 Worker 可变引用（生命周期管理：start/stop）
    #[inline]
    pub fn worker_mut(&mut self) -> &mut Worker {
        &mut self.worker
    }

    /// 获取 L7 服务器引用（观测：指标导出 / 审计快照）
    #[inline]
    pub fn server(&self) -> &ProtocolServer {
        &self.server
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use zenith_net::source_admission::SourceAdmissionEngine;
    use zenith_net::worker::WorkerState;
    use zenith_tls::CertGeneration;

    use crate::app::{App, success_response};
    use crate::quic_server::QuicServerConfig;
    use crate::server::ProtocolServer;

    // ─────────────────────────────────────────────────────────────────────
    // 测试证书（CN=localhost，与 zenith-tls 测试证书同源）
    // ─────────────────────────────────────────────────────────────────────
    const TEST_CERT_PEM: &[u8] = b"-----BEGIN CERTIFICATE-----
MIIDCTCCAfGgAwIBAgIUEUD9CUfA86J7odmW+9fgH5y3ziIwDQYJKoZIhvcNAQEL
BQAwFDESMBAGA1UEAwwJbG9jYWxob3N0MB4XDTI2MDcyOTIyMDAxMFoXDTI3MDcy
OTIyMDAxMFowFDESMBAGA1UEAwwJbG9jYWxob3N0MIIBIjANBgkqhkiG9w0BAQEF
AAOCAQ8AMIIBCgKCAQEAuAcuUu9Ajdh//C3n75jRuI0CAfI1EdX/SzDAUZoK1Vbi
+7ZQJ2y7YXvJEa03/CYMn7Qr7Cj/NW+7HZNvxD4o6gAMrNdB4qIEOEM/mSQwXryH
ELJb/it/mC66Sklm3hYjNx0naPLf/5ZGlXuBjr1Um5dT5V1H9BG/tssO2jFbcGpo
MnbZ16J0LO4Jgh8ojaRqzu408CpFnYLUueNwNeG/T+3LUxFpdWNf5uvu4ErW0R4/
rWH4dPlS6/UktQ92jxHHukikPj8hnDkv0H4TfNn+H3cIRIA76wNqif6UTPYPLtkU
4WfvUKHPsto4SQSvG2nk31wtWJZ7PFIS611KpnbXYQIDAQABo1MwUTAdBgNVHQ4E
FgQUxsaRnqWYFIFm8wcXRUMeUGWNOdYwHwYDVR0jBBgwFoAUxsaRnqWYFIFm8wcX
RUMeUGWNOdYwDwYDVR0TAQH/BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAgAzn
7GtDuMU1hLiQdXohLzZh2bu1ySmas3ZoAm/Ydbs6+wQA7CaGZ+Q0gTb0CZZUR60r
6qQvD3JkYqNuakUHZlDWZ6NboO9O3F4SybRR3F9Y08kjog7jORAFqh/QpfVgh2Ep
psDAK2TxIFKaLG4It4wE68x9FhHQZfJIekSe1qJXAd6CYmhbZRCIu/5xScijCpJ7
BO0mTj7WU1u3bj8/7jld4ct5dbtEIHpF4rq/5uSCbqPNvA8sqA4i4+beK3aEiXl1
yCPpOR133rFfOg/t0FiqKy1Vtj+kDJgVhzjPrw1w3L50h6iRmug52ozLYMdyOQOI
rEpfPnlOnBBdLz8hjA==
-----END CERTIFICATE-----
";

    const TEST_KEY_PEM: &[u8] = b"-----BEGIN PRIVATE KEY-----
MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQC4By5S70CN2H/8
LefvmNG4jQIB8jUR1f9LMMBRmgrVVuL7tlAnbLthe8kRrTf8JgyftCvsKP81b7sd
k2/EPijqAAys10HiogQ4Qz+ZJDBevIcQslv+K3+YLrpKSWbeFiM3HSdo8t//lkaV
e4GOvVSbl1PlXUf0Eb+2yw7aMVtwamgydtnXonQs7gmCHyiNpGrO7jTwKkWdgtS5
43A14b9P7ctTEWl1Y1/m6+7gStbRHj+tYfh0+VLr9SS1D3aPEce6SKQ+PyGcOS/Q
fhN82f4fdwhEgDvrA2qJ/pRM9g8u2RThZ+9Qoc+y2jhJBK8baeTfXC1Ylns8UhLr
XUqmdtdhAgMBAAECggEAEh9y6mvxWYa2o+kJbEkKbjhEuFhX7Ze7enYkmmSnKHdU
ByHfJuLIWUNNe9YpK0W7/IZLxQgMigCk1rbMTPEqKlEy7lqMfHskGz5UJwqvUMUU
MArAkHlMKXqAkgxEex6G/Uh7txQkBxGPhe0RxzLSADiY5H+ZNGoDDUdWARrXPGyz
ZWlNY/nwg5qKO8zQePNAtUvHUWtNq3G8achi96NsFuuxuXwLKOx60JCMcg5BpRpb
2v8F+2r0ZjngkhI+wYGE7E5E/BdbbooHb1MFyajc+C/GH+sOhSfMMthjwKiSSaIB
/Jpqp9ETJ4GsxTxZTTOszyM5yPBIiPIeuwR0LwXRWQKBgQDjwBJuUqNl0CNMj7I7
6FmPrylzY7JKQB3q1SXqVPKeuPArd2BNBjGX6Ic9k0lLtmcDcsHnKNSXZaTb6UQw
APFlsdWBkWfaNeVh7SLlhez1MWwryuWCw/V8RdMR8MiOcZ+QfjPBLpUgJi/e6CLE
kvv8806vSKwhXQf1Ho43cmhrvwKBgQDO2sOBcYHupfjqOzWewJfR+rOs9F3/YYr/
fZQpdRu4z8zPfTfeR7O8NhvTtNJuzwi/yOjas1DEriBql8zUVDRHJZibj43qvuG6
Ua/cTmbcnyiEv15V0nmv91rEPgTtaf6/iz/zVvlKYT0Zj3wIxpX3rTiw6Ba3wD7I
81EM4K4E3wKBgQCVtmgnP4mL3xOlO3y1ptpg+ossAChuaNGB0lXXQbovnnC6kgGr
AFxPeJqWXqC69Y+oE6LlWtDNKRMDQMcDK0uERy+LudLj/bPo+KKM8MnAsJlj/D99
A2X3KEtEqtybzpNOv7cz0XRUKuYjCMP6JokhUauyy/njAK2/czOXvUxpLwKBgQCX
h+haRdViBpGevPsdrYZKGzZOR8EoGMOjP9IuwIwrSYaGpPstSSdgg97EqpzQ8bc+
DyaNN3i+a7RxgXxaOskFKYRuyK20vlpLjBWg9IojqjAbdrjbc9ES18fVJH2lkdU9
afvR/e+mzi7dL6A0KY2on2t9JLenqhwURzIjld/EzwKBgEoHKFsBZQqYoTBjw46g
VP3U0WBNPxtLlFNFWTh7SVAXTDkSOi4j5YeSEzGt4EHdd+RSVSjap0Yq+kLE3ZzB
hDJ2OdlHsaaSJcZ9+Vnnj44fSAJUKZcAQhq9CfZKcxvQiDuxTP3GZRl0urUdiedO
n6Pgv5yWjDwcNLUlPWhljvBK
-----END PRIVATE KEY-----
";

    // ─────────────────────────────────────────────────────────────────────
    // 测试辅助
    // ─────────────────────────────────────────────────────────────────────

    /// 创建模拟模式 Worker（小 Ring，零拷贝关闭）
    fn make_worker() -> Worker {
        let config = zenith_net::worker::XskConfig {
            ifindex: 0,
            queue_id: 0,
            zero_copy: false,
            fill_ring_size: 256,
            rx_ring_size: 256,
            tx_ring_size: 256,
            completion_ring_size: 256,
            shared_umem: false,
            frame_size: 4096,
            headroom: 0,
            ..Default::default()
        };
        Worker::new(0, config, 256, SourceAdmissionEngine::allow_all())
            .expect("模拟 Worker 构造必须成功")
    }

    /// 创建 socketless QUIC 绑定的 L7 服务器
    fn make_server() -> ProtocolServer {
        let mut app = App::new();
        app.get("/", |_req, _rm| {
            Ok(success_response("hello from h3", "text/plain"))
        });
        let mut server = ProtocolServer::new(app);
        let cert = CertGeneration::from_pem(TEST_CERT_PEM, TEST_KEY_PEM)
            .expect("测试证书必须可解析");
        let cfg = QuicServerConfig {
            bind_addr: "10.0.0.1:443".parse().expect("合法地址"),
            ..QuicServerConfig::default()
        };
        server
            .bind_quic_socketless(cfg, &cert)
            .expect("socketless 绑定必须成功");
        server
    }

    /// 构造 Eth + IPv4 + UDP 帧（载荷调用方给定）
    ///
    /// 对端 172.16.0.1:12345（MAC AA:BB:CC:DD:EE:FF）→ 本机 10.0.0.1:443（MAC 11:22:33:44:55:66）
    fn build_udp_frame(payload: &[u8]) -> Vec<u8> {
        let ip_total = (20 + 8 + payload.len()) as u16;
        let udp_len = (8 + payload.len()) as u16;
        let mut frame = vec![0u8; 14 + ip_total as usize];

        // Ethernet
        frame[0..6].copy_from_slice(&[0x11, 0x22, 0x33, 0x44, 0x55, 0x66]); // dst = 本机
        frame[6..12].copy_from_slice(&[0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]); // src = 对端
        frame[12] = 0x08;
        frame[13] = 0x00;

        // IPv4
        let ip = 14;
        frame[ip] = 0x45;
        frame[ip + 2..ip + 4].copy_from_slice(&ip_total.to_be_bytes());
        frame[ip + 8] = 64;
        frame[ip + 9] = 17; // UDP
        frame[ip + 12..ip + 16].copy_from_slice(&[172, 16, 0, 1]); // src
        frame[ip + 16..ip + 20].copy_from_slice(&[10, 0, 0, 1]); // dst

        // UDP
        let l4 = ip + 20;
        frame[l4..l4 + 2].copy_from_slice(&12345u16.to_be_bytes());
        frame[l4 + 2..l4 + 4].copy_from_slice(&443u16.to_be_bytes());
        frame[l4 + 4..l4 + 6].copy_from_slice(&udp_len.to_be_bytes());

        frame[l4 + 8..].copy_from_slice(payload);
        frame
    }

    /// 构造结构合法的 QUIC Long Header（Initial 形态）载荷
    ///
    /// 加密部分为无效数据 —— 服务端会创建连接、尝试解密、失败后 fail-closed
    /// 静默丢弃（RFC 9000 防放大语义），返回空响应。用于验证桥接管道贯通，
    /// 不依赖客户端密码学栈。
    fn build_initial_like_payload() -> Vec<u8> {
        let mut p = Vec::with_capacity(64);
        p.push(0xC3); // Long + Fixed + Initial + PN len 4
        p.extend_from_slice(&1u32.to_be_bytes()); // version = QUIC v1
        p.push(8); // DCID len
        p.extend_from_slice(&[0xC1, 0xC2, 0xC3, 0xC4, 0xC5, 0xC6, 0xC7, 0xC8]); // DCID
        p.push(8); // SCID len
        p.extend_from_slice(&[0xD1, 0xD2, 0xD3, 0xD4, 0xD5, 0xD6, 0xD7, 0xD8]); // SCID
        p.push(0); // Token len = 0
        p.extend_from_slice(&[0x40, 0x14]); // Length varint = 20（PN 4 + 载荷 16）
        p.extend_from_slice(&[0, 0, 0, 0]); // Packet number
        p.extend_from_slice(&[0xAB; 16]); // 无效加密载荷（解密必失败）
        p
    }

    // ─────────────────────────────────────────────────────────────────────
    // 测试
    // ─────────────────────────────────────────────────────────────────────

    #[test]
    fn test_bridge_requires_socketless_quic() {
        let worker = make_worker();
        // 未绑定 QUIC 的服务器 → 构造必须 fail-closed 拒绝
        let app = App::new();
        let server = ProtocolServer::new(app);
        match AfxdpH3Bridge::new(worker, server) {
            Err(BridgeError::QuicNotBound) => {}
            other => panic!("未绑定 QUIC 时必须返回 QuicNotBound，实际: {}", match other {
                Ok(_) => "Ok".to_string(),
                Err(e) => e.to_string(),
            }),
        }
    }

    #[test]
    fn test_socketless_bind_has_no_kernel_socket() {
        let server = make_server();
        // socketless 模式：无内核 UDP socket（铁则 1：quic_local_addr 为 None）
        assert!(server.quic_local_addr().is_none());
    }

    #[test]
    fn test_bridge_cycle_delivers_datagram_to_l7() {
        let mut worker = make_worker();
        worker.start();
        // 预填充 Fill Ring：模拟模式下 simulate_rx_transfer 的描述符来自 Fill Ring，
        // 须先跑一个空周期填充（与 test_bridge_admission_deny_blocks_delivery 一致）
        worker.process_cycle().expect("预填充周期必须成功");
        let server = make_server();
        let mut bridge = AfxdpH3Bridge::new(worker, server).expect("桥构造必须成功");

        // 注入 1 帧：Eth/IPv4/UDP + QUIC Initial 形态载荷
        let frame = build_udp_frame(&build_initial_like_payload());
        bridge.worker_mut().simulate_rx_transfer(1, frame.len(), |_idx, buf| {
            buf[..frame.len()].copy_from_slice(&frame);
        });

        let processed = bridge.run_cycle().expect("桥周期必须成功");
        assert_eq!(processed, 1, "Worker 必须处理 1 个包");

        let stats = bridge.stats();
        assert_eq!(stats.cycles_completed, 1);
        assert_eq!(stats.datagrams_delivered, 1, "数据报必须交付 L7");
        // 无效加密 Initial：QUIC 服务器正确返回 CONNECTION_CLOSE（RFC 9000 §10.2），
        // 不计入 l7_errors（协议错误而非处理异常），但 l7_responses >= 1
        assert_eq!(stats.l7_errors, 0, "CONNECTION_CLOSE 不应计为 L7 错误");
        // QUIC 服务器对不可解密 Initial 返回 CONNECTION_CLOSE 帧（RFC 9000 §10.3）
        assert!(stats.l7_responses >= 1, "CONNECTION_CLOSE 响应应生成 >=1 个响应包");
        assert!(stats.tx_frames_queued >= 1, "CONNECTION_CLOSE 应注入 TX");
    }

    #[test]
    fn test_bridge_admission_deny_blocks_delivery() {
        let config = zenith_net::worker::XskConfig {
            ifindex: 0,
            queue_id: 0,
            zero_copy: false,
            fill_ring_size: 256,
            rx_ring_size: 256,
            tx_ring_size: 256,
            completion_ring_size: 256,
            shared_umem: false,
            frame_size: 4096,
            headroom: 0,
            ..Default::default()
        };
        // 默认拒绝引擎：任何包都不放行
        let mut worker = Worker::new(
            0,
            config,
            256,
            SourceAdmissionEngine::deny_all(),
        )
        .expect("Worker 构造必须成功");
        worker.start();
        // 预填充 Fill Ring（同上）
        worker.process_cycle().expect("预填充周期必须成功");
        let server = make_server();
        let mut bridge = AfxdpH3Bridge::new(worker, server).expect("桥构造必须成功");

        let frame = build_udp_frame(&build_initial_like_payload());
        bridge.worker_mut().simulate_rx_transfer(1, frame.len(), |_idx, buf| {
            buf[..frame.len()].copy_from_slice(&frame);
        });

        let processed = bridge.run_cycle().expect("桥周期必须成功");
        assert_eq!(processed, 1, "拒绝的包也计入处理数");
        let stats = bridge.stats();
        assert_eq!(stats.datagrams_delivered, 0, "准入拒绝的数据报禁止交付 L7");
        assert_eq!(bridge.worker().stats().rejected_packets, 1);
    }

    #[test]
    fn test_bridge_tx_injection_path() {
        // TX 路径确定性验证：手工构造数据报 → 回包帧 → queue_tx_data → TX Ring
        let mut worker = make_worker();
        worker.start();

        let dgram = UdpDatagram {
            src_ip: zenith_net::IpAddr::V4([172, 16, 0, 1]),
            dst_ip: zenith_net::IpAddr::V4([10, 0, 0, 1]),
            src_port: 12345,
            dst_port: 443,
            src_mac: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF],
            dst_mac: [0x11, 0x22, 0x33, 0x44, 0x55, 0x66],
            payload: smallvec::SmallVec::new(),
        };
        let payload = b"quic-response-bytes";
        let mut frame_buf = [0u8; 2048];
        let len = dgram
            .build_reply_frame(payload, &mut frame_buf)
            .expect("回包帧构造必须成功");

        let frame_idx = worker
            .queue_tx_data(&frame_buf[..len])
            .expect("TX 注入必须成功");
        assert!(frame_idx < 256, "返回值是注入帧的索引，必须落在帧池容量内");

        // 模拟内核发送完成 → 帧回收（守恒闭环）
        worker.simulate_tx_complete(1);
        let stats = worker.stats();
        assert_eq!(stats.tx_packets, 0, "queue_tx_data 不计入 rx/tx 转发统计（独立注入路径）");
    }

    #[test]
    fn test_bridge_worker_state_passthrough() {
        let worker = make_worker();
        let server = make_server();
        let mut bridge = AfxdpH3Bridge::new(worker, server).expect("桥构造必须成功");

        // 未 start 的 Worker：桥周期必须报错（Worker 状态机 fail-closed）
        assert!(bridge.run_cycle().is_err());

        bridge.worker_mut().start();
        assert_eq!(bridge.worker().state(), WorkerState::Running);
        assert!(bridge.run_cycle().is_ok(), "空周期必须成功");
        assert_eq!(bridge.stats().cycles_completed, 1);
    }
}