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        let join = std::thread::spawn(move || {
25            // The receiver may already be gone (test dropped early); that's fine.
26            let _ = tx.send(run());
27        });
28        Self { rx, join }
29    }
30
31    /// Block until the pipeline exits or `timeout` elapses; `None` on timeout.
32    ///
33    /// Use this to wait for a pipeline that stops on its own (e.g. a
34    /// `Fail`-policy fatal) without a manual shutdown trigger.
35    pub fn wait_exit(&self, timeout: Duration) -> Option<Result<ExitReport, StartError>> {
36        self.rx.recv_timeout(timeout).ok()
37    }
38
39    /// Block until the pipeline exits and join its thread, returning the run
40    /// result. Panics if the thread panicked.
41    pub fn join(self) -> Result<ExitReport, StartError> {
42        let report = self.rx.recv().expect("pipeline thread dropped its result");
43        self.join.join().expect("pipeline thread panicked");
44        report
45    }
46}
47
48/// How often [`wait_until`] re-checks its predicate. Small enough that the
49/// cadence is not itself the latency a test measures, coarse enough not to spin
50/// a core.
51const POLL_INTERVAL: Duration = Duration::from_millis(5);
52
53/// Poll `check` until it returns `true` or `timeout` elapses; panics with `what`
54/// on timeout.
55pub fn wait_until(timeout: Duration, what: &str, mut check: impl FnMut() -> bool) {
56    let deadline = Instant::now() + timeout;
57    while Instant::now() < deadline {
58        if check() {
59            return;
60        }
61        std::thread::sleep(POLL_INTERVAL);
62    }
63    panic!("timed out after {timeout:?} waiting for: {what}");
64}
spate_test/run.rs

spate_test/
run.rs
