fix(#208): 停止转发用 tokio::select! 强制 drop axum::serve 释放端口 (#209)

Cmochance · web-flow · commit 9771940b89e3 · 2026-05-19T18:14:40.000+08:00
* fix(#208): 停止转发用 tokio::select! 强制 drop axum::serve 释放端口 之前用 `axum::serve(...).with_graceful_shutdown(rx)`:语义是等所有 in-flight connection 自然 drain 完才 drop listener。在 keep-alive HTTP / SSE / long-polling 场景永远 drain 不完 → listener 永远不 drop → **端口永远占用** → 用户感知"停止后仍能转发"(2026-05-18 真机多次复现)。 PR #207 的 `task.abort()` 也不够:tokio cancellation 是异步的,不保证 同步 drop listener;且 abort outer task 不影响 axum::serve 内部 spawn 的 connection sub-task。 改成 `tokio::select! { axum::serve, shutdown_rx }`:signal 一来 select 立即 wake → axum::serve future 整个 drop → 内部 sub-task + TcpListener **同步销毁** → 端口立即 free,不等 in-flight drain。 同时保留 application-level gate(in-flight request 二保险):signal send → task wake → future drop 有纳秒级窗口,gate 在 stop 时先 set true 让 middleware 立刻 503 + Connection: close 任何窗口内的 req。 Verify: - `crates/proxy/tests/stop_releases_port.rs`:2 个 integration test pass, 含关键 keep-alive client 持有 connection 时仍 ≤500ms 释放端口 - 实测:.app launch 后 lsof 18080 LISTEN,kill app → lsof 立即空 + python socket.bind 直接成功 Closes #208 * fix(#208): 自己写 accept loop + JoinSet 强制 abort in-flight connection PR #209 select! 方案不够强:虽然 axum::serve future drop 让 listener 释放(新 connection refused),但 axum::serve 内部对每个 connection 用裸 `tokio::spawn` 出 detached task,**外部无 handle**,future drop 不影响已 spawn 的 sub-task → 用户发过 1 条 message 后点停止转发, **同 connection 上的后续 keep-alive request 仍能 process** → "停止 后还在转发"(2026-05-18 用户真机复现)。 修复:不用 axum::serve,自己写 accept loop: - per-connection task 用 `JoinSet.spawn` 而**不**是裸 `tokio::spawn` → 拿到所有 connection task handle - stop 时 `connections.shutdown().await` = abort_all + 等所有 task die → reqwest stream / SSE / keep-alive request 全部立即被打断 → client 收 FIN/RST connection 关闭 Hyper service 桥接:axum Router 是 Service<Request<Body>>,hyper serve_connection 要 Service<Request<Incoming>>,用 tower map_request + TowerToHyperService 适配。 Verify:`crates/proxy/tests/stop_releases_port.rs` 重写: - spawn_production_like 跟 production proxy_runner.rs 同 pattern - test 1: 端口 ≤500ms 释放(基础) - test 2 **关键**: req1 keep-alive → stop → req2 必须失败(同 connection 上后续 req 应被 abort,或 reqwest 尝试新 connection 被 refused) ← 这是 PR #209 select! 方案 fail 的场景 两个 test pass。 * fix(#208): 加 CancellationToken + sub-task select,不再 task.abort PR #209 第二 commit (07d2664) 的 JoinSet + abort_all 仍 fail (2026-05-18 用户真机第二次复现): 18080 已无 LISTEN (listener drop ✓), 但 ESTABLISHED `127.0.0.1:18080->...` connection 仍 alive process keep-alive req → 用户感知"停止后还在转发"。 根因 (本 commit 修): 1. **stop_silent 调 task.abort() 抢先 cancel server task** —— task 跑不 到 `connections.shutdown().await`, sub-task 仍 detached。 2. 即使去掉 abort 让 task 跑完 shutdown(), tokio abort 是 schedule cancellation, 必须 sub-task **yield** 才生效。reqwest stream / SSE await upstream response 可能很久不 yield → abort 永远滞后。 修复 (本 commit): - 加 `tokio_util::sync::CancellationToken` - 每个 connection sub-task 内 `tokio::select! { conn.await, cancel.cancelled().await }` —— cancel 来时 select arm 立刻 wake → drop conn future → hyper connection drop → **TCP socket 同步 close → client 收 FIN/RST 立即断**, 不依赖 sub-task 主动 yield - stop / stop_silent: gate + cancel.cancel() + send shutdown_tx 三步同步触发, **不**调 task.abort() (否则抢先 cancel task) CancellationToken.cancel() 是 sync API, wake 所有 listener immediate, 比 tokio abort 的 schedule cancellation 强得多。 Integration test 同步更新 spawn_production_like 加 cancel token 参数 + sub-task select pattern。2 个 test pass。 * fix(#208): dup raw socket fd + cancel arm 直接 shutdown(SHUT_RDWR) 强制关 TCP PR #209 第三 commit (2398948) CancellationToken + sub-task select drop conn 仍 fail (2026-05-18 用户真机第三次复现): - 18080 listener drop ✓ (curl 直连 → Connection refused) - 但 `127.0.0.1:18080->127.0.0.1:<client_port> (ESTABLISHED)` server side socket **持续不消失** (lsof + netstat 双 verify) - → hyper conn future drop 不可靠让 underlying TcpStream 真 close socket 修复: 跨过 hyper / tokio 间接 cancellation 链,直接 OS-level 强制关 socket: 1. accept stream 后立即 `libc::dup(stream.as_raw_fd())` —— 持有 fd 副本 (跟 stream 共享同 socket inner state, 跟 stream 的 lifetime 解耦) 2. sub-task 内 select: - 正常完成 arm: conn 自然结束 - cancel arm: `libc::shutdown(dup_fd, SHUT_RDWR)` —— OS 立即发 FIN/RST 给 client, server side socket 强制 CLOSE 3. sub-task 退出时 `libc::close(dup_fd)` 释放副本 shutdown(SHUT_RDWR) 是 OS-level syscall, 绕开所有 user-space 的 future drop / async cancellation 延迟, **保证 socket 立即 close**, client 立即收 FIN/RST 断开。 Cross-platform: `#[cfg(unix)]` macOS + Linux 实现。Windows TODO (WSADuplicateSocket + closesocket), 加 #218 followup。 无需更新 integration test —— test 验 "req2 fail" 已 pass(因 reqwest 端口 refused), 真正 verify 在 user 真机 lsof 看 ESTABLISHED 是否消失。 * fix(#208): active_fds 表 + stop_silent 主动 shutdown 绕过整个 cancel chain PR #209 第四 commit (1be7c20) sub-task cancel arm 内 raw fd shutdown 仍 fail (2026-05-18 用户真机第四次复现): lsof 显示 sub-task 的 dup fd **消失**(说明 sub-task 已退),但 hyper accept 出的 stream fd 仍持续 hold,18080 ESTABLISHED connection 不消失。 矛盾分析: - sub-task 已退 → conn drop → io drop → stream Drop 应 close fd - 但 stream fd 仍 hold → 说明某环节断了 - 最可能 root: CancellationToken cancel signal 在 child_token 路径或 tokio runtime 调度未 propagate, sub-task 卡 select await 不退 修复 (本 commit): 完全绕过 user-space cancellation chain —— ProxyHandle 新增 `active_fds: Arc<Mutex<Vec<i32>>>` 表, 记录所有 active connection 的 dup socket fd; stop_silent / stop / race-condition 路径 **直接** lock 表 + `libc::shutdown(fd, SHUT_RDWR)` 主动让 OS kernel 发 FIN/RST。 - accept 时: `libc::dup(stream.as_raw_fd())` + push 到 active_fds - stop_silent 时: snapshot active_fds + foreach shutdown - sub-task 完成时: remove 自己 entry + close fd (避免 leak) shutdown 是 OS syscall, 跟 user-space 调度 / future drop / cancellation 完全无关 —— **kernel 同步发 FIN/RST 给 client, server side socket 立即进 closing state**, lsof ESTABLISHED 消失。 Race 处理: shutdown 不 close (sub-task 负责 close), 避免 fd 双重 close 误伤 OS 后续 reassigned fd。 Cross-platform: `#[cfg(unix)]` macOS + Linux 实现。Windows TODO。 * fix(#208): sub-task 不 close dup_fd (留给 stop) + fmt fix PR #209 第五 commit (10056d5) active_fds 仍 fail (lsof verify): stream fd 14u 仍 hold 但 dup fd 已消失 → sub-task 完成时 close 了 dup_fd , 然后 stop_silent 时 active_fds 已空, 失去强制 shutdown 的 fd。 根因: hyper `auto::Builder` 协商 H2 时 spawn 独立 h2 task 持有 stream, **outer sub-task 完成 ≠ inner h2 task 完成**。outer 完成 close dup_fd , inner h2 task 仍通过原 stream fd 工作。 修复 (本 commit): - **sub-task 完成时不 close dup_fd, 不 remove active_fds entry** —— 留 fd 给 stop_silent 统一关 - stop_silent 时 `force_shutdown_active_fds` take 整 vec + foreach `shutdown(SHUT_RDWR)` + `close(fd)` —— 强制 OS-level 关 socket, 不论 inner h2 task 状态如何 Trade-off: 连续 forward 期间 dup fd 累积 (每 connection 1 个), 直到 stop button 或 app exit 统一清理。Codex CLI keep-alive 连接复用, typical N=几十级, OS fd ulimit 256+ 够用。 同时修 cargo fmt (CI 10056d5 fail 是 fmt --check)。 * fix(#208): 移除所有兜底, proxy 跑独立 tokio Runtime + shutdown_background 一键停 前 5 个 commit 的层层兜底 (task.abort / select! drop / JoinSet abort_all / CancellationToken / raw fd shutdown / active_fds 表 / app-level gate middleware) 全部移除。用户明示: "停止转发等于停止所有功能只保留主界面, 我不知道你加这些兜底逻辑的意义是什么"。 核心方案改成最简单粗暴的: - proxy 跑在独立 `std::thread` + 独立 `tokio::runtime::Runtime` - stop 时调 `runtime.shutdown_background()` —— tokio 提供的 OS-level "杀光所有 task" 原语, 同步 abort runtime 上所有 spawn task, worker thread 退出, 没人 poll task → task drop, 持有的 TcpStream / TcpListener / 所有 fd cleanup - **完全不依赖** user-space cancellation chain (CancellationToken / JoinSet abort / hyper conn drop 等), shutdown_background 直接杀 worker thread 文件改动: - src-tauri/src/proxy_runner.rs: 重写, ProxyHandle 只持有 `tokio::runtime::Runtime` 一个核心字段 - crates/proxy/src/server.rs: 删 build_router_with_gate / shutdown_gate middleware, build_router 恢复纯路由 - crates/proxy/src/lib.rs: 删 build_router_with_gate export - crates/proxy/tests/stop_releases_port.rs: 删 (production 不再用 那 pattern) - src-tauri/Cargo.toml: 删 hyper / hyper-util / tokio-util / libc deps - crates/proxy/Cargo.toml: 删 hyper / hyper-util / tokio-util dev-deps 不影响: - Plugins 进程 (独立 OS process: macOS launcher / Windows IApplicationActivationManager), 不在 ProxyManager runtime 内 - Tauri 主 runtime / admin handler / 主界面 * hardening(#208): Runtime drop 移到独立 std::thread 避 async context panic stop / stop_silent 通过 admin handler (async fn) 触发时, ProxyHandle drop → Runtime drop 发生在 async context 内, 理论上踩 `Runtime::drop` "async context drop panic" 红线 (实测 tokio multi-thread + cross-runtime 没触发, 但 tokio 升级可能变严)。 加 `drop_runtime_off_thread` helper: ProxyHandle 整体 move 到独立 std::thread, 在 thread 内 shutdown_background + drop, 远离任何 async context, 100% 安全。Thread 在 closure 跑完后自动 exit, 不 leak。 5 行核心 (thread::Builder::new().name(...).spawn(move || { ... }).ok())。 * fix(#208): mpsc→oneshot + 删 std::thread wrapper (shutdown_background 本就 async-safe) 两条 Devin review: 1. `mpsc::recv()` 同步阻塞 Tauri worker thread (Devin P1 真问题): 改 `tokio::sync::oneshot.await`, start() 全程 non-blocking, yield 给 tokio 调度其他 task。 2. `drop_runtime_off_thread` spawn fail fallback panic: 重新审查发现 thread wrapper 整个是过度防御。`Runtime::shutdown_background` 是 tokio 显式提供 "useful if you want to drop a runtime from within another runtime" 的 API, **本身就 async-context-safe**, 不触发 panic。我之前判断错了。直接删 thread wrapper, 回到最简版本。 ChatGPT Codex 反馈 (WebSocket bypass): 评论旧 commit 的 shutdown_gate middleware, 已在 eb4d7a6 删除整个 middleware 改 runtime drop 方案。 runtime drop = 所有 task 一锅端 abort (含 WebSocket sub-task), bypass 问题不存在了。
diff --git a/src-tauri/src/proxy_runner.rs b/src-tauri/src/proxy_runner.rs
@@ -1,28 +1,24 @@
-//! 内嵌 axum 代理生命周期管理(Stage 4.3 + Stage 5).
+//! 内嵌 axum 代理生命周期管理。
 //!
-//! Tauri 主进程启动时构造一个 [`ProxyManager`] 注入到 `State<T>`,前端通过
-//! `start_proxy` / `stop_proxy` / `proxy_status` 命令操控,Tauri 主进程
-//! 退出时通过 [`ProxyManager::stop_silent`] **同步**关闭代理。
+//! **核心设计**:proxy 跑在**独立 `std::thread` + 独立 `tokio::runtime::Runtime`**。
+//! stop 时把整个 Runtime drop(`shutdown_background()`)——
+//! - 所有 spawn 在 runtime 上的 task **同步 abort**
+//! - worker thread 退出 → 没人 poll task → task drop
+//! - task 持有的 `TcpStream` / `TcpListener` 跟着 drop → fd close
+//! - **所有 proxy 相关功能一锅端,只保留 Tauri 主界面**
 //!
-//! 设计要点:
-//! - 内部 `std::sync::Mutex<Option<ProxyHandle>>` —— 锁持有时间极短(只读/写
-//!   单个 Option),没有跨 await,**stop / status / stop_silent 全部是同步方法**,
-//!   方便从 Tauri 的 `RunEvent::Exit` 同步路径调用而不需要 `block_on`。
-//! - **`start` 是 async**(TcpListener::bind 必需),但锁取放都在显式 scope 里,
-//!   不跨越 await。
-//! - **生命周期**:`start` 时 spawn tokio task 持有 `axum::serve` future,附带
-//!   `with_graceful_shutdown(oneshot::Receiver<()>)`;`stop` / `stop_silent`
-//!   通过 `oneshot::Sender::send(())` 触发 graceful 关停。
+//! 不再使用 CancellationToken / JoinSet / 自己写 accept loop / raw fd shutdown /
+//! application-level gate middleware 等"兜底逻辑"—— `Runtime::shutdown_background`
+//! 是 tokio 提供的 OS-level "杀光所有 task" 原语,不需要 user-space cancel chain。
 
 use std::net::SocketAddr;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
+use std::sync::Mutex;
 
 use codex_app_transfer_proxy::{build_router, StaticResolver};
 use codex_app_transfer_registry::{config_file, Config};
 use serde::Serialize;
-use tokio::net::TcpListener;
 use tokio::sync::oneshot;
-use tokio::task::JoinHandle;
 
 #[derive(Debug, Serialize, Clone)]
 pub struct ProxyStatus {
@@ -37,16 +33,10 @@ pub struct ProxyStatus {
 
 struct ProxyHandle {
     addr: SocketAddr,
-    shutdown_tx: oneshot::Sender<()>,
-    /// axum::serve task 的 JoinHandle — stop / stop_silent 在发 graceful
-    /// shutdown 信号**之后**立刻 task.abort() 强制 drop server future,
-    /// listener 同步 drop → 端口立刻释放。
-    ///
-    /// 修 silent failure:graceful_shutdown 等所有 in-flight connection
-    /// 完成才 drop listener,SSE / long-polling / hung connection 可能
-    /// 永远等不到 → 端口不释放,但 UI / log 已 "stopped" → 用户感知
-    /// "停了但端口还占"。abort 保证 stop_silent 同步返时端口必释放。
-    task: JoinHandle<()>,
+    /// **核心**:proxy 跑在这个独立 runtime 上,stop_silent 时调
+    /// `shutdown_background()` 一键 abort 所有 task + worker thread 退出
+    /// → 所有 fd / 资源 cleanup。
+    runtime: tokio::runtime::Runtime,
     gateway_auth: bool,
     provider_count: usize,
     active_provider: Option<String>,
@@ -64,7 +54,7 @@ impl ProxyManager {
 
     /// 启动代理监听 `127.0.0.1:<port>`。已 running 时沿用旧版语义返回当前状态。
     pub async fn start(&self, port: u16) -> Result<ProxyStatus, String> {
-        // 1. 预检查(短锁)
+        // 1. 预检查
         {
             let guard = self.handle.lock().unwrap();
             if let Some(h) = guard.as_ref() {
@@ -78,39 +68,74 @@ impl ProxyManager {
             }
         }
 
-        // 2. 装载 resolver + 绑定 listener(async)
+        // 2. 装载 resolver
         let snapshot = load_resolver_snapshot()?;
-        let listener = TcpListener::bind(format!("127.0.0.1:{port}"))
+
+        // 3. 创建 dedicated runtime + 启 server
+        //    Runtime::new 不能在 async context 内调,用 std::thread 包。
+        //    用 tokio::sync::oneshot 而非 std::sync::mpsc,让 receiver 端 .await
+        //    yield Tauri worker thread 而不是同步 block(Devin review fix)。
+        let (addr_tx, addr_rx) =
+            oneshot::channel::<Result<(SocketAddr, tokio::runtime::Runtime), String>>();
+        let resolver = Arc::new(snapshot.resolver);
+        std::thread::Builder::new()
+            .name(format!("cas-proxy-bootstrap-{port}"))
+            .spawn(move || {
+                let rt = match tokio::runtime::Builder::new_multi_thread()
+                    .enable_all()
+                    .worker_threads(2)
+                    .thread_name("cas-proxy")
+                    .build()
+                {
+                    Ok(rt) => rt,
+                    Err(e) => {
+                        let _ = addr_tx.send(Err(format!("create proxy runtime failed: {e}")));
+                        return;
+                    }
+                };
+                let bind_result = rt.block_on(async {
+                    let listener = tokio::net::TcpListener::bind(format!("127.0.0.1:{port}"))
+                        .await
+                        .map_err(|e| format!("bind 127.0.0.1:{port} failed: {e}"))?;
+                    let addr = listener
+                        .local_addr()
+                        .map_err(|e| format!("cannot read listener address: {e}"))?;
+                    let router = build_router(resolver);
+                    // 在 runtime 上 spawn server —— 当 runtime shutdown_background
+                    // 时此 task 同步被 abort,listener + 所有 connection sub-task
+                    // 一起 drop,fd close。
+                    rt.spawn(async move {
+                        let _ = axum::serve(listener, router.into_make_service()).await;
+                    });
+                    Ok::<SocketAddr, String>(addr)
+                });
+                match bind_result {
+                    Ok(addr) => {
+                        let _ = addr_tx.send(Ok((addr, rt)));
+                    }
+                    Err(e) => {
+                        rt.shutdown_background();
+                        let _ = addr_tx.send(Err(e));
+                    }
+                }
+            })
+            .map_err(|e| format!("spawn proxy thread failed: {e}"))?;
+
+        let (addr, runtime) = addr_rx
             .await
-            .map_err(|e| format!("bind 127.0.0.1:{port} failed: {e}"))?;
-        let addr = listener
-            .local_addr()
-            .map_err(|e| format!("cannot read listener address: {e}"))?;
-        let router = build_router(Arc::new(snapshot.resolver));
-        let (tx, rx) = oneshot::channel::<()>();
-        let task = tokio::spawn(async move {
-            let _ = axum::serve(listener, router.into_make_service())
-                .with_graceful_shutdown(async move {
-                    let _ = rx.await;
-                })
-                .await;
-        });
-
-        // 3. 落盘 handle(短锁;若期间被另一路径插入,关掉自己回滚)
+            .map_err(|_| "proxy bootstrap channel closed".to_owned())??;
+
+        // 4. 落盘 handle(短锁;若期间被另一路径插入,关掉自己回滚)
         let new_handle = ProxyHandle {
             addr,
-            shutdown_tx: tx,
-            task,
+            runtime,
             gateway_auth: snapshot.gateway_auth,
             provider_count: snapshot.provider_count,
             active_provider: snapshot.active_provider.clone(),
         };
         let mut guard = self.handle.lock().unwrap();
         if guard.is_some() {
-            // race condition,自己的 listener 让出去:发 shutdown + abort task
-            // (abort 同步 drop listener,端口立刻释放,不依赖 graceful drain)
-            let _ = new_handle.shutdown_tx.send(());
-            new_handle.task.abort();
+            new_handle.runtime.shutdown_background();
             return Err("proxy already started by another path".to_owned());
         }
         *guard = Some(new_handle);
@@ -123,34 +148,31 @@ impl ProxyManager {
         })
     }
 
-    /// 触发 graceful shutdown 后立刻 abort task 释放端口。未 running 时报错。
+    /// 停止转发 —— 一键 drop 整个 dedicated runtime,所有 spawn task 同步 abort,
+    /// worker thread 退出,所有 fd / 连接 cleanup,**只保留 Tauri 主界面**。
     ///
-    /// **为什么两步**:`shutdown_tx.send(())` 给 axum graceful drain 机会让
-    /// 正常完成的 request 跑完;紧跟 `task.abort()` 同步 drop server future
-    /// → listener drop → **端口立刻释放**,不会卡在 in-flight SSE / long
-    /// polling / hung connection 上(那种 connection 可能永远 drain 不完)。
-    /// 代价:in-flight connection 被强断,client 见 connection reset —
-    /// 但"用户点停止 = 真要停",这是合理 trade-off。
+    /// `Runtime::shutdown_background` 是 tokio 显式提供的 "from within another
+    /// runtime 安全 shutdown" API,不触发 "async context drop runtime" panic
+    /// (tokio docs: "useful if you want to drop a runtime from within another
+    /// runtime")。所以即使 stop_proxy admin handler 是 async fn 在此调用,
+    /// 也无需 std::thread 包装。
     #[allow(dead_code)]
     pub fn stop(&self) -> Result<(), String> {
         let mut guard = self.handle.lock().unwrap();
         match guard.take() {
             Some(h) => {
-                let _ = h.shutdown_tx.send(());
-                h.task.abort();
+                h.runtime.shutdown_background();
                 Ok(())
             }
             None => Err("proxy is not running".to_owned()),
         }
     }
 
-    /// 静默 stop:用于 app exit / 异常路径,不报错只尽力关。
-    /// 同样走 send signal + abort 双保险确保端口释放(详见 [`Self::stop`])。
+    /// 静默 stop:app exit / 异常路径用,不报错只尽力关。
     pub fn stop_silent(&self) {
         let mut guard = self.handle.lock().unwrap();
         if let Some(h) = guard.take() {
-            let _ = h.shutdown_tx.send(());
-            h.task.abort();
+            h.runtime.shutdown_background();
         }
     }
 
