reifydb_testing/util/
wait.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::time::Instant;
5
6use reifydb_value::value::duration::Duration;
7use tokio::time::sleep;
8
9pub fn default_timeout() -> Duration {
10	Duration::from_seconds(5).unwrap()
11}
12
13pub fn default_poll_interval() -> Duration {
14	Duration::from_milliseconds(1).unwrap()
15}
16
17pub async fn wait_for_condition<F>(condition: F, timeout: Duration, poll_interval: Duration, timeout_message: &str)
18where
19	F: Fn() -> bool,
20{
21	#[allow(clippy::disallowed_methods)]
22	let start = Instant::now();
23	let mut poll_count = 0u64;
24
25	while !condition() {
26		if start.elapsed() > timeout.to_std() {
27			println!(
28				"[DEBUG:await] TIMEOUT elapsed={:.1}s polls={poll_count} msg={timeout_message}",
29				start.elapsed().as_secs_f64()
30			);
31			panic!("Timeout after {:?}: {}", timeout, timeout_message);
32		}
33		poll_count += 1;
34		if poll_count.is_multiple_of(1000) {
35			println!(
36				"[DEBUG:await] poll #{poll_count} elapsed={:.1}s msg={timeout_message}",
37				start.elapsed().as_secs_f64()
38			);
39		}
40		sleep(poll_interval.to_std()).await;
41	}
42	println!(
43		"[DEBUG:await] condition met after {poll_count} polls elapsed={:.3}s msg={timeout_message}",
44		start.elapsed().as_secs_f64()
45	);
46}
47
48pub async fn wait_for<F>(condition: F, message: &str)
49where
50	F: Fn() -> bool,
51{
52	wait_for_condition(condition, default_timeout(), default_poll_interval(), message).await;
53}
54
55#[cfg(test)]
56pub mod tests {
57	use std::{sync::Arc, thread};
58
59	use reifydb_runtime::sync::mutex::Mutex;
60
61	use super::*;
62
63	#[tokio::test]
64	async fn test_wait_for_immediate() {
65		wait_for(|| true, "Should not timeout").await;
66	}
67
68	#[tokio::test]
69	async fn test_wait_for_becomes_true() {
70		let counter = Arc::new(Mutex::new(0));
71		let counter_clone = counter.clone();
72
73		thread::spawn(move || {
74			thread::sleep(Duration::from_milliseconds(50).unwrap().to_std());
75			*counter_clone.lock() = 5;
76		});
77
78		wait_for(|| *counter.lock() == 5, "Counter should reach 5").await;
79
80		assert_eq!(*counter.lock(), 5);
81	}
82
83	#[tokio::test]
84	#[should_panic(expected = "Timeout after")]
85	async fn test_wait_for_timeout() {
86		wait_for_condition(
87			|| false,
88			Duration::from_milliseconds(10).unwrap(),
89			Duration::from_milliseconds(1).unwrap(),
90			"Condition never becomes true",
91		)
92		.await;
93	}
94}
reifydb_testing/util/wait.rs

reifydb_testing/util/
wait.rs
