nextest_runner/runner/
internal_events.rs

1// Copyright (c) The nextest Contributors
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! Internal events used between the runner components.
5//!
6//! These events often mirror those in [`crate::reporter::events`], but are used
7//! within the runner. They'll often carry additional information that the
8//! reporter doesn't need to know about.
9
10use super::{SetupScriptPacket, TestPacket};
11use crate::{
12    config::{
13        elements::{FlakyResult, JunitFlakyFailStatus, ReportSkipPolicy},
14        scripts::{ScriptId, SetupScriptConfig},
15    },
16    errors::DisplayErrorChain,
17    list::TestInstance,
18    output_spec::LiveSpec,
19    reporter::{
20        TestOutputDisplay, UnitErrorDescription,
21        events::{
22            ChildExecutionOutputDescription, ErrorSummary, ExecuteStatus, ExecutionResult,
23            ExecutionResultDescription, InfoResponse, OutputErrorSlice, RetryData,
24            SetupScriptEnvMap, SetupScriptExecuteStatus, StressIndex, TestSlotAssignment, UnitKind,
25            UnitState,
26        },
27    },
28    signal::ShutdownEvent,
29    test_output::ChildExecutionOutput,
30    time::StopwatchSnapshot,
31};
32use nextest_metadata::MismatchReason;
33use std::time::Duration;
34use tokio::{
35    sync::{
36        mpsc::{UnboundedReceiver, UnboundedSender},
37        oneshot,
38    },
39    task::JoinError,
40};
41
42/// An internal event.
43///
44/// These events are sent by the executor (the part that actually runs
45/// executables) to the dispatcher (the part of the runner that coordinates with
46/// the external world).
47#[derive(Debug)]
48pub(super) enum ExecutorEvent<'a> {
49    SetupScriptStarted {
50        stress_index: Option<StressIndex>,
51        script_id: ScriptId,
52        config: &'a SetupScriptConfig,
53        program: String,
54        index: usize,
55        total: usize,
56        // See the note in the `Started` variant.
57        req_rx_tx: oneshot::Sender<UnboundedReceiver<RunUnitRequest<'a>>>,
58    },
59    SetupScriptSlow {
60        stress_index: Option<StressIndex>,
61        script_id: ScriptId,
62        config: &'a SetupScriptConfig,
63        program: String,
64        elapsed: Duration,
65        will_terminate: Option<Duration>,
66    },
67    SetupScriptFinished {
68        stress_index: Option<StressIndex>,
69        script_id: ScriptId,
70        config: &'a SetupScriptConfig,
71        program: String,
72        index: usize,
73        total: usize,
74        status: SetupScriptExecuteStatus<LiveSpec>,
75    },
76    Started {
77        stress_index: Option<StressIndex>,
78        test_instance: TestInstance<'a>,
79        slot_assignment: TestSlotAssignment,
80        command_line: Vec<String>,
81        // The channel over which to return the unit request.
82        //
83        // The callback context is solely responsible for coordinating the
84        // creation of all channels, such that it acts as the source of truth
85        // for which units to broadcast messages out to. This oneshot channel is
86        // used to let each test instance know to go ahead and start running
87        // tests.
88        //
89        // Why do we use unbounded channels? Mostly to make life simpler --
90        // these are low-traffic channels that we don't expect to be backed up.
91        req_rx_tx: oneshot::Sender<UnboundedReceiver<RunUnitRequest<'a>>>,
92        // The configured result for flaky tests.
93        flaky_result: FlakyResult,
94    },
95    Slow {
96        stress_index: Option<StressIndex>,
97        test_instance: TestInstance<'a>,
98        retry_data: RetryData,
99        elapsed: Duration,
100        will_terminate: Option<Duration>,
101    },
102    AttemptFailedWillRetry {
103        stress_index: Option<StressIndex>,
104        test_instance: TestInstance<'a>,
105        failure_output: TestOutputDisplay,
106        run_status: ExecuteStatus<LiveSpec>,
107        delay_before_next_attempt: Duration,
108    },
109    RetryStarted {
110        stress_index: Option<StressIndex>,
111        test_instance: TestInstance<'a>,
112        slot_assignment: TestSlotAssignment,
113        retry_data: RetryData,
114        command_line: Vec<String>,
115        // This is used to indicate that the dispatcher still wants to run the test.
116        tx: oneshot::Sender<()>,
117    },
118    Finished {
119        stress_index: Option<StressIndex>,
120        test_instance: TestInstance<'a>,
121        success_output: TestOutputDisplay,
122        failure_output: TestOutputDisplay,
123        junit_store_success_output: bool,
124        junit_store_failure_output: bool,
125        junit_flaky_fail_status: JunitFlakyFailStatus,
126        last_run_status: ExecuteStatus<LiveSpec>,
127    },
128    Skipped {
129        stress_index: Option<StressIndex>,
130        test_instance: TestInstance<'a>,
131        reason: MismatchReason,
132        junit_report_skipped: ReportSkipPolicy,
133    },
134}
135
136#[derive(Clone, Copy)]
137pub(super) enum UnitExecuteStatus<'a, 'status> {
138    Test(&'status InternalExecuteStatus<'a>),
139    SetupScript(&'status InternalSetupScriptExecuteStatus<'a>),
140}
141
142impl<'a> UnitExecuteStatus<'a, '_> {
143    pub(super) fn info_response(&self) -> InfoResponse<'a> {
144        match self {
145            Self::Test(status) => status.test.info_response(
146                UnitState::Exited {
147                    result: ExecutionResultDescription::from(status.result),
148                    time_taken: status.stopwatch_end.active,
149                    slow_after: status.slow_after,
150                },
151                status.output.clone(),
152            ),
153            Self::SetupScript(status) => status.script.info_response(
154                UnitState::Exited {
155                    result: ExecutionResultDescription::from(status.result),
156                    time_taken: status.stopwatch_end.active,
157                    slow_after: status.slow_after,
158                },
159                status.output.clone(),
160            ),
161        }
162    }
163}
164
165pub(super) struct InternalExecuteStatus<'a> {
166    pub(super) test: TestPacket<'a>,
167    pub(super) slow_after: Option<Duration>,
168    pub(super) output: ChildExecutionOutput,
169    pub(super) result: ExecutionResult,
170    pub(super) stopwatch_end: StopwatchSnapshot,
171}
172
173impl InternalExecuteStatus<'_> {
174    pub(super) fn into_external(self) -> ExecuteStatus<LiveSpec> {
175        let output: ChildExecutionOutputDescription<LiveSpec> = self.output.into();
176
177        // Compute the error summary and output error slice using
178        // UnitErrorDescription.
179        let desc = UnitErrorDescription::new(UnitKind::Test, &output);
180        let error_summary = desc.all_error_list().map(|errors| ErrorSummary {
181            short_message: errors.short_message(),
182            description: DisplayErrorChain::new(errors).to_string(),
183        });
184        let output_error_slice = desc.output_slice().map(|slice| OutputErrorSlice {
185            slice: slice.to_string(),
186            start: slice.combined_subslice().map(|s| s.start).unwrap_or(0),
187        });
188
189        ExecuteStatus {
190            retry_data: self.test.retry_data(),
191            output,
192            result: self.result.into(),
193            start_time: self.stopwatch_end.start_time.fixed_offset(),
194            time_taken: self.stopwatch_end.active,
195            is_slow: self.slow_after.is_some(),
196            delay_before_start: self.test.delay_before_start(),
197            error_summary,
198            output_error_slice,
199        }
200    }
201}
202
203pub(super) struct InternalSetupScriptExecuteStatus<'a> {
204    pub(super) script: SetupScriptPacket<'a>,
205    pub(super) slow_after: Option<Duration>,
206    pub(super) output: ChildExecutionOutput,
207    pub(super) result: ExecutionResult,
208    pub(super) stopwatch_end: StopwatchSnapshot,
209    pub(super) env_map: Option<SetupScriptEnvMap>,
210}
211
212impl InternalSetupScriptExecuteStatus<'_> {
213    pub(super) fn into_external(self) -> SetupScriptExecuteStatus<LiveSpec> {
214        let output: ChildExecutionOutputDescription<LiveSpec> = self.output.into();
215
216        // Compute the error summary using UnitErrorDescription.
217        // Setup scripts don't have output_error_slice since that's only for
218        // tests (setup scripts can fail in all kinds of ways, while tests fail
219        // in more predictable ones).
220        let desc = UnitErrorDescription::new(UnitKind::Script, &output);
221        let error_summary = desc.all_error_list().map(|errors| ErrorSummary {
222            short_message: errors.short_message(),
223            description: DisplayErrorChain::new(errors).to_string(),
224        });
225
226        SetupScriptExecuteStatus {
227            output,
228            result: self.result.into(),
229            start_time: self.stopwatch_end.start_time.fixed_offset(),
230            time_taken: self.stopwatch_end.active,
231            is_slow: self.slow_after.is_some(),
232            env_map: self.env_map,
233            error_summary,
234        }
235    }
236}
237
238/// Events sent from the dispatcher to individual unit execution tasks.
239#[derive(Clone, Debug)]
240pub(super) enum RunUnitRequest<'a> {
241    Signal(SignalRequest),
242    /// Non-signal cancellation requests (e.g. test failures) which should cause
243    /// tests to exit in some states.
244    OtherCancel,
245    Query(RunUnitQuery<'a>),
246}
247
248impl<'a> RunUnitRequest<'a> {
249    pub(super) fn drain(self, status: UnitExecuteStatus<'a, '_>) {
250        match self {
251            #[cfg(unix)]
252            Self::Signal(SignalRequest::Stop(sender)) => {
253                // The receiver being dead isn't really important.
254                let _ = sender.send(());
255            }
256            #[cfg(unix)]
257            Self::Signal(SignalRequest::Continue) => {}
258            Self::Signal(SignalRequest::Shutdown(_)) => {}
259            Self::OtherCancel => {}
260            Self::Query(RunUnitQuery::GetInfo(tx)) => {
261                // The receiver being dead isn't really important.
262                _ = tx.send(status.info_response());
263            }
264        }
265    }
266}
267
268#[derive(Clone, Debug)]
269pub(super) enum SignalRequest {
270    // The mpsc sender is used by each test to indicate that the stop signal has been sent.
271    #[cfg(unix)]
272    Stop(UnboundedSender<()>),
273    #[cfg(unix)]
274    Continue,
275    Shutdown(ShutdownRequest),
276}
277
278#[derive(Copy, Clone, Debug, Eq, PartialEq)]
279pub(super) enum ShutdownRequest {
280    Once(ShutdownEvent),
281    Twice,
282}
283
284#[derive(Clone, Debug)]
285pub(super) enum RunUnitQuery<'a> {
286    GetInfo(UnboundedSender<InfoResponse<'a>>),
287}
288
289#[derive(Clone, Copy, Debug, Eq, PartialEq)]
290pub(super) enum InternalTerminateReason {
291    Timeout,
292    Signal(ShutdownRequest),
293}
294
295pub(super) enum RunnerTaskState {
296    Finished { child_join_errors: Vec<JoinError> },
297    Cancelled,
298}
299
300impl RunnerTaskState {
301    /// Mark a runner task as finished and having not run any children.
302    pub(super) fn finished_no_children() -> Self {
303        Self::Finished {
304            child_join_errors: Vec::new(),
305        }
306    }
307}
308
309#[derive(Clone, Copy, Debug)]
310#[must_use]
311pub(super) enum HandleSignalResult {
312    /// A job control signal was delivered.
313    #[cfg(unix)]
314    JobControl,
315
316    /// The child was terminated.
317    #[cfg_attr(not(windows), expect(dead_code))]
318    Terminated(TerminateChildResult),
319}
320
321#[derive(Clone, Copy, Debug)]
322#[must_use]
323pub(super) enum TerminateChildResult {
324    /// The child process exited without being forcibly killed.
325    Exited,
326
327    /// The child process was forcibly killed.
328    Killed,
329}
nextest_runner/runner/internal_events.rs

nextest_runner/runner/
internal_events.rs
