use crate::{Clock, Timer};
use futures::channel::oneshot;
use futures::future::BoxFuture;
use jiff::Timestamp;
use std::sync::{Arc, Mutex};
use std::time::Duration;
pub struct FrozenClock {
instant: Timestamp,
}
impl FrozenClock {
pub fn at(rfc3339: &str) -> Self {
let instant = rfc3339.parse().unwrap_or(Timestamp::UNIX_EPOCH);
Self { instant }
}
pub fn at_instant(instant: Timestamp) -> Self {
Self { instant }
}
}
impl Clock for FrozenClock {
fn now(&self) -> Timestamp {
self.instant
}
}
type PendingSleeps = Vec<(Timestamp, oneshot::Sender<()>)>;
pub struct MockClock {
inner: Arc<Mutex<Timestamp>>,
pending: Arc<Mutex<PendingSleeps>>,
}
impl MockClock {
pub fn new(start: Timestamp) -> Self {
Self {
inner: Arc::new(Mutex::new(start)),
pending: Arc::new(Mutex::new(Vec::new())),
}
}
pub fn set_now(&self, instant: Timestamp) {
let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
*guard = instant;
drop(guard);
self.fire_pending();
}
pub fn advance_by(&self, dur: Duration) {
let new_now = {
let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
*guard = guard.checked_add(dur).unwrap_or(*guard);
*guard
};
self.fire_ready(new_now);
}
pub fn advance_and_fire(&self, dur: Duration) {
self.advance_by(dur);
}
fn current(&self) -> Timestamp {
*self.inner.lock().unwrap_or_else(|e| e.into_inner())
}
fn fire_pending(&self) {
let now = self.current();
self.fire_ready(now);
}
fn fire_ready(&self, now: Timestamp) {
let to_fire: Vec<oneshot::Sender<()>> = {
let mut pending = self.pending.lock().unwrap_or_else(|e| e.into_inner());
let mut fired = Vec::new();
let mut remaining: PendingSleeps = Vec::new();
let mut original: PendingSleeps = Vec::new();
std::mem::swap(&mut *pending, &mut original);
for (wake_at, sender) in original {
if wake_at <= now {
fired.push(sender);
} else {
remaining.push((wake_at, sender));
}
}
*pending = remaining;
fired
};
for sender in to_fire {
let _ = sender.send(());
}
}
pub(crate) fn register_sleep(&self, wake_at: Timestamp) -> oneshot::Receiver<()> {
let (tx, rx) = oneshot::channel();
let now = self.current();
if wake_at <= now {
let _ = tx.send(());
} else {
let mut pending = self.pending.lock().unwrap_or_else(|e| e.into_inner());
pending.push((wake_at, tx));
}
rx
}
}
impl Clock for MockClock {
fn now(&self) -> Timestamp {
self.current()
}
}
pub struct AdvanceableTimer {
clock: Arc<MockClock>,
}
impl AdvanceableTimer {
pub fn new(clock: Arc<MockClock>) -> Self {
Self { clock }
}
}
impl Timer for AdvanceableTimer {
fn sleep(&self, dur: Duration) -> BoxFuture<'static, ()> {
let now = self.clock.current();
let wake_at = now.checked_add(dur).unwrap_or(now);
let rx = self.clock.register_sleep(wake_at);
Box::pin(async move {
let _ = rx.await;
})
}
fn next_tick(&self) -> BoxFuture<'static, ()> {
Box::pin(futures::future::ready(()))
}
}
use std::sync::Arc as StdArc;
pub fn frozen_at(rfc3339: &str) -> StdArc<dyn Clock> {
StdArc::new(FrozenClock::at(rfc3339))
}
pub fn mock_pair(start: Timestamp) -> (StdArc<MockClock>, StdArc<AdvanceableTimer>) {
let clock = StdArc::new(MockClock::new(start));
let timer = StdArc::new(AdvanceableTimer::new(StdArc::clone(&clock)));
(clock, timer)
}
#[cfg(test)]
mod tests {
use super::*;
fn t0() -> Timestamp {
"2026-05-10T00:00:00Z".parse().expect("valid RFC 3339")
}
#[test]
fn frozen_at_parses_rfc3339() {
let clock = FrozenClock::at("2026-05-10T00:00:00Z");
assert_eq!(clock.now(), t0());
}
#[test]
fn frozen_clock_at_instant() {
let clock = FrozenClock::at_instant(t0());
assert_eq!(clock.now(), t0());
}
#[test]
fn frozen_clock_never_advances() {
let clock = FrozenClock::at_instant(t0());
let a = clock.now();
let b = clock.now();
assert_eq!(a, b);
}
#[test]
fn frozen_clock_unix_millis() {
let clock = FrozenClock::at("1970-01-01T00:00:01Z");
assert_eq!(clock.now_unix_millis(), 1_000);
}
#[test]
fn frozen_at_helper_trait_dispatch() {
let c: StdArc<dyn Clock> = frozen_at("2026-05-10T00:00:00Z");
assert_eq!(c.now(), t0());
}
#[test]
fn mock_clock_starts_at_given_instant() {
let clock = MockClock::new(t0());
assert_eq!(clock.now(), t0());
}
#[test]
fn mock_clock_advance_by_increases_time() {
let clock = MockClock::new(t0());
clock.advance_by(Duration::from_secs(1));
assert_eq!(clock.now(), t0() + jiff::SignedDuration::from_secs(1));
}
#[test]
fn mock_clock_set_now_updates_instant() {
let clock = MockClock::new(t0());
let new_now = t0() + jiff::SignedDuration::from_hours(5);
clock.set_now(new_now);
assert_eq!(clock.now(), new_now);
}
#[test]
fn mock_clock_trait_dispatch() {
let c: StdArc<dyn Clock> = StdArc::new(MockClock::new(t0()));
assert_eq!(c.now(), t0());
}
#[tokio::test]
async fn advanceable_timer_sleep_resolves_after_advance() {
let (clock, timer) = mock_pair(t0());
let sleep_fut = timer.sleep(Duration::from_secs(60));
let clock2 = StdArc::clone(&clock);
tokio::spawn(async move {
clock2.advance_and_fire(Duration::from_secs(60));
})
.await
.ok();
sleep_fut.await; }
#[tokio::test]
async fn advanceable_timer_partial_advance_fires_only_due_sleeps() {
let (clock, timer) = mock_pair(t0());
let sleep_30 = timer.sleep(Duration::from_secs(30));
let sleep_60 = timer.sleep(Duration::from_secs(60));
clock.advance_and_fire(Duration::from_secs(45));
sleep_30.await;
clock.advance_and_fire(Duration::from_secs(15));
sleep_60.await;
}
#[tokio::test]
async fn advanceable_timer_next_tick_resolves_immediately() {
let (_, timer) = mock_pair(t0());
timer.next_tick().await;
}
#[test]
fn mock_pair_helper_returns_linked_pair() {
let (clock, timer) = mock_pair(t0());
let _: StdArc<MockClock> = clock;
let _: StdArc<AdvanceableTimer> = timer;
}
#[test]
fn frozen_clock_send_sync() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<FrozenClock>();
}
#[test]
fn mock_clock_send_sync() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<MockClock>();
assert_send_sync::<AdvanceableTimer>();
}
}