//! The local replica: the Automerge document together with the epoch it belongs to. //! //! A remake replaces the document with a new one (for compaction: a snapshot) whose //! history is unrelated to the old one. Syncing two unrelated histories would make Automerge merge //! them, turning every key into a conflict with its own copy, so nodes on different //! epochs must never sync. Keeping the epoch and the document under one lock makes it //! impossible to label a sync message from one document with the other's epoch. use std::fmt; use std::ops::{Deref, DerefMut}; use std::sync::Arc; use super::{ChangeHash, Document, Transaction}; use tokio::sync::watch; /// One generation of the swarm document. /// /// * `generation` orders epochs: every remake increments it, so a node can tell /// whether a peer is behind or ahead. /// * `id` is the hash of the epoch's snapshot change. It tells two epochs of the same /// generation apart, which only happens if two remakes committed concurrently. #[derive(Clone, Copy, PartialEq, Eq, Hash)] pub(crate) struct Epoch { pub(crate) generation: u64, pub(crate) id: [u8; 32], } /// How a peer's epoch relates to ours. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum EpochRelation { Same, /// The peer is on an older generation and has to migrate. PeerBehind, /// The peer is on a newer generation; we have to migrate. PeerAhead, /// Same generation, different snapshot: competing remakes. Never sync. Diverged, } impl Epoch { /// Swarms that have never been compacted. pub(crate) const GENESIS: Epoch = Epoch { generation: 0, id: [0; 32], }; pub(crate) fn relation_to(&self, peer: &Epoch) -> EpochRelation { use std::cmp::Ordering; match peer.generation.cmp(&self.generation) { Ordering::Less => EpochRelation::PeerBehind, Ordering::Greater => EpochRelation::PeerAhead, Ordering::Equal if peer.id == self.id => EpochRelation::Same, Ordering::Equal => EpochRelation::Diverged, } } pub(crate) fn id_hex(&self) -> String { 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 for EpochRecord { fn from(epoch: Epoch) -> Self { Self { generation: epoch.generation, id: epoch.id_hex(), } } } impl fmt::Display for Epoch { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { if *self == Epoch::GENESIS { write!(f, "0:genesis") } else { write!(f, "{}:{}", self.generation, &self.id_hex()[..12]) } } } impl fmt::Debug for Epoch { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fmt::Display::fmt(self, f) } } /// The document and its epoch, always locked and swapped together. /// /// Derefs to the document, so code that only needs the document keeps working with a /// guard of `Replica`. /// /// 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`]. /// /// 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(crate) struct Replica { pub(crate) epoch: Epoch, /// 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: 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, stamp: watch::Sender, /// 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, } /// 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, to: Vec, patches: automerge::PatchLog, } impl Log { fn at(epoch_seq: u32, heads: Vec) -> Self { Self { epoch_seq, from: heads.clone(), to: heads, patches: automerge::PatchLog::active(), } } } fn sorted_heads(doc: &Document) -> Vec { 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 /// together: keepalives and HELLOs read them without the document's lock. #[derive(Clone, Debug)] pub(crate) struct Stamp { pub(crate) epoch: Epoch, pub(crate) epoch_seq: u32, /// Sorted. pub(crate) heads: Arc<[ChangeHash]>, } impl Replica { 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 } /// The document's heads as of the last refresh, sorted. pub(crate) fn heads(&self) -> &[ChangeHash] { &self.heads } /// Call after changing the document or its epoch (under the write lock). Publishes /// the new [`Stamp`]. pub(crate) 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(crate) fn stamps(&self) -> watch::Receiver { 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( &mut self, at: crate::clock::UnixMillis, writer: &Writer, f: impl FnOnce(&mut Transaction<'_>) -> Result, ) -> Result { 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( &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 { 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 = Document; fn deref(&self) -> &Document { &self.doc } } impl DerefMut for Replica { fn deref_mut(&mut self) -> &mut Document { &mut self.doc } } #[cfg(test)] mod tests { use super::*; fn epoch(generation: u64, byte: u8) -> Epoch { Epoch { generation, id: [byte; 32], } } #[test] fn relation() { let local = epoch(3, 1); assert_eq!(local.relation_to(&epoch(3, 1)), EpochRelation::Same); assert_eq!(local.relation_to(&epoch(2, 9)), EpochRelation::PeerBehind); assert_eq!(local.relation_to(&epoch(4, 9)), EpochRelation::PeerAhead); 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 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, as the change's author (its node ID), /// and the user it wrote for through the API, if any, as its message. 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 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( doc: &mut Document, at: crate::clock::UnixMillis, writer: &Writer, f: impl FnOnce(&mut Transaction<'_>) -> Result, ) -> Result { 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( doc: &mut Document, at: crate::clock::UnixMillis, writer: &Writer, log: &mut automerge::PatchLog, f: impl FnOnce(&mut Transaction<'_>) -> Result, ) -> (Result, 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( mut transaction: Transaction<'_>, at: crate::clock::UnixMillis, writer: &Writer, f: impl FnOnce(&mut Transaction<'_>) -> Result, ) -> (Result, Option) { 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(user) = writer.user { options.set_message(user); } 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); }