propagate clock figures

7 files changed, 26 insertions, 11 deletions

diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 1b20ee9..6eab2d1 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -1781,7 +1781,7 @@ components:
           description: The peers currently listed.
         propagation:
           type: object
-          description: Time from a change's commit elsewhere until applied here, once caught up (includes clock skew).
+          description: Time from a change's commit elsewhere until applied here, once caught up, by the peers' clocks (includes the writer's clock skew).
           required: [count, sumMs, ahead, buckets]
           properties:
             count:
@@ -1790,7 +1790,7 @@ components:
               type: integer
             ahead:
               type: integer
-              description: Changes stamped later than our clock on arrival.
+              description: Changes stamped later than their arrival (writer's clock ahead).
             buckets:
               type: array
               description: Not cumulative; the last bucket has no bound.
diff --git a/src/api/stats.rs b/src/api/stats.rs
index d274fd9..558507a 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -67,13 +67,14 @@ struct DiskStats {
 }
 
 /// 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.
+/// once caught up, by the peers' clocks (`swarm::skew`). Includes the writer's clock
+/// skew.
 #[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).
+    /// Changes stamped later than their arrival (writer's clock ahead).
     ahead: u64,
     /// Not cumulative; the last bucket has no bound.
     buckets: Vec<PropagationBucket>,
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 6774e74..0a1d75d 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -744,7 +744,7 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String {
     out.family(
         "propagation_delay_seconds",
         "histogram",
-        "Time from a change's commit elsewhere until applied here (includes clock skew).",
+        "Time from a change's commit elsewhere until applied here, by the peers' clocks (includes the writer's clock skew).",
     );
     out.histogram(
         "propagation_delay_seconds",
@@ -759,7 +759,7 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String {
     out.single(
         "propagation_clock_ahead_total",
         "counter",
-        "Changes stamped later than this node's clock on arrival.",
+        "Changes stamped later than their arrival, by the peers' clocks.",
         propagation.ahead as f64,
     );
 
diff --git a/src/node.rs b/src/node.rs
index e70b205..ac4c252 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -116,6 +116,7 @@ impl Node {
             alias: config.alias,
             electorate_window: config.electorate_window,
             clock_skew_warn: config.clock_skew_warn,
+            clock_ahead_ms: Default::default(),
             clock_skew_warned: Default::default(),
             blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)),
             links: Default::default(),
diff --git a/src/state/mod.rs b/src/state/mod.rs
index cf50b2a..ec8823a 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -116,6 +116,9 @@ pub(crate) struct AppState {
     /// This node's clock is off when it differs from most of its peers' by more than
     /// this (`--clock-skew-warn`, `swarm::skew`); 0: never.
     pub(crate) clock_skew_warn: Duration,
+    /// How far this node's clock is ahead of its peers' (their median, `swarm::skew`), as
+    /// last measured; 0 before. The propagation figures take it off.
+    pub(crate) clock_ahead_ms: Arc<std::sync::atomic::AtomicI64>,
     /// When the log last said this node's clock is off, while it is (`swarm::skew`).
     pub(crate) clock_skew_warned: Arc<std::sync::Mutex<Option<tokio::time::Instant>>>,
     /// Chunked transfers served and fetched (`swarm::blob`).
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