diff --git a/src/channels/base.rs b/src/channels/base.rs index e929399..1e40dd6 100644 --- a/src/channels/base.rs +++ b/src/channels/base.rs @@ -8,6 +8,9 @@ pub enum ChannelError { ConfigError(String), ConnectionError(String), SendError(String), + /// Channel 内部队列已满——不可重试,立即丢弃。 + /// 重试只会浪费退避时间并阻塞 channel 队列消费。 + ChannelFull, BusError(String), Other(String), } @@ -18,6 +21,7 @@ impl std::fmt::Display for ChannelError { ChannelError::ConfigError(s) => write!(f, "Config error: {}", s), ChannelError::ConnectionError(s) => write!(f, "Connection error: {}", s), ChannelError::SendError(s) => write!(f, "Send error: {}", s), + ChannelError::ChannelFull => write!(f, "Channel queue full"), ChannelError::BusError(s) => write!(f, "Bus error: {}", s), ChannelError::Other(s) => write!(f, "Error: {}", s), } diff --git a/src/channels/cli.rs b/src/channels/cli.rs index 1c18ede..6528218 100644 --- a/src/channels/cli.rs +++ b/src/channels/cli.rs @@ -102,13 +102,20 @@ impl Channel for CliChannel { // 使用 try_send 避免阻塞 dispatcher——dispatcher 是单线程顺序处理, // 若 writer task 卡在 ws_sender.send() 上,send().await 会阻塞, - // 导致所有连接的实时消息被阻塞。try_send 满时立即返回错误, - // dispatcher 记录后继续处理下一条消息。 + // 导致所有连接的实时消息被阻塞。try_send 满时返回 ChannelFull + // (不可重试,立即丢弃);关闭时返回 SendError(可重试)。 for outbound in ws_outbound_from_outbound_message(&msg) { - connection - .sender - .try_send(outbound) - .map_err(|_| ChannelError::SendError("CLI websocket sender closed or full".to_string()))?; + match connection.sender.try_send(outbound) { + Ok(()) => {} + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + return Err(ChannelError::ChannelFull); + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + return Err(ChannelError::SendError( + "CLI websocket sender closed".to_string(), + )); + } + } } Ok(()) diff --git a/src/gateway/outbound_dispatcher.rs b/src/gateway/outbound_dispatcher.rs index 0aeac41..756d124 100644 --- a/src/gateway/outbound_dispatcher.rs +++ b/src/gateway/outbound_dispatcher.rs @@ -89,12 +89,23 @@ impl OutboundDispatcher { break; }; - if let Err(error) = Self::send_with_retry(&*channel, msg).await { - tracing::error!( - channel = %channel_name, - error = %error, - "Failed to send message after retries" - ); + match Self::send_with_retry(&*channel, msg).await { + Ok(()) => {} + Err(ChannelError::ChannelFull) => { + // 队列满是不可重试的——send_with_retry 已跳过重试。 + // 记 warn 而非 error:这是预期的背压丢弃。 + tracing::warn!( + channel = %channel_name, + "Message dropped: channel queue full" + ); + } + Err(error) => { + tracing::error!( + channel = %channel_name, + error = %error, + "Failed to send message after retries" + ); + } } } // dispatcher 退出时取消所有 sender task @@ -177,6 +188,9 @@ impl OutboundDispatcher { /// 发送消息,失败时按 `[1, 2, 4]` 秒间隔重试。 /// + /// `ChannelFull` 不可重试——队列满时重试只会浪费退避时间并阻塞 + /// channel 队列消费。立即返回让 sender task 尽快处理下一条消息。 + /// /// 仅在单个 channel 的 sender task 内执行——重试 sleep 只阻塞 /// 该 channel 的发送,不影响其他 channel。 async fn send_with_retry( @@ -186,6 +200,8 @@ impl OutboundDispatcher { for (attempt_index, delay) in RETRY_DELAYS_SECS.iter().enumerate() { match channel.send(msg.clone()).await { Ok(()) => return Ok(()), + // 队列满:不可重试,立即返回 + Err(ChannelError::ChannelFull) => return Err(ChannelError::ChannelFull), Err(error) if attempt_index < RETRY_DELAYS_SECS.len() - 1 => { tracing::warn!( attempt = attempt_index + 1, @@ -217,6 +233,8 @@ mod tests { received: Arc, delay_ms: u64, fail_first_n: u32, + /// 始终返回 `ChannelFull`,用于验证 `send_with_retry` 不重试队列满错误。 + always_full: bool, call_count: Arc, } @@ -227,6 +245,7 @@ mod tests { received: Arc::new(AtomicU32::new(0)), delay_ms: 0, fail_first_n: 0, + always_full: false, call_count: Arc::new(AtomicU32::new(0)), } } @@ -240,6 +259,11 @@ mod tests { self.fail_first_n = n; self } + + fn with_channel_full(mut self) -> Self { + self.always_full = true; + self + } } #[async_trait] @@ -262,6 +286,12 @@ mod tests { async fn send(&self, _msg: OutboundMessage) -> Result<(), ChannelError> { let count = self.call_count.fetch_add(1, Ordering::SeqCst); + + // 队列满:不可重试错误,用于验证 send_with_retry 立即返回 + if self.always_full { + return Err(ChannelError::ChannelFull); + } + if (count as u32) < self.fail_first_n { return Err(ChannelError::SendError("simulated failure".to_string())); } @@ -467,4 +497,35 @@ mod tests { // 直接 abort dispatcher 及其 sender task 即可清理。 dispatcher_handle.abort(); } + + #[tokio::test] + async fn test_send_with_retry_no_retry_on_channel_full() { + // 验证:send_with_retry 遇到 ChannelFull 应立即返回,不重试。 + // 若错误地重试,会 sleep 1+2+4=7 秒,测试将在超时阈值内失败。 + let channel = TestChannel::new("full").with_channel_full(); + let call_count = channel.call_count.clone(); + + let msg = make_message("full", "chat-1", "dropped"); + + // 500ms 阈值:远小于首次重试间隔 1s,足以区分"立即返回"与"至少一次重试" + let result = tokio::time::timeout(Duration::from_millis(500), async { + OutboundDispatcher::send_with_retry(&channel, msg).await + }) + .await + .expect("send_with_retry should return immediately on ChannelFull, not retry"); + + // 必须返回 ChannelFull 错误 + assert!( + matches!(result, Err(ChannelError::ChannelFull)), + "expected ChannelFull error, got {:?}", + result + ); + + // send 只被调用一次——证明没有重试 + assert_eq!( + call_count.load(Ordering::SeqCst), + 1, + "send should be called exactly once when ChannelFull is returned" + ); + } }