diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/access/screen.rs | 27 | ||||
| -rw-r--r-- | src/api/stats.rs | 8 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 6 | ||||
| -rw-r--r-- | src/state/host.rs | 62 | ||||
| -rw-r--r-- | src/swarm/refusals.rs | 5 |
5 files changed, 102 insertions, 6 deletions
diff --git a/src/access/screen.rs b/src/access/screen.rs index e15b407..89ca007 100644 --- a/src/access/screen.rs +++ b/src/access/screen.rs @@ -70,6 +70,9 @@ pub(crate) struct Screening { refusals: Refusals, /// The members refused as a whole, by node ID (`refusing`). refusing: std::collections::BTreeMap<String, Refusing>, + /// Changes this node holds that peers say the swarm's rules refuse, and its own do + /// too, whose drop would take much: each with the peers that said so (`hold_drop`). + held_drops: HashMap<ChangeHash, BTreeSet<String>>, } /// A member refused as a whole: since when, for which change of its, and why. @@ -139,6 +142,30 @@ impl Screening { } } + /// `peer` says `root`, held here, is refused by the swarm's rules, its drop taking + /// much (`AppState::drop_taken`): how many peers said so of it, this one with them. + pub(crate) fn hold_drop(&mut self, root: ChangeHash, peer: &str) -> usize { + let named = self.held_drops.entry(root).or_default(); + named.insert(peer.to_owned()); + named.len() + } + + /// Drops held, for another peer to say so too (`hold_drop`). + pub(crate) fn drops_held(&self) -> usize { + self.held_drops.len() + } + + /// `dropped`, changes this node took unjudged and dropped (`AppState::drop_taken`), + /// kept with those refused, the oldest going past `MAX_REFUSED`; logged once, there. + pub(crate) fn keep_dropped(&mut self, dropped: Vec<Refused>) { + for dropped in dropped { + self.refused.push_back(dropped); + 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() diff --git a/src/api/stats.rs b/src/api/stats.rs index 1f0701d..90b4fc2 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -130,6 +130,10 @@ struct SyncStats { /// Peers refused as a whole, their own changes having broken the swarm's rules /// (`access::screen`): sent nothing. refused_peers: usize, + /// Changes peers say the swarm's rules refuse, which this node took unjudged and its + /// rules refuse too, whose drop takes much: held for a second member to say so + /// (`AppState::drop_taken`). + drops_held: usize, /// The largest sync message received since start (its Automerge part). Larger /// transfers come as deltas (blobs). largest_received_bytes: u64, @@ -378,6 +382,10 @@ pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsR last_from_neighbour_secs, diverged_peers, refused_peers, + drops_held: { + use crate::locks::MutexExt; + state.screening.lock_or_recover().drops_held() + }, largest_received_bytes: node.largest_sync_received, }, relay: RelayStats { diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 41d224a..dd49751 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -640,6 +640,12 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String { "Peers refused as a whole, their own changes having broken the swarm's rules.", stats.sync.refused_peers as f64, ); + out.single( + "drops_held", + "gauge", + "Changes peers say the swarm's rules refuse, ours too, whose drop takes much: held for a second member to say so.", + stats.sync.drops_held as f64, + ); let relay = &stats.relay; out.single( "announce_rounds_total", 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 diff --git a/src/swarm/refusals.rs b/src/swarm/refusals.rs index c1949eb..af4252c 100644 --- a/src/swarm/refusals.rs +++ b/src/swarm/refusals.rs @@ -10,7 +10,8 @@ //! 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 +//! among them quarantined, the others' listed as refused (`AppState::drop_taken`); one +//! taking much, once a second member says so too. 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). @@ -85,7 +86,7 @@ pub(crate) async fn received(state: &AppState, peer: &str, message: RefusalsMess }; 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; + state.drop_taken(hash, peer, &why).await; } else if !taken.is_empty() { warn!(peer = %peer, changes = ?taken, "A peer refuses changes our rules take: its policy or ours is off"); } |