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/// A wait longer than any process lives. An instant or deadline that would
82/// overflow saturates to this far ahead instead of panicking.
83pub const FAR_FUTURE: Duration = Duration::from_secs(100 * 365 * 24 * 60 * 60);
84
85/// The instant `duration` after `now`, saturating at [`FAR_FUTURE`].
86pub fn saturating_after(now: Instant, duration: Duration) -> Instant {
87    now.checked_add(duration)
88        .unwrap_or_else(|| now + FAR_FUTURE)
89}
90
91/// The longest wait one browser timer holds: a longer one fires at once.
92const MAX_TIMER: Duration = Duration::from_millis(i32::MAX as u64);
93
94/// Split a wait into pieces one browser timer can hold, in order. gloo panics
95/// past `u32::MAX` milliseconds and a timer fires at once past `i32::MAX`.
96#[cfg_attr(not(target_arch = "wasm32"), allow(dead_code))]
97fn timer_pieces(duration: Duration) -> impl Iterator<Item = Duration> {
98    let (full, rest) = timer_split(duration);
99    (0..full)
100        .map(|_| MAX_TIMER)
101        .chain((!rest.is_zero() || full == 0).then_some(rest))
102}
103
104/// The number of full browser timers in `duration`, and the remainder.
105#[cfg_attr(not(target_arch = "wasm32"), allow(dead_code))]
106fn timer_split(duration: Duration) -> (u128, Duration) {
107    let timer = MAX_TIMER.as_nanos();
108    let rest = u64::try_from(duration.as_nanos() % timer).expect("a remainder fits one timer");
109    (duration.as_nanos() / timer, Duration::from_nanos(rest))
110}
111
112/// Waits at least `duration`, as [`Instant`] measures it.
113#[doc(hidden)]
114pub async fn sleep(duration: Duration) {
115    wasm_or_native! {
116        native => {tokio::time::sleep(duration).await},
117        wasm => {
118            // A wait too long for the clock saturates, and so do its timers.
119            let now = Instant::now();
120            let deadline = saturating_after(now, duration);
121            for piece in timer_pieces(deadline.saturating_duration_since(now)) {
122                gloo_timers::future::sleep(piece).await;
123            }
124            // A browser timer drops the part of a wait below 1 ms, and Node
125            // can fire one up to 1 ms early, so a rate-limit wait could end
126            // just before its delay. Wait out what is left.
127            loop {
128                let left = deadline.saturating_duration_since(Instant::now());
129                if left.is_zero() {
130                    break;
131                }
132                gloo_timers::future::sleep(left.max(Duration::from_millis(1))).await;
133            }
134        },
135    }
136}
137
138pub fn interval_stream(
139    period: crate::time::Duration,
140) -> impl futures::Stream<Item = crate::time::Instant> {
141    use futures::StreamExt;
142    wasm_or_native! {
143        wasm => {gloo_timers::future::IntervalStream::new(period.as_millis().min(u32::MAX as u128) as u32).map(|_| crate::time::Instant::now())},
144        native => {tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(period)).map(|t| t.into_std())},
145    }
146}
147
148/// Draw a random delay in `[0, jitter]`. Mirrors the jitter draw in
149/// [`crate::retry`] so the same cross-platform `rand` path is exercised.
150///
151/// Public so workers that compute their own per-iteration sleeps (rather than
152/// driving [`jittered_interval_stream`]) can de-synchronize fleet-wide wakes.
153pub fn rand_offset(jitter: crate::time::Duration) -> crate::time::Duration {
154    use rand::RngExt;
155    if jitter.is_zero() {
156        return crate::time::Duration::ZERO;
157    }
158    let distr = rand::distr::Uniform::new_inclusive(crate::time::Duration::ZERO, jitter).unwrap();
159    rand::rng().sample(distr)
160}
161
162/// Like [`interval_stream`], but de-synchronizes fleets of clients that boot
163/// together:
164/// - a one-time startup offset in `[0, jitter]` is awaited before the first
165///   tick, so clients booted at the same time don't tick in lockstep;
166/// - each subsequent tick adds a fresh random delay in `[0, jitter]`, so they
167///   don't re-converge over time.
168///
169/// `jitter == Duration::ZERO` degenerates to [`interval_stream`] and pulls no
170/// randomness — the zero-jitter path is byte-for-byte the old behavior.
171pub fn jittered_interval_stream(
172    base: crate::time::Duration,
173    jitter: crate::time::Duration,
174) -> impl futures::Stream<Item = crate::time::Instant> + Unpin {
175    use futures::StreamExt;
176
177    // The jittered branch composes `then`/`flat_map` over per-tick sleep
178    // futures, which makes the stream `!Unpin`. Workers consume the stream by
179    // value in a `while let` loop (like `interval_stream`), so box it to keep
180    // the same `Unpin` ergonomics. Native must stay `Send` (workers spawn on a
181    // multi-thread runtime); wasm cannot require `Send`.
182    let jittered = move || {
183        let startup_offset = rand_offset(jitter);
184        // Sleep the startup offset once, then on every base-interval tick sleep
185        // an additional per-tick jitter before yielding the instant.
186        futures::stream::once(async move {
187            sleep(startup_offset).await;
188        })
189        .flat_map(move |_| {
190            interval_stream(base).then(move |instant| async move {
191                sleep(rand_offset(jitter)).await;
192                instant
193            })
194        })
195    };
196
197    wasm_or_native! {
198        wasm => {
199            if jitter.is_zero() {
200                interval_stream(base).boxed_local()
201            } else {
202                jittered().boxed_local()
203            }
204        },
205        native => {
206            if jitter.is_zero() {
207                interval_stream(base).boxed()
208            } else {
209                jittered().boxed()
210            }
211        },
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218    use crate::ErrorCode;
219
220    /// Every piece fits one browser timer, and the pieces sum to the wait.
221    #[xmtp_macro::test]
222    fn a_long_wait_splits_into_timer_pieces() {
223        for duration in [
224            Duration::ZERO,
225            MAX_TIMER,
226            MAX_TIMER + Duration::from_millis(1),
227            Duration::from_secs(60 * 24 * 60 * 60),
228        ] {
229            let pieces: Vec<_> = timer_pieces(duration).collect();
230            assert!(!pieces.is_empty());
231            assert!(pieces.iter().all(|piece| *piece <= MAX_TIMER));
232            assert_eq!(pieces.iter().sum::<Duration>(), duration);
233        }
234    }
235
236    /// A wait of more than `u32::MAX` timers still splits into pieces that each
237    /// fit one timer.
238    #[xmtp_macro::test]
239    fn a_wait_past_u32_max_timers_keeps_every_piece_bounded() {
240        let extra = Duration::from_millis(5);
241        let duration = MAX_TIMER * u32::MAX + MAX_TIMER * 2 + extra;
242        assert_eq!(timer_split(duration), (u128::from(u32::MAX) + 2, extra));
243        assert!(
244            timer_pieces(Duration::MAX)
245                .take(3)
246                .all(|piece| piece == MAX_TIMER)
247        );
248    }
249
250    /// A sleep never ends before its duration, including a part below 1 ms,
251    /// which a browser timer drops.
252    #[xmtp_macro::test]
253    async fn a_sleep_lasts_at_least_its_duration() {
254        for duration in [Duration::from_micros(700), Duration::from_micros(1_500)] {
255            for _ in 0..10 {
256                let started = Instant::now();
257                sleep(duration).await;
258                assert!(
259                    started.elapsed() >= duration,
260                    "{duration:?} sleep ended early"
261                );
262            }
263        }
264    }
265
266    /// A delay no clock can hold saturates instead of panicking.
267    #[xmtp_macro::test]
268    fn an_unrepresentable_delay_saturates() {
269        let now = Instant::now();
270        let until = saturating_after(now, Duration::MAX);
271        assert!(until >= now + FAR_FUTURE);
272        assert_eq!(
273            saturating_after(now, Duration::from_secs(5)),
274            now + Duration::from_secs(5)
275        );
276    }
277
278    #[test]
279    fn test_expired_error_code() {
280        let expired = Expired;
281        assert_eq!(expired.error_code(), "Expired");
282    }
283
284    #[test]
285    fn test_expired_display() {
286        let expired = Expired;
287        assert_eq!(format!("{}", expired), "timer duration expired");
288    }
289
290    #[test]
291    fn test_expired_description() {
292        use std::error::Error;
293        #[allow(deprecated)]
294        let desc = Expired.description();
295        assert_eq!(desc, "Timer duration expired");
296    }
297
298    // Jitter tests rely on tokio's paused virtual clock, native-only.
299    #[cfg(not(target_arch = "wasm32"))]
300    mod jitter {
301        use super::super::jittered_interval_stream;
302        use crate::time::Duration;
303        use futures::StreamExt;
304
305        // Under `start_paused`, the runtime auto-advances the virtual clock to
306        // the next pending timer whenever all tasks are idle. So we measure how
307        // far the clock moved (`Instant::now` elapsed) rather than driving it by
308        // hand — that sidesteps ordering pitfalls between `advance` and when the
309        // stream's sleeps get armed.
310
311        #[tokio::test(start_paused = true)]
312        async fn zero_jitter_uses_base_period() {
313            let base = Duration::from_secs(10);
314            let start = tokio::time::Instant::now();
315            let mut s = Box::pin(jittered_interval_stream(base, Duration::ZERO));
316            // First tick is immediate (tokio interval fires at t=0, no offset).
317            s.next().await.unwrap();
318            assert_eq!(start.elapsed(), Duration::ZERO, "first tick is immediate");
319            // Second tick lands exactly `base` later — no jitter.
320            s.next().await.unwrap();
321            assert_eq!(start.elapsed(), base, "second tick after exactly base");
322        }
323
324        #[tokio::test(start_paused = true)]
325        async fn jitter_keeps_ticks_within_bounds() {
326            let base = Duration::from_secs(10);
327            let jitter = Duration::from_secs(5);
328            let start = tokio::time::Instant::now();
329            let mut s = Box::pin(jittered_interval_stream(base, jitter));
330            // First yield = startup_offset + first per-tick jitter, each in
331            // [0, jitter]; the base interval fires at t=0. So [0, 2*jitter].
332            s.next().await.unwrap();
333            let first = start.elapsed();
334            assert!(
335                first <= jitter * 2,
336                "first tick within 2*jitter, got {first:?}"
337            );
338            // The underlying tokio interval anchors tick N at N*base, so its own
339            // schedule self-corrects for prior per-tick jitter. Yield N lands at
340            // startup_offset + N*base + jitter_N. The gap between consecutive
341            // yields is therefore base + (jitter_next - jitter_prev), bounded by
342            // [base - jitter, base + jitter].
343            s.next().await.unwrap();
344            let gap = start.elapsed() - first;
345            assert!(
346                gap >= base.saturating_sub(jitter),
347                "gap >= base - jitter, got {gap:?}"
348            );
349            assert!(gap <= base + jitter, "gap <= base + jitter, got {gap:?}");
350        }
351    }
352}
xmtp_common/time.rs

xmtp_common/
time.rs
