1 files changed, 12 insertions, 4 deletions
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 7385ae2..e8248bf 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -299,7 +299,11 @@ 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();
+            }
             let wait = match peer_state.liveness {
                 Liveness::Dormant { .. } => state.timing.dormant_probe,
                 Liveness::Active => peer_state.hello_backoff.next(),
@@ -413,7 +417,7 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S
 /// neighbours get no more HELLOs, keepalives or sync openings from us; their session is
 /// reset, so we don't keep pushing changes to them either. They are still answered if
 /// they contact us.
-async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) {
+pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) {
     for peer in &change.added {
         remember_peer(state, peer).await;
         if let Err(error) = send_hello_once(state, peer).await {
@@ -492,6 +496,7 @@ 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(),
@@ -529,9 +534,8 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                     .unwrap_or_else(|poisoned| poisoned.into_inner())
                     .retain(|peer| known.contains_key(peer));
             }
-            if let Some(change) = topology::refresh(&state).await {
-                apply_neighbour_change(&state, change).await;
-            }
+            // Fallback for what doesn't notify (`topology::run_worker`).
+            state.topology_changed.notify_one();
             last_forget_check = Some(Instant::now());
         }
     }
@@ -802,6 +806,10 @@ pub(crate) async fn handle_message(
         let peer_state_mutex = state.get_or_create_peer_state(&peer).await;
         let mut peer_state = peer_state_mutex.lock().await;
         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(Instant::now()), previously_seen)
     };
     crate::swarm::source::heard_from(state, &peer);