large messages over slow links are deltas

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);