1 files changed, 139 insertions, 0 deletions
diff --git a/src/swarm/refusals.rs b/src/swarm/refusals.rs
new file mode 100644
index 0000000..2e5399a
--- /dev/null
+++ b/src/swarm/refusals.rs
@@ -0,0 +1,139 @@
+//! Telling a peer which of the changes it brought were refused (REFUSALS), so a node
+//! that took a change the swarm's rules refuse finds out and sets itself right.
+//!
+//! A node takes a change the others refuse only if it didn't judge it as they do: under
+//! a policy it didn't know yet (it receives without judging then, `access::policy`), or
+//! one that came after. Everything it writes from then on builds on that change and is
+//! refused too, and nobody tells it so. Hence: a node that refuses changes a peer
+//! brought tells it which broke a rule (`tell`), at most every `TELL_EVERY` and
+//! `MAX_LISTED` of them. The peer doesn't take the word for it (a member could tell
+//! anything): it judges each it holds itself, as its rules have them now
+//! (`received`). If its rules refuse one too, it took it unjudged, and drops it with
+//! everything built on it: its document as it would be without them, its own writes
+//! among them quarantined (`AppState::drop_taken`). If its rules take it, the two
+//! policies disagree, which it logs; finding out which is right is left
+//! (`future-work.md`, Write protection by checking changes).
+
+use std::time::Duration;
+
+use tracing::{debug, warn};
+
+use crate::document::replica::Epoch;
+use crate::document::{ChangeHash, ReadDoc};
+use crate::state::AppState;
+use crate::transport::{ParsedMessage, RefusalsMessage};
+
+/// Changes listed in one REFUSALS, at most.
+pub(crate) const MAX_LISTED: usize = 64;
+/// How often a peer is told, at most.
+pub(crate) const TELL_EVERY: Duration = Duration::from_secs(30);
+
+/// Tells `peer` that `changes` it brought, of our epoch, broke the swarm's rules.
+pub(crate) async fn tell(
+    state: &AppState,
+    peer: &str,
+    epoch: Epoch,
+    epoch_seq: u32,
+    mut changes: Vec<ChangeHash>,
+) {
+    changes.truncate(MAX_LISTED);
+    let message = ParsedMessage::Refusals(RefusalsMessage {
+        epoch,
+        epoch_seq,
+        changes,
+    });
+    if let Err(error) = state.transport.send_message(peer, &message).await {
+        // Told again with the next refusal.
+        debug!(peer = %peer, error = %format!("{error:#}"), "Couldn't tell refusals");
+    }
+}
+
+/// REFUSALS from `peer`, of our epoch: each change listed that we hold judged as our
+/// rules have it now; if they refuse one, we took it unjudged, and drop it.
+pub(crate) async fn received(state: &AppState, peer: &str, message: RefusalsMessage) {
+    let doc = std::sync::Arc::clone(&state.doc);
+    let changes: Vec<ChangeHash> = message.changes.into_iter().take(MAX_LISTED).collect();
+    let judged = tokio::task::spawn_blocking(move || {
+        let replica = doc.blocking_read();
+        if replica.epoch != message.epoch {
+            return None;
+        }
+        let policy = crate::access::policy::of(&replica.doc);
+        if !policy.enforced() || matches!(policy, crate::access::policy::Policy::Unknown(_)) {
+            // We can't judge as they do: receiving without judging until upgraded.
+            return None;
+        }
+        let mut refused = None;
+        let mut taken = Vec::new();
+        for hash in changes {
+            match judge_held(&replica.doc, replica.epoch, &hash) {
+                Some(Err(why)) => {
+                    refused = Some((hash, why));
+                    break;
+                }
+                Some(Ok(())) => taken.push(hash),
+                None => {}
+            }
+        }
+        Some((refused, taken))
+    })
+    .await
+    .ok()
+    .flatten();
+    let Some((refused, taken)) = judged else {
+        return;
+    };
+    if let Some((hash, why)) = refused {
+        warn!(peer = %peer, change = %hash, why = %why, "We hold a change the swarm's rules refuse, taken unjudged: dropping it and what builds on it (our writes among them go to the quarantine)");
+        state.drop_taken(hash).await;
+    } else if !taken.is_empty() {
+        warn!(peer = %peer, changes = ?taken, "A peer refuses changes our rules take: its policy or ours is off");
+    }
+}
+
+/// `hash`, which `doc` (of `epoch`) holds, judged by its rules now: why they refuse it,
+/// if they do; None if `doc` doesn't hold it.
+fn judge_held(
+    doc: &crate::document::Document,
+    epoch: Epoch,
+    hash: &ChangeHash,
+) -> Option<Result<(), String>> {
+    let change = doc.get_change_by_hash(hash)?;
+    let decoded = change.decode();
+    let ops = crate::document::inspect::Inspector::new(doc)
+        .ops(std::slice::from_ref(&decoded))
+        .pop()
+        .unwrap_or_default();
+    let paths = ops.iter().map(|op| op.path.clone()).collect();
+    let committer = doc
+        .get_author_for_actor(change.actor_id())
+        .and_then(|author| {
+            std::str::from_utf8(author.as_bytes())
+                .ok()
+                .map(str::to_owned)
+        });
+    let user = change
+        .message()
+        .and_then(crate::document::replica::user_of)
+        .map(str::to_owned);
+    let deps = change.deps().to_vec();
+    let judged = crate::access::rules::judge(
+        doc,
+        epoch,
+        &crate::access::rules::Change {
+            hash: *hash,
+            deps: &deps,
+            seen: &deps,
+            committer: committer.as_deref(),
+            user: user.as_deref(),
+            paths: &paths,
+            ops: &ops,
+        },
+    );
+    Some(match judged {
+        Ok(()) => Ok(()),
+        Err(crate::access::rules::Refusal::Rule(why)) => Err(why),
+        // Its dependencies are in the document: it is.
+        Err(crate::access::rules::Refusal::Unready) => Ok(()),
+    })
+}