1use crate::{BoxFuture, Clock, Timer};
2use jiff::Timestamp;
3use std::future::Future;
4use std::pin::Pin;
5use std::sync::{Arc, Mutex, PoisonError};
6use std::task::{Context, Poll, Waker};
7use std::time::Duration;
8
9#[derive(Debug)]
13pub struct FrozenClock {
14 instant: Timestamp,
15}
16
17impl FrozenClock {
18 pub fn at(rfc3339: &str) -> Self {
21 let instant = rfc3339.parse().unwrap_or(Timestamp::UNIX_EPOCH);
22 Self { instant }
23 }
24
25 pub fn at_instant(instant: Timestamp) -> Self {
27 Self { instant }
28 }
29}
30
31impl Clock for FrozenClock {
32 fn now(&self) -> Timestamp {
33 self.instant
34 }
35}
36
37type PendingSleeps = Vec<(Timestamp, Alarm)>;
40
41#[derive(Debug, Default)]
43struct Slot {
44 rung: bool,
45 waker: Option<Waker>,
46}
47
48#[derive(Debug)]
52struct Alarm(Arc<Mutex<Slot>>);
53
54impl Drop for Alarm {
55 fn drop(&mut self) {
56 let waker = {
57 let mut slot = self.0.lock().unwrap_or_else(PoisonError::into_inner);
58 slot.rung = true;
59 slot.waker.take()
60 };
61 if let Some(waker) = waker {
62 waker.wake();
63 }
64 }
65}
66
67#[derive(Debug)]
69struct Sleep(Arc<Mutex<Slot>>);
70
71impl Future for Sleep {
72 type Output = ();
73
74 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
75 let mut slot = self.0.lock().unwrap_or_else(PoisonError::into_inner);
76 if slot.rung {
77 Poll::Ready(())
78 } else {
79 slot.waker = Some(cx.waker().clone());
80 Poll::Pending
81 }
82 }
83}
84
85#[derive(Debug)]
87pub struct MockClock {
88 inner: Arc<Mutex<Timestamp>>,
89 pending: Arc<Mutex<PendingSleeps>>,
90}
91
92impl MockClock {
93 pub fn new(start: Timestamp) -> Self {
95 Self {
96 inner: Arc::new(Mutex::new(start)),
97 pending: Arc::new(Mutex::new(Vec::new())),
98 }
99 }
100
101 pub fn set_now(&self, instant: Timestamp) {
104 let mut guard = self.inner.lock().unwrap_or_else(PoisonError::into_inner);
105 *guard = instant;
106 drop(guard);
107 self.fire_pending();
108 }
109
110 pub fn advance_by(&self, dur: Duration) {
113 let new_now = {
114 let mut guard = self.inner.lock().unwrap_or_else(PoisonError::into_inner);
115 *guard = guard.checked_add(dur).unwrap_or(*guard);
116 *guard
117 };
118 self.fire_ready(new_now);
119 }
120
121 pub fn advance_and_fire(&self, dur: Duration) {
124 self.advance_by(dur);
125 }
126
127 fn current(&self) -> Timestamp {
128 *self.inner.lock().unwrap_or_else(PoisonError::into_inner)
129 }
130
131 fn fire_pending(&self) {
132 let now = self.current();
133 self.fire_ready(now);
134 }
135
136 fn fire_ready(&self, now: Timestamp) {
137 let to_fire: PendingSleeps = {
140 let mut pending = self.pending.lock().unwrap_or_else(PoisonError::into_inner);
141 let (fired, remaining): (PendingSleeps, PendingSleeps) = std::mem::take(&mut *pending)
142 .into_iter()
143 .partition(|(wake_at, _)| *wake_at <= now);
144 *pending = remaining;
145 fired
146 };
147 drop(to_fire);
149 }
150
151 fn register_sleep(&self, wake_at: Timestamp) -> Sleep {
152 let slot = Arc::new(Mutex::new(Slot::default()));
153 let alarm = Alarm(Arc::clone(&slot));
154 if wake_at <= self.current() {
155 drop(alarm);
157 } else {
158 let mut pending = self.pending.lock().unwrap_or_else(PoisonError::into_inner);
159 pending.push((wake_at, alarm));
160 }
161 Sleep(slot)
162 }
163}
164
165impl Clock for MockClock {
166 fn now(&self) -> Timestamp {
167 self.current()
168 }
169}
170
171#[derive(Debug)]
176pub struct AdvanceableTimer {
177 clock: Arc<MockClock>,
178}
179
180impl AdvanceableTimer {
181 pub fn new(clock: Arc<MockClock>) -> Self {
183 Self { clock }
184 }
185}
186
187impl Timer for AdvanceableTimer {
188 fn sleep(&self, dur: Duration) -> BoxFuture<'static, ()> {
189 let now = self.clock.current();
190 let wake_at = now.checked_add(dur).unwrap_or(now);
191 Box::pin(self.clock.register_sleep(wake_at))
192 }
193
194 fn next_tick(&self) -> BoxFuture<'static, ()> {
195 Box::pin(std::future::ready(()))
196 }
197}
198
199pub fn frozen_at(rfc3339: &str) -> Arc<dyn Clock> {
203 Arc::new(FrozenClock::at(rfc3339))
204}
205
206pub fn mock_pair(start: Timestamp) -> (Arc<MockClock>, Arc<AdvanceableTimer>) {
208 let clock = Arc::new(MockClock::new(start));
209 let timer = Arc::new(AdvanceableTimer::new(Arc::clone(&clock)));
210 (clock, timer)
211}
212
213#[cfg(test)]
214mod tests {
215 #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
216 use super::*;
217
218 fn t0() -> Timestamp {
219 "2026-05-10T00:00:00Z".parse().expect("valid RFC 3339")
220 }
221
222 #[test]
225 fn frozen_at_parses_rfc3339() {
226 let clock = FrozenClock::at("2026-05-10T00:00:00Z");
227 assert_eq!(clock.now(), t0());
228 }
229
230 #[test]
231 fn frozen_clock_at_instant() {
232 let clock = FrozenClock::at_instant(t0());
233 assert_eq!(clock.now(), t0());
234 }
235
236 #[test]
237 fn frozen_clock_never_advances() {
238 let clock = FrozenClock::at_instant(t0());
239 let a = clock.now();
240 let b = clock.now();
241 assert_eq!(a, b);
242 }
243
244 #[test]
245 fn frozen_clock_unix_millis() {
246 let clock = FrozenClock::at("1970-01-01T00:00:01Z");
247 assert_eq!(clock.now_unix_millis(), 1_000);
248 }
249
250 #[test]
251 fn frozen_at_helper_trait_dispatch() {
252 let c: Arc<dyn Clock> = frozen_at("2026-05-10T00:00:00Z");
253 assert_eq!(c.now(), t0());
254 }
255
256 #[test]
259 fn mock_clock_starts_at_given_instant() {
260 let clock = MockClock::new(t0());
261 assert_eq!(clock.now(), t0());
262 }
263
264 #[test]
265 fn mock_clock_advance_by_increases_time() {
266 let clock = MockClock::new(t0());
267 clock.advance_by(Duration::from_secs(1));
268 assert_eq!(clock.now(), t0() + jiff::SignedDuration::from_secs(1));
269 }
270
271 #[test]
272 fn mock_clock_set_now_updates_instant() {
273 let clock = MockClock::new(t0());
274 let new_now = t0() + jiff::SignedDuration::from_hours(5);
275 clock.set_now(new_now);
276 assert_eq!(clock.now(), new_now);
277 }
278
279 #[test]
280 fn mock_clock_trait_dispatch() {
281 let c: Arc<dyn Clock> = Arc::new(MockClock::new(t0()));
282 assert_eq!(c.now(), t0());
283 }
284
285 #[tokio::test]
288 async fn advanceable_timer_sleep_resolves_after_advance() {
289 let (clock, timer) = mock_pair(t0());
290 let sleep_fut = timer.sleep(Duration::from_secs(60));
291 let clock2 = Arc::clone(&clock);
293 tokio::spawn(async move {
294 clock2.advance_and_fire(Duration::from_secs(60));
295 })
296 .await
297 .ok();
298 sleep_fut.await; }
300
301 #[tokio::test]
302 async fn advanceable_timer_partial_advance_fires_only_due_sleeps() {
303 let (clock, timer) = mock_pair(t0());
304 let sleep_30 = timer.sleep(Duration::from_secs(30));
305 let sleep_60 = timer.sleep(Duration::from_secs(60));
306
307 clock.advance_and_fire(Duration::from_secs(45));
308
309 sleep_30.await;
311
312 clock.advance_and_fire(Duration::from_secs(15));
314 sleep_60.await;
315 }
316
317 #[tokio::test]
318 async fn a_dropped_clock_wakes_its_pending_sleeps() {
319 let (clock, timer) = mock_pair(t0());
320 let sleep = timer.sleep(Duration::from_secs(60));
321 drop(timer);
322 drop(clock);
323 sleep.await;
324 }
325
326 #[tokio::test]
327 async fn advanceable_timer_next_tick_resolves_immediately() {
328 let (_, timer) = mock_pair(t0());
329 timer.next_tick().await;
330 }
331
332 #[test]
333 fn mock_pair_helper_returns_linked_pair() {
334 let (clock, timer) = mock_pair(t0());
335 let _: Arc<MockClock> = clock;
336 let _: Arc<AdvanceableTimer> = timer;
337 }
338
339 #[test]
340 fn frozen_clock_send_sync() {
341 fn assert_send_sync<T: Send + Sync>() {}
342 assert_send_sync::<FrozenClock>();
343 }
344
345 #[test]
346 fn mock_clock_send_sync() {
347 fn assert_send_sync<T: Send + Sync>() {}
348 assert_send_sync::<MockClock>();
349 assert_send_sync::<AdvanceableTimer>();
350 }
351}