[PATCH pve-cluster 05/12] rust: notify: add the socket server

[Hannes Laimer <h.laimer@proxmox.com>]

Daemons learn about changes in /etc/pve by polling the file versions
over IPC, and inotify only sees writes made on the local node. Add a
server that pushes each mutation it is handed to the connections whose
subscription matches it. It runs on its own thread, and only root and
one configured group can connect to its Unix socket.

A memdb mutation holds the database mutex, so the server must never
wait on a client when it is handed an event. It just puts the event
into a bounded channel that every client reads at its own pace. A
client that falls behind by more than the channel holds gets one
resync in place of what it missed. If a write to a client does not
complete within 30 seconds, its connection is dropped.

No event history is kept beyond the channel. A new subscription starts
at the current state, so the server answers a subscribe with a resync
and the client reads everything once. A resync from a lagged channel
asks for the same.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/rust/pmxcfs-notify/Cargo.toml    |   3 +
 src/rust/pmxcfs-notify/src/conn.rs   | 196 +++++++++-
 src/rust/pmxcfs-notify/src/lib.rs    |  11 +-
 src/rust/pmxcfs-notify/src/server.rs | 542 +++++++++++++++++++++++++++
 4 files changed, 749 insertions(+), 3 deletions(-)
 create mode 100644 src/rust/pmxcfs-notify/src/server.rs

diff --git a/src/rust/pmxcfs-notify/Cargo.toml b/src/rust/pmxcfs-notify/Cargo.toml
index 5a49c75..bd29936 100644
--- a/src/rust/pmxcfs-notify/Cargo.toml
+++ b/src/rust/pmxcfs-notify/Cargo.toml
@@ -10,5 +10,8 @@ rust-version.workspace = true
 
 [dependencies]
 anyhow.workspace = true
+log.workspace = true
+nix.workspace = true
 serde.workspace = true
 serde_json.workspace = true
+tokio.workspace = true
diff --git a/src/rust/pmxcfs-notify/src/conn.rs b/src/rust/pmxcfs-notify/src/conn.rs
index 451504c..60319f5 100644
--- a/src/rust/pmxcfs-notify/src/conn.rs
+++ b/src/rust/pmxcfs-notify/src/conn.rs
@@ -1,13 +1,36 @@
 use std::collections::BTreeMap;
+use std::io;
+use std::time::Duration;
 
-use crate::protocol::Params;
+use serde_json::{Value, json};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::UnixStream;
+use tokio::sync::broadcast::{self, error::RecvError};
+use tokio::time::timeout;
+
+use crate::protocol::{Command, Event, Message, PROTOCOL_VERSION, Params};
+use crate::server::{Entry, Slot};
 use crate::template::Template;
 
+const MAX_LINE: usize = 64 * 1024;
+#[cfg(not(test))]
+const WRITE_TIMEOUT: Duration = Duration::from_secs(30);
+#[cfg(test)]
+const WRITE_TIMEOUT: Duration = Duration::from_secs(2);
+
 pub(crate) struct Conn {
     privileged: bool,
     templates: BTreeMap<String, Template>,
 }
 
+#[derive(Debug)]
+enum ReadError {
+    Eof,
+    TooLong,
+    NotUtf8,
+    Io(io::Error),
+}
+
 impl Conn {
     pub(crate) fn new(privileged: bool) -> Self {
         Self {
@@ -41,6 +64,118 @@ impl Conn {
     }
 }
 
+pub(crate) async fn serve(mut stream: UnixStream, privileged: bool, slot: Slot) {
+    let mut conn = Conn::new(privileged);
+    let mut events: Option<broadcast::Receiver<Entry>> = None;
+    let mut inbuf = Vec::new();
+    loop {
+        let line = tokio::select! {
+            line = read_line(&mut stream, &mut inbuf) => {
+                let line = match line {
+                    Ok(line) => line,
+                    Err(ReadError::Eof) => return,
+                    Err(ReadError::TooLong) => {
+                        log::warn!("closing connection, line too long");
+                        return;
+                    }
+                    Err(ReadError::NotUtf8) => {
+                        log::warn!("closing connection, line is not UTF-8");
+                        return;
+                    }
+                    Err(ReadError::Io(err)) => {
+                        log::debug!("closing connection: {err}");
+                        return;
+                    }
+                };
+                match serde_json::from_str::<Command>(&line) {
+                    Ok(Command::Hello) => {
+                        Message::Ok(json!({ "protocol": PROTOCOL_VERSION })).line()
+                    }
+                    Ok(Command::Subscribe { patterns }) => {
+                        conn.templates = patterns;
+                        events = Some(slot.events.subscribe());
+                        Message::Ok(Value::Null).line() + &Message::resync().line()
+                    }
+                    Ok(Command::Unsubscribe) => {
+                        conn.templates.clear();
+                        events = None;
+                        Message::Ok(Value::Null).line()
+                    }
+                    Err(err) => Message::Error(err.to_string()).line(),
+                }
+            }
+            event = next_event(&mut events) => match event {
+                Ok(Entry::Resync) => Message::resync().line(),
+                Ok(Entry::Change { kind, path, to }) => {
+                    match conn.matches(&path, to.as_deref()) {
+                        Some(params) => Message::Event(Event {
+                            kind,
+                            path: Some(&path),
+                            to: to.as_deref(),
+                            params: Some(params),
+                        })
+                        .line(),
+                        None => continue,
+                    }
+                }
+                // the resync stands for everything up to now, so a fresh receiver starting there
+                // replaces one that still holds older entries
+                Err(RecvError::Lagged(_)) => {
+                    events = Some(slot.events.subscribe());
+                    Message::resync().line()
+                }
+                Err(RecvError::Closed) => return,
+            },
+        };
+        if !write(&mut stream, &line).await {
+            return;
+        }
+    }
+}
+
+async fn next_event(events: &mut Option<broadcast::Receiver<Entry>>) -> Result<Entry, RecvError> {
+    match events {
+        Some(events) => events.recv().await,
+        None => std::future::pending().await,
+    }
+}
+
+async fn read_line(stream: &mut UnixStream, inbuf: &mut Vec<u8>) -> Result<String, ReadError> {
+    loop {
+        if let Some(end) = inbuf.iter().position(|&b| b == b'\n') {
+            let line = inbuf[..end].to_vec();
+            inbuf.drain(..=end);
+            return String::from_utf8(line).map_err(|_| ReadError::NotUtf8);
+        }
+        if inbuf.len() > MAX_LINE {
+            return Err(ReadError::TooLong);
+        }
+        let mut chunk = [0u8; 4096];
+        match stream.read(&mut chunk).await {
+            Ok(0) => return Err(ReadError::Eof),
+            Ok(n) => inbuf.extend_from_slice(&chunk[..n]),
+            Err(err) if err.kind() == io::ErrorKind::Interrupted => {}
+            Err(err) => return Err(ReadError::Io(err)),
+        }
+    }
+}
+
+/// Returns whether the line went out. A client whose socket stays full for the whole timeout is
+/// taken to have stopped reading and is cut off.
+async fn write(stream: &mut UnixStream, line: &str) -> bool {
+    match timeout(WRITE_TIMEOUT, stream.write_all(line.as_bytes())).await {
+        Ok(Ok(())) => true,
+        Ok(Err(err)) => {
+            log::debug!("closing connection: {err}");
+            false
+        }
+        Err(_) => {
+            log::warn!("closing connection, client stopped reading");
+            false
+        }
+    }
+}
+
 // Same rule as path_is_private in the daemon.
 fn is_private(path: &str) -> bool {
     fn priv_component(path: &str) -> bool {
@@ -67,6 +202,65 @@ mod tests {
         conn
     }
 
+    async fn line_or_pending(
+        stream: &mut UnixStream,
+        inbuf: &mut Vec<u8>,
+    ) -> Option<Result<String, ReadError>> {
+        timeout(Duration::from_millis(100), read_line(stream, inbuf))
+            .await
+            .ok()
+    }
+
+    #[tokio::test]
+    async fn reassembles_partial_lines() {
+        let (mut server, mut client) = UnixStream::pair().unwrap();
+        let mut inbuf = Vec::new();
+        client.write_all(b"{\"command\":").await.unwrap();
+        assert!(line_or_pending(&mut server, &mut inbuf).await.is_none());
+        client
+            .write_all(b"\"hello\"}\n{\"command\":\"unsubscribe\"}\n")
+            .await
+            .unwrap();
+        assert_eq!(
+            read_line(&mut server, &mut inbuf).await.unwrap(),
+            "{\"command\":\"hello\"}"
+        );
+        assert_eq!(
+            read_line(&mut server, &mut inbuf).await.unwrap(),
+            "{\"command\":\"unsubscribe\"}"
+        );
+        assert!(line_or_pending(&mut server, &mut inbuf).await.is_none());
+        assert!(inbuf.is_empty());
+    }
+
+    #[tokio::test]
+    async fn rejects_overlong_and_non_utf8_lines() {
+        let (mut server, mut client) = UnixStream::pair().unwrap();
+        let mut inbuf = Vec::new();
+        tokio::spawn(async move {
+            client.write_all(&vec![b'x'; MAX_LINE + 1]).await.unwrap();
+            std::future::pending::<()>().await;
+        });
+        assert!(matches!(
+            read_line(&mut server, &mut inbuf).await,
+            Err(ReadError::TooLong)
+        ));
+
+        let (mut server, mut client) = UnixStream::pair().unwrap();
+        let mut inbuf = Vec::new();
+        client.write_all(b"\xff\xfe\n").await.unwrap();
+        assert!(matches!(
+            read_line(&mut server, &mut inbuf).await,
+            Err(ReadError::NotUtf8)
+        ));
+
+        drop(client);
+        assert!(matches!(
+            read_line(&mut server, &mut inbuf).await,
+            Err(ReadError::Eof)
+        ));
+    }
+
     #[test]
     fn matches_per_template() {
         let conn = subscribed(
diff --git a/src/rust/pmxcfs-notify/src/lib.rs b/src/rust/pmxcfs-notify/src/lib.rs
index 18e7d2f..c152d43 100644
--- a/src/rust/pmxcfs-notify/src/lib.rs
+++ b/src/rust/pmxcfs-notify/src/lib.rs
@@ -12,13 +12,20 @@
 //! `nodes/{node}/qemu-server/{vmid}.conf` captures node and vmid, and `{path...}` matches every
 //! path. Connections admitted by their group id rather than as root never receive events that
 //! mention a private path, just as IPC and FUSE hide private files from them.
-
-#![allow(dead_code)]
+//!
+//! Nothing is kept for a client beyond a bounded backlog. A client that falls further behind gets
+//! one resync event in place of what it missed, the same event every client gets when the daemon
+//! replaces its state. A subscription starts at the current state, so its reply is followed by a
+//! resync. A resync means reading everything again.
 
 mod conn;
 mod protocol;
+mod server;
 mod template;
 
+pub use protocol::EventKind;
+pub use server::{Config, Server};
+
 /// Events and templates name memdb paths without a leading slash.
 pub(crate) fn normalize_path(path: &str) -> &str {
     path.trim_start_matches('/')
diff --git a/src/rust/pmxcfs-notify/src/server.rs b/src/rust/pmxcfs-notify/src/server.rs
new file mode 100644
index 0000000..fe5defc
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/server.rs
@@ -0,0 +1,542 @@
+use std::fs;
+use std::io;
+use std::os::unix::fs::PermissionsExt;
+use std::os::unix::net::UnixListener as StdUnixListener;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::thread::JoinHandle;
+use std::time::Duration;
+
+use anyhow::{Context, Error};
+use nix::sys::socket::{setsockopt, sockopt};
+use nix::sys::stat::{Mode, umask};
+use tokio::net::{UnixListener, UnixStream};
+use tokio::sync::{Notify, broadcast};
+
+use crate::conn;
+use crate::normalize_path;
+use crate::protocol::EventKind;
+
+const MAX_CONNECTIONS: usize = 128;
+const SNDBUF: usize = 2 * 1024 * 1024;
+
+pub struct Config {
+    pub path: PathBuf,
+    pub gid: u32,
+    /// Events kept for a client that reads slower than they arrive.
+    pub capacity: usize,
+}
+
+// Reference counted strings, so handing a copy to every connection clones only pointers.
+#[derive(Clone)]
+pub(crate) enum Entry {
+    Resync,
+    Change {
+        kind: EventKind,
+        path: Arc<str>,
+        to: Option<Arc<str>>,
+    },
+}
+
+pub(crate) struct Shared {
+    pub(crate) events: broadcast::Sender<Entry>,
+    stop: Notify,
+    connections: AtomicUsize,
+    gid: u32,
+}
+
+/// A connection's place under the cap, held for as long as its task runs.
+pub(crate) struct Slot(Arc<Shared>);
+
+impl std::ops::Deref for Slot {
+    type Target = Shared;
+
+    fn deref(&self) -> &Shared {
+        &self.0
+    }
+}
+
+impl Drop for Slot {
+    fn drop(&mut self) {
+        self.0.connections.fetch_sub(1, Ordering::Relaxed);
+    }
+}
+
+pub struct Server {
+    shared: Arc<Shared>,
+    thread: Option<JoinHandle<()>>,
+}
+
+impl Server {
+    /// Changes the process umask around the bind, so no other thread may be creating files while
+    /// this runs.
+    pub fn start(config: Config) -> Result<Server, Error> {
+        let listener = bind(&config.path, config.gid)?;
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_io()
+            .enable_time()
+            .build()?;
+        let listener = {
+            let _runtime = runtime.enter();
+            UnixListener::from_std(listener)?
+        };
+        let (events, _) = broadcast::channel(config.capacity);
+        let shared = Arc::new(Shared {
+            events,
+            stop: Notify::new(),
+            connections: AtomicUsize::new(0),
+            gid: config.gid,
+        });
+        let thread = {
+            let shared = Arc::clone(&shared);
+            std::thread::Builder::new()
+                .name("pmxcfs-notify".to_owned())
+                .spawn(move || {
+                    let _socket = RemoveOnExit(config.path);
+                    runtime.block_on(accept_loop(listener, shared));
+                })?
+        };
+        Ok(Server {
+            shared,
+            thread: Some(thread),
+        })
+    }
+
+    /// Hands the mutation to every subscribed connection without waiting on any of them, so it
+    /// never blocks the caller. Every kind but a resync needs a path.
+    pub fn emit(&self, kind: EventKind, path: Option<&str>, to: Option<&str>) {
+        let entry = match (kind, path) {
+            (EventKind::Resync, _) => Entry::Resync,
+            (kind, Some(path)) => Entry::Change {
+                kind,
+                path: normalize_path(path).into(),
+                to: to.map(|to| normalize_path(to).into()),
+            },
+            (kind, None) => {
+                log::warn!("ignoring {kind:?} event without a path");
+                return;
+            }
+        };
+        // an error only says that no connection is subscribed
+        let _ = self.shared.events.send(entry);
+    }
+
+    pub fn shutdown(mut self) {
+        self.stop_thread();
+    }
+
+    fn stop_thread(&mut self) {
+        let Some(thread) = self.thread.take() else {
+            return;
+        };
+        self.shared.stop.notify_one();
+        let _ = thread.join();
+    }
+
+    #[cfg(test)]
+    pub(crate) fn connection_count(&self) -> usize {
+        self.shared.connections.load(Ordering::Relaxed)
+    }
+}
+
+impl Drop for Server {
+    fn drop(&mut self) {
+        self.stop_thread();
+    }
+}
+
+// The socket file goes away with the thread however the thread ends, a panic included, so no stale
+// socket is left behind.
+struct RemoveOnExit(PathBuf);
+
+impl Drop for RemoveOnExit {
+    fn drop(&mut self) {
+        let _ = fs::remove_file(&self.0);
+    }
+}
+
+// The socket is created with mode 0700, so between bind and chmod nobody but its owner can connect.
+fn bind(path: &Path, gid: u32) -> Result<StdUnixListener, Error> {
+    match fs::remove_file(path) {
+        Ok(()) => {}
+        Err(err) if err.kind() == io::ErrorKind::NotFound => {}
+        Err(err) => {
+            return Err(err).with_context(|| format!("failed to remove stale {}", path.display()));
+        }
+    }
+    let mask = umask(Mode::from_bits_truncate(0o077));
+    let listener = StdUnixListener::bind(path);
+    umask(mask);
+    let listener = listener.with_context(|| format!("failed to bind {}", path.display()))?;
+    listener.set_nonblocking(true)?;
+    std::os::unix::fs::chown(path, None, Some(gid))?;
+    fs::set_permissions(path, fs::Permissions::from_mode(0o660))?;
+    Ok(listener)
+}
+
+async fn accept_loop(listener: UnixListener, shared: Arc<Shared>) {
+    let mut accept_failing = false;
+    loop {
+        let stream = tokio::select! {
+            accepted = listener.accept() => match accepted {
+                Ok((stream, _)) => {
+                    accept_failing = false;
+                    stream
+                }
+                Err(err) => {
+                    // a failed accept leaves the connection queued and the listener readable, so
+                    // an immediate retry would spin. A lasting cause like exhausted descriptors is
+                    // logged once, not on every round
+                    if !accept_failing {
+                        log::warn!("accept failed: {err}");
+                        accept_failing = true;
+                    }
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                    continue;
+                }
+            },
+            () = shared.stop.notified() => return,
+        };
+        admit(stream, &shared);
+    }
+}
+
+fn admit(stream: UnixStream, shared: &Arc<Shared>) {
+    let cred = match stream.peer_cred() {
+        Ok(cred) => cred,
+        Err(err) => {
+            log::warn!("failed to get peer credentials: {err}");
+            return;
+        }
+    };
+    let privileged = cred.uid() == 0 && cred.gid() == 0;
+    if !privileged && cred.gid() != shared.gid {
+        log::warn!(
+            "connection from bad user {}/{} rejected",
+            cred.uid(),
+            cred.gid()
+        );
+        return;
+    }
+    if shared.connections.load(Ordering::Relaxed) >= MAX_CONNECTIONS {
+        log::warn!("connection limit of {MAX_CONNECTIONS} reached, refusing");
+        return;
+    }
+    // root may raise the send buffer past wmem_max, which gives a slow client room for thousands
+    // of events before it falls behind
+    if setsockopt(&stream, sockopt::SndBufForce, &SNDBUF).is_err()
+        && setsockopt(&stream, sockopt::SndBuf, &SNDBUF).is_err()
+    {
+        log::debug!("could not raise the send buffer of a connection");
+    }
+    shared.connections.fetch_add(1, Ordering::Relaxed);
+    tokio::spawn(conn::serve(stream, privileged, Slot(Arc::clone(shared))));
+}
+
+#[cfg(test)]
+mod tests {
+    use std::io::{BufRead, BufReader, Write};
+    use std::os::unix::fs::MetadataExt;
+    use std::os::unix::net::UnixStream;
+    use std::time::Instant;
+
+    use serde_json::{Value, json};
+
+    use super::*;
+
+    static COUNTER: AtomicUsize = AtomicUsize::new(0);
+
+    struct TempDir(PathBuf);
+
+    impl Drop for TempDir {
+        fn drop(&mut self) {
+            let _ = fs::remove_dir_all(&self.0);
+        }
+    }
+
+    fn start(capacity: usize) -> (Server, PathBuf, TempDir) {
+        let n = COUNTER.fetch_add(1, Ordering::SeqCst);
+        let dir = std::env::temp_dir().join(format!("pmxcfs-notify-{}-{n}", std::process::id()));
+        fs::create_dir_all(&dir).unwrap();
+        let gid = fs::metadata(&dir).unwrap().gid();
+        let path = dir.join("sock");
+        let server = Server::start(Config {
+            path: path.clone(),
+            gid,
+            capacity,
+        })
+        .unwrap();
+        (server, path, TempDir(dir))
+    }
+
+    fn connect(path: &PathBuf) -> (UnixStream, BufReader<UnixStream>) {
+        let stream = UnixStream::connect(path).unwrap();
+        stream
+            .set_read_timeout(Some(Duration::from_secs(5)))
+            .unwrap();
+        let reader = BufReader::new(stream.try_clone().unwrap());
+        (stream, reader)
+    }
+
+    fn request(stream: &mut UnixStream, reader: &mut BufReader<UnixStream>, line: &str) -> String {
+        stream.write_all(line.as_bytes()).unwrap();
+        stream.write_all(b"\n").unwrap();
+        let mut reply = String::new();
+        reader.read_line(&mut reply).unwrap();
+        reply
+    }
+
+    fn subscribe(stream: &mut UnixStream, reader: &mut BufReader<UnixStream>, patterns: &str) {
+        let line = format!("{{\"command\":\"subscribe\",\"args\":{{\"patterns\":{patterns}}}}}");
+        assert_eq!(request(stream, reader, &line), "{\"ok\":null}\n");
+        assert_eq!(event(reader), json!({ "type": "resync" }));
+    }
+
+    fn event(reader: &mut BufReader<UnixStream>) -> Value {
+        let mut line = String::new();
+        reader.read_line(&mut line).unwrap();
+        serde_json::from_str::<Value>(&line).unwrap()["event"].clone()
+    }
+
+    /// The next event if one arrives within a short timeout, None otherwise.
+    fn pending(reader: &mut BufReader<UnixStream>) -> Option<Value> {
+        let stream = reader.get_ref();
+        stream
+            .set_read_timeout(Some(Duration::from_millis(300)))
+            .unwrap();
+        let mut line = String::new();
+        let result = reader.read_line(&mut line);
+        reader
+            .get_ref()
+            .set_read_timeout(Some(Duration::from_secs(5)))
+            .unwrap();
+        match result {
+            Ok(n) if n > 0 => Some(serde_json::from_str::<Value>(&line).unwrap()["event"].clone()),
+            _ => None,
+        }
+    }
+
+    fn wait_for(mut condition: impl FnMut() -> bool) {
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while !condition() {
+            assert!(Instant::now() < deadline, "condition not met in time");
+            std::thread::sleep(Duration::from_millis(10));
+        }
+    }
+
+    #[test]
+    fn subscribe_and_receive() {
+        let (server, path, _dir) = start(64);
+        let meta = fs::metadata(&path).unwrap();
+        assert_eq!(meta.mode() & 0o777, 0o660);
+
+        let (mut stream, mut reader) = connect(&path);
+        assert_eq!(
+            request(&mut stream, &mut reader, r#"{"command":"hello"}"#),
+            "{\"ok\":{\"protocol\":1}}\n"
+        );
+        assert_eq!(server.connection_count(), 1);
+        let reply = request(&mut stream, &mut reader, r#"{"command":"bogus"}"#);
+        assert!(
+            reply.starts_with("{\"error\":\"unknown variant `bogus`"),
+            "{reply}"
+        );
+        server.emit(EventKind::Write, Some("datacenter.cfg"), None);
+        subscribe(
+            &mut stream,
+            &mut reader,
+            r#"{"guest":"/nodes/{node}/qemu-server/{vmid}.conf","dc":"datacenter.cfg"}"#,
+        );
+        // a rejected subscription leaves the previous one in place
+        let reply = request(
+            &mut stream,
+            &mut reader,
+            r#"{"command":"subscribe","args":{"patterns":{"bad":"{a}{b}"}}}"#,
+        );
+        assert!(
+            reply.starts_with("{\"error\":\"only one placeholder per component in '{a}{b}'"),
+            "{reply}"
+        );
+
+        server.emit(EventKind::Write, Some("storage.cfg"), None);
+        server.emit(
+            EventKind::Create,
+            Some("/nodes/n1/qemu-server/100.conf"),
+            None,
+        );
+        server.emit(
+            EventKind::Write,
+            Some("nodes/n1/qemu-server/100.conf.tmp.1"),
+            None,
+        );
+        server.emit(
+            EventKind::Rename,
+            Some("nodes/n1/qemu-server/100.conf.tmp.1"),
+            Some("datacenter.cfg"),
+        );
+        server.emit(EventKind::Resync, None, None);
+
+        let create = event(&mut reader);
+        assert_eq!(
+            create,
+            json!({
+                "type": "create",
+                "path": "nodes/n1/qemu-server/100.conf",
+                "params": { "guest": [{ "node": "n1", "vmid": "100" }] },
+            })
+        );
+        let rename = event(&mut reader);
+        assert_eq!(rename["type"], "rename");
+        assert_eq!(rename["to"], "datacenter.cfg");
+        assert_eq!(rename["params"], json!({ "dc": [{}] }));
+        assert_eq!(event(&mut reader), json!({ "type": "resync" }));
+
+        assert_eq!(
+            request(&mut stream, &mut reader, r#"{"command":"unsubscribe"}"#),
+            "{\"ok\":null}\n"
+        );
+        server.emit(EventKind::Write, Some("datacenter.cfg"), None);
+        server.emit(EventKind::Resync, None, None);
+        // nothing emitted while unsubscribed shows up after subscribing again
+        subscribe(&mut stream, &mut reader, r#"{"dc":"datacenter.cfg"}"#);
+        server.emit(EventKind::Write, Some("datacenter.cfg"), None);
+        assert_eq!(event(&mut reader)["type"], "write");
+        assert!(pending(&mut reader).is_none());
+
+        server.shutdown();
+        let mut line = String::new();
+        assert_eq!(reader.read_line(&mut line).unwrap(), 0);
+        assert!(!path.exists());
+    }
+
+    #[test]
+    fn private_paths_follow_credentials() {
+        let (server, path, _dir) = start(64);
+        let meta = fs::metadata(&path).unwrap();
+        let privileged = meta.uid() == 0 && meta.gid() == 0;
+
+        let (mut stream, mut reader) = connect(&path);
+        subscribe(&mut stream, &mut reader, r#"{"all":"{path...}"}"#);
+        server.emit(EventKind::Write, Some("nodes/n1/priv/ssl.key"), None);
+        server.emit(EventKind::Write, Some("nodes/n1/pve-ssl.pem"), None);
+
+        if privileged {
+            assert_eq!(event(&mut reader)["path"], "nodes/n1/priv/ssl.key");
+        }
+        let public = event(&mut reader);
+        assert_eq!(public["path"], "nodes/n1/pve-ssl.pem");
+        assert_eq!(
+            public["params"],
+            json!({ "all": [{ "path": "nodes/n1/pve-ssl.pem" }] })
+        );
+    }
+
+    #[test]
+    fn client_close_is_noticed() {
+        let (server, path, _dir) = start(64);
+        let (mut stream, mut reader) = connect(&path);
+        request(&mut stream, &mut reader, r#"{"command":"hello"}"#);
+        assert_eq!(server.connection_count(), 1);
+        drop(reader);
+        drop(stream);
+        wait_for(|| server.connection_count() == 0);
+    }
+
+    #[test]
+    fn connections_are_capped() {
+        let (server, path, _dir) = start(64);
+        let mut streams: Vec<UnixStream> = (0..MAX_CONNECTIONS)
+            .map(|_| UnixStream::connect(&path).unwrap())
+            .collect();
+        wait_for(|| server.connection_count() == MAX_CONNECTIONS);
+
+        let (_stream, mut reader) = connect(&path);
+        let mut line = String::new();
+        assert_eq!(reader.read_line(&mut line).unwrap(), 0, "one too many");
+
+        drop(streams.pop());
+        wait_for(|| server.connection_count() == MAX_CONNECTIONS - 1);
+        let (mut stream, mut reader) = connect(&path);
+        assert_eq!(
+            request(&mut stream, &mut reader, r#"{"command":"hello"}"#),
+            "{\"ok\":{\"protocol\":1}}\n"
+        );
+    }
+
+    fn numbered(n: usize) -> String {
+        format!("{}/f{n}", "x".repeat(3000))
+    }
+
+    /// Fills the client's socket while it is not reading. Returns the number of events emitted,
+    /// numbered from 1 upward.
+    fn burst(server: &Server) -> usize {
+        let count = 2000;
+        for n in 1..=count {
+            server.emit(EventKind::Write, Some(&numbered(n)), None);
+        }
+        count
+    }
+
+    #[test]
+    fn a_slow_client_is_resynced() {
+        let (server, path, _dir) = start(100);
+        let (mut stream, mut reader) = connect(&path);
+        subscribe(&mut stream, &mut reader, r#"{"all":"{filler}/f{n}"}"#);
+        let count = burst(&server);
+
+        // where the client falls behind depends on scheduling. It always gets at least one resync
+        // and no event twice, and after the last resync a gapless run of events up to the newest
+        // one, if any
+        let mut events = vec![event(&mut reader)];
+        while let Some(ev) = pending(&mut reader) {
+            events.push(ev);
+        }
+        let number = |ev: &Value| -> usize {
+            ev["params"]["all"][0]["n"]
+                .as_str()
+                .unwrap()
+                .parse()
+                .unwrap()
+        };
+        let writes: Vec<usize> = events
+            .iter()
+            .filter(|ev| ev["type"] == "write")
+            .map(number)
+            .collect();
+        assert!(writes.len() < count, "nothing was missed");
+        assert!(
+            writes.windows(2).all(|w| w[0] < w[1]),
+            "an event arrived twice"
+        );
+        let last_resync = events
+            .iter()
+            .rposition(|ev| ev["type"] == "resync")
+            .expect("the client never fell behind");
+        let tail: Vec<usize> = events[last_resync + 1..].iter().map(number).collect();
+        assert!(
+            tail.windows(2).all(|w| w[1] == w[0] + 1),
+            "gap after the last resync"
+        );
+        if let Some(last) = tail.last() {
+            assert_eq!(*last, count, "the newest event never arrived");
+        }
+
+        server.emit(EventKind::Write, Some(&numbered(0)), None);
+        assert_eq!(number(&event(&mut reader)), 0);
+        assert_eq!(server.connection_count(), 1);
+    }
+
+    #[test]
+    fn a_client_that_stopped_reading_is_cut_off() {
+        // room for the whole burst, so the socket fills instead of the channel
+        let (server, path, _dir) = start(4096);
+        let (mut stream, mut reader) = connect(&path);
+        subscribe(&mut stream, &mut reader, r#"{"all":"{path...}"}"#);
+        burst(&server);
+        wait_for(|| server.connection_count() == 0);
+        // what the socket still held is readable, then EOF rather than a reset
+        let mut sink = Vec::new();
+        assert!(std::io::Read::read_to_end(&mut stream, &mut sink).is_ok());
+    }
+}
-- 
2.47.3