#![cfg_attr(not(test), deny(clippy::disallowed_methods))]
use std::sync::{Arc, LazyLock, Mutex};
use std::time::Duration;
use chrono::{DateTime, Utc};
#[allow(
clippy::disallowed_methods,
reason = "this IS the seam: the single process-monotonic origin every real \
MonotonicInstant is measured from. There is nothing further to \
inject it from."
)]
static MONOTONIC_ORIGIN: LazyLock<std::time::Instant> = LazyLock::new(std::time::Instant::now);
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct MonotonicInstant(Duration);
impl MonotonicInstant {
pub const ORIGIN: Self = Self(Duration::ZERO);
#[must_use]
pub const fn from_origin_elapsed(since_origin: Duration) -> Self {
Self(since_origin)
}
#[must_use]
pub const fn since_origin(self) -> Duration {
self.0
}
#[must_use]
pub const fn saturating_duration_since(self, earlier: Self) -> Duration {
self.0.saturating_sub(earlier.0)
}
#[must_use]
pub const fn elapsed_at(self, now: Self) -> Duration {
now.saturating_duration_since(self)
}
#[must_use]
pub const fn checked_add(self, duration: Duration) -> Option<Self> {
match self.0.checked_add(duration) {
Some(sum) => Some(Self(sum)),
None => None,
}
}
#[must_use]
pub const fn saturating_add(self, duration: Duration) -> Self {
Self(self.0.saturating_add(duration))
}
}
pub trait ClockSource: Send + Sync + 'static {
fn now(&self) -> DateTime<Utc>;
fn monotonic(&self) -> MonotonicInstant {
MonotonicInstant::from_origin_elapsed(MONOTONIC_ORIGIN.elapsed())
}
}
impl ClockSource for std::sync::Arc<dyn ClockSource> {
fn now(&self) -> DateTime<Utc> {
(**self).now()
}
fn monotonic(&self) -> MonotonicInstant {
(**self).monotonic()
}
}
#[derive(Debug, Clone, Copy)]
pub struct Clock(DateTime<Utc>, MonotonicInstant);
impl Clock {
#[must_use]
pub const fn now(&self) -> DateTime<Utc> {
self.0
}
#[must_use]
pub const fn monotonic(&self) -> MonotonicInstant {
self.1
}
}
impl std::ops::Deref for Clock {
type Target = DateTime<Utc>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl axum::extract::FromRequestParts<crate::state::AppState> for Clock {
type Rejection = std::convert::Infallible;
async fn from_request_parts(
_parts: &mut axum::http::request::Parts,
state: &crate::state::AppState,
) -> Result<Self, Self::Rejection> {
let clock = state.clock();
Ok(Self(clock.now(), clock.monotonic()))
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct SystemClock;
impl ClockSource for SystemClock {
#[allow(
clippy::disallowed_methods,
reason = "this IS the seam: SystemClock is the production ClockSource, \
the one place the real wall clock is allowed to be read."
)]
fn now(&self) -> DateTime<Utc> {
Utc::now()
}
}
#[derive(Debug, Clone, Copy)]
pub struct FixedClock(DateTime<Utc>);
impl FixedClock {
#[must_use]
pub const fn at(dt: DateTime<Utc>) -> Self {
Self(dt)
}
}
impl ClockSource for FixedClock {
fn now(&self) -> DateTime<Utc> {
self.0
}
fn monotonic(&self) -> MonotonicInstant {
MonotonicInstant::ORIGIN
}
}
#[derive(Clone, Debug)]
pub struct TickingClock {
current: Arc<Mutex<DateTime<Utc>>>,
start: DateTime<Utc>,
}
impl TickingClock {
#[must_use]
pub fn starting_at(dt: DateTime<Utc>) -> Self {
Self {
current: Arc::new(Mutex::new(dt)),
start: dt,
}
}
pub fn advance(&self, duration: std::time::Duration) {
let mut guard = self
.current
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Ok(delta) = chrono::Duration::from_std(duration) {
*guard += delta;
}
}
}
impl ClockSource for TickingClock {
fn now(&self) -> DateTime<Utc> {
*self
.current
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn monotonic(&self) -> MonotonicInstant {
let elapsed = self
.now()
.signed_duration_since(self.start)
.to_std()
.unwrap_or(Duration::ZERO);
MonotonicInstant::from_origin_elapsed(elapsed)
}
}
#[must_use]
pub fn monotonic_now() -> MonotonicInstant {
SystemClock.monotonic()
}
thread_local! {
static AMBIENT: std::cell::RefCell<Vec<AmbientEntry>> =
const { std::cell::RefCell::new(Vec::new()) };
}
struct AmbientEntry {
alive: Arc<std::sync::atomic::AtomicBool>,
clock: Arc<dyn ClockSource>,
floor: Duration,
start: Duration,
}
static INSTANT_HIGH_WATER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
fn instant_floor() -> Duration {
let high_water =
Duration::from_nanos(INSTANT_HIGH_WATER.load(std::sync::atomic::Ordering::Relaxed));
SystemClock.monotonic().since_origin().max(high_water)
}
#[derive(Debug)]
pub(crate) struct AmbientGuard {
alive: Arc<std::sync::atomic::AtomicBool>,
}
impl Drop for AmbientGuard {
fn drop(&mut self) {
self.alive
.store(false, std::sync::atomic::Ordering::Release);
let _ = AMBIENT.try_with(|stack| {
if let Ok(mut stack) = stack.try_borrow_mut() {
stack.retain(|entry| !Arc::ptr_eq(&entry.alive, &self.alive));
}
});
}
}
pub(crate) fn install_ambient(clock: Arc<dyn ClockSource>) -> AmbientGuard {
let start = clock.monotonic().since_origin();
install_ambient_at(clock, instant_floor(), start)
}
fn install_ambient_at(
clock: Arc<dyn ClockSource>,
floor: Duration,
start: Duration,
) -> AmbientGuard {
let alive = Arc::new(std::sync::atomic::AtomicBool::new(true));
AMBIENT.with(|stack| {
stack.borrow_mut().push(AmbientEntry {
alive: Arc::clone(&alive),
clock,
floor,
start,
});
});
AmbientGuard { alive }
}
fn ambient_entry() -> Option<(Arc<dyn ClockSource>, Duration, Duration)> {
AMBIENT
.try_with(|stack| {
let mut stack = stack.try_borrow_mut().ok()?;
stack.retain(|entry| entry.alive.load(std::sync::atomic::Ordering::Acquire));
stack
.last()
.map(|entry| (Arc::clone(&entry.clock), entry.floor, entry.start))
})
.ok()
.flatten()
}
pub(crate) fn spawn_blocking<F, R>(f: F) -> tokio::task::JoinHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
let carried = ambient_entry();
#[allow(
clippy::disallowed_methods,
reason = "the sanctioned spawn_blocking: it carries the ambient clock"
)]
tokio::task::spawn_blocking(move || {
let _guard = carried.map(|(clock, floor, start)| install_ambient_at(clock, floor, start));
f()
})
}
thread_local! {
static THREAD_LATEST_INSTANT: std::cell::Cell<Duration> =
const { std::cell::Cell::new(Duration::ZERO) };
}
fn keep_thread_instants_forward(offset: Duration) -> Duration {
let latest = THREAD_LATEST_INSTANT
.try_with(std::cell::Cell::get)
.unwrap_or(Duration::ZERO);
let offset = if offset < latest {
let behind = latest.saturating_sub(offset);
let _ = AMBIENT.try_with(|stack| {
if let Ok(mut stack) = stack.try_borrow_mut()
&& let Some(entry) = stack.last_mut()
{
entry.floor = entry.floor.saturating_add(behind);
}
});
latest
} else {
offset
};
let _ = THREAD_LATEST_INSTANT.try_with(|cell| cell.set(offset));
offset
}
fn with_ambient<T>(read: impl FnOnce(&dyn ClockSource) -> T) -> T {
match ambient_entry() {
Some((clock, _, _)) => read(clock.as_ref()),
None => read(&SystemClock),
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct AmbientClock;
impl ClockSource for AmbientClock {
fn now(&self) -> DateTime<Utc> {
ambient_now()
}
fn monotonic(&self) -> MonotonicInstant {
ambient_monotonic()
}
}
#[must_use]
pub fn ambient_now() -> DateTime<Utc> {
with_ambient(ClockSource::now)
}
#[must_use]
pub fn ambient_monotonic() -> MonotonicInstant {
with_ambient(ClockSource::monotonic)
}
#[must_use]
pub fn ambient_instant() -> std::time::Instant {
let offset = match ambient_entry() {
Some((clock, floor, start)) => keep_thread_instants_forward(
floor.saturating_add(clock.monotonic().since_origin().saturating_sub(start)),
),
None => SystemClock.monotonic().since_origin(),
};
INSTANT_HIGH_WATER.fetch_max(
u64::try_from(offset.as_nanos()).unwrap_or(u64::MAX),
std::sync::atomic::Ordering::Relaxed,
);
*MONOTONIC_ORIGIN + offset
}
pub(crate) fn system_instant() -> std::time::Instant {
*MONOTONIC_ORIGIN + SystemClock.monotonic().since_origin()
}
#[must_use]
pub fn ambient_system_time() -> std::time::SystemTime {
std::time::UNIX_EPOCH + clock_unix_duration(&AmbientClock)
}
#[must_use]
pub fn clock_unix_secs(clock: &dyn ClockSource) -> u64 {
clock_unix_duration(clock).as_secs()
}
#[must_use]
pub fn clock_unix_duration(clock: &dyn ClockSource) -> std::time::Duration {
let now = clock.now();
let ts = now.timestamp();
if ts >= 0 {
std::time::Duration::new(ts.cast_unsigned(), now.timestamp_subsec_nanos())
} else {
std::time::Duration::ZERO
}
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::TimeZone;
#[test]
fn system_clock_returns_time_close_to_utc_now() {
let clock = SystemClock;
let a = clock.now();
let b = Utc::now();
assert!(
(b - a).num_seconds().abs() < 1,
"SystemClock should be within 1s of Utc::now()"
);
}
#[test]
fn fixed_clock_always_returns_same_time() {
let pinned = Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap();
let clock = FixedClock::at(pinned);
assert_eq!(clock.now(), pinned);
assert_eq!(clock.now(), pinned);
}
#[test]
fn ticking_clock_starts_at_given_time() {
let start = Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap();
let clock = TickingClock::starting_at(start);
assert_eq!(clock.now(), start);
}
#[test]
fn ticking_clock_advances_correctly() {
let start = Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap();
let clock = TickingClock::starting_at(start);
clock.advance(std::time::Duration::from_secs(3600));
assert_eq!(clock.now(), start + chrono::Duration::hours(1));
}
#[test]
fn ticking_clock_clone_shares_state() {
let start = Utc.with_ymd_and_hms(2025, 6, 1, 12, 0, 0).unwrap();
let clock = TickingClock::starting_at(start);
let clone = clock.clone();
clock.advance(std::time::Duration::from_secs(86400));
assert_eq!(clone.now(), start + chrono::Duration::days(1));
}
#[test]
fn clock_unix_secs_uses_clock_timestamp() {
let pinned = Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap();
let clock = FixedClock::at(pinned);
let secs = clock_unix_secs(&clock);
assert_eq!(secs, pinned.timestamp().cast_unsigned());
}
#[test]
fn monotonic_saturates_instead_of_underflowing_on_inversion() {
let early = MonotonicInstant::from_origin_elapsed(Duration::from_secs(1));
let late = MonotonicInstant::from_origin_elapsed(Duration::from_secs(5));
assert_eq!(
late.saturating_duration_since(early),
Duration::from_secs(4)
);
assert_eq!(early.saturating_duration_since(late), Duration::ZERO);
assert_eq!(early.elapsed_at(late), Duration::from_secs(4));
}
#[test]
fn monotonic_add_saturates_instead_of_panicking() {
let base = MonotonicInstant::ORIGIN;
assert_eq!(
base.saturating_add(Duration::from_secs(30)).since_origin(),
Duration::from_secs(30)
);
assert_eq!(
base.checked_add(Duration::MAX),
Some(MonotonicInstant::from_origin_elapsed(Duration::MAX))
);
let high = MonotonicInstant::from_origin_elapsed(Duration::MAX);
assert_eq!(high.checked_add(Duration::from_secs(1)), None);
assert_eq!(high.saturating_add(Duration::from_secs(1)), high);
}
#[test]
fn ticking_clock_monotonic_moves_in_lockstep_with_wall_time() {
let start = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let clock = TickingClock::starting_at(start);
assert_eq!(clock.monotonic(), MonotonicInstant::ORIGIN);
clock.advance(Duration::from_secs(90));
assert_eq!(clock.monotonic().since_origin(), Duration::from_secs(90));
assert_eq!(clock.now(), start + chrono::Duration::seconds(90));
let clone = clock.clone();
clock.advance(Duration::from_secs(10));
assert_eq!(clone.monotonic().since_origin(), Duration::from_secs(100));
}
#[test]
fn ticking_clock_monotonic_is_reproducible_across_instances() {
let start = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let a = TickingClock::starting_at(start);
let b = TickingClock::starting_at(start);
for step in [1u64, 60, 3_600, 86_400] {
a.advance(Duration::from_secs(step));
b.advance(Duration::from_secs(step));
}
assert_eq!(a.monotonic(), b.monotonic());
assert_eq!(a.monotonic().since_origin(), Duration::from_secs(90_061));
}
#[test]
fn fixed_clock_monotonic_never_advances() {
let clock = FixedClock::at(Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap());
let first = clock.monotonic();
std::thread::yield_now();
assert_eq!(first, MonotonicInstant::ORIGIN);
assert_eq!(clock.monotonic(), first);
}
#[test]
fn system_clock_monotonic_is_real_and_non_decreasing() {
let clock = SystemClock;
let first = clock.monotonic();
let second = clock.monotonic();
assert!(
second >= first,
"the real monotonic clock must never go backwards"
);
assert!(monotonic_now() >= second);
let base = clock.monotonic();
let mut advanced = base;
for _ in 0..50_000_000u64 {
advanced = clock.monotonic();
if advanced > base {
break;
}
}
assert!(
advanced > base,
"ClockSource::monotonic's default body must advance with real time"
);
}
#[test]
fn a_custom_clock_keeps_compiling_and_gets_real_monotonic_by_default() {
#[derive(Debug)]
struct LegacyClock;
impl ClockSource for LegacyClock {
fn now(&self) -> DateTime<Utc> {
Utc.with_ymd_and_hms(1999, 12, 31, 23, 59, 59).unwrap()
}
}
let clock = LegacyClock;
let first = clock.monotonic();
assert!(clock.monotonic() >= first);
}
#[test]
fn a_backwards_wall_clock_cannot_produce_a_negative_elapsed() {
let start = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let clock = TickingClock::starting_at(start);
clock.advance(Duration::from_secs(10));
let later = clock.monotonic();
let earlier = MonotonicInstant::from_origin_elapsed(Duration::from_secs(100));
assert_eq!(later.saturating_duration_since(earlier), Duration::ZERO);
}
#[test]
fn clock_unix_secs_exact_epoch() {
let epoch = Utc.with_ymd_and_hms(1970, 1, 1, 0, 0, 0).unwrap();
let clock = FixedClock::at(epoch);
assert_eq!(clock_unix_secs(&clock), 0);
}
#[test]
fn clock_unix_duration_zero_for_pre_epoch() {
let pre_epoch = Utc.with_ymd_and_hms(1969, 12, 31, 23, 59, 59).unwrap();
let clock = FixedClock::at(pre_epoch);
assert_eq!(clock_unix_duration(&clock), std::time::Duration::ZERO);
}
#[test]
fn clock_unix_duration_exact_epoch_with_nanos() {
let epoch_with_nanos =
Utc.with_ymd_and_hms(1970, 1, 1, 0, 0, 0).unwrap() + chrono::Duration::nanoseconds(100);
let clock = FixedClock::at(epoch_with_nanos);
assert_eq!(
clock_unix_duration(&clock),
std::time::Duration::from_nanos(100)
);
}
#[test]
fn clock_unix_duration_post_epoch() {
let post_epoch = Utc.with_ymd_and_hms(1970, 1, 1, 0, 0, 1).unwrap();
let clock = FixedClock::at(post_epoch);
assert_eq!(
clock_unix_duration(&clock),
std::time::Duration::from_secs(1)
);
}
#[test]
fn ambient_clock_is_installed_nested_and_removed() {
let pinned = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let later = Utc.with_ymd_and_hms(2021, 1, 1, 0, 0, 0).unwrap();
let outer = install_ambient(Arc::new(FixedClock::at(pinned)));
assert_eq!(ambient_now(), pinned);
let inner = install_ambient(Arc::new(FixedClock::at(later)));
assert_eq!(ambient_now(), later, "the newest clock wins");
drop(inner);
assert_eq!(ambient_now(), pinned);
drop(outer);
assert!(ambient_now() > later, "the system clock again");
}
#[test]
fn a_later_timeline_starts_after_earlier_instants() {
let epoch = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let hour = std::time::Duration::from_secs(3600);
let first = TickingClock::starting_at(epoch);
let guard = install_ambient(Arc::new(first.clone()));
let start = ambient_instant();
first.advance(hour);
let stored = ambient_instant();
assert_eq!(stored.saturating_duration_since(start), hour);
drop(guard);
let second = TickingClock::starting_at(epoch);
let _guard = install_ambient(Arc::new(second.clone()));
let fresh = ambient_instant();
assert!(fresh >= stored, "a new timeline does not go back");
second.advance(hour);
assert_eq!(ambient_instant().saturating_duration_since(fresh), hour);
}
#[test]
fn framework_code_spawns_blocking_work_through_the_ambient_helper() {
fn visit(dir: &std::path::Path, offenders: &mut Vec<String>) {
for entry in std::fs::read_dir(dir).unwrap() {
let path = entry.unwrap().path();
if path.is_dir() {
visit(&path, offenders);
} else if path.extension().is_some_and(|ext| ext == "rs")
&& !path.ends_with("src/time.rs")
&& std::fs::read_to_string(&path)
.unwrap()
.contains("task::spawn_blocking")
{
offenders.push(path.display().to_string());
}
}
}
let mut offenders = Vec::new();
visit(
&std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"),
&mut offenders,
);
assert!(
offenders.is_empty(),
"use crate::time::spawn_blocking instead of tokio::task::spawn_blocking in {offenders:?}"
);
}
#[tokio::test]
async fn spawn_blocking_reads_the_callers_ambient_clock() {
let epoch = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let clock = TickingClock::starting_at(epoch);
let guard = install_ambient(Arc::new(clock.clone()));
clock.advance(std::time::Duration::from_secs(3600));
let here = (ambient_now(), ambient_instant());
let there = spawn_blocking(|| (ambient_now(), ambient_instant()))
.await
.unwrap();
assert_eq!(there, here, "the blocking thread reads the same clock");
drop(guard);
let there = spawn_blocking(ambient_now).await.unwrap();
assert!((Utc::now() - there).num_seconds().abs() < 5);
}
#[test]
fn an_outer_timeline_goes_on_after_a_nested_one() {
let epoch = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let hour = std::time::Duration::from_secs(3600);
let outer = TickingClock::starting_at(epoch);
let _outer = install_ambient(Arc::new(outer.clone()));
let before = ambient_instant();
let inner = TickingClock::starting_at(epoch);
let guard = install_ambient(Arc::new(inner.clone()));
inner.advance(hour);
let stored = ambient_instant();
assert!(stored.saturating_duration_since(before) >= hour);
drop(guard);
let back = ambient_instant();
assert!(back >= stored, "the outer timeline does not go back");
outer.advance(hour);
assert_eq!(ambient_instant().saturating_duration_since(back), hour);
}
#[test]
fn ambient_guard_dropped_on_another_thread_uninstalls() {
let pinned = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let guard = install_ambient(Arc::new(FixedClock::at(pinned)));
assert_eq!(ambient_now(), pinned);
std::thread::spawn(move || drop(guard)).join().unwrap();
assert_ne!(
ambient_now(),
pinned,
"the entry is skipped once its guard drops"
);
}
}