1 files changed, 13 insertions, 4 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 80d6d56..a05e346 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -141,10 +141,10 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
             tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> {
                 // Label the message with the epoch of the document it was generated
                 // from, read under the same lock.
-                let (epoch, message) = {
+                let (epoch, epoch_seq, message) = {
                     let mut replica = doc.blocking_write();
                     let mut peer_state = psm.blocking_lock();
-                    let epoch = replica.epoch;
+                    let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
                     let sent_before = peer_state.sync.state.sent_hashes.clone();
                     let message = {
                         let sync_doc = replica.doc.sync();
@@ -178,7 +178,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                             regenerate: false,
                         });
                     }
-                    (epoch, message)
+                    (epoch, epoch_seq, message)
                 };
                 let Some(message) = message else {
                     return Ok(None);
@@ -186,6 +186,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                 let envelope = transport::SyncEnvelope {
                     swarm_id,
                     epoch,
+                    epoch_seq,
                     message,
                     ready,
                 };
@@ -329,6 +330,7 @@ pub(crate) async fn handle_sync(
         let epoch = envelope.epoch;
         let message = envelope.message;
         let peer = peer.to_owned();
+        let clock = Arc::clone(&state.clock);
 
         // Some((caught_up, heads changed)) when applied, None when the message belongs
         // to another epoch.
@@ -365,7 +367,7 @@ pub(crate) async fn handle_sync(
             stats.heads_seen(heads_after.len());
             let changed = heads_after != heads_before;
             if changed && measure_propagation {
-                let now_ms = crate::state::unix_millis() as i64;
+                let now_ms = clock.now_millis() as i64;
                 // Time 0: not stamped (snapshots, older nodes).
                 let delays: Vec<i64> = replica
                     .doc
@@ -607,6 +609,13 @@ pub(crate) async fn run_change_publisher(state: AppState) {
             warn!(path = ?path, "Concurrent writes produced a conflict");
         }
 
+        if paths
+            .iter()
+            .any(|path| crate::swarm::topology::affects_neighbours(path))
+        {
+            state.topology_changed.notify_one();
+        }
+
         // Fails only when nobody is subscribed.
         let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
     }