perf(gateway): send_with_retry 对 ChannelFull 不重试,加速队列消费

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 仅调用一次)
This commit is contained in:
oudecheng 2026-08-05 11:14:53 +08:00
parent deab3699fe
commit 6fc6d5628f
3 changed files with 84 additions and 12 deletions

View File

@ -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),
}

View File

@ -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(())

View File

@ -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<AtomicU32>,
delay_ms: u64,
fail_first_n: u32,
/// 始终返回 `ChannelFull`,用于验证 `send_with_retry` 不重试队列满错误。
always_full: bool,
call_count: Arc<AtomicU32>,
}
@ -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"
);
}
}