From 7323e8ed8598cda9784493aa3334d947120b53d8 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 7 Nov 2026 22:35:22 +0100 Subject: swarm: changes taken unjudged, dropped (REFUSALS) A node refusing changes a peer brought tells it which broke a rule; the peer judges them itself, and if its rules refuse them too, drops them with all built on them (its document forked without them, a fresh actor), quarantining its own writes among them. Wire version 42. Auth plan M6 (rejoin on taint). --- src/swarm/mod.rs | 1 + src/swarm/peer/inbound.rs | 10 ++++ src/swarm/refusals.rs | 139 ++++++++++++++++++++++++++++++++++++++++++++++ src/swarm/sync.rs | 19 +++++-- 4 files changed, 165 insertions(+), 4 deletions(-) create mode 100644 src/swarm/refusals.rs (limited to 'src/swarm') diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 0cb4fb8..d872474 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -15,6 +15,7 @@ pub(crate) mod pair; pub(crate) mod peer; pub(crate) mod plumtree; pub(crate) mod reach; +pub(crate) mod refusals; pub(crate) mod relay; pub(crate) mod rtt; pub(crate) mod signatures; diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index 9b5db8f..a7633b3 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -272,6 +272,16 @@ pub(crate) async fn handle_message( crate::swarm::signatures::received(state, message).await; } } + transport::ParsedMessage::Refusals(message) => { + // From members, of our epoch: judging costs (`swarm::refusals`). + let member = matches!( + peer_state.lock().await.standing, + crate::state::Standing::Member { .. } + ); + if member && same_epoch { + crate::swarm::refusals::received(state, &peer, message).await; + } + } transport::ParsedMessage::Probe(request) => { crate::swarm::working_links::answer(state, &peer, request).await; } 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, +) { + 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 = 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> { + 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(()), + }) +} diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 9f928ad..7601d69 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -571,6 +571,7 @@ pub(crate) async fn handle_sync( bool, (Vec<(String, Vec)>, bool), Option, + Option<(u32, Vec)>, ); tokio::task::spawn_blocking(move || -> Result> { // Before the lock: inflating and decoding a large message takes a while. No @@ -618,6 +619,7 @@ pub(crate) async fn handle_sync( let epoch_seq = replica.epoch_seq; let mut message = message; let mut refused_again = false; + let mut roots = Vec::new(); replica .logged_signed(|doc, known, log| { let result = (|| { @@ -630,7 +632,7 @@ pub(crate) async fn handle_sync( if required { known.offer(doc, signatures.clone()); } - refused_again = crate::access::screen::apply_bytes( + let outcome = crate::access::screen::apply_bytes( doc, epoch, epoch_seq, @@ -638,8 +640,9 @@ pub(crate) async fn handle_sync( &screening, required.then_some(&mut *known), log, - )? - .refused_again; + )?; + refused_again = outcome.refused_again; + roots = outcome.roots; } sync.receive_logged(doc, message, log) })(); @@ -649,6 +652,10 @@ pub(crate) async fn handle_sync( peer_state.last_sync_rx = Some(Instant::now()); // It offers again only what we refused, and would as fast as we answer: we // answer later, later each time (`SyncSession::refused_again`). + // What it brought that broke the rules, told to it now and then + // (`swarm::refusals`). + let tell = (!roots.is_empty() && peer_state.sync.refusals_due(Instant::now())) + .then(|| (replica.epoch_seq, std::mem::take(&mut roots))); let back_off = if refused_again { Some(peer_state.sync.refused_again(Instant::now())) } else { @@ -744,16 +751,20 @@ pub(crate) async fn handle_sync( carried, authors, back_off, + tell, ))) }) .await .context("Sync task panicked")?? }; - let Some((caught_up, changed, carried, (authors, unauthored), back_off)) = applied else { + let Some((caught_up, changed, carried, (authors, unauthored), back_off, tell)) = applied else { debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); return Ok(()); }; + if let Some((epoch_seq, roots)) = tell { + crate::swarm::refusals::tell(state, peer, epoch, epoch_seq, roots).await; + } if let Some(delay) = back_off { debug!( peer = %peer, -- cgit v1.3.1