xmtp_common/
time.rs

1//! Time primitives for native and WebAssembly
2
3use crate::{if_native, if_wasm, wasm_or_native};
4use std::fmt;
5
6if_native! {
7    pub use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
8}
9if_wasm! {
10    pub use web_time::{Duration, Instant, SystemTime, UNIX_EPOCH};
11}
12
13#[derive(Debug)]
14pub struct Expired;
15
16impl std::error::Error for Expired {
17    fn description(&self) -> &str {
18        "Timer duration expired"
19    }
20}
21
22if_native! {
23    impl From<tokio::time::error::Elapsed> for Expired {
24        fn from(_: tokio::time::error::Elapsed) -> Expired {
25            Expired
26        }
27    }
28}
29
30impl fmt::Display for Expired {
31    fn fmt(&self, f: &mut fmt::Formatter) -> std::fmt::Result {
32        write!(f, "timer duration expired")
33    }
34}
35
36impl crate::ErrorCode for Expired {
37    fn error_code(&self) -> &'static str {
38        "Expired"
39    }
40}
41
42fn duration_since_epoch() -> Duration {
43    SystemTime::now()
44        .duration_since(UNIX_EPOCH)
45        .expect("Time went backwards")
46}
47
48pub fn now_ns() -> i64 {
49    duration_since_epoch().as_nanos() as i64
50}
51
52pub fn now_ms() -> u64 {
53    duration_since_epoch().as_millis() as u64
54}
55pub fn now_secs() -> i64 {
56    duration_since_epoch().as_secs() as i64
57}
58
59pub async fn timeout<F>(duration: Duration, future: F) -> Result<F::Output, Expired>
60where
61    F: std::future::IntoFuture,
62{
63    wasm_or_native! {
64        wasm => {
65            use futures::future::Either::*;
66            let millis = duration.as_millis().min(u32::MAX as u128) as u32;
67            let timeout = gloo_timers::future::TimeoutFuture::new(millis);
68            let future = future.into_future();
69            futures::pin_mut!(future);
70            match futures::future::select(timeout, future).await {
71                Left(_) => Err(Expired),
72                Right((value, _)) => Ok(value),
73            }
74        },
75        native => {
76            tokio::time::timeout(duration, future).await.map_err(Into::into)
77        }
78    }
79}
80
81#[doc(hidden)]
82pub async fn sleep(duration: Duration) {
83    wasm_or_native! {
84        native => {tokio::time::sleep(duration).await},
85        wasm => {gloo_timers::future::sleep(duration).await},
86    }
87}
88
89pub fn interval_stream(
90    period: crate::time::Duration,
91) -> impl futures::Stream<Item = crate::time::Instant> {
92    use futures::StreamExt;
93    wasm_or_native! {
94        wasm => {gloo_timers::future::IntervalStream::new(period.as_millis().min(u32::MAX as u128) as u32).map(|_| crate::time::Instant::now())},
95        native => {tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(period)).map(|t| t.into_std())},
96    }
97}
98
99/// Draw a random delay in `[0, jitter]`. Mirrors the jitter draw in
100/// [`crate::retry`] so the same cross-platform `rand` path is exercised.
101///
102/// Public so workers that compute their own per-iteration sleeps (rather than
103/// driving [`jittered_interval_stream`]) can de-synchronize fleet-wide wakes.
104pub fn rand_offset(jitter: crate::time::Duration) -> crate::time::Duration {
105    use rand::RngExt;
106    if jitter.is_zero() {
107        return crate::time::Duration::ZERO;
108    }
109    let distr = rand::distr::Uniform::new_inclusive(crate::time::Duration::ZERO, jitter).unwrap();
110    rand::rng().sample(distr)
111}
112
113/// Like [`interval_stream`], but de-synchronizes fleets of clients that boot
114/// together:
115/// - a one-time startup offset in `[0, jitter]` is awaited before the first
116///   tick, so clients booted at the same time don't tick in lockstep;
117/// - each subsequent tick adds a fresh random delay in `[0, jitter]`, so they
118///   don't re-converge over time.
119///
120/// `jitter == Duration::ZERO` degenerates to [`interval_stream`] and pulls no
121/// randomness — the zero-jitter path is byte-for-byte the old behavior.
122pub fn jittered_interval_stream(
123    base: crate::time::Duration,
124    jitter: crate::time::Duration,
125) -> impl futures::Stream<Item = crate::time::Instant> + Unpin {
126    use futures::StreamExt;
127
128    // The jittered branch composes `then`/`flat_map` over per-tick sleep
129    // futures, which makes the stream `!Unpin`. Workers consume the stream by
130    // value in a `while let` loop (like `interval_stream`), so box it to keep
131    // the same `Unpin` ergonomics. Native must stay `Send` (workers spawn on a
132    // multi-thread runtime); wasm cannot require `Send`.
133    let jittered = move || {
134        let startup_offset = rand_offset(jitter);
135        // Sleep the startup offset once, then on every base-interval tick sleep
136        // an additional per-tick jitter before yielding the instant.
137        futures::stream::once(async move {
138            sleep(startup_offset).await;
139        })
140        .flat_map(move |_| {
141            interval_stream(base).then(move |instant| async move {
142                sleep(rand_offset(jitter)).await;
143                instant
144            })
145        })
146    };
147
148    wasm_or_native! {
149        wasm => {
150            if jitter.is_zero() {
151                interval_stream(base).boxed_local()
152            } else {
153                jittered().boxed_local()
154            }
155        },
156        native => {
157            if jitter.is_zero() {
158                interval_stream(base).boxed()
159            } else {
160                jittered().boxed()
161            }
162        },
163    }
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169    use crate::ErrorCode;
170
171    #[test]
172    fn test_expired_error_code() {
173        let expired = Expired;
174        assert_eq!(expired.error_code(), "Expired");
175    }
176
177    #[test]
178    fn test_expired_display() {
179        let expired = Expired;
180        assert_eq!(format!("{}", expired), "timer duration expired");
181    }
182
183    #[test]
184    fn test_expired_description() {
185        use std::error::Error;
186        #[allow(deprecated)]
187        let desc = Expired.description();
188        assert_eq!(desc, "Timer duration expired");
189    }
190
191    #[test]
192    fn test_now_ns_returns_positive() {
193        let ns = now_ns();
194        assert!(ns > 0, "now_ns should return a positive value");
195    }
196
197    #[test]
198    fn test_now_ms_returns_positive() {
199        let ms = now_ms();
200        assert!(ms > 0, "now_ms should return a positive value");
201    }
202
203    #[test]
204    fn test_now_secs_returns_positive() {
205        let secs = now_secs();
206        assert!(secs > 0, "now_secs should return a positive value");
207    }
208
209    // Jitter tests rely on tokio's paused virtual clock, native-only.
210    #[cfg(not(target_arch = "wasm32"))]
211    mod jitter {
212        use super::super::jittered_interval_stream;
213        use crate::time::Duration;
214        use futures::StreamExt;
215
216        // Under `start_paused`, the runtime auto-advances the virtual clock to
217        // the next pending timer whenever all tasks are idle. So we measure how
218        // far the clock moved (`Instant::now` elapsed) rather than driving it by
219        // hand — that sidesteps ordering pitfalls between `advance` and when the
220        // stream's sleeps get armed.
221
222        #[tokio::test(start_paused = true)]
223        async fn zero_jitter_uses_base_period() {
224            let base = Duration::from_secs(10);
225            let start = tokio::time::Instant::now();
226            let mut s = Box::pin(jittered_interval_stream(base, Duration::ZERO));
227            // First tick is immediate (tokio interval fires at t=0, no offset).
228            s.next().await.unwrap();
229            assert_eq!(start.elapsed(), Duration::ZERO, "first tick is immediate");
230            // Second tick lands exactly `base` later — no jitter.
231            s.next().await.unwrap();
232            assert_eq!(start.elapsed(), base, "second tick after exactly base");
233        }
234
235        #[tokio::test(start_paused = true)]
236        async fn jitter_keeps_ticks_within_bounds() {
237            let base = Duration::from_secs(10);
238            let jitter = Duration::from_secs(5);
239            let start = tokio::time::Instant::now();
240            let mut s = Box::pin(jittered_interval_stream(base, jitter));
241            // First yield = startup_offset + first per-tick jitter, each in
242            // [0, jitter]; the base interval fires at t=0. So [0, 2*jitter].
243            s.next().await.unwrap();
244            let first = start.elapsed();
245            assert!(
246                first <= jitter * 2,
247                "first tick within 2*jitter, got {first:?}"
248            );
249            // The underlying tokio interval anchors tick N at N*base, so its own
250            // schedule self-corrects for prior per-tick jitter. Yield N lands at
251            // startup_offset + N*base + jitter_N. The gap between consecutive
252            // yields is therefore base + (jitter_next - jitter_prev), bounded by
253            // [base - jitter, base + jitter].
254            s.next().await.unwrap();
255            let gap = start.elapsed() - first;
256            assert!(
257                gap >= base.saturating_sub(jitter),
258                "gap >= base - jitter, got {gap:?}"
259            );
260            assert!(gap <= base + jitter, "gap <= base + jitter, got {gap:?}");
261        }
262    }
263}
xmtp_common/time.rs

xmtp_common/
time.rs
