add throughput stats

1 files changed, 14 insertions, 1 deletions

diff --git a/src/state.rs b/src/state.rs
index b8944ab..d8f0bbf 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -16,6 +16,7 @@ use crate::backoff::Backoff;
 use crate::document::replica::{Epoch, Replica};
 use crate::remake::{Host, PeerSummary, RemakeState};
 use crate::swarm::sync::{ChangeSet, SyncEvent};
+use crate::swarm::throughput::{SyncBytes, ThroughputTable};
 use crate::transport;
 use crate::transport::firewall::Firewall;
 
@@ -204,6 +205,9 @@ pub(crate) struct SyncSession {
 #[derive(Clone, Debug)]
 pub(crate) struct SyncReplyWait {
     pub(crate) deadline: Instant,
+    /// When the first of the unanswered messages was generated, and the sync bytes
+    /// exchanged with the peer by then (`swarm::throughput`).
+    pub(crate) started: (Instant, SyncBytes),
     /// Changes first sent in the unanswered messages.
     pub(crate) sent: Vec<automerge::ChangeHash>,
 }
@@ -266,13 +270,20 @@ impl SyncSession {
     }
 
     /// A sync message went out; `sent` are the changes it carried for the first time.
-    pub(crate) fn expect_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) {
+    /// `bytes` are the sync bytes exchanged with the peer before it.
+    pub(crate) fn expect_reply(
+        &mut self,
+        sent: Vec<automerge::ChangeHash>,
+        now: Instant,
+        bytes: SyncBytes,
+    ) {
         match &mut self.reply {
             // Still waiting for an earlier one: keep its deadline, remember these too.
             Some(wait) => wait.sent.extend(sent),
             None => {
                 self.reply = Some(SyncReplyWait {
                     deadline: now + self.reply_backoff.current(),
+                    started: (now, bytes),
                     sent,
                 });
             }
@@ -586,6 +597,8 @@ pub(crate) struct AppState {
     pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>,
     /// Round-trip times to peers (`swarm::rtt`).
     pub(crate) rtt: Arc<std::sync::Mutex<crate::swarm::rtt::RttTable>>,
+    /// Sync throughput with peers (`swarm::throughput`).
+    pub(crate) throughput: Arc<std::sync::Mutex<ThroughputTable>>,
     /// The neighbour we are pulling changes from, and since when (`swarm::relay`).
     pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>,
     /// Our neighbours, once known (`swarm::topology`).