add writer trees

1 files changed, 15 insertions, 6 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 37c3b0a..403e0c1 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -144,7 +144,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
             move || -> (u64, Result<Option<transport::EncodedMessage>>) {
                 // Label the message with the epoch of the document it was generated
                 // from, read under the same lock.
-                let (session, epoch, epoch_seq, body) = {
+                let (session, epoch, epoch_seq, body, writer) = {
                     let replica = doc.blocking_read();
                     let mut peer_state = psm.blocking_lock();
                     let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
@@ -171,7 +171,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                         );
                         transport::SyncBody::Delta(spec)
                     });
-                    (peer_state.sync.id(), epoch, epoch_seq, body)
+                    // Whose change it passes on, if any: this message carries it.
+                    let writer = peer_state.tree_writer.take();
+                    (peer_state.sync.id(), epoch, epoch_seq, body, writer)
                 };
                 let Some(body) = body else {
                     return (session, Ok(None));
@@ -182,6 +184,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     epoch_seq,
                     body,
                     ready,
+                    writer,
                 };
                 let encoded = transport
                     .encode(&peer_owned, &ParsedMessage::Sync(envelope))
@@ -341,6 +344,7 @@ pub(crate) async fn handle_sync(
 
     let epoch_label = envelope.epoch;
     let from_ready = envelope.ready;
+    let writer = envelope.writer;
     let payload = match envelope.body {
         transport::SyncBody::Message(payload) => payload,
         transport::SyncBody::Delta(spec) => {
@@ -462,7 +466,7 @@ pub(crate) async fn handle_sync(
     }
     // Caught up: our other neighbours may lack what the source had.
     if changed || just_caught_up {
-        crate::swarm::relay::announce(state, peer).await;
+        crate::swarm::relay::announce(state, peer, writer.filter(|_| changed)).await;
     }
 
     state
@@ -490,7 +494,7 @@ fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: tran
             Ok(changed) => {
                 state.notify_store_changed();
                 if changed {
-                    crate::swarm::relay::announce(&state, &peer).await;
+                    crate::swarm::relay::announce(&state, &peer, None).await;
                 }
             }
             Err(error) => {
@@ -598,12 +602,17 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
                 peers_guard.keys().cloned().collect()
             };
 
-            // Our own changes go to our neighbours; they pass them on. Others only get
-            // answers to what they send us.
+            // Our own changes go to our neighbours, saying they're ours: they pass them
+            // on along our tree (`swarm::tree`). Others only get answers to what they
+            // send us.
+            let own = crate::swarm::topology::pair_key(state.local_id());
             for peer in peer_keys {
                 if peer_swarm(&state, &peer).await.is_some()
                     && crate::swarm::topology::is_neighbour(&state, &peer)
                 {
+                    if let Some(peer_state) = state.get_peer_state(&peer).await {
+                        peer_state.lock().await.tree_writer = Some(own);
+                    }
                     dirty_peers.insert(peer);
                 }
             }