From f137a43b44684959a5e1de1f1e54566d8995926f Mon Sep 17 00:00:00 2001
From: tv <tv@krebsco.de>
Date: Wed, 4 Nov 2026 22:15:34 +0100
Subject: throughput from delivered message

---
 src/swarm/throughput.rs | 52 +++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 50 insertions(+), 2 deletions(-)

(limited to 'src/swarm/throughput.rs')

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();
-- 
cgit v1.3.1