From 8ecbfca939a1ed09ba92cd00444a110b28b5bd8c Mon Sep 17 00:00:00 2001
From: lbl <1791778603@qq.com>
Date: Fri, 21 Aug 2026 02:25:40 +0800
Subject: [PATCH] =?UTF-8?q?fix(quic):=20=E5=85=A5=E7=AB=99=20channel=20?=
 =?UTF-8?q?=E6=BB=A1=E6=97=B6=E4=B8=A2=E5=8C=85=E8=80=8C=E9=9D=9E=E9=98=BB?=
 =?UTF-8?q?=E5=A1=9E=E6=95=B4=E6=9D=A1=E6=8E=A5=E6=94=B6=E5=BE=AA=E7=8E=AF?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

QuicDataInbound::send 此前用异步 send 等待 channel 容量，消费端
（QUIC endpoint 驱动）处理不过来时，256 容量的 channel 一满就
阻塞整条接收循环，所有对端的入站流量被头部阻塞。

改为 try_send：满则丢包并告警（IP 包语义下可接受），关闭才报错。

测试：test_send_drops_when_channel_full（满时不阻塞）、
test_send_errors_when_channel_closed（关闭后报错）。
---
 .../quic_over/enhanced_io/enhanced_inbound.rs | 55 +++++++++++++++++--
 1 file changed, 51 insertions(+), 4 deletions(-)

diff --git a/vnt-core/src/enhanced_tunnel/quic_over/enhanced_io/enhanced_inbound.rs b/vnt-core/src/enhanced_tunnel/quic_over/enhanced_io/enhanced_inbound.rs
index ae6183c..6174fea 100644
--- a/vnt-core/src/enhanced_tunnel/quic_over/enhanced_io/enhanced_inbound.rs
+++ b/vnt-core/src/enhanced_tunnel/quic_over/enhanced_io/enhanced_inbound.rs
@@ -19,10 +19,18 @@ pub struct QuicDataInbound {
 }
 impl QuicDataInbound {
     pub async fn send(&self, data: Bytes, addr: Ipv4Addr) -> anyhow::Result<()> {
-        self.sender
-            .send((data, addr))
-            .await
-            .map_err(|_e| anyhow!("quic data inbound error"))
+        match self.sender.try_send((data, addr)) {
+            Ok(()) => Ok(()),
+            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
+                // 消费端处理不过来时丢包：channel 满不能阻塞整条 QUIC 接收循环，
+                // 否则一个慢消费者会卡住所有对端的入站流量
+                log::warn!("quic data inbound channel full, dropping packet from {addr}");
+                Ok(())
+            }
+            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
+                Err(anyhow!("quic data inbound error"))
+            }
+        }
     }
 }
 impl Debug for QuicInnerInboundReceiver {
@@ -91,3 +99,42 @@ impl QuicInnerInboundReceiver {
         }
     }
 }
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    /// channel 满时 send 必须立即返回（丢包），不能阻塞接收循环
+    #[tokio::test]
+    async fn test_send_drops_when_channel_full() {
+        let (inbound, _receiver) = create_enhanced_inbound();
+        // 填满 channel（容量 256）
+        for _ in 0..256 {
+            inbound
+                .send(Bytes::from_static(b"x"), Ipv4Addr::LOCALHOST)
+                .await
+                .unwrap();
+        }
+        // 再发送：旧实现会永久阻塞，修复后应立即返回 Ok（丢包）
+        let rs = tokio::time::timeout(
+            Duration::from_millis(200),
+            inbound.send(Bytes::from_static(b"y"), Ipv4Addr::LOCALHOST),
+        )
+        .await;
+        assert!(rs.is_ok(), "send blocked on full channel");
+        rs.unwrap().unwrap();
+    }
+
+    /// channel 关闭后 send 返回错误
+    #[tokio::test]
+    async fn test_send_errors_when_channel_closed() {
+        let (inbound, receiver) = create_enhanced_inbound();
+        drop(receiver);
+        let rs = inbound
+            .send(Bytes::from_static(b"x"), Ipv4Addr::LOCALHOST)
+            .await;
+        assert!(rs.is_err());
+    }
+}