aboutsummaryrefslogtreecommitdiffstats
path: root/src/document/replica.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/document/replica.rs')
-rw-r--r--src/document/replica.rs324
1 files changed, 311 insertions, 13 deletions
diff --git a/src/document/replica.rs b/src/document/replica.rs
index fe76c12..03325eb 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -10,7 +10,7 @@ use std::fmt;
use std::ops::{Deref, DerefMut};
use std::sync::Arc;
-use automerge::{AutoCommit, ChangeHash};
+use super::{ChangeHash, Document, Transaction};
use tokio::sync::watch;
/// One generation of the swarm document.
@@ -55,7 +55,25 @@ impl Epoch {
}
pub(crate) fn id_hex(&self) -> String {
- self.id.iter().map(|byte| format!("{byte:02x}")).collect()
+ crate::transport::Hex(&self.id).to_string()
+ }
+}
+
+/// An epoch as records and the API show it: its generation and its snapshot hash in
+/// hex (all zeros for the genesis epoch).
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct EpochRecord {
+ pub(crate) generation: u64,
+ pub(crate) id: String,
+}
+
+impl From<Epoch> for EpochRecord {
+ fn from(epoch: Epoch) -> Self {
+ Self {
+ generation: epoch.generation,
+ id: epoch.id_hex(),
+ }
}
}
@@ -77,13 +95,12 @@ impl fmt::Debug for Epoch {
/// The document and its epoch, always locked and swapped together.
///
-/// Derefs to the document, so code that only needs the `AutoCommit` keeps working
-/// with a guard of `Replica`.
+/// Derefs to the document, so code that only needs the document keeps working with a
+/// guard of `Replica`.
///
-/// The heads are cached: `AutoCommit::get_heads` takes `&mut self` (it closes a pending
-/// transaction), so without the cache every keepalive and status read would need the
-/// write lock. Every path that changes the document under the write lock calls
-/// [`Replica::refresh_heads`] before releasing it (`AppState::write_doc`, `mutate_doc`,
+/// The heads are cached, sorted, so keepalives and status reads (and the stamp below)
+/// don't recompute them. Every path that changes the document under the write lock calls
+/// [`Replica::refresh_heads`] before releasing it (`AppState::mutate_doc`,
/// `switch_epoch`, the sync apply); readers take the read lock and use
/// [`Replica::heads`].
///
@@ -95,9 +112,46 @@ pub(crate) struct Replica {
/// How many times this node changed epoch (switch, rejoin, abandoned rejoin) in this
/// run: messages carry it, so a late one can't flip a peer's view of our epoch back.
pub(crate) epoch_seq: u32,
- pub(crate) doc: AutoCommit,
+ pub(crate) doc: Document,
+ /// Whether writes go in (`AppState::mutate_doc`). False while a rejoin syncs a fresh
+ /// document: a write would land in a document the epoch's snapshot hasn't reached
+ /// yet, or be thrown away with it if the rejoin is abandoned. Cleared and set again
+ /// in the same lock as the swap (`Host::switch_epoch`), so no write checked
+ /// before a swap lands after it.
+ pub(crate) writable: bool,
heads: Vec<ChangeHash>,
stamp: watch::Sender<Stamp>,
+ /// What local writes and sync changed since the change publisher last looked
+ /// (`changed_since`), so it needn't diff the document for it.
+ log: std::sync::Mutex<Log>,
+}
+
+/// The patches of the changes from `from` to `to` (sorted heads) of the document of
+/// `epoch_seq`, as they were made. Changes made otherwise (applied from a delta, a
+/// merge) aren't in it: the heads then aren't `to` any more, and whoever looks diffs
+/// instead. Another document (an epoch switch) starts a log of its own.
+struct Log {
+ epoch_seq: u32,
+ from: Vec<ChangeHash>,
+ to: Vec<ChangeHash>,
+ patches: automerge::PatchLog,
+}
+
+impl Log {
+ fn at(epoch_seq: u32, heads: Vec<ChangeHash>) -> Self {
+ Self {
+ epoch_seq,
+ from: heads.clone(),
+ to: heads,
+ patches: automerge::PatchLog::active(),
+ }
+ }
+}
+
+fn sorted_heads(doc: &Document) -> Vec<ChangeHash> {
+ let mut heads = doc.get_heads();
+ heads.sort_unstable();
+ heads
}
/// The epoch label and the heads of the replica, as of its last refresh, published
@@ -111,18 +165,21 @@ pub(crate) struct Stamp {
}
impl Replica {
- pub(crate) fn new(epoch: Epoch, doc: AutoCommit) -> Self {
+ pub(crate) fn new(epoch: Epoch, doc: Document) -> Self {
let (stamp, _) = watch::channel(Stamp {
epoch,
epoch_seq: 0,
heads: Arc::from([]),
});
+ let log = std::sync::Mutex::new(Log::at(0, sorted_heads(&doc)));
let mut replica = Self {
epoch,
epoch_seq: 0,
doc,
+ writable: true,
heads: Vec::new(),
stamp,
+ log,
};
replica.refresh_heads();
replica
@@ -149,18 +206,77 @@ impl Replica {
pub(crate) fn stamps(&self) -> watch::Receiver<Stamp> {
self.stamp.subscribe()
}
+
+ /// The writes `f` makes, as one change stamped `at` if it succeeds (receivers tell
+ /// by the stamp how long the change took to reach them, `swarm::stats`), by
+ /// `writer`, dropped if not. Doesn't refresh the heads: the caller does, under the
+ /// same lock.
+ pub(crate) fn transact<T, E>(
+ &mut self,
+ at: crate::clock::UnixMillis,
+ writer: &Writer,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
+ ) -> Result<T, E> {
+ self.logged(|doc, log| transact_logged(doc, at, writer, log, f))
+ }
+
+ /// Runs `f`, which changes the document and records what it changed in the log it
+ /// is given; it returns whether the log still holds everything (a rolled-back
+ /// transaction takes the log with it).
+ pub(crate) fn logged<T>(
+ &mut self,
+ f: impl FnOnce(&mut Document, &mut automerge::PatchLog) -> (T, bool),
+ ) -> T {
+ use crate::locks::MutexExt;
+ let before = sorted_heads(&self.doc);
+ let mut log = self.log.lock_or_recover();
+ if log.epoch_seq != self.epoch_seq || log.to != before {
+ // Another document, or changed otherwise since: what the log holds doesn't
+ // lead here.
+ *log = Log::at(self.epoch_seq, before);
+ }
+ let (out, whole) = f(&mut self.doc, &mut log.patches);
+ let after = sorted_heads(&self.doc);
+ if whole {
+ log.to = after;
+ } else {
+ *log = Log::at(self.epoch_seq, after);
+ }
+ out
+ }
+
+ /// The store paths changed since `before` (sorted heads, of this document) to
+ /// now (`store::changed`): from the log where it covers exactly that, else by a
+ /// diff. Starts the log again from now.
+ pub(crate) fn changed_since(
+ &self,
+ before: &[ChangeHash],
+ lists: super::store::InLists,
+ ) -> Vec<super::store::Changed> {
+ use crate::locks::MutexExt;
+ let now = sorted_heads(&self.doc);
+ let mut log = self.log.lock_or_recover();
+ let changed = if log.epoch_seq == self.epoch_seq && log.from == before && log.to == now {
+ let patches = self.doc.make_patches(&mut log.patches);
+ super::store::changed_in(&patches, lists)
+ } else {
+ super::store::changed(&self.doc, before, &now, lists)
+ };
+ *log = Log::at(self.epoch_seq, now);
+ changed
+ }
}
impl Deref for Replica {
- type Target = AutoCommit;
+ type Target = Document;
- fn deref(&self) -> &AutoCommit {
+ fn deref(&self) -> &Document {
&self.doc
}
}
impl DerefMut for Replica {
- fn deref_mut(&mut self) -> &mut AutoCommit {
+ fn deref_mut(&mut self) -> &mut Document {
&mut self.doc
}
}
@@ -185,9 +301,191 @@ mod tests {
assert_eq!(local.relation_to(&epoch(3, 2)), EpochRelation::Diverged);
}
+ /// A rolled-back write leaves its writer unused in the document; saving must not
+ /// trip over it (Automerge refuses to save a document with unused writers).
+ #[test]
+ fn saves_after_a_rolled_back_write() {
+ use crate::clock::UnixMillis;
+ use crate::document::store;
+ use serde_json::json;
+
+ let path = vec!["a".to_owned()];
+ let mut replica = Replica::new(Epoch::GENESIS, Document::new());
+
+ let writer = Writer {
+ node: "node",
+ user: None,
+ };
+ let refused: Result<(), &str> = replica.transact(UnixMillis(1), &writer, |doc| {
+ store::write_path(doc, &path, &json!(1)).unwrap();
+ Err("refused")
+ });
+ assert!(refused.is_err());
+ let loaded = super::super::load(&super::super::save(&replica.doc)).unwrap();
+ assert!(loaded.get_heads().is_empty());
+ assert_eq!(store::read_path(&loaded, &path).unwrap(), None);
+
+ // The same writer commits afterwards, and that saves too.
+ replica
+ .transact(UnixMillis(2), &writer, |doc| {
+ store::write_path(doc, &path, &json!(2))
+ })
+ .unwrap();
+ replica.refresh_heads();
+ let loaded = super::super::load(&super::super::save(&replica.doc)).unwrap();
+ let mut heads = loaded.get_heads();
+ heads.sort_unstable();
+ assert_eq!(heads, replica.heads());
+ assert_eq!(store::read_path(&loaded, &path).unwrap(), Some(json!(2)));
+ }
+
+ #[test]
+ fn messages_name_their_user_in_a_trailer() {
+ let writer = Writer {
+ node: "node",
+ user: Some("tv"),
+ };
+ let message = writer.message().unwrap();
+ assert_eq!(message, "User: tv");
+ assert_eq!(user_of(&message), Some("tv"));
+ assert_eq!(
+ user_of("Moved the settings.\n\nUser: tv\nSigned-off-by: someone"),
+ Some("tv")
+ );
+ // Before trailers: the user itself.
+ assert_eq!(user_of("tv"), Some("tv"));
+ assert_eq!(user_of("Moved the settings."), None);
+ assert_eq!(user_of("Moved: the settings\nand more"), None);
+ assert_eq!(user_of("Signed-off-by: someone"), None);
+ }
+
#[test]
fn display() {
assert_eq!(Epoch::GENESIS.to_string(), "0:genesis");
assert_eq!(epoch(2, 0xab).to_string(), "2:abababababab");
}
}
+
+/// Who makes a change, as it records: the node that commits it, as its Automerge author
+/// (its node ID), and the user it wrote for through the API, if any, in its message
+/// (`User:`, `USER_TRAILER`). That user, at that node, is who wrote it: user names are
+/// each node's own (`--htpasswd`). Every node holds every change, so who wrote what is
+/// known swarm-wide (`GET /admin/changes`).
+pub(crate) struct Writer<'a> {
+ pub(crate) node: &'a str,
+ pub(crate) user: Option<&'a str>,
+}
+
+/// The trailer of a change's message naming its user (`Writer`), as Git's
+/// `Signed-off-by:` names who signed off.
+pub(crate) const USER_TRAILER: &str = "User";
+
+impl Writer<'_> {
+ /// The change's message: its trailers (`trailers`), if it has any.
+ fn message(&self) -> Option<String> {
+ self.user.map(|user| format!("{USER_TRAILER}: {user}"))
+ }
+}
+
+/// The trailers of a change's message, as in Git: the `Key: value` lines of its last
+/// paragraph, if each of its lines is one.
+pub(crate) fn trailers(message: &str) -> Vec<(&str, &str)> {
+ let last = message.trim_end().rsplit("\n\n").next().unwrap_or("");
+ let trailers: Option<Vec<(&str, &str)>> = last
+ .lines()
+ .map(|line| {
+ let (key, value) = line.split_once(": ")?;
+ (!key.is_empty() && key.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-'))
+ .then(|| (key, value.trim()))
+ })
+ .collect();
+ trailers.unwrap_or_default()
+}
+
+/// The user a change's message names (`USER_TRAILER`). Changes made before messages had
+/// trailers have the user as their whole message: one word, without a `:`
+/// (`--htpasswd` names have none).
+pub(crate) fn user_of(message: &str) -> Option<&str> {
+ let trailers = trailers(message);
+ if trailers.is_empty() {
+ let message = message.trim();
+ let word =
+ !message.is_empty() && !message.contains(|c: char| c == ':' || c.is_whitespace());
+ return word.then_some(message);
+ }
+ trailers
+ .into_iter()
+ .find(|(key, _)| key.eq_ignore_ascii_case(USER_TRAILER))
+ .map(|(_, user)| user)
+}
+
+/// The writes `f` makes to `doc`, as one change stamped `at` by `writer` if it succeeds,
+/// dropped if not ([`Replica::transact`], and writes into the document before a rejoin).
+pub(crate) fn transact<T, E>(
+ doc: &mut Document,
+ at: crate::clock::UnixMillis,
+ writer: &Writer,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
+) -> Result<T, E> {
+ transact_logged(doc, at, writer, &mut automerge::PatchLog::inactive(), f).0
+}
+
+/// [`transact`], what it changes recorded in `log`; with whether `log` still holds
+/// everything (`Replica::logged`).
+fn transact_logged<T, E>(
+ doc: &mut Document,
+ at: crate::clock::UnixMillis,
+ writer: &Writer,
+ log: &mut automerge::PatchLog,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
+) -> (Result<T, E>, bool) {
+ authored(doc, writer.node);
+ let taken = std::mem::replace(log, automerge::PatchLog::inactive());
+ if let Ok(transaction) = doc.transaction_log_patches(taken) {
+ let (out, logged) = run(transaction, at, writer, f);
+ let whole = logged.is_some();
+ if let Some(logged) = logged {
+ *log = logged;
+ }
+ return (out, whole);
+ }
+ // Another document's log (the epoch switched): this change goes unlogged.
+ (run(doc.transaction(), at, writer, f).0, false)
+}
+
+/// Runs `f` in `transaction`, committing what it writes if it succeeds (stamped `at`,
+/// by `writer`) with the transaction's log, rolling it back if not.
+fn run<T, E>(
+ mut transaction: Transaction<'_>,
+ at: crate::clock::UnixMillis,
+ writer: &Writer,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
+) -> (Result<T, E>, Option<automerge::PatchLog>) {
+ let out = f(&mut transaction);
+ if out.is_err() {
+ transaction.rollback();
+ return (out, None);
+ }
+ let mut options =
+ automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64);
+ if let Some(message) = writer.message() {
+ options.set_message(message);
+ }
+ let (_, logged) = transaction.commit_with(options);
+ (out, Some(logged))
+}
+
+/// Makes `node` the author of `doc`'s changes from now on. Setting an author gives the
+/// document a new actor; this node keeps its own (`with_actor`), which it tracks
+/// (`RemakeState::own_actors`, the document file), as only it writes with it.
+fn authored(doc: &mut Document, node: &str) {
+ if doc
+ .get_author()
+ .is_some_and(|author| author.as_bytes() == node.as_bytes())
+ {
+ return;
+ }
+ let actor = doc.get_actor().clone();
+ doc.set_author(Some(automerge::Author::from(node.as_bytes().to_vec())));
+ doc.set_actor(actor);
+}