zenith-net 0.1.0

//! 单队列 Worker 数据面循环
//!
//! 集成 AF_XDP、FramePool、SourceAdmissionEngine 的核心数据处理循环：
//! - 单 Owner 设计，零锁零堆分配热路径
//! - 批量处理所有 Ring 操作
//! - 步数预算防止畸形包消耗
//! - 所有权守恒校验每个周期执行
//! - 所有缓冲区预分配，热路径零堆分配
//!
//! # 数据流
//! ```text
//! FramePool → FillRing → RXRing → 协议解析 + 准入 → TXRing → CompletionRing → FramePool
//! ```

use crate::error::{NetError, Result};
use crate::packet::{parse_packet, L4Protocol};
use crate::protocol_expectation::ProtocolExpectation;
use crate::source_admission::{AdmissionAction, IpAddr, SourceAdmissionEngine};
use crate::transport::{
    parse_quic_header_with_dcid_len, QuicAction, QuicConnParams, QuicConnectionTable, QuicHeaderType,
};

use std::sync::Arc;

use zenith_foundation::{FrameId, FramePool};
use zenith_linux::descriptor::XdpDesc;
use zenith_linux::umem::{UmemConfig, UmemManager};
use zenith_linux::xsk::XskSocket;

// XskConfig 为 Worker::new 系列公开构造函数的参数类型，经本模块 re-export
// 保证公开 API 参数类型对下游 crate 可达（无需直接依赖 zenith-linux）
pub use zenith_linux::xsk::XskConfig;

/// Worker 状态
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkerState {
    /// 已创建
    Created,
    /// 正在运行
    Running,
    /// 已停止
    Stopped,
    /// 已出错
    Error,
}

/// Worker 统计信息
#[derive(Debug, Clone, Copy, Default)]
pub struct WorkerStats {
    /// 总接收包数
    pub rx_packets: u64,
    /// 总发送包数
    pub tx_packets: u64,
    /// 准入拒绝包数
    pub rejected_packets: u64,
    /// 解析失败包数
    pub parse_errors: u64,
    /// 帧池分配次数
    pub frame_allocs: u64,
    /// 帧池释放次数
    pub frame_frees: u64,
    /// 隔离区帧数
    pub quarantined_frames: u64,
    /// 处理周期数
    pub cycles_completed: u64,
    /// CPU 使用率（0.0 - 1.0）
    pub cpu_usage: f64,
    /// 当前队列深度
    pub queue_depth: u64,
    /// P99 延迟（微秒）
    pub latency_p99: u64,
    /// 吞吐量（包/秒）
    pub throughput: u64,
    /// 非预期包丢弃数（协议预期检查未通过）
    pub unexpected_dropped: u64,
}

/// 最大批量大小（每个循环周期处理的最大帧数）
const MAX_BATCH_SIZE: usize = 64;

/// 帧大小（与 UMEM 帧大小一致）
const FRAME_SIZE: usize = 4096;

/// 单队列 Worker
///
/// 核心数据面处理循环：
/// 1. 填充 Fill Ring（从 FramePool 分配帧）
/// 2. 批量接收（从 RX Ring 读取）
/// 3. 协议解析与准入检查
/// 4. 批量发送（到 TX Ring）
/// 5. 从 Completion Ring 回收帧归还 FramePool
///
/// # 设计约束
/// - 单线程：无 Arc/RwLock/Mutex
/// - 热路径零堆分配：所有缓冲区预分配
/// - 批量处理：Ring 操作仅使用 batch API
/// - 所有权守恒：每个周期结束时校验
pub struct Worker {
    /// Worker ID
    id: u32,
    /// XSK Socket
    xsk: XskSocket,
    /// 帧池
    frame_pool: FramePool,
    /// 准入引擎
    admission: SourceAdmissionEngine,
    /// Worker 状态
    state: WorkerState,
    /// 统计信息
    stats: WorkerStats,
    /// 域 ID（用于帧池分配）
    domain_id: u32,
    /// UMEM 模拟缓冲区（存储帧数据，仅在模拟模式下使用）
    umem_buf: Vec<u8>,
    /// 真实 UMEM 管理器（Some=真实内核 AF_XDP 模式；None=模拟模式）
    ///
    /// 持有 Arc 强引用以保证：
    /// - XskSocket 通过 bind() 引用的 mmap 区域在 Worker 生命周期内始终有效
    /// - Worker drop 时，若这是最后一个 Arc 持有者，UmemManager 的 drop 会 munmap
    umem_manager: Option<Arc<UmemManager>>,
    /// 预分配 RX 描述符缓冲区（热路径零堆分配）
    rx_desc_buf: [XdpDesc; MAX_BATCH_SIZE],
    /// RX 描述符数量（本次周期接收到的）
    rx_desc_count: usize,
    /// 准入拒绝时是否隔离帧
    quarantine_on_error: bool,
    /// 准入拒绝时是否隔离帧
    quarantine_on_deny: bool,
    /// QUIC 连接表（预分配固定容量）
    quic_table: QuicConnectionTable,
    /// 周期耗时采样环（64 槽 u64 微秒，栈上零堆分配）
    latency_samples: [u64; 64],
    /// 采样环写入位置（取模环绕）
    latency_sample_idx: usize,
    /// 有效采样数（饱和于 64，P99 仅在有效样本上计算）
    latency_sample_count: usize,
    /// 上周期结束时刻（cpu_usage 的墙钟分母）
    last_cycle_end: Option<std::time::Instant>,
    /// QUIC 统计
    quic_stats: WorkerQuicStats,
    /// 帧分配代际侧表（ABA 防护）
    ///
    /// 分配帧时捕获 `FrameToken::generation()`；释放/隔离时以此作为期望值
    /// 传给 `release_by_id`/`quarantine_by_id`。若帧已被释放并重新分配
    /// （FramePool 代际已推进），陈旧释放请求的期望代际与池当前代际不匹配，
    /// 被 fail-closed 拒绝——此前以"读取池当前代际"自证，代际校验形同虚设。
    alloc_generations: Vec<u64>,
    /// 协议预期配置（None = 不检查；Some = 准入放行后二次校验）
    protocol_expectation: Option<ProtocolExpectation>,
}

/// 计算延迟样本的 P99（微秒）
///
/// 仅在最近 `valid` 个有效样本上计算（样本环未填满时忽略未写入的零槽）；
/// 64 槽栈上排序，零堆分配。`valid` 为 0 时返回 0（无负载，诚实语义）。
fn compute_p99(samples: &[u64; 64], valid: usize) -> u64 {
    let valid = valid.min(64);
    if valid == 0 {
        return 0;
    }
    let mut sorted = *samples;
    sorted.sort_unstable();
    // rank ∈ 1..=valid（ceil(0.99 * valid)），从升序数组尾部的有效区取值
    let rank = (valid * 99).div_ceil(100).clamp(1, valid);
    sorted[64 - valid + (rank - 1)]
}

/// 准入放行的 UDP 数据报（L7 数据出口）
///
/// 由 [`Worker::process_cycle_with_udp_sink`] 对每个准入放行的 UDP 包交付一次。
/// 载荷从 UMEM 帧拷贝（v1 语义；零拷贝视图需延长帧借用生命周期，留待后续优化）。
///
/// 除载荷外还携带完整的 L2-L4 回包上下文（对端/本端 MAC、双向 IP、双向端口），
/// 使 L7 服务可直接用 [`Self::build_reply_frame`] 构造回包帧注入 TX Ring，
/// 全程无需内核 socket（铁则 1 合规）。
#[derive(Debug, Clone)]
pub struct UdpDatagram {
    /// 源 IP（数据报发送方）
    pub src_ip: IpAddr,
    /// 目的 IP（本机侧）
    pub dst_ip: IpAddr,
    /// 源端口
    pub src_port: u16,
    /// 目的端口（本机服务端口）
    pub dst_port: u16,
    /// 源 MAC（对端）
    pub src_mac: [u8; 6],
    /// 目的 MAC（本机网卡）
    pub dst_mac: [u8; 6],
    /// UDP 载荷（不含 L2-L4 头）：SmallVec 栈内联 2048B，
    /// 覆盖以太网 MTU(1500)/QUIC 初始核(1200+)场景，交付 L7 全程零堆分配；
    /// 超大载荷（巨型帧/分片重组场景）溢出时自动回落堆（语义不变）
    pub payload: smallvec::SmallVec<[u8; 2048]>,
}

impl UdpDatagram {
    /// 转换为标准库 SocketAddr（供 L7 服务寻址回包）
    pub fn src_socket_addr(&self) -> std::net::SocketAddr {
        let ip = match self.src_ip {
            IpAddr::V4(b) => std::net::IpAddr::V4(std::net::Ipv4Addr::from(b)),
            IpAddr::V6(b) => std::net::IpAddr::V6(std::net::Ipv6Addr::from(b)),
            // NET-019：跨族通配映射为 IPv4 通配地址
            IpAddr::Any => std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED),
        };
        std::net::SocketAddr::new(ip, self.src_port)
    }

    /// 构造回包帧（Eth + IPv4 + UDP + payload），地址/端口全部对调
    ///
    /// - 仅支持 IPv4（IPv6 的 UDP 校验和为强制项，需伪头计算，留待后续；返回 None 诚实拒绝）
    /// - UDP 校验和置 0（RFC 768：IPv4 下可选）；IPv4 头校验和真实计算（强制项）
    /// - checked 算术：`buf` 容量不足或载荷超 MTU 上限时返回 None（fail-closed，绝不截断）
    ///
    /// # 返回
    /// `Some(帧长)` 写入成功；`None` 无法构造（调用方应计丢弃并继续，禁止 panic）
    pub fn build_reply_frame(&self, payload: &[u8], buf: &mut [u8]) -> Option<usize> {
        use crate::packet::{ETH_HEADER_LEN, UDP_HEADER_LEN};

        const IPV4_HEADER_LEN: usize = 20;
        /// 单帧载荷上限：Eth(14) + IPv4(20) + UDP(8) + 载荷 ≤ u16 字段上限
        const MAX_IP_TOTAL: usize = u16::MAX as usize;

        let (local_ip, peer_ip) = match (self.dst_ip, self.src_ip) {
            (IpAddr::V4(local), IpAddr::V4(peer)) => (local, peer),
            // IPv6 回包需强制 UDP 校验和（伪头计算），v1 诚实拒绝
            _ => return None,
        };

        let total_len = ETH_HEADER_LEN
            .checked_add(IPV4_HEADER_LEN)?
            .checked_add(UDP_HEADER_LEN)?
            .checked_add(payload.len())?;
        let ip_total = IPV4_HEADER_LEN.checked_add(UDP_HEADER_LEN)?.checked_add(payload.len())?;
        if total_len > buf.len() || ip_total > MAX_IP_TOTAL {
            return None;
        }

        // ── Ethernet：MAC 对调，ethertype = IPv4
        buf[0..6].copy_from_slice(&self.src_mac); // 目的 = 对端
        buf[6..12].copy_from_slice(&self.dst_mac); // 源 = 本机
        buf[12] = 0x08;
        buf[13] = 0x00;

        // ── IPv4 头（20 字节，无选项）
        let ip = ETH_HEADER_LEN;
        buf[ip] = 0x45; // version=4, IHL=5
        buf[ip + 1] = 0; // DSCP/ECN
        buf[ip + 2..ip + 4].copy_from_slice(&(ip_total as u16).to_be_bytes());
        buf[ip + 4] = 0; // Identification
        buf[ip + 5] = 0;
        buf[ip + 6] = 0x40; // Flags: DF（禁止分片，QUIC 要求）
        buf[ip + 7] = 0; // Fragment offset
        buf[ip + 8] = 64; // TTL
        buf[ip + 9] = crate::packet::ip_proto::UDP;
        buf[ip + 10] = 0; // 校验和先置 0 再计算
        buf[ip + 11] = 0;
        buf[ip + 12..ip + 16].copy_from_slice(&local_ip);
        buf[ip + 16..ip + 20].copy_from_slice(&peer_ip);
        let cksum = crate::packet::compute_ipv4_checksum(&buf[ip..ip + IPV4_HEADER_LEN]);
        buf[ip + 10..ip + 12].copy_from_slice(&cksum.to_be_bytes());

        // ── UDP 头：端口对调，length 含头（RFC 768），校验和置 0（IPv4 可选）
        let l4 = ip + IPV4_HEADER_LEN;
        // checked 算术 + try_from 形式化：`as u16` 截断语义禁止（§4.4 fail-closed）。
        // 前置校验 ip_total ≤ MAX_IP_TOTAL 已蕴含 udp_len < u16::MAX，
        // try_from 失败属不可达，但仍以 None 拒绝而非 unwrap/panic。
        let udp_len = u16::try_from(UDP_HEADER_LEN.checked_add(payload.len())?).ok()?;
        buf[l4..l4 + 2].copy_from_slice(&self.dst_port.to_be_bytes());
        buf[l4 + 2..l4 + 4].copy_from_slice(&self.src_port.to_be_bytes());
        buf[l4 + 4..l4 + 6].copy_from_slice(&udp_len.to_be_bytes());
        buf[l4 + 6] = 0;
        buf[l4 + 7] = 0;

        // ── 载荷
        buf[l4 + UDP_HEADER_LEN..total_len].copy_from_slice(payload);
        Some(total_len)
    }
}

/// Worker QUIC 统计
#[derive(Debug, Clone, Copy, Default)]
pub struct WorkerQuicStats {
    /// QUIC 包总数
    pub quic_packets: u64,
    /// QUIC 新连接数
    pub quic_new_connections: u64,
    /// QUIC 放大攻击拦截数
    pub quic_amplification_blocked: u64,
    /// QUIC 无效包数
    pub quic_invalid_packets: u64,
}

impl Worker {
    /// 创建新的 Worker
    ///
    /// # 参数
    /// * `id` - Worker 唯一 ID
    /// * `xsk_config` - XSK Socket 配置
    /// * `frame_pool_capacity` - 帧池容量
    /// * `admission` - 预配置的准入引擎
    ///
    /// # 返回
    /// * `Result<Self>` - 新 Worker 或错误
    pub fn new(
        id: u32,
        xsk_config: XskConfig,
        frame_pool_capacity: u32,
        admission: SourceAdmissionEngine,
    ) -> Result<Self> {
        let xsk = XskSocket::new(xsk_config)?;
        let frame_pool = FramePool::new(
            format!("worker-{}", id),
            frame_pool_capacity,
            FRAME_SIZE as u32,
        );

        let umem_buf = vec![0u8; (frame_pool_capacity as usize) * FRAME_SIZE];

        Ok(Self {
            id,
            xsk,
            frame_pool,
            admission,
            state: WorkerState::Created,
            stats: WorkerStats::default(),
            domain_id: id,
            umem_buf,
            umem_manager: None,
            rx_desc_buf: [XdpDesc::zero(); MAX_BATCH_SIZE],
            rx_desc_count: 0,
            quarantine_on_error: false,
            quarantine_on_deny: false,
            quic_table: QuicConnectionTable::new(64),
            latency_samples: [0u64; 64],
            latency_sample_idx: 0,
            latency_sample_count: 0,
            last_cycle_end: None,
            quic_stats: WorkerQuicStats::default(),
            alloc_generations: vec![0u64; frame_pool_capacity as usize],
            protocol_expectation: None,
        })
    }

    /// 创建 Worker 并使用外部 FramePool
    pub fn with_frame_pool(
        id: u32,
        xsk_config: XskConfig,
        frame_pool: FramePool,
        admission: SourceAdmissionEngine,
    ) -> Result<Self> {
        let capacity = frame_pool.capacity();
        let xsk = XskSocket::new(xsk_config)?;

        let umem_buf = vec![0u8; (capacity as usize) * FRAME_SIZE];

        Ok(Self {
            id,
            xsk,
            frame_pool,
            admission,
            state: WorkerState::Created,
            stats: WorkerStats::default(),
            domain_id: id,
            umem_buf,
            umem_manager: None,
            rx_desc_buf: [XdpDesc::zero(); MAX_BATCH_SIZE],
            rx_desc_count: 0,
            quarantine_on_error: false,
            quarantine_on_deny: false,
            quic_table: QuicConnectionTable::new(64),
            latency_samples: [0u64; 64],
            latency_sample_idx: 0,
            latency_sample_count: 0,
            last_cycle_end: None,
            quic_stats: WorkerQuicStats::default(),
            alloc_generations: vec![0u64; capacity as usize],
            protocol_expectation: None,
        })
    }

    /// 创建真实 AF_XDP 模式的 Worker（严格 fail-closed：任何一步失败返回错误，绝不静默降级到模拟）
    ///
    /// 完整串联真实系统调用链：
    /// 1. 分配 PAGE_SIZE 对齐的 UMEM 配置
    /// 2. `UmemManager::new + create()` → 真实 `libc::mmap`（含 HugePage/`mlock`/`madvise`）
    /// 3. `XskSocket::new` → `create_socket()` → 真实 `libc::socket(AF_XDP, SOCK_RAW, 0)`
    /// 4. `XskSocket::configure()` → 真实 `setsockopt(SO_RCVBUF | SO_SNDBUF, 1MB)`
    /// 5. `XskSocket::bind(Arc<UmemManager>)` → 真实 `libc::bind(sockaddr_xdp{ifindex,queue_id,flags})`
    ///
    /// 完成后：
    /// - `self.xsk.state == XskState::Bound`（非 Created/Configured）
    /// - `self.umem_manager.is_some()`，持有 Arc 强引用保证 mmap 区域生命周期
    /// - `self.is_real_af_xdp() == true`
    ///
    /// 失败场景：任何系统调用失败（ENOMEM/EPERM/EADDRNOTAVAIL/ENODEV）直接向上传播错误。
    /// **本方法从不回退到模拟模式**——如需模拟请使用 `Worker::new`。
    pub fn with_real_af_xdp(
        id: u32,
        xsk_config: XskConfig,
        frame_pool_capacity: u32,
        admission: SourceAdmissionEngine,
        use_hugepage: bool,
        lock_memory: bool,
    ) -> Result<Self> {
        // ── UMEM 计算：帧数 * FRAME_SIZE，并按 PAGE_SIZE 对齐
        let umem_size_raw = (frame_pool_capacity as usize)
            .checked_mul(FRAME_SIZE)
            .ok_or(NetError::Internal("frame_pool_capacity * FRAME_SIZE overflow"))?;

        let page_size = {
            // 使用 zenith_linux::page_size() 安全封装：内部 unsafe 已隔离，
            // 此处 zenith-net 保持 #![deny(unsafe_code)] 零 unsafe。
            zenith_linux::page_size()
        };
        // 向上对齐 page_size
        let umem_size = (umem_size_raw + page_size - 1) & !(page_size - 1);

        let mut umem_mgr = UmemManager::new(UmemConfig {
            size: umem_size,
            hugepage: use_hugepage,
            locked: lock_memory,
            shared: false,
        })?;
        umem_mgr.create()?; // 真实 mmap / mlock / madvise
        let umem_arc: Arc<UmemManager> = Arc::new(umem_mgr);

        // ── XSK Socket：真实 socket() + setsockopt() + bind(2) 到内核 ifindex+queue_id
        let mut xsk = XskSocket::new(xsk_config)?;
        xsk.create_socket()?;   // libc::socket(AF_XDP, SOCK_RAW, 0)
        xsk.configure()?;       // SO_RCVBUF + SO_SNDBUF = 1MB
        xsk.bind(Arc::clone(&umem_arc))?; // libc::bind(sockaddr_xdp{...})

        let frame_pool = FramePool::new(
            format!("worker-{id}"),
            frame_pool_capacity,
            FRAME_SIZE as u32,
        );
        let umem_buf = vec![0u8; (frame_pool_capacity as usize) * FRAME_SIZE];

        Ok(Self {
            id,
            xsk,
            frame_pool,
            admission,
            state: WorkerState::Created,
            stats: WorkerStats::default(),
            domain_id: id,
            umem_buf,
            umem_manager: Some(umem_arc),
            rx_desc_buf: [XdpDesc::zero(); MAX_BATCH_SIZE],
            rx_desc_count: 0,
            quarantine_on_error: false,
            quarantine_on_deny: false,
            quic_table: QuicConnectionTable::new(64),
            latency_samples: [0u64; 64],
            latency_sample_idx: 0,
            latency_sample_count: 0,
            last_cycle_end: None,
            quic_stats: WorkerQuicStats::default(),
            alloc_generations: vec![0u64; frame_pool_capacity as usize],
            protocol_expectation: None,
        })
    }

    /// Worker 是否运行在真实内核 AF_XDP 模式（非模拟）
    ///
    /// 严格双重校验：
    /// 1. `xsk.get_fd().is_ok()` → 已完成 create_socket() 的 fd 创建，非模拟模式
    /// 2. `umem_manager.is_some()` → 已完成真实 mmap 的 UMEM 持有
    #[inline]
    pub fn is_real_af_xdp(&self) -> bool {
        self.xsk.get_fd().is_ok() && self.umem_manager.is_some()
    }

    /// 把本 Worker 的 XSK socket 注册进 XSKMAP（`queue_id → xsk_fd`）
    ///
    /// XDP 程序通过 `xsk_map` 将命中数据包重定向到对应队列的 socket；
    /// 未注册时真实数据面处于哑死状态（网卡收到包但 no socket 可投递）。
    /// 真实 AF_XDP 初始化的必经一步：`with_real_af_xdp` 完成 socket+bind
    /// 之后、进入数据面循环之前调用一次。
    ///
    /// # Fail-Closed
    /// - 模拟模式或 socket 未创建（`get_fd` 无 fd）→ 返回错误，绝不静默跳过
    /// - map 更新失败（权限/map 不存在）→ 原样向上传播
    #[cfg(all(feature = "linux", target_os = "linux"))]
    pub fn register_xsk(&self, maps: &zenith_ebpf::BpfMaps<'_>) -> Result<()> {
        let fd = self.xsk.get_fd()?;
        // i32 → u32：禁 as 截断；fd ≥ 0 由内核保证，仍 u32::try_from fail-closed
        let fd_u32 = u32::try_from(fd).map_err(|_| {
            NetError::Internal("xsk fd negative (kernel ABI 异常)")
        })?;
        let queue_id = self.xsk.queue_id();
        maps.update_xsk(queue_id, fd_u32)?;
        tracing::debug!(
            worker_id = self.id,
            queue_id,
            xsk_fd = fd,
            "XSKMAP registered: queue -> xsk_fd"
        );
        Ok(())
    }

    /// 获取真实 UMEM 管理器引用（真实模式下 Some；模拟模式 None）
    #[inline]
    pub fn umem_manager(&self) -> Option<&Arc<UmemManager>> {
        self.umem_manager.as_ref()
    }

    /// 获取 QUIC 连接表引用
    #[inline]
    pub fn quic_table(&self) -> &QuicConnectionTable {
        &self.quic_table
    }

    /// 获取 QUIC 统计
    #[inline]
    pub fn quic_stats(&self) -> WorkerQuicStats {
        self.quic_stats
    }

    /// 获取 QUIC 连接表可变引用
    #[inline]
    pub fn quic_table_mut(&mut self) -> &mut QuicConnectionTable {
        &mut self.quic_table
    }

    /// 处理一个 QUIC UDP 数据包
    ///
    /// QUIC 数据包基于 UDP 承载（通常为端口 443）。
    /// 本方法负责：
    /// 1. 解析 QUIC 包头
    /// 2. Long Header：按 DCID 查找或创建连接
    /// 3. Retry Token 三倍防放大验证
    /// 4. Short Header：必须已建立连接，否则视为无效包
    /// 5. 统计放大攻击风险
    pub fn handle_quic_packet(
        &mut self,
        data: &[u8],
        remote_addr: IpAddr,
        remote_port: u16,
        local_addr: IpAddr,
        local_port: u16,
        ip_version: crate::packet::IpVersion,
    ) -> QuicAction {
        let hdr = match parse_quic_header_with_dcid_len(data, 8) {
            Some(h) => h,
            None => {
                self.quic_stats.quic_invalid_packets += 1;
                return QuicAction::InvalidPacket;
            }
        };
        self.quic_stats.quic_packets += 1;

        match hdr.header_type {
            QuicHeaderType::Short => {
                // 必须已有连接
                let conn_idx = match self.quic_table.find_by_dcid(&hdr.dcid[..hdr.dcid_len as usize]) {
                    Some(idx) => idx,
                    None => {
                        self.quic_stats.quic_invalid_packets += 1;
                        return QuicAction::InvalidPacket;
                    }
                };
                if let Some(c) = self.quic_table.get(conn_idx) {
                    c.observe_packet_number(hdr.packet_number);
                    c.record_rx_bytes(data.len() as u64);
                }
                QuicAction::DataReceived {
                    conn_idx,
                    stream_id: hdr.packet_number,
                }
            }
            QuicHeaderType::VersionNegotiation | QuicHeaderType::Retry => {
                // Retry 包本身作为服务端下发，客户端不应在此处理
                // 对服务端：Retry 包无需新建连接，直接忽略
                QuicAction::InvalidPacket
            }
            QuicHeaderType::Long => {
                // Initial / Handshake / ZeroRTT
                // 1. 查找已有连接（以 DCID 为键）
                if let Some(idx) = self
                    .quic_table
                    .find_by_dcid(&hdr.dcid[..hdr.dcid_len as usize])
                    && let Some(c) = self.quic_table.get(idx) {
                        c.observe_packet_number(hdr.packet_number);
                        c.record_rx_bytes(data.len() as u64);
                        // 验证 Token
                        if hdr.has_token()
                            && let Ok(ok) = c.verify_token(
                                &hdr.token[..hdr.token_len as usize],
                            )
                                && !ok {
                                    self.quic_stats.quic_amplification_blocked += 1;
                                    return QuicAction::AmplificationBlocked { conn_idx: idx };
                                }
                        return QuicAction::DataReceived {
                            conn_idx: idx,
                            stream_id: hdr.packet_number,
                        };
                    }
                // 2. 新建连接（Initial 包）
                match self.quic_table.allocate(QuicConnParams {
                    scid: &hdr.dcid[..hdr.dcid_len as usize],
                    dcid: &hdr.scid[..hdr.scid_len as usize],
                    remote_addr,
                    remote_port,
                    local_addr,
                    local_port,
                    ip_version,
                }) {
                    Ok(idx) => {
                        if let Some(c) = self.quic_table.get(idx) {
                            c.observe_packet_number(hdr.packet_number);
                            c.record_rx_bytes(data.len() as u64);
                            if hdr.has_token() {
                                // 首次连接保存 Token 用于下次验证
                                c.retry_token[..hdr.token_len as usize]
                                    .copy_from_slice(&hdr.token[..hdr.token_len as usize]);
                                c.retry_token_len = hdr.token_len;
                            }
                        }
                        self.quic_stats.quic_new_connections += 1;
                        QuicAction::NewConnection { conn_idx: idx }
                    }
                    Err(_) => QuicAction::InvalidPacket,
                }
            }
        }
    }

    /// 获取 Worker ID
    #[inline]
    pub fn id(&self) -> u32 {
        self.id
    }

    /// 获取 Worker 状态
    #[inline]
    pub fn state(&self) -> WorkerState {
        self.state
    }

    /// 获取统计信息
    #[inline]
    pub fn stats(&self) -> WorkerStats {
        self.stats
    }

    /// 获取准入引擎的可变引用
    #[inline]
    pub fn admission_mut(&mut self) -> &mut SourceAdmissionEngine {
        &mut self.admission
    }

    /// 获取准入引擎的不可变引用
    #[inline]
    pub fn admission(&self) -> &SourceAdmissionEngine {
        &self.admission
    }

    /// 设置解析错误时是否隔离帧
    #[inline]
    pub fn set_quarantine_on_error(&mut self, enabled: bool) {
        self.quarantine_on_error = enabled;
    }

    /// 设置准入拒绝时是否隔离帧
    #[inline]
    pub fn set_quarantine_on_deny(&mut self, enabled: bool) {
        self.quarantine_on_deny = enabled;
    }

    /// 获取解析错误隔离配置
    #[inline]
    pub fn quarantine_on_error(&self) -> bool {
        self.quarantine_on_error
    }

    /// 获取准入拒绝隔离配置
    #[inline]
    pub fn quarantine_on_deny(&self) -> bool {
        self.quarantine_on_deny
    }

    /// 设置协议预期配置（None = 禁用检查）
    #[inline]
    pub fn set_protocol_expectation(&mut self, expectation: Option<ProtocolExpectation>) {
        self.protocol_expectation = expectation;
    }

    /// 获取协议预期配置引用
    #[inline]
    pub fn protocol_expectation(&self) -> Option<&ProtocolExpectation> {
        self.protocol_expectation.as_ref()
    }

    /// 获取帧池引用
    #[inline]
    pub fn frame_pool(&self) -> &FramePool {
        &self.frame_pool
    }

    /// 获取帧池可变引用
    ///
    /// 用于运行时级别执行 quarantine / recover / release 等修改帧状态的操作。
    #[inline]
    pub fn frame_pool_mut(&mut self) -> &mut FramePool {
        &mut self.frame_pool
    }

    /// 启动 Worker
    pub fn start(&mut self) {
        self.state = WorkerState::Running;
    }

    /// 停止 Worker
    pub fn stop(&mut self) {
        self.state = WorkerState::Stopped;
    }

    /// 获取 UMEM 中指定帧的数据
    ///
    /// # 参数
    /// * `frame_idx` - 帧索引
    ///
    /// # 返回
    /// * `Option<&[u8]>` - 帧数据切片；越界返回 None
    ///
    /// # 安全
    /// 使用 checked 算术计算偏移与边界，越界返回 `None` 而非 panic
    ///（符合 AGENT.md §5.1「checked 算术」与「禁止 unwrap/越界」铁律）。
    #[inline]
    pub fn get_frame_data(&self, frame_idx: u32) -> Option<&[u8]> {
        let idx = frame_idx as usize;
        let offset = idx.checked_mul(FRAME_SIZE)?;
        let end = offset.checked_add(FRAME_SIZE)?;
        match &self.umem_manager {
            // 真实 AF_XDP 模式：读取内核 DMA 写入的真实 UMEM mmap 区域
            // （此前错误地读取独立的 umem_buf 全零堆缓冲，真实数据面断裂）
            Some(umem) => umem.slice(offset, FRAME_SIZE),
            // 模拟模式：读取用户态模拟缓冲
            None => self.umem_buf.get(offset..end),
        }
    }

    /// 获取 UMEM 中指定帧的可变数据
    ///
    /// # 安全
    /// 同 [`Self::get_frame_data`]，越界返回 `None`。
    /// 真实模式下写入真实 UMEM mmap（TX 路径：内核随后从该区域发送）。
    #[inline]
    pub fn get_frame_data_mut(&mut self, frame_idx: u32) -> Option<&mut [u8]> {
        let idx = frame_idx as usize;
        let offset = idx.checked_mul(FRAME_SIZE)?;
        let end = offset.checked_add(FRAME_SIZE)?;
        match &self.umem_manager {
            // 真实 AF_XDP 模式：写真实 UMEM mmap（单 Owner Worker 顺序访问，
            // 同一帧无并发活跃借用，满足 UmemManager::slice_mut 契约）
            Some(umem) => umem.slice_mut(offset, FRAME_SIZE),
            None => self.umem_buf.get_mut(offset..end),
        }
    }

    /// 模拟内核行为：将数据从 Fill Ring 转移到 RX Ring
    ///
    /// 在真实环境中，此步骤由内核完成（网卡 DMA 收到数据包后
    /// 将描述符从 Fill Ring 转移到 RX Ring）。
    /// 在模拟模式下，我们手动执行此操作。
    ///
    /// # 参数
    /// * `data_len` - 每个帧要填充的数据长度
    /// * `data_generator` - 生成测试数据的闭包
    pub fn simulate_rx_transfer<F>(&mut self, count: u32, data_len: usize, data_generator: F)
    where
        F: Fn(u32, &mut [u8]),
    {
        let mut descs: [XdpDesc; MAX_BATCH_SIZE] = [XdpDesc::zero(); MAX_BATCH_SIZE];
        let mut desc_count = 0usize;
        for _ in 0..(count as usize).min(MAX_BATCH_SIZE) {
            if let Ok(desc) = self.xsk.fill_ring_mut().dequeue_batch(1)
                && let Some(d) = desc.first() {
                    let frame_idx = (d.addr / FRAME_SIZE as u64) as u32;
                    // 安全访问：越界描述符直接丢弃（守恒铁律）
                    let Some(data) = self.get_frame_data_mut(frame_idx) else {
                        // 越界描述符回收入 Fill Ring，避免丢帧所有权
                        let _ = self.frame_pool.quarantine_by_id(
                            FrameId::new(frame_idx),
                            self.frame_generation(frame_idx),
                            "frame_idx_out_of_range",
                        );
                        self.stats.quarantined_frames += 1;
                        continue;
                    };
                    let len = data_len.min(FRAME_SIZE);
                    data_generator(frame_idx, &mut data[..len]);
                    descs[desc_count] = *d;
                    desc_count += 1;
                }
        }
        if desc_count > 0 {
            let _ = self.xsk.rx_ring_mut().enqueue_batch(&descs[..desc_count]);
        }
    }

    /// 模拟内核完成发送：将 TX Ring 描述符转移到 Completion Ring
    ///
    /// 在真实环境中，此步骤由内核在网卡发送完成后执行。
    pub fn simulate_tx_complete(&mut self, count: u32) {
        if let Ok(descs) = self.xsk.tx_ring_mut().dequeue_batch(count)
            && !descs.is_empty() {
                let _ = self.xsk.completion_ring_mut().enqueue_batch(&descs);
            }
    }

    /// 执行一个数据面处理周期
    ///
    /// 包装器：统一测量周期耗时并刷新 WorkerStats 的
    /// queue_depth / latency_p99 / throughput / cpu_usage（AutoOptimizer 真实输入）。
    ///
    /// # 返回
    /// * `Result<u32>` - 本周期处理的包数
    pub fn process_cycle(&mut self) -> Result<u32> {
        let cycle_start = std::time::Instant::now();
        let result = self.process_cycle_inner();
        let processed = result.as_ref().ok().copied().unwrap_or(0);
        self.finish_cycle_metrics(cycle_start, processed);
        result
    }

    /// 周期收尾测量：写入延迟样本环、P99、队列深度、吞吐、CPU 占用率
    ///
    /// 全部 saturating/除零防护；64 槽采样环栈上排序，热路径零堆分配。
    fn finish_cycle_metrics(&mut self, cycle_start: std::time::Instant, processed: u32) {
        let elapsed_us = cycle_start.elapsed().as_micros() as u64;

        // 延迟样本环 + P99（仅在有效样本上计算）
        self.latency_samples[self.latency_sample_idx] = elapsed_us;
        self.latency_sample_idx = (self.latency_sample_idx + 1) % 64;
        self.latency_sample_count = (self.latency_sample_count + 1).min(64);
        self.stats.latency_p99 =
            compute_p99(&self.latency_samples, self.latency_sample_count);

        // 队列深度 = 本周期 RX 描述符数
        self.stats.queue_depth = self.rx_desc_count as u64;

        // 吞吐 = 本周期处理包数换算 pps（checked_div：零耗时按已达最大速率计）
        let processed_pps = u64::from(processed).saturating_mul(1_000_000);
        self.stats.throughput = processed_pps.checked_div(elapsed_us).unwrap_or(processed_pps);

        // CPU 占用率 = 本周期忙时 / 距上周期结束的墙钟间隔（clamp 0.0..=1.0）
        let now = std::time::Instant::now();
        if let Some(last) = self.last_cycle_end {
            let wall_us = now.duration_since(last).as_micros() as u64;
            self.stats.cpu_usage = if wall_us > 0 {
                (elapsed_us as f64 / wall_us as f64).clamp(0.0, 1.0)
            } else {
                // 背靠背周期（零间隔）：视为满载
                1.0
            };
        }
        self.last_cycle_end = Some(now);
    }

    /// 执行一个数据面处理周期，并将准入放行的 UDP 数据报交付给 L7 汇
    ///
    /// 与 [`Self::process_cycle`] 的唯一区别：Allow 的 UDP 包在转入 TX Ring 前，
    /// 其载荷经 `sink` 交付（L7 数据出口，供 AF_XDP → QUIC/HTTP3 桥接消费）。
    /// 测量语义（queue_depth/latency_p99/throughput/cpu_usage）完全一致。
    pub fn process_cycle_with_udp_sink(
        &mut self,
        sink: &mut dyn FnMut(UdpDatagram),
    ) -> Result<u32> {
        let cycle_start = std::time::Instant::now();
        let result = self.process_cycle_inner_with_sink(Some(sink));
        let processed = result.as_ref().ok().copied().unwrap_or(0);
        self.finish_cycle_metrics(cycle_start, processed);
        result
    }

    /// 数据面处理周期内部实现（Phase 0-5）
    fn process_cycle_inner(&mut self) -> Result<u32> {
        self.process_cycle_inner_with_sink(None)
    }

    /// 数据面处理周期内部实现（可选 UDP 数据出口）
    fn process_cycle_inner_with_sink(
        &mut self,
        sink: Option<&mut dyn FnMut(UdpDatagram)>,
    ) -> Result<u32> {
        if self.state != WorkerState::Running {
            return Err(NetError::WorkerState {
                reason: "worker not running",
            });
        }

        // Phase 0: 先回收上一周期完成的帧
        self.recycle_completed_frames()?;

        // Phase 1: 填充 Fill Ring
        let filled = self.fill_fill_ring()?;

        // Phase 2: 批量接收（从 RX Ring 读取）
        let received = self.rx_batch_preallocated()?;

        if received == 0 && filled == 0 {
            self.stats.cycles_completed += 1;
            return Ok(0);
        }

        // Phase 3: 处理数据包（解析 + 准入）
        // 先复制描述符到栈上以避免借用冲突（可变/不可变同时借用）
        let mut descs_copy: [XdpDesc; MAX_BATCH_SIZE] = [XdpDesc::zero(); MAX_BATCH_SIZE];
        descs_copy[..self.rx_desc_count].copy_from_slice(&self.rx_desc_buf[..self.rx_desc_count]);
        let (processed, allow_count, deny_count) = self
            .process_packets_batch_with_sink(&descs_copy[..self.rx_desc_count], sink)?;

        // Phase 4: 准入拒绝的帧归还帧池
        self.recycle_rejected_frames(deny_count)?;

        // Phase 5: 允许通过的帧提交到 TX Ring
        self.submit_to_tx_ring(allow_count)?;

        self.stats.cycles_completed += 1;
        Ok(processed)
    }

    /// Phase 1: 填充 Fill Ring
    ///
    /// 从帧池分配帧，填充到 XSK Fill Ring。
    /// 同时初始化 UMEM 缓冲区。
    ///
    /// 批量语义（AGENT.md §6.1.1）：先批量分配帧到栈上数组，
    /// 再一次 `enqueue_batch` 单内存屏障入队 n 个描述符，
    /// 取代此前逐帧单元素入队（每帧一次 Release store）。
    #[inline]
    fn fill_fill_ring(&mut self) -> Result<u32> {
        let available = self.xsk.fill_ring().available_space();
        if available == 0 {
            return Ok(0);
        }

        let to_fill = (available.min(MAX_BATCH_SIZE as u32)) as usize;
        let mut descs: [XdpDesc; MAX_BATCH_SIZE] = [XdpDesc::zero(); MAX_BATCH_SIZE];
        let mut tokens: [Option<zenith_foundation::FrameToken>; MAX_BATCH_SIZE] =
            [const { None }; MAX_BATCH_SIZE];
        let mut count = 0usize;

        // 阶段 1：批量分配帧并清零（栈上数组暂存，热路径零堆分配）
        for (desc_slot, token_slot) in descs.iter_mut().zip(tokens.iter_mut()).take(to_fill) {
            match self.frame_pool.allocate(self.domain_id) {
                Ok(token) => {
                    let frame_idx = token.frame_id().value();
                    // 捕获分配时代际到侧表（ABA 防护：释放时以此自证，
                    // 取代此前"读取池当前代际"的无效自证）
                    if let Some(g) = self.alloc_generations.get_mut(frame_idx as usize) {
                        *g = token.generation();
                    }

                    // 写入 UMEM 数据（清零表示空帧）
                    // 安全访问：frame_idx 由 FramePool 分配，理论上必在范围内；
                    // 仍使用 checked 访问以满足「禁止 unwrap/越界」铁律。
                    if let Some(data) = self.get_frame_data_mut(frame_idx) {
                        data.fill(0);
                    } else {
                        // 越界：立即隔离此帧（手持真实 token 代际，非自证）
                        let _ = self.frame_pool.quarantine_by_id(
                            FrameId::new(frame_idx),
                            token.generation(),
                            "frame_idx_out_of_range",
                        );
                        self.stats.quarantined_frames += 1;
                        continue;
                    }

                    *desc_slot = XdpDesc {
                        addr: (frame_idx as u64) << 12,
                        len: 0,
                        options: 0,
                    };
                    *token_slot = Some(token);
                    count += 1;
                }
                Err(_) => break,
            }
        }

        if count == 0 {
            return Ok(0);
        }

        // 阶段 2：单次批量入队（单内存屏障处理 n 元素）
        let enqueued = self
            .xsk
            .fill_ring_mut()
            .enqueue_batch(&descs[..count])? as usize;

        // 阶段 3：所有权结算——已入队的 token 所有权转移给 ring，
        // 未入队的归还帧池（禁止泄漏）
        for (i, token_slot) in tokens.iter_mut().enumerate().take(count) {
            if i < enqueued {
                self.stats.frame_allocs += 1;
                // token 被 ring 消费（转移所有权给内核/ring），
                // FrameToken 未实现 Drop，此处显式丢弃以标记所有权转移
                let _ = token_slot.take();
            } else if let Some(token) = token_slot.take() {
                let _ = self.frame_pool.release(token);
            }
        }

        Ok(enqueued as u32)
    }

    /// Phase 2: 从 RX Ring 批量接收（预分配缓冲区，零堆分配）
    #[inline]
    fn rx_batch_preallocated(&mut self) -> Result<usize> {
        let available = self.xsk.rx_ring().available_data();
        if available == 0 {
            self.rx_desc_count = 0;
            return Ok(0);
        }

        let to_receive = available.min(MAX_BATCH_SIZE as u32);
        let received = self
            .xsk
            .rx_ring_mut()
            .dequeue_batch_to(&mut self.rx_desc_buf[..to_receive as usize])?;

        self.rx_desc_count = received as usize;
        self.stats.rx_packets += received as u64;
        Ok(self.rx_desc_count)
    }

    /// Phase 3: 批量处理数据包（可选 UDP 数据出口）
    ///
    /// 对接收到的每个包进行协议解析和准入判定；处理完的描述符根据准入动作
    /// 分发到 TX Ring 或 Completion Ring。当 `sink` 为 Some 时，准入放行的
    /// UDP 包额外交付一份载荷拷贝给 L7 汇（帧所有权语义不变）。
    ///
    /// # 返回
    /// * `(processed, allow_count, deny_count)` - 处理总数、允许数、拒绝数
    fn process_packets_batch_with_sink(
        &mut self,
        descs: &[XdpDesc],
        mut sink: Option<&mut dyn FnMut(UdpDatagram)>,
    ) -> Result<(u32, u32, u32)> {
        if descs.is_empty() {
            return Ok((0, 0, 0));
        }

        let mut processed = 0u32;
        let mut allow_count = 0u32;
        let mut deny_count = 0u32;

        for desc in descs {
            if desc.is_zero() {
                continue;
            }

            let frame_idx = (desc.addr / FRAME_SIZE as u64) as u32;
            // 安全访问：越界描述符进入隔离区（守恒铁律）
            let Some(data) = self.get_frame_data(frame_idx) else {
                self.stats.parse_errors += 1;
                deny_count += 1;
                processed += 1;
                let _ = self.frame_pool.quarantine_by_id(
                    FrameId::new(frame_idx),
                    self.frame_generation(frame_idx),
                    "frame_idx_out_of_range",
                );
                self.stats.quarantined_frames += 1;
                let _ = self.xsk.completion_ring_mut().enqueue_batch(&[*desc]);
                continue;
            };

            // 解析协议
            let parsed = match parse_packet(data) {
                Ok(p) => p,
                Err(_e) => {
                    self.stats.parse_errors += 1;
                    deny_count += 1;
                    processed += 1;

                    // 根据配置决定是隔离还是直接回收
                    if self.quarantine_on_error {
                        let _ = self.frame_pool.quarantine_by_id(
                            FrameId::new(frame_idx),
                            self.frame_generation(frame_idx),
                            "parse_error",
                        );
                        self.stats.quarantined_frames += 1;
                    }

                    // 解析失败的帧进入 Completion Ring 回收
                    let _ = self.xsk.completion_ring_mut().enqueue_batch(&[*desc]);
                    continue;
                }
            };

            // 提取五元组信息
            let (src_ip, src_port, dst_port, protocol) = self.extract_flow_info(&parsed);

            // 准入判定
            let action = self
                .admission
                .evaluate(src_ip, src_port, dst_port, protocol);

            match action {
                AdmissionAction::Allow => {
                    // 准入放行后：协议预期检查（端口白名单 / TTL / 分片策略）
                    if let Some(ref expectation) = self.protocol_expectation {
                        let ttl = parsed.ipv4.map(|ip| ip.ttl()).unwrap_or(64);
                        let is_fragment = parsed.ipv4.map(|ip| {
                            ip.fragment_offset() > 0 || ip.more_fragments()
                        }).unwrap_or(false);
                        if !expectation.is_packet_expected(protocol, dst_port, ttl, is_fragment) {
                            // 非预期包：静默丢弃（回收到 Completion Ring）
                            self.stats.unexpected_dropped += 1;
                            let _ = self.xsk.completion_ring_mut().enqueue_batch(&[*desc]);
                            processed += 1;
                            continue;
                        }
                    }
                    // L7 数据出口：准入放行的 UDP 包交付载荷给汇（桥接 L7 服务）
                    if let Some(sink_fn) = sink.as_deref_mut()
                        && protocol == crate::packet::ip_proto::UDP
                        && let Some(payload) = Self::extract_udp_payload(&parsed)
                    {
                        // 回包上下文：本机侧 IP（IP 头目的地址）+ 双向 MAC
                        let dst_ip = match parsed.ip_version {
                            crate::packet::IpVersion::V4 => parsed
                                .ipv4
                                .map(|i| IpAddr::V4(i.dst_ip()))
                                .unwrap_or(IpAddr::V4_WILDCARD),
                            crate::packet::IpVersion::V6 => parsed
                                .ipv6
                                .map(|i| IpAddr::V6(i.dst_ip()))
                                .unwrap_or(IpAddr::V6_WILDCARD),
                            crate::packet::IpVersion::Unknown => IpAddr::V4_WILDCARD,
                        };
                        sink_fn(UdpDatagram {
                            src_ip,
                            dst_ip,
                            src_port,
                            dst_port,
                            src_mac: parsed.eth.src_mac(),
                            dst_mac: parsed.eth.dst_mac(),
                            // SmallVec：≤2048B 栈内联，从 UMEM 帧拷贝零堆分配
                            payload: smallvec::SmallVec::from_slice(payload),
                        });
                    }
                    // 允许通过：描述符转入 TX Ring（模拟发送路径）
                    let _ = self.xsk.tx_ring_mut().enqueue_batch(&[*desc]);
                    allow_count += 1;
                    processed += 1;
                }
                AdmissionAction::Deny => {
                    // 拒绝：根据配置决定是隔离还是直接回收
                    if self.quarantine_on_deny {
                        let _ = self.frame_pool.quarantine_by_id(
                            FrameId::new(frame_idx),
                            self.frame_generation(frame_idx),
                            "admission_deny",
                        );
                        self.stats.quarantined_frames += 1;
                    }

                    // 描述符转入 Completion Ring 回收
                    let _ = self.xsk.completion_ring_mut().enqueue_batch(&[*desc]);
                    self.stats.rejected_packets += 1;
                    deny_count += 1;
                    processed += 1;
                }
            }
        }

        Ok((processed, allow_count, deny_count))
    }

    /// 从解析后的数据包提取 UDP 载荷（checked 算术，越界返回 None）
    ///
    /// 布局：Eth(14) [+ VLAN(4)] + IP 头（v4 IHL / v6 40） + UDP 头(8) + 载荷。
    /// 载荷长度以 UDP length 字段为准（含头，RFC 768），并夹紧到帧实际长度。
    fn extract_udp_payload<'a>(parsed: &crate::packet::ParsedPacket<'a>) -> Option<&'a [u8]> {
        use crate::packet::{ETH_HEADER_LEN, IPV6_HEADER_LEN, IpVersion, UDP_HEADER_LEN, VLAN_TAG_LEN};
        let udp = parsed.udp?;
        let eth_len = ETH_HEADER_LEN
            .checked_add(if parsed.vlan.is_some() { VLAN_TAG_LEN } else { 0 })?;
        let ip_len = match parsed.ip_version {
            IpVersion::V4 => parsed.ipv4?.header_length(),
            IpVersion::V6 => IPV6_HEADER_LEN,
            IpVersion::Unknown => return None,
        };
        let l4_start = eth_len.checked_add(ip_len)?;
        let udp_total = u16::from_be(udp.length) as usize;
        if udp_total < UDP_HEADER_LEN {
            return None;
        }
        let payload_start = l4_start.checked_add(UDP_HEADER_LEN)?;
        let payload_end = l4_start.checked_add(udp_total)?;
        if payload_end > parsed.raw.len() || payload_start > payload_end {
            return None;
        }
        Some(&parsed.raw[payload_start..payload_end])
    }

    /// 从解析后的数据包提取五元组信息
    #[inline]
    fn extract_flow_info(
        &self,
        parsed: &crate::packet::ParsedPacket<'_>,
    ) -> (IpAddr, u16, u16, u8) {
        let src_ip = match parsed.ip_version {
            crate::packet::IpVersion::V4 => {
                if let Some(ipv4) = parsed.ipv4 {
                    IpAddr::V4(ipv4.src_ip())
                } else {
                    IpAddr::V4_WILDCARD
                }
            }
            crate::packet::IpVersion::V6 => {
                if let Some(ipv6) = parsed.ipv6 {
                    IpAddr::V6(ipv6.src_ip())
                } else {
                    IpAddr::V6_WILDCARD
                }
            }
            crate::packet::IpVersion::Unknown => IpAddr::V4_WILDCARD,
        };

        let src_port = match parsed.l4_proto {
            L4Protocol::Tcp => parsed.tcp.map(|t| t.src_port()).unwrap_or(0),
            L4Protocol::Udp => parsed.udp.map(|u| u.src_port()).unwrap_or(0),
            _ => 0,
        };

        let dst_port = match parsed.l4_proto {
            L4Protocol::Tcp => parsed.tcp.map(|t| t.dst_port()).unwrap_or(0),
            L4Protocol::Udp => parsed.udp.map(|u| u.dst_port()).unwrap_or(0),
            _ => 0,
        };

        let protocol = match parsed.ip_version {
            crate::packet::IpVersion::V4 => {
                parsed.ipv4.map(|i| i.protocol()).unwrap_or(0)
            }
            crate::packet::IpVersion::V6 => {
                parsed.ipv6.map(|i| i.next_header()).unwrap_or(0)
            }
            crate::packet::IpVersion::Unknown => 0,
        };

        (src_ip, src_port, dst_port, protocol)
    }

    /// Phase 5: 回收准入拒绝的帧
    ///
    /// 拒绝的帧已经在 Completion Ring 中，直接回收即可。
    #[inline]
    fn recycle_rejected_frames(&mut self, count: u32) -> Result<()> {
        if count == 0 {
            return Ok(());
        }
        self.recycle_completed_frames_inner(count)?;
        Ok(())
    }

    /// Phase 6: 提交允许通过的帧到 TX Ring
    ///
    /// 模拟模式：手动将 TX Ring 描述符转移到 Completion Ring（模拟内核发送完成）。
    /// 真实模式：内核在网卡 DMA 完成后异步产生 Completion 事件，
    /// 由下一周期 Phase 0 的 `recycle_completed_frames` 回收——
    /// 此处**禁止**模拟完成（否则同一描述符被模拟完成与内核完成双重回收）。
    #[inline]
    fn submit_to_tx_ring(&mut self, count: u32) -> Result<()> {
        if count == 0 {
            return Ok(());
        }

        if self.is_real_af_xdp() {
            // 真实模式：通知内核发送（NEED_WAKEUP 置位时才 syscall），
            // Completion 由内核异步产生，下一周期 Phase 0 回收
            self.xsk.notify_tx()?;
            self.stats.tx_packets += count as u64;
        } else {
            // 模拟内核完成发送
            self.simulate_tx_complete(count);
            self.stats.tx_packets += count as u64;

            // 回收已完成发送的帧
            self.recycle_completed_frames_inner(count)?;
        }

        Ok(())
    }

    /// 从 Completion Ring 回收已完成的帧
    #[inline]
    fn recycle_completed_frames(&mut self) -> Result<()> {
        let available = self.xsk.completion_ring().available_data();
        if available == 0 {
            return Ok(());
        }
        let count = available.min(MAX_BATCH_SIZE as u32);
        self.recycle_completed_frames_inner(count)?;
        Ok(())
    }

    /// 内部回收实现
    #[inline]
    fn recycle_completed_frames_inner(&mut self, count: u32) -> Result<()> {
        if count == 0 {
            return Ok(());
        }

        let mut buffer: [XdpDesc; MAX_BATCH_SIZE] = [XdpDesc::zero(); MAX_BATCH_SIZE];
        let dequeued = self
            .xsk
            .completion_ring_mut()
            .dequeue_batch_to(&mut buffer[..count as usize])?;

        for desc in buffer.iter().take(dequeued as usize) {
            if desc.is_zero() {
                continue;
            }

            let frame_idx = (desc.addr / FRAME_SIZE as u64) as u32;

            // 归还帧池（由 FramePool 管理所有权状态）
            if let Some(frame_info) = self.frame_pool.get_frame_info(FrameId::new(frame_idx))
                && frame_info.state() == zenith_foundation::FrameState::Allocated {
                    let _ = self.frame_pool.release_by_id(
                        FrameId::new(frame_idx),
                        self.frame_generation(frame_idx),
                    );
                    self.stats.frame_frees += 1;
                }
        }

        Ok(())
    }

    /// 验证所有权守恒
    pub fn verify_conservation(&self) -> bool {
        self.frame_pool.verify_conservation().is_ok()
    }

    /// 获取帧池信息
    pub fn frame_pool_info(&self) -> (u32, u32, u32) {
        (
            self.frame_pool.capacity(),
            self.frame_pool.allocated_count(),
            self.frame_pool.quarantined_count(),
        )
    }

    /// 读取指定帧**分配时捕获**的代际号（ABA 防护，用于所有权验证）
    ///
    /// 返回 `alloc_generations` 侧表中本 Worker 最近一次分配该帧时记录的
    /// 代际。与 FramePool 当前代际无关：若帧已被释放并重新分配（池代际
    /// 已推进），本返回值即成为陈旧代际，`release_by_id`/`quarantine_by_id`
    /// 将 fail-closed 拒绝——代际校验由此恢复真实 ABA 防护语义。
    ///
    /// # Arguments
    /// * `frame_idx` - 帧索引
    ///
    /// # Returns
    /// * 分配时捕获的代际号，或 0（从未分配/越界时）
    #[inline]
    fn frame_generation(&self, frame_idx: u32) -> u64 {
        self.alloc_generations
            .get(frame_idx as usize)
            .copied()
            .unwrap_or(0)
    }

    /// 将出站数据注入 TX Ring（L7 桥接回包入口）
    ///
    /// 从帧池分配一帧、拷贝数据、入队 TX Ring；帧所有权随描述符转移给
    /// TX 路径（与准入放行帧同一回收链路：Completion → 归还帧池）。
    ///
    /// # 参数
    /// * `data` - 出站帧字节（必须 ≤ FRAME_SIZE，超出 fail-closed 返回错误）
    ///
    /// # 返回
    /// * `Ok(u32)` - 注入的帧索引
    /// * `Err(_)` - 数据超长 / 帧池耗尽 / 帧索引越界
    pub fn queue_tx_data(&mut self, data: &[u8]) -> Result<u32> {
        if data.len() > FRAME_SIZE {
            return Err(NetError::InvalidOperation {
                reason: format!(
                    "tx data {} bytes exceeds frame size {}",
                    data.len(),
                    FRAME_SIZE
                ),
            });
        }
        let token = self.frame_pool.allocate(self.domain_id)?;
        let frame_idx = token.frame_id().value();
        let buf = self.get_frame_data_mut(frame_idx).ok_or(NetError::Internal(
            "allocated frame index out of umem range",
        ))?;
        buf[..data.len()].copy_from_slice(data);
        // 帧尾清零（避免泄漏上一帧残留数据到线路上）
        if data.len() < FRAME_SIZE {
            buf[data.len()..].fill(0);
        }
        // 捕获分配时代际到侧表（ABA 防护：Completion Ring 回收时以此自证，
        // 与 fill_fill_ring 同一模式；缺失则 release_by_id 因代际不匹配
        // fail-closed 拒绝释放，帧永久泄漏）
        if let Some(g) = self.alloc_generations.get_mut(frame_idx as usize) {
            *g = token.generation();
        }

        let desc = XdpDesc {
            addr: (frame_idx as u64) << 12,
            len: data.len() as u32,
            options: 0,
        };
        self.xsk.tx_ring_mut().enqueue_batch(&[desc])?;
        // token 所有权随描述符转移给 TX 路径（与 fill_fill_ring 同一语义）
        let _ = token;
        Ok(frame_idx)
    }
}

impl Drop for Worker {
    fn drop(&mut self) {
        self.state = WorkerState::Stopped;
    }
}

impl std::fmt::Debug for Worker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Worker")
            .field("id", &self.id)
            .field("state", &self.state)
            .field("stats", &self.stats)
            .field("domain_id", &self.domain_id)
            .field("rx_desc_count", &self.rx_desc_count)
            .finish()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::packet::*;
    use crate::source_admission::{AdmissionRule, ProtoMatch};

    #[test]
    fn test_compute_p99_empty() {
        assert_eq!(compute_p99(&[0u64; 64], 0), 0);
    }

    #[test]
    fn test_compute_p99_partial_valid() {
        // 仅 4 个有效样本（其余零槽不得参与）
        let mut samples = [0u64; 64];
        samples[0] = 10;
        samples[1] = 20;
        samples[2] = 30;
        samples[3] = 100;
        // ceil(0.99*4)=4 → 取有效区最大值 100
        assert_eq!(compute_p99(&samples, 4), 100);
    }

    #[test]
    fn test_compute_p99_full_ring() {
        let mut samples = [0u64; 64];
        for (i, s) in samples.iter_mut().enumerate() {
            *s = (i as u64) + 1; // 1..=64
        }
        // ceil(0.99*64)=64 → 最大值 64
        assert_eq!(compute_p99(&samples, 64), 64);
    }

    #[test]
    fn test_worker_stats_real_writes() {
        let mut worker = Worker::new(
            0,
            create_test_config(),
            64,
            SourceAdmissionEngine::allow_all(),
        )
        .expect("worker 构造应成功");
        worker.start();
        // 驱动两个周期：第一周期建立 last_cycle_end 基线，第二周期产生 cpu_usage
        worker.process_cycle().expect("周期 1 应成功");
        std::thread::sleep(std::time::Duration::from_millis(1));
        worker.process_cycle().expect("周期 2 应成功");
        let stats = worker.stats();
        assert_eq!(stats.cycles_completed, 2);
        // 四字段必须真实写入（非恒零默认值）
        assert!(stats.latency_p99 > 0, "latency_p99 应非零: {stats:?}");
        assert!(stats.cpu_usage > 0.0 && stats.cpu_usage <= 1.0,
            "cpu_usage 应在 (0,1] 区间: {}", stats.cpu_usage);
        // queue_depth/throughput 零负载下为 0 属诚实语义，仅断言字段可读
        let _ = stats.queue_depth;
        let _ = stats.throughput;
    }

    fn create_test_config() -> XskConfig {
        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: FRAME_SIZE as u32,
            headroom: 0,
            ..Default::default()
        }
    }

    /// 创建一个最小的合法以太网 + IPv4 + TCP 数据包
    fn create_test_packet(frame: &mut [u8], dst_mac: [u8; 6], src_mac: [u8; 6], src_ip: [u8; 4], dst_ip: [u8; 4], src_port: u16, dst_port: u16) -> usize {
        let offset = ETH_HEADER_LEN;

        // Ethernet header
        frame[0..6].copy_from_slice(&dst_mac);
        frame[6..12].copy_from_slice(&src_mac);
        frame[12] = 0x08; // IPv4
        frame[13] = 0x00;

        // IPv4 header
        frame[offset] = 0x45; // version=4, ihl=5
        frame[offset + 1] = 0x00; // TOS
        // total length (20 + 20 = 40)
        frame[offset + 2] = 0x00;
        frame[offset + 3] = 40;
        // identification
        frame[offset + 4] = 0x00;
        frame[offset + 5] = 0x01;
        // flags + fragment
        frame[offset + 6] = 0x40;
        frame[offset + 7] = 0x00;
        // TTL
        frame[offset + 8] = 64;
        // protocol = TCP
        frame[offset + 9] = ip_proto::TCP;
        // checksum (simple, not correct for real usage)
        frame[offset + 10] = 0x00;
        frame[offset + 11] = 0x00;
        // src IP
        frame[offset + 12..offset + 16].copy_from_slice(&src_ip);
        // dst IP
        frame[offset + 16..offset + 20].copy_from_slice(&dst_ip);

        // TCP header
        let tcp_offset = offset + IPV4_MIN_HEADER_LEN;
        // src port
        frame[tcp_offset] = (src_port >> 8) as u8;
        frame[tcp_offset + 1] = (src_port & 0xFF) as u8;
        // dst port
        frame[tcp_offset + 2] = (dst_port >> 8) as u8;
        frame[tcp_offset + 3] = (dst_port & 0xFF) as u8;
        // seq num
        frame[tcp_offset + 4..tcp_offset + 8].copy_from_slice(&[0x00, 0x00, 0x00, 0x01]);
        // ack num
        frame[tcp_offset + 8..tcp_offset + 12].copy_from_slice(&[0x00, 0x00, 0x00, 0x00]);
        // data offset (5 << 4) + flags (SYN=0x02)
        frame[tcp_offset + 12] = 0x50;
        frame[tcp_offset + 13] = 0x02;
        // window
        frame[tcp_offset + 14] = 0x10;
        frame[tcp_offset + 15] = 0x00;
        // checksum
        frame[tcp_offset + 16] = 0x00;
        frame[tcp_offset + 17] = 0x00;

        ETH_HEADER_LEN + IPV4_MIN_HEADER_LEN + TCP_MIN_HEADER_LEN
    }

    #[test]
    fn test_worker_creation() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::new(1, config, 256, admission);
        assert!(worker.is_ok());

        let worker = worker.unwrap();
        assert_eq!(worker.id(), 1);
        assert_eq!(worker.state(), WorkerState::Created);
        assert_eq!(worker.frame_pool().capacity(), 256);
    }

    #[test]
    fn test_worker_start_stop() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        assert_eq!(worker.state(), WorkerState::Created);

        worker.start();
        assert_eq!(worker.state(), WorkerState::Running);

        worker.stop();
        assert_eq!(worker.state(), WorkerState::Stopped);
    }

    #[test]
    fn test_worker_not_running() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();

        let result = worker.process_cycle();
        assert!(result.is_err());
    }

    #[test]
    fn test_worker_process_cycle_empty() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        worker.start();

        // 空周期（Fill Ring 填帧，但 RX Ring 无数据）
        let result = worker.process_cycle();
        assert!(result.is_ok());

        let processed = result.unwrap();
        assert_eq!(processed, 0);

        let stats = worker.stats();
        assert!(stats.frame_allocs > 0); // 有帧被分配到 Fill Ring
        assert_eq!(stats.rx_packets, 0);
    }

    #[test]
    fn test_worker_with_admission_deny() {
        let config = create_test_config();
        let mut admission = SourceAdmissionEngine::deny_all();

        // 添加一条允许规则
        admission
            .add_rule(crate::source_admission::AdmissionRule {
                id: 1,
                src_ip: IpAddr::V4_WILDCARD,
                prefix_len: 0,
                src_port: 0,
                dst_port: 8080,
                proto: crate::source_admission::ProtoMatch::Tcp,
                action: AdmissionAction::Allow,
                enabled: true,
            })
            .unwrap();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        worker.start();

        let result = worker.process_cycle();
        assert!(result.is_ok());
    }

    #[test]
    fn test_worker_conservation_initial() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::new(1, config, 256, admission).unwrap();
        assert!(worker.verify_conservation());
    }

    #[test]
    fn test_worker_with_frame_pool() {
        let config = create_test_config();
        let pool = FramePool::new("external", 512, FRAME_SIZE as u32);
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::with_frame_pool(1, config, pool, admission);
        assert!(worker.is_ok());

        let worker = worker.unwrap();
        let (capacity, allocated, quarantined) = worker.frame_pool_info();
        assert_eq!(capacity, 512);
        assert_eq!(allocated, 0);
        assert_eq!(quarantined, 0);
    }

    #[test]
    fn test_worker_full_cycle_with_data() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        worker.start();

        // 先执行一个周期来填充 Fill Ring
        let result = worker.process_cycle();
        assert!(result.is_ok());

        // 模拟内核行为：将数据从 Fill Ring 转移到 RX Ring
        // 创建一个 TCP SYN 包
        let packet_data = |_frame_idx: u32, data: &mut [u8]| {
            create_test_packet(
                data,
                [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF],
                [0x11, 0x22, 0x33, 0x44, 0x55, 0x66],
                [10, 0, 0, 1],
                [10, 0, 0, 2],
                12345,
                80,
            );
        };

        worker.simulate_rx_transfer(4, FRAME_SIZE, packet_data);

        // 再次执行周期来处理数据
        let result = worker.process_cycle();
        assert!(result.is_ok());
        let processed = result.unwrap();
        assert_eq!(processed, 4); // 4 个包都应该被处理
    }

    #[test]
    fn test_worker_admission_deny_specific() {
        let config = create_test_config();
        let mut admission = SourceAdmissionEngine::deny_all();

        // 只允许 10.0.0.1 的 TCP 8080
        admission
            .add_rule(crate::source_admission::AdmissionRule {
                id: 1,
                src_ip: IpAddr::V4([10, 0, 0, 1]),
                prefix_len: 0,
                src_port: 0,
                dst_port: 8080,
                proto: crate::source_admission::ProtoMatch::Tcp,
                action: AdmissionAction::Allow,
                enabled: true,
            })
            .unwrap();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        worker.start();

        // 先填充 Fill Ring
        let _ = worker.process_cycle();

        // 模拟：创建一个来自 10.0.0.1:12345 → 10.0.0.2:80 的 TCP 包（端口不匹配，应该被拒绝）
        let packet_data = |_frame_idx: u32, data: &mut [u8]| {
            create_test_packet(
                data,
                [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF],
                [0x11, 0x22, 0x33, 0x44, 0x55, 0x66],
                [10, 0, 0, 1],
                [10, 0, 0, 2],
                12345,
                80, // 端口 80，规则只允许 8080
            );
        };

        worker.simulate_rx_transfer(2, FRAME_SIZE, packet_data);

        let result = worker.process_cycle();
        assert!(result.is_ok());
        let processed = result.unwrap();
        assert_eq!(processed, 2); // 被拒绝但仍被处理

        let stats = worker.stats();
        assert_eq!(stats.rejected_packets, 2); // 2 个都被拒绝
    }

    #[test]
    fn test_worker_multiple_cycles() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        worker.start();

        for _ in 0..5 {
            let result = worker.process_cycle();
            assert!(result.is_ok());
        }

        let stats = worker.stats();
        assert!(stats.cycles_completed >= 5);
    }

    #[test]
    fn test_extract_flow_info_ipv4_tcp() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::new(1, config, 256, admission).unwrap();

        // 构造 IPv4 TCP 数据包
        let mut data = vec![0u8; 100];
        create_test_packet(
            &mut data,
            [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF],
            [0x11, 0x22, 0x33, 0x44, 0x55, 0x66],
            [192, 168, 1, 100],
            [10, 0, 0, 1],
            54321,
            443,
        );

        let parsed = parse_packet(&data).unwrap();
        let (src_ip, src_port, dst_port, protocol) = worker.extract_flow_info(&parsed);

        assert_eq!(src_ip, IpAddr::V4([192, 168, 1, 100]));
        assert_eq!(src_port, 54321);
        assert_eq!(dst_port, 443);
        assert_eq!(protocol, 6); // TCP
    }

    #[test]
    fn test_extract_flow_info_udp() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::new(1, config, 256, admission).unwrap();

        // 构造 IPv4 UDP 数据包
        let mut data = vec![0u8; 100];
        let offset = ETH_HEADER_LEN;
        // Ethernet
        data[0..6].copy_from_slice(&[0xAA; 6]);
        data[6..12].copy_from_slice(&[0xBB; 6]);
        data[12] = 0x08;
        data[13] = 0x00;
        // IPv4
        data[offset] = 0x45;
        data[offset + 2] = 0x00;
        data[offset + 3] = 28; // total length
        data[offset + 8] = 64;
        data[offset + 9] = ip_proto::UDP;
        data[offset + 12..offset + 16].copy_from_slice(&[172, 16, 0, 1]);
        data[offset + 16..offset + 20].copy_from_slice(&[8, 8, 8, 8]);
        // UDP
        let udp_offset = offset + IPV4_MIN_HEADER_LEN;
        data[udp_offset] = 0x00;
        data[udp_offset + 1] = 53; // src port = 53 (DNS)
        data[udp_offset + 2] = 0x00;
        data[udp_offset + 3] = 53; // dst port = 53
        data[udp_offset + 4] = 0x00;
        data[udp_offset + 5] = 8; // length

        let parsed = parse_packet(&data).unwrap();
        let (src_ip, src_port, dst_port, protocol) = worker.extract_flow_info(&parsed);

        assert_eq!(src_ip, IpAddr::V4([172, 16, 0, 1]));
        assert_eq!(src_port, 53);
        assert_eq!(dst_port, 53);
        assert_eq!(protocol, 17); // UDP
    }

    /// 构造测试用 UDP 数据报（IPv4：对端 172.16.0.1:12345 → 本机 10.0.0.1:443）
    fn make_test_datagram() -> UdpDatagram {
        UdpDatagram {
            src_ip: IpAddr::V4([172, 16, 0, 1]),
            dst_ip: 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::from_slice(b"request"),
        }
    }

    #[test]
    fn test_build_reply_frame_layout() {
        let dgram = make_test_datagram();
        let payload = b"response-payload";
        let mut buf = [0u8; 2048];

        let len = dgram.build_reply_frame(payload, &mut buf).unwrap();
        assert_eq!(len, 14 + 20 + 8 + payload.len());

        // Ethernet：MAC 对调 + ethertype=IPv4
        assert_eq!(&buf[0..6], &[0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]); // dst = 对端
        assert_eq!(&buf[6..12], &[0x11, 0x22, 0x33, 0x44, 0x55, 0x66]); // src = 本机
        assert_eq!(&buf[12..14], &[0x08, 0x00]);

        // IPv4：版本/IHL、总长、DF、TTL、协议、地址对调
        let ip = 14;
        assert_eq!(buf[ip], 0x45);
        assert_eq!(u16::from_be_bytes([buf[ip + 2], buf[ip + 3]]), (20 + 8 + payload.len()) as u16);
        assert_eq!(buf[ip + 6], 0x40, "DF 位必须置位（QUIC 禁止分片）");
        assert_eq!(buf[ip + 8], 64);
        assert_eq!(buf[ip + 9], 17);
        assert_eq!(&buf[ip + 12..ip + 16], &[10, 0, 0, 1]); // src = 本机
        assert_eq!(&buf[ip + 16..ip + 20], &[172, 16, 0, 1]); // dst = 对端

        // IPv4 头校验和：全头（含校验和字段）一反码和必须为 0（RFC 1071 验证法）
        let mut sum: u32 = 0;
        for pair in buf[ip..ip + 20].chunks_exact(2) {
            sum += u32::from(u16::from_be_bytes([pair[0], pair[1]]));
        }
        while (sum >> 16) != 0 {
            sum = (sum & 0xFFFF) + (sum >> 16);
        }
        assert_eq!(sum as u16, 0xFFFF, "IPv4 头校验和验证失败");

        // UDP：端口对调、length 含头、校验和 0（IPv4 可选）
        let l4 = ip + 20;
        assert_eq!(u16::from_be_bytes([buf[l4], buf[l4 + 1]]), 443); // src = 本机服务端口
        assert_eq!(u16::from_be_bytes([buf[l4 + 2], buf[l4 + 3]]), 12345); // dst = 对端
        assert_eq!(u16::from_be_bytes([buf[l4 + 4], buf[l4 + 5]]), (8 + payload.len()) as u16);
        assert_eq!(&buf[l4 + 6..l4 + 8], &[0, 0]);

        // 载荷原样
        assert_eq!(&buf[l4 + 8..len], payload);
    }

    #[test]
    fn test_build_reply_frame_fail_closed() {
        let dgram = make_test_datagram();
        let payload = b"x";

        // 缓冲区不足 → None（绝不截断）
        let mut tiny = [0u8; 41]; // 需要 14+20+8+1 = 43
        assert!(dgram.build_reply_frame(payload, &mut tiny).is_none());

        // IPv6 数据报 → None（v1 诚实拒绝：UDP 校验和为强制项）
        let v6 = UdpDatagram {
            src_ip: IpAddr::V6([0u8; 16]),
            dst_ip: IpAddr::V6([0u8; 16]),
            ..make_test_datagram()
        };
        let mut buf = [0u8; 2048];
        assert!(v6.build_reply_frame(payload, &mut buf).is_none());

        // 载荷超 u16 上限 → None
        let huge = vec![0u8; 70000];
        assert!(dgram.build_reply_frame(&huge, &mut vec![0u8; 70100]).is_none());
    }

    #[test]
    fn test_udp_datagram_src_socket_addr() {
        let dgram = make_test_datagram();
        let addr = dgram.src_socket_addr();
        assert_eq!(addr.ip(), std::net::IpAddr::V4(std::net::Ipv4Addr::new(172, 16, 0, 1)));
        assert_eq!(addr.port(), 12345);
    }

    #[test]
    fn test_worker_rx_desc_preallocated() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        worker.start();

        // 验证预分配缓冲区初始状态
        assert_eq!(worker.rx_desc_count, 0);

        // 执行空周期
        let result = worker.process_cycle();
        assert!(result.is_ok());
        assert_eq!(worker.rx_desc_count, 0);
    }

    #[test]
    fn test_worker_quarantine_config() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        assert!(!worker.quarantine_on_error());
        assert!(!worker.quarantine_on_deny());

        worker.set_quarantine_on_error(true);
        worker.set_quarantine_on_deny(true);
        assert!(worker.quarantine_on_error());
        assert!(worker.quarantine_on_deny());
    }

    #[test]
    fn test_worker_id() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::new(42, config, 256, admission).unwrap();
        assert_eq!(worker.id(), 42);
    }

    #[test]
    fn test_worker_state_transitions() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        assert_eq!(worker.state(), WorkerState::Created);

        worker.start();
        assert_eq!(worker.state(), WorkerState::Running);

        worker.stop();
        assert_eq!(worker.state(), WorkerState::Stopped);
    }

    #[test]
    fn test_worker_stats_initial() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::new(1, config, 256, admission).unwrap();
        let stats = worker.stats();
        assert_eq!(stats.rx_packets, 0);
        assert_eq!(stats.tx_packets, 0);
        assert_eq!(stats.rejected_packets, 0);
        assert_eq!(stats.parse_errors, 0);
    }

    #[test]
    fn test_worker_double_start() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        worker.start();
        assert_eq!(worker.state(), WorkerState::Running);

        worker.start();
        assert_eq!(worker.state(), WorkerState::Running);
    }

    #[test]
    fn test_worker_stop_before_start() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let mut worker = Worker::new(1, config, 256, admission).unwrap();
        worker.stop();
        assert_eq!(worker.state(), WorkerState::Stopped);
    }

    #[test]
    fn test_worker_frame_pool_info() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::new(1, config, 256, admission).unwrap();
        let (capacity, allocated, quarantined) = worker.frame_pool_info();
        assert_eq!(capacity, 256);
        assert_eq!(allocated, 0);
        assert_eq!(quarantined, 0);
    }

    #[test]
    fn test_worker_admission_engine_link() {
        let config = create_test_config();
        let mut admission = SourceAdmissionEngine::deny_all();
        admission
            .add_rule(AdmissionRule {
                id: 1,
                src_ip: IpAddr::V4([10, 0, 0, 1]),
                prefix_len: 0,
                src_port: 0,
                dst_port: 0,
                proto: ProtoMatch::Any,
                action: AdmissionAction::Allow,
                enabled: true,
            })
            .unwrap();

        let worker = Worker::new(1, config, 256, admission).unwrap();
        let action = worker.admission().evaluate(
            IpAddr::V4([10, 0, 0, 1]),
            0,
            0,
            6,
        );
        assert_eq!(action, AdmissionAction::Allow);

        let action2 = worker.admission().evaluate(
            IpAddr::V4([10, 0, 0, 2]),
            0,
            0,
            6,
        );
        assert_eq!(action2, AdmissionAction::Deny);
    }

    #[test]
    fn test_worker_state_enum() {
        let states = vec![
            WorkerState::Created,
            WorkerState::Running,
            WorkerState::Stopped,
            WorkerState::Error,
        ];
        for state in states {
            let debug = format!("{:?}", state);
            assert!(!debug.is_empty());
        }
    }

    #[test]
    fn test_worker_with_deny_all_admission() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::deny_all();

        let worker = Worker::new(1, config, 256, admission);
        assert!(worker.is_ok());
    }

    #[test]
    fn test_extract_flow_info_icmp() {
        let config = create_test_config();
        let admission = SourceAdmissionEngine::allow_all();

        let worker = Worker::new(1, config, 256, admission).unwrap();

        let mut data = vec![0u8; 100];
        let offset = ETH_HEADER_LEN;
        data[0..6].copy_from_slice(&[0xAA; 6]);
        data[6..12].copy_from_slice(&[0xBB; 6]);
        data[12] = 0x08;
        data[13] = 0x00;
        data[offset] = 0x45;
        data[offset + 2] = 0x00;
        data[offset + 3] = 28;
        data[offset + 8] = 64;
        data[offset + 9] = ip_proto::ICMP;
        data[offset + 12..offset + 16].copy_from_slice(&[10, 0, 0, 1]);
        data[offset + 16..offset + 20].copy_from_slice(&[10, 0, 0, 2]);

        let parsed = parse_packet(&data).unwrap();
        let (src_ip, src_port, dst_port, protocol) = worker.extract_flow_info(&parsed);

        assert_eq!(src_ip, IpAddr::V4([10, 0, 0, 1]));
        assert_eq!(src_port, 0);
        assert_eq!(dst_port, 0);
        assert_eq!(protocol, 1);
    }
}