summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/main.rs1
-rw-r--r--src/node.rs1
-rw-r--r--src/state/mod.rs2
-rw-r--r--src/swarm/peer/keepalive.rs22
-rw-r--r--src/swarm/peer/mod.rs2
5 files changed, 26 insertions, 2 deletions
diff --git a/src/main.rs b/src/main.rs
index dee0a32..628a71e 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -267,6 +267,7 @@ async fn main() -> Result<()> {
}
info!("HTTP API server stopped");
+ swarm::peer::farewell(&stopping).await;
swarm::state_file::update(&stopping, true).await;
document::file::flush(&stopping).await;
Ok(())
diff --git a/src/node.rs b/src/node.rs
index ed0832c..49e4194 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -92,6 +92,7 @@ impl Node {
sync_events: sync_tx,
change_tx,
initial_sync_done: Arc::new(AtomicBool::new(false)),
+ stopping: Arc::new(AtomicBool::new(false)),
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 04bdb3c..beac886 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -73,6 +73,8 @@ 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>,
pub(crate) change_events: broadcast::Sender<Arc<ChangeSet>>,
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index a593ac9..b22d954 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -2,6 +2,7 @@
//! documents differ while sync is idle.
use std::sync::atomic::Ordering;
+use std::time::Duration;
use anyhow::Result;
use tokio::time::Instant;
@@ -23,6 +24,25 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
.await
}
+/// How long shutting down waits for the last PINGs (`farewell`).
+const FAREWELL_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// At shutdown: one last PING to each neighbour, saying we aren't ready. A neighbour
+/// catching up from us picks another source now, instead of once we go dormant. Bounded
+/// in time: our Mycelium may be gone already.
+pub(crate) async fn farewell(state: &AppState) {
+ state.stopping.store(true, Ordering::Release);
+ let mut pings = tokio::task::JoinSet::new();
+ for (peer, _) in state.peers_snapshot().await {
+ if topology::is_neighbour(state, &peer) {
+ let state = state.clone();
+ pings.spawn(async move { send_ping(&state, &peer).await });
+ }
+ }
+ let all_sent = async { while pings.join_next().await.is_some() {} };
+ let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).await;
+}
+
/// Our keepalive for `peer`.
pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
// Heads and epoch are published together (`Replica::stamps`), so they always match,
@@ -48,7 +68,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
heads,
epoch,
epoch_seq,
- ready: state.is_ready().await,
+ ready: state.is_ready().await && !state.stopping.load(Ordering::Acquire),
syncs_with_you,
}
}
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index 2e5c1dc..0249f3b 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -24,6 +24,6 @@ pub(crate) use handshake::{
apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once,
};
pub(crate) use inbound::handle_message;
-pub(crate) use keepalive::send_ping;
+pub(crate) use keepalive::{farewell, send_ping};
pub(crate) use lanes::SyncLanes;
pub(crate) use worker::run_peer_worker;