read stamps sync and lock free

1 files changed, 35 insertions, 1 deletions

diff --git a/src/document/replica.rs b/src/document/replica.rs
index db47223..0cab0be 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -8,8 +8,10 @@
 
 use std::fmt;
 use std::ops::{Deref, DerefMut};
+use std::sync::Arc;
 
 use automerge::{AutoCommit, ChangeHash};
+use tokio::sync::watch;
 
 /// One generation of the swarm document.
 ///
@@ -84,6 +86,10 @@ impl fmt::Debug for Epoch {
 /// [`Replica::refresh_heads`] before releasing it (`AppState::write_doc`, `mutate_doc`,
 /// `switch_epoch`, the sync apply); readers take the read lock and use
 /// [`Replica::heads`].
+///
+/// The refresh also publishes a [`Stamp`], for what must not wait for the lock: a
+/// tokio read lock waits behind a writer, and applying a large sync message holds the
+/// write lock for seconds.
 pub struct Replica {
     pub epoch: Epoch,
     /// How many times this node changed epoch (switch, rejoin, abandoned rejoin) in this
@@ -91,15 +97,32 @@ pub struct Replica {
     pub epoch_seq: u32,
     pub doc: AutoCommit,
     heads: Vec<ChangeHash>,
+    stamp: watch::Sender<Stamp>,
+}
+
+/// The epoch label and the heads of the replica, as of its last refresh, published
+/// together: keepalives and HELLOs read them without the document's lock.
+#[derive(Clone, Debug)]
+pub struct Stamp {
+    pub epoch: Epoch,
+    pub epoch_seq: u32,
+    /// Sorted.
+    pub heads: Arc<[ChangeHash]>,
 }
 
 impl Replica {
     pub fn new(epoch: Epoch, doc: AutoCommit) -> Self {
+        let (stamp, _) = watch::channel(Stamp {
+            epoch,
+            epoch_seq: 0,
+            heads: Arc::from([]),
+        });
         let mut replica = Self {
             epoch,
             epoch_seq: 0,
             doc,
             heads: Vec::new(),
+            stamp,
         };
         replica.refresh_heads();
         replica
@@ -110,10 +133,21 @@ impl Replica {
         &self.heads
     }
 
-    /// Call after changing the document (under the write lock).
+    /// Call after changing the document or its epoch (under the write lock). Publishes
+    /// the new [`Stamp`].
     pub fn refresh_heads(&mut self) {
         self.heads = self.doc.get_heads();
         self.heads.sort_unstable();
+        self.stamp.send_replace(Stamp {
+            epoch: self.epoch,
+            epoch_seq: self.epoch_seq,
+            heads: Arc::from(self.heads.as_slice()),
+        });
+    }
+
+    /// The published stamps, from the last refresh on.
+    pub fn stamps(&self) -> watch::Receiver<Stamp> {
+        self.stamp.subscribe()
     }
 }