//! 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, heads: Vec, stamp: watch::Sender, } /// 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 mut replica = Self { epoch, epoch_seq: 0, doc, heads: Vec::new(), stamp, }; 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`), 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, f: impl FnOnce(&mut Transaction<'_>) -> Result, ) -> Result { let mut transaction = self.doc.transaction(); let out = f(&mut transaction); match &out { Ok(_) => { transaction.commit_with( automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64), ); } Err(_) => { transaction.rollback(); } } out } } 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 refused: Result<(), &str> = replica.transact(UnixMillis(1), |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), |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"); } }