path: root/src/document/replica.rs

//! 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<Epoch> 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<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
/// 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<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 = 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<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(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);
}