dasboard cannot block shutdown

4 files changed, 117 insertions, 49 deletions

diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index a8b697e..9a17d6b 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -27,6 +27,7 @@
   #connection { margin-left: auto; color: var(--muted); }
   #connection.live { color: var(--ok); }
   #connection.down { color: var(--bad); }
+  body.stale main { opacity: .55; }
   /* Swarm on the left, the full height; peers (as tall as their rows, up to 8) above
      the figures (the rest) on the right. */
   main { flex: 1; min-height: 0; display: grid; gap: 16px; padding: 16px;
@@ -428,23 +429,45 @@ document.getElementById("graph").addEventListener("click", () => focus(null));
 
 // --- Events ---
 const connection = document.getElementById("connection");
-const events = new EventSource("dashboard/events");
-events.addEventListener("open", () => {
-  connection.textContent = "live"; connection.className = "live";
-});
-events.addEventListener("error", () => {
-  // EventSource reconnects by itself; the node sends a snapshot first again.
-  connection.textContent = "reconnecting…"; connection.className = "down";
-});
-events.addEventListener("snapshot", event => {
-  state = JSON.parse(event.data); stateAt = Date.now(); render();
-});
-events.addEventListener("patch", event => {
-  if (!state) return;
-  state = mergePatch(state, JSON.parse(event.data)); stateAt = Date.now(); render();
-});
+function status(text, live) {
+  connection.textContent = text;
+  connection.className = live ? "live" : "down";
+  document.body.classList.toggle("stale", !live && state !== null);
+}
+// While not live, what is shown stays as it was last (ages don't count on).
+let live = false;
+let events = null;
+function connect() {
+  events = new EventSource("dashboard/events");
+  events.addEventListener("snapshot", event => {
+    state = JSON.parse(event.data); stateAt = Date.now(); live = true;
+    status("live", true); render();
+  });
+  events.addEventListener("patch", event => {
+    if (!state) return;
+    state = mergePatch(state, JSON.parse(event.data)); stateAt = Date.now(); render();
+  });
+  // The node stops: it says so, and ends the stream. Back once it is.
+  events.addEventListener("offline", () => {
+    events.close(); live = false;
+    status("node stopped; reconnecting…", false);
+    setTimeout(connect, 3000);
+  });
+  events.addEventListener("error", () => {
+    live = false;
+    if (events.readyState === EventSource.CLOSED) {
+      // Given up (an HTTP error, e.g. the node is up but not serving the page yet).
+      status("disconnected; retrying…", false);
+      setTimeout(connect, 5000);
+    } else {
+      // EventSource reconnects by itself; the node sends a snapshot first again.
+      status("reconnecting…", false);
+    }
+  });
+}
+connect();
 // Ages count on between events; nothing else changes without one.
-setInterval(render, 1000);
+setInterval(() => { if (live) render(); }, 1000);
 window.addEventListener("resize", () => { lastShape = ""; render(); });
 </script>
 </body>
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index 6c7f1c5..92271ef 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -18,7 +18,7 @@ use std::convert::Infallible;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
-use axum::http::{HeaderValue, header::CONTENT_TYPE};
+use axum::http::{HeaderValue, StatusCode, header::CONTENT_TYPE};
 use axum::response::sse::{Event, KeepAlive, Sse};
 use axum::response::{Html, IntoResponse, Response};
 use axum::routing::get;
@@ -57,6 +57,9 @@ struct Hub {
     /// Each patch, to the clients connected.
     patches: broadcast::Sender<Arc<Published>>,
     control: Mutex<Control>,
+    /// True once the node stops: every client gets an `offline` event, and its stream
+    /// ends (so the HTTP server needn't wait for clients to go).
+    closing: watch::Receiver<bool>,
 }
 
 struct Control {
@@ -66,7 +69,7 @@ struct Control {
 }
 
 impl Hub {
-    fn new(state: AppState) -> Arc<Self> {
+    fn new(state: AppState, closing: watch::Receiver<bool>) -> Arc<Self> {
         let (latest, _) = watch::channel(None);
         let (patches, _) = broadcast::channel(64);
         Arc::new(Self {
@@ -77,6 +80,7 @@ impl Hub {
                 running: false,
                 wanted_until: Instant::now(),
             }),
+            closing,
         })
     }
 
@@ -84,7 +88,7 @@ impl Hub {
     fn want(self: &Arc<Self>) {
         let mut control = self.control.lock_or_recover();
         control.wanted_until = Instant::now() + LINGER;
-        if !control.running {
+        if !control.running && !*self.closing.borrow() {
             control.running = true;
             // What it published before it stopped is stale.
             self.latest.send_replace(None);
@@ -92,24 +96,25 @@ impl Hub {
         }
     }
 
-    /// The latest snapshot, once there is one.
-    async fn current(&self) -> Arc<Published> {
+    /// The latest snapshot, once there is one; None if the node stops first.
+    async fn current(&self) -> Option<Arc<Published>> {
         let mut latest = self.latest.subscribe();
-        loop {
-            // The borrow (a lock guard, not Send) ends here, before any later await.
-            let published = latest
-                .wait_for(Option::is_some)
-                .await
-                .ok()
-                .and_then(|current| current.clone());
-            if let Some(published) = published {
-                return published;
+        let mut closing = self.closing.clone();
+        // The borrows (lock guards, not Send) end within the arms.
+        tokio::select! {
+            published = latest.wait_for(Option::is_some) => {
+                published.ok().and_then(|current| current.clone())
             }
-            // The sender lives as long as the hub: not reached.
-            tokio::time::sleep(TICK).await;
+            _ = closing.wait_for(|closing| *closing) => None,
         }
     }
 
+    /// Resolves once the node stops (or its signal is gone).
+    async fn closed(&self) {
+        let mut closing = self.closing.clone();
+        let _ = closing.wait_for(|closing| *closing).await;
+    }
+
     async fn run(self: Arc<Self>) {
         let mut last: Option<JsonValue> = None;
         let mut seq = 0u64;
@@ -143,7 +148,9 @@ impl Hub {
             tokio::time::sleep(TICK).await;
             {
                 let mut control = self.control.lock_or_recover();
-                if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until {
+                let unwanted =
+                    self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until;
+                if unwanted || *self.closing.borrow() {
                     control.running = false;
                     return;
                 }
@@ -301,12 +308,13 @@ fn merge_patch(old: &JsonValue, new: &JsonValue) -> Option<JsonValue> {
 }
 
 /// The dashboard's routes (authenticated, as the rest of `/admin`), with their hub.
-pub(crate) fn router(state: AppState) -> Router<AppState> {
+/// `closing` turns true when the node stops (`Hub::closing`).
+pub(crate) fn router(state: AppState, closing: watch::Receiver<bool>) -> Router<AppState> {
     Router::new()
         .route("/admin/dashboard", get(page))
         .route("/admin/dashboard/state", get(current_state))
         .route("/admin/dashboard/events", get(events))
-        .layer(Extension(Hub::new(state)))
+        .layer(Extension(Hub::new(state, closing)))
 }
 
 /// `GET /admin/dashboard`: the page.
@@ -317,7 +325,9 @@ async fn page() -> Html<&'static str> {
 /// `GET /admin/dashboard/state`: the latest snapshot, for a client that wants it whole.
 async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response {
     hub.want();
-    let current = hub.current().await;
+    let Some(current) = hub.current().await else {
+        return StatusCode::SERVICE_UNAVAILABLE.into_response();
+    };
     (
         [(CONTENT_TYPE, HeaderValue::from_static("application/json"))],
         current.json.clone(),
@@ -327,17 +337,30 @@ async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response {
 
 /// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a
 /// `patch` whenever something changed (each with its number as the event's ID; a
-/// client that fell behind gets a `snapshot` again).
+/// client that fell behind gets a `snapshot` again), and `offline` when the node
+/// stops, which ends the stream.
 async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
     // Subscribed before the snapshot is read: no patch after it is missed.
     let mut patches = hub.patches.subscribe();
     hub.want();
     let first = hub.current().await;
     let stream = async_stream::stream! {
+        let Some(first) = first else {
+            yield Ok::<_, Infallible>(offline_event());
+            return;
+        };
         let mut seen = first.seq;
-        yield Ok::<_, Infallible>(snapshot_event(&first));
+        yield Ok(snapshot_event(&first));
         loop {
-            match patches.recv().await {
+            let received = tokio::select! {
+                received = patches.recv() => Some(received),
+                () = hub.closed() => None,
+            };
+            let Some(received) = received else {
+                yield Ok(offline_event());
+                break;
+            };
+            match received {
                 Ok(patch) if patch.seq <= seen => {}
                 Ok(patch) => {
                     seen = patch.seq;
@@ -347,7 +370,10 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
                         .data(patch.json.clone()));
                 }
                 Err(broadcast::error::RecvError::Lagged(_)) => {
-                    let current = hub.current().await;
+                    let Some(current) = hub.current().await else {
+                        yield Ok(offline_event());
+                        break;
+                    };
                     seen = current.seq;
                     yield Ok(snapshot_event(&current));
                 }
@@ -358,6 +384,11 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
     Sse::new(stream).keep_alive(KeepAlive::default())
 }
 
+/// The node stops: the client may reconnect once it is back.
+fn offline_event() -> Event {
+    Event::default().event("offline").data("{}")
+}
+
 fn snapshot_event(published: &Published) -> Event {
     Event::default()
         .event("snapshot")
diff --git a/src/api/mod.rs b/src/api/mod.rs
index fd05669..3f60cd6 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -35,7 +35,14 @@ use crate::state::AppState;
 
 /// Builds the complete HTTP router. Everything except the API description (`/swagger-ui`,
 /// `/openapi.yaml`), `/health` and `/ready` is authenticated.
-pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> Router {
+/// `closing` turns true when the node stops: long-lived responses (the dashboard's
+/// events) end then.
+pub(crate) fn router(
+    state: AppState,
+    auth_state: AuthState,
+    debug_api: bool,
+    closing: tokio::sync::watch::Receiver<bool>,
+) -> Router {
     let mut api: Router<AppState> = Router::new()
         .route("/store", get(store::get_store_root))
         .route(
@@ -59,8 +66,10 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) ->
 
     #[cfg(feature = "dashboard")]
     {
-        api = api.merge(dashboard::router(state.clone()));
+        api = api.merge(dashboard::router(state.clone(), closing));
     }
+    #[cfg(not(feature = "dashboard"))]
+    let _ = closing;
 
     if debug_api {
         warn!("Debug API enabled");
diff --git a/src/main.rs b/src/main.rs
index da79668..533f85b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -243,7 +243,14 @@ async fn main() -> Result<()> {
     // The API is up while the node joins: what it serves then is the handlers' business
     // (`docs/lifecycle.md`), and /peers and /admin/stats show how joining goes.
     let joined = AtomicBool::new(false);
-    let app = api::router(state.clone(), auth_state, args.debug_api);
+    // True once the node stops: the server drains, and long-lived responses end.
+    let (closing_tx, closing_rx) = watch::channel(false);
+    let app = api::router(
+        state.clone(),
+        auth_state,
+        args.debug_api,
+        closing_rx.clone(),
+    );
 
     let listener = tokio::net::TcpListener::bind(args.bind_addr)
         .await
@@ -251,10 +258,8 @@ async fn main() -> Result<()> {
 
     info!(address = %listener.local_addr()?, "HTTP API listening");
 
-    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
-
     let server_builder = axum::serve(listener, app).with_graceful_shutdown(async move {
-        let _ = shutdown_rx.await;
+        stopped(closing_rx).await;
         info!("Shutdown signal received. Draining API connections...")
     });
 
@@ -281,14 +286,14 @@ async fn main() -> Result<()> {
 
     tokio::select! {
         failed = joining.as_mut() => {
-            let _ = shutdown_tx.send(());
+            closing_tx.send_replace(true);
             let _ = server_fut.as_mut().await;
             return failed;
         }
 
         exit_status = state.transport.wait_for_exit() => {
             error!(status = ?exit_status, "A transport stopped");
-            let _ = shutdown_tx.send(());
+            closing_tx.send_replace(true);
             let _ = server_fut.as_mut().await;
         }
 
@@ -301,7 +306,7 @@ async fn main() -> Result<()> {
         }
 
         _ = stopped(stop.clone()) => {
-            let _ = shutdown_tx.send(());
+            closing_tx.send_replace(true);
             let _ = server_fut.as_mut().await;
         }
     }