coven_foundation/
clock.rs

1//! The wall clock that timestamps use (§7.2); the only reader of system time.
2
3use std::sync::Arc;
4use std::time::{Duration, SystemTime, UNIX_EPOCH};
5use std::{future::Future, pin::Pin};
6
7// The system's wall-clock deadline timer, one per platform; elsewhere a
8// wait for a wall-clock time fails.
9#[cfg(target_vendor = "apple")]
10#[path = "clock_apple.rs"]
11mod wall_timer;
12#[cfg(any(target_os = "linux", target_os = "android"))]
13#[path = "clock_linux.rs"]
14mod wall_timer;
15#[cfg(windows)]
16#[path = "clock_windows.rs"]
17mod wall_timer;
18
19/// The wall clock that timestamps use (§7.2).
20///
21/// Returning `SystemTime` preserves pre-epoch and out-of-range readings so the
22/// timestamp layer can reject them when converting to 48-bit milliseconds.
23pub trait Clock: Send + Sync {
24    /// The wall clock's current time.
25    fn now(&self) -> SystemTime;
26    /// Monotonic time since an arbitrary origin, counting time the system
27    /// spent asleep (§16.6). Readings never decrease within a process; only
28    /// differences between two readings of one clock mean anything.
29    fn elapsed(&self) -> Duration;
30    /// End once `duration` of elapsed time has passed: a deadline that
31    /// work races, such as a sign-in's wait for its callback or a stalled
32    /// transfer's, or an interval, such as the idle time between passes.
33    /// System clocks use the runtime's monotonic timer; controlled clocks
34    /// wake only when their supplied time advances.
35    fn deadline(&self, duration: Duration) -> Pin<Box<dyn Future<Output = ()> + Send>> {
36        Box::pin(tokio::time::sleep(duration))
37    }
38    /// Wait out `wait` before trying again: a backoff, a provider's
39    /// cooldown, or the spacing between requests the code chose. It passes
40    /// as a deadline of the same length does; a test's clock may move its
41    /// time to the wait's end as the code arms it, since no work races it.
42    fn wait_out(&self, wait: Duration) -> Pin<Box<dyn Future<Output = ()> + Send>> {
43        self.deadline(wait)
44    }
45    /// Wait until the wall clock reads `at` or later, however it gets
46    /// there: by running, or by being set past it (§13.2). A time already
47    /// reached ends the wait at once, and a clock set back moves the wait's
48    /// end with it. The clock itself says when it reaches `at`; nothing reads
49    /// it again on a timer. A clock that can't say fails the wait.
50    fn sleep_until(
51        &self,
52        at: SystemTime,
53    ) -> Pin<Box<dyn Future<Output = Result<(), WallTimerError>> + Send>>;
54}
55
56/// A clock couldn't wait for a wall-clock time (§13.2): the system refused
57/// its timer, or the clock has none.
58#[derive(Clone, Debug, thiserror::Error)]
59pub enum WallTimerError {
60    /// The clock has no timer for a wall-clock time: a platform without one,
61    /// or a clock that can't say when its reading reaches it.
62    #[error("this clock has no wall-clock timer coven can wait on")]
63    NoTimer,
64    /// The time is past what the system's wall timer holds.
65    #[error("the time is past what the system's wall timer holds")]
66    OutOfRange {
67        /// The failed conversion, where the system's timer takes the time
68        /// as an integer of another size.
69        #[source]
70        source: Option<std::num::TryFromIntError>,
71    },
72    /// The system couldn't create its timer source.
73    #[error("the system couldn't create a timer source")]
74    TimerSource,
75    /// A call that arms or waits on the system's timer failed.
76    #[error("the system's `{call}` failed")]
77    System {
78        /// The call that failed.
79        call: &'static str,
80        /// What the system answered.
81        #[source]
82        source: Arc<std::io::Error>,
83    },
84    /// An app's own clock failed to wait.
85    #[error("the clock failed to wait for a wall-clock time")]
86    Clock(#[source] Arc<dyn std::error::Error + Send + Sync>),
87}
88
89impl WallTimerError {
90    /// A failure of an app's own clock, which it reports as this.
91    pub fn clock(failure: impl std::error::Error + Send + Sync + 'static) -> Self {
92        Self::Clock(Arc::new(failure))
93    }
94
95    /// The system's answer to a failed `call`.
96    #[cfg(any(target_os = "linux", target_os = "android", windows))]
97    pub(crate) fn system(call: &'static str, source: impl Into<std::io::Error>) -> Self {
98        Self::System { call, source: Arc::new(source.into()) }
99    }
100}
101
102/// A shared wall clock, supplied when the store opens.
103pub type ClockRef = Arc<dyn Clock>;
104
105/// The system wall clock.
106pub struct SystemClock;
107
108impl Clock for SystemClock {
109    fn now(&self) -> SystemTime {
110        SystemTime::now()
111    }
112    fn elapsed(&self) -> Duration {
113        boot_time()
114    }
115    /// The system's own timer for an absolute wall-clock time, which the
116    /// system moves when its clock is set. A time at or before the epoch
117    /// has passed: no timer is armed for it, since Linux's timer can't hold
118    /// one.
119    fn sleep_until(
120        &self,
121        at: SystemTime,
122    ) -> Pin<Box<dyn Future<Output = Result<(), WallTimerError>> + Send>> {
123        if at <= UNIX_EPOCH {
124            return Box::pin(std::future::ready(Ok(())));
125        }
126        wall_timer_sleep(at)
127    }
128}
129
130#[cfg(any(target_vendor = "apple", target_os = "linux", target_os = "android", windows))]
131fn wall_timer_sleep(
132    at: SystemTime,
133) -> Pin<Box<dyn Future<Output = Result<(), WallTimerError>> + Send>> {
134    Box::pin(wall_timer::sleep_until(at))
135}
136
137/// Elsewhere, as on the BSDs, coven has no timer the system moves with its
138/// clock, and fails the wait rather than read the clock again on a timer.
139#[cfg(not(any(target_vendor = "apple", target_os = "linux", target_os = "android", windows)))]
140fn wall_timer_sleep(
141    _at: SystemTime,
142) -> Pin<Box<dyn Future<Output = Result<(), WallTimerError>> + Send>> {
143    Box::pin(std::future::ready(Err(WallTimerError::NoTimer)))
144}
145
146/// Linux and Android count suspension only in `CLOCK_BOOTTIME`.
147#[cfg(any(target_os = "linux", target_os = "android"))]
148fn boot_time() -> Duration {
149    timespec(rustix::time::clock_gettime(rustix::time::ClockId::Boottime))
150}
151
152/// Darwin's `CLOCK_MONOTONIC` keeps advancing while the system sleeps,
153/// unlike `CLOCK_UPTIME_RAW` and `std::time::Instant`.
154#[cfg(target_vendor = "apple")]
155fn boot_time() -> Duration {
156    timespec(rustix::time::clock_gettime(rustix::time::ClockId::Monotonic))
157}
158
159/// Other Unix systems: `CLOCK_MONOTONIC`, whose treatment of suspension is
160/// the platform's own.
161#[cfg(all(unix, not(any(target_os = "linux", target_os = "android", target_vendor = "apple"))))]
162fn boot_time() -> Duration {
163    timespec(rustix::time::clock_gettime(rustix::time::ClockId::Monotonic))
164}
165
166#[cfg(unix)]
167fn timespec(time: rustix::time::Timespec) -> Duration {
168    Duration::new(
169        u64::try_from(time.tv_sec).expect("monotonic seconds are nonnegative"),
170        u32::try_from(time.tv_nsec).expect("nanoseconds below one second"),
171    )
172}
173
174/// Windows interrupt time counts sleep and hibernation, in 100 ns units.
175#[cfg(windows)]
176fn boot_time() -> Duration {
177    let mut ticks = 0u64;
178    // SAFETY: the out pointer refers to a live, writable u64.
179    unsafe { windows_sys::Win32::System::WindowsProgramming::QueryInterruptTime(&mut ticks) };
180    Duration::from_nanos(ticks.saturating_mul(100))
181}
182
183/// How long a test waits on an event before failing instead of hanging.
184/// Every wait it bounds ends on the code's own event, so only code that never
185/// sends it reaches this. It is not a measure of speed, so it sits far above
186/// what any step takes on a loaded machine (the slowest whole tests take about
187/// 260s there). nextest ends a test at `terminate-after` periods in
188/// `.config/nextest.toml`, which stays a period above this so the test's own
189/// report of where it hung prints first; `clock_tests` checks it.
190#[cfg(any(test, feature = "test-utils"))]
191pub const HANG_GUARD: Duration = Duration::from_secs(300);
192
193/// `future`'s output, or `None` once `HANG_GUARD` of real time passes
194/// first: a test waiting on the system fails instead of hanging, whatever
195/// clock the store under it reads.
196#[cfg(any(test, feature = "test-utils"))]
197pub async fn within<F: Future>(future: F) -> Option<F::Output> {
198    tokio::time::timeout(HANG_GUARD, future).await.ok()
199}
200
201/// Block this thread for `HANG_GUARD` of real time: what a test process's
202/// hang report waits out before it prints.
203#[cfg(any(test, feature = "test-utils"))]
204pub fn wait_out_hang_guard() {
205    std::thread::sleep(HANG_GUARD);
206}
207
208/// A clock that returns the instant supplied by a test until it is set again.
209#[cfg(feature = "test-utils")]
210pub struct FixedClock {
211    time: tokio::sync::watch::Sender<FixedTime>,
212    /// How many sleeps have begun, and the latest one's deadline by
213    /// `elapsed`.
214    sleeps: tokio::sync::watch::Sender<(u64, Duration)>,
215}
216
217#[cfg(feature = "test-utils")]
218struct FixedTime {
219    wall: SystemTime,
220    elapsed: Duration,
221}
222
223#[cfg(feature = "test-utils")]
224impl FixedClock {
225    /// A clock initially set to `now`.
226    pub fn new(now: SystemTime) -> Self {
227        Self {
228            time: tokio::sync::watch::channel(FixedTime { wall: now, elapsed: Duration::ZERO }).0,
229            sleeps: tokio::sync::watch::channel((0, Duration::ZERO)).0,
230        }
231    }
232
233    /// Set the time every subsequent read returns, including a backward jump.
234    pub fn set(&self, now: SystemTime) {
235        self.time.send_modify(|time| {
236            if let Ok(elapsed) = now.duration_since(time.wall) {
237                time.elapsed += elapsed;
238            }
239            time.wall = now;
240        });
241    }
242
243    /// How many deadlines and waits have begun on this clock so far.
244    pub fn sleeps_begun(&self) -> u64 {
245        self.sleeps.borrow().0
246    }
247
248    /// Wait until a deadline or wait begins after the first `seen`, and
249    /// return how many have begun and the latest one's end by `elapsed`,
250    /// so a test can set the clock to exactly the wait the code armed.
251    pub fn sleep_begun_after(
252        &self,
253        seen: u64,
254    ) -> impl Future<Output = (u64, Duration)> + Send + 'static {
255        let mut sleeps = self.sleeps.subscribe();
256        async move { *sleeps.wait_for(|(begun, _)| *begun > seen).await.expect("the clock is kept") }
257    }
258}
259
260#[cfg(feature = "test-utils")]
261impl Clock for FixedClock {
262    fn now(&self) -> SystemTime {
263        self.time.borrow().wall
264    }
265    /// Forward settings advance it, as sleep would; backward ones do not.
266    fn elapsed(&self) -> Duration {
267        self.time.borrow().elapsed
268    }
269    fn deadline(&self, duration: Duration) -> Pin<Box<dyn Future<Output = ()> + Send>> {
270        let mut time = self.time.subscribe();
271        let deadline = time.borrow().elapsed + duration;
272        self.sleeps.send_modify(|(begun, latest)| {
273            *begun += 1;
274            *latest = deadline;
275        });
276        Box::pin(async move {
277            time.wait_for(|time| time.elapsed >= deadline)
278                .await
279                .expect("clock owner retains its sender");
280        })
281    }
282    /// Wakes when a setting reaches `at`; one before it, forward or back,
283    /// does not.
284    fn sleep_until(
285        &self,
286        at: SystemTime,
287    ) -> Pin<Box<dyn Future<Output = Result<(), WallTimerError>> + Send>> {
288        let mut time = self.time.subscribe();
289        Box::pin(async move {
290            time.wait_for(|time| time.wall >= at).await.expect("clock owner retains its sender");
291            Ok(())
292        })
293    }
294}
295
296#[cfg(feature = "test-utils")]
297#[path = "clock_polling.rs"]
298mod polling;
299#[cfg(feature = "test-utils")]
300pub use polling::PollingClock;
301
302/// A clock that calls a supplied function on every read.
303#[cfg(feature = "test-utils")]
304pub struct ClosureClock<F>(
305    /// The function called on each read.
306    pub F,
307);
308
309#[cfg(feature = "test-utils")]
310impl<F: Fn() -> SystemTime + Send + Sync> Clock for ClosureClock<F> {
311    fn now(&self) -> SystemTime {
312        (self.0)()
313    }
314    /// A supplied wall reading has no monotonic counterpart: it never advances.
315    fn elapsed(&self) -> Duration {
316        Duration::ZERO
317    }
318    /// Nothing says when the function's reading reaches a time.
319    fn sleep_until(
320        &self,
321        _at: SystemTime,
322    ) -> Pin<Box<dyn Future<Output = Result<(), WallTimerError>> + Send>> {
323        Box::pin(std::future::ready(Err(WallTimerError::NoTimer)))
324    }
325}
326
327/// The system wall waits a test listens to the release of, by deadline: a
328/// system timer is the process's, so the waits on it are the process's.
329#[cfg(test)]
330static RELEASED: std::sync::Mutex<Vec<(SystemTime, tokio::sync::oneshot::Sender<()>)>> =
331    std::sync::Mutex::new(Vec::new());
332
333/// Told once a system wall wait for `at` has released its timer, so the
334/// timer's handler can no longer run.
335#[cfg(test)]
336fn when_released(at: SystemTime) -> tokio::sync::oneshot::Receiver<()> {
337    let (told, telling) = tokio::sync::oneshot::channel();
338    crate::locks::lock_unpoisoned(&RELEASED).push((at, told));
339    telling
340}
341
342/// A system wall wait for `at` has released its timer: tell whoever waits
343/// for that.
344#[cfg(all(
345    test,
346    any(target_vendor = "apple", target_os = "linux", target_os = "android", windows)
347))]
348fn released(at: SystemTime) {
349    let mut watched = crate::locks::lock_unpoisoned(&RELEASED);
350    let (found, rest): (Vec<_>, Vec<_>) =
351        std::mem::take(&mut *watched).into_iter().partition(|(watched, _)| *watched == at);
352    *watched = rest;
353    for (_, told) in found {
354        #[expect(
355            clippy::let_underscore_must_use,
356            reason = "a test that stopped listening has no use for it"
357        )]
358        let _ = told.send(());
359    }
360}
361
362/// Dropped after a wait's timer, whose handle closing releases it there.
363#[cfg(all(test, any(target_os = "linux", target_os = "android", windows)))]
364struct ReleasedOnDrop(SystemTime);
365
366#[cfg(all(test, any(target_os = "linux", target_os = "android", windows)))]
367impl Drop for ReleasedOnDrop {
368    fn drop(&mut self) {
369        released(self.0);
370    }
371}
372
373#[cfg(test)]
374#[path = "clock_tests.rs"]
375mod tests;
coven_foundation/clock.rs

coven_foundation/
clock.rs
