blockworx/log/

dag.rs

//! The history graph: changes keyed by content address, and the canonical
//! order they replay in.
//!
//! The linearization is *not* what makes the merge correct — registers commute,
//! so any order folds to the same state. It exists so that two replicas
//! produce the same snapshot bytes, so undo can find an actor's latest change,
//! and so collision reporting is reproducible.

use std::cmp::Reverse;
use std::collections::{BTreeMap, BTreeSet, BinaryHeap};

use crate::log::change::Change;
use crate::log::id::{ActorId, ChangeHash, Lamport};

/// Changes held by hash. Inserting the same change twice is a no-op, which is
/// what makes ingest idempotent at the graph level as well as the fold's.
#[derive(Clone, Debug, Default)]
pub struct Dag {
    changes: BTreeMap<ChangeHash, Change>,
}

impl Dag {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn len(&self) -> usize {
        self.changes.len()
    }

    pub fn is_empty(&self) -> bool {
        self.changes.is_empty()
    }

    pub fn contains(&self, hash: ChangeHash) -> bool {
        self.changes.contains_key(&hash)
    }

    pub fn get(&self, hash: ChangeHash) -> Option<&Change> {
        self.changes.get(&hash)
    }

    pub fn insert(&mut self, change: Change) -> ChangeHash {
        let hash = change.hash();
        self.changes.entry(hash).or_insert(change);
        hash
    }

    pub fn iter(&self) -> impl Iterator<Item = (ChangeHash, &Change)> {
        self.changes.iter().map(|(hash, change)| (*hash, change))
    }

    /// Parents referenced by changes held here but not themselves held.
    ///
    /// Non-empty means either the history is shallow — compacted down to a
    /// baseline — or changes arrived before their ancestors and want
    /// buffering.
    pub fn missing_parents(&self) -> BTreeSet<ChangeHash> {
        self.changes
            .values()
            .flat_map(Change::parents)
            .copied()
            .filter(|parent| !self.changes.contains_key(parent))
            .collect()
    }

    /// The changes nothing else builds on — what a peer is told about, and
    /// what the next local change takes as its parents. Sorted, so two
    /// replicas holding the same history describe it identically.
    pub fn heads(&self) -> Vec<ChangeHash> {
        let claimed: BTreeSet<ChangeHash> = self
            .changes
            .values()
            .flat_map(Change::parents)
            .copied()
            .collect();
        self.changes
            .keys()
            .copied()
            .filter(|hash| !claimed.contains(hash))
            .collect()
    }

    /// Every change in replay order: parents before children, concurrent
    /// changes broken by `(lamport, actor)` and finally by hash.
    ///
    /// A parent that is not held is treated as already replayed, so a shallow
    /// history linearizes rather than deadlocking. Whether such a history is
    /// acceptable is the caller's call — [`missing_parents`](Self::missing_parents)
    /// is how they find out.
    pub fn linearize(&self) -> Vec<ChangeHash> {
        let mut waiting_on: BTreeMap<ChangeHash, usize> = BTreeMap::new();
        let mut dependents: BTreeMap<ChangeHash, Vec<ChangeHash>> = BTreeMap::new();

        for (hash, change) in &self.changes {
            let held_parents: Vec<ChangeHash> = change
                .parents()
                .iter()
                .copied()
                .filter(|p| self.changes.contains_key(p))
                .collect();
            waiting_on.insert(*hash, held_parents.len());
            for parent in held_parents {
                dependents.entry(parent).or_default().push(*hash);
            }
        }

        let mut ready: BinaryHeap<Reverse<Key>> = waiting_on
            .iter()
            .filter(|(_, count)| **count == 0)
            .map(|(hash, _)| Reverse(self.key(*hash)))
            .collect();

        let mut order = Vec::with_capacity(self.changes.len());
        while let Some(Reverse(Key { hash, .. })) = ready.pop() {
            order.push(hash);
            for dependent in dependents.get(&hash).into_iter().flatten() {
                let count = waiting_on.entry(*dependent).or_default();
                *count = count.saturating_sub(1);
                if *count == 0 {
                    ready.push(Reverse(self.key(*dependent)));
                }
            }
        }
        order
    }

    fn key(&self, hash: ChangeHash) -> Key {
        let change = self.changes.get(&hash);
        Key {
            lamport: change.map_or(Lamport::ZERO, |c| c.lamport()),
            actor: change.map_or_else(|| ActorId::from_uuid(uuid::Uuid::nil()), |c| c.actor()),
            hash,
        }
    }
}

/// Sort key for concurrent changes. The hash is the final tiebreak so the
/// order is total even if two changes somehow shared a stamp.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
struct Key {
    lamport: Lamport,
    actor: ActorId,
    hash: ChangeHash,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::log::change::SemanticLabel;
    use crate::log::command::{Command, ElementKind, PropSet};
    use crate::log::id::{ElementId, Stamp};

    fn actor(byte: u8) -> ActorId {
        ActorId::from_uuid(uuid::Uuid::from_bytes([byte; 16]))
    }

    fn change(
        lamport: u64,
        actor_byte: u8,
        parents: impl IntoIterator<Item = ChangeHash>,
    ) -> Change {
        Change::new(
            Stamp {
                lamport: Lamport::new(lamport),
                actor: actor(actor_byte),
            },
            parents,
            SemanticLabel::new(format!("l{lamport}a{actor_byte}")),
            vec![Command::Create {
                id: ElementId::new(),
                kind: ElementKind::Block,
                parent: ElementId::DOCUMENT,
                init: PropSet::new(),
            }],
        )
    }

    fn position(order: &[ChangeHash], hash: ChangeHash) -> usize {
        order
            .iter()
            .position(|h| *h == hash)
            .expect("every held change is linearized")
    }

    #[test]
    fn inserting_a_change_twice_holds_it_once() {
        let mut dag = Dag::new();
        let c = change(1, 1, []);
        let first = dag.insert(c.clone());
        let second = dag.insert(c);
        assert_eq!(first, second);
        assert_eq!(dag.len(), 1);
    }

    #[test]
    fn a_chain_has_one_head_and_a_fork_has_two() {
        let mut dag = Dag::new();
        let root = dag.insert(change(1, 1, []));
        let next = dag.insert(change(2, 1, [root]));
        assert_eq!(dag.heads(), vec![next]);

        let branch = dag.insert(change(2, 2, [root]));
        let mut heads = dag.heads();
        heads.sort_unstable();
        let mut expected = vec![next, branch];
        expected.sort_unstable();
        assert_eq!(heads, expected);
    }

    #[test]
    fn a_merge_change_collapses_the_heads() {
        let mut dag = Dag::new();
        let root = dag.insert(change(1, 1, []));
        let left = dag.insert(change(2, 1, [root]));
        let right = dag.insert(change(2, 2, [root]));
        assert_eq!(dag.heads().len(), 2);

        let merged = change(3, 1, [left, right]);
        assert!(merged.is_merge());
        let merge = dag.insert(merged);
        assert_eq!(dag.heads(), vec![merge]);
    }

    #[test]
    fn linearization_puts_parents_first() {
        let mut dag = Dag::new();
        let root = dag.insert(change(1, 1, []));
        let left = dag.insert(change(2, 1, [root]));
        let right = dag.insert(change(2, 2, [root]));
        let merge = dag.insert(change(3, 1, [left, right]));

        let order = dag.linearize();
        assert_eq!(order.len(), 4);
        assert!(position(&order, root) < position(&order, left));
        assert!(position(&order, root) < position(&order, right));
        assert!(position(&order, left) < position(&order, merge));
        assert!(position(&order, right) < position(&order, merge));
    }

    /// Two replicas receive the same changes in opposite orders; the
    /// linearization has to be a function of the graph, not of arrival.
    #[test]
    fn linearization_does_not_depend_on_insertion_order() {
        let root = change(1, 1, []);
        let root_hash = root.hash();
        let left = change(2, 1, [root_hash]);
        let right = change(2, 2, [root_hash]);
        let merge = change(3, 1, [left.hash(), right.hash()]);

        let mut forwards = Dag::new();
        for c in [&root, &left, &right, &merge] {
            forwards.insert(c.clone());
        }
        let mut backwards = Dag::new();
        for c in [&merge, &right, &left, &root] {
            backwards.insert(c.clone());
        }

        assert_eq!(forwards.linearize(), backwards.linearize());
    }

    /// Concurrent siblings are ordered by stamp, so the lower lamport comes
    /// first regardless of hash.
    #[test]
    fn concurrent_changes_are_ordered_by_stamp() {
        let root = change(1, 1, []);
        let root_hash = root.hash();
        let earlier = change(2, 9, [root_hash]);
        let later = change(5, 1, [root_hash]);

        let mut dag = Dag::new();
        for c in [&root, &later, &earlier] {
            dag.insert(c.clone());
        }
        let order = dag.linearize();
        assert!(position(&order, earlier.hash()) < position(&order, later.hash()));
    }

    #[test]
    fn a_parent_that_is_not_held_is_reported_and_does_not_stall_replay() {
        let absent = ChangeHash::from_bytes([0xEE; 32]);
        let mut dag = Dag::new();
        let orphan = dag.insert(change(4, 1, [absent]));

        assert_eq!(
            dag.missing_parents(),
            [absent].into_iter().collect::<BTreeSet<_>>()
        );
        assert_eq!(dag.linearize(), vec![orphan]);
    }

    #[test]
    fn a_complete_history_reports_no_missing_parents() {
        let mut dag = Dag::new();
        let root = dag.insert(change(1, 1, []));
        dag.insert(change(2, 1, [root]));
        assert!(dag.missing_parents().is_empty());
    }
}