From af5727b5e1f1ca95f50630800db147e1b94716b5 Mon Sep 17 00:00:00 2001 From: Parker Hornstein Date: Wed, 21 Oct 2026 16:25:06 -0500 Subject: [PATCH] Sync: stale reports spend no budget and log once a minute Follow-ups to #106 from review: - Reports too far from now for the wire are left out before the sync byte budget is charged, so current reports are not deferred for bytes never sent. - They are logged at most once a minute, with a count and a few tracks, so replaying old data does not flood the log. - The engine tests' t0 is computed once per run, so a run across midnight UTC keeps one time. Co-Authored-By: Claude Opus 5.5 --- crates/ot-server/src/engine.rs | 22 +++++++++------ crates/ot-server/src/engine/sync.rs | 42 ++++++++++++++++++++++------- 2 files changed, 46 insertions(+), 18 deletions(-) diff --git a/crates/ot-server/src/engine.rs b/crates/ot-server/src/engine.rs index 52659fa..3f8fd4b 100644 --- a/crates/ot-server/src/engine.rs +++ b/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, /// 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 { - 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> = 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] diff --git a/crates/ot-server/src/engine/sync.rs b/crates/ot-server/src/engine/sync.rs index 4762a05..0bf2804 100644 --- a/crates/ot-server/src/engine/sync.rs +++ b/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, Vec) = 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 = 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,