announcements answered only if heads differ

14 files changed, 50 insertions, 12 deletions

diff --git a/docs/architecture.md b/docs/architecture.md
index cc738c1..1a320fc 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -232,7 +232,7 @@ the member list makes its neighbour.
 replies, admit (swarm key), mark the peer seen, record its epoch (ignoring late ones),
 then by type: HELLO answered, HELLO_REPLY establishes membership and discovers peers,
 sync queued on its lane if on our epoch (held back from everyone but the source while catching
-up), STATUS answered with a STATUS_REPLY and compared for divergence, blobs served or collected,
+up), STATUS answered with a STATUS_REPLY (an announcement only if the heads differ) and compared for divergence, blobs served or collected,
 LINKS answered with the node's links if the asker is a member, and again on each change while it subscribes, or with other members' links it asks for the asker, which can't reach them (`swarm::links`, for the dashboard);
 PING and PONG (the paths layer answered the PING already) only mark the peer seen, and
 after a PONG the peer's round-trip time is its path's again.
diff --git a/docs/peers.md b/docs/peers.md
index 9937c63..9868c0e 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -280,8 +280,9 @@ backbone (with 20 stable of 250 members and k = 12: about 65 links per stable me
 
 Changes travel through a partial mesh by announcement and pull ([`swarm/relay.rs`](../src/swarm/relay.rs)): a node
 pushes its own writes to its neighbours; after applying changes from one neighbour it
-sends its other neighbours a STATUS with its new heads, and a neighbour that lacks those
-heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks
+sends its other neighbours a STATUS with its new heads (marked as an announcement: no
+keepalive, and answered only by a neighbour whose heads differ, so one that has the
+changes already sends nothing back), and a neighbour that lacks those heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks
 for what's missing). It pulls from one neighbour at a time (15 s timeout); neighbours
 that announce heads it lacks meanwhile wait their turn and are pulled from as soon as
 the current pull is answered, if they still have something it lacks. The regular
diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index 7dbb722..04dbed3 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -110,6 +110,7 @@ fn status(node: &AppState) -> ParsedMessage {
         ready: false,
         syncs_with_you: false,
         leaving: false,
+        announcing: false,
     })
 }
 
diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index daa588f..7879ac7 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -335,6 +335,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
             ready: true,
             syncs_with_you: false,
             leaving: false,
+            announcing: false,
         })
     };
     let a = &a;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 18d29d5..1454c3e 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -143,8 +143,15 @@ pub(crate) async fn handle_message(
             crate::swarm::blob::handle(state, &peer, blob).await;
         }
         transport::ParsedMessage::Status(remote) => {
-            status_received(state, &peer).await;
+            // An announcement is no keepalive; it is answered only if our heads differ,
+            // so the announcer learns what we have that it lacks.
+            if !remote.announcing {
+                status_received(state, &peer).await;
+            }
             let local = on_status(state, &peer, &remote, reactivated).await;
+            if remote.announcing && local.heads == remote.heads {
+                return Ok(());
+            }
             if let Err(error) = state
                 .transport
                 .send_message(&peer, &ParsedMessage::StatusReply(local))
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 8597dcd..196b8b6 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -32,6 +32,20 @@ pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
         .await
 }
 
+/// Sends `peer` a STATUS announcing our new heads (`swarm::relay`): no keepalive, so it
+/// leaves the keepalives' schedule as it is, and it is answered only if the peer's heads
+/// differ (`Status::announcing`).
+pub(crate) async fn announce_heads(state: &AppState, peer: &str) -> Result<()> {
+    let status = transport::Status {
+        announcing: true,
+        ..local_status(state, peer).await
+    };
+    state
+        .transport
+        .send_message(peer, &ParsedMessage::Status(status))
+        .await
+}
+
 /// Sends `peer` a PING (the paths layer measures the path with it).
 pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
     let nonce = crate::transport::paths::random_nonce();
@@ -173,6 +187,7 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta
         ready: state.is_ready().await && !state.is_closing(),
         syncs_with_you,
         leaving: state.is_closing(),
+        announcing: false,
     }
 }
 
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index cf427a2..1826ec6 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -26,6 +26,6 @@ pub(crate) use handshake::{
     send_hello_once,
 };
 pub(crate) use inbound::handle_message;
-pub(crate) use keepalive::{announce, farewell, send_status};
+pub(crate) use keepalive::{announce, announce_heads, farewell, send_status};
 pub(crate) use lanes::SyncLanes;
 pub(crate) use worker::{Due, run_peer_worker};
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 1393d50..a4a4034 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -2,8 +2,9 @@
 //!
 //! A node pushes its own changes to its neighbours, but nothing it receives goes any
 //! further by itself. So after applying changes from one neighbour, a node *announces*
-//! its new heads to its other neighbours with a STATUS (a keepalive: heads, about 100
-//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the
+//! its new heads to its other neighbours with a STATUS (heads, about 100 bytes, not the
+//! changes; no keepalive, and answered only by a neighbour whose heads differ). A
+//! neighbour that sees heads it lacks *pulls*: it enters the
 //! announced heads as the peer's heads in its sync state, Automerge asks for what's
 //! missing (`need`), and the announcer answers with exactly those changes. Each node
 //! receives a change about once; the rest is announcements. Regular keepalives work the
@@ -11,7 +12,7 @@
 //!
 //! Announcements are coalesced: the first applied change schedules one announcement
 //! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
-//! round costs one STATUS and one STATUS_REPLY per neighbour (plus the pulls), whatever the
+//! round costs one STATUS per neighbour (plus the pulls and their replies), whatever the
 //! write rate; the delay is the price per hop. Without coalescing, every applied sync
 //! message would announce to every neighbour, and announcements would outnumber the
 //! data they announce (about 15 to 1 at 250 nodes).
@@ -31,7 +32,7 @@ use tracing::debug;
 
 use crate::locks::MutexExt;
 use crate::state::{AppState, Standing};
-use crate::swarm::peer::send_status;
+use crate::swarm::peer::announce_heads;
 use crate::swarm::topology;
 use crate::transport::{self, HeadsSummary};
 
@@ -86,7 +87,7 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) {
     }
     state.stats.announce_round();
     for peer in neighbours {
-        if let Err(error) = send_status(state, &peer).await {
+        if let Err(error) = announce_heads(state, &peer).await {
             debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
         }
     }
diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs
index 0f52e23..1d7c6ac 100644
--- a/src/transport/libp2p/tests.rs
+++ b/src/transport/libp2p/tests.rs
@@ -59,6 +59,7 @@ fn status() -> ParsedMessage {
         ready: true,
         syncs_with_you: false,
         leaving: false,
+        announcing: false,
     })
 }
 
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index b8b57af..25f9b7c 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -202,6 +202,10 @@ pub(crate) struct Status {
     /// The sender is shutting down: its farewell (`keepalive::farewell`). The
     /// recipient holds it dormant at once, rather than after `dormant_after` of silence.
     pub(crate) leaving: bool,
+    /// The sender announces new heads (`swarm::relay`), rather than keeping alive: a
+    /// STATUS answered only if the recipient's heads differ. Most announcements reach a
+    /// neighbour that has the changes already, which has nothing to tell.
+    pub(crate) announcing: bool,
 }
 
 /// Cheap to clone: large parts are shared (`bytes::Bytes`).
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 9078134..9850ef0 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -717,6 +717,7 @@ mod tests {
             ready: true,
             syncs_with_you: true,
             leaving: false,
+            announcing: false,
         };
         let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap();
         // Mycelium's message data: a length byte, the topic, then our payload.
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 164574d..29b3c23 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -1513,6 +1513,7 @@ mod tests {
             ready: false,
             syncs_with_you: false,
             leaving: false,
+            announcing: false,
         })
     }
 
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index 2f167dc..7a679fc 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -387,6 +387,7 @@ mod tests {
             ready: true,
             syncs_with_you: false,
             leaving: false,
+            announcing: false,
         })
     }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index d72fba4..d3bf39d 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 15;
+pub(crate) const VERSION: u8 = 16;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -369,6 +369,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
         ready: true,
         syncs_with_you: true,
         leaving: false,
+        announcing: false,
     };
     let proof = IdentityProof {
         key: [8; 32],
@@ -549,6 +550,7 @@ mod tests {
             ready: true,
             syncs_with_you: false,
             leaving: false,
+            announcing: false,
         };
         let decoded =
             decode(&encode(&ParsedMessage::StatusReply(status.clone())).unwrap()).unwrap();
@@ -572,6 +574,7 @@ mod tests {
             ready: true,
             syncs_with_you: true,
             leaving: false,
+            announcing: false,
         };
         assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err());
     }
@@ -669,10 +672,11 @@ mod tests {
             ready: true,
             syncs_with_you: false,
             leaving: false,
+            announcing: true,
         });
         assert_eq!(
             hex(&encode(&status).unwrap()),
-            format!("{v}03969290c400{epoch_hex}01c3c2c2")
+            format!("{v}03979290c400{epoch_hex}01c3c2c2c3")
         );
 
         let who = ParsedMessage::Who(IdentityProof {