large messages over slow links are deltas

1 files changed, 23 insertions, 0 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 3f66d50..ce9a2bd 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -87,6 +87,10 @@ pub(crate) struct Blobs {
 /// lost message costs too much to send again whole, and a slow link may not deliver it
 /// within the transport's delivery limit.
 const DELTA_THRESHOLD_CHUNKS: usize = 16;
+/// The longest a sync message should take whole at the throughput measured to its peer
+/// (`delta_threshold_for`): well within what a transport allows a message before its send
+/// fails (10 s, plus its size at 1 MiB/s, on libp2p).
+const WHOLE_WITHIN: Duration = Duration::from_secs(5);
 
 /// The largest sync message accepted: the threshold of a sender with the largest chunks
 /// it may have configured. Anything larger comes from a node that doesn't keep to it.
@@ -115,6 +119,25 @@ impl Blobs {
         DELTA_THRESHOLD_CHUNKS * self.chunk_size as usize
     }
 
+    /// [`delta_threshold`](Self::delta_threshold) for one peer: lower where a message
+    /// that large wouldn't make it whole. `rate`: the measured throughput to it, bytes
+    /// per second (`swarm::throughput`), at which a message whole should take at most
+    /// `WHOLE_WITHIN`; `failed`: the largest message to it that didn't make it lately
+    /// (`PeerState::failed_whole`), of which messages half as large go as deltas, if it
+    /// was larger than two chunks (smaller ones are lost for other reasons than their
+    /// size). Never below a chunk: a delta then is no smaller per message.
+    pub(crate) fn delta_threshold_for(&self, rate: Option<f64>, failed: Option<usize>) -> usize {
+        let chunk = self.chunk_size as usize;
+        let mut threshold = self.delta_threshold();
+        if let Some(rate) = rate.filter(|rate| rate.is_finite() && *rate > 0.0) {
+            threshold = threshold.min((rate * WHOLE_WITHIN.as_secs_f64()) as usize);
+        }
+        if let Some(failed) = failed.filter(|failed| *failed > 2 * chunk) {
+            threshold = threshold.min(failed / 2);
+        }
+        threshold.max(chunk)
+    }
+
     /// Starts downloading a delta offered by `peer`; false if one from it is running.
     pub(crate) fn start_delta(&self, peer: &str) -> bool {
         self.deltas.lock_or_recover().insert(peer.to_owned())