diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/signatures.rs | 2 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 33 |
2 files changed, 29 insertions, 6 deletions
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); |