spate_test/
run.rs

1//! Blocking test-wait helpers: bounded pipeline-exit waits and a generic
2//! condition poll, so whole-pipeline tests don't hand-roll
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`] instead of
11/// polling [`JoinHandle::is_finished`].
12///
13/// [`recv_timeout`]: crossbeam_channel::Receiver::recv_timeout
14#[derive(Debug)]
15pub struct PipelineRun {
16    rx: crossbeam_channel::Receiver<Result<ExitReport, StartError>>,
17    join: JoinHandle<()>,
18}
19
20impl PipelineRun {
21    /// Spawn `run` (typically `move || runtime.run()`) on a new thread; its
22    /// result is sent over the channel when it returns.
23    pub fn spawn(run: impl FnOnce() -> Result<ExitReport, StartError> + Send + 'static) -> Self {
24        let (tx, rx) = crossbeam_channel::bounded(1);
25        let join = std::thread::spawn(move || {
26            // The receiver may already be gone (test dropped early); that's fine.
27            let _ = tx.send(run());
28        });
29        Self { rx, join }
30    }
31
32    /// Block until the pipeline exits or `timeout` elapses; `None` on timeout.
33    ///
34    /// Use this to wait for a pipeline that stops on its own (e.g. a
35    /// `Fail`-policy fatal) without a manual shutdown trigger.
36    pub fn wait_exit(&self, timeout: Duration) -> Option<Result<ExitReport, StartError>> {
37        self.rx.recv_timeout(timeout).ok()
38    }
39
40    /// Block until the pipeline exits and join its thread, returning the run
41    /// result. Panics if the thread panicked.
42    pub fn join(self) -> Result<ExitReport, StartError> {
43        let report = self.rx.recv().expect("pipeline thread dropped its result");
44        self.join.join().expect("pipeline thread panicked");
45        report
46    }
47}
48
49/// How often [`wait_until`] re-checks its predicate. Small enough that the
50/// cadence is not itself the latency a test measures — at the old 250ms every
51/// wait paid up to a quarter second of pure sleep, and the suites use dozens of
52/// them — but coarse enough not to spin a core.
53const POLL_INTERVAL: Duration = Duration::from_millis(5);
54
55/// Poll `check` until it returns `true` or `timeout` elapses; panics with `what`
56/// on timeout.
57pub fn wait_until(timeout: Duration, what: &str, mut check: impl FnMut() -> bool) {
58    let deadline = Instant::now() + timeout;
59    while Instant::now() < deadline {
60        if check() {
61            return;
62        }
63        std::thread::sleep(POLL_INTERVAL);
64    }
65    panic!("timed out after {timeout:?} waiting for: {what}");
66}
spate_test/run.rs

spate_test/
run.rs
