add propagation stats

4 files changed, 126 insertions, 1 deletions

diff --git a/src/api/stats.rs b/src/api/stats.rs
index e426c7a..3822b5f 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -24,6 +24,27 @@ pub(crate) struct StatsResponse {
     traffic: PeerTraffic,
     /// The peers listed in /peers together (a forgotten peer's are gone).
     incidents: Incidents,
+    propagation: PropagationStats,
+}
+
+/// How long changes written elsewhere took from their commit there until applied here,
+/// once caught up. Includes the clock skew between the writer and us.
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct PropagationStats {
+    count: u64,
+    sum_ms: u64,
+    /// Changes stamped later than our clock on arrival (writer's clock ahead).
+    ahead: u64,
+    /// Not cumulative; the last bucket has no bound.
+    buckets: Vec<PropagationBucket>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct PropagationBucket {
+    le_ms: Option<u64>,
+    count: u64,
 }
 
 #[derive(Debug, Serialize)]
@@ -139,5 +160,15 @@ pub(crate) async fn get_stats(State(state): State<AppState>) -> Json<StatsRespon
         stale_hello_replies: node.stale_hello_replies,
         traffic: state.traffic.total(),
         incidents,
+        propagation: PropagationStats {
+            count: node.propagation.count,
+            sum_ms: node.propagation.sum_ms,
+            ahead: node.propagation.ahead,
+            buckets: node
+                .propagation
+                .buckets()
+                .map(|(le_ms, count)| PropagationBucket { le_ms, count })
+                .collect(),
+        },
     })
 }
diff --git a/src/state.rs b/src/state.rs
index dfabf8f..b8944ab 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -803,7 +803,12 @@ impl AppState {
                 match &result {
                     // One API write = one Automerge change.
                     Ok(_) => {
-                        replica.doc.commit();
+                        // Stamped with our clock (milliseconds), so receivers can tell
+                        // how long the change took to reach them (`swarm::stats`).
+                        replica.doc.commit_with(
+                            automerge::transaction::CommitOptions::default()
+                                .with_time(unix_millis() as i64),
+                        );
                     }
                     // Discard partial ops (e.g. old value deleted, new one not written).
                     Err(_) => {
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);
+    }
+}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index adf8146..09e10f4 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -307,6 +307,8 @@ pub(crate) async fn handle_sync(
         let peer_state_mutex = Arc::clone(&peer_state_mutex);
         let stats = Arc::clone(&state.stats);
         let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
+        // While catching up, everything arrives late; that says nothing about propagation.
+        let measure_propagation = !check_caught_up;
         let epoch = envelope.epoch;
         let message = envelope.message;
         let peer = peer.to_owned();
@@ -338,6 +340,19 @@ pub(crate) async fn handle_sync(
             let heads_after = replica.doc.get_heads();
             stats.heads_seen(heads_after.len());
             let changed = heads_after != heads_before;
+            if changed && measure_propagation {
+                let now_ms = crate::state::unix_millis() as i64;
+                // Time 0: not stamped (snapshots, older nodes).
+                let delays: Vec<i64> = replica
+                    .doc
+                    .get_changes(&heads_before)
+                    .iter()
+                    .map(|change| change.timestamp())
+                    .filter(|&time| time > 0)
+                    .map(|time| now_ms - time)
+                    .collect();
+                stats.propagated(delays);
+            }
 
             Ok(Some((
                 // Only from a ready peer: a peer still catching up has empty heads, which