1 files changed, 74 insertions, 0 deletions

diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index 79e5c83..fd88167 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -29,6 +29,52 @@ struct Inner {
     stale_hello_replies: u64,
     saved_size: Option<SavedSize>,
     measuring: bool,
+    propagation: Propagation,
+}
+
+/// Upper bounds (inclusive, milliseconds) of the propagation delay buckets; a last
+/// bucket takes the rest.
+const PROPAGATION_BUCKETS_MS: [u64; 11] = [
+    100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000, 60_000, 300_000, 3_600_000,
+];
+
+/// How long changes written elsewhere took to reach us: from the writer's commit (its
+/// clock) to applying them here (ours), so clock skew between the two is part of it.
+#[derive(Debug, Clone, Copy, Default)]
+pub(crate) struct Propagation {
+    /// Per bucket of `PROPAGATION_BUCKETS_MS`, then the rest; not cumulative.
+    pub(crate) buckets: [u64; PROPAGATION_BUCKETS_MS.len() + 1],
+    pub(crate) count: u64,
+    pub(crate) sum_ms: u64,
+    /// Changes stamped later than our clock said on arrival (counted in the first
+    /// bucket): the writer's clock is ahead of ours.
+    pub(crate) ahead: u64,
+}
+
+impl Propagation {
+    fn record(&mut self, delay_ms: i64) {
+        if delay_ms < 0 {
+            self.ahead += 1;
+        }
+        let delay_ms = delay_ms.max(0) as u64;
+        let bucket = PROPAGATION_BUCKETS_MS
+            .iter()
+            .position(|&bound| delay_ms <= bound)
+            .unwrap_or(PROPAGATION_BUCKETS_MS.len());
+        self.buckets[bucket] += 1;
+        self.count += 1;
+        self.sum_ms += delay_ms;
+    }
+
+    /// Each bucket's upper bound (None for the last) and count.
+    pub(crate) fn buckets(&self) -> impl Iterator<Item = (Option<u64>, u64)> + '_ {
+        PROPAGATION_BUCKETS_MS
+            .iter()
+            .copied()
+            .map(Some)
+            .chain([None])
+            .zip(self.buckets.iter().copied())
+    }
 }
 
 /// The document's size as saved (compressed), when it was measured and how long that
@@ -57,6 +103,7 @@ pub(crate) struct Snapshot {
     pub(crate) catch_up_sources: u32,
     pub(crate) max_heads: usize,
     pub(crate) stale_hello_replies: u64,
+    pub(crate) propagation: Propagation,
 }
 
 impl Default for NodeStats {
@@ -93,6 +140,14 @@ impl NodeStats {
         inner.max_heads = inner.max_heads.max(count);
     }
 
+    /// Changes from elsewhere were applied, `delays_ms` after they were committed.
+    pub(crate) fn propagated(&self, delays_ms: impl IntoIterator<Item = i64>) {
+        let mut inner = self.lock();
+        for delay_ms in delays_ms {
+            inner.propagation.record(delay_ms);
+        }
+    }
+
     pub(crate) fn stale_hello_reply(&self) {
         self.lock().stale_hello_replies += 1;
     }
@@ -131,6 +186,7 @@ impl NodeStats {
             catch_up_sources: inner.catch_up_sources,
             max_heads: inner.max_heads,
             stale_hello_replies: inner.stale_hello_replies,
+            propagation: inner.propagation,
         }
     }
 }
@@ -177,3 +233,21 @@ pub(crate) fn measure_saved_size(state: &AppState) {
         state.stats.finish_measuring(measured);
     });
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn propagation_buckets() {
+        let mut propagation = Propagation::default();
+        for delay_ms in [-20, 100, 101, 7_200_000] {
+            propagation.record(delay_ms);
+        }
+        let buckets: Vec<_> = propagation.buckets().filter(|(_, n)| *n > 0).collect();
+        assert_eq!(buckets, [(Some(100), 2), (Some(250), 1), (None, 1)]);
+        assert_eq!(propagation.ahead, 1);
+        assert_eq!(propagation.count, 4);
+        assert_eq!(propagation.sum_ms, 7_200_201);
+    }
+}