summaryrefslogtreecommitdiffstats
path: root/src/document/replica.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/document/replica.rs')
-rw-r--r--src/document/replica.rs36
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()
}
}