blockworx/doc_ng/

reconcile.rs

//! The reconciliation suite: one [`Host`](crate::doc_ng::session::Host),
//! N [`ClientSession`](crate::doc_ng::session::ClientSession)s, and
//! explicit message queues.
//!
//! Convergence alone proves nothing here: everyone folds one canonical
//! linearization, so it holds however the merge rule compares. What can
//! actually be wrong is the client's optimistic layer — rebuilds against
//! a moving confirmed head, ack/broadcast interleavings, queue discipline
//! — and the merge rule itself, which needs an independent oracle rather
//! than agreement (see `assert_log_semantics`).
//!
//! Interleaving is randomized **across** queues and never within one:
//! per-connection FIFO is the transport's guarantee, and the sessions
//! assert contiguity against it.

#[cfg(test)]
mod tests {
    use std::collections::VecDeque;

    use proptest::prelude::*;
    use uuid::Uuid;

    use crate::doc_ng::{
        block_model::{BlockInit, BlockUpdate, Icon, LabelInit, LabelUpdate},
        commit::Commit,
        document::{Document, TitleBlockUpdate},
        fixtures::{commit, projection},
        geometry::{FracVal, GridPoint, GridRect, GridSize},
        id::BlockId,
        opcode::{Crud, OpCodes},
        rev::{Confirmed, Rev},
        session::{ClientSession, Host, Nonce},
        values::LabelSide,
    };

    const CLIENTS: usize = 3;

    /// What the server sends. Modelled here rather than in a `protocol.rs`
    /// so the suite does not pre-empt phase 3's transport design; the
    /// shapes are the spec's §7 minus `Welcome`.
    #[derive(Debug, Clone)]
    enum Message {
        Committed { nonce: Nonce, rev: Rev },
        Rejected { nonce: Nonce },
        Apply { rev: Rev, commit: Commit },
    }

    struct Client {
        session: ClientSession,
        /// Submitted, in flight toward the host.
        outbox: VecDeque<(Nonce, Commit)>,
        /// Sequenced, in flight toward this client. FIFO, always.
        inbox: VecDeque<Message>,
        /// The harness's own shadow of the session's pending queue. Kept
        /// independently so the oracle never consults the thing it checks.
        pending: VecDeque<(Nonce, Commit)>,
        next_block: u8,
    }

    struct World {
        host: Host,
        clients: Vec<Client>,
    }

    /// Distinct per client, so concurrent creates cannot collide.
    fn block_of(client: usize, nth: u8) -> BlockId {
        let mut bytes = [0u8; 16];
        bytes[0] = u8::try_from(client).expect("few clients");
        bytes[1] = nth;
        // Never the nil uuid, which means "the document" as a parent.
        bytes[2] = 0xb1;
        BlockId::from_uuid(Uuid::from_bytes(bytes))
    }

    fn rect(x: i32) -> GridRect {
        GridRect {
            top_left: GridPoint { x, y: 0 },
            size: GridSize { w: 4, h: 4 },
        }
    }

    fn label_init(name: &str) -> LabelInit {
        LabelInit {
            name: name.into(),
            side: LabelSide::Bottom,
            offset: FracVal::from(1.5),
            hidden: false,
        }
    }

    fn block_init(parent: BlockId, name: &str) -> BlockInit {
        BlockInit {
            parent,
            rect: rect(0),
            locked: false,
            title: label_init(name),
            type_label: label_init("kind"),
            icon: Icon::default(),
        }
    }

    fn create_block(id: BlockId, name: &str) -> OpCodes {
        OpCodes::Block(id, Crud::Create(block_init(BlockId::NULL, name)))
    }

    fn rename(id: BlockId, name: &str) -> OpCodes {
        OpCodes::Block(
            id,
            Crud::Update(BlockUpdate::Title(LabelUpdate::Name(name.into()))),
        )
    }

    fn move_to(id: BlockId, x: i32) -> OpCodes {
        OpCodes::Block(id, Crud::Update(BlockUpdate::Rect(rect(x))))
    }

    fn reparent(child: BlockId, parent: BlockId) -> OpCodes {
        OpCodes::Block(child, Crud::Update(BlockUpdate::Parent(parent)))
    }

    impl World {
        fn new() -> Self {
            Self {
                host: Host::default(),
                clients: (0..CLIENTS)
                    .map(|_| Client {
                        session: ClientSession::default(),
                        outbox: VecDeque::new(),
                        inbox: VecDeque::new(),
                        pending: VecDeque::new(),
                        next_block: 0,
                    })
                    .collect(),
            }
        }

        /// Seal, journal, predict, and put on the wire. Returns false if
        /// the client refused its own edit, which the random pool is
        /// built never to provoke.
        fn edit(&mut self, client: usize, commit: Commit) -> bool {
            let session = &mut self.clients[client].session;
            let Ok(nonce) = session.submit(commit) else {
                return false;
            };
            let outbound = session
                .last_submission()
                .expect("a submission is queued")
                .1
                .clone();
            self.clients[client]
                .outbox
                .push_back((nonce, outbound.clone()));
            self.clients[client].pending.push_back((nonce, outbound));
            true
        }

        /// The host takes this client's oldest in-flight submission.
        fn sequence(&mut self, client: usize) {
            let Some((nonce, commit)) = self.clients[client].outbox.pop_front() else {
                return;
            };
            match self.host.ingest(&commit) {
                Ok(rev) => {
                    self.clients[client]
                        .inbox
                        .push_back(Message::Committed { nonce, rev });
                    for (ix, other) in self.clients.iter_mut().enumerate() {
                        if ix != client {
                            other.inbox.push_back(Message::Apply {
                                rev,
                                commit: commit.clone(),
                            });
                        }
                    }
                }
                Err(_) => self.clients[client]
                    .inbox
                    .push_back(Message::Rejected { nonce }),
            }
        }

        fn deliver(&mut self, client: usize, count: usize) {
            for _ in 0..count {
                let Some(message) = self.clients[client].inbox.pop_front() else {
                    return;
                };
                let client = &mut self.clients[client];
                match message {
                    Message::Committed { nonce, rev } => {
                        client
                            .session
                            .committed(nonce, rev)
                            .expect("an ack for the front of the queue lands");
                        let (shadow, _) = client.pending.pop_front().expect("a pending entry");
                        assert_eq!(shadow, nonce, "answers arrive in submission order");
                    }
                    Message::Rejected { nonce } => {
                        client.session.rejected(nonce).expect("the rejection lands");
                        let (shadow, _) = client.pending.pop_front().expect("a pending entry");
                        assert_eq!(shadow, nonce);
                    }
                    Message::Apply { rev, commit } => {
                        client
                            .session
                            .apply_foreign(rev, &commit)
                            .expect("a sequenced commit lands");
                    }
                }
            }
        }

        fn quiesce(&mut self) {
            for client in 0..CLIENTS {
                while !self.clients[client].outbox.is_empty() {
                    self.sequence(client);
                }
            }
            for client in 0..CLIENTS {
                let waiting = self.clients[client].inbox.len();
                self.deliver(client, waiting);
            }
        }

        /// **The oracle.** Rebuilt from the *host's* log rather than from
        /// the session's incrementally maintained `confirmed`, so a session
        /// that mistracks its head is caught rather than agreed with.
        fn assert_oracle(&self) {
            for (ix, client) in self.clients.iter().enumerate() {
                let confirmed_count =
                    usize::try_from(client.session.rev().get()).expect("the log fits in memory");
                let mut replayed = Document::<Confirmed>::default();
                for commit in &self.host.commits_after(Rev::ZERO)[..confirmed_count] {
                    replayed = replayed
                        .try_apply(commit)
                        .expect("a sequenced commit folds on replay");
                }
                let mut predicted = replayed.predict();
                for (_, commit) in &client.pending {
                    predicted = predicted
                        .try_apply(commit)
                        .expect("the random pool never authors a commit that cannot fold");
                }
                assert_eq!(
                    client.session.optimistic().content_hash(),
                    predicted.content_hash(),
                    "client {ix}: optimistic != confirmed-log ⊕ pending",
                );
            }
        }

        /// **The semantic oracle.** Convergence and the prediction oracle
        /// are both blind to the merge *rule*: everyone folds one canonical
        /// linearization, so `f(x) == f(x)` holds however `Register::apply`
        /// compares. Replaying the log with dumb sequential last-wins
        /// bookkeeping — plain assignment, no `WriteOrder`, no `Register` —
        /// is an independent implementation to disagree with.
        fn assert_log_semantics(&self) {
            let mut title = String::new();
            let mut blocks: std::collections::BTreeMap<BlockId, (GridRect, String, bool)> =
                std::collections::BTreeMap::new();

            for commit in self.host.commits_after(Rev::ZERO) {
                for op in commit.ops() {
                    match op {
                        OpCodes::Document(TitleBlockUpdate::Name(name)) => title = name.clone(),
                        OpCodes::Block(id, Crud::Create(init)) => {
                            blocks.insert(*id, (init.rect, init.title.name.clone(), true));
                        }
                        OpCodes::Block(id, Crud::Update(BlockUpdate::Rect(to))) => {
                            blocks.entry(*id).or_default().0 = *to;
                        }
                        OpCodes::Block(
                            id,
                            Crud::Update(BlockUpdate::Title(LabelUpdate::Name(name))),
                        ) => {
                            blocks.entry(*id).or_default().1 = name.clone();
                        }
                        OpCodes::Block(id, Crud::Delete) => {
                            blocks.entry(*id).or_default().2 = false;
                        }
                        OpCodes::Block(id, Crud::Restore) => {
                            blocks.entry(*id).or_default().2 = true;
                        }
                        _ => panic!("the random pool authored an op this oracle cannot replay"),
                    }
                }
            }

            let expected: Vec<_> = blocks
                .into_iter()
                .filter(|(_, (_, _, alive))| *alive)
                .map(|(id, (rect, name, _))| (id, BlockId::NULL, rect, name))
                .collect();
            assert_eq!(
                projection(self.host.state()),
                expected,
                "the fold disagrees with last-write-wins over the log",
            );
            assert_eq!(
                self.host.state().title_block().name.as_ref(),
                &title,
                "the singleton's register disagrees with the log's last write",
            );
        }

        fn assert_converged(&self) {
            let host = self.host.state().content_hash();
            for (ix, client) in self.clients.iter().enumerate() {
                assert!(client.pending.is_empty(), "client {ix} is not quiesced");
                assert_eq!(
                    client.session.confirmed().content_hash(),
                    host,
                    "client {ix}'s confirmed document differs from the host's",
                );
                assert_eq!(
                    client.session.optimistic().content_hash(),
                    host,
                    "client {ix}'s prediction differs from the host's document",
                );
            }
        }

        fn block_ids(&self, client: usize) -> Vec<BlockId> {
            let mut ids: Vec<_> = self.clients[client]
                .session
                .optimistic()
                .blocks()
                .map(|(id, _)| id)
                .collect();
            ids.sort();
            ids
        }

        fn title_of(&self, client: usize, id: BlockId) -> String {
            self.clients[client]
                .session
                .confirmed()
                .block(&id)
                .expect("the block is in the document")
                .as_ref()
                .title
                .name
                .as_ref()
                .clone()
        }
    }

    #[derive(Debug, Clone, Copy)]
    enum Edit {
        Create,
        Move {
            which: usize,
            x: i32,
        },
        Rename {
            which: usize,
            name: u8,
        },
        Delete {
            which: usize,
        },
        Restore {
            which: usize,
        },
        Title {
            name: u8,
        },
        /// One commit writing one register twice — the intra-commit
        /// ordering case, exercised inside the random drive as well as in
        /// its own scenario.
        MoveTwice {
            which: usize,
            first: i32,
            second: i32,
        },
    }

    #[derive(Debug, Clone, Copy)]
    enum Step {
        Edit { client: usize, edit: Edit },
        Sequence { client: usize },
        Deliver { client: usize, count: usize },
    }

    impl World {
        /// Every edit the pool authors is valid against any state it could
        /// meet: fresh ids, existing targets, and no reparenting — the one
        /// op a foreign commit can invalidate. Rejection is provoked
        /// deliberately in its own scenario instead.
        fn apply_edit(&mut self, client: usize, edit: Edit) {
            let ids = self.block_ids(client);
            let pick = |which: usize| ids[which % ids.len()];
            let ops = match edit {
                Edit::Create => None,
                _ if ids.is_empty() => None,
                Edit::Move { which, x } => Some(vec![move_to(pick(which), x)]),
                Edit::Rename { which, name } => {
                    Some(vec![rename(pick(which), &format!("n{name}"))])
                }
                Edit::Delete { which } => Some(vec![OpCodes::Block(pick(which), Crud::Delete)]),
                Edit::Restore { which } => Some(vec![OpCodes::Block(pick(which), Crud::Restore)]),
                Edit::Title { name } => Some(vec![OpCodes::Document(TitleBlockUpdate::Name(
                    format!("doc{name}"),
                ))]),
                Edit::MoveTwice {
                    which,
                    first,
                    second,
                } => {
                    let id = pick(which);
                    Some(vec![move_to(id, first), move_to(id, second)])
                }
            };

            let ops = ops.unwrap_or_else(|| {
                let nth = self.clients[client].next_block;
                self.clients[client].next_block = nth.wrapping_add(1);
                vec![create_block(block_of(client, nth), "fresh")]
            });
            assert!(
                self.edit(client, commit("edit", ops)),
                "the random pool must never author an edit the client refuses",
            );
        }

        fn step(&mut self, step: Step) {
            match step {
                Step::Edit { client, edit } => self.apply_edit(client, edit),
                Step::Sequence { client } => self.sequence(client),
                Step::Deliver { client, count } => self.deliver(client, count),
            }
        }
    }

    fn any_edit() -> impl Strategy<Value = Edit> {
        prop_oneof![
            2 => Just(Edit::Create),
            3 => (any::<usize>(), -50i32..50).prop_map(|(which, x)| Edit::Move { which, x }),
            3 => (any::<usize>(), any::<u8>()).prop_map(|(which, name)| Edit::Rename { which, name }),
            1 => any::<usize>().prop_map(|which| Edit::Delete { which }),
            1 => any::<usize>().prop_map(|which| Edit::Restore { which }),
            1 => any::<u8>().prop_map(|name| Edit::Title { name }),
            2 => (any::<usize>(), -50i32..50, -50i32..50)
                .prop_map(|(which, first, second)| Edit::MoveTwice { which, first, second }),
        ]
    }

    fn any_step() -> impl Strategy<Value = Step> {
        prop_oneof![
            3 => (0..CLIENTS, any_edit()).prop_map(|(client, edit)| Step::Edit { client, edit }),
            3 => (0..CLIENTS).prop_map(|client| Step::Sequence { client }),
            3 => (0..CLIENTS, 0usize..4).prop_map(|(client, count)| Step::Deliver { client, count }),
        ]
    }

    proptest! {
        /// The two properties that matter, checked together: the oracle
        /// after **every** step (a prediction is never allowed to drift,
        /// not merely to end up right), and byte-identical convergence
        /// once the wires drain.
        #[test]
        fn predictions_reconcile_and_everyone_converges(steps in prop::collection::vec(any_step(), 1..80)) {
            let mut world = World::new();
            for step in steps {
                world.step(step);
                world.assert_oracle();
            }
            world.quiesce();
            world.assert_oracle();
            world.assert_converged();
            world.assert_log_semantics();
        }
    }

    /// A semantic oracle: not "they agree" but *which value won*. A fold
    /// that dropped every write would agree with itself perfectly.
    #[test]
    fn the_later_commit_in_server_order_wins_the_register_everywhere() {
        let mut world = World::new();
        let id = block_of(0, 0);
        world.edit(0, commit("create", vec![create_block(id, "original")]));
        world.quiesce();

        // Both clients rename the same register, neither having seen the
        // other — the concurrent write the whole design is about.
        world.edit(0, commit("rename", vec![rename(id, "from client 0")]));
        world.edit(1, commit("rename", vec![rename(id, "from client 1")]));
        assert_eq!(
            world.title_of(0, id),
            "original",
            "neither rename is confirmed yet",
        );

        // The host takes client 1's first, so client 0's is the later rev.
        world.sequence(1);
        world.sequence(0);
        world.quiesce();

        for client in 0..CLIENTS {
            assert_eq!(
                world.title_of(client, id),
                "from client 0",
                "client {client} must show the value the later rev wrote",
            );
        }
        world.assert_converged();
    }

    /// The losing client's own value stands in its prediction until the
    /// winning commit is confirmed — the Figma suppression rule, which is
    /// the reason unacked writes sort above confirmed ones.
    #[test]
    fn an_unacked_local_write_outranks_an_incoming_confirmed_one() {
        let mut world = World::new();
        let id = block_of(0, 0);
        world.edit(0, commit("create", vec![create_block(id, "original")]));
        world.quiesce();

        world.edit(1, commit("rename", vec![rename(id, "local")]));
        world.edit(0, commit("rename", vec![rename(id, "remote")]));
        world.sequence(0);
        world.deliver(1, 1);

        assert_eq!(
            world.clients[1]
                .session
                .optimistic()
                .block(&id)
                .expect("the block is present")
                .as_ref()
                .title
                .name
                .as_ref()
                .as_str(),
            "local",
            "the unacked local edit still suppresses the confirmed one",
        );
        assert_eq!(
            world.title_of(1, id),
            "remote",
            "while the confirmed document has already taken the foreign value",
        );

        world.quiesce();
        world.assert_converged();
    }

    /// Intra-commit ordering, end to end through a session: `seq` resolves
    /// a commit that writes one register twice, and every client sees the
    /// last write.
    #[test]
    fn a_commit_writing_one_register_twice_ends_on_its_last_write() {
        let mut world = World::new();
        let id = block_of(0, 0);
        world.edit(0, commit("create", vec![create_block(id, "b")]));
        world.quiesce();

        world.edit(0, commit("drag", vec![move_to(id, 5), move_to(id, 9)]));
        world.quiesce();

        for client in 0..CLIENTS {
            assert_eq!(
                *world.clients[client]
                    .session
                    .confirmed()
                    .block(&id)
                    .expect("the block is present")
                    .as_ref()
                    .rect
                    .as_ref(),
                rect(9),
                "client {client} must end on the commit's last write",
            );
        }
    }

    /// Re-delivery is refused by the contiguity check, not absorbed by the
    /// fold — and refused *loudly*, because a repeated rev means the
    /// transport broke a guarantee the sessions are built on.
    ///
    /// The rest of idempotence lives elsewhere: a re-folded commit mints a
    /// *different* rev, so fold-level re-apply is not idempotent by design,
    /// and the register keeps the original guarantee (an equal write order
    /// loses, `register.rs`).
    #[test]
    #[should_panic(expected = "gap in the per-connection rev sequence")]
    fn re_delivering_a_confirmed_commit_is_refused() {
        let mut world = World::new();
        let id = block_of(0, 0);
        world.edit(0, commit("create", vec![create_block(id, "b")]));
        world.quiesce();

        let replayed = world.host.commits_after(Rev::ZERO)[0].clone();
        let stale = world.clients[1].session.rev();
        let _ = world.clients[1].session.apply_foreign(stale, &replayed);
    }

    /// Rejection safety. The cycle can only be seen by the host: each
    /// client's reparent is valid against what it holds, and the second to
    /// be sequenced closes the loop.
    #[test]
    fn a_commit_the_host_refuses_is_dropped_and_the_client_re_converges() {
        let mut world = World::new();
        let (first, second) = (block_of(0, 0), block_of(0, 1));
        world.edit(
            0,
            commit(
                "two blocks",
                vec![create_block(first, "a"), create_block(second, "b")],
            ),
        );
        world.quiesce();

        world.edit(0, commit("nest a under b", vec![reparent(first, second)]));
        world.edit(1, commit("nest b under a", vec![reparent(second, first)]));

        world.sequence(1);
        let host_after_first = world.host.state().content_hash();
        world.sequence(0);
        assert_eq!(
            world.host.state().content_hash(),
            host_after_first,
            "the refused commit must leave the host untouched",
        );
        assert_eq!(
            world.host.rev(),
            Rev::new(2),
            "and must consume no rev — the log cannot gap",
        );

        world.quiesce();
        world.assert_converged();
        assert_eq!(
            *world.clients[0]
                .session
                .confirmed()
                .block(&first)
                .expect("the block is present")
                .as_ref()
                .parent
                .as_ref(),
            BlockId::NULL,
            "client 0's refused reparent is gone from its document",
        );
    }

    /// The undo round trip through a session, over the cases the journal's
    /// rules turn on. The edit is asserted **observable** first, or an
    /// inverse that restored nothing would pass vacuously.
    #[test]
    fn undo_restores_the_visible_document_through_the_wire() {
        let id = block_of(0, 0);
        let created = block_of(0, 1);
        for edit in [
            commit("move", vec![move_to(id, 7)]),
            commit("rename", vec![rename(id, "renamed")]),
            commit("delete", vec![OpCodes::Block(id, Crud::Delete)]),
            commit(
                "create",
                vec![create_block(created, "new"), move_to(created, 3)],
            ),
            commit("drag", vec![move_to(id, 4), move_to(id, 8)]),
            commit(
                "multi",
                vec![
                    move_to(id, 2),
                    rename(id, "multi"),
                    create_block(created, "c"),
                ],
            ),
        ] {
            let mut world = World::new();
            world.edit(0, commit("create", vec![create_block(id, "original")]));
            world.quiesce();
            let before = projection(world.clients[0].session.confirmed());

            let label = edit.label().to_owned();
            world.edit(0, edit);
            world.quiesce();
            assert_ne!(
                projection(world.clients[0].session.confirmed()),
                before,
                "{label}: the edit must be observable or the round trip proves nothing",
            );

            let undone = world.clients[0]
                .session
                .undo()
                .expect("something to undo")
                .expect("the inverse folds");
            let outbound = world.clients[0]
                .session
                .last_submission()
                .expect("the undo is queued")
                .1
                .clone();
            world.clients[0]
                .outbox
                .push_back((undone, outbound.clone()));
            world.clients[0].pending.push_back((undone, outbound));
            world.quiesce();

            assert_eq!(
                projection(world.clients[0].session.confirmed()),
                before,
                "{label}: undo must restore the visible document",
            );
            world.assert_converged();
        }
    }

    /// A tombstoned block is restorable, and the restore travels: delete
    /// and restore are inverses through the wire, not only in the fold.
    #[test]
    fn a_delete_and_its_restore_travel_to_every_client() {
        let mut world = World::new();
        let id = block_of(0, 0);
        world.edit(0, commit("create", vec![create_block(id, "b")]));
        world.quiesce();

        world.edit(0, commit("delete", vec![OpCodes::Block(id, Crud::Delete)]));
        world.quiesce();
        for client in 0..CLIENTS {
            assert!(
                projection(world.clients[client].session.confirmed()).is_empty(),
                "client {client} must not show a tombstoned block",
            );
        }

        world.edit(
            1,
            commit("restore", vec![OpCodes::Block(id, Crud::Restore)]),
        );
        world.quiesce();
        for client in 0..CLIENTS {
            assert_eq!(
                world.title_of(client, id),
                "b",
                "client {client} must see the restored block, inner values intact",
            );
        }
        world.assert_converged();
    }
}