From 63904a0bf5cc510b7d74ebeaee240106c4205aea Mon Sep 17 00:00:00 2001 From: fylorn <249551762+fylorn@users.noreply.github.com> Date: Fri, 25 Sep 2026 08:42:18 +0800 Subject: [PATCH] test: stop handing core ports that were just released CI failed once in tw-control's remote tests with `gw.listen.port_taken` on 127.0.0.1:34393 (narrowing_allow_from_closes_the_connections_it_no_longer_lets_in). free_port() bound 127.0.0.1:0, read the port and closed the socket, and the test then wrote that number into the config for core to bind. A released port goes back to the system's ephemeral range, which parallel tests draw from whenever they bind port 0 or open a connection. Linux picks those ports at random, so now and then another test took the number before core bound it. The same "bind 0, read the port, release it" step was in 38 places that start a gateway with tw_gateway::serve, and in the hot-reload, listen and remote-port tests. - tw_gateway::serve binds before it returns and hands back the address it bound. The 38 call sites pass 127.0.0.1:0 and use the returned address, so the system picks the port and nobody can take it in between. serve is only used by tests. - Where the number has to exist before anything binds it, tests take it from spare_port(): the remote control port (the config refuses 0 on purpose, config.remote_port_zero), the port PUT /listen saves, and the hot-reload tests that move between two ports or keep one port across addresses. spare_port() picks at random from REMOTE_PORT_RANGE (20000-32000), below the ephemeral ranges of Linux, macOS and Windows, so the system never hands these ports out itself. It holds each number with a UDP socket for the life of the process. UDP and TCP ports are separate, so core still binds the TCP port, but another test, or another test process running at the same time, cannot get the same number. - The mirror case: a port that should have nobody listening (an unreachable upstream or proxy) was also a released ephemeral port, which a parallel test could start listening on. Those come from the same range now, held the same way. Reproduced on macOS, where ephemeral ports are handed out in sequence and the race does not show without help: a background process walks the ephemeral range with bind(127.0.0.1:0) while holding the last 4000 sockets. Running the 20 affected test binaries at once, six rounds, origin/main failed 12 and 13 of 120 runs (remote, listen, hotreload; the remote failures are the same port_taken as on CI). With this change, 0 of 120 in each of three such runs. Sixteen copies of the remote test binary at once, 20 rounds: 0 of 320. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + crates/tw-control/Cargo.toml | 2 + crates/tw-control/tests/common/mod.rs | 50 ++++++ crates/tw-control/tests/listen.rs | 37 ++-- crates/tw-control/tests/live_state.rs | 16 +- crates/tw-control/tests/remote.rs | 27 ++- crates/tw-gateway/src/l1.rs | 19 ++- crates/tw-gateway/src/listen.rs | 19 +++ crates/tw-gateway/src/server.rs | 14 +- crates/tw-gateway/tests/chatgpt.rs | 8 +- crates/tw-gateway/tests/common/mod.rs | 49 ++++++ crates/tw-gateway/tests/conversion.rs | 8 +- crates/tw-gateway/tests/dialect.rs | 9 +- crates/tw-gateway/tests/endings.rs | 18 +- crates/tw-gateway/tests/headers_and_paths.rs | 16 +- crates/tw-gateway/tests/hotreload.rs | 58 ++----- crates/tw-gateway/tests/live.rs | 8 +- crates/tw-gateway/tests/m1_acceptance.rs | 8 +- crates/tw-gateway/tests/m5_redact.rs | 17 +- crates/tw-gateway/tests/m5_request_guards.rs | 9 +- crates/tw-gateway/tests/m5_toolwall.rs | 18 +- crates/tw-gateway/tests/models.rs | 9 +- crates/tw-gateway/tests/notices.rs | 8 +- crates/tw-gateway/tests/oauth.rs | 7 +- crates/tw-gateway/tests/passthrough.rs | 167 +++++++------------ crates/tw-gateway/tests/ws.rs | 24 ++- 26 files changed, 322 insertions(+), 304 deletions(-) create mode 100644 crates/tw-control/tests/common/mod.rs create mode 100644 crates/tw-gateway/tests/common/mod.rs diff --git a/Cargo.lock b/Cargo.lock index eee56b07..f37751b9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2544,6 +2544,7 @@ dependencies = [ "hyper", "hyper-util", "libc", + "rand 0.10.2", "reqwest", "serde", "serde_json", diff --git a/crates/tw-control/Cargo.toml b/crates/tw-control/Cargo.toml index ab8273a6..d7e9cd55 100644 --- a/crates/tw-control/Cargo.toml +++ b/crates/tw-control/Cargo.toml @@ -58,4 +58,6 @@ tw-api = { workspace = true } tw-types = { workspace = true } # 网卡那条断言要对着本机真实的清单比，不写死某个平台的名字 tw-config = { workspace = true } +# 测试挑端口：从远程控制端口那一段里随机挑（见 tests/common） +rand = { workspace = true } tokio = { workspace = true, features = ["rt", "macros"] } diff --git a/crates/tw-control/tests/common/mod.rs b/crates/tw-control/tests/common/mod.rs new file mode 100644 index 00000000..78b505a8 --- /dev/null +++ b/crates/tw-control/tests/common/mod.rs @@ -0,0 +1,50 @@ +//! 几个测试文件共用的小工具。 + +use std::net::SocketAddr; + +/// 一个空着的端口号：**系统不会自己把它分出去**，也不会再给第二个测试。 +/// +/// 号要在谁都没绑它之前就定下来时用它：写进配置或请求（远程控制端口 —— 配置 +/// 不收 0，见 `config.remote_port_zero`；`PUT /listen` 要保存的端口），或者要它 +/// 一直没人听（连不上的代理）。只要网关起来就行的地方用不着它： +/// `tw_gateway::serve` 自己绑 0，把真的地址交回来。 +/// +/// **不能绑 0 拿号再放掉。**放掉的号回到系统的临时端口段（Linux 默认 +/// 32768–60999，macOS 和 Windows 49152–65535），并行的测试绑 0、发起连接都从 +/// 那一段取号：要绑它的，core 绑之前号就被拿走，报一个和代码无关的 +/// `gw.listen.port_taken`；要它没人听的，别的测试在上面起了服务，「连不上」就 +/// 连上了。这里从那一段下面挑，和 core 挑远程控制端口同一段。 +/// +/// **号用 UDP 占着，直到进程退出。**UDP 和 TCP 是两套端口：占着 UDP 的这个号， +/// TCP 的照样绑得上；而别的用例、同时在跑的别的测试进程（两个工作目录各跑一遍） +/// 来拿同一个号时，系统说它已经被占了。只在进程里记账的话，后一种管不到。 +pub fn spare_port() -> u16 { + use std::sync::{Mutex, PoisonError}; + + static HELD: Mutex> = Mutex::new(Vec::new()); + for _ in 0..1000 { + let port = rand::random_range(tw_config::REMOTE_PORT_RANGE); + let Ok(hold) = std::net::UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], port))) else { + continue; + }; + if free(port, [127, 0, 0, 1]) && free(port, [0, 0, 0, 0]) { + HELD.lock() + .unwrap_or_else(PoisonError::into_inner) + .push(hold); + return port; + } + } + panic!("no spare port left in {:?}", tw_config::REMOTE_PORT_RANGE); +} + +/// 这个端口的 TCP 此刻绑不绑得上。别的程序正占着的（开发机上可能真有一个 core +/// 在这一段里听）、系统留作他用的（Windows 会成段地保留端口）算绑不上；别的错误 +/// 说明这台机器出了别的问题，直接报出来。 +fn free(port: u16, ip: [u8; 4]) -> bool { + use std::io::ErrorKind; + match std::net::TcpListener::bind(SocketAddr::from((ip, port))) { + Ok(_) => true, + Err(e) if matches!(e.kind(), ErrorKind::AddrInUse | ErrorKind::PermissionDenied) => false, + Err(e) => panic!("binding port {port} to try it failed: {e}"), + } +} diff --git a/crates/tw-control/tests/listen.rs b/crates/tw-control/tests/listen.rs index c20cccc8..78c83f4f 100644 --- a/crates/tw-control/tests/listen.rs +++ b/crates/tw-control/tests/listen.rs @@ -11,6 +11,9 @@ use axum::http::{Request, StatusCode}; use tower::ServiceExt; use tw_control::{ConfigManager, ControlState}; +mod common; +use common::spare_port; + struct Bed { dir: tempfile::TempDir, app: axum::Router, @@ -78,14 +81,6 @@ async fn call( ) } -fn free_port() -> u16 { - std::net::TcpListener::bind("127.0.0.1:0") - .unwrap() - .local_addr() - .unwrap() - .port() -} - fn yaml(port: u16) -> String { format!( "version: 1 @@ -103,8 +98,8 @@ listen: #[tokio::test] async fn saving_writes_what_was_chosen_and_leaves_no_trace_of_defaults() { - let b = bed(&yaml(free_port())); - let port = free_port(); + let b = bed(&yaml(spare_port())); + let port = spare_port(); let (st, v) = call( &b.app, "PUT", @@ -143,7 +138,7 @@ async fn saving_writes_what_was_chosen_and_leaves_no_trace_of_defaults() { async fn a_port_in_use_is_refused_and_nothing_is_written() { // **这个接口存在的理由。**写进去之后才发现绑不上，网关守着旧地址而配置 // 文件说着新地址 —— 两边从那一刻起各说各的 - let start = free_port(); + let start = spare_port(); let b = bed(&yaml(start)); let squatter = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let taken = squatter.local_addr().unwrap().port(); @@ -167,13 +162,13 @@ async fn a_port_in_use_is_refused_and_nothing_is_written() { #[tokio::test] async fn an_interface_that_is_not_there_is_refused_by_name() { - let b = bed(&yaml(free_port())); + let b = bed(&yaml(spare_port())); let before = b.file(); let (st, v) = call( &b.app, "PUT", "/listen", - serde_json::json!({ "bind": "en97", "port": free_port(), "allow_from": [] }), + serde_json::json!({ "bind": "en97", "port": spare_port(), "allow_from": [] }), ) .await; assert_eq!(st, StatusCode::CONFLICT, "{v}"); @@ -200,7 +195,7 @@ async fn an_interface_that_is_not_there_is_refused_by_name() { &b.app, "PUT", "/listen", - serde_json::json!({ "bind": "192.168.1.5\twifi", "port": free_port(), "allow_from": [] }), + serde_json::json!({ "bind": "192.168.1.5\twifi", "port": spare_port(), "allow_from": [] }), ) .await; assert_eq!(st, StatusCode::BAD_REQUEST, "{v}"); @@ -221,7 +216,7 @@ async fn an_interface_that_is_not_there_is_refused_by_name() { async fn the_status_follows_the_listener_after_a_save() { // 界面左下角的地址以前是启动时记的一次：改了端口，网关已经在新端口上 // 服务，状态里还写着旧的 - let p1 = free_port(); + let p1 = spare_port(); let b = bed(&yaml(p1)); let gw = b.gw.clone(); let want = tw_config::try_parse(&b.file()) @@ -236,7 +231,7 @@ async fn the_status_follows_the_listener_after_a_save() { let (_, s) = call(&b.app, "GET", "/status", serde_json::Value::Null).await; assert_eq!(s["gateway_addr"], format!("127.0.0.1:{p1}")); - let p2 = free_port(); + let p2 = spare_port(); let (st, v) = call( &b.app, "PUT", @@ -260,12 +255,12 @@ async fn the_status_follows_the_listener_after_a_save() { #[tokio::test] async fn a_stale_version_is_refused_like_any_other_edit() { - let b = bed(&yaml(free_port())); + let b = bed(&yaml(spare_port())); let (st, _) = call( &b.app, "PUT", "/listen", - serde_json::json!({ "bind": "loopback", "port": free_port(), "base_version": "nope", "allow_from": [] }), + serde_json::json!({ "bind": "loopback", "port": spare_port(), "base_version": "nope", "allow_from": [] }), ) .await; assert_eq!(st, StatusCode::CONFLICT); @@ -296,8 +291,8 @@ listen: async fn an_empty_allow_list_is_written_down_and_the_default_one_is_not() { // 不写 = 默认名单，所以空的必须写成 `[]`；和默认名单一样的不写， // 配置文件不因为存了一次就多出几行 - let b = bed(&yaml(free_port())); - let port = free_port(); + let b = bed(&yaml(spare_port())); + let port = spare_port(); let save = |allow: serde_json::Value| serde_json::json!({ "bind": "all", "port": port, "allow_from": allow }); let (st, v) = call(&b.app, "PUT", "/listen", save(serde_json::json!([]))).await; assert_eq!(st, StatusCode::OK, "{v}"); @@ -319,7 +314,7 @@ async fn an_empty_allow_list_is_written_down_and_the_default_one_is_not() { #[tokio::test] async fn interfaces_come_one_per_name() { // 配置里按名字存：同一张网卡列两行，选第二行等于选第一行 - let b = bed(&yaml(free_port())); + let b = bed(&yaml(spare_port())); let (st, v) = call(&b.app, "GET", "/interfaces", serde_json::Value::Null).await; assert_eq!(st, StatusCode::OK, "{v}"); let names: Vec<&str> = v diff --git a/crates/tw-control/tests/live_state.rs b/crates/tw-control/tests/live_state.rs index bb1cc74f..6a0b58b9 100644 --- a/crates/tw-control/tests/live_state.rs +++ b/crates/tw-control/tests/live_state.rs @@ -13,6 +13,9 @@ use axum::http::{Request, StatusCode}; use tower::ServiceExt; use tw_control::{ConfigManager, ControlState}; +mod common; +use common::spare_port; + fn control( d: &tempfile::TempDir, yaml: &str, @@ -52,11 +55,9 @@ async fn get(app: &axum::Router, path: &str) -> (StatusCode, serde_json::Value) /// 把网关真的跑起来：凭据和代理的状态只有真实转发才会碰到。 async fn serve(gw: tw_gateway::AppState) -> SocketAddr { - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(gw, addr).await.unwrap() }); + let addr = tw_gateway::serve(gw, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(60)).await; addr } @@ -76,10 +77,9 @@ async fn ask(gw: SocketAddr) -> u16 { .as_u16() } -/// 一个确定没人在听的地址。 +/// 一个确定没人在听的地址（见 [`spare_port`]）。 fn dead_addr() -> SocketAddr { - let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - l.local_addr().unwrap() + SocketAddr::from(([127, 0, 0, 1], spare_port())) } // ---------------------------------------------------------------- 还在跑的请求 diff --git a/crates/tw-control/tests/remote.rs b/crates/tw-control/tests/remote.rs index ffb33b8c..6d63b383 100644 --- a/crates/tw-control/tests/remote.rs +++ b/crates/tw-control/tests/remote.rs @@ -14,6 +14,9 @@ use tw_api::control::{Address, ControlKey}; use tw_control::{ConfigManager, ControlState}; use tw_link::LinkError; +mod common; +use common::spare_port; + const KEY: &str = "c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00"; fn yaml(port: u16, enabled: bool, allow: &str) -> String { @@ -26,14 +29,6 @@ fn yaml_at(bind: &str, port: u16, enabled: bool, allow: &str) -> String { ) } -fn free_port() -> u16 { - std::net::TcpListener::bind("127.0.0.1:0") - .unwrap() - .local_addr() - .unwrap() - .port() -} - struct Bed { dir: tempfile::TempDir, state: ControlState, @@ -176,7 +171,7 @@ async fn send( /// 放行的来源握得上手，状态里说得出开着、听在哪。 #[tokio::test] async fn an_allowed_source_gets_in_and_the_status_says_where_it_listens() { - let port = free_port(); + let port = spare_port(); let b = bed(yaml(port, true, "[127.0.0.1]")).await; let l = b.until(|l| l.addr.is_some()).await; assert_eq!(l.addr.unwrap().port(), port); @@ -196,7 +191,7 @@ async fn an_allowed_source_gets_in_and_the_status_says_where_it_listens() { /// 不是「钥匙不对」。 #[tokio::test] async fn a_source_outside_allow_from_is_closed_without_a_word() { - let port = free_port(); + let port = spare_port(); let b = bed(yaml(port, true, "[10.0.0.0/8]")).await; let l = b.until(|l| l.addr.is_some()).await; let mut raw = tokio::net::TcpStream::connect(l.addr.unwrap()) @@ -229,7 +224,7 @@ async fn a_source_outside_allow_from_is_closed_without_a_word() { /// 同一个来源一分钟里握手失败五次，之后连对的钥匙也直接关掉。 #[tokio::test] async fn five_wrong_keys_from_one_source_bench_it() { - let port = free_port(); + let port = spare_port(); let b = bed(yaml(port, true, "[127.0.0.1]")).await; let addr = b.until(|l| l.addr.is_some()).await.addr.unwrap(); let wrong = "1".repeat(64); @@ -259,7 +254,7 @@ async fn five_wrong_keys_from_one_source_bench_it() { /// 同样的事从本机的通道做得了。 #[tokio::test] async fn a_remote_connection_cannot_stop_the_core_take_diagnostics_or_move_its_own_door() { - let port = free_port(); + let port = spare_port(); let b = bed(yaml(port, true, "[127.0.0.1]")).await; let addr = b.until(|l| l.addr.is_some()).await.addr.unwrap(); let (mut r, _) = open_tcp(addr, KEY).await.unwrap(); @@ -350,7 +345,7 @@ async fn a_remote_connection_cannot_stop_the_core_take_diagnostics_or_move_its_o /// 旧的并说为什么，本机的通道照常。 #[tokio::test] async fn it_follows_the_configuration_live() { - let p1 = free_port(); + let p1 = spare_port(); let b = bed(yaml(p1, true, "[127.0.0.1]")).await; let a1 = b.until(|l| l.addr.is_some()).await.addr.unwrap(); let (mut s, conn) = open_tcp(a1, KEY).await.unwrap(); @@ -374,7 +369,7 @@ async fn it_follows_the_configuration_live() { )); // 换个端口打开 - let p2 = free_port(); + let p2 = spare_port(); b.rewrite(&yaml(p2, true, "[127.0.0.1]")).await; let a2 = b .until(|l| l.addr.is_some_and(|a| a.port() == p2)) @@ -421,7 +416,7 @@ async fn a_port_that_cannot_be_bound_at_start_does_not_take_the_local_channel_do /// 却仍放行它的，连接照常。 #[tokio::test] async fn narrowing_allow_from_closes_the_connections_it_no_longer_lets_in() { - let port = free_port(); + let port = spare_port(); let b = bed(yaml(port, true, "[127.0.0.1]")).await; let addr = b.until(|l| l.addr.is_some()).await.addr.unwrap(); let (mut s, conn) = open_tcp(addr, KEY).await.unwrap(); @@ -458,7 +453,7 @@ async fn narrowing_allow_from_closes_the_connections_it_no_longer_lets_in() { /// 旧的自己。要能当场换过去，不报错。 #[tokio::test] async fn the_same_port_moves_between_all_and_a_specific_address_live() { - let port = free_port(); + let port = spare_port(); let b = bed(yaml_at("all", port, true, "[127.0.0.1]")).await; let l = b.until(|l| l.addr.is_some()).await; assert!(l.addr.unwrap().ip().is_unspecified(), "{l:?}"); diff --git a/crates/tw-gateway/src/l1.rs b/crates/tw-gateway/src/l1.rs index 6f149fbe..07613cf6 100644 --- a/crates/tw-gateway/src/l1.rs +++ b/crates/tw-gateway/src/l1.rs @@ -888,14 +888,25 @@ mod tests { /// 并行测试里别的用例就可能绑到它 —— 这个文件里有七处在绑随机端口， /// 其中几处还起了真的 SOCKS5 服务器。于是「死端口」活了过来， /// 测试看到的不是 TCP 拒绝而是一次握手。Windows 上撞得尤其勤。 + /// 放掉之前先连一次确认它是死的也不够：确认完了照样会被绑去。 /// - /// 所以拿到之后先自己连一次确认它是死的，活着就换一个。 + /// 所以从系统不会自己分出去的那一段里挑（和 core 挑远程控制端口同一段）， + /// 号用 UDP 占着直到进程退出。UDP 和 TCP 是两套端口，占着 UDP 的号不妨碍 + /// TCP 上没人听；而别的测试挑号时也先占 UDP 的同一个号（见集成测试的 + /// `spare_port`），占不到就换，不会拿它去绑。 async fn dead_port() -> SocketAddr { + use std::sync::{Mutex, PoisonError}; + static HELD: Mutex> = Mutex::new(Vec::new()); for _ in 0..64 { - let l = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let a = l.local_addr().unwrap(); - drop(l); + let port = rand::random_range(tw_config::REMOTE_PORT_RANGE); + let a = SocketAddr::from(([127, 0, 0, 1], port)); + let Ok(hold) = std::net::UdpSocket::bind(a) else { + continue; + }; if tokio::net::TcpStream::connect(a).await.is_err() { + HELD.lock() + .unwrap_or_else(PoisonError::into_inner) + .push(hold); return a; } } diff --git a/crates/tw-gateway/src/listen.rs b/crates/tw-gateway/src/listen.rs index deb7f9a3..54b4d219 100644 --- a/crates/tw-gateway/src/listen.rs +++ b/crates/tw-gateway/src/listen.rs @@ -220,6 +220,25 @@ fn snapshot(bound: &[Bound], error: Option) -> Listening { } } +/// 起服务，**不跟配置走**，绑上了就交回真的地址。见 [`crate::server::serve`]。 +pub(crate) async fn serve_detached( + state: AppState, + want: SocketAddr, +) -> std::io::Result { + let listener = TcpListener::bind(want) + .await + .map_err(|e| std::io::Error::new(e.kind(), format!("{}", bind_failure(want, &e))))?; + let bound = start(&state, want, listener)?; + let actual = bound.actual; + state.set_listening(snapshot(std::slice::from_ref(&bound), None)); + // 丢掉 `Bound` 就停止接新连接：把它留在一个不会结束的任务里 + tokio::spawn(async move { + let _bound = bound; + std::future::pending::<()>().await + }); + Ok(actual) +} + /// 起服务。`follow` 为真时**跟着配置里的监听地址走**（「温」那一级热重载）。 /// /// 命令行给了 `--port` 时 `follow` 为假：那是一个显式的覆盖，不该被配置 diff --git a/crates/tw-gateway/src/server.rs b/crates/tw-gateway/src/server.rs index 7c373263..9a1b247c 100644 --- a/crates/tw-gateway/src/server.rs +++ b/crates/tw-gateway/src/server.rs @@ -218,9 +218,17 @@ struct Sender { key: Option, } -/// 在一个地址上起服务，**不跟配置走**。测试和命令行 `--port` 用它。 -pub async fn serve(state: AppState, addr: std::net::SocketAddr) -> std::io::Result<()> { - crate::listen::serve_at(state, vec![addr], false).await +/// 在一个地址上起服务，**不跟配置走**。绑上了就返回，交回真的地址；服务在后台 +/// 一直跑到运行时结束。测试用它。 +/// +/// **端口给 0，由系统挑**，从返回值拿到真的端口。别先绑一个 0 端口拿号、放掉、 +/// 再把号交给这里：放掉之后号回到系统手里，并行的测试绑 0、发起连接都会被分到 +/// 它，赶在这里绑之前拿走就是一次和代码无关的「端口被占」。 +pub async fn serve( + state: AppState, + addr: std::net::SocketAddr, +) -> std::io::Result { + crate::listen::serve_detached(state, addr).await } /// 上游回了话的一跳：`served` 或者 `status`。 diff --git a/crates/tw-gateway/tests/chatgpt.rs b/crates/tw-gateway/tests/chatgpt.rs index be66fb6d..3f2f8db6 100644 --- a/crates/tw-gateway/tests/chatgpt.rs +++ b/crates/tw-gateway/tests/chatgpt.rs @@ -237,11 +237,9 @@ async fn start_gateway( let state = tw_gateway::AppState::new(cfg).unwrap(); let rx = state.bus.subscribe(); let handed = state.clone(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(60)).await; (addr, handed, rx) } diff --git a/crates/tw-gateway/tests/common/mod.rs b/crates/tw-gateway/tests/common/mod.rs new file mode 100644 index 00000000..ebc7250e --- /dev/null +++ b/crates/tw-gateway/tests/common/mod.rs @@ -0,0 +1,49 @@ +//! 几个测试文件共用的小工具。 + +use std::net::SocketAddr; + +/// 一个空着的端口号：**系统不会自己把它分出去**，也不会再给第二个测试。 +/// +/// 号要在谁都没绑它之前就定下来时用它：写进配置（换端口、同一个端口换地址）， +/// 或者要它一直没人听（连不上的上游）。只要网关起来就行的地方用不着它： +/// `tw_gateway::serve` 自己绑 0，把真的地址交回来。 +/// +/// **不能绑 0 拿号再放掉。**放掉的号回到系统的临时端口段（Linux 默认 +/// 32768–60999，macOS 和 Windows 49152–65535），并行的测试绑 0、发起连接都从 +/// 那一段取号：要绑它的，网关绑之前号就被拿走，报一个和代码无关的 +/// `gw.listen.port_taken`；要它没人听的，别的测试在上面起了服务，「连不上」就 +/// 连上了。这里从那一段下面挑，和 core 挑远程控制端口同一段。 +/// +/// **号用 UDP 占着，直到进程退出。**UDP 和 TCP 是两套端口：占着 UDP 的这个号， +/// TCP 的照样绑得上；而别的用例、同时在跑的别的测试进程（两个工作目录各跑一遍） +/// 来拿同一个号时，系统说它已经被占了。只在进程里记账的话，后一种管不到。 +pub fn spare_port() -> u16 { + use std::sync::{Mutex, PoisonError}; + + static HELD: Mutex> = Mutex::new(Vec::new()); + for _ in 0..1000 { + let port = rand::random_range(tw_config::REMOTE_PORT_RANGE); + let Ok(hold) = std::net::UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], port))) else { + continue; + }; + if free(port, [127, 0, 0, 1]) && free(port, [0, 0, 0, 0]) { + HELD.lock() + .unwrap_or_else(PoisonError::into_inner) + .push(hold); + return port; + } + } + panic!("no spare port left in {:?}", tw_config::REMOTE_PORT_RANGE); +} + +/// 这个端口的 TCP 此刻绑不绑得上。别的程序正占着的（开发机上可能真有一个 core +/// 在这一段里听）、系统留作他用的（Windows 会成段地保留端口）算绑不上；别的错误 +/// 说明这台机器出了别的问题，直接报出来。 +fn free(port: u16, ip: [u8; 4]) -> bool { + use std::io::ErrorKind; + match std::net::TcpListener::bind(SocketAddr::from((ip, port))) { + Ok(_) => true, + Err(e) if matches!(e.kind(), ErrorKind::AddrInUse | ErrorKind::PermissionDenied) => false, + Err(e) => panic!("binding port {port} to try it failed: {e}"), + } +} diff --git a/crates/tw-gateway/tests/conversion.rs b/crates/tw-gateway/tests/conversion.rs index 06d18d4d..c6652a9d 100644 --- a/crates/tw-gateway/tests/conversion.rs +++ b/crates/tw-gateway/tests/conversion.rs @@ -106,11 +106,9 @@ async fn gateway_with( }; let state = tw_gateway::AppState::new(cfg).unwrap(); let rx = state.bus.subscribe(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; (addr, rx) } diff --git a/crates/tw-gateway/tests/dialect.rs b/crates/tw-gateway/tests/dialect.rs index 5782811d..17bbcd47 100644 --- a/crates/tw-gateway/tests/dialect.rs +++ b/crates/tw-gateway/tests/dialect.rs @@ -113,12 +113,9 @@ async fn start_gateway( }; let state = tw_gateway::AppState::new(cfg).unwrap(); let rx = state.bus.subscribe(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - let s2 = state.clone(); - tokio::spawn(async move { tw_gateway::serve(s2, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; (addr, rx) } diff --git a/crates/tw-gateway/tests/endings.rs b/crates/tw-gateway/tests/endings.rs index 49a69c6e..25dc0dca 100644 --- a/crates/tw-gateway/tests/endings.rs +++ b/crates/tw-gateway/tests/endings.rs @@ -20,6 +20,9 @@ use tokio::sync::broadcast::Receiver; use tw_api::Event; use tw_config::{Client, Config, Provider, Security, SecurityMode}; +mod common; +use common::spare_port; + /// Anthropic 流的第一帧。**输入和缓存读在这里就是齐的**，输出是个占位的 /// 1 —— 累计输出要等流的末尾才报。 /// 客户端要的模型名。**每一种结局都要带着它**（见 `model_of`） @@ -209,11 +212,9 @@ async fn serve_with_bus(cfg: Config) -> (SocketAddr, Receiver, tw_observe let state = tw_gateway::AppState::new(cfg).unwrap(); let bus = state.bus.clone(); let events = bus.subscribe(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; (addr, events, bus) } @@ -612,11 +613,8 @@ async fn a_websocket_session_the_client_closes_is_finished() { #[tokio::test] async fn a_websocket_whose_upstream_cannot_be_reached_is_failed() { - // 绑一个端口再立刻放掉，拿到一个确定没人在听的端口 - let dead = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; + // 一个确定没人在听的端口（见 `spare_port`） + let dead = SocketAddr::from(([127, 0, 0, 1], spare_port())); let (gw, mut events) = serve(cfg(provider(dead))).await; let mut c = ws_connect(gw).await; // 网关会先说一句为什么，再关掉连接 diff --git a/crates/tw-gateway/tests/headers_and_paths.rs b/crates/tw-gateway/tests/headers_and_paths.rs index 7000bd62..ded9f9df 100644 --- a/crates/tw-gateway/tests/headers_and_paths.rs +++ b/crates/tw-gateway/tests/headers_and_paths.rs @@ -62,11 +62,9 @@ async fn start_gateway(provider: Provider) -> SocketAddr { ..Default::default() }; let state = tw_gateway::AppState::new(cfg).unwrap(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; addr } @@ -259,11 +257,9 @@ async fn a_disabled_key_is_refused_and_says_it_was_disabled_on_purpose() { ..Default::default() }; let state = tw_gateway::AppState::new(cfg).unwrap(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let resp = reqwest::Client::new() diff --git a/crates/tw-gateway/tests/hotreload.rs b/crates/tw-gateway/tests/hotreload.rs index 1cca2897..d8a0c0e5 100644 --- a/crates/tw-gateway/tests/hotreload.rs +++ b/crates/tw-gateway/tests/hotreload.rs @@ -14,6 +14,9 @@ use axum::extract::State; use axum::routing::any; use tw_config::{Client, Config, Provider}; +mod common; +use common::spare_port; + async fn counting_upstream(name: &'static str) -> (SocketAddr, Arc) { let hits = Arc::new(AtomicUsize::new(0)); let h = hits.clone(); @@ -60,12 +63,9 @@ fn provider(name: &str, at: SocketAddr) -> Provider { } async fn serve(state: tw_gateway::AppState) -> SocketAddr { - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - let s = state.clone(); - tokio::spawn(async move { tw_gateway::serve(s, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; addr } @@ -493,15 +493,7 @@ async fn changing_the_port_actually_moves_the_listener() { // 反应」比报错难查得多。 let (up, _) = counting_upstream("a").await; let mut c = cfg(vec![provider("a", up)], vec![]); - // 先占两个端口拿号，再放掉 - let (p1, p2) = { - let a = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let b = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - ( - a.local_addr().unwrap().port(), - b.local_addr().unwrap().port(), - ) - }; + let (p1, p2) = (spare_port(), spare_port()); c.listen.gateway.port = p1; let state = tw_gateway::AppState::new(c.clone()).unwrap(); let mut events = state.bus.subscribe(); @@ -568,12 +560,11 @@ async fn a_port_already_taken_keeps_the_old_listener_and_says_why() { // 同一份配置拉起来、再失败 —— 用户改了个端口，换来的是所有客户端断线 let (up, _) = counting_upstream("a").await; let mut c = cfg(vec![provider("a", up)], vec![]); - let squatter = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let taken = squatter.local_addr().unwrap().port(); - let p1 = { - let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - l.local_addr().unwrap().port() - }; + // 占着的程序后面会退出、把端口让给网关：放掉的那一下不能有别人插进来， + // 所以它占的也是一个系统不会转手的号 + let taken = spare_port(); + let squatter = std::net::TcpListener::bind(("127.0.0.1", taken)).unwrap(); + let p1 = spare_port(); c.listen.gateway.port = p1; let state = tw_gateway::AppState::new(c.clone()).unwrap(); let mut events = state.bus.subscribe(); @@ -626,14 +617,7 @@ async fn a_request_in_flight_survives_the_listener_being_rebuilt() { a }; let mut c = cfg(vec![provider("slow", slow)], vec![]); - let (p1, p2) = { - let a = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let b = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - ( - a.local_addr().unwrap().port(), - b.local_addr().unwrap().port(), - ) - }; + let (p1, p2) = (spare_port(), spare_port()); c.listen.gateway.port = p1; let state = tw_gateway::AppState::new(c.clone()).unwrap(); following(&state, &c).await; @@ -667,14 +651,6 @@ async fn a_request_in_flight_survives_the_listener_being_rebuilt() { assert!(r.text().await.unwrap().contains("slow")); } -fn free_port() -> u16 { - std::net::TcpListener::bind("127.0.0.1:0") - .unwrap() - .local_addr() - .unwrap() - .port() -} - fn bind(raw: &str) -> tw_config::Bind { serde_yaml_ng::from_str(raw).unwrap() } @@ -700,7 +676,7 @@ async fn switching_between_all_interfaces_and_one_address_on_the_same_port_takes // 报「端口被占」然后守着旧的，要重启才生效 let (up, _) = counting_upstream("a").await; let mut c = cfg(vec![provider("a", up)], vec![]); - let port = free_port(); + let port = spare_port(); c.listen.gateway.port = port; c.listen.gateway.bind = bind("all"); let state = tw_gateway::AppState::new(c.clone()).unwrap(); @@ -732,7 +708,7 @@ async fn the_pre_save_check_does_not_call_our_own_port_taken() { // 不该回答「被别的程序占了」—— 那样这份设置根本存不进去 let (up, _) = counting_upstream("a").await; let mut c = cfg(vec![provider("a", up)], vec![]); - c.listen.gateway.port = free_port(); + c.listen.gateway.port = spare_port(); c.listen.gateway.bind = bind("all"); let state = tw_gateway::AppState::new(c.clone()).unwrap(); following(&state, &c).await; @@ -767,7 +743,7 @@ async fn a_request_in_flight_survives_the_old_listener_giving_way() { a }; let mut c = cfg(vec![provider("slow", slow)], vec![]); - let port = free_port(); + let port = spare_port(); c.listen.gateway.port = port; c.listen.gateway.bind = bind("all"); let state = tw_gateway::AppState::new(c.clone()).unwrap(); @@ -807,7 +783,7 @@ async fn when_the_new_address_is_really_taken_the_old_one_is_taken_back() { // // 场景：0.0.0.0:p 在听，换到 [::1]:p，而 [::1]:p 被别人占着。同端口，所以 // 走「先让出来」那条路；让了也没用，于是回到 0.0.0.0:p - let port = free_port(); + let port = spare_port(); let Ok(squatter) = std::net::TcpListener::bind(("::1", port)) else { eprintln!("no IPv6 loopback here; skipping"); return; diff --git a/crates/tw-gateway/tests/live.rs b/crates/tw-gateway/tests/live.rs index 50af585c..1c5c46d0 100644 --- a/crates/tw-gateway/tests/live.rs +++ b/crates/tw-gateway/tests/live.rs @@ -67,11 +67,9 @@ async fn serve(upstream: SocketAddr) -> (SocketAddr, tw_gateway::live::Live) { }; let state = tw_gateway::AppState::new(cfg).unwrap(); let live = state.live.clone(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; (addr, live) } diff --git a/crates/tw-gateway/tests/m1_acceptance.rs b/crates/tw-gateway/tests/m1_acceptance.rs index b0f9be1b..bce80530 100644 --- a/crates/tw-gateway/tests/m1_acceptance.rs +++ b/crates/tw-gateway/tests/m1_acceptance.rs @@ -47,11 +47,9 @@ fn provider(name: &str, at: SocketAddr) -> Provider { async fn serve(cfg: Config) -> SocketAddr { let state = tw_gateway::AppState::new(cfg).unwrap(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; addr } diff --git a/crates/tw-gateway/tests/m5_redact.rs b/crates/tw-gateway/tests/m5_redact.rs index 63a56f01..d6af207b 100644 --- a/crates/tw-gateway/tests/m5_redact.rs +++ b/crates/tw-gateway/tests/m5_redact.rs @@ -104,11 +104,9 @@ async fn start_with(providers: Vec, redact: RedactPolicy) -> SocketAdd ..Default::default() }; let state = tw_gateway::AppState::new(cfg).unwrap(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; addr } @@ -308,12 +306,9 @@ async fn the_ui_is_told_what_was_replaced_without_being_told_the_value() { }; let state = tw_gateway::AppState::new(cfg).unwrap(); let mut rx = state.bus.subscribe(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - let s2 = state.clone(); - tokio::spawn(async move { tw_gateway::serve(s2, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; ask(addr, &body_with_key(), false).await; diff --git a/crates/tw-gateway/tests/m5_request_guards.rs b/crates/tw-gateway/tests/m5_request_guards.rs index 4797fd32..d9b002e7 100644 --- a/crates/tw-gateway/tests/m5_request_guards.rs +++ b/crates/tw-gateway/tests/m5_request_guards.rs @@ -112,12 +112,9 @@ async fn send( ) -> (Reply, tokio::sync::broadcast::Receiver) { let state = tw_gateway::AppState::new(cfg).unwrap(); let rx = state.bus.subscribe(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - let s2 = state.clone(); - tokio::spawn(async move { tw_gateway::serve(s2, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let r = reqwest::Client::new() .post(format!("http://{addr}{path}")) diff --git a/crates/tw-gateway/tests/m5_toolwall.rs b/crates/tw-gateway/tests/m5_toolwall.rs index 371616ea..c2771cc7 100644 --- a/crates/tw-gateway/tests/m5_toolwall.rs +++ b/crates/tw-gateway/tests/m5_toolwall.rs @@ -101,12 +101,9 @@ fn with_policy(up: SocketAddr, inspect_tools: ToolPolicy) -> Config { async fn run(cfg: Config) -> (String, tokio::sync::broadcast::Receiver) { let state = tw_gateway::AppState::new(cfg).unwrap(); let rx = state.bus.subscribe(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - let s2 = state.clone(); - tokio::spawn(async move { tw_gateway::serve(s2, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let body = reqwest::Client::new() @@ -394,12 +391,9 @@ async fn start_json_upstream(body: String) -> SocketAddr { async fn run_whole(cfg: Config) -> (String, tokio::sync::broadcast::Receiver) { let state = tw_gateway::AppState::new(cfg).unwrap(); let rx = state.bus.subscribe(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - let s2 = state.clone(); - tokio::spawn(async move { tw_gateway::serve(s2, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let body = reqwest::Client::new() diff --git a/crates/tw-gateway/tests/models.rs b/crates/tw-gateway/tests/models.rs index 2c60966a..81c941e4 100644 --- a/crates/tw-gateway/tests/models.rs +++ b/crates/tw-gateway/tests/models.rs @@ -91,12 +91,9 @@ fn grouped(mut cfg: Config, order: &[&str]) -> Config { } async fn serve(state: tw_gateway::AppState) -> SocketAddr { - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - let s = state.clone(); - tokio::spawn(async move { tw_gateway::serve(s, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; addr } diff --git a/crates/tw-gateway/tests/notices.rs b/crates/tw-gateway/tests/notices.rs index ae68126d..a4ff92aa 100644 --- a/crates/tw-gateway/tests/notices.rs +++ b/crates/tw-gateway/tests/notices.rs @@ -138,11 +138,9 @@ async fn serve( let state = tw_gateway::AppState::new(cfg).unwrap(); let rx = state.bus.subscribe(); let kept = state.clone(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(60)).await; (addr, rx, kept) } diff --git a/crates/tw-gateway/tests/oauth.rs b/crates/tw-gateway/tests/oauth.rs index 14febb25..59601571 100644 --- a/crates/tw-gateway/tests/oauth.rs +++ b/crates/tw-gateway/tests/oauth.rs @@ -158,10 +158,9 @@ async fn start_gateway(providers: Vec) -> (SocketAddr, tw_gateway::App }; let state = tw_gateway::AppState::new(cfg).unwrap(); let handed = state.clone(); - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = l.local_addr().unwrap(); - drop(l); - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(60)).await; (addr, handed) } diff --git a/crates/tw-gateway/tests/passthrough.rs b/crates/tw-gateway/tests/passthrough.rs index 3bd7f578..c18b9405 100644 --- a/crates/tw-gateway/tests/passthrough.rs +++ b/crates/tw-gateway/tests/passthrough.rs @@ -16,6 +16,9 @@ use axum::http::HeaderMap; use axum::routing::post; use tw_config::{Client, Config, Listen, Provider}; +mod common; +use common::spare_port; + /// 假上游收到的东西，测试拿它来断言。 #[derive(Default, Debug)] struct Seen { @@ -83,13 +86,9 @@ async fn start_gateway(upstream: SocketAddr) -> SocketAddr { ..Default::default() }; let state = tw_gateway::AppState::new(cfg).unwrap(); - // 先 bind 拿端口，再放掉让 serve 自己 bind —— serve 需要自己建 - // 监听器才能带上 connect info（对端地址）。 - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; addr } @@ -259,13 +258,9 @@ async fn a_request_emits_the_four_lifecycle_events_in_order() { }; let state = tw_gateway::AppState::new(cfg).unwrap(); let mut rx = state.bus.subscribe(); - // 先 bind 拿端口，再放掉让 serve 自己 bind —— serve 需要自己建 - // 监听器才能带上 connect info（对端地址）。 - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let resp = reqwest::Client::new() @@ -342,13 +337,10 @@ async fn next_lifecycle(rx: &mut tokio::sync::broadcast::Receiver /// 超时，不告诉你卡在哪一行。 #[tokio::test(flavor = "multi_thread")] async fn an_unreachable_upstream_emits_a_failure_event_and_a_502() { - // 绑一个端口再立刻放掉，这样能拿到一个**确定没人在听**的端口。 + // 一个**确定没人在听**的端口（见 `spare_port`）。 // 不要写死一个「大概没人用」的端口号：低位端口在 macOS 上可能被 // 防火墙黑洞掉，表现为连接挂住十秒而不是立刻被拒。 - let dead_port = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap().port() - }; + let dead_port = spare_port(); let cfg = Config { retention: Default::default(), default_route: None, @@ -371,13 +363,9 @@ async fn an_unreachable_upstream_emits_a_failure_event_and_a_502() { }; let state = tw_gateway::AppState::new(cfg).unwrap(); let mut rx = state.bus.subscribe(); - // 先 bind 拿端口，再放掉让 serve 自己 bind —— serve 需要自己建 - // 监听器才能带上 connect info（对端地址）。 - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let resp = tokio::time::timeout( @@ -467,13 +455,9 @@ async fn a_rule_sends_opus_to_one_upstream_and_everything_else_to_another() { ])], }; let state = tw_gateway::AppState::new(cfg).unwrap(); - // 先 bind 拿端口，再放掉让 serve 自己 bind —— serve 需要自己建 - // 监听器才能带上 connect info（对端地址）。 - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let send = |model: &str| { @@ -544,13 +528,9 @@ async fn with_no_routes_at_all_requests_still_go_somewhere() { ..Default::default() }; let state = tw_gateway::AppState::new(cfg).unwrap(); - // 先 bind 拿端口，再放掉让 serve 自己 bind —— serve 需要自己建 - // 监听器才能带上 connect info（对端地址）。 - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let r = reqwest::Client::new() @@ -594,13 +574,9 @@ fn cfg_with(providers: Vec, routes: Vec) -> Config { async fn serve_cfg(cfg: Config) -> SocketAddr { let state = tw_gateway::AppState::new(cfg).unwrap(); - // 先 bind 拿端口，再放掉让 serve 自己 bind —— serve 需要自己建 - // 监听器才能带上 connect info（对端地址）。 - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; addr } @@ -1316,12 +1292,9 @@ async fn intercepting_a_probe_emits_its_own_event_not_a_request_pair() { vec![], ); let state = tw_gateway::AppState::new(cfg).unwrap(); - let addr = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - let st = state.clone(); - tokio::spawn(async move { tw_gateway::serve(st, addr).await.unwrap() }); + let addr = tw_gateway::serve(state.clone(), ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; // 监听起来时报的那一条不算：这里只看请求带出来的事件 let mut rx = state.bus.subscribe(); @@ -1708,11 +1681,9 @@ async fn a_stream_that_dies_midway_says_so_instead_of_just_stopping() { ); let state = tw_gateway::AppState::new(cfg).unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let text = reqwest::Client::new() @@ -1803,11 +1774,9 @@ async fn the_upstream_usage_reaches_the_event_stream_without_buffering_the_respo )) .unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let text = reqwest::Client::new() @@ -1857,11 +1826,9 @@ async fn an_upstream_that_gives_no_usage_reports_none_rather_than_zeroes() { )) .unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; send_to(gw).await; @@ -1892,11 +1859,9 @@ async fn a_key_pasted_into_a_prompt_is_noticed_but_the_request_goes_through_unto )) .unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let raw = r#"{"model":"m","messages":[{"role":"user","content":"我的 key 是 sk-ant-api03-abcdefghijklmnopqrstuvwxyz1234"}]}"#; @@ -1958,11 +1923,9 @@ async fn turning_the_detector_off_stops_it_looking_at_all() { cfg.security.redact.mode = tw_config::SecurityMode::Off; let state = tw_gateway::AppState::new(cfg).unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; reqwest::Client::new() .post(format!("http://{gw}/v1/messages")) @@ -2027,11 +1990,9 @@ async fn the_attempt_chain_records_every_hop_and_why_each_one_failed() { }])]; let state = tw_gateway::AppState::new(cfg).unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; assert_eq!(send_to(gw).await.status(), 200); @@ -2087,11 +2048,9 @@ async fn a_request_that_succeeds_first_try_still_has_a_chain_of_one() { )) .unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; send_to(gw).await; @@ -2123,11 +2082,9 @@ async fn a_request_that_fails_everywhere_still_reports_the_chain() { )) .unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; assert_ne!(send_to(gw).await.status(), 200); @@ -2178,11 +2135,9 @@ async fn reporting_quota_does_not_change_how_an_upstream_is_billed() { )) .unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; for n in 1..=2 { @@ -2211,11 +2166,9 @@ async fn free_billing_in_the_config_goes_out_with_the_request() { cfg.providers[0].billing = tw_config::Billing::Free; let state = tw_gateway::AppState::new(cfg).unwrap(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; send_to(gw).await; @@ -2260,11 +2213,9 @@ async fn a_quota_reset_is_kept_as_the_moment_it_happens() { .unwrap(); let kept = state.clone(); let mut rx = state.bus.subscribe(); - let gw = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; - tokio::spawn(async move { tw_gateway::serve(state, gw).await.unwrap() }); + let gw = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; send_to(gw).await; diff --git a/crates/tw-gateway/tests/ws.rs b/crates/tw-gateway/tests/ws.rs index 5c6b629f..958808ce 100644 --- a/crates/tw-gateway/tests/ws.rs +++ b/crates/tw-gateway/tests/ws.rs @@ -17,6 +17,9 @@ use tokio::sync::broadcast::Receiver; use tw_api::Event; use tw_config::{Client, Config, Listen, Provider, Security, SecurityMode}; +mod common; +use common::spare_port; + const USER_KEY: &str = "sk-ant-api03-USERSOWNKEYAAAAAAAAAAAAAA"; /// 假上游：记下收到的每一帧，然后按剧本回。 @@ -119,10 +122,9 @@ async fn start_gateway(up: SocketAddr, mode: SecurityMode, inspect: SecurityMode ..Default::default() }; let state = tw_gateway::AppState::new(cfg).unwrap(); - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = l.local_addr().unwrap(); - drop(l); - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(60)).await; addr } @@ -275,10 +277,9 @@ async fn an_upgrade_without_a_gateway_key_is_refused() { async fn serve(cfg: Config) -> (SocketAddr, Receiver) { let state = tw_gateway::AppState::new(cfg).unwrap(); let events = state.bus.subscribe(); - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = l.local_addr().unwrap(); - drop(l); - tokio::spawn(async move { tw_gateway::serve(state, addr).await.unwrap() }); + let addr = tw_gateway::serve(state, ([127, 0, 0, 1], 0).into()) + .await + .unwrap(); tokio::time::sleep(Duration::from_millis(60)).await; (addr, events) } @@ -443,11 +444,8 @@ async fn an_upstream_that_refuses_the_upgrade_is_reported_with_its_status() { /// 连不上：**失败的那一跳也要报**，而且说清为什么 —— 失败的时候恰恰最需要看它。 #[tokio::test] async fn an_unreachable_upstream_is_reported_as_a_failed_hop() { - // 绑一个端口再立刻放掉，拿到一个确定没人在听的端口 - let dead = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - l.local_addr().unwrap() - }; + // 一个确定没人在听的端口（见 `spare_port`） + let dead = SocketAddr::from(([127, 0, 0, 1], spare_port())); let (gw, mut events) = serve(routed_to_an_account(dead)).await; let _c = connect(gw).await;