propagate clock figures

2 files changed, 15 insertions, 5 deletions

diff --git a/src/swarm/skew.rs b/src/swarm/skew.rs
index 1b98ff5..af11e5e 100644
--- a/src/swarm/skew.rs
+++ b/src/swarm/skew.rs
@@ -1,8 +1,9 @@
 //! Whether this node's clock agrees with its peers' (`Transport::clock_offset`, from
 //! their PONGs). Each peer's clock is known within half a round trip; this node's own
 //! is off when it differs from most of its peers' by more than `--clock-skew-warn` for
-//! sure, then the log says so (every `WARN_EVERY` while it lasts). A diagnostic only:
-//! setting clocks is the operating system's (NTP).
+//! sure, then the log says so (every `WARN_EVERY` while it lasts). Clocks are the
+//! operating system's to set (NTP); the propagation figures only take this node's own
+//! offset off (`AppState::clock_ahead_ms`).
 //!
 //! Two nodes alone can't tell whose clock is off: each warns.
 
@@ -75,10 +76,16 @@ pub(crate) async fn measure(state: &AppState) -> Skew {
     of(&offsets, state.clock_skew_warn)
 }
 
-/// Says in the log when this node's clock is off, and when it agrees again. From the
-/// peer worker, every few seconds.
+/// Notes how far this node's clock is ahead of its peers' (`AppState::clock_ahead_ms`),
+/// and says in the log when it is off, and when it agrees again. From the peer worker,
+/// every few seconds.
 pub(crate) async fn check(state: &AppState) {
     let skew = measure(state).await;
+    if let Some(offset_ms) = skew.offset_ms {
+        state
+            .clock_ahead_ms
+            .store(offset_ms, std::sync::atomic::Ordering::Relaxed);
+    }
     let now = Instant::now();
     let mut warned = state.clock_skew_warned.lock_or_recover();
     if skew.off {
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 51144ac..18c017c 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -350,6 +350,7 @@ pub(crate) async fn handle_sync(
         let message = payload;
         let peer = peer.to_owned();
         let clock = Arc::clone(&state.clock);
+        let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
 
         // Some((caught_up, heads changed)) when applied, None when the message belongs
         // to another epoch.
@@ -386,7 +387,9 @@ pub(crate) async fn handle_sync(
             stats.heads_seen(heads_after.len());
             let changed = heads_after != heads_before;
             if changed && measure_propagation {
-                let now_ms = clock.now_millis().as_u64() as i64;
+                // Now by the peers' clocks (their median): this node's own clock skew
+                // doesn't count; the writer's still does.
+                let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms;
                 // Time 0: not stamped (snapshots, older nodes).
                 let delays: Vec<i64> = replica
                     .doc