enrich keepalive content

Add ready and syncs_with_you flags as well as actual heads, so the peer
- ready: same meaning as in the sync message
- syncs_with_you: sender counts recipient as member and in same epoch
- heads: allows to tell who's behind or if both sides have changes

1 files changed, 39 insertions, 26 deletions

diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 1e8b91d..f98bf17 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -6,7 +6,6 @@ use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow, bail};
 use automerge::AutoCommit;
-use sha2::{Digest, Sha256};
 use tokio::sync::Mutex;
 use tokio::time::{Instant, MissedTickBehavior};
 use tracing::{debug, info, instrument, trace, warn};
@@ -98,23 +97,40 @@ async fn apply_forget_records(state: &AppState) {
 
 // --- keepalive and self-healing ----------------------------------------------
 
-fn heads_digest(doc: &mut AutoCommit) -> transport::HeadsDigest {
-    let mut heads = doc.get_heads();
-    heads.sort_unstable();
-    let mut hasher = Sha256::new();
-    for head in &heads {
-        hasher.update(head.0);
+/// Our keepalive for `peer`.
+pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
+    // AutoCommit::get_heads takes &mut self (it closes the pending transaction).
+    // Heads and epoch come from the same locked replica, so they always match.
+    let (heads, epoch) = {
+        let mut replica = state.doc.write().await;
+        (
+            transport::HeadsSummary::of(replica.doc.get_heads()),
+            replica.epoch,
+        )
+    };
+    let syncs_with_you = match state.get_peer_state(peer).await {
+        Some(peer_state) => {
+            let peer_state = peer_state.lock().await;
+            peer_state.liveness == Liveness::Active
+                && matches!(peer_state.standing, Standing::Member { .. })
+                && peer_state.epoch == Some(epoch)
+        }
+        None => false,
+    };
+    transport::Keepalive {
+        heads,
+        epoch,
+        ready: state.is_ready().await,
+        syncs_with_you,
     }
-    hasher.finalize().into()
 }
 
-pub(crate) async fn local_keepalive(state: &AppState) -> transport::Keepalive {
-    // AutoCommit::get_heads takes &mut self (it closes the pending transaction).
-    // Digest and epoch come from the same locked replica, so they always match.
-    let mut replica = state.doc.write().await;
-    transport::Keepalive {
-        heads_digest: heads_digest(&mut replica.doc),
-        epoch: replica.epoch,
+/// What a peer's keepalive says about it.
+async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) {
+    if let Some(peer_state) = state.get_peer_state(peer).await {
+        let mut peer_state = peer_state.lock().await;
+        peer_state.reports_ready = Some(remote.ready);
+        peer_state.syncs_with_us = Some(remote.syncs_with_you);
     }
 }
 
@@ -124,7 +140,7 @@ async fn heal_if_diverged(
     local: &transport::Keepalive,
     remote: &transport::Keepalive,
 ) {
-    // Digests of documents from different epochs are never equal and never converge.
+    // Heads of documents from different epochs are never equal and never converge.
     if local.epoch != remote.epoch {
         return;
     }
@@ -135,18 +151,13 @@ async fn heal_if_diverged(
 
     let heal = {
         let mut peer_state = peer_state_mutex.lock().await;
-        if local.heads_digest == remote.heads_digest {
+        if local.heads == remote.heads {
             // Same document: whatever answer we were waiting for, there is nothing left
             // to send. (Automerge answers only when it has something to say, so a sync
             // message to a peer that already has everything gets no reply at all.)
             peer_state.sync.reply_settled();
         }
-        peer_state.needs_heal(
-            &local.heads_digest,
-            &remote.heads_digest,
-            &state.timing,
-            Instant::now(),
-        )
+        peer_state.needs_heal(&local.heads, &remote.heads, &state.timing, Instant::now())
     };
 
     if heal {
@@ -432,7 +443,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             }
 
             if ping_due {
-                let local = local_keepalive(&state).await;
+                let local = local_keepalive(&state, peer).await;
                 if let Err(error) = state
                     .transport
                     .send_message(peer, &ParsedMessage::Ping(local))
@@ -747,7 +758,8 @@ pub(crate) async fn handle_message(
             if reactivated || peer_swarm(state, &peer).await.is_none() {
                 ensure_handshake(state, &peer).await;
             }
-            let local = local_keepalive(state).await;
+            note_keepalive(state, &peer, &remote).await;
+            let local = local_keepalive(state, &peer).await;
             heal_if_diverged(state, &peer, &local, &remote).await;
             if let Err(error) = state
                 .transport
@@ -761,7 +773,8 @@ pub(crate) async fn handle_message(
             if reactivated || peer_swarm(state, &peer).await.is_none() {
                 ensure_handshake(state, &peer).await;
             }
-            let local = local_keepalive(state).await;
+            note_keepalive(state, &peer, &remote).await;
+            let local = local_keepalive(state, &peer).await;
             heal_if_diverged(state, &peer, &local, &remote).await;
         }
     }