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>,
}

/// 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,
            writable: true,
            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<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`), 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,
        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
    ) -> Result<T, E> {
        transact(&mut self.doc, at, f)
    }
}

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");
    }
}

/// The writes `f` makes to `doc`, as one change stamped `at` 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,
    f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
) -> Result<T, E> {
    let mut transaction = 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
}