ci_engine/
exec.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Sequential local check orchestration.
3
4use std::{collections::BTreeMap, path::Path, time::Instant};
5
6use ci_config::{Check, CiConfig, Trigger};
7use crypto::{Conclusion, FailureClass};
8
9use crate::{
10    cache::{prepare_caches, restore_worktree_cache_dirs, save_caches},
11    classify::{Disposition, classify},
12    env::HermeticEnv,
13    model::{AttemptRecord, CheckResult, ExecutionContext, RunControls, RunOptions},
14    proc_group::ProcGroupRegistry,
15    process::{RunOutput, run_process},
16    result::{CompletedRun, finalize, infra_result, skipped_result},
17    result_cache::{ResultCache, ResultCacheError, SpotCheck, with_cache},
18};
19
20/// Run every check with default controls.
21pub fn run_checks(
22    config: &CiConfig,
23    context: &ExecutionContext,
24    options: &RunOptions<'_>,
25) -> Result<Vec<CheckResult>, ResultCacheError> {
26    run_checks_with(config, context, options, &RunControls::default())
27}
28
29/// Run checks with explicit trigger/cache/environment controls.
30pub fn run_checks_with(
31    config: &CiConfig,
32    context: &ExecutionContext,
33    options: &RunOptions<'_>,
34    controls: &RunControls<'_>,
35) -> Result<Vec<CheckResult>, ResultCacheError> {
36    let default_environment = HermeticEnv::new();
37    let environment = controls.hermetic_env.unwrap_or(&default_environment);
38    let default_cache_root = options.workdir.join(".hci-cache");
39    let cache_root = controls.cache_root.unwrap_or(&default_cache_root);
40    let resolved = ResolvedRun {
41        options,
42        environment,
43        cache_root,
44        proc_groups: controls.proc_groups.as_ref(),
45        result_cache: controls.result_cache,
46        spot_check: controls.spot_check,
47    };
48    let results = config
49        .checks
50        .iter()
51        .map(|check| match &controls.trigger {
52            Some(trigger) if !check_runs_for_trigger(&check.triggers, trigger) => {
53                Ok(skipped_result(check, context, &resolved))
54            }
55            _ => run_one_check(check, context, &resolved),
56        })
57        .collect::<Result<Vec<_>, _>>()?;
58    restore_worktree_cache_dirs(options.workdir, &config.checks);
59    Ok(results)
60}
61
62pub(crate) struct ResolvedRun<'a> {
63    pub(crate) options: &'a RunOptions<'a>,
64    pub(crate) environment: &'a HermeticEnv,
65    pub(crate) cache_root: &'a Path,
66    pub(crate) proc_groups: Option<&'a ProcGroupRegistry>,
67    pub(crate) result_cache: Option<&'a dyn ResultCache>,
68    pub(crate) spot_check: SpotCheck,
69}
70
71fn run_one_check(
72    check: &Check,
73    context: &ExecutionContext,
74    run: &ResolvedRun<'_>,
75) -> Result<CheckResult, ResultCacheError> {
76    let service_environment = declared_service_env(check);
77    let key_environment = run
78        .environment
79        .build(&check.env, &service_environment, &BTreeMap::new());
80    with_cache(check, context, run, &key_environment, || {
81        run_one_check_uncached(check, context, run, &service_environment)
82    })
83}
84
85fn declared_service_env(check: &Check) -> BTreeMap<String, String> {
86    check
87        .services
88        .iter()
89        .flat_map(|service| {
90            service
91                .env
92                .iter()
93                .map(|entry| (entry.0.clone(), entry.1.clone()))
94        })
95        .collect()
96}
97
98fn run_one_check_uncached(
99    check: &Check,
100    context: &ExecutionContext,
101    run: &ResolvedRun<'_>,
102    service_environment: &BTreeMap<String, String>,
103) -> CheckResult {
104    let started_at = (run.options.now_rfc3339)();
105    let started = Instant::now();
106    let caches = match prepare_caches(
107        &check.name,
108        &check.cache_paths,
109        run.options.workdir,
110        run.cache_root,
111    ) {
112        Ok(caches) => caches,
113        Err(error) => {
114            return infra_result(
115                check,
116                context,
117                run,
118                started_at,
119                started.elapsed(),
120                "cache_paths",
121                &error.to_string(),
122            );
123        }
124    };
125    let services = match run.options.services.up(&check.services) {
126        Ok(services) => services,
127        Err(error) => {
128            return infra_result(
129                check,
130                context,
131                run,
132                started_at,
133                started.elapsed(),
134                "service_provisioning",
135                &format!("service provisioning failed: {error}"),
136            );
137        }
138    };
139    let environment = run
140        .environment
141        .build(&check.env, service_environment, &caches.env);
142    let (last, attempts) = run_attempts(check, run, &environment);
143    let _ = run.options.services.down(services);
144    if let Err(error) = save_caches(&caches) {
145        return infra_result(
146            check,
147            context,
148            run,
149            started_at,
150            started.elapsed(),
151            "cache_save",
152            &error.to_string(),
153        );
154    }
155    finalize(
156        check,
157        context,
158        CompletedRun {
159            output: last,
160            attempts,
161            environment,
162            started_at,
163            finished_at: (run.options.now_rfc3339)(),
164            duration: started.elapsed(),
165        },
166    )
167}
168
169fn run_attempts(
170    check: &Check,
171    run: &ResolvedRun<'_>,
172    environment: &BTreeMap<String, String>,
173) -> (RunOutput, Vec<AttemptRecord>) {
174    let mut last = RunOutput::default();
175    let mut records = Vec::new();
176    for attempt in 1..=check.retry.max.saturating_add(1) {
177        let started = Instant::now();
178        last = run_process(check, run.options.workdir, environment, run.proc_groups);
179        let flake = last.disposition != Disposition::Success
180            && matches_flake_signature(check, &last.combined_output);
181        records.push(AttemptRecord {
182            attempt,
183            conclusion: disposition_conclusion(last.disposition, &last.combined_output),
184            duration_ms: u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
185            flake_matched: flake,
186        });
187        if last.disposition == Disposition::Success || !flake {
188            break;
189        }
190    }
191    (last, records)
192}
193
194fn matches_flake_signature(check: &Check, output: &str) -> bool {
195    check.retry.flake_signatures.iter().any(|pattern| {
196        regex::Regex::new(pattern)
197            .map(|regex| regex.is_match(output))
198            .unwrap_or(false)
199    })
200}
201
202fn disposition_conclusion(disposition: Disposition, output: &str) -> Conclusion {
203    match classify(disposition, output) {
204        None => Conclusion::Success,
205        Some(FailureClass::Timeout) => Conclusion::TimedOut,
206        Some(FailureClass::Infra) => Conclusion::InfraError,
207        Some(_) => Conclusion::Failure,
208    }
209}
210
211fn check_runs_for_trigger(check: &[Trigger], pick: &Trigger) -> bool {
212    if check.is_empty() {
213        return matches!(pick, Trigger::Push);
214    }
215    check.iter().any(|trigger| {
216        matches!(
217            (trigger, pick),
218            (Trigger::Push, Trigger::Push)
219                | (Trigger::Manual, Trigger::Manual)
220                | (Trigger::Cron(_), Trigger::Cron(_))
221        )
222    })
223}
ci_engine/exec.rs

ci_engine/
exec.rs
