zenith-net 0.1.0

//! TCP 状态机实现
//!
//! 完整的 TCP 有限状态机（RFC 793 + RFC 5681）：
//! - 所有 TCP 状态（CLOSED/LISTEN/SYN-SENT/SYN-RECEIVED/ESTABLISHED/FIN-WAIT-1/FIN-WAIT-2/CLOSE-WAIT/CLOSING/TIME-WAIT）
//! - 连接表：预分配固定容量、单 Owner、零堆分配、无锁 O(1) 查找
//! - 序列号管理：滑动窗口、ISN 生成、序号比较
//! - 重传队列：超时重传、快速重传

use crate::error::NetError;
use crate::packet::{IpVersion, TcpHeader};
use crate::source_admission::IpAddr;
use crate::transport::acceptor::BindTable;

/// TCP 状态（完整状态机）
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ConnectionState {
    /// 已关闭（初始态）
    Closed,
    /// 监听中（服务端等待 SYN）
    Listen,
    /// SYN 已发送（客户端等待 SYN+ACK）
    SynSent,
    /// SYN 已接收（服务端等待 ACK）
    SynReceived,
    /// 已建立（数据传输）
    Established,
    /// FIN-WAIT-1（主动关闭，等待 ACK 或 FIN）
    FinWait1,
    /// FIN-WAIT-2（等待对端 FIN）
    FinWait2,
    /// CLOSE-WAIT（被动关闭，等待应用层 close）
    CloseWait,
    /// 正在关闭（双方 FIN）
    Closing,
    /// TIME-WAIT（等待 2MSL）
    TimeWait,
    /// LAST-ACK（等待对端 ACK 我们的 FIN，被动关闭最后阶段）
    LastAck,
}

impl ConnectionState {
    /// 是否为活跃状态（非 CLOSED）
    #[inline]
    pub fn is_active(&self) -> bool {
        !matches!(self, ConnectionState::Closed)
    }

    /// 是否可以接收数据
    #[inline]
    pub fn can_receive_data(&self) -> bool {
        matches!(self, ConnectionState::Established | ConnectionState::CloseWait)
    }

    /// 是否可以发送数据
    #[inline]
    pub fn can_send_data(&self) -> bool {
        matches!(self, ConnectionState::Established)
    }
}

/// TCP 连接四元组键（Copy 类型，零堆分配）
///
/// 格式：{src_ip, src_port, dst_ip, dst_port, ip_version}
/// 使用折叠哈希实现 O(1) 查找
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct ConnectionKey {
    /// 源 IP
    pub src_ip: IpAddr,
    /// 源端口
    pub src_port: u16,
    /// 目标 IP
    pub dst_ip: IpAddr,
    /// 目标端口
    pub dst_port: u16,
    /// IP 版本
    pub ip_version: IpVersion,
}

impl ConnectionKey {
    /// 创建新的连接键
    #[inline]
    pub fn new(
        src_ip: IpAddr,
        src_port: u16,
        dst_ip: IpAddr,
        dst_port: u16,
        ip_version: IpVersion,
    ) -> Self {
        Self {
            src_ip,
            src_port,
            dst_ip,
            dst_port,
            ip_version,
        }
    }

    /// 计算哈希值（折叠哈希，高位异或低位）
    ///
    /// 使用折叠哈希将四元组压缩为 u64，保证哈希分布均匀。
    /// 实现委托给 transport 层共享的 [`crate::transport::hash_flow_tuple`]
    /// （与 UDP 会话表同源，禁止复制实现）。
    #[inline]
    pub fn hash(&self) -> u64 {
        super::hash_flow_tuple(
            &self.src_ip,
            self.src_port,
            &self.dst_ip,
            self.dst_port,
            self.ip_version,
        )
    }

    /// 生成反向连接键（服务端视角）
    #[inline]
    pub fn reversed(&self) -> Self {
        Self {
            src_ip: self.dst_ip,
            src_port: self.dst_port,
            dst_ip: self.src_ip,
            dst_port: self.src_port,
            ip_version: self.ip_version,
        }
    }
}

/// 单个 TCP 连接槽（预分配，零堆分配）
#[derive(Debug, Clone)]
pub struct TcpConnection {
    /// 连接键
    pub key: ConnectionKey,
    /// 连接状态
    pub state: ConnectionState,
    /// 本端发送序号（ISS → 下一个待发送的序号）
    pub snd_nxt: u32,
    /// 发送未确认序号（SND.UNA：最早未被确认的发送序号；RFC 9293 §3.10）
    pub snd_una: u32,
    /// 本端接收序号（IRS → 期望下一个接收的序号）
    pub rcv_nxt: u32,
    /// 发送窗口大小
    pub snd_wnd: u16,
    /// 接收窗口大小
    pub rcv_wnd: u16,
    /// 拥塞窗口
    pub cwnd: u32,
    /// 起始发送序号（ISS）
    pub iss: u32,
    /// 起始接收序号（IRS）
    pub irs: u32,
    /// 最近活动时间戳（用于超时回收）
    pub last_active: u64,
    /// 重传次数
    pub retransmit_count: u8,
    /// 快速恢复标志
    pub in_fast_recovery: bool,
    /// 连接创建时间
    pub created_at: u64,
    /// 各类型定时器句柄（None = 无活跃定时器）
    ///
    /// # 极致性能
    /// 索引由 [`TimerType::as_index`] 确定，长度 = [`TIMER_TYPE_COUNT`]。
    /// 连接拆除时直接用句柄调用 [`TimerWheel::remove_by_handle`]（O(1)），
    /// 避免 [`TimerWheel::remove_timer`] 的 O(SLOT_COUNT) 全表扫描。
    pub timer_handles: [Option<u32>; crate::transport::timer::TIMER_TYPE_COUNT],
    /// 平滑 RTT (微秒), RFC 6298 Jacobson 算法
    pub srtt_us: u64,
    /// RTT 方差 (微秒)
    pub rttvar_us: u64,
    /// 最后 SYN 时间戳 (微秒)
    pub last_syn_ts: u64,
    /// SYN-ACK 到达时间戳 (微秒)
    pub syn_ack_ts: u64,
    /// 是否已获得首个 RTT 样本
    pub has_rtt_sample: bool,
}

impl TcpConnection {
    /// 创建空连接（CLOSED 状态）
    #[inline]
    pub fn empty() -> Self {
        Self {
            key: ConnectionKey::new(IpAddr::V4([0; 4]), 0, IpAddr::V4([0; 4]), 0, IpVersion::V4),
            state: ConnectionState::Closed,
            snd_nxt: 0,
            snd_una: 0,
            rcv_nxt: 0,
            snd_wnd: 0,
            rcv_wnd: 0,
            cwnd: 0,
            iss: 0,
            irs: 0,
            last_active: 0,
            retransmit_count: 0,
            in_fast_recovery: false,
            created_at: 0,
            timer_handles: [None; crate::transport::timer::TIMER_TYPE_COUNT],
            srtt_us: 0,
            rttvar_us: 0,
            last_syn_ts: 0,
            syn_ack_ts: 0,
            has_rtt_sample: false,
        }
    }

    /// 是否为空闲槽（可复用）
    #[inline]
    pub fn is_free(&self) -> bool {
        self.state == ConnectionState::Closed
    }

    /// 是否为 TIME_WAIT 状态（需要等待 2MSL）
    #[inline]
    pub fn is_time_wait(&self) -> bool {
        self.state == ConnectionState::TimeWait
    }

    /// 初始化新连接（设置初始状态）
    #[inline]
    pub fn init(&mut self, key: ConnectionKey, iss: u32, irs: u32, now: u64) {
        self.key = key;
        self.state = ConnectionState::Listen;
        self.iss = iss;
        self.irs = irs;
        self.snd_nxt = iss;
        self.snd_una = iss;
        self.rcv_nxt = irs;
        self.snd_wnd = 0xFFFF;
        self.rcv_wnd = 0xFFFF;
        self.cwnd = 64;
        self.last_active = now;
        self.created_at = now;
        self.retransmit_count = 0;
        self.in_fast_recovery = false;
        // RTT 估计字段重置（防止槽复用时残留旧值）
        self.srtt_us = 0;
        self.rttvar_us = 0;
        self.last_syn_ts = 0;
        self.syn_ack_ts = 0;
        self.has_rtt_sample = false;
    }

    /// 检查序号是否在接收窗口内
    #[inline]
    pub fn is_seq_in_rcv_window(&self, seq: u32) -> bool {
        let diff = seq.wrapping_sub(self.rcv_nxt);
        diff < self.rcv_wnd as u32
    }

    /// 检查序号是否在发送窗口内
    #[inline]
    pub fn is_seq_in_snd_window(&self, seq: u32) -> bool {
        let diff = seq.wrapping_sub(self.snd_nxt);
        diff < self.snd_wnd as u32
    }

    /// 32 位序号严格小于比较（考虑回绕，RFC 793 序号空间）
    ///
    /// `a < b` 当且仅当 `(b - a)` 落在前向半空间 `[1, 2^31)`，
    /// 即 `b` 在 `a` 之前不超过 2^31-1 字节。
    #[inline]
    pub fn seq_lt(a: u32, b: u32) -> bool {
        (b.wrapping_sub(a) as i32) > 0
    }

    /// 32 位序号小于等于比较（考虑回绕）
    #[inline]
    pub fn seq_le(a: u32, b: u32) -> bool {
        (b.wrapping_sub(a) as i32) >= 0
    }

    /// 32 位序号严格大于比较（考虑回绕）
    #[inline]
    pub fn seq_gt(a: u32, b: u32) -> bool {
        (a.wrapping_sub(b) as i32) > 0
    }

    /// 32 位序号大于等于比较（考虑回绕）
    #[inline]
    pub fn seq_ge(a: u32, b: u32) -> bool {
        (a.wrapping_sub(b) as i32) >= 0
    }

    /// RFC 6298 Jacobson RTT 估计
    ///
    /// 首次测量: SRTT = R, RTTVAR = R / 2
    /// 后续更新: RTTVAR = 3/4 * RTTVAR + 1/4 * |SRTT - R|
    ///           SRTT = 7/8 * SRTT + 1/8 * R
    ///
    /// # 参数
    /// * `r_us` - 本次 RTT 测量值（微秒）
    pub fn update_rtt(&mut self, r_us: u64) {
        if !self.has_rtt_sample {
            self.srtt_us = r_us;
            self.rttvar_us = r_us / 2;
            self.has_rtt_sample = true;
        } else {
            let diff = if self.srtt_us > r_us {
                self.srtt_us - r_us
            } else {
                r_us - self.srtt_us
            };
            self.rttvar_us = (self.rttvar_us * 3 + diff) / 4;
            self.srtt_us = (self.srtt_us * 7 + r_us) / 8;
        }
    }
}

/// TCP 统计信息
#[derive(Debug, Clone, Copy, Default)]
pub struct TcpStats {
    /// 总连接数
    pub total_connections: u64,
    /// 活跃连接数
    pub active_connections: u64,
    /// SYN 接收数
    pub syn_received: u64,
    /// SYN+ACK 发送数
    pub syn_ack_sent: u64,
    /// ACK 接收数
    pub ack_received: u64,
    /// RST 接收数
    pub rst_received: u64,
    /// FIN 接收数
    pub fin_received: u64,
    /// 重传次数
    pub retransmits: u64,
    /// 超时回收连接数
    pub timeouts: u64,
    /// 连接超时关闭数
    pub connection_timeouts: u64,
    /// 拒绝连接数
    pub rejected: u64,
    /// 孤儿 FIN 丢弃数（FIN 到达但无对应连接）
    pub orphan_fin_dropped: u64,
}

/// 最大连接表容量（硬上限，启动时确定）
pub const MAX_CONNECTIONS: usize = 65536;

/// 哈希桶数量（必须为 2 的幂）
pub const HASH_BUCKETS: usize = 16384;

/// 连接表（预分配，单 Owner，零堆分配，无锁）
///
/// 使用开放寻址哈希（线性探测）：
/// - 预分配固定容量的连接槽数组
/// - O(1) 平均查找（哈希冲突线性探测）
/// - 单线程访问，无锁设计
/// - 硬上限约束：容量在启动时确定，运行期不可增长
///
/// # 与 [`crate::transport::udp::UdpSessionTable`] 的关系（泛型化评估结论）
/// 两者同为「预分配槽数组 + u32 桶开放寻址 + 线性探测」骨架，但此处
/// **有意保持单态化副本以保内联**，不做泛型化，原因：
/// - 插入语义不同：本表插入时会复用指向 CLOSED 槽的桶（见 [`Self::insert`]），
///   UDP 侧无此逻辑；泛型骨架需引入行为钩子才能统一，反而模糊热路径；
/// - 槽类型、空闲判定（`state == Closed` vs `!active`）、键提取方式均不同，
///   泛型化只能以闭包/trait 钩子参数化，虽静态分发但阻碍编译期内联优化；
/// - 桶数常量不同（[`HASH_BUCKETS`] = 16384 vs UDP 侧 8192）。
///
/// 两侧共享的部分（IP 折叠哈希、五元组哈希）已提取至
/// [`crate::transport::hash_flow_tuple`]。
#[derive(Debug)]
pub struct ConnectionTable {
    /// 连接槽数组（预分配）
    slots: Vec<TcpConnection>,
    /// 桶索引数组（指向 slots 索引，或 0 表示空）
    /// 使用 u32 存储，0 表示空槽（因为连接索引从 1 开始）
    buckets: Vec<u32>,
    /// 空闲槽索引栈（O(1) 分配/回收）
    ///
    /// 初始化时逆序填入 0..capacity，pop() 时按 0, 1, 2, ... 顺序返回。
    /// insert 时 pop 取用，remove/scan_timeouts 时 push 回收。
    /// 状态机直接置 Closed 的槽不在栈中，由 find_free_slot 回退线性扫描兜底。
    free_stack: Vec<u32>,
    /// 已使用连接数
    count: usize,
    /// 最大容量
    capacity: usize,
}

impl ConnectionTable {
    /// 创建连接表（预分配）
    ///
    /// # 参数
    /// * `capacity` - 最大连接数（硬上限）
    pub fn new(capacity: usize) -> Self {
        let cap = capacity.clamp(1, MAX_CONNECTIONS);
        let mut slots = Vec::with_capacity(cap);
        for _ in 0..cap {
            slots.push(TcpConnection::empty());
        }
        let buckets = vec![0u32; HASH_BUCKETS];
        // 空闲槽栈：逆序填入，pop() 时按 0, 1, 2, ... 顺序返回
        let mut free_stack = Vec::with_capacity(cap);
        for i in (0..cap).rev() {
            free_stack.push(i as u32);
        }
        Self {
            slots,
            buckets,
            free_stack,
            count: 0,
            capacity: cap,
        }
    }

    /// 获取当前连接数
    #[inline]
    pub fn count(&self) -> usize {
        self.count
    }

    /// 获取最大容量
    #[inline]
    pub fn capacity(&self) -> usize {
        self.capacity
    }

    /// 查找连接（O(1) 平均）
    ///
    /// # 参数
    /// * `key` - 连接键
    ///
    /// # 返回
    /// * `Option<usize>` - 连接槽索引
    #[inline]
    pub fn find(&self, key: &ConnectionKey) -> Option<usize> {
        let bucket_idx = (key.hash() as usize) & (HASH_BUCKETS - 1);
        let mut i = bucket_idx;
        let mut probe = 0;
        loop {
            let raw = self.buckets[i];
            if raw == 0 {
                return None;
            }
            if raw == 0xFFFF_FFFFu32 {
                // tombstone: continue probing
                probe += 1;
                if probe >= HASH_BUCKETS {
                    return None;
                }
                i = (i + 1) & (HASH_BUCKETS - 1);
                continue;
            }
            let real_idx = (raw as usize) - 1;
            let slot = &self.slots[real_idx];
            if slot.state != ConnectionState::Closed && slot.key == *key {
                return Some(real_idx);
            }
            probe += 1;
            if probe >= HASH_BUCKETS {
                return None;
            }
            i = (i + 1) & (HASH_BUCKETS - 1);
        }
    }

    /// 获取可变连接引用
    #[inline]
    pub fn get_mut(&mut self, idx: usize) -> Option<&mut TcpConnection> {
        if idx < self.slots.len() {
            Some(&mut self.slots[idx])
        } else {
            None
        }
    }

    /// 获取连接引用
    #[inline]
    pub fn get(&self, idx: usize) -> Option<&TcpConnection> {
        if idx < self.slots.len() {
            Some(&self.slots[idx])
        } else {
            None
        }
    }

    /// 插入新连接
    ///
    /// # 参数
    /// * `key` - 连接键
    /// * `iss` - 初始发送序号
    /// * `irs` - 初始接收序号
    /// * `now` - 当前时间戳
    ///
    /// # 返回
    /// * `Result<usize, NetError>` - 连接槽索引
    pub fn insert(
        &mut self,
        key: ConnectionKey,
        iss: u32,
        irs: u32,
        now: u64,
    ) -> Result<usize, NetError> {
        // 检查容量
        if self.count >= self.capacity {
            return Err(NetError::ConnectionTableFull {
                capacity: self.capacity,
            });
        }

        // 检查是否已存在
        if self.find(&key).is_some() {
            return Err(NetError::ConnectionExists {
                key: format!("{:?}", key),
            });
        }

        // 查找空闲槽（fail-closed：满表返回错误，绝不静默复用槽 0）
        let slot_idx = self.find_free_slot().ok_or(NetError::ConnectionTableFull {
            capacity: self.capacity,
        })?;

        // 插入哈希表（线性探测，与 find 保持一致）
        let hash = key.hash();
        let bucket_idx = (hash as usize) & (HASH_BUCKETS - 1);
        let mut i = bucket_idx;
        let mut probe = 0;
        loop {
            let existing_slot_idx = self.buckets[i];
            if existing_slot_idx == 0 || existing_slot_idx == 0xFFFF_FFFFu32 {
                // 空桶或 tombstone：直接插入
                self.buckets[i] = (slot_idx as u32) + 1;
                break;
            }
            // 检查是否是已删除的槽（槽存在但 CLOSED）
            // 桶中存储的是 (slot_idx + 1)，需减 1 还原为实际索引（与 find 一致）
            let existing_slot = &self.slots[(existing_slot_idx as usize) - 1];
            if existing_slot.state == ConnectionState::Closed {
                // 复用此桶
                self.buckets[i] = (slot_idx as u32) + 1;
                break;
            }
            // 继续探测
            probe += 1;
            if probe >= HASH_BUCKETS {
                return Err(NetError::HashTableFull);
            }
            i = (i + 1) & (HASH_BUCKETS - 1);
        }

        // 初始化连接
        self.slots[slot_idx].init(key, iss, irs, now);
        self.count += 1;

        Ok(slot_idx)
    }

    /// 查找空闲槽（O(1) 空闲栈弹出，回退线性扫描兜底）
    ///
    /// 优先从 `free_stack` 弹出（O(1)），栈空时回退线性扫描，
    /// 以回收状态机直接置 Closed 但未入栈的槽。
    ///
    /// # 返回
    /// * `Some(idx)` - 空闲槽索引
    /// * `None` - 表已满（调用方须 fail-closed，禁止静默复用槽 0）
    #[inline]
    fn find_free_slot(&mut self) -> Option<usize> {
        // O(1)：优先从空闲栈弹出
        if let Some(idx) = self.free_stack.pop() {
            return Some(idx as usize);
        }
        // 回退：线性扫描（处理状态机直接置 Closed 但未入栈的槽）
        (0..self.capacity).find(|&i| self.slots[i].is_free())
    }

    /// 移除连接（标记为 CLOSED）
    ///
    /// # 参数
    /// * `key` - 连接键
    ///
    /// # 返回
    /// * `bool` - 是否成功移除
    pub fn remove(&mut self, key: &ConnectionKey) -> bool {
        if let Some(idx) = self.find(key) {
            // 标记为 CLOSED
            self.slots[idx].state = ConnectionState::Closed;
            // 回收空闲槽索引到栈中（供下次 insert O(1) 取用）
            self.free_stack.push(idx as u32);

            // 从哈希桶中清除（需要找到并清除对应的桶）
            let bucket_idx = (key.hash() as usize) & (HASH_BUCKETS - 1);
            let mut i = bucket_idx;
            let mut probe = 0;
            loop {
                let slot_idx = self.buckets[i];
                if slot_idx == 0 {
                    break;
                }
                if slot_idx as usize == idx + 1 {
                    self.buckets[i] = 0xFFFF_FFFFu32; // tombstone
                    break;
                }
                probe += 1;
                if probe >= HASH_BUCKETS {
                    break;
                }
                i = (i + 1) & (HASH_BUCKETS - 1);
            }

            self.count -= 1;
            true
        } else {
            false
        }
    }

    /// 关闭指定索引处的连接并完成账本回收（O(1)）
    ///
    /// 这是连接拆除热路径的统一入口：负责将槽位置为 `Closed`、
    /// 递减 `count`、把槽索引入空闲栈、并将哈希桶标记为 tombstone。
    /// 之前的实现（RST / 定时器超时路径）仅置 `Closed` 却不递减 `count`、
    /// 不入空闲栈，导致表"假满"、新 SYN 被拒（慢速 DoS）。
    ///
    /// # 参数
    /// * `idx` - 连接槽索引
    ///
    /// # 返回
    /// * `bool` - 是否实际关闭了一个活跃连接（已关闭则返回 `false`，避免重复计数）
    pub fn close_by_index(&mut self, idx: usize) -> bool {
        if idx >= self.slots.len() || !self.slots[idx].state.is_active() {
            return false;
        }
        let key = self.slots[idx].key;
        self.slots[idx].state = ConnectionState::Closed;
        self.free_stack.push(idx as u32);
        self.count = self.count.saturating_sub(1);

        // 将哈希桶标记为 tombstone，避免残留桶指向已关闭槽
        let bucket_idx = (key.hash() as usize) & (HASH_BUCKETS - 1);
        let mut i = bucket_idx;
        let mut probe = 0;
        loop {
            let slot_idx = self.buckets[i];
            if slot_idx == 0 {
                break;
            }
            if slot_idx as usize == idx + 1 {
                self.buckets[i] = 0xFFFF_FFFFu32;
                break;
            }
            probe += 1;
            if probe >= HASH_BUCKETS {
                break;
            }
            i = (i + 1) & (HASH_BUCKETS - 1);
        }
        true
    }

    /// 扫描超时连接（调用者在主循环中定期调用）
    ///
    /// # 参数
    /// * `now` - 当前时间戳
    /// * `timeout_ms` - 超时时间（毫秒）
    ///
    /// # 返回
    /// * 超时的连接键列表
    pub fn scan_timeouts(&mut self, now: u64, timeout_ms: u64) -> Vec<ConnectionKey> {
        let mut timed_out = Vec::new();
        for (i, slot) in self.slots.iter_mut().enumerate() {
            if slot.state.is_active() && now.saturating_sub(slot.last_active) > timeout_ms {
                timed_out.push(slot.key);
                slot.state = ConnectionState::Closed;
                // 回收空闲槽索引到栈中
                self.free_stack.push(i as u32);
            }
        }
        // 重新构建哈希表（简化实现）
        self.rebuild_hash();
        self.count = self.count.saturating_sub(timed_out.len());
        timed_out
    }

    /// 重建哈希表（在扫描超时后调用）
    fn rebuild_hash(&mut self) {
        self.buckets.fill(0);
        for i in 0..self.slots.len() {
            if self.slots[i].state.is_active() {
                let key = self.slots[i].key;
                let bucket_idx = (key.hash() as usize) & (HASH_BUCKETS - 1);
                let mut idx = bucket_idx;
                let mut probe = 0;
                loop {
                    if self.buckets[idx] == 0 {
                        self.buckets[idx] = (i as u32) + 1;
                        break;
                    }
                    probe += 1;
                    if probe >= HASH_BUCKETS {
                        break;
                    }
                    idx = (idx + 1) & (HASH_BUCKETS - 1);
                }
            }
        }
    }

    /// 遍历活跃连接
    pub fn iter_active(&self) -> impl Iterator<Item = &TcpConnection> {
        self.slots.iter().filter(|s| s.state.is_active())
    }

    /// 遍历活跃连接（可变）
    pub fn iter_active_mut(&mut self) -> impl Iterator<Item = &mut TcpConnection> {
        self.slots.iter_mut().filter(|s| s.state.is_active())
    }

    /// 获取底层槽数组的迭代器（用于批量处理）
    pub fn slots(&self) -> &[TcpConnection] {
        &self.slots
    }
}

/// TCP 状态机（核心逻辑）
///
/// 实现完整的 TCP 有限状态机转换逻辑：
/// - 基于 RFC 793 的状态转换
/// - 序号验证与窗口管理
/// - SYN flood 防护
/// - 快速重传与拥塞控制
/// - 定时器驱动的超时与重传
#[derive(Debug)]
pub struct TcpStateMachine {
    /// 连接表
    table: ConnectionTable,
    /// 统计信息
    stats: TcpStats,
    /// 时间轮定时器
    timer_wheel: crate::transport::timer::TimerWheel,
    /// 监听端点注册表（NET-020：被动打开监听校验）
    ///
    /// `Some(table)` 时，`handle_new_syn` 会校验目标 (dst_ip, dst_port) 是否处于
    /// LISTEN 状态，非监听端口一律丢弃 SYN（fail-closed）。`None` 表示状态机作为
    /// 独立参考实现运行、未接入数据面，保持兼容行为（不拦截）。
    bind_table: Option<BindTable>,
}

impl TcpStateMachine {
    /// 创建 TCP 状态机
    ///
    /// # 参数
    /// * `max_connections` - 最大连接数
    pub fn new(max_connections: usize) -> Self {
        Self {
            table: ConnectionTable::new(max_connections),
            stats: TcpStats::default(),
            timer_wheel: crate::transport::timer::TimerWheel::new(1),
            bind_table: None,
        }
    }

    /// 挂载监听端点注册表（NET-020）
    ///
    /// 挂载后，被动打开（SYN 处理）会强制校验目标端口处于 LISTEN 状态，
    /// 非监听端口丢弃 SYN。集成点：AF_XDP Worker 注册监听端口后调用。
    #[inline]
    pub fn attach_bind_table(&mut self, table: BindTable) {
        self.bind_table = Some(table);
    }

    /// 获取监听端点注册表引用
    #[inline]
    pub fn bind_table(&self) -> Option<&BindTable> {
        self.bind_table.as_ref()
    }

    /// 获取监听端点注册表可变引用
    #[inline]
    pub fn bind_table_mut(&mut self) -> Option<&mut BindTable> {
        self.bind_table.as_mut()
    }

    /// 获取时间轮引用
    #[inline]
    pub fn timer_wheel(&self) -> &crate::transport::timer::TimerWheel {
        &self.timer_wheel
    }

    /// 获取时间轮可变引用
    #[inline]
    pub fn timer_wheel_mut(&mut self) -> &mut crate::transport::timer::TimerWheel {
        &mut self.timer_wheel
    }

    /// 获取连接表不可变引用
    #[inline]
    pub fn table(&self) -> &ConnectionTable {
        &self.table
    }

    /// 获取连接表可变引用
    #[inline]
    pub fn table_mut(&mut self) -> &mut ConnectionTable {
        &mut self.table
    }

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

    /// 定时器驱动的超时处理
    ///
    /// 推进时间轮并处理过期定时器：
    /// - TcpConnectionTimeout: 关闭处于 SYN-RECEIVED 的连接
    /// - TcpRetransmit: 触发重传计数增加
    /// - TcpTimeWait: 从 TIME-WAIT 状态转为 CLOSED
    /// - TcpFinWait2: 从 FIN-WAIT-2 转为 CLOSED
    /// - UdpSessionTimeout: UDP 会话超时（转发到 UDP 处理）
    ///
    /// # Arguments
    /// * `elapsed_ms` - 经过的毫秒数
    ///
    /// # Returns
    /// * `Vec<TimerAction>` - 需要上层处理的动作列表
    pub fn run_tick(&mut self, elapsed_ms: u64) -> Vec<crate::transport::timer::TimerAction> {
        use crate::transport::timer::TimerType;

        let expired = self.timer_wheel.advance(elapsed_ms);

        for action in &expired {
            match action.timer_type {
                TimerType::TcpConnectionTimeout => {
                    // 关闭连接（仅对活跃连接生效，完成 count/空闲栈/哈希桶回收）
                    if self.table.close_by_index(action.target_idx) {
                        self.stats.connection_timeouts += 1;
                        self.stats.active_connections =
                            self.stats.active_connections.saturating_sub(1);
                    }
                }
                TimerType::TcpRetransmit => {
                    if let Some(conn) = self.table.get_mut(action.target_idx) {
                        conn.retransmit_count = conn.retransmit_count.saturating_add(1);
                        conn.cwnd = conn.cwnd.saturating_mul(2).max(1);
                        self.stats.retransmits += 1;
                    }
                }
                TimerType::TcpTimeWait => {
                    // 从 TIME-WAIT 关闭
                    if self.table.close_by_index(action.target_idx) {
                        self.stats.active_connections =
                            self.stats.active_connections.saturating_sub(1);
                    }
                }
                TimerType::TcpFinWait2 => {
                    if self.table.close_by_index(action.target_idx) {
                        self.stats.active_connections =
                            self.stats.active_connections.saturating_sub(1);
                    }
                }
                TimerType::UdpSessionTimeout => {
                    // UDP 会话超时 - UDP 层处理
                }
                TimerType::QuicPto | TimerType::QuicCloseTimeout | TimerType::QuicHandshakeTimeout => {
                    // QUIC 定时器 - QUIC 层处理
                }
            }
        }

        // SmallVec → Vec（pub API 兼容；≤16 个过期定时器时源数据本就在栈上）
        expired.to_vec()
    }

    /// 为指定连接添加定时器
    ///
    /// # 极致性能
    /// 返回的句柄会存入 `TcpConnection::timer_handles`，后续 `remove_timers_for`
    /// 可直接用句柄调用 O(1) 的 `remove_by_handle`，避免全表扫描。
    ///
    /// # Arguments
    /// * `duration_ms` - 延迟毫秒
    /// * `timer_type` - 定时器类型
    /// * `connection_idx` - 连接索引
    pub fn add_timer(
        &mut self,
        duration_ms: u64,
        timer_type: crate::transport::timer::TimerType,
        connection_idx: usize,
    ) -> Option<u32> {
        let handle = self.timer_wheel.add_timer(duration_ms, timer_type, connection_idx)?;
        // 将句柄存入连接，供 remove_timers_for O(1) 删除
        if let Some(conn) = self.table.get_mut(connection_idx) {
            conn.timer_handles[timer_type.as_index()] = Some(handle);
        }
        Some(handle)
    }

    /// 移除指定连接的所有定时器
    ///
    /// # 极致性能：O(TIMER_TYPE_COUNT) = O(8)
    /// 通过 `timer_handles` 中存储的句柄直接调用 `remove_by_handle`（O(1)），
    /// 而非 `remove_timer` 的 O(SLOT_COUNT × 链长度) 全表扫描。
    /// 连接拆除热路径的关键优化。
    pub fn remove_timers_for(&mut self, connection_idx: usize) {
        if let Some(conn) = self.table.get_mut(connection_idx) {
            for slot in conn.timer_handles.iter_mut() {
                if let Some(h) = slot.take() {
                    let _ = self.timer_wheel.remove_by_handle(h);
                }
            }
        }
    }

    /// 生成初始发送序号（ISN）
    ///
    /// RFC 6528 合规实现：熵源为操作系统 CSPRNG
    /// （[`zenith_foundation::random::random_u64`]），并叠加连接四元组折叠哈希作为
    /// per-connection 偏移，保证不同连接的 ISN 互不相关且攻击者不可预测。
    ///
    /// # 安全说明
    /// 历史实现为固定种子（`0xDEADBEEF`）LCG，ISN 完全可预测，
    /// 违反 RFC 6528，已移除（含 LCG 状态字段与固定种子）。
    #[inline]
    fn generate_isn(&self, key: &ConnectionKey) -> u32 {
        // 熵源必须是 CSPRNG；四元组哈希仅提供 per-connection 偏移
        (zenith_foundation::random::random_u64() as u32) ^ (key.hash() as u32)
    }

    /// 处理入站 TCP 包
    ///
    /// # 参数
    /// * `key` - 连接四元组键
    /// * `tcp` - TCP 头
    /// * `payload_len` - TCP 载荷长度（字节）
    /// * `now` - 当前时间戳
    ///
    /// # 返回
    /// * `TcpAction` - 应对的动作
    pub fn handle_packet(
        &mut self,
        key: &ConnectionKey,
        tcp: &TcpHeader,
        payload_len: usize,
        now: u64,
    ) -> TcpAction {
        // 1. 查找现有连接
        if let Some(idx) = self.table.find(key) {
            return self.handle_existing_connection(idx, key, tcp, payload_len, now);
        }

        // 2. 检查是否有反向连接键（服务端监听）
        let reversed = key.reversed();
        if let Some(idx) = self.table.find(&reversed) {
            return self.handle_server_side(idx, key, tcp, payload_len, now);
        }

        // 3. 新连接：处理 SYN
        if tcp.syn() && !tcp.ack() {
            return self.handle_new_syn(key, tcp, now);
        }

        // 4. 孤儿 FIN：记录并丢弃（不回 RST，防止 FIN 扫描放大）
        if tcp.fin() {
            self.stats.orphan_fin_dropped += 1;
            self.stats.rejected += 1;
            return TcpAction::Drop;
        }

        // 5. 无匹配连接，发送 RST
        self.stats.rejected += 1;
        TcpAction::SendRst
    }

    /// 处理新 SYN 包（服务端接收）
    fn handle_new_syn(
        &mut self,
        key: &ConnectionKey,
        tcp: &TcpHeader,
        now: u64,
    ) -> TcpAction {
        // NET-020：被动打开校验（fail-closed）。
        // 若已挂载 BindTable，则目标 (dst_ip, dst_port) 必须处于 LISTEN 状态，
        // 否则丢弃 SYN（非监听端口建连拒绝）。未挂载时保持兼容行为。
        if let Some(bind) = &self.bind_table {
            let dst = crate::NetAddr::from_ip_addr(key.dst_ip, key.dst_port);
            if !bind.is_listening(dst) {
                self.stats.rejected += 1;
                return TcpAction::Drop;
            }
        }

        // 检查连接表容量（SYN flood 防护）
        if self.table.count() >= self.table.capacity() {
            self.stats.rejected += 1;
            return TcpAction::Drop;
        }

        // 以反向键存储（服务端视角：server→client）
        let server_key = key.reversed();
        let iss = self.generate_isn(&server_key);
        let irs = tcp.seq_num().wrapping_add(1);

        match self.table.insert(server_key, iss, irs, now) {
            Ok(idx) => {
                // 进入 SYN-RECEIVED 状态
                if let Some(conn) = self.table.get_mut(idx) {
                    conn.state = ConnectionState::SynReceived;
                    conn.snd_nxt = iss.wrapping_add(1); // SYN 消耗 1 序号
                    // 服务端接收窗口保留自身通告值（init 默认 0xFFFF），
                    // 不得用客户端广告窗口覆盖（否则客户端通告 0 窗口时
                    // 本端接收窗口为 0，NET-021 窗口校验会拒绝合法 FIN/RST）。
                    conn.last_active = now;
                    // 记录 SYN 到达时间，用于后续 RTT 估计
                    conn.last_syn_ts = now;
                }
                self.stats.syn_received += 1;
                self.stats.active_connections += 1;
                self.stats.total_connections += 1;

                TcpAction::SendSynAck {
                    syn_seq: iss,
                    ack_seq: irs,
                    window: 0xFFFF,
                }
            }
            Err(_) => {
                self.stats.rejected += 1;
                TcpAction::Drop
            }
        }
    }

    /// 处理服务端已建立的连接
    fn handle_server_side(
        &mut self,
        idx: usize,
        _key: &ConnectionKey,
        tcp: &TcpHeader,
        payload_len: usize,
        now: u64,
    ) -> TcpAction {
        // RST 处理（在借用连接前处理，以便 close_by_index 完成 count/空闲栈/哈希桶回收）
        // NET-021：off-path RST 防护——仅当 SEQ 落在接收窗口内才关闭连接，
        // 否则丢弃（防伪造 RST 强制断连）。
        let rst_acceptable = match self.table.get(idx) {
            Some(c) => c.is_seq_in_rcv_window(tcp.seq_num()),
            None => false,
        };
        if tcp.rst() {
            if rst_acceptable {
                self.stats.rst_received += 1;
                if self.table.close_by_index(idx) {
                    self.stats.active_connections =
                        self.stats.active_connections.saturating_sub(1);
                }
            }
            return TcpAction::Drop;
        }

        // LastAck → Closed：收到确认我们 FIN 的 ACK 后关闭（服务端主动关闭最后阶段）。
        // 在借用前检测，以便 close_by_index 完成账本回收。
        let last_ack_closed = match self.table.get(idx) {
            Some(c) => {
                c.state == ConnectionState::LastAck
                    && tcp.ack()
                    && TcpConnection::seq_ge(c.snd_una, c.snd_nxt)
            }
            None => false,
        };
        if last_ack_closed {
            if self.table.close_by_index(idx) {
                self.stats.active_connections = self.stats.active_connections.saturating_sub(1);
            }
            return TcpAction::Drop;
        }

        let conn = match self.table.get_mut(idx) {
            Some(c) => c,
            None => return TcpAction::Drop,
        };

        // 更新活跃时间
        conn.last_active = now;

        // ACK 处理（RFC 9293 §3.10：先验证 ACK 合法性，非法 ACK 丢弃整段，再处理 FIN/数据）
        // SynReceived 状态例外：三次握手的最后一个 ACK 是握手完成段，
        // 其 ack_num 应确认服务端的 SYN（snd_nxt = ISN+1），
        // 但测试/实际实现中 ack_num 可能尚未对齐，状态转换优先于 ACK 验证
        if tcp.ack() && conn.state != ConnectionState::SynReceived {
            self.stats.ack_received += 1;
            let ack_num = tcp.ack_num();

            // 1) ack_num > snd_nxt：确认了尚未发送的数据 → 非法 ACK。
            //    发送 challenge ACK 并丢弃该报文段（不推进 snd_nxt/snd_una，不处理 FIN/数据）。
            if TcpConnection::seq_gt(ack_num, conn.snd_nxt) {
                return TcpAction::Ack {
                    ack_seq: conn.rcv_nxt,
                    window: 0xFFFF,
                };
            }

            // 2) snd_una <= ack_num <= snd_nxt：合法 ACK。
            //    仅当 ack_num > snd_una（新数据被确认）时才推进累积确认指针 snd_una
            //    并相应推进 snd_nxt；ack_num == snd_una 为重复 ACK，接受但不推进。
            if TcpConnection::seq_gt(ack_num, conn.snd_una) {
                conn.snd_una = ack_num;
                conn.snd_nxt = ack_num;
            }
            // 3) ack_num < snd_una：旧的重传 ACK，忽略确认推进，但仍继续处理其它字段。

            // 更新发送窗口（合法 ACK 与旧重复 ACK 均处理窗口更新）
            conn.snd_wnd = tcp.window_size();
        }

        // FIN 处理
        // NET-021：off-path FIN 防护——仅当 SEQ 落在接收窗口内才触发状态迁移，
        // 否则保持 CloseWait/状态不变（防伪造 FIN 强制断连）。
        if tcp.fin() && conn.is_seq_in_rcv_window(tcp.seq_num()) {
            conn.rcv_nxt = conn.rcv_nxt.wrapping_add(1);
            conn.state = ConnectionState::CloseWait;
            self.stats.fin_received += 1;
        }

        // 数据处理
        if payload_len > 0 && conn.state.can_receive_data() {
            let next_seq = tcp.seq_num().wrapping_add(payload_len as u32);
            if TcpConnection::seq_gt(next_seq, conn.rcv_nxt) {
                conn.rcv_nxt = next_seq;
            }
        }

        // 状态转换
        

        match conn.state {
            ConnectionState::SynReceived => {
                // 验证三次握手最终 ACK 的合法性：ack_num 必须落在
                // (snd_una, snd_nxt] 区间内（RFC 9293 §3.10），
                // 防止攻击者用任意 ack_num 的 ACK 完成握手
                if tcp.ack()
                    && TcpConnection::seq_gt(tcp.ack_num(), conn.snd_una)
                    && TcpConnection::seq_le(tcp.ack_num(), conn.snd_nxt)
                {
                    conn.state = ConnectionState::Established;
                    conn.snd_una = tcp.ack_num();
                    // RFC 6298: 用 SYN 到 ACK 的往返时间作为首个 RTT 样本
                    let r = now.saturating_sub(conn.last_syn_ts);
                    conn.update_rtt(r);
                    self.stats.ack_received += 1;
                    TcpAction::Established
                } else if tcp.ack() {
                    // 非法 ACK：发送 challenge ACK，不完成握手
                    TcpAction::Ack {
                        ack_seq: conn.rcv_nxt,
                        window: 0xFFFF,
                    }
                } else {
                    TcpAction::Drop
                }
            }
            ConnectionState::Established => TcpAction::Ack {
                ack_seq: conn.rcv_nxt,
                window: 0xFFFF,
            },
            ConnectionState::CloseWait => TcpAction::Close,
            _ => TcpAction::Drop,
        }
    }

    /// 处理客户端已建立的连接
    fn handle_existing_connection(
        &mut self,
        idx: usize,
        _key: &ConnectionKey,
        tcp: &TcpHeader,
        payload_len: usize,
        now: u64,
    ) -> TcpAction {
        // RST 处理（在借用连接前处理，以便 close_by_index 完成 count/空闲栈/哈希桶回收）
        // NET-021：off-path RST 防护——仅当 SEQ 落在接收窗口内才关闭连接。
        let rst_acceptable = match self.table.get(idx) {
            Some(c) => c.is_seq_in_rcv_window(tcp.seq_num()),
            None => false,
        };
        if tcp.rst() {
            if rst_acceptable {
                self.stats.rst_received += 1;
                if self.table.close_by_index(idx) {
                    self.stats.active_connections =
                        self.stats.active_connections.saturating_sub(1);
                }
            }
            return TcpAction::Drop;
        }

        // LastAck → Closed：收到确认我们 FIN 的 ACK 后关闭（服务端主动关闭最后阶段）。
        // 在借用前检测，以便 close_by_index 完成账本回收。
        let last_ack_closed = match self.table.get(idx) {
            Some(c) => {
                c.state == ConnectionState::LastAck
                    && tcp.ack()
                    && TcpConnection::seq_ge(c.snd_una, c.snd_nxt)
            }
            None => false,
        };
        if last_ack_closed {
            if self.table.close_by_index(idx) {
                self.stats.active_connections = self.stats.active_connections.saturating_sub(1);
            }
            return TcpAction::Drop;
        }

        let conn = match self.table.get_mut(idx) {
            Some(c) => c,
            None => return TcpAction::Drop,
        };

        conn.last_active = now;

        // FIN 处理
        // NET-021：off-path FIN 防护——仅当 SEQ 落在接收窗口内才触发状态迁移，
        // 否则保持不变（防伪造 FIN 强制断连）。
        if tcp.fin() && conn.is_seq_in_rcv_window(tcp.seq_num()) {
            conn.rcv_nxt = conn.rcv_nxt.wrapping_add(1);
            conn.state = match conn.state {
                ConnectionState::FinWait1 => ConnectionState::Closing,
                ConnectionState::FinWait2 => ConnectionState::TimeWait,
                _ => ConnectionState::CloseWait,
            };
            self.stats.fin_received += 1;
        }

        // ACK 处理（RFC 9293 §3.10：验证 ACK 合法性，与 handle_server_side 一致）
        // SynReceived 状态例外：三次握手的最后一个 ACK 是握手完成段，
        // 其 ack_num 必须确认服务端的 SYN（snd_nxt = ISN+1），
        // 通用 ACK 验证会提前推进 snd_una/snd_nxt 导致握手 ACK 校验失效，
        // 因此跳过通用验证，由下方状态机分支单独校验（与 handle_server_side 一致）。
        // SynSent 状态例外（NET-028）：SYN-ACK 的 ack_num 必须等于 ISN+1，
        // 同样由下方状态机分支单独校验（通用 ACK 验证会以 ack_num>snd_nxt 提前拦截）。
        if tcp.ack()
            && !matches!(
                conn.state,
                ConnectionState::SynReceived | ConnectionState::SynSent
            )
        {
            self.stats.ack_received += 1;
            let ack_num = tcp.ack_num();

            // 1) ack_num > snd_nxt：确认了尚未发送的数据 → 非法 ACK。
            //    发送 challenge ACK 并丢弃该报文段（不推进 snd_una/snd_nxt）。
            if TcpConnection::seq_gt(ack_num, conn.snd_nxt) {
                return TcpAction::Ack {
                    ack_seq: conn.rcv_nxt,
                    window: 0xFFFF,
                };
            }

            // 2) snd_una < ack_num <= snd_nxt：合法 ACK，推进累积确认指针。
            if TcpConnection::seq_gt(ack_num, conn.snd_una) {
                conn.snd_una = ack_num;
                conn.snd_nxt = ack_num;
            }
            // 3) ack_num <= snd_una：旧的重传 ACK，忽略确认推进。

            // 更新发送窗口（合法 ACK 与旧重复 ACK 均处理窗口更新）
            conn.snd_wnd = tcp.window_size();
        }

        // 数据处理
        if payload_len > 0 && conn.state.can_receive_data() {
            let next_seq = tcp.seq_num().wrapping_add(payload_len as u32);
            if TcpConnection::seq_gt(next_seq, conn.rcv_nxt) {
                conn.rcv_nxt = next_seq;
            }
        }

        match conn.state {
            ConnectionState::SynSent => {
                // NET-028：SYN-ACK 必须确认我们发送的 SYN——ack_num 必须等于 ISN+1
                //（snd_nxt = ISN）。原实现未校验 ISN，攻击者可伪造任意 ack_num
                // 完成握手（规避 SYN 校验）。校验失败 fail-closed 丢弃。
                if tcp.syn()
                    && tcp.ack()
                    && tcp.ack_num() == conn.snd_nxt.wrapping_add(1)
                {
                    conn.state = ConnectionState::Established;
                    conn.rcv_nxt = tcp.seq_num().wrapping_add(1);
                    conn.snd_nxt = tcp.ack_num();
                    // RFC 6298: 用 SYN 到 SYN-ACK 的往返时间作为首个 RTT 样本
                    let r = now.saturating_sub(conn.last_syn_ts);
                    conn.update_rtt(r);
                    self.stats.ack_received += 1;
                    TcpAction::Established
                } else {
                    // 非法 SYN-ACK（ack_num != ISN+1）：丢弃（fail-closed）
                    TcpAction::Drop
                }
            }
            // RFC 9293 §3.10：SynReceived 收到 ACK → 需验证 ack_num 确认了我们的 SYN
            // snd_una = ISN_server，snd_nxt = ISN_server + 1（SYN-ACK 已发送）
            // 合法 ACK 要求 snd_una < ack_num <= snd_nxt，即 ack_num == snd_nxt（确认 SYN）
            // 通用 ACK 验证已对 SynReceived 跳过，此处为唯一校验点（与 handle_server_side 一致）
            ConnectionState::SynReceived => {
                if tcp.ack() {
                    let ack_num = tcp.ack_num();
                    // RFC 9293 §3.10: ACK must acknowledge our SYN
                    // Acceptable ACK: snd_una < ack_num <= snd_nxt
                    if TcpConnection::seq_gt(ack_num, conn.snd_una)
                        && TcpConnection::seq_le(ack_num, conn.snd_nxt)
                    {
                        conn.state = ConnectionState::Established;
                        // RFC 6298: 用 SYN 到 ACK 的往返时间作为 RTT 样本
                        let r = now.saturating_sub(conn.last_syn_ts);
                        conn.update_rtt(r);
                        self.stats.ack_received += 1;
                        // 推进累积确认指针
                        conn.snd_una = ack_num;
                        TcpAction::Established
                    } else {
                        // 非法 ACK：ack_num 不在 (snd_una, snd_nxt] 范围内，丢弃（fail-closed）
                        TcpAction::Drop
                    }
                } else {
                    TcpAction::Drop
                }
            }
            ConnectionState::Established => TcpAction::Ack {
                ack_seq: conn.rcv_nxt,
                window: 0xFFFF,
            },
            ConnectionState::FinWait1 | ConnectionState::FinWait2 => {
                // FIN_WAIT_1 → FIN_WAIT_2: 当我们的 FIN 被 ACK 确认后转换
                if conn.state == ConnectionState::FinWait1
                    && tcp.ack()
                    && TcpConnection::seq_ge(conn.snd_una, conn.snd_nxt.wrapping_sub(1))
                {
                    conn.state = ConnectionState::FinWait2;
                }
                TcpAction::Ack {
                    ack_seq: conn.rcv_nxt,
                    window: 0xFFFF,
                }
            }
            // CLOSING → TIME_WAIT: 收到 ACK 确认我们的 FIN 后转换
            ConnectionState::Closing => {
                if tcp.ack()
                    && TcpConnection::seq_ge(conn.snd_una, conn.snd_nxt.wrapping_sub(1))
                {
                    conn.state = ConnectionState::TimeWait;
                }
                TcpAction::Ack {
                    ack_seq: conn.rcv_nxt,
                    window: 0xFFFF,
                }
            }
            // LAST_ACK → Closed 已在函数顶部统一处理（close_by_index 完成账本回收），
            // 此处仅兜底未满足关闭条件的报文（非 ACK / ACK 未确认我们的 FIN），直接丢弃。
            ConnectionState::LastAck => TcpAction::Drop,
            _ => TcpAction::Drop,
        }
    }

    /// 发送 SYN（客户端主动连接）
    ///
    /// # 参数
    /// * `key` - 连接键
    /// * `now` - 当前时间戳
    ///
    /// # 返回
    /// * `Result<u32, NetError>` - 连接槽索引
    pub fn send_syn(
        &mut self,
        key: ConnectionKey,
        now: u64,
    ) -> Result<usize, NetError> {
        let iss = self.generate_isn(&key);
        let irs = 0; // 客户端 IRS 在收到 SYN+ACK 后确定
        let idx = self.table.insert(key, iss, irs, now)?;
        if let Some(conn) = self.table.get_mut(idx) {
            conn.state = ConnectionState::SynSent;
            conn.snd_nxt = iss;
            conn.last_active = now;
            // 记录 SYN 发送时间，用于后续 RTT 估计
            conn.last_syn_ts = now;
        }
        self.stats.active_connections += 1;
        self.stats.total_connections += 1;
        Ok(idx)
    }

    /// 主动关闭连接
    ///
    /// # 参数
    /// * `key` - 连接键
    /// * `now` - 当前时间戳
    ///
    /// # 返回
    /// * `TcpAction` - 应发送的动作
    pub fn close_connection(
        &mut self,
        key: &ConnectionKey,
        now: u64,
    ) -> TcpAction {
        if let Some(idx) = self.table.find(key) {
            // 借用前判定是否可发送 FIN，避免 else 分支借用 `conn` 后无法调用 close_by_index
            let can_send = self.table.get(idx).map(|c| c.state.can_send_data()).unwrap_or(false);
            if can_send {
                if let Some(conn) = self.table.get_mut(idx) {
                    // 发送 FIN
                    conn.state = ConnectionState::FinWait1;
                    conn.snd_nxt = conn.snd_nxt.wrapping_add(1);
                    conn.last_active = now;
                }
                return TcpAction::SendFin;
            } else {
                // 无法发送 FIN：直接关闭并完成 count/空闲栈/哈希桶回收
                if self.table.close_by_index(idx) {
                    self.stats.active_connections = self.stats.active_connections.saturating_sub(1);
                }
                return TcpAction::Drop;
            }
        }
        self.stats.rejected += 1;
        TcpAction::Drop
    }

    /// 被动关闭（CLOSE_WAIT 处理）
    ///
    /// # 参数
    /// * `key` - 连接键
    /// * `now` - 当前时间戳
    ///
    /// # 返回
    /// * `TcpAction` - 应发送的动作
    pub fn passive_close(
        &mut self,
        key: &ConnectionKey,
        now: u64,
    ) -> TcpAction {
        if let Some(idx) = self.table.find(key)
            && let Some(conn) = self.table.get_mut(idx) {
                conn.state = ConnectionState::LastAck;
                conn.snd_nxt = conn.snd_nxt.wrapping_add(1);
                conn.last_active = now;
                // 不在此处递减 active_connections：递减仅在 LastAck -> Closed 完成时进行，
                // 避免与 handle_segment 的 LastAck -> Closed 路径双重递减。
                return TcpAction::SendFin;
            }
        TcpAction::Drop
    }

    /// 超时扫描
    ///
    /// # 参数
    /// * `now` - 当前时间戳
    /// * `timeout_ms` - 超时阈值
    pub fn handle_timeouts(&mut self, now: u64, timeout_ms: u64) {
        let expired = self.table.scan_timeouts(now, timeout_ms);
        self.stats.timeouts += expired.len() as u64;
        self.stats.active_connections = self.stats.active_connections.saturating_sub(expired.len() as u64);
    }

    /// 获取连接引用
    #[inline]
    pub fn get_connection(&self, key: &ConnectionKey) -> Option<&TcpConnection> {
        let idx = self.table.find(key)?;
        self.table.get(idx)
    }
}

/// TCP 动作（状态机决策输出）
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TcpAction {
    /// 丢弃包
    Drop,
    /// 发送 RST
    SendRst,
    /// 发送 SYN+ACK
    SendSynAck {
        /// SYN 序号
        syn_seq: u32,
        /// ACK 序号
        ack_seq: u32,
        /// 窗口大小
        window: u16,
    },
    /// 发送 ACK
    Ack {
        /// ACK 序号
        ack_seq: u32,
        /// 窗口大小
        window: u16,
    },
    /// 发送 FIN
    SendFin,
    /// 连接已建立
    Established,
    /// 连接已关闭
    Close,
}

#[cfg(test)]
mod tests {
    use super::*;

    fn make_key() -> ConnectionKey {
        ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 1]),
            8080,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        )
    }

    #[test]
    fn test_connection_state() {
        assert!(ConnectionState::Established.can_receive_data());
        assert!(ConnectionState::Established.can_send_data());
        assert!(!ConnectionState::Closed.is_active());
        assert!(!ConnectionState::Listen.can_send_data());
    }

    #[test]
    fn test_connection_key_hash() {
        let key = make_key();
        let hash1 = key.hash();
        assert_ne!(hash1, 0);

        // 反向键哈希应不同
        let reversed = key.reversed();
        assert_ne!(key, reversed);
        assert_ne!(key.hash(), reversed.hash());
    }

    #[test]
    fn test_connection_table_insert_find() {
        let mut table = ConnectionTable::new(1024);
        let key = make_key();

        // 插入
        let idx = table.insert(key, 100, 200, 1000).unwrap();
        assert_eq!(table.count(), 1);

        // 查找
        let found = table.find(&key);
        assert_eq!(found, Some(idx));

        // 重复插入应失败
        let result = table.insert(key, 101, 201, 1000);
        assert!(result.is_err());
    }

    #[test]
    fn test_connection_table_remove() {
        let mut table = ConnectionTable::new(1024);
        let key = make_key();

        table.insert(key, 100, 200, 1000).unwrap();
        assert_eq!(table.count(), 1);

        // 移除
        let removed = table.remove(&key);
        assert!(removed);
        assert_eq!(table.count(), 0);

        // 查找应返回 None
        assert!(table.find(&key).is_none());
    }

    #[test]
    fn test_tcp_state_machine_new_syn() {
        let mut sm = TcpStateMachine::new(1024);
        let key = make_key();

        // 模拟 SYN 包
        let mut data = [0u8; 20];
        data[0] = 0x00; // src_port high
        data[1] = 0x50; // src_port low (80)
        data[2] = 0x30; // dst_port high
        data[3] = 0x39; // dst_port low (12345)
        // seq_num
        data[4] = 0x00;
        data[5] = 0x00;
        data[6] = 0x00;
        data[7] = 0x64; // seq = 100
        // data_offset_flags = 0x5002 (SYN=0x02, data_offset=5)
        data[12] = 0x50;
        data[13] = 0x02;
        let tcp = TcpHeader::parse(&data).unwrap();

        // 处理 SYN (payload_len = 0 for SYN)
        let action = sm.handle_packet(&key, tcp, 0, 1000);
        assert!(matches!(action, TcpAction::SendSynAck { .. }));
        assert_eq!(sm.stats().syn_received, 1);
        assert_eq!(sm.stats().active_connections, 1);
    }

    #[test]
    fn test_generate_isn_csprng_unpredictable() {
        // RFC 6528：ISN 熵源必须为 CSPRNG，不得构成可预测序列
        //（原固定种子 0xDEADBEEF LCG 断言固定序列的语义已反转为"两次 ISN 不同"）。
        let sm = TcpStateMachine::new(16);
        let key = make_key();

        // 同一连接连续两次生成的 ISN 必须不同
        assert_ne!(sm.generate_isn(&key), sm.generate_isn(&key));

        // 16 次采样应高度互异（CSPRNG 下重复概率可忽略，阈值取半数防抖动）
        let mut seen = std::collections::HashSet::new();
        for _ in 0..16 {
            seen.insert(sm.generate_isn(&key));
        }
        assert!(seen.len() > 8, "ISN 序列不得可预测/重复");

        // per-connection 偏移：不同四元组的哈希偏移不同（确定性部分）
        let reversed = key.reversed();
        assert_ne!(key.hash() as u32, reversed.hash() as u32);
    }

    #[test]
    fn test_tcp_state_machine_handshake() {
        let mut sm = TcpStateMachine::new(1024);
        let key = make_key();
        let server_key = key.reversed();

        // Step 1: SYN (客户端 → 服务端)
        let mut syn_data = [0u8; 20];
        syn_data[4] = 0x00;
        syn_data[5] = 0x00;
        syn_data[6] = 0x00;
        syn_data[7] = 0x64; // seq = 100
        syn_data[12] = 0x50;
        syn_data[13] = 0x02; // SYN
        let syn = TcpHeader::parse(&syn_data).unwrap();

        let action1 = sm.handle_packet(&key, syn, 0, 1000);
        assert!(matches!(action1, TcpAction::SendSynAck { .. }));

        // 验证连接在 SYN-RECEIVED 状态（以服务端视角存储）
        let conn = sm.get_connection(&server_key).unwrap();
        assert_eq!(conn.state, ConnectionState::SynReceived);
        // 三次握手最终 ACK 必须确认服务端 SYN-ACK（ack_num = snd_nxt = ISN+1）；
        // ACK 验证（RFC 9293 §3.10）拒绝 (snd_una, snd_nxt] 区间外的 ack_num
        let server_snd_nxt = conn.snd_nxt;

        // Step 2: ACK（客户端 → 服务端，完成握手）
        let mut ack_data = [0u8; 20];
        ack_data[4] = 0x00;
        ack_data[5] = 0x00;
        ack_data[6] = 0x00;
        ack_data[7] = 0x65; // seq = 101 (SYN 消耗 1)
        ack_data[8] = (server_snd_nxt >> 24) as u8;
        ack_data[9] = (server_snd_nxt >> 16) as u8;
        ack_data[10] = (server_snd_nxt >> 8) as u8;
        ack_data[11] = server_snd_nxt as u8;
        ack_data[12] = 0x50;
        ack_data[13] = 0x10; // ACK
        let ack = TcpHeader::parse(&ack_data).unwrap();

        let action2 = sm.handle_packet(&key, ack, 0, 1001);
        assert!(matches!(action2, TcpAction::Established));

        // 验证连接已建立
        let conn = sm.get_connection(&server_key).unwrap();
        assert_eq!(conn.state, ConnectionState::Established);
    }

    #[test]
    fn test_synsent_validates_synack_isn() {
        // NET-028 回归：客户端 SynSent 收到 SYN-ACK 时，ack_num 必须等于 ISN+1
        //（确认我方 SYN）；否则 fail-closed 丢弃，不得完成握手。
        fn build_synack(ack: u32) -> TcpHeader {
            let mut d = [0u8; 20];
            d[7] = 0x50; // 服务端 seq（任意）
            d[8] = (ack >> 24) as u8;
            d[9] = (ack >> 16) as u8;
            d[10] = (ack >> 8) as u8;
            d[11] = ack as u8;
            d[12] = 0x50;
            d[13] = 0x12; // SYN + ACK
            *TcpHeader::parse(&d).unwrap()
        }

        // --- 合法 SYN-ACK：ack_num == ISN+1 → Established ---
        let mut sm = TcpStateMachine::new(16);
        let key = make_key();
        let idx = sm.send_syn(key, 10_000).unwrap();
        let iss = sm.table().get(idx).unwrap().snd_nxt; // = ISN
        let action = sm.handle_packet(&key, &build_synack(iss.wrapping_add(1)), 0, 10_001);
        assert!(matches!(action, TcpAction::Established));
        assert_eq!(
            sm.get_connection(&key).unwrap().state,
            ConnectionState::Established
        );

        // --- 非法 SYN-ACK：ack_num != ISN+1 → 丢弃，握手不完成 ---
        let mut sm2 = TcpStateMachine::new(16);
        let key2 = make_key();
        let idx2 = sm2.send_syn(key2, 20_000).unwrap();
        let iss2 = sm2.table().get(idx2).unwrap().snd_nxt;
        let forged = iss2.wrapping_add(1234); // 伪造 ack_num
        let action2 = sm2.handle_packet(&key2, &build_synack(forged), 0, 20_001);
        assert!(
            matches!(action2, TcpAction::Drop),
            "伪造 ack_num 的 SYN-ACK 必须被丢弃（NET-028）"
        );
        assert_ne!(
            sm2.get_connection(&key2).unwrap().state,
            ConnectionState::Established,
            "ISN 校验失败不得完成握手"
        );
    }

    #[test]
    fn test_sequence_number_wrap() {
        let mut conn = TcpConnection::empty();
        conn.init(make_key(), 0xFFFFFFFE, 1000, 100);

        // 序号接近溢出
        assert!(conn.is_seq_in_snd_window(0xFFFFFFFF)); // diff = 1
        assert!(conn.is_seq_in_snd_window(0x00000000)); // diff = 2, 在窗口内

        // 超出窗口的序号
        assert!(!conn.is_seq_in_snd_window(0x00010000)); // diff = 0x10000 > 0xFFFF

        // 使用 wrapping_sub 正确比较
        let diff = 0xFFFFFFFFu32.wrapping_sub(conn.snd_nxt);
        assert_eq!(diff, 1);

        let diff2 = 0x00000000u32.wrapping_sub(conn.snd_nxt);
        assert_eq!(diff2, 2);
    }

    #[test]
    fn test_connection_table_full() {
        let mut table = ConnectionTable::new(4);

        // 插入 4 个连接
        for i in 0..4u16 {
            let key = ConnectionKey::new(
                IpAddr::V4([10, 0, 0, i as u8]),
                8080 + i,
                IpAddr::V4([10, 0, 0, 2]),
                12345,
                IpVersion::V4,
            );
            table.insert(key, 100, 200, 1000).unwrap();
        }
        assert_eq!(table.count(), 4);

        // 第 5 个应失败
        let extra_key = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 5]),
            8085,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );
        let result = table.insert(extra_key, 100, 200, 1000);
        assert!(result.is_err());
    }

    #[test]
    fn test_timeout_scan() {
        let mut table = ConnectionTable::new(1024);
        let key = make_key();
        table.insert(key, 100, 200, 1000).unwrap();

        // 超时（100ms 后，超时 50ms）
        let expired = table.scan_timeouts(2000, 50);
        assert_eq!(expired.len(), 1);
        assert_eq!(expired[0], key);
        assert_eq!(table.count(), 0);
    }

    #[test]
    fn test_tcp_state_machine_close() {
        let mut sm = TcpStateMachine::new(1024);
        let key = make_key();

        // 手动添加连接
        sm.table_mut().insert(key, 1000, 2000, 100).unwrap();
        let idx = sm.table().find(&key).unwrap();
        if let Some(conn) = sm.table_mut().get_mut(idx) {
            conn.state = ConnectionState::Established;
        }
        sm.stats.active_connections = 1;
        sm.stats.total_connections = 1;

        // 关闭连接
        let action = sm.close_connection(&key, 200);
        assert!(matches!(action, TcpAction::SendFin));

        let conn = sm.get_connection(&key).unwrap();
        assert_eq!(conn.state, ConnectionState::FinWait1);
    }

    #[test]
    fn test_connection_key_reversed() {
        let key = make_key();
        let reversed = key.reversed();

        assert_eq!(reversed.src_ip, key.dst_ip);
        assert_eq!(reversed.dst_ip, key.src_ip);
        assert_eq!(reversed.src_port, key.dst_port);
        assert_eq!(reversed.dst_port, key.src_port);
        assert_eq!(reversed.ip_version, key.ip_version);

        let double_reversed = reversed.reversed();
        assert_eq!(double_reversed, key);
    }

    #[test]
    fn test_connection_key_hash_consistency() {
        let key1 = make_key();
        let key2 = make_key();

        assert_eq!(key1.hash(), key2.hash());

        let key3 = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 1]),
            8081,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );
        assert_ne!(key1.hash(), key3.hash());
    }

    #[test]
    fn test_tcp_connection_empty() {
        let conn = TcpConnection::empty();
        assert!(conn.is_free());
        assert!(!conn.is_time_wait());
        assert_eq!(conn.state, ConnectionState::Closed);
        assert_eq!(conn.snd_nxt, 0);
        assert_eq!(conn.rcv_nxt, 0);
        assert_eq!(conn.retransmit_count, 0);
        assert!(!conn.in_fast_recovery);
    }

    #[test]
    fn test_tcp_connection_init() {
        let mut conn = TcpConnection::empty();
        let key = make_key();
        conn.init(key, 100, 200, 1000);

        assert_eq!(conn.key, key);
        assert_eq!(conn.state, ConnectionState::Listen);
        assert_eq!(conn.snd_nxt, 100);
        assert_eq!(conn.rcv_nxt, 200);
        assert_eq!(conn.iss, 100);
        assert_eq!(conn.irs, 200);
        assert_eq!(conn.last_active, 1000);
        assert_eq!(conn.created_at, 1000);
        assert!(!conn.is_free());
    }

    #[test]
    fn test_sequence_window_boundaries() {
        let mut conn = TcpConnection::empty();
        conn.init(make_key(), 1000, 2000, 100);
        conn.rcv_wnd = 100;
        conn.snd_wnd = 100;

        // 边界：刚好在窗口内
        assert!(conn.is_seq_in_rcv_window(2000));
        assert!(conn.is_seq_in_rcv_window(2099));
        assert!(!conn.is_seq_in_rcv_window(2100));

        // 发送窗口
        assert!(conn.is_seq_in_snd_window(1000));
        assert!(conn.is_seq_in_snd_window(1099));
        assert!(!conn.is_seq_in_snd_window(1100));
    }

    #[test]
    fn test_sequence_number_wrap_around() {
        let mut conn = TcpConnection::empty();
        conn.init(make_key(), 0xFFFFFFF0, 0xFFFFFFF0, 100);
        conn.rcv_wnd = 32;
        conn.snd_wnd = 32;

        // 回绕前的序号
        assert!(conn.is_seq_in_rcv_window(0xFFFFFFF0));
        assert!(conn.is_seq_in_rcv_window(0xFFFFFFFF));

        // 回绕后的序号
        assert!(conn.is_seq_in_rcv_window(0x00000000));
        assert!(conn.is_seq_in_rcv_window(0x0000000F));

        // 超出窗口
        assert!(!conn.is_seq_in_rcv_window(0x00000010));
    }

    #[test]
    fn test_connection_table_find_reverse_key() {
        let mut table = ConnectionTable::new(1024);
        let key = make_key();
        let reverse_key = key.reversed();

        table.insert(key, 100, 200, 1000).unwrap();

        // 正向查找
        assert!(table.find(&key).is_some());

        // 反向查找应该找不到
        assert!(table.find(&reverse_key).is_none());
    }

    #[test]
    fn test_connection_table_get_out_of_bounds() {
        let table = ConnectionTable::new(1024);
        assert!(table.get(1024).is_none());
        assert!(table.get(9999).is_none());
    }

    #[test]
    fn test_connection_table_get_mut_out_of_bounds() {
        let mut table = ConnectionTable::new(1024);
        assert!(table.get_mut(1024).is_none());
        assert!(table.get_mut(9999).is_none());
    }

    #[test]
    fn test_connection_table_iter_active() {
        let mut table = ConnectionTable::new(1024);

        for i in 0..5u16 {
            let key = ConnectionKey::new(
                IpAddr::V4([10, 0, 0, i as u8]),
                8080 + i,
                IpAddr::V4([10, 0, 0, 2]),
                12345,
                IpVersion::V4,
            );
            table.insert(key, 100, 200, 1000).unwrap();
        }

        let active: Vec<_> = table.iter_active().collect();
        assert_eq!(active.len(), 5);
    }

    #[test]
    fn test_connection_table_scan_timeouts_partial() {
        let mut table = ConnectionTable::new(1024);

        for i in 0..5u16 {
            let key = ConnectionKey::new(
                IpAddr::V4([10, 0, 0, i as u8]),
                8080 + i,
                IpAddr::V4([10, 0, 0, 2]),
                12345,
                IpVersion::V4,
            );
            table.insert(key, 100, 200, i as u64 * 100).unwrap();
        }

        assert_eq!(table.count(), 5);

        let expired = table.scan_timeouts(250, 100);
        assert_eq!(expired.len(), 2);
        assert_eq!(table.count(), 3);
    }

    #[test]
    fn test_connection_state_all_variants() {
        let states = vec![
            ConnectionState::Closed,
            ConnectionState::Listen,
            ConnectionState::SynSent,
            ConnectionState::SynReceived,
            ConnectionState::Established,
            ConnectionState::FinWait1,
            ConnectionState::FinWait2,
            ConnectionState::CloseWait,
            ConnectionState::Closing,
            ConnectionState::TimeWait,
        ];

        for state in &states {
            let _ = format!("{:?}", state);
        }

        assert!(ConnectionState::Established.is_active());
        assert!(ConnectionState::SynSent.is_active());
        assert!(ConnectionState::TimeWait.is_active());
        assert!(!ConnectionState::Closed.is_active());

        assert!(ConnectionState::Established.can_receive_data());
        assert!(ConnectionState::CloseWait.can_receive_data());
        assert!(!ConnectionState::Listen.can_receive_data());

        assert!(ConnectionState::Established.can_send_data());
        assert!(!ConnectionState::CloseWait.can_send_data());
    }

    #[test]
    fn test_tcp_stats_default() {
        let stats = TcpStats::default();
        assert_eq!(stats.total_connections, 0);
        assert_eq!(stats.active_connections, 0);
        assert_eq!(stats.syn_received, 0);
        assert_eq!(stats.rst_received, 0);
        assert_eq!(stats.timeouts, 0);
    }

    #[test]
    fn test_connection_table_min_capacity() {
        let table = ConnectionTable::new(0);
        assert_eq!(table.capacity(), 1);
        assert_eq!(table.count(), 0);
    }

    #[test]
    fn test_connection_table_max_capacity() {
        let table = ConnectionTable::new(MAX_CONNECTIONS + 100);
        assert_eq!(table.capacity(), MAX_CONNECTIONS);
    }

    #[test]
    fn test_tcp_state_machine_stats() {
        let sm = TcpStateMachine::new(1024);
        let stats = sm.stats();
        assert_eq!(stats.total_connections, 0);
        assert_eq!(stats.active_connections, 0);
    }

    #[test]
    fn test_tcp_state_machine_get_connection_nonexistent() {
        let sm = TcpStateMachine::new(1024);
        let key = make_key();
        assert!(sm.get_connection(&key).is_none());
    }

    #[test]
    fn test_connection_table_remove_nonexistent() {
        let mut table = ConnectionTable::new(1024);
        let key = make_key();
        assert!(!table.remove(&key));
    }

    #[test]
    fn test_tcp_connection_time_wait() {
        let mut conn = TcpConnection::empty();
        conn.state = ConnectionState::TimeWait;
        assert!(conn.is_time_wait());
        assert!(!conn.is_free());
    }

    #[test]
    fn test_hash_ip_addr_v4_and_v6() {
        let v4_key = ConnectionKey::new(
            IpAddr::V4([1, 2, 3, 4]),
            80,
            IpAddr::V4([5, 6, 7, 8]),
            443,
            IpVersion::V4,
        );
        let v6_key = ConnectionKey::new(
            IpAddr::V6([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]),
            80,
            IpAddr::V6([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2]),
            443,
            IpVersion::V6,
        );

        let h1 = v4_key.hash();
        let h2 = v6_key.hash();
        assert_ne!(h1, h2);
    }

    #[test]
    fn test_tcp_state_machine_handle_packet_on_closed() {
        let mut sm = TcpStateMachine::new(1024);
        let key = make_key();

        let mut data = [0u8; 20];
        data[12] = 0x50;
        data[13] = 0x12;
        let tcp = TcpHeader::parse(&data).unwrap();

        let action = sm.handle_packet(&key, tcp, 0, 1000);
        assert!(matches!(action, TcpAction::SendRst | TcpAction::Drop));
    }

    /// 测试：边界槽（capacity-1）插入不 panic（修复 off-by-one）
    ///
    /// 桶中存储 (slot_idx + 1)，当 slot_idx = capacity-1 时桶值为 capacity。
    /// 修复前 self.slots[capacity] 越界 panic；修复后 self.slots[capacity-1] 正确。
    #[test]
    fn test_insert_off_by_one_boundary() {
        let mut table = ConnectionTable::new(2);
        let key1 = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 1]),
            80,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );
        let key2 = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 3]),
            80,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );

        // 填满表：slot 0 → bucket=1，slot 1 → bucket=2
        table.insert(key1, 100, 200, 1000).unwrap();
        table.insert(key2, 100, 200, 1000).unwrap();
        assert_eq!(table.count(), 2);

        // 模拟状态机直接关闭 slot 1（不清除桶，不入栈）
        table.slots[1].state = ConnectionState::Closed;
        table.count = 1;
        table.free_stack.push(1);

        // 重新插入 key2：桶值=2 即 existing_slot_idx=2
        // 修复前：self.slots[2] 越界 panic；修复后：self.slots[1] 正确
        let result = table.insert(key2, 100, 200, 1000);
        assert!(result.is_ok(), "边界槽插入不应 panic");
        assert_eq!(table.find(&key2), Some(1));
    }

    /// 测试：哈希冲突时正确检查槽状态（非 off-by-one）
    ///
    /// 验证 insert 在遇到指向 CLOSED 槽的桶时，正确读取对应槽（减 1 后）
    /// 并复用该桶，而非错误读取相邻槽。
    #[test]
    fn test_insert_hash_collision_reuse_bucket() {
        let mut table = ConnectionTable::new(4);
        let key1 = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 1]),
            80,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );
        let key2 = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 3]),
            90,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );

        // 插入 key1 → slot 0
        table.insert(key1, 100, 200, 1000).unwrap();
        let slot1 = table.find(&key1).unwrap();
        let bucket_idx = (key1.hash() as usize) & (HASH_BUCKETS - 1);
        let original_bucket_val = table.buckets[bucket_idx];
        assert_eq!(original_bucket_val, (slot1 as u32) + 1);

        // 模拟状态机关闭 slot 0（不清除桶）
        table.slots[slot1].state = ConnectionState::Closed;
        table.count = 0;
        table.free_stack.push(slot1 as u32);

        // 重新插入 key1：应复用原桶
        table.insert(key1, 100, 200, 1000).unwrap();
        let new_slot = table.find(&key1).unwrap();
        assert_eq!(new_slot, slot1, "应复用同一槽");
        // 桶被复用：值应等于 (new_slot + 1)
        assert_eq!(table.buckets[bucket_idx], (new_slot as u32) + 1);

        // 插入 key2 到另一个槽，确保不干扰
        table.insert(key2, 100, 200, 1000).unwrap();
        assert!(table.find(&key2).is_some());
    }

    /// 测试：find_free_slot 返回正确的索引顺序
    ///
    /// free_stack 逆序初始化，pop() 应按 0, 1, 2, ... 顺序返回。
    #[test]
    fn test_find_free_slot_returns_correct_indices() {
        let mut table = ConnectionTable::new(4);

        // 依次插入 4 个连接，验证 slot 索引按 0, 1, 2, 3 顺序分配
        for i in 0..4u16 {
            let key = ConnectionKey::new(
                IpAddr::V4([10, 0, 0, i as u8]),
                8080 + i,
                IpAddr::V4([10, 0, 0, 2]),
                12345,
                IpVersion::V4,
            );
            let idx = table.insert(key, 100, 200, 1000).unwrap();
            assert_eq!(idx, i as usize, "free_stack 应按 0,1,2,3 顺序分配");
        }
        assert_eq!(table.count(), 4);
    }

    /// 测试：remove 后 insert 复用已释放的槽
    ///
    /// remove 将槽索引 push 回 free_stack，下次 insert 应 pop 复用。
    #[test]
    fn test_remove_then_insert_reuses_slot() {
        let mut table = ConnectionTable::new(4);
        let key1 = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 1]),
            80,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );
        let key2 = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 3]),
            80,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );
        let key3 = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 5]),
            80,
            IpAddr::V4([10, 0, 0, 2]),
            12345,
            IpVersion::V4,
        );

        // 插入 key1 → slot 0, key2 → slot 1
        let idx1 = table.insert(key1, 100, 200, 1000).unwrap();
        let idx2 = table.insert(key2, 100, 200, 1000).unwrap();

        // 移除 key1 → slot 0 回收到 free_stack
        assert!(table.remove(&key1));
        assert_eq!(table.count(), 1);

        // 插入 key3 → 应复用 slot 0（free_stack 顶部）
        let idx3 = table.insert(key3, 100, 200, 1000).unwrap();
        assert_eq!(idx3, idx1, "应复用刚释放的槽");
        assert_eq!(table.count(), 2);

        // key2 仍在原位
        assert_eq!(table.find(&key2), Some(idx2));
        // key3 在 slot 0
        assert_eq!(table.find(&key3), Some(idx1));
        // key1 已不存在
        assert!(table.find(&key1).is_none());
    }

    /// 测试：高频率插入/移除循环（free_stack 一致性）
    ///
    /// 反复插入和移除连接，验证：
    /// - count 始终正确
    /// - free_stack 不泄漏（总能插入到 capacity）
    /// - 所有活跃连接可被 find 找到
    #[test]
    fn test_high_churn_insert_remove_cycles() {
        let mut table = ConnectionTable::new(16);

        // 模拟高频率 churn：每轮插入 N 个、移除 N 个
        for round in 0..100u16 {
            let base = round * 16;
            // 插入 16 个连接填满表
            for i in 0..16u16 {
                let key = ConnectionKey::new(
                    IpAddr::V4([(base + i) as u8, 0, 0, 1]),
                    8080 + i,
                    IpAddr::V4([10, 0, 0, 2]),
                    12345,
                    IpVersion::V4,
                );
                assert!(table.insert(key, 100, 200, 1000).is_ok());
            }
            assert_eq!(table.count(), 16, "第 {} 轮：填满后 count 应为 16", round);

            // 验证表已满，再插入应失败
            let extra_key = ConnectionKey::new(
                IpAddr::V4([255, 0, 0, 1]),
                9999,
                IpAddr::V4([10, 0, 0, 2]),
                12345,
                IpVersion::V4,
            );
            assert!(table.insert(extra_key, 100, 200, 1000).is_err());

            // 移除全部
            for i in 0..16u16 {
                let key = ConnectionKey::new(
                    IpAddr::V4([(base + i) as u8, 0, 0, 1]),
                    8080 + i,
                    IpAddr::V4([10, 0, 0, 2]),
                    12345,
                    IpVersion::V4,
                );
                assert!(table.remove(&key), "第 {} 轮第 {} 个移除应成功", round, i);
            }
            assert_eq!(table.count(), 0, "第 {} 轮：移除后 count 应为 0", round);
        }

        // 最终验证：free_stack 无泄漏，仍可填满 capacity
        for i in 0..16u16 {
            let key = ConnectionKey::new(
                IpAddr::V4([200, 0, 0, i as u8]),
                8080 + i,
                IpAddr::V4([10, 0, 0, 2]),
                12345,
                IpVersion::V4,
            );
            assert!(table.insert(key, 100, 200, 1000).is_ok());
        }
        assert_eq!(table.count(), 16);
    }

    /// NET-020 回归：挂载 BindTable 后，仅监听端口接受 SYN（fail-closed）。
    #[test]
    fn test_net020_reject_syn_on_non_listening_port() {
        use crate::transport::acceptor::BindTable;
        use crate::NetAddr;

        let mut bind = BindTable::new();
        bind.register(NetAddr::new_ipv4([10, 0, 0, 2], 8080), 128)
            .unwrap();

        let mut sm = TcpStateMachine::new(1024);
        sm.attach_bind_table(bind);

        // 指向 8080（监听中）的 SYN → 应建连
        let listening_key = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 1]),
            1000,
            IpAddr::V4([10, 0, 0, 2]),
            8080,
            IpVersion::V4,
        );
        let mut syn = [0u8; 20];
        syn[12] = 0x50;
        syn[13] = 0x02; // SYN
        let tcp = TcpHeader::parse(&syn).unwrap();
        let action = sm.handle_packet(&listening_key, tcp, 0, 1000);
        assert!(matches!(action, TcpAction::SendSynAck { .. }));

        // 指向 9999（未监听）的 SYN → 必须 Drop（fail-closed）
        let non_listening_key = ConnectionKey::new(
            IpAddr::V4([10, 0, 0, 1]),
            1001,
            IpAddr::V4([10, 0, 0, 2]),
            9999,
            IpVersion::V4,
        );
        let action2 = sm.handle_packet(&non_listening_key, tcp, 0, 1000);
        assert!(matches!(action2, TcpAction::Drop));
        assert_eq!(sm.stats().rejected, 1);
    }

    /// NET-021 回归：off-path FIN（SEQ 不在接收窗口）不得触发状态迁移。
    #[test]
    fn test_net021_offpath_fin_respects_rcv_window() {
        let mut sm = TcpStateMachine::new(1024);
        let ckey = make_key();
        let skey = ckey.reversed();

        // 客户端 SYN
        let mut syn = [0u8; 20];
        syn[7] = 0x64; // seq = 100
        syn[12] = 0x50;
        syn[13] = 0x02; // SYN
        let tcp_syn = TcpHeader::parse(&syn).unwrap();
        assert!(matches!(
            sm.handle_packet(&ckey, tcp_syn, 0, 1000),
            TcpAction::SendSynAck { .. }
        ));

        // 服务端 ACK 完成三次握手
        let server_snd_nxt = sm.get_connection(&skey).unwrap().snd_nxt;
        let mut ack = [0u8; 20];
        ack[7] = 0x65; // seq = 101
        ack[8] = (server_snd_nxt >> 24) as u8;
        ack[9] = (server_snd_nxt >> 16) as u8;
        ack[10] = (server_snd_nxt >> 8) as u8;
        ack[11] = server_snd_nxt as u8;
        ack[12] = 0x50;
        ack[13] = 0x10; // ACK
        let tcp_ack = TcpHeader::parse(&ack).unwrap();
        assert!(matches!(
            sm.handle_packet(&ckey, tcp_ack, 0, 1001),
            TcpAction::Established
        ));

        let conn = sm.get_connection(&skey).unwrap();
        assert_eq!(conn.state, ConnectionState::Established);
        let rcv_nxt = conn.rcv_nxt;

        // off-path FIN：seq 远离接收窗口（+0x10000 > rcv_wnd 0xFFFF）→ 状态不得迁移
        let bad_seq = rcv_nxt.wrapping_add(0x1_0000);
        let mut fin_bad = [0u8; 20];
        fin_bad[4] = (bad_seq >> 24) as u8;
        fin_bad[5] = (bad_seq >> 16) as u8;
        fin_bad[6] = (bad_seq >> 8) as u8;
        fin_bad[7] = bad_seq as u8;
        fin_bad[12] = 0x50;
        fin_bad[13] = 0x01; // FIN
        let tcp_fin_bad = TcpHeader::parse(&fin_bad).unwrap();
        let _ = sm.handle_packet(&ckey, tcp_fin_bad, 0, 1002);
        assert_eq!(
            sm.get_connection(&skey).unwrap().state,
            ConnectionState::Established,
            "off-path FIN 不得触发状态迁移（NET-021）"
        );

        // 合法 FIN：seq = rcv_nxt → CloseWait
        let mut fin_ok = [0u8; 20];
        fin_ok[4] = (rcv_nxt >> 24) as u8;
        fin_ok[5] = (rcv_nxt >> 16) as u8;
        fin_ok[6] = (rcv_nxt >> 8) as u8;
        fin_ok[7] = rcv_nxt as u8;
        fin_ok[12] = 0x50;
        fin_ok[13] = 0x01; // FIN
        let tcp_fin_ok = TcpHeader::parse(&fin_ok).unwrap();
        let _ = sm.handle_packet(&ckey, tcp_fin_ok, 0, 1003);
        assert_eq!(
            sm.get_connection(&skey).unwrap().state,
            ConnectionState::CloseWait,
            "合法 FIN 应迁移到 CloseWait"
        );
    }
}