From 8d70536fd267e7ee730c390b6b47c03ddbef7df9 Mon Sep 17 00:00:00 2001
From: tv <tv@krebsco.de>
Date: Mon, 26 Oct 2026 18:51:01 +0100
Subject: read stamps sync and lock free

---
 src/swarm/blob.rs           |  4 ++--
 src/swarm/peer/handshake.rs |  4 ++--
 src/swarm/peer/keepalive.rs | 20 +++++++++++++-------
 src/swarm/relay.rs          |  5 +++--
 4 files changed, 20 insertions(+), 13 deletions(-)

(limited to 'src/swarm')

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index c153a17..4b69562 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -166,7 +166,7 @@ async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> {
 }
 
 async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) {
-    let (epoch, epoch_seq) = state.epoch_stamp().await;
+    let (epoch, epoch_seq) = state.epoch_stamp();
     let message = ParsedMessage::Blob(BlobMessage {
         epoch,
         epoch_seq,
@@ -220,7 +220,7 @@ impl Drop for Registration<'_> {
 }
 
 async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) {
-    let (epoch, epoch_seq) = state.epoch_stamp().await;
+    let (epoch, epoch_seq) = state.epoch_stamp();
     let message = ParsedMessage::Blob(BlobMessage {
         epoch,
         epoch_seq,
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.
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 4bfc2b6..34451c9 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -103,8 +103,9 @@ pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transp
     if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
         return;
     }
-    // The common case, the same heads, needs only the cached heads (read lock).
-    if HeadsSummary::of(state.doc.read().await.heads().to_vec()) == remote.heads {
+    // The common case, the same heads, needs only the published heads.
+    let ours = HeadsSummary::of(state.stamp.borrow().heads.to_vec());
+    if ours == remote.heads {
         return;
     }
     let missing = match &remote.heads {
-- 
cgit v1.3.1