spate_test/
run.rs

1//! Blocking test-wait helpers: bounded pipeline-exit waits and a generic
2//! condition poll. Whole-pipeline tests use these in place of hand-rolled
3//! `deadline + thread::sleep` busy-loops.
4
5use spate_core::pipeline::{ExitReport, StartError};
6use std::thread::JoinHandle;
7use std::time::{Duration, Instant};
8
9/// A pipeline running on its own OS thread, with its exit result delivered
10/// over a channel so tests can wait with a bounded [`recv_timeout`].
11///
12/// [`recv_timeout`]: crossbeam_channel::Receiver::recv_timeout
13#[derive(Debug)]
14pub struct PipelineRun {
15    rx: crossbeam_channel::Receiver<Result<ExitReport, StartError>>,
16    join: JoinHandle<()>,
17}
18
19impl PipelineRun {
20    /// Spawn `run` (typically `move || runtime.run()`) on a new thread; its
21    /// result is sent over the channel when it returns.
22    pub fn spawn(run: impl FnOnce() -> Result<ExitReport, StartError> + Send + 'static) -> Self {
23        let (tx, rx) = crossbeam_channel::bounded(1);
24        #[expect(
25            clippy::disallowed_methods,
26            reason = "the thread ends when `run` returns; `PipelineRun::join` joins it, and dropping a run never blocks"
27        )]
28        let join = std::thread::spawn(move || {
29            // The receiver may already be gone (test dropped early); that's fine.
30            let _ = tx.send(run());
31        });
32        Self { rx, join }
33    }
34
35    /// Block until the pipeline exits or `timeout` elapses; `None` on timeout.
36    ///
37    /// Use this to wait for a pipeline that stops on its own (e.g. a
38    /// `Fail`-policy fatal) without a manual shutdown trigger.
39    pub fn wait_exit(&self, timeout: Duration) -> Option<Result<ExitReport, StartError>> {
40        self.rx.recv_timeout(timeout).ok()
41    }
42
43    /// Block until the pipeline exits and join its thread, returning the run
44    /// result. Panics if the thread panicked.
45    pub fn join(self) -> Result<ExitReport, StartError> {
46        let report = self.rx.recv().expect("pipeline thread dropped its result");
47        self.join.join().expect("pipeline thread panicked");
48        report
49    }
50}
51
52/// How often [`wait_until`] re-checks its predicate. Small enough that the
53/// cadence is not itself the latency a test measures, coarse enough not to spin
54/// a core.
55const POLL_INTERVAL: Duration = Duration::from_millis(5);
56
57/// Poll `check` until it returns `true` or `timeout` elapses; panics with `what`
58/// on timeout.
59pub fn wait_until(timeout: Duration, what: &str, mut check: impl FnMut() -> bool) {
60    if poll_until(timeout, || check().then_some(())).is_none() {
61        panic!("timed out after {timeout:?} waiting for: {what}");
62    }
63}
64
65/// Poll `check` until it returns a value, or `None` once `timeout` elapses.
66pub(crate) fn poll_until<T>(timeout: Duration, mut check: impl FnMut() -> Option<T>) -> Option<T> {
67    let deadline = Instant::now() + timeout;
68    while Instant::now() < deadline {
69        if let Some(value) = check() {
70            return Some(value);
71        }
72        std::thread::sleep(POLL_INTERVAL);
73    }
74    None
75}
spate_test/run.rs

spate_test/
run.rs
