fix: 文件描述符软上限提到 65536，匿名推送限 1000 条同时在线

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"

src/api.rs

@@ -635,6 +635,31 @@ pub struct Stream {
     gzip: Option<String>,
 }
 
+/// 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<Option<tokio::sync::SemaphorePermit<'static>>, ()> {
+    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<String>, gzip: bool) {
@@ -693,7 +722,7 @@ async fn stream_live(app: Shared, mut socket: WebSocket, session: Option<String>
         } 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

src/main.rs

@@ -396,6 +396,41 @@ fn parse_args() -> Result<Args> {
     })
 }
 
+/// 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<IpAddr> {
     [("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(),