diff --git a/src/core/payload.rs b/src/core/payload.rs index 971d503..e518376 100644 --- a/src/core/payload.rs +++ b/src/core/payload.rs @@ -196,6 +196,7 @@ mod tests { vec![(ProviderId::Claude, PollOutcome::Ok(windows))], &strings(), Instant::now(), + SystemTime::now(), ); state } @@ -251,6 +252,7 @@ mod tests { vec![(ProviderId::Claude, PollOutcome::AuthFailed)], &strings(), Instant::now(), + SystemTime::now(), ); let snap = state.snapshots.get(&ProviderId::Claude); let p = bubble_payload(ProviderId::Claude, snap, &strings(), false, CANVAS, now()); @@ -269,6 +271,7 @@ mod tests { vec![(ProviderId::Claude, PollOutcome::Failed)], &strings(), Instant::now(), + SystemTime::now(), ); let snap = state.snapshots.get(&ProviderId::Claude); let panel = panel_payload(ProviderId::Claude, snap, &strings(), false); diff --git a/src/core/settings.rs b/src/core/settings.rs index b037393..5e7370e 100644 --- a/src/core/settings.rs +++ b/src/core/settings.rs @@ -231,10 +231,17 @@ pub fn load() -> Settings { load_from(&settings_path()) } -/// [`load`] from an explicit path. +/// [`load`] from an explicit path. A file that is not a JSON object at all +/// is copied to `settings.json.bak` first, since the next save replaces it. pub fn load_from(path: &Path) -> Settings { match std::fs::read_to_string(path) { - Ok(content) => parse_and_normalise(&content), + Ok(content) => { + let parsed = parse(&content); + if parsed.is_none() { + back_up_unreadable(path); + } + normalise(parsed.unwrap_or_default()) + } Err(err) => { if err.kind() != std::io::ErrorKind::NotFound { log::warn!("settings read failed ({}): {err}", path.display()); @@ -244,13 +251,65 @@ pub fn load_from(path: &Path) -> Settings { } } -/// Parse settings JSON, falling back to the defaults when it is corrupt, -/// then enforce the invariants a hand-edited file may break. +/// Keep a copy of a settings file that could not be read at all, so the +/// user can recover it by hand. +fn back_up_unreadable(path: &Path) { + let backup = path.with_extension("json.bak"); + match std::fs::copy(path, &backup) { + Ok(_) => log::warn!( + "settings file is not valid JSON; kept a copy at {}", + backup.display() + ), + Err(err) => log::warn!( + "settings file is not valid JSON and could not be copied to {}: {err}", + backup.display() + ), + } +} + +/// Parse settings JSON, keeping every readable key, then enforce the +/// invariants a hand-edited file may break. A file that is not a JSON +/// object yields the defaults. pub fn parse_and_normalise(content: &str) -> Settings { - let mut settings: Settings = serde_json::from_str(content).unwrap_or_else(|err| { + normalise(parse(content).unwrap_or_default()) +} + +/// Parse settings JSON. A value of the wrong type for one key costs only +/// that key, which falls back to its default; `None` when the content is +/// not a JSON object. +fn parse(content: &str) -> Option { + let err = match serde_json::from_str::(content) { + Ok(settings) => return Some(settings), + Err(err) => err, + }; + let Ok(serde_json::Value::Object(file)) = serde_json::from_str(content) else { log::warn!("settings file is not valid; using defaults: {err}"); - Settings::default() - }); + return None; + }; + log::warn!("settings file has invalid values; keeping the valid keys: {err}"); + let serde_json::Value::Object(mut merged) = serde_json::to_value(Settings::default()).ok()? + else { + return None; + }; + for (key, value) in file { + // Unknown keys are ignored, as the strict parse does. + if !merged.contains_key(&key) { + continue; + } + let mut trial = merged.clone(); + trial.insert(key.clone(), value.clone()); + match serde_json::from_value::(serde_json::Value::Object(trial)) { + Ok(_) => { + merged.insert(key, value); + } + Err(err) => log::warn!("settings key {key:?} ignored: {err}"), + } + } + serde_json::from_value(serde_json::Value::Object(merged)).ok() +} + +/// Enforce the invariants a hand-edited file may break. +fn normalise(mut settings: Settings) -> Settings { // At least one live provider must be visible. Otherwise the app has nothing to show. if !settings.has_enabled_live_provider() { settings.show_claude_code = true; @@ -398,6 +457,66 @@ mod tests { assert!(settings.widget_visible); } + #[test] + fn one_wrongly_typed_key_keeps_the_others() { + let settings = parse_and_normalise( + r#"{"bubble_size_logical":250.5,"show_codex":true,"poll_interval_ms":60000}"#, + ); + assert_eq!(settings.bubble_size_logical, DEFAULT_BUBBLE_SIZE); + assert!(settings.show_codex); + assert_eq!(settings.poll_interval_ms, POLL_1_MIN); + // Keys missing from the file keep their defaults. + assert!(settings.show_claude_code); + assert_eq!( + settings.update_check_interval_secs, + Some(UPDATE_CHECK_HOURLY_SECS) + ); + + let settings = parse_and_normalise( + r#"{"show_codex":"yes","widget_visible":false,"last_update_check_unix":5, + "bubble_positions":{"claude":[1,2]},"unknown":1}"#, + ); + assert!(!settings.show_codex); + assert!(!settings.widget_visible); + assert_eq!(settings.last_update_check_unix, Some(5)); + assert_eq!( + settings.bubble_positions.get(ProviderId::Claude), + Some((1, 2)) + ); + } + + fn temp_dir(tag: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "ccub-settings-{tag}-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or_default() + )) + } + + #[test] + fn an_unreadable_file_is_backed_up_before_it_can_be_replaced() { + let dir = temp_dir("backup"); + std::fs::create_dir_all(&dir).expect("mkdir"); + let path = dir.join(SETTINGS_FILE); + let backup = path.with_extension("json.bak"); + std::fs::write(&path, "{ not json").expect("write"); + assert!(load_from(&path).show_claude_code); + assert_eq!( + std::fs::read_to_string(&backup).expect("backup"), + "{ not json" + ); + + // A file with only a bad value is salvaged, not backed up. + std::fs::remove_file(&backup).expect("remove backup"); + std::fs::write(&path, r#"{"show_codex":true,"poll_interval_ms":"x"}"#).expect("write"); + assert!(load_from(&path).show_codex); + assert!(!backup.exists()); + std::fs::remove_dir_all(&dir).expect("cleanup"); + } + #[test] fn normalise_restores_a_live_provider_and_clamps_size() { let settings = parse_and_normalise( diff --git a/src/core/state.rs b/src/core/state.rs index 2928336..549c8df 100644 --- a/src/core/state.rs +++ b/src/core/state.rs @@ -9,7 +9,7 @@ use std::collections::HashMap; use std::time::{Duration, Instant, SystemTime}; use crate::core::i18n::{self, LocaleStrings}; -use crate::usage::{ProviderId, UsageWindows}; +use crate::usage::{ProviderId, UsageWindows, Window}; /// Minimum gap between two cooldown-gated alerts (threshold and session /// expired), so notifications stay calm. @@ -18,6 +18,10 @@ pub const ALERT_COOLDOWN: Duration = Duration::from_secs(30 * 60); /// Usage thresholds that raise an alert the poll they are crossed. pub const ALERT_THRESHOLDS: [u8; 2] = [80, 95]; +/// How far a provider's reported reset time may wander between polls of +/// the same window. +const RESET_JITTER: Duration = Duration::from_secs(60); + /// Shortest countdown re-render delay. const MIN_COUNTDOWN_DELAY: Duration = Duration::from_secs(1); @@ -111,12 +115,15 @@ pub struct UsageState { impl UsageState { /// Apply one poll's results. A provider fires an alert only the poll its /// session usage crosses 80 or 95, and each alert must pass the - /// cooldown. An empty batch changes nothing. + /// cooldown. Usage in a session window that replaced the stored one + /// counts as climbing from zero. An empty batch changes nothing. + /// `now` drives the cooldown, `wall_now` the window-reset check. pub fn apply_results( &mut self, results: Vec<(ProviderId, PollOutcome)>, strings: &LocaleStrings, now: Instant, + wall_now: SystemTime, ) -> Applied { let mut applied = Applied::default(); if results.is_empty() { @@ -128,7 +135,12 @@ impl UsageState { let entry = self.snapshots.entry(id).or_default(); match outcome { PollOutcome::Ok(windows) => { - let old_pct = entry.windows.primary.utilization; + let old_pct = + if is_new_window(&entry.windows.primary, &windows.primary, wall_now) { + 0.0 + } else { + entry.windows.primary.utilization + }; let new_pct = windows.primary.utilization; for threshold in ALERT_THRESHOLDS { if old_pct < threshold as f64 && new_pct >= threshold as f64 { @@ -202,6 +214,26 @@ impl UsageState { } } +/// Whether `new` is a later window than the stored `old` one: the stored +/// window's reset time has passed, or the reset time moved forward by more +/// than [`RESET_JITTER`]. A reset time within the jitter of the stored one +/// is the same window, even when that time has passed, since the provider +/// still reports it. +fn is_new_window(old: &Window, new: &Window, wall_now: SystemTime) -> bool { + let Some(old_reset) = old.resets_at else { + return false; + }; + let near = |t: SystemTime| { + let gap = t.duration_since(old_reset).unwrap_or_else(|e| e.duration()); + gap <= RESET_JITTER + }; + match new.resets_at { + Some(new_reset) if near(new_reset) => false, + Some(new_reset) => old_reset <= wall_now || new_reset > old_reset, + None => old_reset <= wall_now, + } +} + /// Four-line tray tooltip: label, session, weekly and the click hint. pub fn tray_tooltip( label: &str, @@ -304,14 +336,131 @@ mod tests { } fn poll(state: &mut UsageState, pct: f64, now: Instant) -> Applied { + poll_window(state, windows(pct), now, SystemTime::now()) + } + + fn poll_window( + state: &mut UsageState, + w: UsageWindows, + now: Instant, + wall_now: SystemTime, + ) -> Applied { let strings = I18n::load().strings().clone(); state.apply_results( - vec![(ProviderId::Claude, PollOutcome::Ok(windows(pct)))], + vec![(ProviderId::Claude, PollOutcome::Ok(w))], &strings, now, + wall_now, ) } + /// A session window at `pct` that resets at `resets_at`. + fn session(pct: f64, resets_at: SystemTime) -> UsageWindows { + let mut w = windows(pct); + w.primary.resets_at = Some(resets_at); + w + } + + const FIVE_HOURS: Duration = Duration::from_secs(5 * 3_600); + + #[test] + fn a_reset_window_alerts_again_on_its_own_crossing() { + let t0 = Instant::now(); + let wall0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_800_000_000); + let reset1 = wall0 + Duration::from_secs(3_600); + let mut state = UsageState::default(); + assert_eq!( + poll_window(&mut state, session(97.0, reset1), t0, wall0).alerts, + [threshold(80)] + ); + // After the reset the new window climbs to 85 and crosses 80 again. + let t1 = t0 + 2 * ALERT_COOLDOWN; + let wall1 = reset1 + Duration::from_secs(600); + assert_eq!( + poll_window(&mut state, session(85.0, reset1 + FIVE_HOURS), t1, wall1).alerts, + [threshold(80)] + ); + } + + #[test] + fn a_reset_detected_by_a_later_reset_time_counts_too() { + // The poll clock lags, but the provider already reports the next + // window: its reset time moved forward by a whole window. + let t0 = Instant::now(); + let wall0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_800_000_000); + let reset1 = wall0 + Duration::from_secs(60); + let mut state = UsageState::default(); + poll_window(&mut state, session(97.0, reset1), t0, wall0); + let t1 = t0 + 2 * ALERT_COOLDOWN; + assert_eq!( + poll_window(&mut state, session(85.0, reset1 + FIVE_HOURS), t1, wall0).alerts, + [threshold(80)] + ); + } + + #[test] + fn a_new_window_crosses_both_thresholds_in_order() { + let t0 = Instant::now(); + let wall0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_800_000_000); + let reset1 = wall0 + Duration::from_secs(3_600); + let reset2 = reset1 + FIVE_HOURS; + let mut state = UsageState::default(); + poll_window(&mut state, session(97.0, reset1), t0, wall0); + // Jumping to 96 crosses 80 and 95; the cooldown keeps the first. + let t1 = t0 + 2 * ALERT_COOLDOWN; + let wall1 = reset1 + Duration::from_secs(60); + assert_eq!( + poll_window(&mut state, session(96.0, reset2), t1, wall1).alerts, + [threshold(80)] + ); + // Climbing in two steps alerts at 80, then at 95 after the cooldown. + let mut state = UsageState::default(); + poll_window(&mut state, session(97.0, reset1), t0, wall0); + assert_eq!( + poll_window(&mut state, session(85.0, reset2), t1, wall1).alerts, + [threshold(80)] + ); + let t2 = t1 + ALERT_COOLDOWN; + assert_eq!( + poll_window(&mut state, session(96.0, reset2), t2, wall1).alerts, + [threshold(95)] + ); + } + + #[test] + fn reset_time_jitter_within_a_window_does_not_realert() { + let t0 = Instant::now(); + let wall0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_800_000_000); + let reset = wall0 + Duration::from_secs(3_600); + let mut state = UsageState::default(); + assert_eq!( + poll_window(&mut state, session(85.0, reset), t0, wall0).alerts, + [threshold(80)] + ); + let later = t0 + 2 * ALERT_COOLDOWN; + let jitter = Duration::from_secs(3); + for (i, reported) in [reset + jitter, reset - jitter, reset] + .into_iter() + .enumerate() + { + let t = later + ALERT_COOLDOWN * u32::try_from(i).expect("small"); + assert!(poll_window(&mut state, session(88.0, reported), t, wall0) + .alerts + .is_empty()); + } + // A poll just past the reset that still reports the same window is + // the same window, not a fresh one at 88%. + let past = reset + Duration::from_secs(5); + assert!(poll_window( + &mut state, + session(88.0, reset), + later + 4 * ALERT_COOLDOWN, + past + ) + .alerts + .is_empty()); + } + fn threshold(t: u8) -> Alert { Alert::Threshold { provider: ProviderId::Claude, @@ -374,6 +523,7 @@ mod tests { ], &strings, Instant::now(), + SystemTime::now(), ); assert_eq!(applied.auth_failures, [ProviderId::Claude]); assert!(!state.last_poll_ok); @@ -398,12 +548,13 @@ mod tests { ], &strings, Instant::now(), + SystemTime::now(), ); assert!(state.last_poll_ok); assert!(applied.auth_failures.is_empty()); assert_eq!(state.snapshots[&ProviderId::ChatGpt].primary_text, "5%"); assert_eq!( - state.apply_results(Vec::new(), &strings, Instant::now()), + state.apply_results(Vec::new(), &strings, Instant::now(), SystemTime::now()), Applied::default() ); assert!(state.last_poll_ok); diff --git a/src/runtime.rs b/src/runtime.rs index f7eef0d..259ddac 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -166,8 +166,11 @@ pub struct Runtime { http: net::Client, registry: Mutex, /// At most one poll runs at a time; a Refresh spam or a timer firing - /// during a slow poll is dropped instead of stacking requests. + /// during a slow poll is folded into one follow-up poll through + /// `poll_requested` instead of stacking requests. poll_in_flight: AtomicBool, + /// A poll was asked for while one was in flight. + poll_requested: AtomicBool, /// The release the last check found, applied by the version menu item. /// Never locked together with `state`. pending_update: Mutex>, @@ -276,30 +279,47 @@ impl Runtime { } } - /// Start a poll on a worker thread unless one is already running. + /// Start a poll on a worker thread. While one is running the request is + /// queued instead, and the worker polls once more when it is done, so a + /// provider enabled mid-poll still gets its data. pub fn spawn_poll(self: &Arc) { if self .poll_in_flight .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) .is_err() { - log::debug!("poll already in flight; skipped"); + self.poll_requested.store(true, Ordering::Release); + log::debug!("poll already in flight; queued another"); return; } let runtime = Arc::clone(self); let spawned = std::thread::Builder::new() .name("poll".into()) - .spawn(move || { - runtime.poll_once(); - runtime.poll_in_flight.store(false, Ordering::Release); - runtime.propagate(); - }); + .spawn(move || runtime.run_polls()); if let Err(err) = spawned { log::error!("poll thread could not start: {err}"); self.poll_in_flight.store(false, Ordering::Release); } } + /// Body of the poll worker: poll until no request is queued, then hand + /// the in-flight flag back. + fn run_polls(self: &Arc) { + loop { + self.poll_once(); + self.propagate(); + if !self.poll_requested.swap(false, Ordering::AcqRel) { + break; + } + } + self.poll_in_flight.store(false, Ordering::Release); + // A request that came after the last check but before the flag was + // cleared was only queued; start it now. + if self.poll_requested.swap(false, Ordering::AcqRel) { + self.spawn_poll(); + } + } + /// Poll every enabled provider, apply the results, then run the CLI /// credential refresh for the providers whose credentials were rejected. /// Blocking; runs on the poll worker. @@ -314,8 +334,10 @@ impl Runtime { .into_iter() .map(|(provider, result)| (provider, poll_outcome(provider, result))) .collect(); - let applied = - self.with_state(|s| s.usage.apply_results(outcomes, &s.strings, Instant::now())); + let applied = self.with_state(|s| { + s.usage + .apply_results(outcomes, &s.strings, Instant::now(), SystemTime::now()) + }); self.deliver_alerts(applied.alerts); if !applied.auth_failures.is_empty() { self.refresh_credentials(&applied.auth_failures); @@ -578,7 +600,12 @@ impl Runtime { // The apply may fail and keep this process running, so the // automatic check must stay armed. log::debug!("update check skipped while an update is applied"); - self.send(Message::UpdateCheckDone { interval_secs }); + // Not a failure: re-arm the full interval, as the apply + // either relaunches or reports its own error. + self.send(Message::UpdateCheckDone { + interval_secs, + succeeded: true, + }); return; } _ => {} @@ -591,19 +618,21 @@ impl Runtime { if let Err(err) = spawned { log::error!("update check thread could not start: {err}"); self.with_state(|s| s.update_status = UpdateStatus::Failed); - self.send(Message::UpdateCheckDone { interval_secs }); + self.send(Message::UpdateCheckDone { + interval_secs, + succeeded: false, + }); self.refresh_shell(false); } } - /// Blocking body of [`Self::check_for_updates`]. + /// Blocking body of [`Self::check_for_updates`]. Only a check that got + /// an answer is stamped as the last check; a failed one leaves the + /// stamp alone and has the scheduler retry sooner. fn run_update_check(&self) { let proxy = update::proxy_url(net::env_proxy().as_ref(), self.http.system_proxy()); let outcome = update::check(&self.ui.app, proxy); - let now_unix = SystemTime::now() - .duration_since(SystemTime::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or_default(); + let succeeded = outcome.is_ok(); let (status, found) = match outcome { Ok(CheckOutcome::UpToDate) => { log::info!("update check: up to date"); @@ -626,13 +655,21 @@ impl Runtime { .unwrap_or_else(PoisonError::into_inner) = found; let (snapshot, interval_secs) = self.with_state(|s| { s.update_status = status; - s.settings.last_update_check_unix = Some(now_unix); - (s.settings.clone(), s.settings.update_check_interval_secs) + let snapshot = succeeded.then(|| { + s.settings.last_update_check_unix = Some(scheduler::unix_now()); + s.settings.clone() + }); + (snapshot, s.settings.update_check_interval_secs) }); - if let Err(err) = settings::save(&snapshot) { - log::warn!("settings save failed: {err}"); + if let Some(snapshot) = snapshot { + if let Err(err) = settings::save(&snapshot) { + log::warn!("settings save failed: {err}"); + } } - self.send(Message::UpdateCheckDone { interval_secs }); + self.send(Message::UpdateCheckDone { + interval_secs, + succeeded, + }); self.refresh_shell(false); } @@ -972,6 +1009,7 @@ fn setup(app: &mut tauri::App, cli: &CliArgs) -> Result<(), Box> { http, registry: Mutex::new(Registry::with_defaults()), poll_in_flight: AtomicBool::new(false), + poll_requested: AtomicBool::new(false), pending_update: Mutex::new(None), scheduler: tx, trays: Trays::default(), diff --git a/src/scheduler.rs b/src/scheduler.rs index f7eafa6..333a0e8 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -6,6 +6,11 @@ //! thread only waits for the nearest deadline or the next message and hands //! each due [`Action`] to its [`SchedulerHost`]. A host must return quickly: //! slow work (HTTP, the CLI refresh) runs on worker threads it spawns. +//! +//! `Instant` stops while the machine sleeps, so the update check, which is +//! due at a wall-clock time, is kept as a unix timestamp, and the thread +//! wakes at least every [`MAX_WAIT`] to notice a resume (see +//! [`resumed_from_sleep`]). use std::io; use std::sync::mpsc::{Receiver, RecvTimeoutError}; @@ -21,6 +26,18 @@ const MIN_POLL_INTERVAL: Duration = Duration::from_secs(10); /// Shortest automatic update-check interval, for the same reason. const MIN_UPDATE_INTERVAL: Duration = Duration::from_secs(60); +/// First retry delay after a failed update check; it doubles per failure +/// in a row and never exceeds the check interval. +const UPDATE_RETRY_BASE: Duration = Duration::from_secs(60); + +/// Longest the thread sleeps between passes, so a resume from sleep or a +/// wall-clock deadline is noticed within about this long. +const MAX_WAIT: Duration = Duration::from_secs(60); + +/// How far wall time may run ahead of `Instant` between two passes before +/// the gap counts as a sleep. +const RESUME_THRESHOLD: Duration = Duration::from_secs(30); + #[derive(Clone, Debug, PartialEq, Eq)] pub enum Message { /// Poll at once. The periodic poll deadline is left as it is. @@ -36,8 +53,13 @@ pub enum Message { interval_secs: Option, last_check_unix: Option, }, - /// An update check finished. Only this re-arms the full interval. - UpdateCheckDone { interval_secs: Option }, + /// An update check finished. Only this re-arms the check: after the + /// full interval when it succeeded, after a growing retry delay capped + /// at the interval when it failed. + UpdateCheckDone { + interval_secs: Option, + succeeded: bool, + }, /// Stop the thread. Shutdown, } @@ -67,7 +89,10 @@ pub struct Deadlines { poll_interval: Duration, poll_at: Instant, countdown_at: Option, - update_at: Option, + /// Unix seconds the next update check is due at. + update_at_unix: Option, + /// Failed update checks in a row, which set the retry delay. + update_failures: u32, } impl Deadlines { @@ -79,12 +104,12 @@ impl Deadlines { poll_interval, poll_at: now + poll_interval, countdown_at: None, - update_at: None, + update_at_unix: None, + update_failures: 0, }; let actions = deadlines.arm_update( config.update_interval_secs, config.last_update_check_unix, - now, now_unix, ); (deadlines, actions) @@ -106,9 +131,20 @@ impl Deadlines { Message::SetUpdateInterval { interval_secs, last_check_unix, - } => self.arm_update(interval_secs, last_check_unix, now, now_unix), - Message::UpdateCheckDone { interval_secs } => { - self.update_at = interval_secs.map(|secs| now + update_interval(secs)); + } => self.arm_update(interval_secs, last_check_unix, now_unix), + Message::UpdateCheckDone { + interval_secs, + succeeded, + } => { + let delay = if succeeded { + self.update_failures = 0; + interval_secs.map(update_interval) + } else { + let delay = interval_secs.map(|secs| self.retry_delay(secs)); + self.update_failures = self.update_failures.saturating_add(1); + delay + }; + self.update_at_unix = delay.map(|d| now_unix.saturating_add(d.as_secs())); Vec::new() } Message::Shutdown => Vec::new(), @@ -118,7 +154,7 @@ impl Deadlines { /// Fire every deadline that has passed. The poll re-arms for a full /// interval; the countdown and update check stay disarmed until the /// host re-arms them (after the re-render, after the check finished). - pub fn take_due(&mut self, now: Instant) -> Vec { + pub fn take_due(&mut self, now: Instant, now_unix: u64) -> Vec { let mut due = Vec::new(); if self.poll_at <= now { due.push(Action::Poll); @@ -128,52 +164,84 @@ impl Deadlines { due.push(Action::Countdown); self.countdown_at = None; } - if self.update_at.is_some_and(|at| at <= now) { + if self.update_at_unix.is_some_and(|at| at <= now_unix) { due.push(Action::CheckForUpdates); - self.update_at = None; + self.update_at_unix = None; } due } + /// The machine woke from sleep: poll at once and re-arm the poll for a + /// full interval from now. The update check needs nothing here, since + /// [`Self::take_due`] compares its wall-clock deadline with `now_unix`. + pub fn resume(&mut self, now: Instant) -> Vec { + self.poll_at = now + self.poll_interval; + vec![Action::Poll] + } + /// Time until the nearest deadline. The poll deadline is always armed, /// so there always is one. - pub fn next_wakeup(&self, now: Instant) -> Duration { - [Some(self.poll_at), self.countdown_at, self.update_at] - .into_iter() - .flatten() - .min() - .map_or(self.poll_interval, |at| at.saturating_duration_since(now)) + pub fn next_wakeup(&self, now: Instant, now_unix: u64) -> Duration { + let update = self + .update_at_unix + .map(|at| Duration::from_secs(at.saturating_sub(now_unix))); + [ + Some(self.poll_at.saturating_duration_since(now)), + self.countdown_at + .map(|at| at.saturating_duration_since(now)), + update, + ] + .into_iter() + .flatten() + .min() + .unwrap_or(self.poll_interval) } fn arm_update( &mut self, interval_secs: Option, last_check_unix: Option, - now: Instant, now_unix: u64, ) -> Vec { let interval_secs = interval_secs.map(|secs| secs.max(MIN_UPDATE_INTERVAL.as_secs())); match update_check_due(last_check_unix, interval_secs, now_unix) { None => { - self.update_at = None; + self.update_at_unix = None; Vec::new() } Some(UpdateCheckDue::Now) => { - self.update_at = None; + self.update_at_unix = None; vec![Action::CheckForUpdates] } Some(UpdateCheckDue::After(remaining)) => { - self.update_at = Some(now + remaining); + self.update_at_unix = Some(now_unix.saturating_add(remaining.as_secs())); Vec::new() } } } + + /// Delay before the next try after the current run of failures: + /// `UPDATE_RETRY_BASE * 2^failures`, capped at the check interval. + fn retry_delay(&self, interval_secs: u64) -> Duration { + let factor = 1u32.checked_shl(self.update_failures).unwrap_or(u32::MAX); + UPDATE_RETRY_BASE + .saturating_mul(factor) + .min(update_interval(interval_secs)) + } } fn update_interval(secs: u64) -> Duration { Duration::from_secs(secs).max(MIN_UPDATE_INTERVAL) } +/// Whether the wall clock ran ahead of `Instant` by more than +/// [`RESUME_THRESHOLD`] between two passes, which means the machine slept +/// (or the clock was set forward). `wall_elapsed` is `None` when the wall +/// clock went backwards. +pub fn resumed_from_sleep(instant_elapsed: Duration, wall_elapsed: Option) -> bool { + wall_elapsed.is_some_and(|wall| wall.saturating_sub(instant_elapsed) > RESUME_THRESHOLD) +} + /// Seconds since the Unix epoch; 0 for a clock set before 1970. pub fn unix_now() -> u64 { SystemTime::now() @@ -202,11 +270,29 @@ where } fn run_loop(mut deadlines: Deadlines, rx: &Receiver, host: &H) { + let mut last_instant = Instant::now(); + let mut last_wall = SystemTime::now(); loop { - for action in deadlines.take_due(Instant::now()) { + let (now, wall) = (Instant::now(), SystemTime::now()); + let instant_elapsed = now.saturating_duration_since(last_instant); + let wall_elapsed = wall.duration_since(last_wall).ok(); + (last_instant, last_wall) = (now, wall); + if resumed_from_sleep(instant_elapsed, wall_elapsed) { + log::info!( + "wall clock moved {wall_elapsed:?} while Instant moved {instant_elapsed:?}; \ + treating it as a resume from sleep" + ); + for action in deadlines.resume(now) { + host.run(action); + } + } + for action in deadlines.take_due(now, unix_now()) { host.run(action); } - let message = match rx.recv_timeout(deadlines.next_wakeup(Instant::now())) { + let wait = deadlines + .next_wakeup(Instant::now(), unix_now()) + .min(MAX_WAIT); + let message = match rx.recv_timeout(wait) { Ok(Message::Shutdown) | Err(RecvTimeoutError::Disconnected) => return, Ok(message) => message, Err(RecvTimeoutError::Timeout) => continue, @@ -243,11 +329,11 @@ mod tests { let t0 = Instant::now(); let (mut d, initial) = Deadlines::new(config(None, None), t0, NOW_UNIX); assert!(initial.is_empty()); - assert_eq!(d.next_wakeup(t0), 5 * MIN); - assert!(d.take_due(t0 + 4 * MIN).is_empty()); - assert_eq!(d.take_due(t0 + 5 * MIN), [Action::Poll]); - assert_eq!(d.next_wakeup(t0 + 5 * MIN), 5 * MIN); - assert_eq!(d.take_due(t0 + 10 * MIN), [Action::Poll]); + assert_eq!(d.next_wakeup(t0, NOW_UNIX), 5 * MIN); + assert!(d.take_due(t0 + 4 * MIN, NOW_UNIX).is_empty()); + assert_eq!(d.take_due(t0 + 5 * MIN, NOW_UNIX), [Action::Poll]); + assert_eq!(d.next_wakeup(t0 + 5 * MIN, NOW_UNIX), 5 * MIN); + assert_eq!(d.take_due(t0 + 10 * MIN, NOW_UNIX), [Action::Poll]); } #[test] @@ -258,7 +344,7 @@ mod tests { d.handle(Message::PollNow, t0 + 2 * MIN, NOW_UNIX), [Action::Poll] ); - assert_eq!(d.take_due(t0 + 5 * MIN), [Action::Poll]); + assert_eq!(d.take_due(t0 + 5 * MIN, NOW_UNIX), [Action::Poll]); } #[test] @@ -269,9 +355,9 @@ mod tests { assert!(d .handle(Message::SetPollInterval(MIN * 15), t1, NOW_UNIX) .is_empty()); - assert!(d.take_due(t0 + 5 * MIN).is_empty()); - assert_eq!(d.next_wakeup(t1), 15 * MIN); - assert_eq!(d.take_due(t1 + 15 * MIN), [Action::Poll]); + assert!(d.take_due(t0 + 5 * MIN, NOW_UNIX).is_empty()); + assert_eq!(d.next_wakeup(t1, NOW_UNIX), 15 * MIN); + assert_eq!(d.take_due(t1 + 15 * MIN, NOW_UNIX), [Action::Poll]); } #[test] @@ -280,7 +366,7 @@ mod tests { let mut cfg = config(None, None); cfg.poll_interval = Duration::ZERO; let (d, _) = Deadlines::new(cfg, t0, NOW_UNIX); - assert_eq!(d.next_wakeup(t0), MIN_POLL_INTERVAL); + assert_eq!(d.next_wakeup(t0, NOW_UNIX), MIN_POLL_INTERVAL); } #[test] @@ -289,13 +375,13 @@ mod tests { let (mut d, _) = Deadlines::new(config(None, None), t0, NOW_UNIX); let delay = Duration::from_secs(31); d.handle(Message::RecomputeCountdown(Some(delay)), t0, NOW_UNIX); - assert_eq!(d.next_wakeup(t0), delay); - assert_eq!(d.take_due(t0 + delay), [Action::Countdown]); - assert!(d.take_due(t0 + 2 * delay).is_empty()); + assert_eq!(d.next_wakeup(t0, NOW_UNIX), delay); + assert_eq!(d.take_due(t0 + delay, NOW_UNIX), [Action::Countdown]); + assert!(d.take_due(t0 + 2 * delay, NOW_UNIX).is_empty()); d.handle(Message::RecomputeCountdown(Some(delay)), t0, NOW_UNIX); d.handle(Message::RecomputeCountdown(None), t0, NOW_UNIX); - assert!(d.take_due(t0 + 2 * delay).is_empty()); + assert!(d.take_due(t0 + 2 * delay, NOW_UNIX).is_empty()); } #[test] @@ -319,44 +405,118 @@ mod tests { NOW_UNIX, ); assert!(initial.is_empty()); - let remaining = Duration::from_secs(3_000); + let remaining = 3_000; assert!(!d - .take_due(t0 + remaining - MIN) + .take_due(t0, NOW_UNIX + remaining - 60) .contains(&Action::CheckForUpdates)); - assert_eq!(d.next_wakeup(t0 + remaining - MIN), MIN); - assert_eq!(d.take_due(t0 + remaining), [Action::CheckForUpdates]); + assert_eq!(d.next_wakeup(t0, NOW_UNIX + remaining - 60), MIN); + assert_eq!( + d.take_due(t0, NOW_UNIX + remaining), + [Action::CheckForUpdates] + ); + } + + /// Whether a check is due at `now_unix`, with `Instant` left at `t0` + /// to show the update deadline follows the wall clock alone. + fn check_due(d: &mut Deadlines, t0: Instant, now_unix: u64) -> bool { + d.take_due(t0, now_unix).contains(&Action::CheckForUpdates) + } + + fn done(interval_secs: Option, succeeded: bool) -> Message { + Message::UpdateCheckDone { + interval_secs, + succeeded, + } } #[test] fn only_a_finished_check_rearms_the_full_interval() { let t0 = Instant::now(); - let hour = Duration::from_secs(UPDATE_CHECK_HOURLY_SECS); - let (mut d, initial) = - Deadlines::new(config(Some(UPDATE_CHECK_HOURLY_SECS), None), t0, NOW_UNIX); + let hour = UPDATE_CHECK_HOURLY_SECS; + let (mut d, initial) = Deadlines::new(config(Some(hour), None), t0, NOW_UNIX); assert_eq!(initial, [Action::CheckForUpdates]); // While the check runs nothing else is scheduled for it. - assert!(!d.take_due(t0 + 3 * hour).contains(&Action::CheckForUpdates)); + assert!(!check_due(&mut d, t0, NOW_UNIX + 3 * hour)); - let t1 = t0 + 3 * hour; - d.handle( - Message::UpdateCheckDone { - interval_secs: Some(UPDATE_CHECK_HOURLY_SECS), - }, - t1, - NOW_UNIX, - ); - assert!(!d.take_due(t1 + hour / 2).contains(&Action::CheckForUpdates)); - assert!(d.take_due(t1 + hour).contains(&Action::CheckForUpdates)); + let t1 = NOW_UNIX + 3 * hour; + d.handle(done(Some(hour), true), t0, t1); + assert!(!check_due(&mut d, t0, t1 + hour / 2)); + assert!(check_due(&mut d, t0, t1 + hour)); // It repeats every interval, not just once. let t2 = t1 + hour; - d.handle( - Message::UpdateCheckDone { - interval_secs: Some(UPDATE_CHECK_HOURLY_SECS), - }, - t2, - NOW_UNIX, - ); - assert!(d.take_due(t2 + hour).contains(&Action::CheckForUpdates)); + d.handle(done(Some(hour), true), t0, t2); + assert!(check_due(&mut d, t0, t2 + hour)); + } + + #[test] + fn a_failed_check_retries_with_a_doubling_delay_capped_at_the_interval() { + let t0 = Instant::now(); + let hour = UPDATE_CHECK_HOURLY_SECS; + let (mut d, _) = Deadlines::new(config(Some(hour), None), t0, NOW_UNIX); + let mut now = NOW_UNIX; + for expected in [60, 120, 240, 480, 960, 1_920, hour, hour] { + d.handle(done(Some(hour), false), t0, now); + assert!(!check_due(&mut d, t0, now + expected - 1)); + assert!(check_due(&mut d, t0, now + expected)); + now += expected; + } + // A success resets the delay: the next failure retries after a minute. + d.handle(done(Some(hour), true), t0, now); + assert!(check_due(&mut d, t0, now + hour)); + now += hour; + d.handle(done(Some(hour), false), t0, now); + assert!(check_due(&mut d, t0, now + 60)); + } + + #[test] + fn a_failed_check_with_auto_check_off_stays_off() { + let t0 = Instant::now(); + let (mut d, _) = Deadlines::new(config(None, None), t0, NOW_UNIX); + d.handle(done(None, false), t0, NOW_UNIX); + assert!(!check_due(&mut d, t0, NOW_UNIX + 7 * DAY)); + } + + #[test] + fn the_update_deadline_follows_the_wall_clock_across_a_sleep() { + // `Instant` does not advance while the machine sleeps, but the + // check is still due once the wall clock passes its time. + let t0 = Instant::now(); + let day = UPDATE_CHECK_DAILY_SECS; + let (mut d, _) = Deadlines::new(config(Some(day), Some(NOW_UNIX)), t0, NOW_UNIX); + assert!(!check_due(&mut d, t0, NOW_UNIX + day - 1)); + assert!(check_due( + &mut d, + t0 + Duration::from_secs(1), + NOW_UNIX + day + )); + } + + #[test] + fn resume_polls_at_once_and_rearms_the_poll() { + let t0 = Instant::now(); + let (mut d, _) = Deadlines::new(config(None, None), t0, NOW_UNIX); + let t1 = t0 + 2 * MIN; + assert_eq!(d.resume(t1), [Action::Poll]); + assert!(d.take_due(t0 + 5 * MIN, NOW_UNIX).is_empty()); + assert_eq!(d.next_wakeup(t1, NOW_UNIX), 5 * MIN); + assert_eq!(d.take_due(t1 + 5 * MIN, NOW_UNIX), [Action::Poll]); + } + + #[test] + fn a_wall_clock_jump_past_instant_counts_as_a_resume() { + let secs = Duration::from_secs; + assert!(!resumed_from_sleep(MIN, Some(MIN))); + // Ordinary scheduling slack is not a resume. + assert!(!resumed_from_sleep(MIN, Some(MIN + secs(5)))); + assert!(!resumed_from_sleep(MIN, Some(MIN + RESUME_THRESHOLD))); + assert!(resumed_from_sleep( + MIN, + Some(MIN + RESUME_THRESHOLD + secs(1)) + )); + assert!(resumed_from_sleep(secs(1), Some(secs(8 * 3_600)))); + // A clock set back is not a resume. + assert!(!resumed_from_sleep(MIN, None)); + assert!(!resumed_from_sleep(MIN, Some(Duration::ZERO))); } #[test] @@ -393,13 +553,9 @@ mod tests { NOW_UNIX, ); assert!(actions.is_empty()); - let remaining = Duration::from_secs(5 * DAY); - assert!(!d - .take_due(t0 + remaining - MIN) - .contains(&Action::CheckForUpdates)); - assert!(d - .take_due(t0 + remaining) - .contains(&Action::CheckForUpdates)); + let remaining = 5 * DAY; + assert!(!check_due(&mut d, t0, NOW_UNIX + remaining - 60)); + assert!(check_due(&mut d, t0, NOW_UNIX + remaining)); } #[test] @@ -418,20 +574,10 @@ mod tests { t0, NOW_UNIX, ); - assert!(!d - .take_due(t0 + Duration::from_secs(7 * DAY)) - .contains(&Action::CheckForUpdates)); + assert!(!check_due(&mut d, t0, NOW_UNIX + 7 * DAY)); // A check that finishes after auto-check was turned off stays off. - d.handle( - Message::UpdateCheckDone { - interval_secs: None, - }, - t0, - NOW_UNIX, - ); - assert!(!d - .take_due(t0 + Duration::from_secs(14 * DAY)) - .contains(&Action::CheckForUpdates)); + d.handle(done(None, true), t0, NOW_UNIX); + assert!(!check_due(&mut d, t0, NOW_UNIX + 14 * DAY)); } /// Records actions on a channel, standing in for the runtime.