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    // Jitter tests rely on tokio's paused virtual clock, native-only.
192    #[cfg(not(target_arch = "wasm32"))]
193    mod jitter {
194        use super::super::jittered_interval_stream;
195        use crate::time::Duration;
196        use futures::StreamExt;
197
198        // Under `start_paused`, the runtime auto-advances the virtual clock to
199        // the next pending timer whenever all tasks are idle. So we measure how
200        // far the clock moved (`Instant::now` elapsed) rather than driving it by
201        // hand — that sidesteps ordering pitfalls between `advance` and when the
202        // stream's sleeps get armed.
203
204        #[tokio::test(start_paused = true)]
205        async fn zero_jitter_uses_base_period() {
206            let base = Duration::from_secs(10);
207            let start = tokio::time::Instant::now();
208            let mut s = Box::pin(jittered_interval_stream(base, Duration::ZERO));
209            // First tick is immediate (tokio interval fires at t=0, no offset).
210            s.next().await.unwrap();
211            assert_eq!(start.elapsed(), Duration::ZERO, "first tick is immediate");
212            // Second tick lands exactly `base` later — no jitter.
213            s.next().await.unwrap();
214            assert_eq!(start.elapsed(), base, "second tick after exactly base");
215        }
216
217        #[tokio::test(start_paused = true)]
218        async fn jitter_keeps_ticks_within_bounds() {
219            let base = Duration::from_secs(10);
220            let jitter = Duration::from_secs(5);
221            let start = tokio::time::Instant::now();
222            let mut s = Box::pin(jittered_interval_stream(base, jitter));
223            // First yield = startup_offset + first per-tick jitter, each in
224            // [0, jitter]; the base interval fires at t=0. So [0, 2*jitter].
225            s.next().await.unwrap();
226            let first = start.elapsed();
227            assert!(
228                first <= jitter * 2,
229                "first tick within 2*jitter, got {first:?}"
230            );
231            // The underlying tokio interval anchors tick N at N*base, so its own
232            // schedule self-corrects for prior per-tick jitter. Yield N lands at
233            // startup_offset + N*base + jitter_N. The gap between consecutive
234            // yields is therefore base + (jitter_next - jitter_prev), bounded by
235            // [base - jitter, base + jitter].
236            s.next().await.unwrap();
237            let gap = start.elapsed() - first;
238            assert!(
239                gap >= base.saturating_sub(jitter),
240                "gap >= base - jitter, got {gap:?}"
241            );
242            assert!(gap <= base + jitter, "gap <= base + jitter, got {gap:?}");
243        }
244    }
245}
xmtp_common/time.rs

xmtp_common/
time.rs
