unify closing logic

6 files changed, 25 insertions, 31 deletions

diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index f0cbbfd..0a502a3 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -381,8 +381,9 @@ fn merge_patch(old: &JsonValue, new: &JsonValue) -> Option<JsonValue> {
 }
 
 /// The dashboard's routes (authenticated, as the rest of `/admin`), with their hub.
-/// `closing` turns true when the node stops (`Hub::closing`).
-pub(crate) fn router(state: AppState, closing: watch::Receiver<bool>) -> Router<AppState> {
+pub(crate) fn router(state: AppState) -> Router<AppState> {
+    // Its streams end when the node shuts down (`AppState::closing`).
+    let closing = state.closing.subscribe();
     Router::new()
         .route("/admin/dashboard", get(page_html))
         .route("/admin/dashboard/state", get(current_state))
diff --git a/src/api/mod.rs b/src/api/mod.rs
index 1f178e7..58967f3 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -35,14 +35,7 @@ use crate::state::AppState;
 
 /// Builds the complete HTTP router. Everything except the API description (`/swagger-ui`,
 /// `/openapi.yaml`), `/health` and `/ready` is authenticated.
-/// `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 {
+pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> Router {
     let mut api: Router<AppState> = Router::new()
         .route("/store", get(store::get_store_root))
         .route(
@@ -67,10 +60,8 @@ pub(crate) fn router(
 
     #[cfg(feature = "dashboard")]
     {
-        api = api.merge(dashboard::router(state.clone(), closing));
+        api = api.merge(dashboard::router(state.clone()));
     }
-    #[cfg(not(feature = "dashboard"))]
-    let _ = closing;
 
     if debug_api {
         warn!("Debug API enabled");
diff --git a/src/main.rs b/src/main.rs
index 533f85b..617ca94 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -243,14 +243,7 @@ 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);
-    // 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 app = api::router(state.clone(), auth_state, args.debug_api);
 
     let listener = tokio::net::TcpListener::bind(args.bind_addr)
         .await
@@ -258,8 +251,10 @@ async fn main() -> Result<()> {
 
     info!(address = %listener.local_addr()?, "HTTP API listening");
 
+    // The node shutting down (`AppState::closing`): the server drains.
+    let closing = state.closing.subscribe();
     let server_builder = axum::serve(listener, app).with_graceful_shutdown(async move {
-        stopped(closing_rx).await;
+        stopped(closing).await;
         info!("Shutdown signal received. Draining API connections...")
     });
 
@@ -286,14 +281,14 @@ async fn main() -> Result<()> {
 
     tokio::select! {
         failed = joining.as_mut() => {
-            closing_tx.send_replace(true);
+            state.closing.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");
-            closing_tx.send_replace(true);
+            state.closing.send_replace(true);
             let _ = server_fut.as_mut().await;
         }
 
@@ -306,7 +301,7 @@ async fn main() -> Result<()> {
         }
 
         _ = stopped(stop.clone()) => {
-            closing_tx.send_replace(true);
+            state.closing.send_replace(true);
             let _ = server_fut.as_mut().await;
         }
     }
diff --git a/src/node.rs b/src/node.rs
index 0b8f663..8adab2b 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -103,7 +103,7 @@ impl Node {
             sync_events: sync_tx,
             change_tx,
             initial_sync_done: Arc::new(AtomicBool::new(false)),
-            stopping: Arc::new(AtomicBool::new(false)),
+            closing: Arc::new(watch::channel(false).0),
             change_events,
             timing: config.timing,
             remake: Arc::new(remake::RemakeState::new(config.clock.now_millis())),
diff --git a/src/state/mod.rs b/src/state/mod.rs
index d7c80c2..2f3d433 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -84,8 +84,10 @@ pub(crate) struct AppState {
     pub(crate) change_tx: watch::Sender<u64>,
 
     pub(crate) initial_sync_done: Arc<AtomicBool>,
-    /// The node is shutting down: its keepalives say it isn't ready (`peer::farewell`).
-    pub(crate) stopping: Arc<AtomicBool>,
+    /// True once the node shuts down (a signal, a failed join, a transport gone): the
+    /// HTTP API drains and its long-lived responses end, and keepalives say the node
+    /// isn't ready and is leaving (`peer::farewell`).
+    pub(crate) closing: Arc<watch::Sender<bool>>,
 
     pub(crate) change_events: broadcast::Sender<Arc<ChangeSet>>,
 
@@ -177,6 +179,11 @@ pub(crate) struct CatchUpSource {
 }
 
 impl AppState {
+    /// Whether the node shuts down (`closing`).
+    pub(crate) fn is_closing(&self) -> bool {
+        *self.closing.borrow()
+    }
+
     pub(crate) fn local_id(&self) -> &str {
         self.transport.local_id()
     }
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 4b529f5..9037fdd 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -120,7 +120,7 @@ pub(crate) fn announce(state: &AppState) {
 /// catching up from us picks another source now, instead of once we go dormant. Bounded
 /// in time: our transport may be gone already.
 pub(crate) async fn farewell(state: &AppState) {
-    state.stopping.store(true, Ordering::Release);
+    state.closing.send_replace(true);
     info!("Saying farewell to our neighbours");
     status_to_neighbours(state, Some(FAREWELL_TIMEOUT)).await;
 }
@@ -145,9 +145,9 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
         heads,
         epoch,
         epoch_seq,
-        ready: state.is_ready().await && !state.stopping.load(Ordering::Acquire),
+        ready: state.is_ready().await && !state.is_closing(),
         syncs_with_you,
-        leaving: state.stopping.load(Ordering::Acquire),
+        leaving: state.is_closing(),
     }
 }