diff options
| -rw-r--r-- | docs/access.md | 6 | ||||
| -rw-r--r-- | docs/measurements.md | 1 | ||||
| -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 |
7 files changed, 108 insertions, 7 deletions
diff --git a/docs/access.md b/docs/access.md index 5983c7c..a8dfed4 100644 --- a/docs/access.md +++ b/docs/access.md @@ -296,7 +296,11 @@ every 30 seconds). The peer doesn't take the word for it: it judges each it hold itself, by its rules as they are now. If they refuse one too, it drops it with everything built on it, its document becoming what it would be without them, and puts its own writes among them in the quarantine (`GET /admin/quarantine`), to be replayed -or discarded; what it writes from then on goes everywhere. If its rules take the +or discarded, and the others' among the refused (`GET /admin/refused`, "taken +unjudged, dropped"); what it writes from then on goes everywhere. A drop taking more +than 100 changes, or more than ten and a tenth of the document, waits until a second member says so too: +its own rules and one peer's word could both be off (logged as an error, counted in +`/admin/stats` as `sync.dropsHeld`). If its rules take the change, the two policies disagree, which it logs. A change whose dependencies haven't come yet waits for them. diff --git a/docs/measurements.md b/docs/measurements.md index 4550357..c377d9c 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -79,6 +79,7 @@ or caught up over many small messages. | `sync.lastFromNeighbourSecs` | `last_sync_from_neighbour_seconds` | Since a sync message from any neighbour was applied. Growing without bound while others write: the node is cut off from the swarm's changes. | | `sync.divergedPeers` | `diverged_peers` | Peers whose heads differ from ours with no sync message in between. | | `sync.refusedPeers` | `refused_peers` | Peers refused as a whole, their own changes having broken the swarm's rules ([`access.md`](access.md)): sent nothing. | +| `sync.dropsHeld` | `drops_held` | Changes a peer says the swarm's rules refuse, which this node took unjudged and its rules refuse too, whose drop would take much: held until a second member says so ([`access.md`](access.md)). Above 0: look at the log (`dropping it takes much`). | | `sync.largestReceivedBytes` | `largest_sync_received_bytes` | The largest sync message received since start. Larger transfers come as deltas (blobs of 16 chunks or more, [`peers.md`](peers.md)), so this stays below 16 × `--blob-chunk` from nodes of this version. | | `relay.announceRounds` | `announce_rounds_total` | Announcement rounds: our new heads to the neighbours after applying their changes (the quick round to those writers' trees leave out, or to all for changes no tree carries, and the slow round to the others; [`peers.md`](peers.md)). | | `relay.treelessRounds` | `treeless_rounds_total` | Of those, quick rounds to all neighbours: some change came that no writer's tree carries (pulled, a delta, from a node with no part in its tree). Many, with every node on the member list: trees miss often. | 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"); } |