unify heard_from logic

1 files changed, 28 insertions, 6 deletions

diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 9037fdd..0e80980 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -16,7 +16,7 @@ use tokio::time::Instant;
 use tracing::{debug, info, warn};
 
 use super::handshake::ensure_handshake;
-use crate::state::AppState;
+use crate::state::{AppState, PeerState};
 use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 use crate::transport::{self, ParsedMessage};
@@ -64,23 +64,45 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes
     let Some(peer_state) = state.get_peer_state(peer).await else {
         return;
     };
-    let reactivated = {
-        let mut peer_state = peer_state.lock().await;
-        if matches!(message, ParsedMessage::Ping(_)) {
+    let ping = matches!(message, ParsedMessage::Ping(_));
+    let (reactivated, _) = heard_from(state, peer, &peer_state, now, |peer_state| {
+        if ping {
             // Its keepalive (or a path probe): the next is ours.
             peer_state.keepalive_received_at = Some(now);
         }
+    })
+    .await;
+    if reactivated {
+        ensure_handshake(state, peer).await;
+    }
+}
+
+/// A message from `peer` came in: it is seen (`PeerState::mark_seen`: active, and
+/// reachable, again), and the catch-up source hears of it. `also` runs on its state
+/// under the same lock. Returns whether it was dormant (the caller re-handshakes), and
+/// when it was last seen before.
+pub(super) async fn heard_from(
+    state: &AppState,
+    peer: &str,
+    peer_state: &tokio::sync::Mutex<PeerState>,
+    now: Instant,
+    also: impl FnOnce(&mut PeerState),
+) -> (bool, Option<Instant>) {
+    let (reactivated, previously_seen) = {
+        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)
+        (peer_state.mark_seen(now), previously_seen)
     };
     crate::swarm::source::heard_from(state, peer);
     if reactivated {
         info!(peer = %peer, "Dormant peer is active again; re-handshaking");
-        ensure_handshake(state, peer).await;
     }
+    (reactivated, previously_seen)
 }
 
 /// How long shutting down waits for the last STATUSes (`farewell`).