diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/access/screen.rs | 80 | ||||
| -rw-r--r-- | src/api/dashboard/dashboard.html | 13 | ||||
| -rw-r--r-- | src/api/peers.rs | 25 | ||||
| -rw-r--r-- | src/api/stats.rs | 8 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 6 | ||||
| -rw-r--r-- | src/simtest/policy.rs | 9 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 8 |
7 files changed, 146 insertions, 3 deletions
diff --git a/src/access/screen.rs b/src/access/screen.rs index a5a8c8a..6773d05 100644 --- a/src/access/screen.rs +++ b/src/access/screen.rs @@ -8,6 +8,12 @@ //! the policy, bans) is what its writer saw, unless one of those ancestors changed it. So a change that //! does (`relevant`) is applied before the next is judged: the rest go in at once. //! +//! A member whose own change broke a rule, as its signature proves (signatures +//! required), is refused as a whole (`Screening::refusing`): the node sends it nothing +//! more, as all it commits builds on what was refused, until a change of its is taken +//! again (it rejoined) or the epoch changes. One that only passed a refused change on +//! isn't: the committer is, whoever brought it. +//! //! A change refused is remembered by its hash (`Refusals`): when it comes again, as its //! writer keeps offering what the node never takes, it is refused again without being //! decoded or logged, and the sync session with whoever sent only such changes backs off @@ -58,6 +64,16 @@ pub(crate) struct Screening { pub(crate) waiting: Waiting, refused: std::collections::VecDeque<Refused>, refusals: Refusals, + /// The members refused as a whole, by node ID (`refusing`). + refusing: std::collections::BTreeMap<String, Refusing>, +} + +/// A member refused as a whole: since when, for which change of its, and why. +#[derive(Debug, Clone)] +pub(crate) struct Refusing { + pub(crate) since: tokio::time::Instant, + pub(crate) change: ChangeHash, + pub(crate) why: String, } /// The hashes of the changes refused, to refuse them again by hash. @@ -117,6 +133,33 @@ impl Screening { pub(crate) fn refused(&self) -> impl DoubleEndedIterator<Item = &Refused> { self.refused.iter() } + + /// Whether `node` is refused as a whole, and since when, for what. + pub(crate) fn refusing(&self, node: &str) -> Option<&Refusing> { + self.refusing.get(node) + } + + /// The members refused as a whole and taken again of `screened`, logged. + fn judge_members(&mut self, screened: &Screened) { + for node in &screened.taken_from { + if self.refusing.remove(node).is_some() { + tracing::info!(node = %node, "A member refused as a whole wrote a change taken: refused no longer"); + } + } + for (node, change, why) in &screened.broke_rules { + if !self.refusing.contains_key(node) { + tracing::warn!(node = %node, change = %change, why = %why, "A member's own change broke the rules: refusing it as a whole, sending it nothing"); + self.refusing.insert( + node.clone(), + Refusing { + since: tokio::time::Instant::now(), + change: *change, + why: why.clone(), + }, + ); + } + } + } } /// Received changes waiting for their dependencies. @@ -136,6 +179,11 @@ pub(crate) struct Screened { pub(crate) refused: Vec<Refused>, /// Received changes refused before (`Refusals`), refused again by their hash. pub(crate) refused_again: usize, + /// Members whose own changes broke a rule, a signature proving it: with the change, + /// and why. + pub(crate) broke_rules: Vec<(String, ChangeHash, String)>, + /// The committers of changes taken. + pub(crate) taken_from: BTreeSet<String>, } /// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`) @@ -297,6 +345,9 @@ pub(crate) fn apply( match judged { Ok(()) => { accepted.push(change.clone()); + if let Some(committer) = committer { + screened.taken_from.insert(committer.to_owned()); + } if relevant(&paths[index]) { // What rules read changed: in before the next is judged. screened.applied += accepted.len(); @@ -313,6 +364,14 @@ pub(crate) fn apply( } Err(Refusal::Rule(why)) => { refused.insert(hash); + // Its committer's own, as a signature by it vouches. + if signatures.is_some() + && let Some(committer) = committer + { + screened + .broke_rules + .push((committer.to_owned(), hash, why.clone())); + } screened.refused.push(refuse(why)); } } @@ -381,6 +440,7 @@ pub(crate) fn apply_bytes( applied: screened.applied, refused_again: received > 0 && screened.refused_again == received, }; + screening.judge_members(&screened); screening.record(screened.refused); Ok(outcome) } @@ -537,6 +597,26 @@ mod tests { assert!(waiting.len() == 0); } + /// A member whose own change broke a rule: refused as a whole, until a change of its + /// is taken. + #[test] + fn members_refused_as_a_whole() { + let mut screening = Screening::default(); + let broke = Screened { + broke_rules: vec![("b".to_owned(), ChangeHash([1; 32]), "why".to_owned())], + ..Screened::default() + }; + screening.judge_members(&broke); + assert_eq!(screening.refusing("b").unwrap().change, ChangeHash([1; 32])); + assert!(screening.refusing("a").is_none()); + let taken = Screened { + taken_from: ["b".to_owned()].into(), + ..Screened::default() + }; + screening.judge_members(&taken); + assert!(screening.refusing("b").is_none()); + } + /// A refused change offered again: refused by its hash, not judged or kept again, /// and so is a new one building on it. #[test] diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index 601b578..fa4d436 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -284,6 +284,7 @@ #peers td.via { max-width: 0; min-width: 8em; width: 100%; overflow: hidden; text-overflow: ellipsis; } /* A peer's status (not the legend's "dormant"). */ #peers .active { color: var(--ok); } #peers .dormant { color: var(--bad); } + #peers .refused { color: var(--bad); font-weight: 600; } #peers .unknown { color: var(--muted); } .tree { font-family: ui-monospace, monospace; font-size: 12px; } /* Every entry indented alike: a branch's triangle sits in the indentation, left of @@ -658,9 +659,15 @@ function renderPeers() { el("td", { title: id + (peer.neighbour ? "" : "\nnot a neighbour: this node only answers it, and syncs with it on its own only as a neighbour") }, peer.alias || el("span", { class: "id" }, short(id))), - // How long dormant is about how long unseen ("seen"): in the tooltip only. - el("td", { class: peer.status, title: peer.status === "dormant" - ? "dormant for " + age(aged("dormantForSecs", peer.dormantForSecs)) : "" }, peer.status), + // How long dormant is about how long unseen ("seen"): in the tooltip only. Refused + // as a whole (its own change broke the swarm's rules) above all. + peer.refused + ? el("td", { class: "refused", title: "refused as a whole for " + + age(aged("secsAgo", peer.refused.secsAgo)) + ": its change " + + short(peer.refused.change) + " broke the swarm's rules (" + peer.refused.why + + "), so this node sends it nothing" }, "refused") + : el("td", { class: peer.status, title: peer.status === "dormant" + ? "dormant for " + age(aged("dormantForSecs", peer.dormantForSecs)) : "" }, peer.status), el("td", {}, age(aged("lastSeenSecsAgo", peer.lastSeenSecsAgo))), el("td", { class: "id via", title: via }, via), el("td", {}, rtt), diff --git a/src/api/peers.rs b/src/api/peers.rs index cd9381f..2bdde96 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -50,6 +50,9 @@ pub(crate) struct PeerStatus { /// How long its heads have differed from ours with no sync message from it in /// between; absent while they match. pub(crate) diverged_for_secs: Option<u64>, + /// Whether we refuse it as a whole, its own change having broken the swarm's rules + /// (`access::screen`): then we send it nothing. + pub(crate) refused: Option<RefusedPeer>, pub(crate) has_pending_delivery: bool, /// Epoch the peer last advertised; absent until it has sent a message. pub(crate) epoch: Option<EpochRecord>, @@ -113,6 +116,16 @@ pub(crate) struct ThroughputInfo { pub(crate) secs_ago: u64, } +/// A peer refused as a whole (`PeerStatus::refused`). +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct RefusedPeer { + pub(crate) secs_ago: u64, + /// The change of its that broke the rules, and why. + pub(crate) change: String, + pub(crate) why: String, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct PeerListResponse { @@ -176,6 +189,17 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse { samples: estimate.samples, secs_ago: now.saturating_duration_since(estimate.at).as_secs(), }); + let refused = { + state + .screening + .lock_or_recover() + .refusing(&peer) + .map(|refusing| RefusedPeer { + secs_ago: now.saturating_duration_since(refusing.since).as_secs(), + change: refusing.change.to_string(), + why: refusing.why.clone(), + }) + }; let ps = peer_state_mutex.lock().await; let (status, dormant_for_secs) = match ps.liveness { @@ -223,6 +247,7 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse { diverged_for_secs: ps .diverged_since .map(|t| now.saturating_duration_since(t).as_secs()), + refused, has_pending_delivery: ps.sync.sending.is_some() || ps.sync.retry_at.is_some(), epoch: ps.epoch.map(EpochRecord::from), same_epoch: ps.epoch.map(|epoch| epoch == local_epoch), diff --git a/src/api/stats.rs b/src/api/stats.rs index 8eaff0c..1f0701d 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -127,6 +127,9 @@ struct SyncStats { last_from_neighbour_secs: Option<u64>, /// Peers whose heads differ from ours while sync is idle (`peers.md`, healing). diverged_peers: usize, + /// Peers refused as a whole, their own changes having broken the swarm's rules + /// (`access::screen`): sent nothing. + refused_peers: usize, /// The largest sync message received since start (its Automerge part). Larger /// transfers come as deltas (blobs). largest_received_bytes: u64, @@ -324,11 +327,15 @@ pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsR let mut incidents = Incidents::default(); let mut last_from_neighbour_secs: Option<u64> = None; let mut diverged_peers = 0; + let mut refused_peers = 0; for peer in list.peers.values() { incidents.add(&peer.incidents); if peer.diverged_for_secs.is_some() { diverged_peers += 1; } + if peer.refused.is_some() { + refused_peers += 1; + } if peer.neighbour && let Some(secs) = peer.last_sync_secs_ago { @@ -370,6 +377,7 @@ pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsR sync: SyncStats { last_from_neighbour_secs, diverged_peers, + refused_peers, largest_received_bytes: node.largest_sync_received, }, relay: RelayStats { diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 038f739..41d224a 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -634,6 +634,12 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String { "Peers whose heads differ from ours while sync is idle.", stats.sync.diverged_peers as f64, ); + out.single( + "refused_peers", + "gauge", + "Peers refused as a whole, their own changes having broken the swarm's rules.", + stats.sync.refused_peers as f64, + ); let relay = &stats.relay; out.single( "announce_rounds_total", diff --git a/src/simtest/policy.rs b/src/simtest/policy.rs index b6be095..eb706bc 100644 --- a/src/simtest/policy.rs +++ b/src/simtest/policy.rs @@ -90,6 +90,15 @@ async fn a_swarm_judging_every_change() { }) .await; assert!(!holds(node, "notes/forged").await); + // Its own change, signed: c refused as a whole, sent nothing more. + use crate::locks::MutexExt; + assert!( + node.screening + .lock_or_recover() + .refusing(c.local_id()) + .is_some(), + "c isn't refused as a whole" + ); } } diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index f5d6bfb..9f928ad 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -36,6 +36,14 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { trace!("Peer was forgotten; skipping sync"); return Ok(()); }; + { + use crate::locks::MutexExt; + if state.screening.lock_or_recover().refusing(peer).is_some() { + // All it commits builds on what we refused (`access::screen`). + trace!("Refusing the peer as a whole; not syncing with it"); + return Ok(()); + } + } let we_ready = state.is_ready().await; |