read stamps sync and lock free

2 files changed, 15 insertions, 9 deletions

diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index f08929a..7e494c6 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -118,7 +118,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
         .as_ref()
         .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer))
         .unwrap_or_default();
-    let (epoch, epoch_seq) = state.epoch_stamp().await;
+    let (epoch, epoch_seq) = state.epoch_stamp();
     // Retries repeat the handshake's number.
     let handshake = match state.get_peer_state(peer).await {
         Some(peer_state) => peer_state.lock().await.handshake,
@@ -209,7 +209,7 @@ pub(super) async fn send_hello_reply(
         .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer))
         .unwrap_or_default();
     let ready = state.is_ready().await;
-    let (epoch, epoch_seq) = state.epoch_stamp().await;
+    let (epoch, epoch_seq) = state.epoch_stamp();
     let reply = transport::HelloReply {
         swarm_id,
         peers,
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 4fcad7e..4ce326e 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -25,14 +25,14 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
 
 /// Our keepalive for `peer`.
 pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
-    // Heads and epoch come from the same locked replica, so they always match. The heads
-    // are cached (`Replica`), so the read lock is enough.
+    // Heads and epoch are published together (`Replica::stamps`), so they always match,
+    // and reading them doesn't wait for a sync message being applied.
     let (heads, epoch, epoch_seq) = {
-        let replica = state.doc.read().await;
+        let stamp = state.stamp.borrow();
         (
-            transport::HeadsSummary::of(replica.heads().to_vec()),
-            replica.epoch,
-            replica.epoch_seq,
+            transport::HeadsSummary::of(stamp.heads.to_vec()),
+            stamp.epoch,
+            stamp.epoch_seq,
         )
     };
     let syncs_with_you = match state.get_peer_state(peer).await {
@@ -62,7 +62,13 @@ pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transp
         peer_state.syncs_with_us = Some(remote.syncs_with_you);
         peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
         drop(peer_state);
-        crate::swarm::relay::pull_if_behind(state, peer, remote).await;
+        // Off the consumer: telling which heads we lack takes the document's lock.
+        {
+            let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone());
+            tokio::spawn(async move {
+                crate::swarm::relay::pull_if_behind(&state, &peer, &remote).await
+            });
+        }
         if became_ready {
             // We may have left the opening to it while it wasn't ready, and nothing else
             // would make us look again.