From 6fc6d5628fad7e2d5fc72f03a583ba196c1e29e1 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Wed, 5 Aug 2026 11:14:53 +0800 Subject: [PATCH] =?UTF-8?q?perf(gateway):=20send=5Fwith=5Fretry=20?= =?UTF-8?q?=E5=AF=B9=20ChannelFull=20=E4=B8=8D=E9=87=8D=E8=AF=95=EF=BC=8C?= =?UTF-8?q?=E5=8A=A0=E9=80=9F=E9=98=9F=E5=88=97=E6=B6=88=E8=B4=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ws 连接队列满时，sender task 原本重试 3 次浪费 1+2+4=7s，期间 channel 队列堆积。新增 ChannelError::ChannelFull 变体区分队列满错误，CliChannel::send 在 try_send Full 时返回该变体；send_with_retry 遇到 ChannelFull 立即返回不重试，sender task 记 warn（预期背压丢弃）而非 error。 - base.rs: 新增 ChannelFull 变体及 Display 实现 - cli.rs: try_send 错误区分 Full（ChannelFull，不可重试）与 Closed（SendError，可重试） - outbound_dispatcher.rs: send_with_retry 短路返回 ChannelFull；run_sender_task 区分日志级别；新增 TestChannel.with_channel_full 及单测验证不重试（500ms 阈值内返回 + send 仅调用一次） --- src/channels/base.rs | 4 ++ src/channels/cli.rs | 19 +++++--- src/gateway/outbound_dispatcher.rs | 73 +++++++++++++++++++++++++++--- 3 files changed, 84 insertions(+), 12 deletions(-) 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" + ); + } }