centralize call to topology_change

4 files changed, 23 insertions, 27 deletions

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(),