diff --git a/Cargo.lock b/Cargo.lock index 076b61d4..318f7e27 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -859,6 +859,12 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + [[package]] name = "litemap" version = "0.8.3" @@ -952,6 +958,7 @@ dependencies = [ "reqwest", "rusqlite", "rust-embed", + "rustix", "serde", "serde_json", "serde_urlencoded", @@ -1310,6 +1317,19 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" +[[package]] +name = "rustix" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "891efababe418670775f199f0d233d84843c227a0949a883ce15b37c78d6629d" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.61.2", +] + [[package]] name = "rustls" version = "0.23.45" diff --git a/Cargo.toml b/Cargo.toml index 26ba71f2..2943298b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,6 +32,9 @@ tower-http = { version = "0.6", features = ["limit", "compression-gzip"] } # tar. tar = { version = "0.4", default-features = false } flate2 = { version = "1", default-features = false, features = ["rust_backend"] } +# The open-file limit, raised at startup. Pure Rust on Linux, so the cross-built +# musl binary gains no C dependency. +rustix = { version = "1", features = ["process"] } [dev-dependencies] serde_urlencoded = "0.7" diff --git a/src/api.rs b/src/api.rs index 0988a4bf..c22b67be 100644 --- a/src/api.rs +++ b/src/api.rs @@ -635,6 +635,31 @@ pub struct Stream { gzip: Option, } +/// How many anonymous streams may be open at once. The fourth bound of the +/// anonymous paths, beside `HISTORY_GATE`, `main::RELAY_GATE` and +/// `auth::PASSWORD_GATE`: a stream holds a descriptor and a task for as long as +/// its page stays open, and at a hundred nodes sends about 7 KB/s of compressed +/// frames, 55 KB/s to a page that does not ask for gzip. A thousand pages are +/// 7 to 55 MB/s outbound. Beyond it a page falls back to polling `/api/nodes`, +/// whose rendered frame every viewer shares. +const PUBLIC_STREAMS: usize = 1000; +static PUBLIC_STREAM_GATE: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(PUBLIC_STREAMS); + +/// A send that blocks this long has no reader left: the pages treat 10 s +/// without a frame as a lost stream and open another. A peer that vanished +/// without a FIN would otherwise keep its slot until TCP gives up, about +/// 15 minutes. +const SEND_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +/// The slot a new stream holds until it ends. A session is the operator's own +/// and is not counted, so whoever holds one can always open the panel. +fn stream_slot(session: Option<&str>) -> Result>, ()> { + if session.is_some() { + return Ok(None); + } + PUBLIC_STREAM_GATE.try_acquire().map(Some).map_err(|_| ()) +} + /// Live stream for the browser. Each connection runs its own timer -- simpler to /// reason about than a fan-out channel -- over a shared snapshot, so a timer /// costs no more than a send. @@ -650,10 +675,14 @@ pub async fn live_ws( if session.is_none() && !app.public_page() { return answer(StatusCode::UNAUTHORIZED, "需要登录后查看"); } - upgrade - .read_buffer_size(SOCKET_BUFFER) - .max_message_size(MAX_FRAME) - .on_upgrade(move |socket| stream_live(app, socket, session, stream.gzip.is_some())) + let Ok(slot) = stream_slot(session.as_deref()) else { + return answer(StatusCode::SERVICE_UNAVAILABLE, "同时在线的访客太多，请稍后再试"); + }; + upgrade.read_buffer_size(SOCKET_BUFFER).max_message_size(MAX_FRAME).on_upgrade(move |socket| async move { + // Held until the stream ends. + let _slot = slot; + stream_live(app, socket, session, stream.gzip.is_some()).await + }) } async fn stream_live(app: Shared, mut socket: WebSocket, session: Option, gzip: bool) { @@ -693,7 +722,7 @@ async fn stream_live(app: Shared, mut socket: WebSocket, session: Option } else { Message::Text(live_snapshot(&app, full)) }; - if socket.send(frame).await.is_err() { + if !matches!(tokio::time::timeout(SEND_TIMEOUT, socket.send(frame)).await, Ok(Ok(()))) { break; } } @@ -3598,6 +3627,17 @@ mod tests { assert_eq!(app.db.node(id).unwrap().unwrap().disk_total, 30i64 << 30, "and no extra write to get it"); } + /// Only anonymous streams are counted, and a slot comes back when its + /// stream ends. `PUBLIC_STREAM_GATE` is touched by no other test. + #[test] + fn anonymous_streams_past_the_gate_are_refused_and_a_session_is_not() { + let held: Vec<_> = (0..PUBLIC_STREAMS).map(|_| stream_slot(None).expect("up to the limit")).collect(); + assert!(stream_slot(None).is_err(), "the stream past the limit must be refused"); + assert!(stream_slot(Some("hash")).is_ok(), "a session opens its stream with every slot taken"); + drop(held); + assert!(stream_slot(None).is_ok(), "a finished stream gives its slot back"); + } + /// `span` bounds one window; this bounds how many are built concurrently. /// Each holds the reader for its entire scan, and the path takes no /// credentials. `PASSWORD_GATE` refuses the same way; `RELAY_GATE` queues diff --git a/src/main.rs b/src/main.rs index ceb18411..d0a3e309 100644 --- a/src/main.rs +++ b/src/main.rs @@ -396,6 +396,41 @@ fn parse_args() -> Result { }) } +/// The soft open-file limit the hub asks for. Every agent holds one descriptor, +/// every open page another, and every request in flight a third. 65536 leaves +/// about 65 for each of the 1000 nodes a hub is sized for. +const NOFILE_TARGET: u64 = 65_536; + +/// Raises the soft open-file limit toward [`NOFILE_TARGET`], capped by the hard +/// one. systemd and Docker start a process at a soft limit of 1024, which Rust +/// does not raise as Go does: with about a thousand connections `accept` fails +/// with EMFILE, which the server retries silently, and the hub stops answering +/// without logging a line (measured: 846 of 1000 agents connected, `/api/me` +/// unanswered). +fn raise_open_files() { + use rustix::process::{getrlimit, setrlimit, Resource}; + let mut limit = getrlimit(Resource::Nofile); + // Unlimited needs nothing. + let Some(soft) = limit.current else { return }; + let wanted = limit.maximum.map_or(NOFILE_TARGET, |hard| hard.min(NOFILE_TARGET)); + let mut now = soft; + if soft < wanted { + limit.current = Some(wanted); + match setrlimit(Resource::Nofile, limit) { + Ok(()) => now = wanted, + Err(e) => warn!("open-file limit could not be raised from {soft}: {e}"), + } + } + if now < NOFILE_TARGET { + warn!( + "open-file limit is {now}, and the hub stops accepting connections near it. Raise \ + LimitNOFILE for the service, or --ulimit nofile for the container" + ); + } else if now > soft { + info!("open-file limit raised from {soft} to {now}"); + } +} + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -417,6 +452,8 @@ async fn main() -> Result<()> { println!("Emergency password: {}", new_password(&Db::open(&args.database)?)?); return Ok(()); } + // After the two arguments that print and exit, so their output stays theirs. + raise_open_files(); std::fs::create_dir_all(&args.themes)?; if let Err(e) = db::temp_files_beside(&args.database) { warn!("SQLite keeps its temporary files in its default directory: {e:#}"); @@ -530,7 +567,7 @@ async fn main() -> Result<()> { .layer(tower_http::limit::RequestBodyLimitLayer::new(64 * 1024)) // The two chunked uploads, merged after that layer rather than beneath // it. They raise the ceiling on a single request to `api::MAX_CHUNK`, - // not on the file behind it: a 256 MiB backup arrives as 64 of the + // not on the file behind it: a 1 GiB backup arrives as 256 of the // panel's 4 MiB pieces, so no reverse proxy needs to know the database // size. The whole-file ceilings live on `total` and are checked on the // first request. @@ -610,8 +647,7 @@ fn advertised_url(site: &str, listen: SocketAddr) -> String { /// This host's own address on its outbound route. Asking the kernel to route a /// datagram it never sends is the cheapest way to select one interface among /// several, and it answers without any network traffic. Behind NAT it yields the -/// private address, since the hub cannot know its public one, which is why -/// install-hub.sh prints the address it looked up instead. +/// private address, since the hub cannot know its public one; `--site` states it. fn outbound_ip() -> Option { [("0.0.0.0:0", "1.1.1.1:80"), ("[::]:0", "[2606:4700:4700::1111]:80")].into_iter().find_map( |(bind, route_to)| { @@ -839,6 +875,20 @@ mod tests { use super::*; use axum::http::{StatusCode, Uri}; + /// The limit only moves up, so raising it here cannot disturb another test. + #[test] + fn the_soft_open_file_limit_reaches_the_target_or_the_hard_one() { + use rustix::process::{getrlimit, Resource}; + raise_open_files(); + let limit = getrlimit(Resource::Nofile); + let wanted = limit.maximum.map_or(NOFILE_TARGET, |hard| hard.min(NOFILE_TARGET)); + assert!( + limit.current.is_none_or(|soft| soft >= wanted), + "soft limit {:?} below {wanted}", + limit.current + ); + } + fn app(site: &str) -> App { App::new( Db::open(":memory:").unwrap(),