From 0bb3924836205a0364aaf0959c7cf387db004794 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 03:54:19 +0100 Subject: unify closing logic --- src/api/dashboard/mod.rs | 5 +++-- src/api/mod.rs | 13 ++----------- src/main.rs | 19 +++++++------------ src/node.rs | 2 +- src/state/mod.rs | 11 +++++++++-- src/swarm/peer/keepalive.rs | 6 +++--- 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 { } /// 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) -> Router { +pub(crate) fn router(state: AppState) -> Router { + // 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, -) -> Router { +pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> Router { let mut api: Router = 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, pub(crate) initial_sync_done: Arc, - /// The node is shutting down: its keepalives say it isn't ready (`peer::farewell`). - pub(crate) stopping: Arc, + /// 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>, pub(crate) change_events: broadcast::Sender>, @@ -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(), } } -- cgit v1.3.1