From 964f78a097a76b809954ddd4c1d5c7055d5a5772 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 05:45:57 +0100 Subject: centralize call to topology_change --- src/swarm/peer/admit.rs | 3 ++- src/swarm/peer/handshake.rs | 15 ++++++--------- src/swarm/peer/keepalive.rs | 24 ++++++++++-------------- src/swarm/peer/worker.rs | 8 +++++--- 4 files changed, 23 insertions(+), 27 deletions(-) (limited to 'src/swarm/peer') diff --git a/src/swarm/peer/admit.rs b/src/swarm/peer/admit.rs index 1800775..6e02150 100644 --- a/src/swarm/peer/admit.rs +++ b/src/swarm/peer/admit.rs @@ -4,6 +4,7 @@ use tracing::{debug, info, trace, warn}; use super::handshake::remember_peer; use crate::state::{AppState, HelloState, Standing}; +use crate::swarm::topology; use crate::transport::ParsedMessage; /// With a swarm key, only peers that proved it get past this point. A HELLO or @@ -50,7 +51,7 @@ pub(super) async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) if peer_state.standing.is_admitted() { warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership"); } - peer_state.standing = Standing::Stranger; + topology::change_peer(state, &mut peer_state, |p| p.standing = Standing::Stranger); peer_state.hello_state = HelloState::NotSent; peer_state.reset_sync(); } diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs index 225a7c9..13a1264 100644 --- a/src/swarm/peer/handshake.rs +++ b/src/swarm/peer/handshake.rs @@ -147,11 +147,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { let mut peer_state = peer_state_mutex.lock().await; if peer_state.hello_state == HelloState::Sending { - let was_unreachable = peer_state.unreachable_since.is_some(); - peer_state.hello_sent(Instant::now()); - if !was_unreachable && peer_state.unreachable_since.is_some() { - state.topology_changed.notify_one(); - } + topology::change_peer(state, &mut peer_state, |p| p.hello_sent(Instant::now())); let wait = peer_state.next_hello_wait(); peer_state.hello_state = HelloState::AwaitingReply { next_attempt: Instant::now() + wait, @@ -402,11 +398,12 @@ pub(super) async fn establish_peer( } if !matches!(peer_state.standing, Standing::Member { .. }) { - peer_state.standing = Standing::Member { - swarm_id: swarm_id.clone(), - }; // A member our list may lack yet: a neighbour then (`topology::refresh`). - state.topology_changed.notify_one(); + topology::change_peer(state, &mut peer_state, |p| { + p.standing = Standing::Member { + swarm_id: swarm_id.clone(), + } + }); } // Stop our retries and sync from a clean slate. A new handshake from the peer always diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 40b8ed3..5e03d0e 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -91,11 +91,9 @@ pub(super) async fn heard_from( let mut peer_state = peer_state.lock().await; also(&mut peer_state); let previously_seen = peer_state.last_seen; - if peer_state.unreachable_since.is_some() { - // Reachable again (`mark_seen` clears it). - state.topology_changed.notify_one(); - } - (peer_state.mark_seen(now), previously_seen) + // Reachable (again). + let reactivated = topology::change_peer(state, &mut peer_state, |p| p.mark_seen(now)); + (reactivated, previously_seen) }; crate::swarm::source::heard_from(state, peer); if reactivated { @@ -187,22 +185,20 @@ pub(super) async fn status_received(state: &AppState, peer: &str) { pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) { if let Some(peer_state) = state.get_peer_state(peer).await { let mut peer_state = peer_state.lock().await; - // Its farewell: dormant now, as if silent for long (a message from it, as its - // HELLO when it's back, makes it active again). - let left = remote.leaving && peer_state.say_farewell(Instant::now()); + let left = topology::change_peer(state, &mut peer_state, |p| { + // A neighbour that only connects out stays one as long as it says so. + p.syncs_with_us = Some(remote.syncs_with_you); + // Its farewell: dormant now, as if silent for long (a message from it, as + // its HELLO when it's back, makes it active again). + remote.leaving && p.say_farewell(Instant::now()) + }); let became_ready = remote.ready && peer_state.reports_ready != Some(true); peer_state.reports_ready = Some(remote.ready); - let changed = - peer_state.syncs_with_us.replace(remote.syncs_with_you) != Some(remote.syncs_with_you); peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); drop(peer_state); if left { info!(peer = %peer, "Peer said farewell; dormant until it is heard from again"); } - if changed || left { - // A neighbour that only connects out stays one as long as it says so. - state.topology_changed.notify_one(); - } // Off the consumer: telling which heads we lack takes the document's lock. { let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone()); diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs index 87d2b3c..797f091 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -51,8 +51,11 @@ pub(crate) async fn run_peer_worker(state: AppState) { let (became_dormant, keepalive, unanswered) = { let mut peer_state = peer_state_mutex.lock().await; - let became_dormant = - ready && neighbour && peer_state.check_dormancy(&state.timing, now); + let became_dormant = ready + && neighbour + && topology::change_peer(&state, &mut peer_state, |p| { + p.check_dormancy(&state.timing, now) + }); let keepalive = if neighbour { peer_state.take_keepalive_due(&state.timing, now) } else { @@ -72,7 +75,6 @@ pub(crate) async fn run_peer_worker(state: AppState) { } if became_dormant { - state.topology_changed.notify_one(); warn!( peer = %peer, silent_secs = state.timing.dormant_after.as_secs(), -- cgit v1.3.1