time blob messages

4 files changed, 46 insertions, 4 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index ce9a2bd..98dee66 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -278,6 +278,7 @@ async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) {
         epoch_seq,
         id,
         body,
+        sent: state.clock.now_millis(),
     });
     if let Err(error) = state.transport.send_message(peer, &message).await {
         debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob answer failed");
@@ -347,6 +348,7 @@ async fn request(
             chunks,
             delta: delta.cloned(),
         },
+        sent: state.clock.now_millis(),
     });
     if let Err(error) = state.transport.send_message(peer, &message).await {
         debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob request failed");
diff --git a/src/swarm/hop.rs b/src/swarm/hop.rs
index 4cd413f..1084ddd 100644
--- a/src/swarm/hop.rs
+++ b/src/swarm/hop.rs
@@ -1,5 +1,6 @@
-//! How long sync messages take from a peer to us, per hop: from the peer's clock when it
-//! generated one (`SyncEnvelope::sent`) to ours when we start handling it, the peer's
+//! How long sync messages (and BLOB messages) take from a peer to us, per hop: from the
+//! peer's clock when it generated one (`SyncEnvelope::sent`, `BlobMessage::sent`) to
+//! ours when we start handling it, the peer's
 //! clock offset taken off (`Transport::clock_offset`, from PING and PONG). So the
 //! transport, the network and the queues at both ends, which a PING's round trip leaves
 //! out (PINGs don't wait behind sync messages).
@@ -61,6 +62,26 @@ impl HopDelays {
     }
 }
 
+/// A message from `peer` that it `sent` (its clock) is here: how long it took, recorded
+/// with the peer's hop delays (`PeerState::hop_delays`); None while its clock isn't
+/// known.
+pub(crate) async fn note(
+    state: &crate::state::AppState,
+    peer: &str,
+    sent: crate::clock::UnixMillis,
+) -> Option<i64> {
+    let offset = state.transport.clock_offset(peer)?;
+    let delay = delay_ms(sent, state.clock.now_millis(), offset.offset_ms);
+    if let Some(peer_state) = state.get_peer_state(peer).await {
+        peer_state
+            .lock()
+            .await
+            .hop_delays
+            .record(delay, Instant::now());
+    }
+    Some(delay)
+}
+
 /// How long a sync message `sent` by a peer (its clock) took to come, as `arrived` (ours),
 /// with the peer's clock `offset_ms` ahead of ours.
 pub(crate) fn delay_ms(
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 53f1aa3..f43a45a 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -15,6 +15,7 @@ use super::keepalive::{
     heal_if_diverged, heard_from, local_status, note_alive, note_status, status_received,
 };
 use super::lanes::SyncLanes;
+use crate::locks::MutexExt;
 use crate::state::AppState;
 use crate::swarm::sync::SyncEvent;
 use crate::transport::{self, ParsedMessage};
@@ -151,6 +152,24 @@ pub(crate) async fn handle_message(
             lanes.enqueue(&peer, envelope);
         }
         transport::ParsedMessage::Blob(blob) => {
+            // How long it took to come, and of a chunk, the throughput (it was asked
+            // for: the request's round trip is no exchange of sync bytes to time).
+            if let Some(delay_ms) = crate::swarm::hop::note(state, &peer, blob.sent).await
+                && delay_ms > 0
+                && let transport::BlobBody::Chunk { data, .. } = &blob.body
+            {
+                let rtt =
+                    state
+                        .lock_rtt()
+                        .effective(&peer, Instant::now(), state.timing.rtt_fresh_for());
+                state.throughput.lock_or_recover().message(
+                    &peer,
+                    data.len() as u64,
+                    Duration::from_millis(delay_ms as u64),
+                    Instant::now(),
+                    rtt,
+                );
+            }
             crate::swarm::blob::handle(state, &peer, blob).await;
         }
         transport::ParsedMessage::Status(remote) => {
diff --git a/src/swarm/throughput.rs b/src/swarm/throughput.rs
index 479239a..ec50693 100644
--- a/src/swarm/throughput.rs
+++ b/src/swarm/throughput.rs
@@ -1,8 +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; 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.
+//! delivered, which gets no answer to time), and from a blob chunk (`swarm::blob`), 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