merge: E2E 排队修复（64KB 窗口 + 拥塞跳帧）

agent/src/capture.rs

@@ -275,7 +275,8 @@ pub fn desktop_size() -> Result<(u32, u32)> {
 
 /// 采集主循环：独立线程运行；连接断开（通道关闭）时退出。
 /// 通道为 watch 单槽（latest-wins）：发送即覆盖，网络慢时不产生积压。
-pub fn run(tx: Sender<Option<Vec<u8>>>, keyreq: Arc<AtomicBool>) {
+/// congestion：发送侧 write 超时置位 → 跳过当前帧（不编码不发送），注入率自适应降速。
+pub fn run(tx: Sender<Option<Vec<u8>>>, keyreq: Arc<AtomicBool>, congestion: Arc<AtomicBool>) {
     let start = Instant::now();
     loop {
         let mut dup = match Duplication::new() {
@@ -333,6 +334,7 @@ pub fn run(tx: Sender<Option<Vec<u8>>>, keyreq: Arc<AtomicBool>) {
         let mut stat_frames = 0u64;
         let mut stat_convert_ms = 0u64;
         let mut stat_encode_ms = 0u64;
+        let mut stat_skipped = 0u64; // 拥塞跳帧计数
 
         macro_rules! emit {
             ($payload:expr) => {{
@@ -357,6 +359,16 @@ pub fn run(tx: Sender<Option<Vec<u8>>>, keyreq: Arc<AtomicBool>) {
                 Ok(true) => {
                     // 主控端请求关键帧（解码异常恢复）
                     let force_key = keyreq.swap(false, Ordering::Relaxed);
+                    // 拥塞反馈：发送侧排队深 → 跳过本帧（NVENC 少输入一帧，参考链连续；
+                    // 解码端收到的帧序 = 编码端实际输出序，无花屏风险）
+                    if congestion.swap(false, Ordering::Relaxed) {
+                        stat_skipped += 1;
+                        let el = iter_start.elapsed();
+                        if el < min_frame {
+                            std::thread::sleep(min_frame - el);
+                        }
+                        continue;
+                    }
                     match &mut encoder {
                         EncoderKind::Nvenc(enc) => {
                             let enc_start = Instant::now();
@@ -412,12 +424,14 @@ pub fn run(tx: Sender<Option<Vec<u8>>>, keyreq: Arc<AtomicBool>) {
                     }
                     if stat_frames > 0 && stat_frames % 120 == 0 {
                         eprintln!(
-                            "[capture] avg convert {}ms, encode {}ms/frame (last 120)",
+                            "[capture] avg convert {}ms, encode {}ms/frame, skipped {} (last 120)",
                             stat_convert_ms / 120,
-                            stat_encode_ms / 120
+                            stat_encode_ms / 120,
+                            stat_skipped
                         );
                         stat_convert_ms = 0;
                         stat_encode_ms = 0;
+                        stat_skipped = 0;
                     }
                     // 帧间隔节流到 ~60fps
                     let el = iter_start.elapsed();

agent/src/main.rs

@@ -43,12 +43,13 @@ async fn main() -> Result<()> {
 
     // 传输参数调优：
     // - 收发窗口放大，避免高 RTT 下吞吐被 窗口/RTT 死锁（默认 512KB 流窗口）
-    // - 发送缓冲上限 512KB：QUIC 内部排队 ≈ 3 帧 @5Mbps，E2E 延迟天花板 ~150ms；
+    // - 发送缓冲上限 64KB：QUIC 内部排队硬上限（实测 WiFi ~3Mbps 吞吐 → E2E ≤ ~170ms；
-    //   配合上游 latest-wins 丢帧，超量部分在进 QUIC 前就被丢弃
+    //   突发吞吐 = 64KB/7ms ≈ 72Mbps，不限制带宽）。注意此窗口为连接级，流 2 输入/Ping
+    //   数据量极小不受影响。配合下方拥塞反馈跳帧，把排队压在数帧以内
     let mut transport = quinn::TransportConfig::default();
     transport.receive_window(quinn::VarInt::from_u32(16 * 1024 * 1024));
     transport.stream_receive_window(quinn::VarInt::from_u32(8 * 1024 * 1024));
-    transport.send_window(512 * 1024);
+    transport.send_window(64 * 1024);
     let mut server_config = quinn::ServerConfig::with_single_cert(vec![cert], key.into())?;
     server_config.transport_config(Arc::new(transport));
 
@@ -87,7 +88,10 @@ async fn main() -> Result<()> {
     let (tx, mut frames_rx) = watch::channel::<Option<Vec<u8>>>(None);
     let keyreq = Arc::new(AtomicBool::new(false));
     let keyreq_capture = keyreq.clone();
-    std::thread::spawn(move || capture::run(tx, keyreq_capture));
+    // 拥塞反馈标志：发送侧 write 超过单帧预算(20ms) → 置位，采集线程跳帧降注入率
+    let congestion = Arc::new(AtomicBool::new(false));
+    let congestion_capture = congestion.clone();
+    std::thread::spawn(move || capture::run(tx, keyreq_capture, congestion_capture));
 
     // 上行控制消息（Pong）经通道交给主循环写流 2，避免多任务争抢
     let (ctl_tx, mut ctl_rx) = mpsc::unbounded_channel::<ControlMsg>();
@@ -127,7 +131,14 @@ async fn main() -> Result<()> {
                     break; // 采集线程退出
                 }
                 if let Some(data) = frames_rx.borrow().clone() {
-                    if send.write_all(&data).await.is_err() {
+                    let wt = std::time::Instant::now();
+                    let ok = send.write_all(&data).await.is_ok();
+                    // write 耗时 > 单帧预算(16.7ms 的 1.2 倍) → QUIC 窗口排队深，
+                    // 通知采集线程跳帧，把注入率降到网络实际吞吐水平
+                    if wt.elapsed() > std::time::Duration::from_millis(20) {
+                        congestion.store(true, Ordering::Relaxed);
+                    }
+                    if !ok {
                         break;
                     }
                 }

viewer/src/main.rs

@@ -362,7 +362,7 @@ fn session(
     };
 
     let mut window = Window::new(
-        "mvp-remote v0.6-HUD (Esc 退出)",
+        "mvp-remote v0.7-LowLatency (Esc 退出)",
         w,
         h,
         WindowOptions {