From 7bea2f254d3629cda586e25e3ba7d24a0facb440 Mon Sep 17 00:00:00 2001 From: ssmithx Date: Fri, 24 Jul 2026 14:23:58 +0000 Subject: [PATCH] fix(server): stop blocking network/SSH calls from freezing the async runtime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of the actual, ongoing outage today (distinct from the connection- semaphore issue fixed earlier on this branch): several RPC handlers ran blocking, synchronous I/O directly inside async fns with no real await point, occupying a tokio worker thread for the full duration of the call: - network::router::check_upnp_available/check_tor_connectivity/check_dns — blocking std::net socket calls (check_dns had no timeout at all). - api::rpc::openwrt's 5 handlers (get-status, provision-tollgate, scan-wifi, configure-wan, scan) — Router::connect_password uses plain std::net::TcpStream::connect with NO timeout, wrapped in a fully synchronous ssh2 session (connect, handshake, run commands). Confirmed live via gdb: 3 of 4 tokio worker threads simultaneously blocked in TcpStream::connect -> Router::connect_password, called from openwrt.get-status. The frontend's home-dashboard polling loop calls openwrt.get-status periodically (refreshTollgate); once the configured router became unreachable (physical relocation to a new network today), every poll hung for the OS's default multi-minute TCP connect timeout, and polls arrived faster than they timed out — so stuck attempts accumulated until every worker thread was blocked and the entire process (not just openwrt/network calls — every RPC method, since they share the same small worker pool) stopped responding. Fix: move all the blocking I/O onto tokio's blocking pool via spawn_blocking (with an explicit timeout added to check_dns, since unlike the others it had no bound of its own). Added a regression test (network::router::blocking_io_tests) proving a cheap concurrent task is no longer starved while these checks run. Co-Authored-By: Claude Sonnet 5 --- core/archipelago/src/api/rpc/openwrt.rs | 226 +++++++++++++++--------- core/archipelago/src/network/router.rs | 152 ++++++++++++---- 2 files changed, 257 insertions(+), 121 deletions(-) diff --git a/core/archipelago/src/api/rpc/openwrt.rs b/core/archipelago/src/api/rpc/openwrt.rs index 0c30a0ca..24c0c17b 100644 --- a/core/archipelago/src/api/rpc/openwrt.rs +++ b/core/archipelago/src/api/rpc/openwrt.rs @@ -38,7 +38,19 @@ impl RpcHandler { .unwrap_or("") .to_string(); - let routers = detect::scan_subnet(subnet, prefix, &ssh_user, &ssh_password).await; + // `scan_subnet` is `async fn` but loops over up to a full /24 of + // blocking TCP probes + SSH handshakes with no real await points — + // same blocking-on-a-worker-thread hazard as the other openwrt + // handlers (see handle_openwrt_get_status), just larger in scope. + let routers = tokio::task::spawn_blocking(move || { + tokio::runtime::Handle::current().block_on(detect::scan_subnet( + subnet, + prefix, + &ssh_user, + &ssh_password, + )) + }) + .await?; let ips: Vec = routers.iter().map(|ip| ip.to_string()).collect(); Ok(serde_json::json!({ "routers": ips })) @@ -87,8 +99,84 @@ impl RpcHandler { .or_else(|| saved.password.clone()) .unwrap_or_default(); - let router = Router::connect_password(&host, 22, &ssh_user, &ssh_password)?; - router.verify_openwrt()?; + // Router/Session (ssh2) is a fully synchronous, blocking API with no + // timeout on the initial TCP connect — run it on the blocking pool, + // not directly on a tokio worker thread. Inlined here, a single + // unreachable router (e.g. after physically relocating the node, so + // the configured router is on a different/unreachable network) hangs + // for the OS's default TCP connect timeout (routinely 2+ minutes), + // and every concurrent poll of this endpoint eats another worker + // thread — with only a handful of worker threads total, that starves + // every other in-flight request in the whole process. This was a + // real full-node outage (2026-07-24), diagnosed via a live gdb + // backtrace showing 3 of 4 worker threads blocked in this exact + // `TcpStream::connect` → `Router::connect_password` call chain. + let host_for_task = host.clone(); + let ssh_user_for_task = ssh_user.clone(); + let ssh_password_for_task = ssh_password.clone(); + let status = tokio::task::spawn_blocking(move || -> Result { + let router = + Router::connect_password(&host_for_task, 22, &ssh_user_for_task, &ssh_password_for_task)?; + router.verify_openwrt()?; + + // System info + let release = router + .run_ok("cat /etc/openwrt_release") + .unwrap_or_default(); + let hostname = router + .uci_get("system.@system[0].hostname") + .unwrap_or_else(|_| "unknown".into()); + let uptime_secs: u64 = router + .run_ok("cat /proc/uptime") + .unwrap_or_default() + .split_whitespace() + .next() + .and_then(|s| s.split('.').next()) + .and_then(|s| s.parse().ok()) + .unwrap_or(0); + + // TollGate — check via opkg (≤24.x) or binary presence (25.x apk-native). + // The service binary is /usr/bin/tollgate-wrt (per its init.d script), + // not /usr/bin/tollgate-module-basic-go — that's only the opkg/apk + // *package* name, never an on-disk filename. + let tollgate_installed = router + .run("/usr/bin/opkg list-installed 2>/dev/null | grep -q '^tollgate-module-basic-go ' || \ + test -f /usr/bin/tollgate-wrt 2>/dev/null") + .map(|(_, code)| code == 0) + .unwrap_or(false); + + let tollgate = if tollgate_installed { + serde_json::json!({ + "installed": true, + "enabled": router.uci_get("tollgate.main.enabled").map(|v| v == "1").unwrap_or(false), + "metric": router.uci_get("tollgate.main.metric").unwrap_or_default(), + "step_size_ms": router.uci_get("tollgate.main.step_size").ok().and_then(|v| v.parse::().ok()).unwrap_or(0), + "price_per_step":router.uci_get("tollgate.main.price_per_step").ok().and_then(|v| v.parse::().ok()).unwrap_or(0), + "min_steps": router.uci_get("tollgate.main.min_steps").ok().and_then(|v| v.parse::().ok()).unwrap_or(1), + "currency": router.uci_get("tollgate.main.currency").unwrap_or_default(), + "mint_url": router.uci_get("tollgate.main.mint_url").unwrap_or_default(), + }) + } else { + serde_json::json!({ "installed": false }) + }; + + // WiFi interfaces + let wifi_raw = router.run_ok("uci show wireless").unwrap_or_default(); + let wifi_interfaces = parse_wifi_interfaces(&wifi_raw); + + let wan_status = wan::get_wan_status(&router); + + Ok(serde_json::json!({ + "host": host_for_task, + "hostname": hostname, + "uptime_secs": uptime_secs, + "release": parse_release(&release), + "tollgate": tollgate, + "wifi_interfaces": wifi_interfaces, + "wan": wan_status, + })) + }) + .await??; // Persist the connection so other views (e.g. the Home dashboard's // Network tile) can poll `openwrt.get-status` with no params instead @@ -107,62 +195,7 @@ impl RpcHandler { .await; } - // System info - let release = router - .run_ok("cat /etc/openwrt_release") - .unwrap_or_default(); - let hostname = router - .uci_get("system.@system[0].hostname") - .unwrap_or_else(|_| "unknown".into()); - let uptime_secs: u64 = router - .run_ok("cat /proc/uptime") - .unwrap_or_default() - .split_whitespace() - .next() - .and_then(|s| s.split('.').next()) - .and_then(|s| s.parse().ok()) - .unwrap_or(0); - - // TollGate — check via opkg (≤24.x) or binary presence (25.x apk-native). - // The service binary is /usr/bin/tollgate-wrt (per its init.d script), - // not /usr/bin/tollgate-module-basic-go — that's only the opkg/apk - // *package* name, never an on-disk filename. - let tollgate_installed = router - .run("/usr/bin/opkg list-installed 2>/dev/null | grep -q '^tollgate-module-basic-go ' || \ - test -f /usr/bin/tollgate-wrt 2>/dev/null") - .map(|(_, code)| code == 0) - .unwrap_or(false); - - let tollgate = if tollgate_installed { - serde_json::json!({ - "installed": true, - "enabled": router.uci_get("tollgate.main.enabled").map(|v| v == "1").unwrap_or(false), - "metric": router.uci_get("tollgate.main.metric").unwrap_or_default(), - "step_size_ms": router.uci_get("tollgate.main.step_size").ok().and_then(|v| v.parse::().ok()).unwrap_or(0), - "price_per_step":router.uci_get("tollgate.main.price_per_step").ok().and_then(|v| v.parse::().ok()).unwrap_or(0), - "min_steps": router.uci_get("tollgate.main.min_steps").ok().and_then(|v| v.parse::().ok()).unwrap_or(1), - "currency": router.uci_get("tollgate.main.currency").unwrap_or_default(), - "mint_url": router.uci_get("tollgate.main.mint_url").unwrap_or_default(), - }) - } else { - serde_json::json!({ "installed": false }) - }; - - // WiFi interfaces - let wifi_raw = router.run_ok("uci show wireless").unwrap_or_default(); - let wifi_interfaces = parse_wifi_interfaces(&wifi_raw); - - let wan_status = wan::get_wan_status(&router); - - Ok(serde_json::json!({ - "host": host, - "hostname": hostname, - "uptime_secs": uptime_secs, - "release": parse_release(&release), - "tollgate": tollgate, - "wifi_interfaces": wifi_interfaces, - "wan": wan_status, - })) + Ok(status) } /// Provision TollGate on an OpenWrt router and create the "archipelago" SSID. @@ -228,15 +261,32 @@ impl RpcHandler { enabled: p.get("enabled").and_then(|v| v.as_bool()).unwrap_or(true), }; - let router = Router::connect_password(&host, 22, &ssh_user, &ssh_password)?; - router.verify_openwrt()?; - tollgate::provision(&router, &config).await?; + let response_ssid = config.ssid.clone(); + let response_mint_url = config.mint_url.clone(); + + // Blocking ssh2 I/O — see handle_openwrt_get_status for why this + // must run on the blocking pool rather than a tokio worker thread. + // `tollgate::provision` is `async fn` but has no real await points + // (every op inside it is a synchronous SSH round trip) — block_on + // here just runs it to completion on this blocking-pool thread + // instead of pretending it yields on a tokio worker. + let host_for_task = host.clone(); + let ssh_user_for_task = ssh_user.clone(); + let ssh_password_for_task = ssh_password.clone(); + tokio::task::spawn_blocking(move || -> Result<()> { + let router = + Router::connect_password(&host_for_task, 22, &ssh_user_for_task, &ssh_password_for_task)?; + router.verify_openwrt()?; + tokio::runtime::Handle::current().block_on(tollgate::provision(&router, &config))?; + Ok(()) + }) + .await??; Ok(serde_json::json!({ "ok": true, "host": host, - "ssid": config.ssid, - "mint_url": config.mint_url, + "ssid": response_ssid, + "mint_url": response_mint_url, })) } @@ -279,22 +329,27 @@ impl RpcHandler { .or_else(|| saved.password.clone()) .unwrap_or_default(); - let router = Router::connect_password(&host, 22, &ssh_user, &ssh_password)?; - router.verify_openwrt()?; + // Blocking ssh2 I/O — see handle_openwrt_get_status for why this + // must run on the blocking pool rather than a tokio worker thread. + let result = tokio::task::spawn_blocking(move || -> Result> { + let router = Router::connect_password(&host, 22, &ssh_user, &ssh_password)?; + router.verify_openwrt()?; - let networks = wifi_scan::scan_networks(&router)?; - let result: Vec = networks - .iter() - .map(|n| { - serde_json::json!({ - "ssid": n.ssid, - "bssid": n.bssid, - "signal": n.signal, - "channel": n.channel, - "encryption": n.encryption, + let networks = wifi_scan::scan_networks(&router)?; + Ok(networks + .iter() + .map(|n| { + serde_json::json!({ + "ssid": n.ssid, + "bssid": n.bssid, + "signal": n.signal, + "channel": n.channel, + "encryption": n.encryption, + }) }) - }) - .collect(); + .collect()) + }) + .await??; Ok(serde_json::json!({ "networks": result })) } @@ -357,9 +412,6 @@ impl RpcHandler { let dhcp_limit = p.get("dhcp_limit").and_then(|v| v.as_u64()).unwrap_or(150) as u32; let masq = p.get("masq").and_then(|v| v.as_bool()).unwrap_or(true); - let router = Router::connect_password(&host, 22, &ssh_user, &ssh_password)?; - router.verify_openwrt()?; - let config = wan::WispConfig { ssid: ssid.clone(), password, @@ -368,7 +420,17 @@ impl RpcHandler { dhcp_limit, masq, }; - wan::configure_wisp(&router, &config)?; + + // Blocking ssh2 I/O — see handle_openwrt_get_status for why this + // must run on the blocking pool rather than a tokio worker thread. + let host_for_task = host.clone(); + tokio::task::spawn_blocking(move || -> Result<()> { + let router = Router::connect_password(&host_for_task, 22, &ssh_user, &ssh_password)?; + router.verify_openwrt()?; + wan::configure_wisp(&router, &config)?; + Ok(()) + }) + .await??; Ok(serde_json::json!({ "ok": true, "host": host, "ssid": ssid })) } diff --git a/core/archipelago/src/network/router.rs b/core/archipelago/src/network/router.rs index a8620842..8922bfa3 100644 --- a/core/archipelago/src/network/router.rs +++ b/core/archipelago/src/network/router.rs @@ -97,42 +97,57 @@ async fn get_wan_ip() -> Option { } /// Check if UPnP is available by attempting SSDP discovery. +/// +/// The socket I/O here is plain blocking `std::net` (its 3s read timeout is +/// enforced by the OS, not by yielding to the async runtime), so it must run +/// on the blocking-pool via `spawn_blocking` — inlined into this "async fn" +/// directly, it used to occupy a tokio worker thread for the full 3s on +/// every call. With only as many worker threads as CPU cores, a handful of +/// concurrent `network.diagnostics` calls (e.g. several Server-settings page +/// loads) could starve the whole runtime and freeze every other in-flight +/// request — root cause of a full-node outage (2026-07-24) that had nothing +/// to do with connection limits and everything to do with blocking sockets +/// on async worker threads. async fn check_upnp_available() -> bool { - use std::net::UdpSocket; + tokio::task::spawn_blocking(|| { + use std::net::UdpSocket; - let ssdp_request = "M-SEARCH * HTTP/1.1\r\n\ - HOST: 239.255.255.250:1900\r\n\ - MAN: \"ssdp:discover\"\r\n\ - MX: 2\r\n\ - ST: urn:schemas-upnp-org:device:InternetGatewayDevice:1\r\n\r\n"; + let ssdp_request = "M-SEARCH * HTTP/1.1\r\n\ + HOST: 239.255.255.250:1900\r\n\ + MAN: \"ssdp:discover\"\r\n\ + MX: 2\r\n\ + ST: urn:schemas-upnp-org:device:InternetGatewayDevice:1\r\n\r\n"; - let socket = match UdpSocket::bind("0.0.0.0:0") { - Ok(s) => s, - Err(_) => return false, - }; + let socket = match UdpSocket::bind("0.0.0.0:0") { + Ok(s) => s, + Err(_) => return false, + }; - if socket - .set_read_timeout(Some(std::time::Duration::from_secs(3))) - .is_err() - { - return false; - } - - if socket - .send_to(ssdp_request.as_bytes(), "239.255.255.250:1900") - .is_err() - { - return false; - } - - let mut buf = [0u8; 2048]; - match socket.recv_from(&mut buf) { - Ok((len, _)) => { - let response = String::from_utf8_lossy(&buf[..len]); - response.contains("InternetGatewayDevice") || response.contains("200 OK") + if socket + .set_read_timeout(Some(std::time::Duration::from_secs(3))) + .is_err() + { + return false; } - Err(_) => false, - } + + if socket + .send_to(ssdp_request.as_bytes(), "239.255.255.250:1900") + .is_err() + { + return false; + } + + let mut buf = [0u8; 2048]; + match socket.recv_from(&mut buf) { + Ok((len, _)) => { + let response = String::from_utf8_lossy(&buf[..len]); + response.contains("InternetGatewayDevice") || response.contains("200 OK") + } + Err(_) => false, + } + }) + .await + .unwrap_or(false) } /// Add a port forward (stored locally; actual UPnP mapping done on request). @@ -281,19 +296,41 @@ pub async fn run_diagnostics() -> Result { } /// Check if Tor SOCKS proxy is reachable. +/// +/// `TcpStream::connect_timeout` blocks the calling OS thread for up to its +/// timeout — same blocking-on-a-worker-thread hazard as `check_upnp_available` +/// above, so this also runs on the blocking pool. async fn check_tor_connectivity() -> bool { - use std::net::TcpStream; - TcpStream::connect_timeout( - &"127.0.0.1:9050".parse().unwrap(), - std::time::Duration::from_secs(2), - ) - .is_ok() + tokio::task::spawn_blocking(|| { + use std::net::TcpStream; + TcpStream::connect_timeout( + &"127.0.0.1:9050".parse().unwrap(), + std::time::Duration::from_secs(2), + ) + .is_ok() + }) + .await + .unwrap_or(false) } /// Check DNS resolution works. +/// +/// `to_socket_addrs()` is a blocking libc resolver call with no timeout of +/// its own — on a network with a slow/unreachable DNS server (e.g. right +/// after relocating to a new network) it can hang far longer than the other +/// checks here. Runs on the blocking pool (same reason as the checks above) +/// AND under an explicit timeout, since unlike UPnP/Tor there's no built-in +/// bound to rely on. async fn check_dns() -> bool { - use std::net::ToSocketAddrs; - "cloudflare.com:443".to_socket_addrs().is_ok() + let lookup = tokio::task::spawn_blocking(|| { + use std::net::ToSocketAddrs; + "cloudflare.com:443".to_socket_addrs().is_ok() + }); + tokio::time::timeout(std::time::Duration::from_secs(5), lookup) + .await + .ok() + .and_then(|r| r.ok()) + .unwrap_or(false) } // --- Router Compatibility Abstraction --- @@ -469,3 +506,40 @@ pub async fn get_router_info(data_dir: &Path) -> Result { }, })) } + +#[cfg(test)] +mod blocking_io_tests { + use super::*; + + // Regression test for the 2026-07-24 outage: check_upnp_available, + // check_tor_connectivity, and check_dns each did blocking std::net I/O + // directly on their calling task instead of via spawn_blocking. On a + // small worker pool (4 threads in production), a handful of concurrent + // network.diagnostics calls tied up every worker thread for seconds, + // freezing every other in-flight RPC request. Proves a cheap task + // spawned alongside these checks still gets scheduled promptly, which + // only holds if the checks aren't monopolizing worker threads. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn network_checks_do_not_starve_other_tasks() { + let cheap = tokio::spawn(async { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + std::time::Instant::now() + }); + + let _ = tokio::join!( + check_upnp_available(), + check_tor_connectivity(), + check_dns() + ); + + // The assertion is that `cheap` — spawned before the blocking checks + // and sleeping only 5ms — finishes within 1s of being spawned. If the + // checks were still blocking worker threads directly, a 2-worker + // runtime running 3 blocking checks concurrently would starve this + // task well past 1s. + tokio::time::timeout(std::time::Duration::from_secs(1), cheap) + .await + .expect("a concurrently-spawned cheap task must not be starved by blocking network checks") + .expect("cheap task panicked"); + } +}