From dd0e4a5120751d8dcb4b42834881f36de17ec168 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 05:45:29 +0100 Subject: start API earlier --- src/main.rs | 50 +++++++++++++++++++++++++++++++------------------- 1 file changed, 31 insertions(+), 19 deletions(-) (limited to 'src/main.rs') diff --git a/src/main.rs b/src/main.rs index 3bdf0ed..8a61200 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,6 +5,7 @@ use std::future::IntoFuture; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use anyhow::{Context, Result}; use automerge::AutoCommit; @@ -314,24 +315,9 @@ async fn main() -> Result<()> { node.spawn_workers(); - let joining = bootstrap::bootstrap(&state, &seeds, start); - tokio::select! { - joined = joining => joined.context("Joining swarm")?, - _ = stopped(stop.clone()) => { - info!("Stopped while joining the swarm"); - return Ok(()); - } - } - - node::spawn_ready_workers(&state); - - info!( - swarm_id = ?state.swarm_id().await, - ready = state.is_ready().await, - "Bootstrap completed" - ); - swarm::state_file::update(&state, true).await; - + // 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); let listener = tokio::net::TcpListener::bind(args.bind_addr) @@ -351,7 +337,30 @@ async fn main() -> Result<()> { let stopping = state.clone(); tokio::pin!(server_fut); + // Joins, then runs until stopped; returns only if joining fails. + let joining = async { + bootstrap::bootstrap(&state, &seeds, start) + .await + .context("Joining swarm")?; + node::spawn_ready_workers(&state); + info!( + swarm_id = ?state.swarm_id().await, + ready = state.is_ready().await, + "Bootstrap completed" + ); + swarm::state_file::update(&state, true).await; + joined.store(true, Ordering::Release); + std::future::pending::>().await + }; + tokio::pin!(joining); + tokio::select! { + failed = joining.as_mut() => { + let _ = shutdown_tx.send(()); + 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(()); @@ -364,7 +373,6 @@ async fn main() -> Result<()> { } else { info!("Axum HTTP engine shutdown cleanly."); } - std::mem::drop(state); } _ = stopped(stop.clone()) => { @@ -374,6 +382,10 @@ async fn main() -> Result<()> { } info!("HTTP API server stopped"); + if !joined.load(Ordering::Acquire) { + info!("Stopped while joining the swarm"); + return Ok(()); + } swarm::peer::farewell(&stopping).await; swarm::state_file::update(&stopping, true).await; document::file::flush(&stopping).await; -- cgit v1.3.1