Sync: leave out a report too old for the wire, never misdate it

crates/ot-server/src/engine.rs

@@ -3027,7 +3027,6 @@ impl Engine {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use chrono::TimeZone;
 
     pub(super) fn obs(at: DateTime<Utc>, name: &str, domain: Domain) -> Observation {
         serde_json::from_value(serde_json::json!({
@@ -3039,8 +3038,15 @@ 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.
     pub(super) fn t0() -> DateTime<Utc> {
-        Utc.with_ymd_and_hms(2026, 9, 24, 12, 0, 0).unwrap()
+        let today = Utc::now().date_naive();
+        (today - chrono::Days::new(1))
+            .and_hms_opt(12, 0, 0)
+            .expect("noon exists")
+            .and_utc()
     }
 
     #[test]

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

@@ -727,6 +727,15 @@ impl Engine {
             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");
+        }
         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;

crates/ot-sync/src/wire.rs

@@ -265,9 +265,29 @@ fn metres_u16(v: f64) -> u16 {
     }
 }
 
+/// A report's time offset from the message's, when it fits the wire's 32
+/// bits (about 24.8 days either side).
+fn time_offset(time_ms: i64, base_ms: u64) -> Option<i32> {
+    time_ms
+        .checked_sub(i64::try_from(base_ms).ok()?)
+        .and_then(|d| i32::try_from(d).ok())
+}
+
+/// Whether a report can travel in a message sent at `base_ms`: its time
+/// must be within about 24.8 days of the send time. [`encode_reports`]
+/// leaves out a report that cannot, rather than send a wrong time.
+pub fn report_time_fits(r: &Report, base_ms: u64) -> bool {
+    time_offset(r.time_ms, base_ms).is_some()
+}
+
 fn write_report(w: &mut W, r: &Report, base_ms: u64) {
     w.uid(r.uid);
-    let dt = (r.time_ms - base_ms as i64).clamp(i32::MIN.into(), i32::MAX.into()) as i32;
+    // Callers keep only reports that fit; a clamp is the last resort.
+    let dt = time_offset(r.time_ms, base_ms).unwrap_or(if r.time_ms < base_ms as i64 {
+        i32::MIN
+    } else {
+        i32::MAX
+    });
     w.i32(dt);
     w.i32((r.lat * 1e7).round() as i32);
     w.i32((r.lon * 1e7).round() as i32);
@@ -469,9 +489,15 @@ pub fn sender(bytes: &[u8]) -> Result<SiteCode, WireError> {
 }
 
 /// Reports in as few messages as fit in [`MAX_UNSIGNED`] bytes each (so
-/// in [`MAX_MESSAGE`] once signed).
+/// in [`MAX_MESSAGE`] once signed). A report whose time is too far from
+/// `hlc` for the wire ([`report_time_fits`]) is left out.
 pub fn encode_reports(site: SiteCode, hlc: Hlc, reports: &[Report]) -> Vec<Vec<u8>> {
-    chunks(reports, HEADER + 1, 255, |r| {
+    let reports: Vec<Report> = reports
+        .iter()
+        .filter(|r| report_time_fits(r, hlc.ms()))
+        .cloned()
+        .collect();
+    chunks(&reports, HEADER + 1, 255, |r| {
         let mut w = W(Vec::new());
         write_report(&mut w, r, hlc.ms());
         w.0.len()
@@ -680,6 +706,27 @@ mod tests {
         assert_eq!(Message::decode(&m.encode()).unwrap(), m);
     }
 
+    #[test]
+    fn a_report_too_old_for_the_wire_is_left_out_not_misdated() {
+        let base = NOW.ms() as i64;
+        let mut at_edge = report(1);
+        at_edge.time_ms = base + i64::from(i32::MIN);
+        let mut too_old = report(2);
+        too_old.time_ms = base + i64::from(i32::MIN) - 1;
+        let mut too_new = report(3);
+        too_new.time_ms = base + i64::from(i32::MAX) + 1;
+        assert!(report_time_fits(&at_edge, NOW.ms()));
+        assert!(!report_time_fits(&too_old, NOW.ms()));
+        assert!(!report_time_fits(&too_new, NOW.ms()));
+
+        let msgs = encode_reports(site("BBB"), NOW, &[too_old, at_edge.clone(), too_new]);
+        assert_eq!(msgs.len(), 1);
+        match Message::decode(&msgs[0]).unwrap().body {
+            Body::Reports(r) => assert_eq!(r, vec![at_edge]),
+            other => panic!("{other:?}"),
+        }
+    }
+
     #[test]
     fn many_reports_split_into_messages_that_fit() {
         let rs: Vec<Report> = (1..=100).map(report).collect();