Merge branch 'worktree-wf_093aad93-ad0-4' into review-fixes

This commit is contained in:
tiennm99 committed 2026-10-23 03:41:26 +07:00
commit 4b11a31756
5 files changed
+574 -117

No files matched your search

+3
View File
@@ -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);
+126 -7
View File
@@ -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<Settings> {
let err = match serde_json::from_str::<Settings>(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::<Settings>(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(
+156 -5
View File
@@ -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);
+60 -22
View File
@@ -166,8 +166,11 @@ pub struct Runtime {
http: net::Client,
registry: Mutex<Registry>,
/// 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<Option<tauri_plugin_updater::Update>>,
@@ -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<Self>) {
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<Self>) {
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<dyn Error>> {
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(),
+229 -83
View File
@@ -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<u64>,
last_check_unix: Option<u64>,
},
/// An update check finished. Only this re-arms the full interval.
UpdateCheckDone { interval_secs: Option<u64> },
/// 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<u64>,
succeeded: bool,
},
/// Stop the thread.
Shutdown,
}
@@ -67,7 +89,10 @@ pub struct Deadlines {
poll_interval: Duration,
poll_at: Instant,
countdown_at: Option<Instant>,
update_at: Option<Instant>,
/// Unix seconds the next update check is due at.
update_at_unix: Option<u64>,
/// 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<Action> {
pub fn take_due(&mut self, now: Instant, now_unix: u64) -> Vec<Action> {
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<Action> {
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<u64>,
last_check_unix: Option<u64>,
now: Instant,
now_unix: u64,
) -> Vec<Action> {
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<Duration>) -> 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<H: SchedulerHost>(mut deadlines: Deadlines, rx: &Receiver<Message>, 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<u64>, 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.