diff --git a/bin/twcore/src/main.rs b/bin/twcore/src/main.rs index 303d71d5..d092a2c1 100644 --- a/bin/twcore/src/main.rs +++ b/bin/twcore/src/main.rs @@ -739,14 +739,10 @@ fn cmd_serve(path: &Path, port: Option, safe: bool, parent: Option) -> if let Some(p) = port { listen.port = p; } - // **`bind` 写的是网卡名时，这一步要问系统。**问不出来就在这里停， - // 而不是带着一个猜出来的地址起监听 —— 错误里说清是哪张网卡，以及 - // 这台机器上真有哪些 - let addrs = listen - .addrs() - .with_context(|| format!("resolving bind: {}", listen.bind))?; - // 配置里写的那个排在最后；日志和报错里说它，不说顺带开着的回环 - let addr = *addrs.last().expect("addrs() never returns an empty list"); + // **`bind` 写的是网卡名时，这里不去问系统。**那张网卡此刻不在（开机自启时 + // WSL 还没起来，它的虚拟网卡也就还没有）不该让网关起不来：监听那一边先只 + // 听回环，网卡出现了再补上，换了地址也跟着换，见 `tw_gateway::serve_at` + let listen_at = format!("{}:{}", listen.bind, listen.port); let rt = tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -907,10 +903,10 @@ fn cmd_serve(path: &Path, port: Option, safe: bool, parent: Option) -> // 探测要打网络，而网关不该因为一次探测慢而起不来。 tw_gateway::models::spawn(state.clone()); - tracing::info!(%addr, "starting"); + tracing::info!(listen = %listen_at, "starting"); tokio::select! { - r = tw_gateway::serve_at(state, addrs, !overridden) => { - r.with_context(|| format!("{addr} could not be listened on. If the port is taken, check whether an earlier instance has fully exited")) + r = tw_gateway::serve_at(state, listen, !overridden) => { + r.with_context(|| format!("{listen_at} could not be listened on. If the port is taken, check whether an earlier instance has fully exited")) } msg = control_dead => { anyhow::bail!("{}", msg.unwrap_or_else(|_| "the control plane stopped".into())) diff --git a/crates/tw-config/src/lib.rs b/crates/tw-config/src/lib.rs index 94bd88bf..40dcb756 100644 --- a/crates/tw-config/src/lib.rs +++ b/crates/tw-config/src/lib.rs @@ -414,8 +414,8 @@ impl Default for GatewayListen { /// /// 写名字而不是写地址，是因为地址会变：DHCP 续租、换个 Wi-Fi， /// `192.168.1.5` 就不在了，网关起不来，而系统给的错误只有一句 -/// 「Can't assign requested address」。名字不会变，每次启动现问系统它 -/// 当下是哪个地址。**写死的地址仍然收** —— 有人就是要钉住那一个。 +/// 「Can't assign requested address」。名字不会变，网关每隔几秒现问系统它 +/// 当下是哪个地址，变了就换过去；网卡暂时不在时先只听回环，等它出现。**写死的地址仍然收** —— 有人就是要钉住那一个。 #[derive(Debug, Clone, PartialEq, Eq, Default)] pub enum Bind { #[default] @@ -450,7 +450,8 @@ pub enum BindError { impl Bind { /// 要监听的那个地址。**网卡名要问系统**，所以这一步可能失败。 /// - /// 每次都现问，不缓存：换网络之后重启网关，拿到的就该是新地址。 + /// 每次都现问，不缓存：换了网络，下一次问拿到的就该是新地址（网关的 + /// 监听那一边每隔几秒问一次，见 `tw_gateway::serve_at`）。 pub fn resolve(&self) -> Result { match self { Bind::Loopback => Ok(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)), diff --git a/crates/tw-config/tests/manual/schema.rs b/crates/tw-config/tests/manual/schema.rs index 974272c7..04c34683 100644 --- a/crates/tw-config/tests/manual/schema.rs +++ b/crates/tw-config/tests/manual/schema.rs @@ -243,8 +243,8 @@ pub fn sections() -> Vec
{ Kind::Bind, Def::Is("loopback"), t( - "`loopback` is this machine only; `all` is every interface; an interface name (`en0`, `eth0`) is looked up at start and follows address changes; a fixed IP address stops working when the address changes. Binding one interface also listens on 127.0.0.1.", - "`loopback` 只有本机；`all` 所有网卡；网卡名（`en0`、`eth0`）在启动时解析，地址变了也能跟上；写死的 IP 地址在地址变化后失效。绑定单张网卡时同时监听 127.0.0.1。", + "`loopback` is this machine only; `all` is every interface; an interface name (`en0`, `eth0`) is looked up again every few seconds and follows address changes, and while the interface is not there the gateway listens on 127.0.0.1 only and adds it once it appears; a fixed IP address stops working when the address changes. Binding one interface also listens on 127.0.0.1.", + "`loopback` 只有本机；`all` 所有网卡；网卡名（`en0`、`eth0`）每隔几秒重新解析，地址变了也能跟上，网卡暂时不在时先只监听 127.0.0.1，出现后再补上；写死的 IP 地址在地址变化后失效。绑定单张网卡时同时监听 127.0.0.1。", ), ), row( diff --git a/crates/tw-control/tests/listen.rs b/crates/tw-control/tests/listen.rs index 78c83f4f..68524f97 100644 --- a/crates/tw-control/tests/listen.rs +++ b/crates/tw-control/tests/listen.rs @@ -219,12 +219,7 @@ async fn the_status_follows_the_listener_after_a_save() { let p1 = spare_port(); let b = bed(&yaml(p1)); let gw = b.gw.clone(); - let want = tw_config::try_parse(&b.file()) - .unwrap() - .listen - .gateway - .addrs() - .unwrap(); + let want = tw_config::try_parse(&b.file()).unwrap().listen.gateway; tokio::spawn(async move { tw_gateway::serve_at(gw, want, true).await.unwrap() }); tokio::time::sleep(Duration::from_millis(80)).await; diff --git a/crates/tw-gateway/src/listen.rs b/crates/tw-gateway/src/listen.rs index 54b4d219..f5c1c923 100644 --- a/crates/tw-gateway/src/listen.rs +++ b/crates/tw-gateway/src/listen.rs @@ -239,35 +239,116 @@ pub(crate) async fn serve_detached( Ok(actual) } +/// 多久再问一次系统：要听的网卡出现了没有、地址变了没有。 +/// +/// **网卡的出现和换地址没有通知可等。**WSL 的虚拟网卡要等 WSL 起来才有，WSL +/// 每重启一次它换一个地址；DHCP 续租、换 Wi-Fi 也一样。问一次是一次 +/// `getifaddrs`（Windows 上是 `GetAdaptersAddresses`），几秒一次不值一提； +/// 而等得太久，刚起来的 WSL 里的客户端就要多连不上几秒。 +const RECHECK: std::time::Duration = std::time::Duration::from_secs(3); + +/// 这一轮要听哪些地址，以及听不全的原因。 +/// +/// **网卡此刻不在，就先只听回环。**绑的是一张网卡时回环本来就要听（见 +/// `GatewayListen::addrs`），网卡那一个等它出现再补上 —— 不能因为一张迟到的 +/// 网卡让整个网关起不来：开机自启时 WSL 还没起，这台电脑上的客户端也跟着 +/// 全部断线。 +fn plan(listen: &tw_config::GatewayListen) -> (Vec, Option) { + match listen.addrs() { + Ok(a) => (a, None), + Err(e) => (loopback_only(listen.port), Some(unresolved(&e))), + } +} + +fn loopback_only(port: u16) -> Vec { + vec![SocketAddr::new(std::net::Ipv4Addr::LOCALHOST.into(), port)] +} + +/// 这个地址此刻不是这台机器的 —— 网卡的地址刚被收走，或者写死的那个地址还没 +/// 出现。**回环和通配地址不算**：它们绑不上从来不是「还没出现」。 +fn absent(a: SocketAddr, e: &std::io::Error) -> bool { + e.kind() == std::io::ErrorKind::AddrNotAvailable + && !a.ip().is_loopback() + && !a.ip().is_unspecified() +} + /// 起服务。`follow` 为真时**跟着配置里的监听地址走**（「温」那一级热重载）。 /// /// 命令行给了 `--port` 时 `follow` 为假：那是一个显式的覆盖，不该被配置 -/// 文件推翻。 +/// 文件推翻 —— 但起始那份设置里的网卡照样跟着：它出现、换地址时补上或换过去。 /// -/// 起始那一次绑不上是致命的，错误带着地址交给调用方 —— 那时还没有旧的 -/// 可以守，守护进程据此重启或进安全模式。之后的每一次换监听都不会让它返回。 -pub async fn serve_at(state: AppState, want: Vec, follow: bool) -> std::io::Result<()> { +/// 起始那一次**回环绑不上是致命的**（多半是端口被占），错误带着地址交给调用方 +/// —— 那时还没有旧的可以守，守护进程据此重启或进安全模式。网卡此刻不在、写死 +/// 的地址此刻不是这台机器的，都不致命：先只听回环，[`RECHECK`] 一轮轮再问， +/// 它出现了就补上。之后的每一次换监听都不会让它返回。 +pub async fn serve_at( + state: AppState, + listen: tw_config::GatewayListen, + follow: bool, +) -> std::io::Result<()> { + let (want, mut note) = plan(&listen); let mut bound = Vec::with_capacity(want.len()); - for (a, l) in bind_all(&want) - .await - .map_err(|(a, e)| std::io::Error::new(e.kind(), format!("{}", bind_failure(a, &e))))? - { - bound.push(start(&state, a, l)?); + for a in want { + match TcpListener::bind(a).await { + Ok(l) => bound.push(start(&state, a, l)?), + Err(e) if absent(a, &e) => { + tracing::warn!(%a, %e, "not an address of this machine yet; listening without it until it is"); + note = Some(bind_failure(a, &e)); + } + Err(e) => { + return Err(std::io::Error::new( + e.kind(), + format!("{}", bind_failure(a, &e)), + )); + } + } } - state.set_listening(snapshot(&bound, None)); - if !follow { - // 监听器各自在自己的任务里跑。这个 future 活着，它们就不收到停止信号 - std::future::pending::<()>().await; + if let Some(e) = ¬e { + tracing::warn!(%e, "listening on loopback only for now; the rest is added once it appears"); } + state.set_listening(snapshot(&bound, note)); + // 眼下这些监听器是照着哪份设置绑的。**设置没变、网卡却解析不出来了**，是 + // 我们跟着的那张网卡没了（WSL 关了）：退回只听回环，等它回来。设置变了 + // 而新的解析不出来，那是另一回事，见下面 + let mut following = (listen.bind.clone(), listen.port); + let mut recheck = tokio::time::interval(RECHECK); + recheck.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + recheck.tick().await; loop { - state.relisten_signal().notified().await; - let want = match state.runtime().config.listen.gateway.addrs() { - Ok(a) => a, - // **新配置的地址算不出来就守住旧的。**`bind` 指着一张刚被拔掉 + // 被叫到的（配置换了、界面上存了一次）和到点再问的，**只有一处不同**： + // 到点再问不去请旧的让出端口 —— 那一下有几毫秒不接新连接，端口真被 + // 别的程序占着时，每几秒来一次就成了每几秒断一次 + let asked = tokio::select! { + _ = state.relisten_signal().notified() => true, + _ = recheck.tick() => false, + }; + let listen = if follow { + state.runtime().config.listen.gateway.clone() + } else { + listen.clone() + }; + let spec = (listen.bind.clone(), listen.port); + let (want, note) = match listen.addrs() { + Ok(a) => (a, None), + // 跟着的那张网卡没了：它那一个退场，回环照听，等它回来再补上。 + // **留着它没有用** —— 那个地址上已经没人连得过来，而它回来时地址 + // 多半换了（WSL 每重启一次换一个）。 + // + // 此刻本来就只在听回环时也一样：没有什么正在工作的要守，改了端口 + // 就该在新端口的回环上听，不必等网卡出现 + Err(e) if spec == following || bound.iter().all(|b| b.want.ip().is_loopback()) => { + if asked || bound.iter().any(|b| !b.want.ip().is_loopback()) { + tracing::warn!(%e, "the interface is gone; listening on loopback only until it is back"); + } + (loopback_only(listen.port), Some(unresolved(&e))) + } + // **新配置的地址算不出来就守住旧的。**`bind` 刚改成一张此刻不在 // 的网卡时，正确的动作不是把一个正在工作的监听器拆掉 —— 那会 - // 让所有客户端立刻断线，而它们本来好好的。 + // 让连着的客户端立刻断线，而它们本来好好的。它出现了再换过去 Err(e) => { - tracing::error!(%e, "the new listen address cannot be resolved; keeping the current one"); + if asked { + tracing::error!(%e, "the new listen address cannot be resolved; keeping the current one"); + } state.set_listening(snapshot(&bound, Some(unresolved(&e)))); continue; } @@ -275,7 +356,8 @@ pub async fn serve_at(state: AppState, want: Vec, follow: bool) -> s let have: Vec = bound.iter().map(|b| b.want).collect(); if want == have { // 通知来了但地址没变（比如又改回去了）。上一次没换成的那句话到此作废 - state.set_listening(snapshot(&bound, None)); + following = spec; + state.set_listening(snapshot(&bound, note)); continue; } // **还要的地址原样留着，只绑新增的** —— 从「仅本机」换到「局域网」时， @@ -295,7 +377,8 @@ pub async fn serve_at(state: AppState, want: Vec, follow: bool) -> s // 挡路的可能是我们自己要退场的那几个（见模块文档）：先让它们放掉 // 端口再试。**只在同端口时这么做** —— 别的端口被占，让出来也没用 Err((a, e)) - if e.kind() == std::io::ErrorKind::AddrInUse + if asked + && e.kind() == std::io::ErrorKind::AddrInUse && gone.iter().any(|b| b.want.port() == a.port()) => { let back: Vec = gone.iter().map(|b| b.want).collect(); @@ -325,7 +408,11 @@ pub async fn serve_at(state: AppState, want: Vec, follow: bool) -> s } } Err((a, e)) => { - tracing::error!(%a, %e, "the new listen address cannot be bound; keeping the current one"); + // 到点再问时每一轮都会走到这里（端口还被占着、网卡的地址还没 + // 真正可用），只在被叫到的那一次记错误，免得日志每几秒一行 + if asked { + tracing::error!(%a, %e, "the new listen address cannot be bound; keeping the current one"); + } bound = keep.into_iter().chain(gone).collect(); bound.sort_by_key(|b| have.iter().position(|w| *w == b.want)); state.set_listening(snapshot(&bound, Some(bind_failure(a, &e)))); @@ -336,9 +423,10 @@ pub async fn serve_at(state: AppState, want: Vec, follow: bool) -> s for (a, l) in listeners { bound.push(start(&state, a, l)?); } + following = spec; // 配置里写的那个排在最后，和 `addrs()` 同一个顺序 bound.sort_by_key(|b| want.iter().position(|w| *w == b.want)); - state.set_listening(snapshot(&bound, None)); + state.set_listening(snapshot(&bound, note)); } } diff --git a/crates/tw-gateway/tests/hotreload.rs b/crates/tw-gateway/tests/hotreload.rs index d8a0c0e5..346f7f2f 100644 --- a/crates/tw-gateway/tests/hotreload.rs +++ b/crates/tw-gateway/tests/hotreload.rs @@ -532,7 +532,7 @@ async fn changing_the_port_actually_moves_the_listener() { /// 起一个跟着配置走的网关，等它绑上。 async fn following(state: &tw_gateway::AppState, c: &Config) { let s = state.clone(); - let want = c.listen.gateway.addrs().unwrap(); + let want = c.listen.gateway.clone(); tokio::spawn(async move { tw_gateway::serve_at(s, want, true).await.unwrap() }); tokio::time::sleep(Duration::from_millis(80)).await; } @@ -810,3 +810,141 @@ async fn when_the_new_address_is_really_taken_the_old_one_is_taken_back() { assert!(ask(gw).await.contains("\"a\""), "旧的地址不该停"); drop(squatter); } + +/// 起一个网关，**起始那一次绑定不许失败**：等它交回监听状态。 +async fn started(state: &tw_gateway::AppState, c: &Config) -> tw_gateway::Listening { + let s = state.clone(); + let listen = c.listen.gateway.clone(); + let serving = tokio::spawn(async move { tw_gateway::serve_at(s, listen, true).await }); + for _ in 0..100 { + assert!( + !serving.is_finished(), + "the gateway gave up: {:?}", + serving.await + ); + let now = state.listening(); + if !now.addrs.is_empty() { + return now; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("the gateway never started listening"); +} + +#[tokio::test] +async fn an_interface_that_is_not_there_yet_does_not_stop_the_gateway_from_starting() { + // **开机自启时 WSL 还没起来**，它的虚拟网卡也就还不在。以前这一步让 twcore + // 直接退出，这台电脑上的客户端跟着全部连不上；现在先只听回环，网卡出现了 + // 再补上 + let (up, _) = counting_upstream("a").await; + let mut c = cfg(vec![provider("a", up)], vec![]); + let port = spare_port(); + c.listen.gateway.port = port; + c.listen.gateway.bind = bind("tw-not-here0"); + let state = tw_gateway::AppState::new(c.clone()).unwrap(); + let now = started(&state, &c).await; + + let lo: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + assert_eq!(now.addrs, [lo]); + assert_eq!( + now.error.as_ref().map(|m| m.code.as_str()), + Some("gw.listen.no_such_nic"), + "界面上要说清为什么只听了本机: {now:?}" + ); + assert!(ask(lo).await.contains("\"a\""), "回环该照常服务"); + + // 换成一个解析得出来的设置：照常换过去，那句话也跟着消失 + let mut next = c.clone(); + next.listen.gateway.bind = bind("all"); + let want = next.listen.gateway.addrs().unwrap(); + state.reload(next).unwrap(); + // `switch` 看见错误就回来，而这时错误本来就在：等地址换过去 + for _ in 0..100 { + if state.listening().addrs == want { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + let now = state.listening(); + assert_eq!(now.addrs, want, "{now:?}"); + assert_eq!(now.error, None, "{now:?}"); + assert!(ask(lo).await.contains("\"a\"")); +} + +#[tokio::test] +async fn an_address_that_is_not_on_this_machine_yet_does_not_stop_the_gateway_from_starting() { + // 写死的地址此刻不是这台机器的（网卡还没拿到它）：和网卡不在一样，先只听 + // 回环。198.51.100.1 是文档专用的地址段，哪台机器上都不会有 + let (up, _) = counting_upstream("a").await; + let mut c = cfg(vec![provider("a", up)], vec![]); + let port = spare_port(); + c.listen.gateway.port = port; + c.listen.gateway.bind = bind("198.51.100.1"); + let state = tw_gateway::AppState::new(c.clone()).unwrap(); + let now = started(&state, &c).await; + + let lo: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + assert_eq!(now.addrs, [lo]); + assert_eq!( + now.error.as_ref().map(|m| m.code.as_str()), + Some("gw.listen.addr_unavailable"), + "{now:?}" + ); + assert!(ask(lo).await.contains("\"a\"")); +} + +#[tokio::test] +async fn a_taken_loopback_port_still_stops_the_gateway_from_starting() { + // 回环那一个绑不上不是「还没出现」：端口被占，照旧交给调用方，守护进程 + // 据此重启或进安全模式 + let squatter = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = squatter.local_addr().unwrap().port(); + let (up, _) = counting_upstream("a").await; + let mut c = cfg(vec![provider("a", up)], vec![]); + c.listen.gateway.port = port; + c.listen.gateway.bind = bind("tw-not-here0"); + let state = tw_gateway::AppState::new(c.clone()).unwrap(); + let r = tokio::time::timeout( + Duration::from_secs(2), + tw_gateway::serve_at(state, c.listen.gateway.clone(), true), + ) + .await + .expect("serve_at should give up rather than wait"); + assert_eq!( + r.unwrap_err().kind(), + std::io::ErrorKind::AddrInUse, + "端口被占要照旧报出来" + ); +} + +#[tokio::test] +async fn a_new_port_takes_effect_on_loopback_while_the_interface_is_still_missing() { + // 网卡还没出现、只在听回环时改了端口：没有什么正在工作的要守，回环换到 + // 新端口上，网卡那一个照旧等它出现 + let (up, _) = counting_upstream("a").await; + let mut c = cfg(vec![provider("a", up)], vec![]); + c.listen.gateway.port = spare_port(); + c.listen.gateway.bind = bind("tw-not-here0"); + let state = tw_gateway::AppState::new(c.clone()).unwrap(); + started(&state, &c).await; + + let p2 = spare_port(); + let lo2: SocketAddr = format!("127.0.0.1:{p2}").parse().unwrap(); + let mut next = c.clone(); + next.listen.gateway.port = p2; + state.reload(next).unwrap(); + for _ in 0..100 { + if state.listening().addrs == [lo2] { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + let now = state.listening(); + assert_eq!(now.addrs, [lo2], "{now:?}"); + assert_eq!( + now.error.as_ref().map(|m| m.code.as_str()), + Some("gw.listen.no_such_nic"), + "{now:?}" + ); + assert!(ask(lo2).await.contains("\"a\"")); +} diff --git a/docs/config.md b/docs/config.md index 5dc20746..aa8cf718 100644 --- a/docs/config.md +++ b/docs/config.md @@ -180,7 +180,7 @@ clients send requests to, and the control channel the desktop app and | Field | Type | Default | Description | |---|---|---|---| -| `bind` | `loopback` \| `all` \| interface name \| IP address | `loopback` | `loopback` is this machine only; `all` is every interface; an interface name (`en0`, `eth0`) is looked up at start and follows address changes; a fixed IP address stops working when the address changes. Binding one interface also listens on 127.0.0.1. | +| `bind` | `loopback` \| `all` \| interface name \| IP address | `loopback` | `loopback` is this machine only; `all` is every interface; an interface name (`en0`, `eth0`) is looked up again every few seconds and follows address changes, and while the interface is not there the gateway listens on 127.0.0.1 only and adds it once it appears; a fixed IP address stops working when the address changes. Binding one interface also listens on 127.0.0.1. | | `port` | integer | `8788` | TCP port of the gateway. | | `allow_from` | list of strings | `[10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, fc00::/7]` | Sources other than this machine that may connect, as CIDR ranges or single addresses. This machine is always allowed. `[]` means this machine only; `0.0.0.0/0` allows everyone and has to be written out. | diff --git a/docs/config.zh-CN.md b/docs/config.zh-CN.md index 3fd04b11..e2a37f93 100644 --- a/docs/config.zh-CN.md +++ b/docs/config.zh-CN.md @@ -127,7 +127,7 @@ core 在哪里接受连接。连接分两种：客户端发送请求的 AI 网 | 字段 | 类型 | 默认值 | 说明 | |---|---|---|---| -| `bind` | `loopback` \| `all` \| 网卡名 \| IP 地址 | `loopback` | `loopback` 只有本机；`all` 所有网卡；网卡名（`en0`、`eth0`）在启动时解析，地址变了也能跟上；写死的 IP 地址在地址变化后失效。绑定单张网卡时同时监听 127.0.0.1。 | +| `bind` | `loopback` \| `all` \| 网卡名 \| IP 地址 | `loopback` | `loopback` 只有本机；`all` 所有网卡；网卡名（`en0`、`eth0`）每隔几秒重新解析，地址变了也能跟上，网卡暂时不在时先只监听 127.0.0.1，出现后再补上；写死的 IP 地址在地址变化后失效。绑定单张网卡时同时监听 127.0.0.1。 | | `port` | 整数 | `8788` | 网关的 TCP 端口。 | | `allow_from` | 字符串列表 | `[10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, fc00::/7]` | 本机以外允许连接的来源，写 CIDR 网段或单个地址。本机始终放行。`[]` 表示只有本机；放行所有来源要明确写 `0.0.0.0/0`。 |