diff options
| -rw-r--r-- | docs/access.md | 10 | ||||
| -rw-r--r-- | docs/dashboard.md | 2 | ||||
| -rw-r--r-- | docs/measurements.md | 2 | ||||
| -rw-r--r-- | docs/openapi.yaml | 11 | ||||
| -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 |
11 files changed, 168 insertions, 6 deletions
diff --git a/docs/access.md b/docs/access.md index 567cebd..31efe4e 100644 --- a/docs/access.md +++ b/docs/access.md @@ -238,7 +238,15 @@ change stays out, and so does whatever builds on it; the node logs it (`Refused change`) and lists the last thousand (`GET /admin/refused`). Offered again, as its writer's node keeps offering what is never taken, it is refused by its hash, not judged or logged again, and a peer offering nothing else is answered later each time, up to a -minute apart. A change whose +minute apart. + +With signatures required, a member whose own change broke a rule (its signature proves +it its own) is refused as a whole: every node sends it nothing more, as all it commits +from then on builds on what was refused, and logs it once. It shows in `GET /peers` +(`refused`, with the change and why), in `/admin/stats` (`sync.refusedPeers`) and on the +dashboard. A node that only passed a refused change on isn't: the committer is, whoever +brought it. It lasts until a change of the member's is taken again (it rejoined and +writes anew), or the epoch changes; forgetting or banning it is the operators'. A change whose dependencies haven't come yet waits for them. What it doesn't do yet: diff --git a/docs/dashboard.md b/docs/dashboard.md index df2d4f9..271592a 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -247,7 +247,7 @@ them: no HELLOs, keepalives or syncing of its own; their tooltip says so), with: | Column | Meaning | |---|---| | peer | Alias, or a shortened node ID (the full one in the tooltip). | -| status | `active`, `dormant` (for how long: in the tooltip) or `unknown` (not heard from since this node started). | +| status | `active`, `dormant` (for how long: in the tooltip) or `unknown` (not heard from since this node started); `refused`, above all, where this node refuses it as a whole, its own change having broken the swarm's rules (the change and why in the tooltip; [`access.md`](access.md)). | | seen | How long ago this node last heard from it. | | via | The address this node sends to it at, and whether `direct` or `relayed`: as much as the box's width leaves, cut short with an ellipsis; all of it in the tooltip. | | rtt | The round trip on that path. | diff --git a/docs/measurements.md b/docs/measurements.md index 4e1099f..8e36a7e 100644 --- a/docs/measurements.md +++ b/docs/measurements.md @@ -78,6 +78,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.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. | @@ -281,6 +282,7 @@ In `/peers`, and as `peer_…` series with a `peer` label: | `destination` | `peer_destination_info{transport,proven,connection}` | Where messages to it go: the address (its transport), whether it proved itself there (else sends wait for its proof), and the connection there: libp2p's `direct` or `relayed`, Tor's `direct` (an onion circuit); none on Mycelium; while `relayed` after a failed hole punch, `punching`: punches failed in a row (`failed`) and when it's tried again (`nextInSecs`). | | `destination.paths[].rttMs`, `.down`, `.connection` | `peer_path_rtt_seconds{transport}`, `peer_path_down{transport}` | Every path to it, best first: its round-trip time (smoothed, from PINGs), whether it is down, and the transport's connection there (`direct` or `relayed`, where it has connections). The first is the one in use: the peer's round-trip time. | | `lastSyncSecsAgo` | `peer_last_sync_seconds` | Since a sync message from it was applied (or the epoch changed). | +| `refused` | | Refused as a whole, its own change having broken the swarm's rules ([`access.md`](access.md)): since when (`secsAgo`), the change, and why. | | `divergedForSecs` | `peer_diverged_seconds` | How long its heads have differed from ours with no sync message from it in between, nor both moving on; absent while they match. Long values precede a healing re-handshake ([`peers.md`](peers.md)). | | `syncThroughput` | `peer_sync_throughput_bytes_per_second` | Sync bytes per second over exchanges of at least 32 KiB, round trip taken off, smoothed; and over messages of at least 32 KiB we awaited nothing for (taken as delivered: no answer to time), and blob chunks (deltas), from how long they took to come (`hopDelay`), half the round trip taken off. Absent before the first large exchange; comes with its age (`secsAgo`) since large exchanges are rare. This is the rate the peer delivers at in practice (its apply time included), not the path's capacity. | | `clock.offsetMs`, `.withinMs` | `peer_clock_offset_seconds`, `peer_clock_offset_within_seconds` | How far its clock is ahead of ours (behind if negative), give or take. Each PONG says when the peer answered, by its clock, which lies between our PING going and the PONG coming: within half the round trip of its middle, however unevenly the two ways took. Of its paths and the last 5 minutes, the narrowest bound is kept; a sample it contradicts (a clock set anew) replaces it at once. Absent before a PONG from it. | diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 7c8bb84..832d941 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -2175,6 +2175,13 @@ components: divergedForSecs: type: [integer, 'null'] description: How long its heads have differed from ours with no sync message from it in between; null while they match. + refused: + type: [object, 'null'] + description: "Refused as a whole, its own change having broken the swarm's rules (`docs/access.md`): sent nothing. Null if not." + properties: + secsAgo: { type: integer } + change: { type: string, description: The change of its that broke the rules. } + why: { type: string } hasPendingDelivery: type: boolean epoch: @@ -2568,12 +2575,14 @@ components: type: integer sync: type: object - required: [divergedPeers, largestReceivedBytes] + required: [divergedPeers, refusedPeers, largestReceivedBytes] properties: lastFromNeighbourSecs: type: [integer, 'null'] divergedPeers: type: integer + refusedPeers: + type: integer largestReceivedBytes: type: integer relay: 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; |