Sync: stale reports spend no budget and log once a minute

crates/ot-server/src/engine.rs

@@ -562,6 +562,8 @@ pub struct Engine {
     sync_budget_bps: f64,
     sync_allowance: f64,
     sync_topped: Instant,
+    /// When reports too old for the wire were last logged.
+    sync_stale_warned: Option<Instant>,
     /// Lines of bearing waiting to be crossed into fixes.
     bearings: nonpoint::Bearings,
     /// Where each ended source track last reported, for scoring replays.
@@ -668,6 +670,7 @@ impl Engine {
             sync_budget_bps: 0.0,
             sync_allowance: 0.0,
             sync_topped: Instant::now(),
+            sync_stale_warned: None,
             bearings: Default::default(),
             #[cfg(test)]
             ended_on: HashMap::new(),
@@ -3038,15 +3041,18 @@ mod tests {
         .unwrap()
     }
 
-    /// Noon UTC yesterday: recent, so that report times stay inside what
-    /// the sync wire carries (a 32-bit millisecond offset from the send
-    /// time, about 24.8 days), and fixed for the length of a test run.
+    /// Noon UTC yesterday, as at the first call: recent, so that report
+    /// times stay inside what the sync wire carries (a 32-bit millisecond
+    /// offset from the send time, about 24.8 days), and the same for the
+    /// whole test run, even one that passes midnight.
     pub(super) fn t0() -> DateTime<Utc> {
-        let today = Utc::now().date_naive();
-        (today - chrono::Days::new(1))
-            .and_hms_opt(12, 0, 0)
-            .expect("noon exists")
-            .and_utc()
+        static T0: std::sync::OnceLock<DateTime<Utc>> = std::sync::OnceLock::new();
+        *T0.get_or_init(|| {
+            (Utc::now().date_naive() - chrono::Days::new(1))
+                .and_hms_opt(12, 0, 0)
+                .expect("noon exists")
+                .and_utc()
+        })
     }
 
     #[test]

crates/ot-server/src/engine/sync.rs

@@ -20,6 +20,9 @@ pub const PEER: &str = "peer:";
 /// How long another node's entry waits for its tracks.
 const PENDING_FOR_MS: i64 = 12 * 3600 * 1000;
 
+/// How often reports left out as too old for the wire are logged.
+const STALE_WARN_EVERY: std::time::Duration = std::time::Duration::from_secs(60);
+
 /// A command's logged form, from what it was given and what it answered:
 /// UIDs it minted or resolved are filled in, and who ran it is left to the
 /// entry.
@@ -726,16 +729,14 @@ impl Engine {
             self.common.site,
             ot_sync::Hlc::new(Utc::now().timestamp_millis() as u64, 0),
         );
-        let (reports, attrs) = self.within_budget(out.reports, out.attrs);
-        let stale: Vec<_> = reports
-            .iter()
-            .filter(|r| !ot_sync::wire::report_time_fits(r, hlc.ms()))
-            .map(|r| r.uid.to_string())
-            .collect();
-        if !stale.is_empty() {
-            tracing::warn!(tracks = ?stale,
-                "not reporting to other nodes: the estimate is more than 24 days from now");
-        }
+        // A report too far from now for the wire is left out before the
+        // budget, so it spends none of it.
+        let (fresh, stale): (Vec<Queued>, Vec<Queued>) = out
+            .reports
+            .into_iter()
+            .partition(|q| ot_sync::wire::report_time_fits(&q.report, hlc.ms()));
+        self.warn_stale(&stale);
+        let (reports, attrs) = self.within_budget(fresh, out.attrs);
         let mut msgs = ot_sync::wire::encode_reports(site, hlc, &reports);
         // As many UIDs as fit in a message once signed.
         let most = (ot_sync::wire::MAX_UNSIGNED - ot_sync::wire::HEADER - 2) / 7;
@@ -758,6 +759,27 @@ impl Engine {
     /// tracks, state changes and retirements, then the largest drifts, then
     /// heartbeats, then attributes. What does not fit waits for the next
     /// flush (it is due again). No budget: everything.
+    /// Say, at most once a minute, that reports are left out because their
+    /// estimates are too far from now for the wire (a replay of old data,
+    /// say): how many, and a few of their tracks.
+    fn warn_stale(&mut self, stale: &[Queued]) {
+        if stale.is_empty()
+            || self
+                .sync_stale_warned
+                .is_some_and(|at| at.elapsed() < STALE_WARN_EVERY)
+        {
+            return;
+        }
+        self.sync_stale_warned = Some(std::time::Instant::now());
+        let some: Vec<String> = stale
+            .iter()
+            .take(5)
+            .map(|q| q.report.uid.to_string())
+            .collect();
+        tracing::warn!(count = stale.len(), tracks = ?some,
+            "not reporting tracks to other nodes: their estimates are more than 24 days from now");
+    }
+
     fn within_budget(
         &mut self,
         mut reports: Vec<Queued>,