minerva 0.2.0

//! Property tests for the node store: the kind-read laws and multi-replica
//! convergence through the composer flows. The axis conformance laws
//! (survivor, semilattice, in-place agreement, restriction) run in
//! `causal_store::foreign::laws::axis` beside the other instances.

extern crate alloc;

use alloc::vec::Vec;

use proptest::prelude::*;

use crate::metis::{Anchor, DotSet, Dotted, Kind, Locus, Node};

use super::{Author, TestNode, rank};

mod convergence;
mod reads;

/// One caller move against a fleet of three composers.
#[derive(Clone, Debug)]
enum Op {
    /// Tag block `key` with a kind, superseding the tag dots observed there
    /// (the register discipline on the tag alone).
    Tag { author: usize, key: u8, kind: usize },
    /// Write a register value under block `key`, superseding the values
    /// observed there.
    Value { author: usize, key: u8 },
    /// Weave one sequence element into block `key` (a pure add).
    Weave { author: usize, key: u8 },
    /// The exclusive re-kind: a fresh tag write superseding everything
    /// observed under block `key`, in one covered delta.
    Rekind { author: usize, key: u8, kind: usize },
    /// Observed-remove of everything under block `key`.
    Retract { author: usize, key: u8 },
    /// One anti-entropy step between two replicas.
    Gossip { from: usize, to: usize },
    /// Raw delta transport: re-deliver an arbitrary earlier write's delta to
    /// a replica. Deltas may arrive out of order, duplicated, or not at all
    /// (a never-picked delta is a loss the final anti-entropy must heal), so
    /// tapes reach the outrun states direct shipping creates, a retract or
    /// re-kind landing before the content it supersedes included (the S190
    /// gate-review class).
    Deliver { pick: usize, to: usize },
}

fn kind_of(index: usize) -> Kind {
    match index % 3 {
        0 => Kind::Register,
        1 => Kind::Map,
        _ => Kind::Sequence,
    }
}

fn arb_op() -> impl Strategy<Value = Op> {
    prop_oneof![
        (0usize..3, 0u8..3, 0usize..3).prop_map(|(author, key, kind)| Op::Tag {
            author,
            key,
            kind
        }),
        (0usize..3, 0u8..3).prop_map(|(author, key)| Op::Value { author, key }),
        (0usize..3, 0u8..3).prop_map(|(author, key)| Op::Weave { author, key }),
        (0usize..3, 0u8..3, 0usize..3).prop_map(|(author, key, kind)| Op::Rekind {
            author,
            key,
            kind
        }),
        (0usize..3, 0u8..3).prop_map(|(author, key)| Op::Retract { author, key }),
        (0usize..3, 0usize..3).prop_map(|(from, to)| Op::Gossip { from, to }),
        (0usize..64, 0usize..3).prop_map(|(pick, to)| Op::Deliver { pick, to }),
    ]
}

fn arb_ops() -> impl Strategy<Value = Vec<Op>> {
    prop::collection::vec(arb_op(), 0..32)
}

/// A page delta: `block` under `key`, nothing else.
fn page_with(key: u8, block: TestNode) -> TestNode {
    let mut page = TestNode::new();
    assert!(page.insert_child(key, block));
    page
}

/// Everything currently observed under block `key` at `held`, as the
/// supersede-set of an exclusive write there.
fn observed_under(held: &TestNode, key: u8) -> DotSet {
    held.children()
        .get(&key)
        .map(Node::observed)
        .unwrap_or_default()
}

/// Applies one op to the fleet. `serial` feeds deterministic weave ranks;
/// `wire` is the log of shipped write deltas the `Deliver` transport
/// re-delivers from.
fn apply(fleet: &mut [Author; 3], wire: &mut Vec<(u32, Dotted<TestNode>)>, op: &Op, serial: u64) {
    match *op {
        Op::Tag { author, key, kind } => {
            let (_, delta) = fleet[author].compose_super(
                |dot| {
                    let mut block = TestNode::new();
                    assert!(block.write_tag(dot, kind_of(kind)));
                    page_with(key, block)
                },
                |held| {
                    held.children()
                        .get(&key)
                        .map(|block| block.tag().observed())
                        .unwrap_or_default()
                },
            );
            wire.push((fleet[author].station(), delta));
        }
        Op::Value { author, key } => {
            let (_, delta) = fleet[author].compose_super(
                |dot| {
                    let mut block = TestNode::new();
                    assert!(block.write_register(dot, "v"));
                    page_with(key, block)
                },
                |held| {
                    held.children()
                        .get(&key)
                        .map(|block| block.register().observed())
                        .unwrap_or_default()
                },
            );
            wire.push((fleet[author].station(), delta));
        }
        Op::Weave { author, key } => {
            let (_, delta) = fleet[author].compose(|dot| {
                let mut block = TestNode::new();
                assert!(block.weave(
                    dot,
                    Locus {
                        anchor: Anchor::Origin,
                        rank: rank(serial),
                    }
                ));
                (page_with(key, block), DotSet::new())
            });
            wire.push((fleet[author].station(), delta));
        }
        Op::Rekind { author, key, kind } => {
            let (_, delta) = fleet[author].compose_super(
                |dot| {
                    let mut block = TestNode::new();
                    assert!(block.write_tag(dot, kind_of(kind)));
                    page_with(key, block)
                },
                |held| observed_under(held, key),
            );
            wire.push((fleet[author].station(), delta));
        }
        Op::Retract { author, key } => {
            let superseded = observed_under(fleet[author].state().store(), key);
            let delta = fleet[author].retract(superseded);
            wire.push((fleet[author].station(), delta));
        }
        Op::Gossip { from, to } => {
            if from != to {
                let delta = fleet[from].owed_to(fleet[to].state().context());
                let station = fleet[from].station();
                let _ = fleet[to].absorb(station, &delta);
            }
        }
        Op::Deliver { pick, to } => {
            if !wire.is_empty() {
                let (station, delta) = wire[pick % wire.len()].clone();
                let _ = fleet[to].absorb(station, &delta);
            }
        }
    }
}

/// Runs a tape against a fresh three-replica fleet and returns it.
fn run(ops: &[Op]) -> [Author; 3] {
    let mut fleet = [Author::new(1), Author::new(2), Author::new(3)];
    let mut wire: Vec<(u32, Dotted<TestNode>)> = Vec::new();
    for (index, op) in ops.iter().enumerate() {
        apply(&mut fleet, &mut wire, op, index as u64);
    }
    fleet
}

/// Full anti-entropy: every ordered pair exchanges owed deltas, twice, so
/// every replica holds every write.
fn settle(fleet: &mut [Author; 3]) {
    for _ in 0..2 {
        for from in 0..3 {
            for to in 0..3 {
                if from != to {
                    let delta = fleet[from].owed_to(fleet[to].state().context());
                    let station = fleet[from].station();
                    let _ = fleet[to].absorb(station, &delta);
                }
            }
        }
    }
}