nautilus_core/time.rs
1// -------------------------------------------------------------------------------------------------
2// Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3// https://nautechsystems.io
4//
5// Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6// You may not use this file except in compliance with the License.
7// You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! The core `AtomicTime` for real-time and static clocks.
17//!
18//! This module provides an atomic time abstraction that supports both real-time and static
19//! clocks. It ensures thread-safe operations and monotonic time retrieval with nanosecond precision.
20//!
21//! # Modes
22//!
23//! - **Real-time mode:** The clock continuously syncs with system wall-clock time (via
24//! [`SystemTime::now()`]). To ensure strict monotonic increments across multiple threads,
25//! the internal updates use an atomic compare-and-exchange loop (`time_since_epoch`).
26//! While this guarantees that every new timestamp is at least one nanosecond greater than the
27//! last, it may introduce higher contention if many threads call it heavily.
28//!
29//! - **Static mode:** The clock is manually controlled via [`AtomicTime::set_time`] or [`AtomicTime::increment_time`],
30//! which can be useful for simulations or backtesting. You can switch modes at runtime using
31//! [`AtomicTime::make_realtime`] or [`AtomicTime::make_static`]. In **static mode**, we use
32//! acquire/release semantics so that updates from one thread can be observed by another;
33//! however, we do not enforce strict global ordering for manual updates. If you need strong,
34//! multi-threaded ordering in **static mode**, you must coordinate higher-level synchronization yourself.
35
36use std::{
37 sync::{
38 OnceLock,
39 atomic::{AtomicBool, AtomicU64, Ordering},
40 },
41 time::{Duration, SystemTime, UNIX_EPOCH},
42};
43
44use crate::{
45 DurationNanos, UnixNanos,
46 datetime::{NANOSECONDS_IN_MICROSECOND, NANOSECONDS_IN_MILLISECOND, NANOSECONDS_IN_SECOND},
47};
48
49/// Global atomic time in **real-time mode** for use across the system.
50///
51/// This clock operates in **real-time mode**, synchronizing with the system clock.
52/// It provides globally unique, strictly increasing timestamps across threads.
53pub static ATOMIC_CLOCK_REALTIME: OnceLock<AtomicTime> = OnceLock::new();
54
55/// Global atomic time in **static mode** for use across the system.
56///
57/// This clock operates in **static mode**, where the time value can be set or incremented
58/// manually. Useful for backtesting or simulated time control.
59pub static ATOMIC_CLOCK_STATIC: OnceLock<AtomicTime> = OnceLock::new();
60
61/// Returns a static reference to the global atomic clock in **real-time mode**.
62///
63/// This clock uses [`AtomicTime::time_since_epoch`] under the hood, ensuring strictly increasing
64/// timestamps across threads.
65pub fn get_atomic_clock_realtime() -> &'static AtomicTime {
66 ATOMIC_CLOCK_REALTIME.get_or_init(AtomicTime::default)
67}
68
69/// Returns a static reference to the global atomic clock in **static mode**.
70///
71/// This clock allows manual time control via [`AtomicTime::set_time`] or [`AtomicTime::increment_time`],
72/// and does not automatically sync with system time.
73pub fn get_atomic_clock_static() -> &'static AtomicTime {
74 ATOMIC_CLOCK_STATIC.get_or_init(|| AtomicTime::new(false, UnixNanos::default()))
75}
76
77/// Returns the duration since the UNIX epoch based on [`SystemTime::now()`].
78///
79/// # Panics
80///
81/// Panics if the system time is set before the UNIX epoch.
82#[inline(always)]
83#[must_use]
84pub fn duration_since_unix_epoch() -> Duration {
85 // The expect() is acceptable here because:
86 // - SystemTime failure indicates catastrophic system clock issues
87 // - This would affect the entire application's ability to function
88 // - Alternative error handling would complicate all time-dependent code paths
89 // - Such failures are extremely rare in practice and indicate hardware/OS problems
90 wall_clock_now()
91 .duration_since(UNIX_EPOCH)
92 .expect("Error calling `SystemTime`")
93}
94
95/// Returns the current wall-clock time as [`SystemTime`].
96///
97/// Under simulation (`simulation` + `cfg(madsim)`), returns virtual wall-clock
98/// time from the madsim deterministic scheduler when called from inside a
99/// madsim runtime. When called outside a runtime (e.g. plain `#[rstest]` test
100/// bodies), falls back to [`SystemTime::now()`], which under `cfg(madsim)` is
101/// libc-intercepted by madsim and resolves to the same real syscall it would
102/// in a normal build. Under normal builds, returns [`SystemTime::now()`].
103///
104/// This is the wall-clock seam. It preserves Unix-epoch semantics (unlike
105/// `tokio::time::Instant` which is monotonic and carries no epoch).
106#[inline(always)]
107#[must_use]
108fn wall_clock_now() -> SystemTime {
109 #[cfg(not(all(feature = "simulation", madsim)))]
110 {
111 SystemTime::now()
112 }
113 #[cfg(all(feature = "simulation", madsim))]
114 {
115 // `try_current` returns `None` when no madsim runtime is active.
116 // Falling back to `SystemTime::now()` matches what madsim's own libc
117 // shim does for `clock_gettime` outside a runtime; production paths
118 // running under simulation are always inside a runtime, so they
119 // continue to receive virtual time.
120 match madsim::time::TimeHandle::try_current() {
121 Some(handle) => handle.now_time(),
122 None => SystemTime::now(),
123 }
124 }
125}
126
127/// Returns the current UNIX time in nanoseconds, based on [`SystemTime::now()`].
128///
129/// # Panics
130///
131/// Panics if the duration in nanoseconds exceeds `u64::MAX`.
132#[inline(always)]
133#[must_use]
134pub fn nanos_since_unix_epoch() -> u64 {
135 u64::try_from(duration_since_unix_epoch().as_nanos())
136 .expect("System time overflow: value exceeds u64::MAX nanoseconds")
137}
138
139/// Represents an atomic timekeeping structure.
140///
141/// [`AtomicTime`] can act as a real-time clock or static clock based on its mode.
142/// It uses an [`AtomicU64`] to atomically update the value using only immutable
143/// references.
144///
145/// The `realtime` flag indicates which mode the clock is currently in.
146/// For concurrency, this struct uses atomic operations with appropriate memory orderings:
147/// - **Acquire/Release** for reading/writing in **static mode**.
148/// - **Compare-and-exchange (`AcqRel`)** in real-time mode to guarantee monotonic increments.
149///
150/// The mode flag and timestamp are private so every update flows through the methods
151/// that uphold the monotonicity and mode invariants.
152#[repr(C)]
153#[derive(Debug)]
154pub struct AtomicTime {
155 realtime: AtomicBool,
156 timestamp_ns: AtomicU64,
157}
158
159impl Default for AtomicTime {
160 /// Creates a new default [`AtomicTime`] instance in **real-time mode**, starting at the current system time.
161 fn default() -> Self {
162 Self::new(true, UnixNanos::default())
163 }
164}
165
166impl AtomicTime {
167 /// Creates a new [`AtomicTime`] instance.
168 ///
169 /// - If `realtime` is `true`, the provided `time` is ignored and the first read starts from
170 /// the current system time.
171 /// - If `realtime` is `false`, this clock starts in **static mode**, with the given `time`
172 /// as its current value.
173 #[must_use]
174 pub fn new(realtime: bool, time: UnixNanos) -> Self {
175 let timestamp_ns = if realtime { 0 } else { time.into() };
176
177 Self {
178 realtime: AtomicBool::new(realtime),
179 timestamp_ns: AtomicU64::new(timestamp_ns),
180 }
181 }
182
183 /// Returns the current time in nanoseconds, based on the clock's mode.
184 ///
185 /// - In **real-time mode**, calls [`AtomicTime::time_since_epoch`], ensuring strictly increasing
186 /// timestamps across threads, using `AcqRel` semantics for the underlying atomic.
187 /// - In **static mode**, reads the stored time using [`Ordering::Acquire`]. Updates by other
188 /// threads using [`AtomicTime::set_time`] or [`AtomicTime::increment_time`] (Release/AcqRel)
189 /// will be visible here.
190 ///
191 /// # Thread Safety
192 ///
193 /// The mode check is not atomic with the subsequent read/update. If another thread
194 /// switches modes between the check and the operation, one stale-mode result may be
195 /// returned. This is intentional: mode switching is a setup-time operation and should
196 /// not occur concurrently with time operations.
197 #[must_use]
198 pub fn get_time_ns(&self) -> UnixNanos {
199 if self.realtime.load(Ordering::Acquire) {
200 self.time_since_epoch()
201 } else {
202 UnixNanos::from(self.timestamp_ns.load(Ordering::Acquire))
203 }
204 }
205
206 /// Returns the current time as microseconds.
207 #[must_use]
208 pub fn get_time_us(&self) -> u64 {
209 self.get_time_ns().as_u64() / NANOSECONDS_IN_MICROSECOND
210 }
211
212 /// Returns the current time as milliseconds.
213 #[must_use]
214 pub fn get_time_ms(&self) -> u64 {
215 self.get_time_ns().as_u64() / NANOSECONDS_IN_MILLISECOND
216 }
217
218 /// Returns the current time as seconds.
219 #[must_use]
220 #[expect(
221 clippy::cast_precision_loss,
222 reason = "Precision loss acceptable for time conversion"
223 )]
224 pub fn get_time(&self) -> f64 {
225 self.get_time_ns().as_f64() / (NANOSECONDS_IN_SECOND as f64)
226 }
227
228 /// Manually sets a new time for the clock (only possible in **static mode**).
229 ///
230 /// This uses an atomic store with [`Ordering::Release`], so any thread reading with
231 /// [`Ordering::Acquire`] will see the updated time. This does *not* enforce a total ordering
232 /// among all threads, but is enough to ensure that once a thread sees this update, it also
233 /// sees all writes made before this call in the writing thread.
234 ///
235 /// Typically used in single-threaded scenarios or coordinated concurrency in **static mode**,
236 /// since there's no global ordering across threads.
237 ///
238 /// # Panics
239 ///
240 /// Panics if invoked when in real-time mode.
241 ///
242 /// # Thread Safety
243 ///
244 /// The mode check is not atomic with the subsequent store. If another thread calls
245 /// `make_realtime()` between the check and store, the invariant can be violated.
246 /// This is intentional: mode switching is a setup-time operation and should not
247 /// occur concurrently with time operations. Callers must ensure mode switches are
248 /// complete before resuming time operations.
249 pub fn set_time(&self, time: UnixNanos) {
250 assert!(
251 !self.realtime.load(Ordering::SeqCst),
252 "Cannot set time while clock is in realtime mode"
253 );
254
255 self.timestamp_ns.store(time.into(), Ordering::Release);
256
257 debug_assert!(
258 !self.realtime.load(Ordering::SeqCst),
259 "Invariant: clock must remain in static mode across `set_time`"
260 );
261 }
262
263 /// Increments the current static-mode time by `delta` and returns the updated value.
264 ///
265 /// Internally this uses [`AtomicU64::try_update`] with [`Ordering::AcqRel`] to ensure the increment is
266 /// atomic and visible to readers using `Acquire` loads.
267 ///
268 /// # Errors
269 ///
270 /// Returns an error if the increment would overflow `u64::MAX` or if called
271 /// while the clock is in real-time mode.
272 ///
273 /// # Thread Safety
274 ///
275 /// The mode check is not atomic with the subsequent update. If another thread calls
276 /// `make_realtime()` between the check and update, the invariant can be violated.
277 /// This is intentional: mode switching is a setup-time operation and should not
278 /// occur concurrently with time operations. Callers must ensure mode switches are
279 /// complete before resuming time operations.
280 pub fn increment_time(&self, delta: DurationNanos) -> anyhow::Result<UnixNanos> {
281 anyhow::ensure!(
282 !self.realtime.load(Ordering::SeqCst),
283 "Cannot increment time while clock is in realtime mode"
284 );
285
286 let previous =
287 match self
288 .timestamp_ns
289 .try_update(Ordering::AcqRel, Ordering::Acquire, |current| {
290 current.checked_add(delta.as_u64())
291 }) {
292 Ok(prev) => prev,
293 Err(_) => anyhow::bail!("Cannot increment time beyond u64::MAX"),
294 };
295
296 debug_assert!(
297 !self.realtime.load(Ordering::SeqCst),
298 "Invariant: clock must remain in static mode across `increment_time`"
299 );
300
301 Ok(UnixNanos::from(previous) + delta)
302 }
303
304 /// Retrieves and updates the current "real-time" clock, returning a strictly increasing
305 /// timestamp based on system time.
306 ///
307 /// Internally:
308 /// - We fetch `now` from [`SystemTime::now()`].
309 /// - We do an atomic compare-and-exchange (using [`Ordering::AcqRel`]) to ensure the stored
310 /// timestamp is never less than the last timestamp.
311 ///
312 /// This ensures:
313 /// 1. **Monotonic increments**: The returned timestamp is strictly greater than the previous
314 /// one (by at least 1 nanosecond).
315 /// 2. **No backward jumps**: If the OS time moves backward, we ignore that shift to preserve
316 /// monotonicity.
317 /// 3. **Visibility**: In a multi-threaded environment, other threads see the updated value
318 /// once this compare-and-exchange completes.
319 ///
320 /// # Panics
321 ///
322 /// Panics if the internal counter has reached `u64::MAX`, which would indicate the process has
323 /// been running for longer than the representable range (~584 years) *or* the clock was
324 /// manually corrupted.
325 pub fn time_since_epoch(&self) -> UnixNanos {
326 // This method guarantees strict consistency but may incur a performance cost under
327 // high contention due to retries in the `compare_exchange` loop.
328 let now = nanos_since_unix_epoch();
329
330 loop {
331 // Acquire to observe the latest stored value
332 let last = self.timestamp_ns.load(Ordering::Acquire);
333
334 // Ensure we never wrap past u64::MAX - treat that as a fatal error
335 let incremented = last
336 .checked_add(1)
337 .expect("AtomicTime overflow: reached u64::MAX");
338 let next = now.max(incremented);
339
340 // AcqRel on success ensures this new value is published,
341 // Acquire on failure reloads if we lost a CAS race.
342 //
343 // Note that under heavy contention (many threads calling this in tight loops),
344 // the CAS loop may increase latency.
345 //
346 // However, in practice, the loop terminates quickly because:
347 // - System time naturally advances between iterations
348 // - Each iteration increments time by at least 1ns, preventing ABA problems
349 // - True contention requiring retry is rare in normal usage patterns
350 //
351 // The concurrent stress test (4 threads × 100k iterations) validates this approach.
352 if self
353 .timestamp_ns
354 .compare_exchange(last, next, Ordering::AcqRel, Ordering::Acquire)
355 .is_ok()
356 {
357 debug_assert!(
358 next > last,
359 "Invariant: time is strictly monotonic across CAS"
360 );
361 return UnixNanos::from(next);
362 }
363 }
364 }
365
366 /// Switches the clock to **real-time mode** (`realtime = true`).
367 ///
368 /// If transitioning from static mode, the internal counter is reset to the current
369 /// wall-clock time so that [`AtomicTime::time_since_epoch`] does not carry forward a
370 /// timestamp set during static mode (e.g. a backtest far in the future).
371 ///
372 /// Uses [`Ordering::SeqCst`] for the mode flag to ensure global ordering.
373 ///
374 /// # Thread Safety
375 ///
376 /// The mode swap and the counter reset are two separate atomic operations. A thread
377 /// reading between them can observe real-time mode with the stale static-mode counter
378 /// and return a timestamp derived from it (potentially far in the future), after which
379 /// the reset moves the clock backwards. Mode switching is a setup-time operation and
380 /// must not run concurrently with time reads.
381 pub fn make_realtime(&self) {
382 if !self.realtime.swap(true, Ordering::SeqCst) {
383 self.timestamp_ns
384 .store(nanos_since_unix_epoch(), Ordering::Release);
385 }
386 }
387
388 /// Switches the clock to **static mode** (`realtime = false`).
389 ///
390 /// If transitioning from real-time mode, the internal counter is snapshotted to the
391 /// current wall-clock time so that subsequent static reads return a reasonable value
392 /// rather than a stale or zero placeholder.
393 ///
394 /// Uses [`Ordering::SeqCst`] for the mode flag to ensure global ordering.
395 ///
396 /// # Thread Safety
397 ///
398 /// The mode swap and the counter snapshot are two separate atomic operations; see
399 /// [`AtomicTime::make_realtime`] for the race this implies. Mode switching is a
400 /// setup-time operation and must not run concurrently with time reads.
401 pub fn make_static(&self) {
402 if self.realtime.swap(false, Ordering::SeqCst) {
403 self.timestamp_ns
404 .store(nanos_since_unix_epoch(), Ordering::Release);
405 }
406 }
407}
408
409#[cfg(test)]
410mod tests {
411 use std::sync::Arc;
412
413 use rstest::*;
414
415 use super::*;
416 use crate::DurationNanos;
417
418 #[rstest]
419 fn test_global_clocks_initialization() {
420 let realtime_clock = get_atomic_clock_realtime();
421 assert!(realtime_clock.get_time_ns().as_u64() > 0);
422
423 let static_clock = get_atomic_clock_static();
424 static_clock.set_time(UnixNanos::from(500_000_000)); // 500 ms
425 assert_eq!(static_clock.get_time_ns().as_u64(), 500_000_000);
426 }
427
428 #[rstest]
429 fn test_mode_switching() {
430 let time = AtomicTime::new(true, UnixNanos::default());
431
432 // Verify real-time mode
433 let realtime_ns = time.get_time_ns();
434 assert!(realtime_ns.as_u64() > 0);
435
436 // Switch to static mode
437 time.make_static();
438 time.set_time(UnixNanos::from(1_000_000_000)); // 1 second
439 let static_ns = time.get_time_ns();
440 assert_eq!(static_ns.as_u64(), 1_000_000_000);
441
442 // Switch back to real-time mode
443 time.make_realtime();
444 let new_realtime_ns = time.get_time_ns();
445 assert!(new_realtime_ns.as_u64() > static_ns.as_u64());
446 }
447
448 #[rstest]
449 #[should_panic(expected = "Cannot set time while clock is in realtime mode")]
450 fn test_set_time_panics_in_realtime_mode() {
451 let clock = AtomicTime::new(true, UnixNanos::default());
452 clock.set_time(UnixNanos::from(123));
453 }
454
455 #[rstest]
456 fn test_increment_time_returns_error_in_realtime_mode() {
457 let clock = AtomicTime::new(true, UnixNanos::default());
458 let result = clock.increment_time(DurationNanos::new(1));
459 assert!(result.is_err());
460 assert!(
461 result
462 .unwrap_err()
463 .to_string()
464 .contains("Cannot increment time while clock is in realtime mode")
465 );
466 }
467
468 #[rstest]
469 #[should_panic(expected = "AtomicTime overflow")]
470 fn test_time_since_epoch_overflow_panics() {
471 use std::sync::atomic::{AtomicBool, AtomicU64};
472
473 // Manually construct a clock with the counter already at u64::MAX
474 let clock = AtomicTime {
475 realtime: AtomicBool::new(true),
476 timestamp_ns: AtomicU64::new(u64::MAX),
477 };
478
479 // This call will attempt to add 1 and must panic
480 let _ = clock.time_since_epoch();
481 }
482
483 #[rstest]
484 fn test_new_realtime_ignores_initial_time() {
485 let before = nanos_since_unix_epoch();
486 let clock = AtomicTime::new(true, UnixNanos::from(u64::MAX));
487 let timestamp = clock.get_time_ns().as_u64();
488 let after = nanos_since_unix_epoch();
489
490 assert!(timestamp >= before);
491 assert!(timestamp <= after);
492 }
493
494 #[rstest]
495 fn test_make_static_snapshots_wall_time() {
496 // A fresh realtime clock that has never been read starts with timestamp_ns = 0.
497 // Switching to static should snapshot wall time, not leave it at 0.
498 let clock = AtomicTime::new(true, UnixNanos::default());
499 clock.make_static();
500 let ts = clock.get_time_ns();
501 assert!(
502 ts.as_u64() > 1_650_000_000_000_000_000,
503 "Expected wall-clock snapshot, was {ts}"
504 );
505 }
506
507 #[rstest]
508 fn test_make_realtime_resets_future_timestamp() {
509 // If static mode set the clock into the future, switching to realtime
510 // should reset to wall time so timestamps are not poisoned.
511 let clock = AtomicTime::new(false, UnixNanos::from(u64::MAX - 1_000));
512 clock.make_realtime();
513 let ts = clock.get_time_ns();
514 // Should be near current wall time, not near u64::MAX
515 let now = nanos_since_unix_epoch();
516 assert!(
517 ts.as_u64() <= now + 1_000_000_000, // within 1 second
518 "Expected wall-clock time, was {ts} (now={now})"
519 );
520 }
521
522 #[rstest]
523 fn test_make_static_idempotent() {
524 // Calling make_static on an already-static clock should not change the time
525 let clock = AtomicTime::new(false, UnixNanos::from(42));
526 clock.make_static();
527 assert_eq!(clock.get_time_ns(), UnixNanos::from(42));
528 }
529
530 #[rstest]
531 fn test_make_realtime_idempotent() {
532 // Calling make_realtime on an already-realtime clock should not reset the counter
533 let clock = AtomicTime::new(true, UnixNanos::default());
534 let ts1 = clock.get_time_ns();
535 clock.make_realtime(); // already realtime, should be a no-op
536 let ts2 = clock.get_time_ns();
537 assert!(ts2 >= ts1);
538 }
539
540 #[rstest]
541 fn test_static_time_is_stable() {
542 // Create a clock in static mode with an initial value
543 let clock = AtomicTime::new(false, UnixNanos::from(42));
544 let time1 = clock.get_time_ns();
545
546 // Sleep a bit to give the system time to change, if the clock were using real-time
547 std::thread::sleep(std::time::Duration::from_millis(10));
548 let time2 = clock.get_time_ns();
549
550 // In static mode, the value should remain unchanged
551 assert_eq!(time1, time2);
552 }
553
554 #[rstest]
555 fn test_increment_time() {
556 // Start in static mode
557 let time = AtomicTime::new(false, UnixNanos::from(0));
558
559 let updated_time = time.increment_time(DurationNanos::new(500)).unwrap();
560 assert_eq!(updated_time.as_u64(), 500);
561
562 let updated_time = time.increment_time(DurationNanos::new(1_000)).unwrap();
563 assert_eq!(updated_time.as_u64(), 1_500);
564 }
565
566 #[rstest]
567 fn test_increment_time_overflow_errors() {
568 let time = AtomicTime::new(false, UnixNanos::from(u64::MAX - 5));
569
570 let err = time.increment_time(DurationNanos::new(10)).unwrap_err();
571 assert_eq!(err.to_string(), "Cannot increment time beyond u64::MAX");
572 }
573
574 #[rstest]
575 fn test_increment_time_after_make_static() {
576 // Switching from realtime snapshots wall time; increments build on the snapshot
577 let clock = AtomicTime::new(true, UnixNanos::default());
578 clock.make_static();
579 let before = clock.get_time_ns();
580 let after = clock.increment_time(DurationNanos::new(1_000)).unwrap();
581 assert_eq!(after, before + DurationNanos::new(1_000));
582 assert_eq!(clock.get_time_ns(), after);
583 }
584
585 #[rstest]
586 fn test_nanos_since_unix_epoch_vs_system_time() {
587 let unix_nanos = nanos_since_unix_epoch();
588 let system_ns = u64::try_from(duration_since_unix_epoch().as_nanos()).unwrap();
589 assert!(unix_nanos.abs_diff(system_ns) < NANOSECONDS_IN_SECOND);
590 }
591
592 #[rstest]
593 fn test_time_since_epoch_monotonicity() {
594 let clock = get_atomic_clock_realtime();
595 let mut previous = clock.time_since_epoch();
596 for _ in 0..1_000_000 {
597 let current = clock.time_since_epoch();
598 assert!(current > previous);
599 previous = current;
600 }
601 }
602
603 #[rstest]
604 fn test_time_since_epoch_strictly_increasing_concurrent() {
605 let time = Arc::new(AtomicTime::new(true, UnixNanos::default()));
606 let num_threads = 4;
607 let iterations = 100_000;
608 let mut handles = Vec::with_capacity(num_threads);
609
610 for thread_id in 0..num_threads {
611 let time_clone = Arc::clone(&time);
612
613 let handle = std::thread::spawn(move || {
614 let mut previous = time_clone.time_since_epoch().as_u64();
615
616 for i in 0..iterations {
617 let current = time_clone.time_since_epoch().as_u64();
618 assert!(
619 current > previous,
620 "Thread {thread_id}: iteration {i}: time did not increase: previous={previous}, current={current}",
621 );
622 previous = current;
623 }
624 });
625
626 handles.push(handle);
627 }
628
629 for handle in handles {
630 handle.join().unwrap();
631 }
632 }
633
634 #[rstest]
635 fn test_duration_since_unix_epoch() {
636 let time = AtomicTime::new(true, UnixNanos::default());
637 let duration = Duration::from_nanos(time.get_time_ns().into());
638 let now = SystemTime::now();
639
640 // Check if the duration is close to the actual difference between now and UNIX_EPOCH
641 let delta = now
642 .duration_since(UNIX_EPOCH)
643 .unwrap()
644 .checked_sub(duration);
645 assert!(delta.unwrap_or_default() < Duration::from_millis(100));
646
647 // Check if the duration is greater than a certain value (assuming the test is run after that point)
648 assert!(duration > Duration::from_mins(27_500_000));
649 }
650
651 #[rstest]
652 fn test_unix_timestamp_is_monotonic_increasing() {
653 let time = AtomicTime::new(true, UnixNanos::default());
654 let result1 = time.get_time();
655 let result2 = time.get_time();
656 let result3 = time.get_time();
657 let result4 = time.get_time();
658 let result5 = time.get_time();
659
660 assert!(result2 >= result1);
661 assert!(result3 >= result2);
662 assert!(result4 >= result3);
663 assert!(result5 >= result4);
664 assert!(result1 > 1_650_000_000.0);
665 }
666
667 #[rstest]
668 fn test_unix_timestamp_ms_is_monotonic_increasing() {
669 let time = AtomicTime::new(true, UnixNanos::default());
670 let result1 = time.get_time_ms();
671 let result2 = time.get_time_ms();
672 let result3 = time.get_time_ms();
673 let result4 = time.get_time_ms();
674 let result5 = time.get_time_ms();
675
676 assert!(result2 >= result1);
677 assert!(result3 >= result2);
678 assert!(result4 >= result3);
679 assert!(result5 >= result4);
680 assert!(result1 >= 1_650_000_000_000);
681 }
682
683 #[rstest]
684 fn test_unix_timestamp_us_is_monotonic_increasing() {
685 let time = AtomicTime::new(true, UnixNanos::default());
686 let result1 = time.get_time_us();
687 let result2 = time.get_time_us();
688 let result3 = time.get_time_us();
689 let result4 = time.get_time_us();
690 let result5 = time.get_time_us();
691
692 assert!(result2 >= result1);
693 assert!(result3 >= result2);
694 assert!(result4 >= result3);
695 assert!(result5 >= result4);
696 assert!(result1 > 1_650_000_000_000_000);
697 }
698
699 #[rstest]
700 fn test_unix_timestamp_ns_is_monotonic_increasing() {
701 let time = AtomicTime::new(true, UnixNanos::default());
702 let result1 = time.get_time_ns();
703 let result2 = time.get_time_ns();
704 let result3 = time.get_time_ns();
705 let result4 = time.get_time_ns();
706 let result5 = time.get_time_ns();
707
708 assert!(result2 >= result1);
709 assert!(result3 >= result2);
710 assert!(result4 >= result3);
711 assert!(result5 >= result4);
712 assert!(result1.as_u64() > 1_650_000_000_000_000_000);
713 }
714
715 #[rstest]
716 fn test_acquire_release_contract_static_mode() {
717 // This test explicitly proves the Acquire/Release memory ordering contract:
718 // - Writer thread uses set_time() which does Release store (see AtomicTime::set_time)
719 // - Reader thread uses get_time_ns() which does Acquire load (see AtomicTime::get_time_ns)
720 // - The Release-Acquire pair ensures all writes before Release are visible after Acquire
721
722 let clock = Arc::new(AtomicTime::new(false, UnixNanos::from(0)));
723 let aux_data = Arc::new(AtomicU64::new(0));
724 let done = Arc::new(AtomicBool::new(false));
725
726 // Writer thread: updates auxiliary data, then releases via set_time
727 let writer_clock = Arc::clone(&clock);
728 let writer_aux = Arc::clone(&aux_data);
729 let writer_done = Arc::clone(&done);
730
731 let writer = std::thread::spawn(move || {
732 for i in 1..=1_000u64 {
733 writer_aux.store(i, Ordering::Relaxed);
734
735 // Release store via set_time creates a release fence - all prior writes (including aux_data)
736 // must be visible to any thread that observes this time value via Acquire load
737 writer_clock.set_time(UnixNanos::from(i * 1000));
738
739 // Yield to encourage interleaving
740 std::thread::yield_now();
741 }
742 writer_done.store(true, Ordering::Release);
743 });
744
745 // Reader thread: acquires via get_time_ns, then checks auxiliary data
746 let reader_clock = Arc::clone(&clock);
747 let reader_aux = Arc::clone(&aux_data);
748 let reader_done = Arc::clone(&done);
749
750 let reader = std::thread::spawn(move || {
751 let mut last_time = 0u64;
752 let mut max_aux_seen = 0u64;
753
754 // Poll until writer is done, with no iteration limit
755 while !reader_done.load(Ordering::Acquire) {
756 let current_time = reader_clock.get_time_ns().as_u64();
757
758 if current_time > last_time {
759 // The Acquire in get_time_ns synchronizes with the Release in set_time,
760 // making aux_data visible
761 let aux_value = reader_aux.load(Ordering::Relaxed);
762
763 // Invariant: aux_value must never go backwards (proves Release-Acquire sync works)
764 if aux_value > 0 {
765 assert!(
766 aux_value >= max_aux_seen,
767 "Acquire/Release contract violated: aux went backwards from {max_aux_seen} to {aux_value}"
768 );
769 max_aux_seen = aux_value;
770 }
771
772 last_time = current_time;
773 }
774
775 std::thread::yield_now();
776 }
777
778 // Check final state after writer completes to ensure we observe updates
779 let final_time = reader_clock.get_time_ns().as_u64();
780 if final_time > last_time {
781 let final_aux = reader_aux.load(Ordering::Relaxed);
782 if final_aux > 0 {
783 assert!(
784 final_aux >= max_aux_seen,
785 "Acquire/Release contract violated: final aux {final_aux} < max {max_aux_seen}"
786 );
787 max_aux_seen = final_aux;
788 }
789 }
790
791 max_aux_seen
792 });
793
794 writer.join().unwrap();
795 let max_observed = reader.join().unwrap();
796
797 // Ensure the reader actually observed updates (not vacuously satisfied)
798 assert!(max_observed > 0, "Reader must observe writer updates");
799 }
800
801 #[rstest]
802 fn test_acquire_release_contract_increment_time() {
803 // Similar test for increment_time, which uses try_update with AcqRel (see AtomicTime::increment_time)
804
805 let clock = Arc::new(AtomicTime::new(false, UnixNanos::from(0)));
806 let aux_data = Arc::new(AtomicU64::new(0));
807 let done = Arc::new(AtomicBool::new(false));
808
809 let writer_clock = Arc::clone(&clock);
810 let writer_aux = Arc::clone(&aux_data);
811 let writer_done = Arc::clone(&done);
812
813 let writer = std::thread::spawn(move || {
814 for i in 1..=1_000u64 {
815 writer_aux.store(i, Ordering::Relaxed);
816 let _ = writer_clock
817 .increment_time(DurationNanos::new(1000))
818 .unwrap();
819 std::thread::yield_now();
820 }
821 writer_done.store(true, Ordering::Release);
822 });
823
824 let reader_clock = Arc::clone(&clock);
825 let reader_aux = Arc::clone(&aux_data);
826 let reader_done = Arc::clone(&done);
827
828 let reader = std::thread::spawn(move || {
829 let mut last_time = 0u64;
830 let mut max_aux = 0u64;
831
832 // Poll until writer is done, with no iteration limit
833 while !reader_done.load(Ordering::Acquire) {
834 let current_time = reader_clock.get_time_ns().as_u64();
835
836 if current_time > last_time {
837 let aux_value = reader_aux.load(Ordering::Relaxed);
838
839 // Invariant: aux_value must never regress (proves AcqRel sync works)
840 if aux_value > 0 {
841 assert!(
842 aux_value >= max_aux,
843 "AcqRel contract violated: aux regressed from {max_aux} to {aux_value}"
844 );
845 max_aux = aux_value;
846 }
847
848 last_time = current_time;
849 }
850
851 std::thread::yield_now();
852 }
853
854 // Check final state after writer completes to ensure we observe updates
855 let final_time = reader_clock.get_time_ns().as_u64();
856 if final_time > last_time {
857 let final_aux = reader_aux.load(Ordering::Relaxed);
858 if final_aux > 0 {
859 assert!(
860 final_aux >= max_aux,
861 "AcqRel contract violated: final aux {final_aux} < max {max_aux}"
862 );
863 max_aux = final_aux;
864 }
865 }
866
867 max_aux
868 });
869
870 writer.join().unwrap();
871 let max_observed = reader.join().unwrap();
872
873 // Ensure the reader actually observed updates (not vacuously satisfied)
874 assert!(max_observed > 0, "Reader must observe writer updates");
875 }
876
877 // The wall-clock seam (`wall_clock_now`) routes through madsim's virtual
878 // clock under simulation. Sleeping for 60 virtual seconds must advance
879 // the value returned by `nanos_since_unix_epoch` by 60s in wall-clock
880 // terms. If the cfg gate fell through to `SystemTime::now()`, the elapsed
881 // value would only reflect real wall-clock time (~0ms) and the assertion
882 // would fail.
883 #[cfg(all(feature = "simulation", madsim))]
884 #[madsim::test]
885 async fn test_wall_clock_advances_with_virtual_time() {
886 let before = nanos_since_unix_epoch();
887 madsim::time::sleep(std::time::Duration::from_mins(1)).await;
888 let after = nanos_since_unix_epoch();
889
890 let elapsed_ns = after.saturating_sub(before);
891 let sixty_seconds_ns = 60 * NANOSECONDS_IN_SECOND;
892 assert!(
893 elapsed_ns >= sixty_seconds_ns,
894 "wall clock did not advance by full virtual sleep: elapsed={elapsed_ns}ns"
895 );
896 }
897}