zenith-runtime 0.1.0

//! TaskOffloadChannel - 无锁双向任务通道
//!
//! 在 DataPlane Worker 与 Blocking Worker Pool 之间建立无锁双向通道，
//! 实现阻塞任务的异步卸载。所有操作均为 O(1) 无锁，零堆分配热路径。
//!
//! # 设计原则
//! - 基于 crossbeam ArrayQueue 的严格 FIFO 顺序
//! - const-generics 定长容量，编译期确定
//! - 任务载荷枚举化，避免堆分配
//! - 零拷贝传递：仅传递指针/ID，不复制数据
//!
//! # 通道方向
//! ```text
//! DataPlane Worker ──submit()──→ Blocking Worker Pool
//! DataPlane Worker ←──poll_result()── Blocking Worker Pool
//! ```

use crossbeam::queue::ArrayQueue;

/// 任务请求（DataPlane → Blocking Pool）
#[derive(Debug, Clone)]
pub struct TaskRequest {
    /// 任务唯一标识。
    pub task_id: u64,
    /// 所属域 ID。
    pub domain_id: u32,
    /// 任务载荷。
    pub payload: TaskPayload,
    /// 相对截止时间（毫秒）：自 `submitted_at` 起的最大允许排队+执行时长。
    pub deadline_ms: u64,
    /// 来源通道 ID（用于结果路由与审计）。
    pub channel_id: u32,
    /// 提交时刻（单调时钟，deadline 相对语义的时间基准）。
    pub submitted_at: std::time::Instant,
}

impl TaskRequest {
    /// 判断任务相对截止时间是否已过（deadline 强制执行）。
    ///
    /// `deadline_ms == 0` 视为"无截止时间"（不强制执行）。
    /// 供 Blocking Worker 在取任务时判定：已超时的任务不再执行，
    /// 而由通道直接产出 `TaskError::Timeout` 结果，避免僵尸任务滞留。
    #[inline]
    pub fn is_expired(&self) -> bool {
        if self.deadline_ms == 0 {
            return false;
        }
        self.submitted_at.elapsed().as_millis() as u64 >= self.deadline_ms
    }
}

/// 任务结果（Blocking Pool → DataPlane）
#[derive(Debug, Clone)]
pub struct TaskResult {
    /// 任务唯一标识。
    pub task_id: u64,
    /// 所属域 ID。
    pub domain_id: u32,
    /// 任务执行结果。
    pub result: Result<TaskOutput, TaskError>,
}

/// 任务载荷（枚举化，热路径零堆分配）
#[derive(Debug, Clone)]
pub enum TaskPayload {
    /// 数据库查询任务。
    DbQuery {
        /// SQL 语句数据。
        sql_data: Vec<u8>,
    },
    /// 文件读取任务。
    FileRead {
        /// 文件路径数据。
        path_data: Vec<u8>,
    },
    /// 外部 RPC 调用任务。
    ExternalRpc {
        /// 请求 URL 数据。
        url_data: Vec<u8>,
        /// 请求体数据。
        body_data: Vec<u8>,
    },
    /// 复杂计算任务。
    ComplexCompute {
        /// 计算输入数据。
        input_data: Vec<u8>,
    },
    /// 自定义任务。
    Custom {
        /// 自定义任务类型标识。
        kind: u32,
        /// 自定义任务数据。
        data: Vec<u8>,
    },
}

impl TaskPayload {
    /// 返回任务载荷的类型标识。
    #[inline]
    pub fn kind(&self) -> u32 {
        match self {
            TaskPayload::DbQuery { .. } => 1,
            TaskPayload::FileRead { .. } => 2,
            TaskPayload::ExternalRpc { .. } => 3,
            TaskPayload::ComplexCompute { .. } => 4,
            TaskPayload::Custom { kind, .. } => *kind,
        }
    }

    /// 判断该任务是否为写操作（数据库查询或外部 RPC）。
    #[inline]
    pub fn is_write(&self) -> bool {
        matches!(
            self,
            TaskPayload::DbQuery { .. } | TaskPayload::ExternalRpc { .. }
        )
    }

    /// 根据载荷数据长度估算近似执行开销。
    #[inline]
    pub fn approximate_cost(&self) -> u32 {
        match self {
            TaskPayload::DbQuery { sql_data } => (sql_data.len() as u32).saturating_mul(10),
            TaskPayload::FileRead { path_data } => (path_data.len() as u32).saturating_mul(20),
            TaskPayload::ExternalRpc { url_data, .. } => (url_data.len() as u32).saturating_mul(50),
            TaskPayload::ComplexCompute { input_data } => (input_data.len() as u32).saturating_mul(100),
            TaskPayload::Custom { data, .. } => (data.len() as u32).saturating_mul(30),
        }
    }
}

/// 任务输出
#[derive(Debug, Clone)]
pub struct TaskOutput {
    /// 输出数据。
    pub data: Vec<u8>,
    /// 状态码。
    pub status_code: u16,
    /// 执行耗时（毫秒）。
    pub elapsed_ms: u32,
}

/// 任务错误
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TaskError {
    /// 任务通道已满，无法投递。
    ChannelFull,
    /// 任务超时。
    Timeout {
        /// 超时任务 ID。
        task_id: u64,
        /// 已耗时间（毫秒）。
        elapsed_ms: u32,
    },
    /// 任务被取消。
    Cancelled {
        /// 被取消任务 ID。
        task_id: u64,
    },
    /// 资源耗尽，无法接收新任务。
    ResourceExhausted,
    /// 无效的任务载荷。
    InvalidPayload {
        /// 无效载荷的类型标识。
        kind: u32,
    },
    /// 内部错误。
    Internal(String),
}

impl std::fmt::Display for TaskError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            TaskError::ChannelFull => write!(f, "task channel is full"),
            TaskError::Timeout { task_id, elapsed_ms } => {
                write!(f, "task {task_id} timed out after {elapsed_ms}ms")
            }
            TaskError::Cancelled { task_id } => write!(f, "task {task_id} was cancelled"),
            TaskError::ResourceExhausted => write!(f, "blocking worker pool exhausted"),
            TaskError::InvalidPayload { kind } => write!(f, "invalid payload kind: {kind}"),
            TaskError::Internal(msg) => write!(f, "internal error: {msg}"),
        }
    }
}

impl std::error::Error for TaskError {}

/// 无锁双向任务通道
///
/// 使用 const-generics 编译期确定容量，所有操作零锁、O(1)。
/// 适用于单 DataPlane Worker 与 Blocking Worker Pool 间的高并发任务传递。
#[derive(Debug)]
pub struct TaskOffloadChannel<const CAP: usize> {
    request_queue: ArrayQueue<TaskRequest>,
    result_queue: ArrayQueue<TaskResult>,
    pending_count: std::sync::atomic::AtomicU64,
    completed_count: std::sync::atomic::AtomicU64,
    channel_id: u32,
}

impl<const CAP: usize> TaskOffloadChannel<CAP> {
    /// 创建指定通道 ID 的无锁任务通道。
    pub fn new(channel_id: u32) -> Self {
        Self {
            request_queue: ArrayQueue::new(CAP),
            result_queue: ArrayQueue::new(CAP),
            pending_count: std::sync::atomic::AtomicU64::new(0),
            completed_count: std::sync::atomic::AtomicU64::new(0),
            channel_id,
        }
    }

    /// 返回通道 ID。
    #[inline]
    pub fn channel_id(&self) -> u32 {
        self.channel_id
    }

    /// 返回通道容量（编译期常量）。
    #[inline]
    pub const fn capacity(&self) -> usize {
        CAP
    }

    /// 返回当前待处理任务数。
    #[inline]
    pub fn pending_count(&self) -> u64 {
        self.pending_count.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// 返回累计已完成任务数。
    #[inline]
    pub fn completed_count(&self) -> u64 {
        self.completed_count.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// 判断请求队列是否为空。
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.request_queue.is_empty()
    }

    /// 返回请求队列中的任务数。
    #[inline]
    pub fn len(&self) -> usize {
        self.request_queue.len()
    }

    /// DataPlane Worker 投递任务（无锁，O(1)）
    #[inline]
    pub fn submit(&self, req: TaskRequest) -> Result<(), TaskError> {
        self.request_queue.push(req).map_err(|_| TaskError::ChannelFull)?;
        self.pending_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        Ok(())
    }

    /// DataPlane Worker 批量投递
    ///
    /// 逐个入队直至通道满。已成功入队的元素计入 `pending_count`；
    /// **未入队的剩余元素通过 `Err` 原样返回**，调用方可据此重试或背压，
    /// 避免部分失败时剩余元素被 `Vec` 消费丢弃。
    ///
    /// # Returns
    /// * `Ok(count)` - 全部元素成功入队，`count` 为入队数（等于入参长度）。
    /// * `Err(remaining)` - 通道满，`remaining` 为尚未入队的元素（含触发失败的那一条）。
    #[inline]
    pub fn submit_batch(&self, reqs: Vec<TaskRequest>) -> Result<usize, Vec<TaskRequest>> {
        let mut iter = reqs.into_iter();
        let mut count = 0usize;
        let mut remaining: Vec<TaskRequest> = Vec::new();
        let mut failed = false;
        for req in iter.by_ref() {
            match self.request_queue.push(req) {
                Ok(()) => count += 1,
                Err(req_back) => {
                    // 通道满：回退触发失败的那一条，并收集迭代器中尚未消费的剩余
                    failed = true;
                    remaining.push(req_back);
                    break;
                }
            }
        }
        if failed {
            remaining.extend(iter);
        }
        if count > 0 {
            self.pending_count
                .fetch_add(count as u64, std::sync::atomic::Ordering::Relaxed);
        }
        if failed {
            Err(remaining)
        } else {
            Ok(count)
        }
    }

    /// Blocking Worker 取任务（无锁，O(1)）
    ///
    /// 已过截止时间的任务不再执行：直接产出 `TaskError::Timeout` 结果并平衡账本，
    /// 继续取下一个未过期任务（deadline 强制执行，杜绝僵尸任务滞留）。
    #[inline]
    pub fn fetch(&self) -> Option<TaskRequest> {
        loop {
            let req = match self.request_queue.pop() {
                Some(r) => r,
                None => return None,
            };
            if !req.is_expired() {
                return Some(req);
            }
            // 任务已超时：不执行，直接产出 Timeout 结果并平衡账本
            let elapsed = req.submitted_at.elapsed().as_millis() as u32;
            let result = TaskResult {
                task_id: req.task_id,
                domain_id: req.domain_id,
                result: Err(TaskError::Timeout {
                    task_id: req.task_id,
                    elapsed_ms: elapsed,
                }),
            };
            // 结果入队失败（结果队列满/背压）：fail-closed 丢弃该 Timeout 结果，
            // 但**不回退 pending**——任务已从请求队列弹出却未产生可观测结果，
            // pending 保持不降即暴露为"滞留账本"，避免 pending 与 completed 失衡
            // （原实现无论 push 成败都 dec_pending，导致结果丢失时账本静默失衡）。
            if self.result_queue.push(result).is_ok() {
                self.completed_count
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                Self::dec_pending(&self.pending_count, 1);
            }
        }
    }

    /// Blocking Worker 批量取任务
    ///
    /// 已过截止时间的任务跳过执行，产出 `TaskError::Timeout` 结果并平衡账本。
    #[inline]
    pub fn fetch_batch(&self, max: usize) -> Vec<TaskRequest> {
        let mut batch = Vec::with_capacity(max);
        while batch.len() < max {
            match self.request_queue.pop() {
                Some(req) => {
                    if !req.is_expired() {
                        batch.push(req);
                    } else {
                        // 已超时：不执行，产出 Timeout 结果并平衡账本
                        let elapsed = req.submitted_at.elapsed().as_millis() as u32;
                        let result = TaskResult {
                            task_id: req.task_id,
                            domain_id: req.domain_id,
                            result: Err(TaskError::Timeout {
                                task_id: req.task_id,
                                elapsed_ms: elapsed,
                            }),
                        };
                        if self.result_queue.push(result).is_ok() {
                            self.completed_count
                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                            Self::dec_pending(&self.pending_count, 1);
                        }
                        // 结果入队失败（结果队列满）：不回退 pending（账本平衡语义同 fetch()）
                    }
                }
                None => break,
            }
        }
        batch
    }

    /// Blocking Worker 回写结果（无锁，O(1)）
    #[inline]
    pub fn complete(&self, result: TaskResult) -> Result<(), TaskError> {
        self.result_queue.push(result).map_err(|_| TaskError::ChannelFull)?;
        self.completed_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        Self::dec_pending(&self.pending_count, 1);
        Ok(())
    }

    /// Blocking Worker 批量回写结果
    ///
    /// 逐个入队直至结果队列满。已成功入队的结果同步递增 `completed_count`、
    /// 递减 `pending_count`；**未入队的剩余结果通过 `Err` 原样返回**，
    /// 避免部分失败时剩余结果被 `Vec` 消费丢弃。
    ///
    /// # Returns
    /// * `Ok(count)` - 全部结果成功入队，`count` 为入队数（等于入参长度）。
    /// * `Err(remaining)` - 结果队列满，`remaining` 为尚未入队的结果（含触发失败的那一条）。
    #[inline]
    pub fn complete_batch(&self, results: Vec<TaskResult>) -> Result<usize, Vec<TaskResult>> {
        let mut iter = results.into_iter();
        let mut count = 0usize;
        let mut remaining: Vec<TaskResult> = Vec::new();
        let mut failed = false;
        for result in iter.by_ref() {
            match self.result_queue.push(result) {
                Ok(()) => count += 1,
                Err(result_back) => {
                    // 队列满：回退触发失败的那一条，并收集迭代器中尚未消费的剩余
                    failed = true;
                    remaining.push(result_back);
                    break;
                }
            }
        }
        if failed {
            remaining.extend(iter);
        }
        if count > 0 {
            self.completed_count
                .fetch_add(count as u64, std::sync::atomic::Ordering::Relaxed);
            Self::dec_pending(&self.pending_count, count as u64);
        }
        if failed {
            Err(remaining)
        } else {
            Ok(count)
        }
    }

    /// 饱和递减 pending 计数（CAS 配对保护）
    ///
    /// complete/complete_batch 允许在无配对 submit 时被调用（容错语义），
    /// 裸 `fetch_sub` 会从 0 下溢回绕到 `u64::MAX`；此处 CAS 饱和到 0。
    #[inline]
    fn dec_pending(counter: &std::sync::atomic::AtomicU64, n: u64) {
        let mut cur = counter.load(std::sync::atomic::Ordering::Relaxed);
        loop {
            let new = cur.saturating_sub(n);
            match counter.compare_exchange_weak(
                cur,
                new,
                std::sync::atomic::Ordering::Relaxed,
                std::sync::atomic::Ordering::Relaxed,
            ) {
                Ok(_) => return,
                Err(actual) => cur = actual,
            }
        }
    }

    /// DataPlane Worker 检查结果（无锁，O(1)）
    #[inline]
    pub fn poll_result(&self) -> Option<TaskResult> {
        self.result_queue.pop()
    }

    /// DataPlane Worker 批量收集结果
    #[inline]
    pub fn poll_results(&self, max: usize) -> Vec<TaskResult> {
        let mut results = Vec::with_capacity(max);
        for _ in 0..max {
            match self.result_queue.pop() {
                Some(result) => results.push(result),
                None => break,
            }
        }
        results
    }

    /// 创建 TaskRequest 构造器（channel_id 写入请求用于结果路由，提交时刻取当前单调时钟）
    #[inline]
    pub fn build_request(
        channel_id: u32,
        task_id: u64,
        domain_id: u32,
        payload: TaskPayload,
        deadline_ms: u64,
    ) -> TaskRequest {
        TaskRequest {
            task_id,
            domain_id,
            payload,
            deadline_ms,
            channel_id,
            submitted_at: std::time::Instant::now(),
        }
    }
}

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

    fn make_payload() -> TaskPayload {
        TaskPayload::DbQuery {
            sql_data: b"SELECT 1".to_vec(),
        }
    }

    fn make_request(task_id: u64) -> TaskRequest {
        TaskRequest {
            task_id,
            domain_id: 1,
            payload: make_payload(),
            deadline_ms: 1000,
            channel_id: 1,
            submitted_at: std::time::Instant::now(),
        }
    }

    fn make_result(task_id: u64) -> TaskResult {
        TaskResult {
            task_id,
            domain_id: 1,
            result: Ok(TaskOutput {
                data: b"OK".to_vec(),
                status_code: 200,
                elapsed_ms: 5,
            }),
        }
    }

    #[test]
    fn test_channel_submit_fetch() {
        let channel = TaskOffloadChannel::<256>::new(1);
        let req = make_request(1);
        channel.submit(req).unwrap();

        assert_eq!(channel.pending_count(), 1);
        assert!(!channel.is_empty());

        let fetched = channel.fetch().unwrap();
        assert_eq!(fetched.task_id, 1);
        assert!(channel.is_empty());
    }

    #[test]
    fn test_channel_complete_poll() {
        let channel = TaskOffloadChannel::<256>::new(1);

        channel.submit(make_request(1)).unwrap();
        channel.fetch().unwrap();
        channel.complete(make_result(1)).unwrap();

        let result = channel.poll_result().unwrap();
        assert_eq!(result.task_id, 1);
        assert!(result.result.is_ok());
        assert_eq!(channel.completed_count(), 1);
    }

    #[test]
    fn test_channel_batch_operations() {
        let channel = TaskOffloadChannel::<256>::new(1);

        let reqs: Vec<TaskRequest> = (0..10).map(make_request).collect();
        let submitted = channel.submit_batch(reqs).unwrap();
        assert_eq!(submitted, 10);
        assert_eq!(channel.pending_count(), 10);

        let fetched = channel.fetch_batch(5);
        assert_eq!(fetched.len(), 5);
        assert_eq!(channel.pending_count(), 10);

        let results: Vec<TaskResult> = fetched
            .iter()
            .map(|r| make_result(r.task_id))
            .collect();
        channel.complete_batch(results).unwrap();
        assert_eq!(channel.completed_count(), 5);
        assert_eq!(channel.pending_count(), 5);

        let polled = channel.poll_results(10);
        assert_eq!(polled.len(), 5);
    }

    #[test]
    fn test_channel_capacity_limit() {
        let channel = TaskOffloadChannel::<4>::new(1);
        for i in 0..4 {
            channel.submit(make_request(i)).unwrap();
        }
        let result = channel.submit(make_request(99));
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), TaskError::ChannelFull));
    }

    #[test]
    fn test_task_payload_methods() {
        let db = TaskPayload::DbQuery {
            sql_data: b"SELECT *".to_vec(),
        };
        assert_eq!(db.kind(), 1);
        assert!(db.is_write());

        let file = TaskPayload::FileRead {
            path_data: b"/etc/passwd".to_vec(),
        };
        assert_eq!(file.kind(), 2);
        assert!(!file.is_write());

        let compute = TaskPayload::ComplexCompute {
            input_data: vec![0; 100],
        };
        assert_eq!(compute.approximate_cost(), 10000);
    }

    #[test]
    fn test_task_error_display() {
        let err = TaskError::Timeout {
            task_id: 42,
            elapsed_ms: 5000,
        };
        assert!(format!("{err}").contains("42"));
        assert!(format!("{err}").contains("5000"));

        let err = TaskError::ChannelFull;
        assert!(format!("{err}").contains("full"));

        let err = TaskError::Cancelled { task_id: 7 };
        assert!(format!("{err}").contains("7"));
    }

    #[test]
    fn test_channel_stats() {
        let channel = TaskOffloadChannel::<256>::new(42);
        assert_eq!(channel.channel_id(), 42);
        assert_eq!(channel.capacity(), 256);
        assert_eq!(channel.pending_count(), 0);
        assert_eq!(channel.completed_count(), 0);

        channel.submit(make_request(1)).unwrap();
        assert_eq!(channel.pending_count(), 1);
    }

    #[test]
    fn test_channel_empty_operations() {
        let channel = TaskOffloadChannel::<16>::new(1);
        assert!(channel.is_empty());
        assert_eq!(channel.len(), 0);
        assert!(channel.fetch().is_none());
        assert!(channel.poll_result().is_none());

        let batch = channel.fetch_batch(5);
        assert!(batch.is_empty());

        let results = channel.poll_results(5);
        assert!(results.is_empty());
    }

    #[test]
    fn test_build_request() {
        let req = TaskOffloadChannel::<16>::build_request(
            7,
            100,
            2,
            TaskPayload::FileRead {
                path_data: b"/tmp/test".to_vec(),
            },
            5000,
        );
        assert_eq!(req.task_id, 100);
        assert_eq!(req.domain_id, 2);
        assert_eq!(req.deadline_ms, 5000);
        // channel_id 必须真实写入请求（不得丢弃）
        assert_eq!(req.channel_id, 7);
        // submitted_at 为构造时刻的单调时钟（刚构造，elapsed 必然极小）
        assert!(req.submitted_at.elapsed().as_secs() < 5);
    }

    #[test]
    fn test_complete_without_submit_no_underflow() {
        // 无配对 submit 的 complete：pending_count 饱和到 0，不得下溢回绕 u64::MAX
        let channel = TaskOffloadChannel::<16>::new(1);
        assert_eq!(channel.pending_count(), 0);
        channel.complete(make_result(1)).unwrap();
        assert_eq!(channel.pending_count(), 0);
        assert_eq!(channel.completed_count(), 1);

        // 批量路径同样饱和保护
        channel.complete_batch(vec![make_result(2), make_result(3)]).unwrap();
        assert_eq!(channel.pending_count(), 0);
        assert_eq!(channel.completed_count(), 3);
    }

    #[test]
    fn test_fetch_expired_task_produces_timeout() {
        // deadline 强制执行：已过截止时间的任务不再执行，产出 Timeout 结果并平衡账本
        let channel = TaskOffloadChannel::<16>::new(1);
        let mut req = make_request(1);
        req.deadline_ms = 1;
        req.submitted_at = std::time::Instant::now() - std::time::Duration::from_secs(60);
        assert_eq!(req.is_expired(), true);
        channel.submit(req).unwrap();

        // fetch 应跳过过期任务（无有效任务可取 → None），并产出 Timeout 结果
        assert!(channel.fetch().is_none());
        let result = channel.poll_result().unwrap();
        assert!(matches!(result.result, Err(TaskError::Timeout { .. })));
        // 账本平衡：pending 归零、completed 计 1
        assert_eq!(channel.pending_count(), 0);
        assert_eq!(channel.completed_count(), 1);
    }

    #[test]
    fn test_fetch_batch_skips_expired_keeps_valid() {
        // 批量取任务：过期任务被跳过转为 Timeout，有效任务正常返回
        let channel = TaskOffloadChannel::<16>::new(1);
        let mut expired = make_request(1);
        expired.deadline_ms = 1;
        expired.submitted_at = std::time::Instant::now() - std::time::Duration::from_secs(60);
        channel.submit(expired).unwrap();
        channel.submit(make_request(2)).unwrap();

        let batch = channel.fetch_batch(8);
        assert_eq!(batch.len(), 1);
        assert_eq!(batch[0].task_id, 2);
        // 过期任务转为 Timeout 结果
        let result = channel.poll_result().unwrap();
        assert!(matches!(result.result, Err(TaskError::Timeout { .. })));
        // 账本：过期任务已结清（pending 减 1），有效任务 task 2 仍待执行（pending 为 1）
        assert_eq!(channel.pending_count(), 1);
        assert_eq!(channel.completed_count(), 1);
    }

    #[test]
    fn test_expired_result_push_failure_does_not_unbalance_pending() {
        // SYS-008：结果队列满（背压）导致 Timeout 结果"静默丢失"时，
        // 不得回退 pending 账本——此前无论 push 成败都 dec_pending，
        // 结果丢失但 pending 下降，账本失衡（pending+completed 不守恒）。
        let channel = TaskOffloadChannel::<4>::new(1);
        // 填满结果队列（4 条占位；complete 的 dec_pending 饱和到 0，不影响账本）
        for i in 0..4 {
            channel.complete(make_result(i)).unwrap();
        }
        assert_eq!(channel.pending_count(), 0);

        // 提交一个过期任务：fetch 将产生 Timeout 结果，但结果队列已满 → push 失败
        let mut expired = make_request(99);
        expired.deadline_ms = 1;
        expired.submitted_at = std::time::Instant::now() - std::time::Duration::from_secs(60);
        channel.submit(expired).unwrap();
        assert_eq!(channel.pending_count(), 1);

        // 结果队列满 → 过期任务被弹出但 Timeout 结果无法入队
        assert!(channel.fetch().is_none());

        // 关键断言：结果 push 失败时 pending 不得被回退（仍为 1），
        // completed 不增加——账本对齐"结果未落地即不算完成"
        assert_eq!(
            channel.pending_count(),
            1,
            "结果队列满导致 Timeout 结果丢失时，pending 不得回退（账本失衡回归）"
        );
        assert_eq!(
            channel.completed_count(),
            4,
            "结果未落地不得计入 completed"
        );
    }
}