aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/sync.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/sync.rs')
-rw-r--r--src/swarm/sync.rs24
1 files changed, 17 insertions, 7 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index c8564c2..b881609 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -128,7 +128,16 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let ready = state.is_ready().await;
// Before this message is counted.
let bytes_before = SyncBytes::of(&state.traffic, peer);
- let delta_threshold = state.blobs.delta_threshold();
+ // Lower towards a peer a message that large wouldn't reach whole: a slow link
+ // takes a delta chunk by chunk, where a message whole would fail and be sent
+ // again, whole, every time.
+ let rate = state
+ .throughput
+ .lock_or_recover()
+ .get(peer)
+ .map(|estimate| estimate.bytes_per_sec);
+ let failed = peer_state_mutex.lock().await.failed_whole(Instant::now());
+ let delta_threshold = state.blobs.delta_threshold_for(rate, failed);
// The session the message comes from, for its send's outcome.
let (session, generated) = tokio::task::spawn_blocking(
@@ -252,12 +261,13 @@ async fn send_generated(
Ok(())
}
Err(error) => {
- let Some(delay) = peer_state_mutex
- .lock()
- .await
- .sync
- .send_failed(session, Instant::now())
- else {
+ let delay = {
+ let mut peer_state = peer_state_mutex.lock().await;
+ let now = Instant::now();
+ peer_state.whole_failed(encoded.payload_len, now);
+ peer_state.sync.send_failed(session, now)
+ };
+ let Some(delay) = delay else {
// The new session may not have opened yet.
debug!("Sync message refused, but a reset already discarded it");
state.request_sync(peer);