2 files changed, 59 insertions, 2 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index a9d98b0..de2d86d 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -440,6 +440,15 @@ pub(crate) async fn handle_sync(
                 throughput
                     .lock_or_recover()
                     .exchange(&peer, wait.started, arrived, rtt);
+            } else if let Some(delay_ms) = hop_delay.filter(|ms| *ms > 0) {
+                // Nothing awaited: as long as it took to come.
+                throughput.lock_or_recover().message(
+                    &peer,
+                    payload_len as u64,
+                    std::time::Duration::from_millis(delay_ms as u64),
+                    arrived.0,
+                    rtt,
+                );
             }
             peer_state.sync.reply_settled();
 
diff --git a/src/swarm/throughput.rs b/src/swarm/throughput.rs
index 85a3fc8..479239a 100644
--- a/src/swarm/throughput.rs
+++ b/src/swarm/throughput.rs
@@ -1,6 +1,8 @@
 //! Sync throughput with peers, measured from sync exchanges we have anyway: from sending
 //! a sync message until the peer's answer arrives, the sync bytes that went either way
-//! with that peer. No transfers of our own.
+//! with that peer; and from a large message we awaited nothing for (one taken as
+//! delivered, which gets no answer to time), how long it took to come (`swarm::hop`). No
+//! transfers of our own.
 //!
 //! Only exchanges of at least `MIN_BYTES` count; smaller ones measure the round trip,
 //! not the throughput. The round trip (`swarm::rtt`) is taken off the elapsed time, but
@@ -81,7 +83,34 @@ impl ThroughputTable {
         if transfer.is_zero() {
             return;
         }
-        let sample = bytes as f64 / transfer.as_secs_f64();
+        self.sample(peer, bytes as f64 / transfer.as_secs_f64(), arrived);
+    }
+
+    /// A sync message of `bytes` from `peer` arrived at `arrived`, `took` after it was
+    /// generated (`swarm::hop`: good to half the round trip, which is taken off here as
+    /// an exchange's round trip is, never more than half). Of messages we awaited
+    /// nothing for, else the exchange counts it.
+    pub(crate) fn message(
+        &mut self,
+        peer: &str,
+        bytes: u64,
+        took: Duration,
+        arrived: Instant,
+        rtt: Option<Duration>,
+    ) {
+        if bytes < MIN_BYTES {
+            return;
+        }
+        let transfer = took
+            .saturating_sub(rtt.unwrap_or_default() / 2)
+            .max(took / 2);
+        if transfer.is_zero() {
+            return;
+        }
+        self.sample(peer, bytes as f64 / transfer.as_secs_f64(), arrived);
+    }
+
+    fn sample(&mut self, peer: &str, sample: f64, arrived: Instant) {
         let estimate = self.peers.entry(peer.to_owned()).or_insert(Estimate {
             bytes_per_sec: sample,
             samples: 0,
@@ -124,6 +153,25 @@ mod tests {
         assert!(table.get("a").is_none());
     }
 
+    /// One-way: half the round trip taken off, at most half the time; small ones don't
+    /// count.
+    #[test]
+    fn messages_count_one_way() {
+        let mut table = ThroughputTable::default();
+        let now = Instant::now();
+        let second = Duration::from_secs(1);
+        table.message("a", 10_000, second, now, None);
+        assert!(table.get("a").is_none());
+        // 1 MB in 1.5 s, of which half a 1 s round trip: 1 MB/s.
+        table.message("a", 1_000_000, second * 3 / 2, now, Some(second));
+        let a = table.get("a").unwrap();
+        assert!((a.bytes_per_sec - 1_000_000.0).abs() < 1.0);
+        // A round trip longer than twice the time takes off only half of it.
+        table.message("b", 1_000_000, second, now, Some(second * 5));
+        let b = table.get("b").unwrap();
+        assert!((b.bytes_per_sec - 2_000_000.0).abs() < 1.0);
+    }
+
     #[test]
     fn round_trip_is_taken_off_but_at_most_half() {
         let mut table = ThroughputTable::default();