path: root/src/access/screen.rs

//! Received changes judged by rules by path before they are applied (`rules::judge`),
//! and applied if allowed (`apply`): a refused change stays out, and so does whatever
//! builds on it; it is kept aside for an operator to look at (`Refused`).
//!
//! A change is judged at the heads its writer saw. Those of a batch aren't in the
//! document yet, so one is judged at the changes of the document its batch ancestors
//! build on: what rules read there (the members, the user registry, remake proposals,
//! the policy, bans) is what its writer saw, unless one of those ancestors changed it. So a change that
//! does (`relevant`) is applied before the next is judged: the rest go in at once.
//!
//! A change whose dependencies are neither in the document nor among those received
//! waits (`Waiting`) until they come, a bounded number; Automerge never sees it before.
//! With signatures required (`access::policy`), so does one no signature by its
//! committer vouches for yet (`Signatures::vouch_batch`), and whatever builds on it,
//! until one comes, but an epoch's root, which no node signs.

use std::collections::{BTreeSet, HashMap, HashSet};

use anyhow::Result;

use super::rules::{self, Refusal};
use crate::document::inspect::{Inspector, Op, Path};
use crate::document::meta;
use crate::document::replica::{Epoch, user_of};
use crate::document::signatures::Signatures;
use crate::document::{Change, ChangeHash, Document, ReadDoc};

/// Changes waiting for their dependencies, kept at most.
const MAX_WAITING: usize = 10_000;
/// Refused changes kept for operators, at most: the oldest go.
pub(crate) const MAX_REFUSED: usize = 1_000;

/// A refused change, and why.
#[derive(Debug, Clone)]
pub(crate) struct Refused {
    pub(crate) change: Change,
    pub(crate) committer: Option<String>,
    pub(crate) user: Option<String>,
    pub(crate) paths: BTreeSet<Path>,
    pub(crate) why: String,
}

/// What screening keeps between batches, for one document (`AppState::screening`): the
/// changes waiting for their dependencies, and those refused, for operators
/// (`GET /admin/refused`).
#[derive(Debug, Default)]
pub(crate) struct Screening {
    /// The document's (`Replica::epoch_seq`): another's start empty (`of`).
    epoch_seq: u32,
    pub(crate) waiting: Waiting,
    refused: std::collections::VecDeque<Refused>,
}

impl Screening {
    /// Itself for the document `epoch_seq`: emptied if it was another's.
    pub(crate) fn of(&mut self, epoch_seq: u32) -> &mut Self {
        if self.epoch_seq != epoch_seq {
            *self = Screening {
                epoch_seq,
                ..Screening::default()
            };
        }
        self
    }

    /// Keeps `refused`, logged, the oldest going past `MAX_REFUSED`.
    pub(crate) fn record(&mut self, refused: Vec<Refused>) {
        for refused in refused {
            tracing::warn!(
                change = %refused.change.hash(),
                committer = refused.committer.as_deref().unwrap_or("none"),
                user = refused.user.as_deref().unwrap_or("none"),
                why = %refused.why,
                "Refused a change (_continuum/access/policy)"
            );
            self.refused.push_back(refused);
            if self.refused.len() > MAX_REFUSED {
                self.refused.pop_front();
            }
        }
    }

    /// Those refused, oldest first.
    pub(crate) fn refused(&self) -> impl DoubleEndedIterator<Item = &Refused> {
        self.refused.iter()
    }
}

/// Received changes waiting for their dependencies.
#[derive(Debug, Default)]
pub(crate) struct Waiting(Vec<Change>);

impl Waiting {
    pub(crate) fn len(&self) -> usize {
        self.0.len()
    }
}

/// What `apply` did: how many changes it applied, and those it refused.
#[derive(Debug, Default)]
pub(crate) struct Screened {
    pub(crate) applied: usize,
    pub(crate) refused: Vec<Refused>,
}

/// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`)
/// those rules allow, what they change recorded in `log`. Those already in `doc` are
/// left out; those whose dependencies are missing wait.
/// With `signatures` (required), those no signature vouches for wait too.
pub(crate) fn apply(
    doc: &mut Document,
    epoch: Epoch,
    received: Vec<Change>,
    waiting: &mut Waiting,
    mut signatures: Option<&mut Signatures>,
    log: &mut crate::document::PatchLog,
) -> Result<Screened> {
    let mut screened = Screened::default();
    // Each once, not in the document yet.
    let mut seen = HashSet::new();
    let candidates: Vec<Change> = std::mem::take(&mut waiting.0)
        .into_iter()
        .chain(received)
        .filter(|change| seen.insert(change.hash()))
        .filter(|change| doc.get_change_by_hash(&change.hash()).is_none())
        .collect();
    let (ready, unready) = in_order(doc, candidates);
    waiting.0 = unready;
    if ready.is_empty() {
        waiting.0.truncate(MAX_WAITING);
        return Ok(screened);
    }

    let decoded: Vec<crate::document::ExpandedChange> =
        ready.iter().map(|change| change.decode()).collect();
    let ops: Vec<Vec<Op>> = Inspector::new(doc).ops(&decoded);
    let paths: Vec<BTreeSet<Path>> = ops
        .iter()
        .map(|ops| ops.iter().map(|op| op.path.clone()).collect())
        .collect();
    // Who made each: its author, or that of its actor's first change (in the document,
    // or among these).
    let mut authors: HashMap<crate::document::ActorId, String> = HashMap::new();
    let committers: Vec<Option<String>> = ready
        .iter()
        .map(|change| {
            let actor = change.actor_id();
            let named = change
                .author()
                .or_else(|| doc.get_author_for_actor(actor))
                .and_then(|author| {
                    std::str::from_utf8(author.as_bytes())
                        .ok()
                        .map(str::to_owned)
                })
                .or_else(|| authors.get(actor).cloned());
            if let Some(named) = &named {
                authors.insert(actor.clone(), named.clone());
            }
            named
        })
        .collect();
    if let Some(signatures) = signatures.as_deref_mut() {
        signatures.vouch_batch(&ready, &committers);
    }

    // The changes of the document each accepted one, not applied yet, builds on.
    let mut frontier: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();
    let mut refused: HashSet<ChangeHash> = HashSet::new();
    let mut unsigned: HashSet<ChangeHash> = HashSet::new();
    let mut accepted: Vec<Change> = Vec::new();
    for (index, change) in ready.iter().enumerate() {
        let hash = change.hash();
        let committer = committers[index].as_deref();
        let user = change.message().and_then(user_of).map(str::to_owned);
        let refuse = |why: String| Refused {
            change: change.clone(),
            committer: committer.map(str::to_owned),
            user: user.clone(),
            paths: paths[index].clone(),
            why,
        };
        if let Some(dep) = change.deps().iter().find(|dep| refused.contains(dep)) {
            refused.insert(hash);
            screened
                .refused
                .push(refuse(format!("it builds on refused change {dep}")));
            continue;
        }
        // No signature vouches for it yet, or for what it builds on: it waits.
        let vouched = |signatures: &Signatures| {
            change.deps().is_empty() || signatures.verified(change.actor_id(), change.seq())
        };
        if signatures
            .as_deref()
            .is_some_and(|signatures| !vouched(signatures))
            || change.deps().iter().any(|dep| unsigned.contains(dep))
        {
            unsigned.insert(hash);
            waiting.0.push(change.clone());
            continue;
        }
        let mut seen_at: BTreeSet<ChangeHash> = BTreeSet::new();
        for dep in change.deps() {
            match frontier.get(dep) {
                Some(heads) => seen_at.extend(heads.iter().copied()),
                None => {
                    seen_at.insert(*dep);
                }
            }
        }
        let seen_at: Vec<ChangeHash> = seen_at.into_iter().collect();
        let judged = rules::judge(
            doc,
            epoch,
            &rules::Change {
                hash,
                deps: change.deps(),
                seen: &seen_at,
                committer,
                user: user.as_deref(),
                paths: &paths[index],
                ops: &ops[index],
            },
        );
        match judged {
            Ok(()) => {
                accepted.push(change.clone());
                if relevant(&paths[index]) {
                    // What rules read changed: in before the next is judged.
                    screened.applied += accepted.len();
                    doc.apply_changes_log_patches(std::mem::take(&mut accepted), log)?;
                    frontier.clear();
                } else {
                    frontier.insert(hash, seen_at);
                }
            }
            Err(Refusal::Unready) => {
                // Can't be: its dependencies are in the document or among these.
                refused.insert(hash);
                screened.refused.push(refuse(Refusal::Unready.to_string()));
            }
            Err(Refusal::Rule(why)) => {
                refused.insert(hash);
                screened.refused.push(refuse(why));
            }
        }
    }
    screened.applied += accepted.len();
    if !accepted.is_empty() {
        doc.apply_changes_log_patches(accepted, log)?;
    }
    waiting.0.truncate(MAX_WAITING);
    Ok(screened)
}

/// [`apply`] of the changes in `bytes`, as sync messages and downloads carry them, with
/// what `screening` keeps for document `epoch_seq` (of `epoch`). Bytes that don't load
/// as changes, or with a misplaced author (`changes::misplaced_author`), are refused
/// whole.
/// With `signatures` (required), those no signature vouches for wait too; and with no
/// bytes, those waiting are judged again (signatures came since).
pub(crate) fn apply_bytes(
    doc: &mut Document,
    epoch: Epoch,
    epoch_seq: u32,
    bytes: &[u8],
    screening: &std::sync::Mutex<Screening>,
    signatures: Option<&mut Signatures>,
    log: &mut crate::document::PatchLog,
) -> Result<usize> {
    use crate::locks::MutexExt;
    if bytes.is_empty() && signatures.is_none() {
        return Ok(0);
    }
    if let Some(misplaced) = crate::document::changes::misplaced_author(bytes) {
        anyhow::bail!("Refused changes: {misplaced}");
    }
    let changes = crate::document::changes::changes_in(bytes)
        .ok_or_else(|| anyhow::anyhow!("Refused changes that don't load"))?;
    let mut screening = screening.lock_or_recover();
    let screening = screening.of(epoch_seq);
    if changes.is_empty() && screening.waiting.len() == 0 {
        return Ok(0);
    }
    let screened = apply(doc, epoch, changes, &mut screening.waiting, signatures, log)?;
    let applied = screened.applied;
    screening.record(screened.refused);
    Ok(applied)
}

/// Whether a change touching `paths` changes what rules read: the members, the user
/// registry, remake records, the policy and the rest of `_continuum/access`, bans, the
/// protocol data as a whole.
fn relevant(paths: &BTreeSet<Path>) -> bool {
    paths.iter().any(|path| match path.as_slice() {
        [] => true,
        [root, ..] if root != meta::ROOT => false,
        [_] => true,
        [_, part, ..] => [
            meta::MEMBERS,
            meta::USERS,
            meta::REMAKES,
            meta::ACCESS,
            meta::BANNED,
        ]
        .contains(&part.as_str()),
    })
}

/// `candidates` in an order where every change comes after its dependencies among them
/// (ready), and those with a dependency neither in `doc` nor among them, nor ready
/// (unready).
fn in_order(doc: &Document, candidates: Vec<Change>) -> (Vec<Change>, Vec<Change>) {
    let mut by_hash: HashMap<ChangeHash, Change> = candidates
        .into_iter()
        .map(|change| (change.hash(), change))
        .collect();
    let mut ready = Vec::new();
    let mut placed: HashSet<ChangeHash> = HashSet::new();
    loop {
        let mut progress = false;
        let mut hashes: Vec<ChangeHash> = by_hash.keys().copied().collect();
        // The same order on every node, whatever order they came in.
        hashes.sort_unstable();
        for hash in hashes {
            let change = &by_hash[&hash];
            if change
                .deps()
                .iter()
                .all(|dep| placed.contains(dep) || doc.get_change_by_hash(dep).is_some())
            {
                placed.insert(hash);
                ready.push(by_hash.remove(&hash).expect("there"));
                progress = true;
            }
        }
        if !progress {
            break;
        }
    }
    (ready, by_hash.into_values().collect())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::access::Right;
    use crate::access::registry;
    use crate::clock::UnixMillis;
    use crate::document::replica::{Writer, transact};
    use crate::document::store;
    use serde_json::json;

    fn path(text: &str) -> Path {
        text.split('/').map(str::to_owned).collect()
    }

    fn write(
        doc: &mut Document,
        node: &str,
        user: Option<&str>,
        at: &str,
        value: serde_json::Value,
    ) {
        let writer = Writer { node, user };
        let at = path(at);
        transact(doc, UnixMillis(1), &writer, |doc| {
            store::write_path(doc, &at, &value)
        })
        .unwrap();
    }

    /// The changes of `doc` after `since`, as received.
    fn received(doc: &Document, since: &[ChangeHash]) -> Vec<Change> {
        doc.get_changes(since).into_iter().collect()
    }

    fn screen(doc: &mut Document, changes: Vec<Change>, waiting: &mut Waiting) -> Screened {
        apply(
            doc,
            Epoch::GENESIS,
            changes,
            waiting,
            None,
            &mut crate::document::PatchLog::inactive(),
        )
        .unwrap()
    }

    /// A swarm whose genesis registers member `a` and user `alice`, who may write
    /// `notes`; written on `theirs`, received by an empty document in one batch.
    fn genesis() -> Document {
        let mut theirs = Document::new();
        let alice = registry::entry("a", 1, &[(path("notes"), &[Right::Write][..])]);
        let writer = Writer {
            node: "a",
            user: None,
        };
        transact(&mut theirs, UnixMillis(1), &writer, |doc| {
            store::write_path(doc, &path("_continuum/swarm"), &json!("s"))?;
            store::write_path(doc, &path("_continuum/users/alice"), &alice)
        })
        .unwrap();
        write(&mut theirs, "a", None, "_continuum/members/a", json!({}));
        theirs
    }

    /// A batch from the genesis on: honest changes applied, a refused one left out with
    /// what builds on it, the rest in.
    #[test]
    fn a_batch_judged_change_by_change() {
        let mut theirs = genesis();
        write(&mut theirs, "a", Some("alice"), "notes/x", json!(1));
        let before_bad = theirs.get_heads();
        write(&mut theirs, "a", Some("alice"), "config/x", json!(2));
        write(&mut theirs, "a", Some("alice"), "notes/y", json!(3));
        // Concurrent with the refused one: in.
        let mut side = theirs.fork_at(&before_bad).unwrap();
        write(&mut side, "a", Some("alice"), "notes/z", json!(4));
        theirs.merge(&mut side).unwrap();

        let mut ours = Document::new();
        let mut waiting = Waiting::default();
        let screened = screen(&mut ours, received(&theirs, &[]), &mut waiting);
        assert_eq!(screened.applied, 4);
        let why: Vec<&str> = screened.refused.iter().map(|r| r.why.as_str()).collect();
        assert_eq!(why.len(), 2, "{why:?}");
        assert!(why[0].contains("may not write /config"), "{why:?}");
        assert!(why[1].contains("builds on refused change"), "{why:?}");
        assert_eq!(
            store::read_path(&ours, &path("notes/x")).unwrap(),
            Some(json!(1))
        );
        assert_eq!(
            store::read_path(&ours, &path("notes/z")).unwrap(),
            Some(json!(4))
        );
        assert_eq!(store::read_path(&ours, &path("config")).unwrap(), None);
        assert!(waiting.len() == 0);
    }

    /// Judged as its writer saw the registry: a grant earlier in the same batch counts.
    #[test]
    fn a_grant_in_the_same_batch_counts() {
        let mut theirs = genesis();
        let bob = registry::entry("a", 1, &[(path("config"), &[Right::Write][..])]);
        write(&mut theirs, "a", Some("alice"), "_continuum/users/bob", bob);
        write(&mut theirs, "a", Some("bob"), "config/x", json!(1));
        let mut ours = Document::new();
        let screened = screen(&mut ours, received(&theirs, &[]), &mut Waiting::default());
        // Alice may not grant (no write on the registry): both refused.
        assert_eq!(screened.refused.len(), 2);

        let mut theirs = genesis();
        let root = registry::entry("a", 1, &registry::all_caps());
        transact(
            &mut theirs,
            UnixMillis(1),
            &Writer {
                node: "a",
                user: None,
            },
            |doc| store::write_path(doc, &path("_continuum/users/root"), &root),
        )
        .unwrap();
        // Written as a member without a user: refused, and so is what builds on it.
        let mut ours = Document::new();
        let screened = screen(&mut ours, received(&theirs, &[]), &mut Waiting::default());
        assert_eq!(screened.refused.len(), 1);
    }

    /// Changes whose dependencies haven't come wait, and are judged once they do.
    #[test]
    fn changes_wait_for_their_dependencies() {
        let mut theirs = genesis();
        write(&mut theirs, "a", Some("alice"), "notes/x", json!(1));
        let middle = theirs.get_heads();
        write(&mut theirs, "a", Some("alice"), "notes/y", json!(2));
        let mut ours = Document::new();
        let mut waiting = Waiting::default();
        // The last first: it waits.
        let last = received(&theirs, &middle);
        let screened = screen(&mut ours, last, &mut waiting);
        assert_eq!(screened.applied, 0);
        assert_eq!(waiting.len(), 1);
        // The rest: all in, the waiting one too.
        let screened = screen(&mut ours, received(&theirs, &[]), &mut waiting);
        assert_eq!(screened.applied, 4);
        assert!(screened.refused.is_empty());
        assert_eq!(waiting.len(), 0);
        assert_eq!(
            store::read_path(&ours, &path("notes/y")).unwrap(),
            Some(json!(2))
        );
    }

    /// With signatures required, changes no signature vouches for wait, with what
    /// builds on them, until one comes; one signature vouches for its actor's earlier
    /// changes too.
    #[test]
    fn unsigned_changes_wait_for_a_signature() {
        let key = crate::identity::NodeKey::from_seed([3; 32]);
        let node = key.id();
        let mut theirs = Document::new();
        let alice = registry::entry(&node, 1, &[(path("notes"), &[Right::Write][..])]);
        transact(
            &mut theirs,
            UnixMillis(1),
            &Writer {
                node: &node,
                user: None,
            },
            |doc| {
                store::write_path(doc, &path("_continuum/swarm"), &json!("s"))?;
                store::write_path(doc, &path("_continuum/users/alice"), &alice)
            },
        )
        .unwrap();
        let root = theirs.get_heads();
        write(
            &mut theirs,
            &node,
            None,
            &format!("_continuum/members/{node}"),
            json!({}),
        );
        write(&mut theirs, &node, Some("alice"), "notes/x", json!(1));
        let last = theirs.get_heads()[0];

        let mut ours = Document::new();
        let mut waiting = Waiting::default();
        let mut signatures = Signatures::default();
        let required =
            |doc: &mut Document, waiting: &mut Waiting, signatures: &mut Signatures, changes| {
                apply(
                    doc,
                    Epoch::GENESIS,
                    changes,
                    waiting,
                    Some(signatures),
                    &mut crate::document::PatchLog::inactive(),
                )
                .unwrap()
            };
        // Unsigned: only the root goes in (no node signs it).
        let screened = required(
            &mut ours,
            &mut waiting,
            &mut signatures,
            received(&theirs, &[]),
        );
        assert_eq!(screened.applied, 1);
        assert_eq!(ours.get_heads(), root);
        assert_eq!(waiting.len(), 2);
        // A forged signature vouches for nothing.
        let forged = crate::identity::NodeKey::from_seed([4; 32]);
        signatures.offer(
            &ours,
            vec![(
                last,
                forged.sign(&crate::document::signatures::message(&last)),
            )],
        );
        let screened = required(&mut ours, &mut waiting, &mut signatures, Vec::new());
        assert_eq!(screened.applied, 0);
        assert_eq!(waiting.len(), 2);
        // The last one's: both in.
        signatures.offer(
            &ours,
            vec![(last, key.sign(&crate::document::signatures::message(&last)))],
        );
        let screened = required(&mut ours, &mut waiting, &mut signatures, Vec::new());
        assert_eq!(screened.applied, 2);
        assert!(screened.refused.is_empty());
        assert_eq!(waiting.len(), 0);
        assert_eq!(
            store::read_path(&ours, &path("notes/x")).unwrap(),
            Some(json!(1))
        );
    }
}