diff options
Diffstat (limited to 'src/state')
| -rw-r--r-- | src/state/host.rs | 62 |
1 files changed, 58 insertions, 4 deletions
diff --git a/src/state/host.rs b/src/state/host.rs index 38be831..feda8a9 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -95,12 +95,21 @@ impl AppState { /// (`swarm::refusals`), and every change built on it: the document becomes the one /// without them (forked where none is), written by a fresh actor of this node's, as /// the old one's later changes are among those dropped; this node's own writes among - /// them go to the quarantine. Every peer syncs anew. Nothing while a rejoin runs, or - /// if the document doesn't hold `root`; nor if `root` is an epoch's root, or nothing - /// would be kept, which no rule can mean (logged as an error). - pub(crate) async fn drop_taken(&self, root: ChangeHash) { + /// them go to the quarantine, the others' to the refused (`GET /admin/refused`, saying + /// so). Every peer syncs anew. Nothing while a rejoin runs, or if the document doesn't + /// hold `root`; nor if `root` is an epoch's root, or nothing would be kept, which no + /// rule can mean (logged as an error). A drop taking much (`BIG_DROP`) waits for a + /// second member to say so (`peer` the first, `why` its rules' reason): its rules and + /// one peer's word might both be off, the swarm's not. + pub(crate) async fn drop_taken(&self, root: ChangeHash, peer: &str, why: &str) { + /// Changes a drop takes, at most, on one peer's word: more, or more than a tenth + /// of the document (past `SMALL_DROP`), wait for another's. + const BIG_DROP: usize = 100; + const SMALL_DROP: usize = 10; let own = self.remake.own_actors(); let state = self.clone(); + let peer = peer.to_owned(); + let why = why.to_owned(); let dropped = tokio::task::spawn_blocking(move || { let mut replica = state.doc.blocking_write(); if !replica.writable { @@ -127,6 +136,19 @@ impl AppState { error!(change = %root, "Dropping a change would leave nothing; kept it"); return None; } + if tainted.len() > BIG_DROP + || (tainted.len() > SMALL_DROP && tainted.len() * 10 > changes.len()) + { + use crate::locks::MutexExt; + let mut screening = state.screening.lock_or_recover(); + let said = screening + .of(replica.epoch_seq) + .hold_drop(root, &peer); + if said < 2 { + error!(change = %root, peer = %peer, changes = tainted.len(), why = %why, "A peer says we hold a change the swarm's rules refuse, and ours do too, but dropping it takes much: waiting for another member to say so"); + return None; + } + } let built_on: std::collections::HashSet<ChangeHash> = kept .iter() .flat_map(|change| change.deps().iter().copied()) @@ -162,9 +184,41 @@ impl AppState { return None; } }; + // The others' changes dropped, for operators (own: to the quarantine). + let theirs: Vec<crate::access::screen::Refused> = changes + .iter() + .filter(|change| tainted.contains(&change.hash())) + .filter(|change| !actors.contains(change.actor_id())) + .map(|change| crate::access::screen::Refused { + change: (*change).clone(), + committer: replica + .doc + .get_author_for_actor(change.actor_id()) + .and_then(|author| { + std::str::from_utf8(author.as_bytes()) + .ok() + .map(str::to_owned) + }), + user: change + .message() + .and_then(crate::document::replica::user_of) + .map(str::to_owned), + paths: Default::default(), + why: if change.hash() == root { + format!("taken unjudged, dropped: {why}") + } else { + format!("taken unjudged, dropped: it builds on {root}") + }, + }) + .collect(); state.remake.add_own_actor(actor); let epoch = replica.epoch; let old = state.swap_document(&mut replica, epoch, fresh); + { + use crate::locks::MutexExt; + let mut screening = state.screening.lock_or_recover(); + screening.of(replica.epoch_seq).keep_dropped(theirs); + } Some((epoch, old, paths, tainted.len())) }) .await |