diff options
| -rw-r--r-- | docs/access.md | 5 | ||||
| -rw-r--r-- | src/access/screen.rs | 157 | ||||
| -rw-r--r-- | src/document/sync.rs | 23 | ||||
| -rw-r--r-- | src/swarm/signatures.rs | 2 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 33 |
5 files changed, 203 insertions, 17 deletions
diff --git a/docs/access.md b/docs/access.md index cc35565..e3dec0b 100644 --- a/docs/access.md +++ b/docs/access.md @@ -232,7 +232,10 @@ longer write as another member; the user a change names is still its node's word In a swarm without a registry, whatever a user would need, any member may. A refused change stays out, and so does whatever builds on it; the node logs it (`Refused a -change`) and lists the last thousand (`GET /admin/refused`). A change whose +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 dependencies haven't come yet waits for them. What it doesn't do yet: diff --git a/src/access/screen.rs b/src/access/screen.rs index 736e04a..a5a8c8a 100644 --- a/src/access/screen.rs +++ b/src/access/screen.rs @@ -8,6 +8,11 @@ //! 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 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 +//! (`Outcome::refused_again`, `SyncSession::refused_again`). +//! //! A change whose dependencies are neither in the document nor among those received //! waits (`Waiting`) until they come, a bounded number; Automerge never sees it before. //! With signatures required (`access::policy`), so does one no signature by its @@ -29,6 +34,9 @@ use crate::document::{Change, ChangeHash, Document, ReadDoc}; const MAX_WAITING: usize = 10_000; /// Refused changes kept for operators, at most: the oldest go. pub(crate) const MAX_REFUSED: usize = 1_000; +/// Hashes of refused changes remembered, at most: the oldest go (and are judged again, +/// and refused again, should they come). +const MAX_REFUSALS: usize = 100_000; /// A refused change, and why. #[derive(Debug, Clone)] @@ -49,6 +57,31 @@ pub(crate) struct Screening { epoch_seq: u32, pub(crate) waiting: Waiting, refused: std::collections::VecDeque<Refused>, + refusals: Refusals, +} + +/// The hashes of the changes refused, to refuse them again by hash. +#[derive(Debug, Default)] +pub(crate) struct Refusals { + known: HashSet<ChangeHash>, + order: std::collections::VecDeque<ChangeHash>, +} + +impl Refusals { + fn contains(&self, hash: &ChangeHash) -> bool { + self.known.contains(hash) + } + + fn insert(&mut self, hash: ChangeHash) { + if self.known.insert(hash) { + self.order.push_back(hash); + if self.order.len() > MAX_REFUSALS + && let Some(oldest) = self.order.pop_front() + { + self.known.remove(&oldest); + } + } + } } impl Screening { @@ -101,29 +134,63 @@ impl Waiting { pub(crate) struct Screened { pub(crate) applied: usize, pub(crate) refused: Vec<Refused>, + /// Received changes refused before (`Refusals`), refused again by their hash. + pub(crate) refused_again: usize, } /// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`) /// those rules allow, what they change recorded in `log`. Those already in `doc` are -/// left out; those whose dependencies are missing wait. -/// With `signatures` (required), those no signature vouches for wait too. +/// left out; those whose dependencies are missing wait; those `refusals` holds are +/// refused again by their hash, and those building on them, and those refused now go +/// into it. With `signatures` (required), those no signature vouches for wait too. pub(crate) fn apply( doc: &mut Document, epoch: Epoch, received: Vec<Change>, waiting: &mut Waiting, + refusals: &mut Refusals, mut signatures: Option<&mut Signatures>, log: &mut crate::document::PatchLog, ) -> Result<Screened> { let mut screened = Screened::default(); - // Each once, not in the document yet. + // Each once, not in the document yet, not refused before. let mut seen = HashSet::new(); let candidates: Vec<Change> = std::mem::take(&mut waiting.0) .into_iter() .chain(received) .filter(|change| seen.insert(change.hash())) .filter(|change| doc.get_change_by_hash(&change.hash()).is_none()) + .filter(|change| { + let again = refusals.contains(&change.hash()); + screened.refused_again += usize::from(again); + !again + }) .collect(); + // Building on one refused before: refused too, without waiting for it (it never + // comes) or judging it. + let (building, candidates): (Vec<Change>, Vec<Change>) = candidates + .into_iter() + .partition(|change| change.deps().iter().any(|dep| refusals.contains(dep))); + for change in building { + let dep = change + .deps() + .iter() + .find(|dep| refusals.contains(dep)) + .copied() + .expect("partitioned so"); + refusals.insert(change.hash()); + screened.refused.push(Refused { + committer: change.author().and_then(|author| { + std::str::from_utf8(author.as_bytes()) + .ok() + .map(str::to_owned) + }), + user: change.message().and_then(user_of).map(str::to_owned), + paths: BTreeSet::new(), + why: format!("it builds on refused change {dep}"), + change, + }); + } let (ready, unready) = in_order(doc, candidates); waiting.0 = unready; if ready.is_empty() { @@ -180,7 +247,11 @@ pub(crate) fn apply( paths: paths[index].clone(), why, }; - if let Some(dep) = change.deps().iter().find(|dep| refused.contains(dep)) { + if let Some(dep) = change + .deps() + .iter() + .find(|dep| refused.contains(dep) || refusals.contains(dep)) + { refused.insert(hash); screened .refused @@ -250,10 +321,23 @@ pub(crate) fn apply( if !accepted.is_empty() { doc.apply_changes_log_patches(accepted, log)?; } + for hash in refused { + refusals.insert(hash); + } waiting.0.truncate(MAX_WAITING); Ok(screened) } +/// What `apply_bytes` did. +#[derive(Debug, Default, Clone, Copy)] +pub(crate) struct Outcome { + /// Changes applied (received, or waiting before). + pub(crate) applied: usize, + /// Whether the bytes carried changes, and all of them refused before: their sender + /// offers again what the node never takes, and its sync session backs off. + pub(crate) refused_again: bool, +} + /// [`apply`] of the changes in `bytes`, as sync messages and downloads carry them, with /// what `screening` keeps for document `epoch_seq` (of `epoch`). Bytes that don't load /// as changes, or with a misplaced author (`changes::misplaced_author`), are refused @@ -268,10 +352,10 @@ pub(crate) fn apply_bytes( screening: &std::sync::Mutex<Screening>, signatures: Option<&mut Signatures>, log: &mut crate::document::PatchLog, -) -> Result<usize> { +) -> Result<Outcome> { use crate::locks::MutexExt; if bytes.is_empty() && signatures.is_none() { - return Ok(0); + return Ok(Outcome::default()); } if let Some(misplaced) = crate::document::changes::misplaced_author(bytes) { anyhow::bail!("Refused changes: {misplaced}"); @@ -281,12 +365,24 @@ pub(crate) fn apply_bytes( let mut screening = screening.lock_or_recover(); let screening = screening.of(epoch_seq); if changes.is_empty() && screening.waiting.len() == 0 { - return Ok(0); + return Ok(Outcome::default()); } - let screened = apply(doc, epoch, changes, &mut screening.waiting, signatures, log)?; - let applied = screened.applied; + let received = changes.len(); + let screened = apply( + doc, + epoch, + changes, + &mut screening.waiting, + &mut screening.refusals, + signatures, + log, + )?; + let outcome = Outcome { + applied: screened.applied, + refused_again: received > 0 && screened.refused_again == received, + }; screening.record(screened.refused); - Ok(applied) + Ok(outcome) } /// Whether a change touching `paths` changes what rules read: the members, the user @@ -382,6 +478,7 @@ mod tests { Epoch::GENESIS, changes, waiting, + &mut Refusals::default(), None, &mut crate::document::PatchLog::inactive(), ) @@ -440,6 +537,45 @@ mod tests { assert!(waiting.len() == 0); } + /// A refused change offered again: refused by its hash, not judged or kept again, + /// and so is a new one building on it. + #[test] + fn refused_again_by_hash() { + let mut theirs = genesis(); + write(&mut theirs, "a", Some("alice"), "config/x", json!(1)); + let mut ours = Document::new(); + let mut waiting = Waiting::default(); + let mut refusals = Refusals::default(); + let mut screen_with = |doc: &mut Document, changes: Vec<Change>| { + apply( + doc, + Epoch::GENESIS, + changes, + &mut waiting, + &mut refusals, + None, + &mut crate::document::PatchLog::inactive(), + ) + .unwrap() + }; + let first = screen_with(&mut ours, received(&theirs, &[])); + assert_eq!( + (first.applied, first.refused.len(), first.refused_again), + (2, 1, 0) + ); + let heads = ours.get_heads(); + let again = screen_with(&mut ours, received(&theirs, &heads)); + assert_eq!( + (again.applied, again.refused.len(), again.refused_again), + (0, 0, 1) + ); + let before = theirs.get_heads(); + write(&mut theirs, "a", Some("alice"), "notes/x", json!(2)); + let building = screen_with(&mut ours, received(&theirs, &before)); + assert_eq!((building.applied, building.refused.len()), (0, 1)); + assert!(building.refused[0].why.contains("builds on refused change")); + } + /// Judged as its writer saw the registry: a grant earlier in the same batch counts. #[test] fn a_grant_in_the_same_batch_counts() { @@ -538,6 +674,7 @@ mod tests { Epoch::GENESIS, changes, waiting, + &mut Refusals::default(), Some(signatures), &mut crate::document::PatchLog::inactive(), ) diff --git a/src/document/sync.rs b/src/document/sync.rs index f124ee8..823d4a3 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -153,6 +153,10 @@ static NEXT_SESSION: AtomicU64 = AtomicU64::new(1); /// while it keeps refusing. const SEND_RETRY: Duration = Duration::from_millis(500); const SEND_RETRY_MAX: Duration = Duration::from_secs(30); +/// The first wait before answering a peer offering only changes we refused, and the +/// longest (`SyncSession::refused_again`). +const REFUSED_WAIT: Duration = Duration::from_secs(1); +const REFUSED_WAIT_MAX: Duration = Duration::from_secs(60); /// One Automerge sync with a peer, from a reset to the next. Everything a reset must /// forget lives here, so a reset replaces it whole and can't miss a field. @@ -205,6 +209,10 @@ pub(crate) struct SyncSession { /// comes late doesn't count: `older_keeps_newer`). said_heads: Option<Vec<crate::document::ChangeHash>>, + /// The wait before answering a peer that offers again only changes we refused + /// (`refused_again`). + refused_backoff: Backoff, + /// The answer we expect to our last sync message(s), if any. pub(crate) reply: Option<SyncReplyWait>, /// How long to wait for it; doubles while answers stay out. @@ -255,6 +263,7 @@ impl SyncSession { known_elsewhere: std::collections::HashMap::new(), assumed_lately: std::collections::VecDeque::new(), said_heads: None, + refused_backoff: Backoff::new(REFUSED_WAIT, REFUSED_WAIT_MAX), reply: None, reply_backoff, } @@ -975,6 +984,20 @@ impl SyncSession { self.id } + /// The peer offered again only changes we refused (`access::screen`): it would as + /// often as we answer, its sync state never learning we won't take them. No message + /// to it before the returned wait, which doubles while it goes on. + pub(crate) fn refused_again(&mut self, now: Instant) -> Duration { + let delay = self.refused_backoff.next(); + self.retry_at = Some(self.retry_at.map_or(now + delay, |at| at.max(now + delay))); + delay + } + + /// The peer brought changes we took (or none refused before): answered at once again. + pub(crate) fn refusals_settled(&mut self) { + self.refused_backoff.reset(); + } + /// Our transport took the message generated from session `session`. Returns whether /// to generate again: a pass was requested while it was handed over, or a reset /// replaced the session meanwhile (any pass requested then was lost with the old diff --git a/src/swarm/signatures.rs b/src/swarm/signatures.rs index 309ccc2..9ea404e 100644 --- a/src/swarm/signatures.rs +++ b/src/swarm/signatures.rs @@ -96,7 +96,7 @@ pub(crate) async fn received(state: &AppState, message: SignaturesMessage) { }); replica.refresh_heads(); return match applied { - Ok(applied) => applied, + Ok(outcome) => outcome.applied, Err(error) => { tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures"); 0 diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index ff7fa5a..f5d6bfb 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -554,13 +554,15 @@ pub(crate) async fn handle_sync( let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed); // Some((caught_up, heads changed, carried changes, the new changes' authors with - // theirs, and whether any has none)) when applied, None when the message belongs - // to another epoch. + // theirs, and whether any has none, how long to wait before answering if it + // brought only changes refused before)) when applied, None when the message + // belongs to another epoch. type Applied = ( bool, bool, bool, (Vec<(String, Vec<crate::document::ChangeHash>)>, bool), + Option<std::time::Duration>, ); tokio::task::spawn_blocking(move || -> Result<Option<Applied>> { // Before the lock: inflating and decoding a large message takes a while. No @@ -607,6 +609,7 @@ pub(crate) async fn handle_sync( let sync = &mut peer_state.sync; let epoch_seq = replica.epoch_seq; let mut message = message; + let mut refused_again = false; replica .logged_signed(|doc, known, log| { let result = (|| { @@ -619,7 +622,7 @@ pub(crate) async fn handle_sync( if required { known.offer(doc, signatures.clone()); } - crate::access::screen::apply_bytes( + refused_again = crate::access::screen::apply_bytes( doc, epoch, epoch_seq, @@ -627,7 +630,8 @@ pub(crate) async fn handle_sync( &screening, required.then_some(&mut *known), log, - )?; + )? + .refused_again; } sync.receive_logged(doc, message, log) })(); @@ -635,6 +639,16 @@ pub(crate) async fn handle_sync( }) .context("Automerge sync failed")?; peer_state.last_sync_rx = Some(Instant::now()); + // It offers again only what we refused, and would as fast as we answer: we + // answer later, later each time (`SyncSession::refused_again`). + let back_off = if refused_again { + Some(peer_state.sync.refused_again(Instant::now())) + } else { + if carried { + peer_state.sync.refusals_settled(); + } + None + }; if let Some(wait) = &peer_state.sync.reply { throughput .lock_or_recover() @@ -721,16 +735,25 @@ pub(crate) async fn handle_sync( changed, carried, authors, + back_off, ))) }) .await .context("Sync task panicked")?? }; - let Some((caught_up, changed, carried, (authors, unauthored))) = applied else { + let Some((caught_up, changed, carried, (authors, unauthored), back_off)) = applied else { debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); return Ok(()); }; + if let Some(delay) = back_off { + debug!( + peer = %peer, + wait_ms = delay.as_millis(), + "Offered only changes refused before; answering later" + ); + state.request_sync_after(peer, delay); + } source::reported(state, peer, from_ready, None); |