horizon-server: the management protocol pushes notifications, starting with server/started and server/stopping

Vanilla's JsonRpcNotificationService broadcasts to every connection of the ManagementServer. Horizon
served the protocol but never sent one. The supervisor now keeps the upgraded connections and a queue
with a thread that writes to them, so the game's output pump and the supervisor never wait on a slow
client. A notification is a JSON-RPC request without an id, its one parameter in an array, as
JsonRPCUtils.createRequest writes it.

server/started goes out at the game's Done line (vanilla: DedicatedServer.initServer, after Done).
server/stopping goes out when the supervisor starts a stop or a restart, or when the game's own stop
logs "Stopping server" (vanilla: DedicatedServer.stopServer, first of all), once per run of the game.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>

crates/horizon-server/src/api.rs

@@ -342,6 +342,8 @@ pub struct Shared {
     /// The event stream's events (events.rs), and the streams' wake-up.
     pub events: Mutex<crate::events::Events>,
     pub event_added: std::sync::Condvar,
+    /// The management protocol's connections, which its notifications go to (management.rs).
+    pub management: crate::management::Clients,
 }
 
 impl Shared {
@@ -372,6 +374,7 @@ impl Shared {
             update: Mutex::new(Updates::default()),
             events: Mutex::default(),
             event_added: std::sync::Condvar::new(),
+            management: crate::management::Clients::default(),
         })
     }
 

crates/horizon-server/src/management.rs

@@ -16,10 +16,69 @@ use serde_json::{Value, json};
 use std::{
     net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs},
     path::Path,
-    sync::Arc,
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicBool, Ordering},
+        mpsc,
+    },
     time::Duration,
 };
 
+/// The connections that have upgraded, and the queue that carries notifications to them, which
+/// `JsonRpcNotificationService.broadcastNotification` does to every `Connection` of the `ManagementServer`.
+/// A queue and a thread, so that a caller (the game's output pump, the supervisor) never waits on a slow client.
+#[derive(Default)]
+pub struct Clients {
+    sockets: Mutex<Vec<(u32, Arc<Socket>)>>,
+    outbox: Mutex<Option<mpsc::Sender<String>>>,
+    /// `server/stopping` went out for this run of the game: a stop comes from the supervisor and from the
+    /// game's own `stop`, and one run says it once.
+    stopping_told: AtomicBool,
+}
+
+/// A connection in `Clients` for as long as this lives.
+struct Registered<'a>(&'a Shared, u32);
+
+impl Drop for Registered<'_> {
+    fn drop(&mut self) {
+        let id = self.1;
+        self.0
+            .management
+            .sockets
+            .lock()
+            .unwrap()
+            .retain(|(client, _)| *client != id);
+    }
+}
+
+/// A notification to every connected client: `short` is the method without `minecraft:` and `params` its one
+/// parameter, as `JsonRPCUtils.createRequest` writes a request without an `id`.
+pub fn notify(shared: &Shared, short: &str, params: Option<Value>) {
+    let mut message = json!({ "jsonrpc": "2.0", "method": format!("minecraft:{short}") });
+    if let Some(params) = params {
+        message["params"] = json!([params]);
+    }
+    if let Some(outbox) = shared.management.outbox.lock().unwrap().as_ref() {
+        let _ = outbox.send(message.to_string());
+    }
+}
+
+/// `NotificationService.serverStarted`, after the game's `Done`: a new run of the game.
+pub fn started(shared: &Shared) {
+    shared
+        .management
+        .stopping_told
+        .store(false, Ordering::SeqCst);
+    notify(shared, "notification/server/started", None);
+}
+
+/// `NotificationService.serverShuttingDown`, which `DedicatedServer.stopServer` sends first of all.
+pub fn stopping(shared: &Shared) {
+    if !shared.management.stopping_told.swap(true, Ordering::SeqCst) {
+        notify(shared, "notification/server/stopping", None);
+    }
+}
+
 /// The management server's settings, read from server.properties as vanilla's `JsonRpc.create` reads them.
 struct Settings {
     host: String,
@@ -150,6 +209,26 @@ fn serve(
     settings: Arc<Settings>,
     listener: TcpListener,
 ) -> std::io::Result<()> {
+    let (outbox, queue) = mpsc::channel::<String>();
+    *shared.management.outbox.lock().unwrap() = Some(outbox);
+    let carrier = shared.clone();
+    std::thread::Builder::new()
+        .name("management-notify".into())
+        .spawn(move || {
+            for text in queue {
+                let sockets: Vec<_> = carrier
+                    .management
+                    .sockets
+                    .lock()
+                    .unwrap()
+                    .iter()
+                    .map(|(_, socket)| socket.clone())
+                    .collect();
+                for socket in sockets {
+                    let _ = socket.send_text(&text);
+                }
+            }
+        })?;
     std::thread::Builder::new()
         .name("management".into())
         .spawn(move || {
@@ -191,7 +270,15 @@ fn connection(shared: &Arc<Shared>, settings: &Settings, stream: &TcpStream) {
         return;
     };
     // `Connection`'s `ClientInfo`: the connections are numbered from 1, as the log lines show them.
-    let client = CONNECTIONS.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
+    let client = CONNECTIONS.fetch_add(1, Ordering::Relaxed) + 1;
+    let socket = Arc::new(socket);
+    shared
+        .management
+        .sockets
+        .lock()
+        .unwrap()
+        .push((client, socket.clone()));
+    let _registered = Registered(shared, client);
     while let Ok(Some(text)) = socket.read_text() {
         match receive(shared, client, &text) {
             Some(Ok(reply)) => {
@@ -645,12 +732,131 @@ mod tests {
     }
 
     fn listen(settings: Settings) -> SocketAddr {
+        listen_on(shared(), settings)
+    }
+
+    fn listen_on(shared: Arc<Shared>, settings: Settings) -> SocketAddr {
         let listener = TcpListener::bind("127.0.0.1:0").unwrap();
         let address = listener.local_addr().unwrap();
-        serve(shared(), Arc::new(settings), listener).unwrap();
+        serve(shared, Arc::new(settings), listener).unwrap();
         address
     }
 
+    /// The upgrade over a plain socket, without the `rpc.discover` that follows it in `discover`.
+    fn upgrade(address: SocketAddr) -> TcpStream {
+        let mut stream = TcpStream::connect(address).unwrap();
+        stream
+            .set_read_timeout(Some(Duration::from_secs(10)))
+            .unwrap();
+        write!(
+            stream,
+            "GET / HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
+             Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\
+             Authorization: Bearer {SECRET}\r\n\r\n"
+        )
+        .unwrap();
+        let mut head = Vec::new();
+        while !head.ends_with(b"\r\n\r\n") {
+            let mut b = [0u8];
+            stream.read_exact(&mut b).unwrap();
+            head.push(b[0]);
+        }
+        assert!(head.starts_with(b"HTTP/1.1 101 "));
+        stream
+    }
+
+    /// The text of the next frame the server sends (unmasked, short enough for a 7-bit length).
+    fn next_text(stream: &mut TcpStream) -> String {
+        let mut head = [0u8; 2];
+        stream.read_exact(&mut head).unwrap();
+        assert_eq!(head[0], 0x81);
+        let mut payload = vec![0u8; usize::from(head[1])];
+        stream.read_exact(&mut payload).unwrap();
+        String::from_utf8(payload).unwrap()
+    }
+
+    /// Waits until `count` connections are in the registry, which the upgrade's reply doesn't guarantee yet.
+    fn registered(shared: &Shared, count: usize) {
+        for _ in 0..500 {
+            if shared.management.sockets.lock().unwrap().len() == count {
+                return;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        panic!("{count} connections never registered");
+    }
+
+    #[test]
+    fn notifications_go_to_every_client_as_vanilla_writes_them() {
+        let shared = shared();
+        let address = listen_on(shared.clone(), settings());
+        let mut first = upgrade(address);
+        let mut second = upgrade(address);
+        registered(&shared, 2);
+        started(&shared);
+        let expected = r#"{"jsonrpc":"2.0","method":"minecraft:notification/server/started"}"#;
+        assert_eq!(next_text(&mut first), expected);
+        assert_eq!(next_text(&mut second), expected);
+        // One stop is told once per run of the game, whoever asks for it first; the next run says it again.
+        stopping(&shared);
+        stopping(&shared);
+        started(&shared);
+        stopping(&shared);
+        let stopping_frame =
+            r#"{"jsonrpc":"2.0","method":"minecraft:notification/server/stopping"}"#;
+        for client in [&mut first, &mut second] {
+            assert_eq!(next_text(client), stopping_frame);
+            assert_eq!(next_text(client), expected);
+            assert_eq!(next_text(client), stopping_frame);
+        }
+        // A parameter goes in an array of one, and a client that left is dropped from the registry.
+        drop(second);
+        notify(
+            &shared,
+            "notification/players/left",
+            Some(json!({ "name": "A" })),
+        );
+        assert_eq!(
+            next_text(&mut first),
+            r#"{"jsonrpc":"2.0","method":"minecraft:notification/players/left","params":[{"name":"A"}]}"#
+        );
+        for _ in 0..500 {
+            if shared.management.sockets.lock().unwrap().len() == 1 {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        assert_eq!(shared.management.sockets.lock().unwrap().len(), 1);
+    }
+
+    #[test]
+    fn every_notification_sent_is_one_discover_lists() {
+        let listed = schema::discover();
+        let names: Vec<&str> = listed["methods"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|m| m["name"].as_str().unwrap())
+            .collect();
+        for sent in [
+            "notification/server/started",
+            "notification/server/stopping",
+        ] {
+            assert!(
+                names.contains(&format!("minecraft:{sent}").as_str()),
+                "{sent}"
+            );
+        }
+    }
+
+    #[test]
+    fn without_a_management_server_a_notification_goes_nowhere() {
+        let shared = shared();
+        started(&shared);
+        stopping(&shared);
+        assert!(shared.management.outbox.lock().unwrap().is_none());
+    }
+
     #[test]
     fn a_websocket_client_discovers_the_methods() {
         let address = listen(settings());

crates/horizon-server/src/supervisor.rs

@@ -498,7 +498,11 @@ fn pump(shared: &Arc<Shared>, from: impl Read + Send + 'static, to_stderr: bool)
                         };
                         if up {
                             shared.emit("phase", serde_json::json!({ "phase": "running" }));
+                            crate::management::started(&shared);
                         }
+                    } else if line.contains("/INFO]: Stopping server") {
+                        // The game's own `stop` (a command in the game, not the supervisor's).
+                        crate::management::stopping(&shared);
                     }
                 }
                 shared.log_line(line);
@@ -648,6 +652,7 @@ fn supervise(
             if command == "stop" && stopping.is_none() {
                 stopping = Some(Instant::now());
                 shared.set_phase(Phase::Stopping);
+                crate::management::stopping(shared);
             }
             if let Some(input) = stdin.as_mut()
                 && writeln!(input, "{line}")
@@ -665,6 +670,7 @@ fn supervise(
             note(shared, "INFO", "Restarting the server");
             stopping = Some(Instant::now());
             shared.set_phase(Phase::Restarting);
+            crate::management::stopping(shared);
             send_stop(&mut stdin);
         }
         if asked {
@@ -673,6 +679,7 @@ fn supervise(
                 note(shared, "INFO", "Stopping the server");
                 stopping = Some(Instant::now());
                 shared.set_phase(Phase::Stopping);
+                crate::management::stopping(shared);
                 send_stop(&mut stdin);
             } else {
                 note(shared, "WARN", "Stopping the game at once, without saving");

docs/api/server.md

@@ -79,8 +79,9 @@ addition; a change to one starts `/api/v2`, served beside `/api/v1`.
 - **Backups**: restore a single dimension.
 - **Mods on a dedicated server**: list, turn on and off, and their settings, once native mods can be built for the
   server ([mods.md](../mods.md), [native-mods.md](../native-mods.md#multiplayer)).
-- **Vanilla's management protocol**: `serversettings/*`, `gamerules`, the notifications and the status
-  heartbeat. They answer an internal error saying so; a kick's translatable message is sent as its key.
+- **Vanilla's management protocol**: `serversettings/*`, `gamerules`, the notifications other than
+  `server/started` and `server/stopping`, and the status heartbeat. The methods answer an internal error saying
+  so; a kick's translatable message is sent as its key.
 - **Compatibility tests**: golden replies per endpoint, and a `server.toml` from each release line.
 
 ## Changelog

docs/server.md

@@ -506,7 +506,11 @@ runtime or new dependency. Server-sent events can replace the polling later if a
     `server/{status,save,stop,system_message}`, and the four lists (`allowlist/*`, `bans/*`, `ip_bans/*`,
     `operators/*`), whose changes the game applies itself (its control endpoint's `management` request, in
     `game/dedicated/management.rs`), so the source, expiry, log lines and kicks are vanilla's;
-  - planned: `serversettings/*` and `gamerules`, the notifications and heartbeat (from the event stream).
+  - notifications: the supervisor pushes them to every connected client as vanilla's `JsonRpcNotificationService`
+    does (a JSON-RPC request without an `id`, its one parameter in an array), through a queue and a thread so a slow
+    client never holds the game's output. Built: `server/started` (the game's `Done`) and `server/stopping`
+    (when the supervisor or the game's own `stop` begins the shutdown, once per run);
+  - planned: `serversettings/*` and `gamerules`, the other notifications and the heartbeat.
 
 ## 9. Benchmark plan (bots against vanilla and Paper)
 

tools/test_management_protocol_parity.py

@@ -20,8 +20,9 @@ sent the same things over TLS, and every answer is compared exactly:
   `server/stop`;
 - the TLS version each server agrees on.
 
-A parse error's `data` is Gson's message in vanilla and serde's in Horizon: left out. The notifications vanilla
-sends meanwhile (`server/started`, `server/saving`...) are set aside: Horizon doesn't send them yet.
+A parse error's `data` is Gson's message in vanilla and serde's in Horizon: left out. The notifications the
+servers send meanwhile (`server/stopping`...) are set aside, and their names are kept in each side's report
+(`notifications`); Horizon sends `server/started` and `server/stopping` so far, not yet the rest.
 """
 import base64
 import json