time blob messages

8 files changed, 55 insertions, 6 deletions

diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index 876c386..8ee1ee5 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -431,7 +431,7 @@ function renderPeers() {
   const table = document.getElementById("peers");
   const headers = { reaches: "How many members besides this node it reaches (an active " +
     "session with each), as it said; +: those of them this node doesn't reach itself (it " +
-    "is a way to them)", hop: "How long its sync messages take to come, by its clock and " +
+    "is a way to them)", hop: "How long its sync and blob messages take to come, by its clock and " +
     "this node's (its offset taken off): this minute's mean, and the last 15 minutes' as " +
     "a line; good to half the round trip, but the trend exactly" };
   const rows = [el("tr", {}, ...["peer", "status", "seen", "via", "rtt", "hop", "clock",
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
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index ae6b593..4ddfe32 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -295,6 +295,9 @@ pub(crate) struct BlobMessage {
     pub(crate) epoch_seq: u32,
     pub(crate) id: BlobId,
     pub(crate) body: BlobBody,
+    /// When the sender sent it, by its clock (as `SyncEnvelope::sent`): the receiver
+    /// takes how long it took to come (`swarm::hop`), and of a chunk, the throughput.
+    pub(crate) sent: crate::clock::UnixMillis,
 }
 
 #[derive(Clone, Serialize, Deserialize)]
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 3e33a9f..3552edf 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -752,6 +752,7 @@ mod tests {
                 index: 5,
                 data: vec![1u8; 64 * 1024].into(),
             },
+            sent: crate::clock::UnixMillis(6),
         });
         let bytes = wire::encode(&chunk).unwrap();
         // A 64 KiB chunk costs only a few bytes of framing.
@@ -773,6 +774,7 @@ mod tests {
                 chunk_size: 64,
                 hashes: vec![[1; 32], [2; 32]],
             },
+            sent: crate::clock::UnixMillis(6),
         });
         let decoded = decode_blob(&wire::encode(&manifest).unwrap());
         assert!(matches!(
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index d1b3353..b3a8ee9 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 26;
+pub(crate) const VERSION: u8 = 27;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -420,6 +420,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
             epoch_seq: 1,
             id: BlobId::Content([5; 32]),
             body: BlobBody::Missing,
+            sent: crate::clock::UnixMillis(9),
         }),
         ParsedMessage::Who(proof.clone()),
         ParsedMessage::Me(proof),
@@ -603,6 +604,7 @@ mod tests {
             epoch_seq: 0,
             id: BlobId::Content([1; 32]),
             body: BlobBody::Missing,
+            sent: crate::clock::UnixMillis(2),
         });
         let payload = encode(&blob).unwrap();
         assert!(matches!(decode(&payload).unwrap(), ParsedMessage::Blob(_)));