blockworx/log/

fold.rs

//! Replay: the one and only writer of document state.
//!
//! Deliberately dumb. It reads nothing but the command in front of it and the
//! register that command names — no I/O, no clock, no randomness, and no
//! re-running of the logic that produced the command. Anything else would make
//! two replicas' replays able to disagree, and every guarantee in this module
//! tree rests on them agreeing.

use crate::log::change::Change;
use crate::log::command::{Command, PropTag, PropWrite};
use crate::log::id::{ElementId, WriteOrder};
use crate::log::state::{DocState, Liveness};

/// A log that cannot be replayed. Both variants mean the caller broke a
/// precondition rather than that the document is odd: changes must arrive
/// after their parents, and identities are never reused.
#[derive(Clone, PartialEq, Eq, Debug, thiserror::Error)]
pub enum FoldError {
    /// A command named an element whose create has not been applied. Causal
    /// delivery makes this unreachable — an author cannot reference what they
    /// had not seen — so it means changes were fed in out of order. Reported
    /// rather than skipped: a skipped command forks the state silently.
    #[error("command {index} names element {id}, which has not been created")]
    UnknownElement { id: ElementId, index: usize },
}

/// Apply one change. Every command of the change shares its stamp and is
/// separated only by its batch index.
pub fn apply(state: &mut DocState, change: &Change) -> Result<(), FoldError> {
    for (index, command) in change.commands().iter().enumerate() {
        apply_command(state, command, change.order_of(index), index)?;
    }
    Ok(())
}

/// Apply a whole sequence, which must already be in causal order.
pub fn fold<'a>(
    state: &mut DocState,
    changes: impl IntoIterator<Item = &'a Change>,
) -> Result<(), FoldError> {
    for change in changes {
        apply(state, change)?;
    }
    Ok(())
}

fn apply_command(
    state: &mut DocState,
    command: &Command,
    order: WriteOrder,
    index: usize,
) -> Result<(), FoldError> {
    let missing = |id| FoldError::UnknownElement { id, index };
    match command {
        Command::Create {
            id,
            kind,
            parent,
            init,
        } => {
            state.ensure(*id, *kind, order);
            for write in init.iter() {
                // The `parent` field is the authority, so anything the init
                // set for that register is overwritten below rather than
                // racing it at equal order.
                if write.tag() != PropTag::Parent {
                    state.write(*id, write.tag(), Some(write.clone()), order);
                }
            }
            state.write(
                *id,
                PropTag::Parent,
                Some(PropWrite::Parent(*parent)),
                order,
            );
            state.set_liveness(*id, Liveness::Live, order);
        }
        Command::SetProp { id, change } => {
            state
                .write(*id, change.tag(), change.new_value(), order)
                .ok_or_else(|| missing(*id))?;
        }
        Command::Delete { id } => {
            state
                .set_liveness(*id, Liveness::Deleted, order)
                .ok_or_else(|| missing(*id))?;
        }
        Command::Restore { id } => {
            state
                .set_liveness(*id, Liveness::Live, order)
                .ok_or_else(|| missing(*id))?;
        }
    }
    Ok(())
}

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

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

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

    fn change(lamport: u64, actor_byte: u8, commands: Vec<Command>) -> Change {
        Change::new(
            Stamp {
                lamport: Lamport::new(lamport),
                actor: actor(actor_byte),
            },
            [],
            SemanticLabel::new("test"),
            commands,
        )
    }

    fn create(id: ElementId) -> Command {
        Command::Create {
            id,
            kind: ElementKind::Block,
            parent: ElementId::DOCUMENT,
            init: PropSet::new(),
        }
    }

    fn set_accent(id: ElementId, from: Option<u8>, to: u8) -> Command {
        Command::SetProp {
            id,
            change: PropChange::Accent {
                old: from,
                new: Some(to),
            },
        }
    }

    fn folded(changes: &[Change]) -> DocState {
        let mut state = DocState::new();
        fold(&mut state, changes).expect("the fixture is in causal order");
        state
    }

    #[test]
    fn a_create_seeds_the_parent_register_so_a_move_is_an_ordinary_write() {
        let id = element(1);
        let state = folded(&[change(1, 1, vec![create(id)])]);
        assert_eq!(
            state.get(id, PropTag::Parent),
            Some(&PropWrite::Parent(ElementId::DOCUMENT))
        );
    }

    /// The create's own `parent` field is what the element's parent is; an
    /// init write for the same register cannot quietly win the tie.
    #[test]
    fn the_create_field_beats_a_parent_in_the_init_set() {
        let id = element(1);
        let state = folded(&[change(
            1,
            1,
            vec![Command::Create {
                id,
                kind: ElementKind::Block,
                parent: ElementId::DOCUMENT,
                init: PropSet::new().with(PropWrite::Parent(element(9))),
            }],
        )]);
        assert_eq!(
            state.get(id, PropTag::Parent),
            Some(&PropWrite::Parent(ElementId::DOCUMENT))
        );
    }

    /// Concurrent writes to one register, delivered both ways round. This is
    /// the property the whole design exists to have.
    #[test]
    fn concurrent_writes_converge_whichever_order_they_arrive() {
        let id = element(1);
        let seed = change(1, 0, vec![create(id)]);
        let alice = change(5, 1, vec![set_accent(id, None, 1)]);
        let bob = change(5, 2, vec![set_accent(id, None, 2)]);

        let one = folded(&[seed.clone(), alice.clone(), bob.clone()]);
        let other = folded(&[seed, bob, alice]);

        assert_eq!(one, other);
        assert_eq!(
            one.get(id, PropTag::Accent),
            Some(&PropWrite::Accent(2)),
            "equal lamports break toward the higher actor id"
        );
    }

    #[test]
    fn delivering_the_same_change_twice_changes_nothing() {
        let id = element(1);
        let seed = change(1, 1, vec![create(id)]);
        let edit = change(2, 1, vec![set_accent(id, None, 4)]);

        let once = folded(&[seed.clone(), edit.clone()]);
        let twice = folded(&[seed.clone(), edit.clone(), edit, seed]);
        assert_eq!(once, twice);
    }

    /// A change that writes the same register twice is ordered only by batch
    /// index, so the second write must land.
    #[test]
    fn a_batch_writing_one_register_twice_ends_on_its_last_write() {
        let id = element(1);
        let state = folded(&[change(
            1,
            1,
            vec![
                create(id),
                set_accent(id, None, 1),
                set_accent(id, Some(1), 2),
            ],
        )]);
        assert_eq!(state.get(id, PropTag::Accent), Some(&PropWrite::Accent(2)));
    }

    #[test]
    fn an_edit_of_an_uncreated_element_is_refused_rather_than_dropped() {
        let mut state = DocState::new();
        let id = element(3);
        let result = apply(&mut state, &change(1, 1, vec![set_accent(id, None, 1)]));
        assert_eq!(result, Err(FoldError::UnknownElement { id, index: 0 }));
    }

    #[test]
    fn deleting_hides_an_element_but_keeps_it_restorable() {
        let id = element(1);
        let mut state = folded(&[change(1, 1, vec![create(id), Command::Delete { id }])]);

        assert!(state.live(id).is_none());
        assert!(state.contains(id), "the tombstone keeps the identity");

        apply(&mut state, &change(2, 1, vec![Command::Restore { id }]))
            .expect("restoring a tombstone");
        assert!(state.live(id).is_some());
    }

    /// Edit-versus-delete resolves by order, not by which arrived first.
    #[test]
    fn an_edit_concurrent_with_a_delete_converges() {
        let id = element(1);
        let seed = change(1, 0, vec![create(id)]);
        let edit = change(4, 1, vec![set_accent(id, None, 5)]);
        let delete = change(6, 2, vec![Command::Delete { id }]);

        let one = folded(&[seed.clone(), edit.clone(), delete.clone()]);
        let other = folded(&[seed, delete, edit]);

        assert_eq!(one, other);
        assert!(
            one.live(id).is_none(),
            "the later-ordered delete decides liveness"
        );
        assert_eq!(
            one.element(id).expect("tombstone").get(PropTag::Accent),
            Some(&PropWrite::Accent(5)),
            "the losing edit is kept, so a restore brings it back with the edit"
        );
    }

    /// Two replicas that saw the same changes in different orders must not
    /// merely be `==` — they must encode identically, since that is what a
    /// snapshot and a sync comparison actually use.
    #[test]
    fn replicas_agree_byte_for_byte_not_just_structurally() {
        let (a, b) = (element(1), element(2));
        let seed = change(1, 0, vec![create(a), create(b)]);
        let alice = change(3, 1, vec![set_accent(a, None, 1)]);
        let bob = change(3, 2, vec![set_accent(b, None, 2)]);

        let one = encode::state_to_bytes(&folded(&[seed.clone(), alice.clone(), bob.clone()]));
        let other = encode::state_to_bytes(&folded(&[seed, bob, alice]));
        assert_eq!(one, other);
    }

    #[test]
    fn replay_is_a_function_of_the_log_alone() {
        let id = element(1);
        let log = [
            change(1, 1, vec![create(id)]),
            change(2, 1, vec![set_accent(id, None, 1)]),
        ];
        assert_eq!(
            encode::state_to_bytes(&folded(&log)),
            encode::state_to_bytes(&folded(&log)),
            "two replays of one log must agree"
        );
    }
}