From 3d6af51dde6f2e2be7bb32f625482e4ba9aa7e69 Mon Sep 17 00:00:00 2001 From: Chris Thompson Date: Tue, 29 Sep 2026 12:32:08 -0600 Subject: [PATCH] Add opt-in parallel doctor group execution Run ready groups in bounded Tokio tasks with ordered actions and reports. Parallel mode blocks dependents after failed or denied prerequisites, including optional checks and --fix=false; it rejects unknown or cyclic dependencies before running fixes. Runs without the flag stay sequential. Serialize child commands with action and known-error prompts through a shared terminal lock, including the default command path. Resource declarations prevent overlap for shared tools. Cancellation aborts active group tasks and terminates their direct child processes. Document the CLI and schema changes and test dependency order, failure, prompt contention, output capture, and cancellation. --- docs/docs/commands/doctor.md | 17 +- docs/docs/models/ScopeDoctorGroup.mdx | 15 + schema/merged.json | 8 + ...pha.com.github.scope.ScopeDoctorGroup.json | 8 + ...lpha.com.github.scope.ScopeKnownError.json | 8 + ....com.github.scope.ScopeReportLocation.json | 8 + src/doctor/check.rs | 23 + src/doctor/commands/run.rs | 74 +- src/doctor/runner.rs | 874 +++++++++++++++++- src/models/v1alpha/doctor_group.rs | 4 + src/shared/analyze/mod.rs | 12 +- src/shared/capture.rs | 123 ++- src/shared/mod.rs | 3 + src/shared/models/internal/doctor_group.rs | 25 + src/shared/report.rs | 4 + 15 files changed, 1171 insertions(+), 35 deletions(-) diff --git a/docs/docs/commands/doctor.md b/docs/docs/commands/doctor.md index 2266819..7aa66f0 100644 --- a/docs/docs/commands/doctor.md +++ b/docs/docs/commands/doctor.md @@ -36,6 +36,18 @@ dependency subtree. By default, any provided fix's will be run. If you don't want to run fixes add `--fix=false` to disable fixing issues. +`--parallel` runs up to two independent groups at once. A group's actions stay in order, +and a dependent group starts only after its prerequisites succeed. Groups with a shared +[`exclusiveResources` declaration](../models/ScopeDoctorGroup.mdx#exclusive-resources) +never overlap. On a fail-fast result, already-running groups finish and no new groups start. +Failed or user-denied prerequisites block their dependents, including when the failed action is +optional or `--fix=false` leaves a check unresolved. Those groups are reported as skipped. +Scope waits for running child commands before showing a fix confirmation; config authors should +give commands that prompt on their own a shared +`terminal` resource so those commands do not compete for input. +Parallel runs reject cycles and references to unknown prerequisites before reporting success. +The default run remains sequential. + When using a [ScopeDoctorGroup](../models/ScopeDoctorGroup.mdx), the checksum of files are stored on disk. If you need to disable caching, add `--no-cache`. ```text @@ -49,7 +61,8 @@ Options: --skip-only When set, only the named groups are removed from the run; their dependencies still run -f, --fix When set, if a fix is specified it will also run [default: true] [possible values: true, false] -n, --no-cache When set cache will be disabled, forcing all file based checks to run - --yolo Automatically approve all fix prompts without asking + -y, --yolo Automatically approve all fix prompts without asking + --parallel Run independent groups concurrently (up to two); failed prerequisites block dependents (excluded default args) ``` @@ -64,4 +77,4 @@ Will print out all doctor checks available, in the order `run` will execute. INFO - ScopeDoctorGroup/path-exists-fix-in-scope-dir Check your shell for basic functionality .scope/doctor-group-in-scope-dir.yaml INFO - ScopeDoctorGroup/path-exists Check your shell for basic functionality .scope/doctor-group-path-exists.yaml INFO - ScopeDoctorGroup/group-1 Check your shell for basic functionality .scope/doctor-group-1.yaml -``` \ No newline at end of file +``` diff --git a/docs/docs/models/ScopeDoctorGroup.mdx b/docs/docs/models/ScopeDoctorGroup.mdx index 911de77..2672cc8 100644 --- a/docs/docs/models/ScopeDoctorGroup.mdx +++ b/docs/docs/models/ScopeDoctorGroup.mdx @@ -89,6 +89,21 @@ dependency that isn't also required by another group still in the run. A shared also required by a non-skipped group) is kept. The skipped group is still reported in the run summary; its dependencies that get trimmed along with it are not. +## Exclusive resources + +Set `spec.exclusiveResources` to a list of names for operations that must not +overlap, such as Homebrew package changes. During `doctor run --parallel`, Scope +holds those names for the full group execution. Two groups sharing a name run +one after the other; unrelated groups may still run together. +Use a shared `terminal` name for groups whose commands can read from the terminal. + +```yaml +spec: + exclusiveResources: + - homebrew + actions: [] +``` + ## Commands A command can either be relative, or use the PATH. diff --git a/schema/merged.json b/schema/merged.json index 0469eb4..0b4d955 100644 --- a/schema/merged.json +++ b/schema/merged.json @@ -169,6 +169,14 @@ "$ref": "#/$defs/DoctorGroupActionSpec" } }, + "exclusiveResources": { + "description": "Resources that this group must hold exclusively while its actions run.", + "type": "array", + "default": [], + "items": { + "type": "string" + } + }, "include": { "description": "Change how a group is handled when building the dependency task graph.\nWhen set to `when-required`, the group will be ignored unless it's required by another\ndependency.", "$ref": "#/$defs/DoctorInclude", diff --git a/schema/v1alpha.com.github.scope.ScopeDoctorGroup.json b/schema/v1alpha.com.github.scope.ScopeDoctorGroup.json index 8c19c13..0308a46 100644 --- a/schema/v1alpha.com.github.scope.ScopeDoctorGroup.json +++ b/schema/v1alpha.com.github.scope.ScopeDoctorGroup.json @@ -185,6 +185,14 @@ "$ref": "#/$defs/DoctorGroupActionSpec" } }, + "exclusiveResources": { + "description": "Resources that this group must hold exclusively while its actions run.", + "type": "array", + "default": [], + "items": { + "type": "string" + } + }, "include": { "description": "Change how a group is handled when building the dependency task graph.\nWhen set to `when-required`, the group will be ignored unless it's required by another\ndependency.", "$ref": "#/$defs/DoctorInclude", diff --git a/schema/v1alpha.com.github.scope.ScopeKnownError.json b/schema/v1alpha.com.github.scope.ScopeKnownError.json index 72eea4e..4b92705 100644 --- a/schema/v1alpha.com.github.scope.ScopeKnownError.json +++ b/schema/v1alpha.com.github.scope.ScopeKnownError.json @@ -185,6 +185,14 @@ "$ref": "#/$defs/DoctorGroupActionSpec" } }, + "exclusiveResources": { + "description": "Resources that this group must hold exclusively while its actions run.", + "type": "array", + "default": [], + "items": { + "type": "string" + } + }, "include": { "description": "Change how a group is handled when building the dependency task graph.\nWhen set to `when-required`, the group will be ignored unless it's required by another\ndependency.", "$ref": "#/$defs/DoctorInclude", diff --git a/schema/v1alpha.com.github.scope.ScopeReportLocation.json b/schema/v1alpha.com.github.scope.ScopeReportLocation.json index e2711f4..80e3ee5 100644 --- a/schema/v1alpha.com.github.scope.ScopeReportLocation.json +++ b/schema/v1alpha.com.github.scope.ScopeReportLocation.json @@ -185,6 +185,14 @@ "$ref": "#/$defs/DoctorGroupActionSpec" } }, + "exclusiveResources": { + "description": "Resources that this group must hold exclusively while its actions run.", + "type": "array", + "default": [], + "items": { + "type": "string" + } + }, "include": { "description": "Change how a group is handled when building the dependency task graph.\nWhen set to `when-required`, the group will be ignored unless it's required by another\ndependency.", "$ref": "#/$defs/DoctorInclude", diff --git a/src/doctor/check.rs b/src/doctor/check.rs index e2d3686..520c959 100644 --- a/src/doctor/check.rs +++ b/src/doctor/check.rs @@ -1677,6 +1677,29 @@ pub(crate) mod tests { Ok(()) } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn known_error_prompt_waits_for_running_terminal_command() -> Result<()> { + let action = build_run_fail_fix_succeed_action(); + let exec_runner = MockExecutionProvider::new(); + let glob_walker = MockGlobWalker::new(); + let mut run = setup_test(vec![action], exec_runner, glob_walker); + run.yolo = false; + run.known_errors.insert( + "ScopeKnownError/test-error".to_string(), + make_known_error("error-pattern", true), + ); + let report = task_report("error-pattern found in output"); + let command = crate::shared::TERMINAL_ACCESS.read().await; + let prompt = tokio::spawn(async move { run.analyze_known_errors(&[report]).await }); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert!(!prompt.is_finished()); + drop(command); + let status = + tokio::time::timeout(std::time::Duration::from_secs(2), prompt).await???; + assert!(matches!(status, AnalyzeStatus::KnownErrorFoundUserDenied)); + Ok(()) + } + #[tokio::test] async fn no_pattern_match_returns_no_known_errors() -> Result<()> { let action = build_run_fail_fix_succeed_action(); diff --git a/src/doctor/commands/run.rs b/src/doctor/commands/run.rs index 0d614bb..f3df367 100644 --- a/src/doctor/commands/run.rs +++ b/src/doctor/commands/run.rs @@ -47,6 +47,9 @@ pub struct DoctorRunArgs { /// Automatically approve all fix prompts without asking #[arg(long, short = 'y', default_value = "false")] pub yolo: bool, + /// Run independent groups concurrently (up to two); failed prerequisites block dependents. + #[arg(long, action = clap::ArgAction::SetTrue, overrides_with = "parallel")] + pub parallel: bool, } fn get_cache(args: &DoctorRunArgs) -> Arc { @@ -192,6 +195,19 @@ pub async fn doctor_run(found_config: &FoundConfig, args: &DoctorRunArgs) -> Res warn!(target: "user", "Could not find any tasks to execute"); } + if args.parallel { + for name in &all_paths { + let Some(group) = found_config.doctor_group.get(name) else { + continue; + }; + for dependency in &group.requires { + if !found_config.doctor_group.contains_key(dependency) { + anyhow::bail!("group {name} needs unknown group {dependency}"); + } + } + } + } + let graph = RunGraph::new(&all_paths, &found_config.doctor_group); let run_groups = RunGroups { group_actions: transform.groups, @@ -201,7 +217,11 @@ pub async fn doctor_run(found_config: &FoundConfig, args: &DoctorRunArgs) -> Res yolo: args.yolo, }; - let result = run_groups.execute().await?; + let result = if args.parallel { + run_groups.execute_parallel().await? + } else { + run_groups.execute().await? + }; report_stdout!("Summary: {}", result); if let Err(e) = transform.file_cache.persist().await { @@ -329,6 +349,7 @@ fn transform_inputs(found_config: &FoundConfig, args: &DoctorRunArgs) -> RunTran #[cfg(test)] mod test { use super::*; + use clap::Parser; use std::collections::BTreeSet; use std::path::PathBuf; @@ -339,6 +360,57 @@ mod test { DoctorGroup, FoundConfig, MockExecutionProvider, OutputCaptureBuilder, SkipSpec, }; + #[test] + fn parallel_flag_is_opt_in_and_idempotent() { + let default = DoctorRunArgs::try_parse_from(["run"]).unwrap(); + assert!(!default.parallel); + let enabled = DoctorRunArgs::try_parse_from(["run", "--parallel", "--parallel"]).unwrap(); + assert!(enabled.parallel); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_run_rejects_unknown_dependency() { + let mut config = FoundConfig::empty(PathBuf::from("/tmp")); + config.doctor_group.insert( + "group".to_string(), + make_root_model_additional( + vec![], + |meta| meta.name("group"), + |group| group.requires(vec!["missing".to_string()]), + ), + ); + let args = DoctorRunArgs { + parallel: true, + no_cache: true, + ..Default::default() + }; + let error = doctor_run(&config, &args).await.unwrap_err(); + assert!( + error + .to_string() + .contains("group group needs unknown group missing") + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_doctor_run_completes_successfully() { + let mut config = FoundConfig::empty(std::env::temp_dir()); + config.doctor_group.insert( + "first".to_string(), + make_root_model_additional(vec![], |meta| meta.name("first"), group_noop), + ); + config.doctor_group.insert( + "second".to_string(), + make_root_model_additional(vec![], |meta| meta.name("second"), group_noop), + ); + let args = DoctorRunArgs { + parallel: true, + no_cache: true, + ..Default::default() + }; + assert_eq!(doctor_run(&config, &args).await.unwrap(), 0); + } + #[test] fn test_will_include_by_default() { let mut fc = FoundConfig::empty(PathBuf::from("/tmp")); diff --git a/src/doctor/runner.rs b/src/doctor/runner.rs index 5c43714..fa73967 100644 --- a/src/doctor/runner.rs +++ b/src/doctor/runner.rs @@ -6,6 +6,7 @@ use crate::prelude::{ SkipSpec, generate_env_vars, progress_bar_without_pos, }; use crate::shared::prelude::DoctorGroup; +use crate::shared::with_terminal_prompt; use anyhow::Result; use colored::Colorize; use opentelemetry::trace::Status; @@ -104,6 +105,8 @@ struct GroupExecutionResult { group_report: GroupReport, } +const MAX_PARALLEL_GROUPS: usize = 2; + pub struct GroupActionContainer where T: DoctorActionRun, @@ -181,10 +184,9 @@ where /// The dependency DAG over a run's runnable groups (i.e. exactly the names in the topological /// `order` passed to [`RunGraph::new`]). An edge points from a dependent to a group it -/// `requires`. Built once during planning and walked recursively during execution, so a -/// dependent group's span is opened before — and therefore encloses — the spans of the groups -/// it depends on (see issue #172: previously every group span was a sibling directly under -/// `doctor run`, regardless of `requires`). +/// `requires`. The sequential executor walks it recursively, so a dependent group's span +/// encloses the spans of its dependencies (see issue #172). The parallel executor uses the +/// same edges for readiness and parents all concurrently runnable groups under `doctor run`. /// /// A "root" is a group no other group in the graph requires — these are the groups that parent /// directly to the top-level `doctor run` span. A group required by more than one other group @@ -192,9 +194,10 @@ where /// it first during the walk. /// /// Nodes are added in `order`, so `NodeIndex::index()` equals a node's position in `order` — -/// [`RunGraph::roots`] and [`RunGraph::dependencies`] sort by that position so the recursive -/// walk's group-by-group execution sequence matches `order` exactly (this is what preserves -/// fail-fast and reporting behavior; see [`RunGroups::execute`]). +/// [`RunGraph::roots`] and [`RunGraph::dependencies`] sort by that position so the sequential +/// recursive walk's group-by-group execution sequence matches `order` exactly (see +/// [`RunGroups::execute`]). Parallel execution uses the same edges for readiness but parents +/// every group span directly under the run because a group may have several dependencies. pub(crate) struct RunGraph { graph: DiGraph, nodes: BTreeMap, @@ -214,9 +217,8 @@ impl RunGraph { /// points edges dependent-to-dependency (so [`RunGraph::roots`]/[`RunGraph::dependencies`] /// read naturally), owns its `String` node weights, and treats a missing name as /// unremarkable (it was legitimately trimmed by planning, e.g. via `--skip`) rather than as - /// a config error. If dependency-graph semantics change — most likely cycle detection, since - /// none exists today (see the fallback pass in [`RunGroups::execute`]) — check whether it - /// should apply to both. + /// a config error. Parallel runs reject cycles before execution; sequential runs use the + /// fallback pass in [`RunGroups::execute`] to run cyclic groups with a warning. pub(crate) fn new(order: &[String], groups: &BTreeMap) -> Self { let mut graph = DiGraph::::new(); let nodes: BTreeMap = order @@ -316,6 +318,200 @@ impl RunGroups where T: DoctorActionRun, { + /// Execute ready groups with bounded concurrency. A completed failure blocks its + /// dependents; a fail-fast result stops new work while already-running work drains. + pub async fn execute_parallel(self) -> Result + where + T: 'static, + { + if !tokio::runtime::Handle::try_current().is_ok_and(|handle| { + handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread + }) { + anyhow::bail!("parallel doctor run requires a multi-thread Tokio runtime"); + } + if let Err(cycle) = petgraph::algo::toposort(&self.graph.graph, None) { + anyhow::bail!( + "cyclic or unsatisfied doctor dependencies involving {}", + self.graph.name(cycle.node_id()) + ); + } + Arc::new(self).execute_parallel_inner().await + } + + async fn execute_parallel_inner(self: Arc) -> Result + where + T: 'static, + { + let header_span = info_span!("doctor run", "indicatif.pb_show" = true); + header_span.pb_set_length((self.graph.len() + self.skipped_groups.len()) as u64); + header_span.pb_set_message("scope doctor run --parallel"); + let _span = header_span.enter(); + + for name in &self.full_order { + if self.skipped_groups.contains(name) { + header_span.pb_inc(1); + warn!(target: "always", "Group skipped, group: \"{}\"", name); + } + } + + let mut pending: BTreeSet = self + .full_order + .iter() + .filter(|name| self.graph.contains(name)) + .cloned() + .collect(); + let mut completed = BTreeMap::::new(); + let mut blocked = BTreeSet::::new(); + let mut held_resources = BTreeSet::::new(); + let mut running_resources = BTreeMap::>::new(); + // JoinSet aborts active groups if the run itself is cancelled. + let mut running = tokio::task::JoinSet::<(String, Result)>::new(); + let mut stop_new = false; + let mut first_error = None; + + while !pending.is_empty() || !running.is_empty() { + // Revisit all pending groups after each completion. The order is stable, so a + // resource conflict never changes which of two ready groups starts first. + for name in &self.full_order { + if !pending.contains(name) { + continue; + } + let node = self.graph.node(name).expect("pending group is in graph"); + let deps = self.graph.dependencies(node); + if stop_new + || deps.iter().any(|dep| { + let dep_name = self.graph.name(*dep); + blocked.contains(dep_name) + || completed.get(dep_name).is_some_and(|result| { + result.status != GroupExecutionStatus::Succeeded + }) + }) + { + pending.remove(name); + blocked.insert(name.clone()); + header_span.pb_inc(1); + warn!(target: "always", "Group skipped, group: \"{}\" (prerequisite failed or run stopped)", name); + continue; + } + if running.len() == MAX_PARALLEL_GROUPS + || !deps + .iter() + .all(|dep| completed.contains_key(self.graph.name(*dep))) + { + continue; + } + + let Some(container) = self.group_actions.get(name) else { + first_error + .get_or_insert(anyhow::anyhow!("no action container for group {name}")); + stop_new = true; + pending.remove(name); + blocked.insert(name.clone()); + header_span.pb_inc(1); + continue; + }; + if container + .group + .exclusive_resources + .iter() + .any(|resource| held_resources.contains(resource)) + { + continue; + } + for resource in &container.group.exclusive_resources { + held_resources.insert(resource.clone()); + } + running_resources.insert(name.clone(), container.group.exclusive_resources.clone()); + pending.remove(name); + let group_name = name.clone(); + let group_span = info_span!( + parent: &header_span, "group", "indicatif.pb_show" = true, + "group.name" = group_name.as_str(), + "otel.name" = format!("group {}", group_name) + ); + group_span.pb_set_length(container.actions.len() as u64); + group_span.pb_set_message(&format!("group {group_name}")); + let run_groups = self.clone(); + let running_name = group_name.clone(); + running.spawn(async move { + let container = run_groups + .group_actions + .get(&group_name) + .expect("scheduled group still has its action container"); + let result = run_groups + .execute_group(&group_span, container) + .instrument(group_span.clone()) + .await; + if let Ok(group_result) = &result + && group_result.status == GroupExecutionStatus::Failed + { + group_span.set_status(Status::Error { + description: std::borrow::Cow::Owned(format!( + "{} group failed", + group_result.group_name + )), + }); + } + (running_name, result) + }); + } + + if let Some(joined) = running.join_next().await { + header_span.pb_inc(1); + match joined { + Ok((name, result)) => { + for resource in running_resources.remove(&name).unwrap_or_default() { + held_resources.remove(&resource); + } + match result { + Ok(group_result) => { + stop_new |= group_result.skip_remaining; + completed.insert(name, group_result); + } + Err(error) => { + stop_new = true; + blocked.insert(name); + first_error.get_or_insert(error); + } + } + } + Err(error) => { + stop_new = true; + first_error.get_or_insert(error.into()); + } + } + } else if !pending.is_empty() { + return Err(anyhow::anyhow!( + "cyclic or unsatisfied doctor dependencies: {}", + pending.iter().cloned().collect::>().join(", ") + )); + } + } + + if let Some(error) = first_error { + return Err(error); + } + + let mut result = PathRunResult { + did_succeed: true, + succeeded_groups: BTreeSet::new(), + failed_group: BTreeSet::new(), + skipped_group: BTreeSet::new(), + group_reports: Vec::new(), + }; + for name in &self.full_order { + if self.skipped_groups.contains(name) { + result.skipped_group.insert(name.clone()); + result.group_reports.push(GroupReport::new(name)); + } else if let Some(group_result) = completed.get(name) { + result.process(group_result); + } else if blocked.contains(name) { + result.skipped_group.insert(name.clone()); + } + } + Ok(result) + } + pub async fn execute(&self) -> Result { let header_span = info_span!("doctor run", "indicatif.pb_show" = true); header_span.pb_set_length((self.graph.len() + self.skipped_groups.len()) as u64); @@ -575,16 +771,18 @@ where } fn prompt_user(prompt_text: &str, maybe_help_text: &Option) -> bool { - tracing_indicatif::suspend_tracing_indicatif(|| { - let prompt = { - let base_prompt = inquire::Confirm::new(prompt_text).with_default(false); - match maybe_help_text { - Some(help_text) => base_prompt.with_help_message(help_text), - None => base_prompt, - } - }; + with_terminal_prompt(|| { + tracing_indicatif::suspend_tracing_indicatif(|| { + let prompt = { + let base_prompt = inquire::Confirm::new(prompt_text).with_default(false); + match maybe_help_text { + Some(help_text) => base_prompt.with_help_message(help_text), + None => base_prompt, + } + }; - prompt.prompt().unwrap_or(false) + prompt.prompt().unwrap_or(false) + }) }) } @@ -869,6 +1067,7 @@ pub fn compute_group_order(params: GroupOrderParams) -> Vec { #[cfg(test)] mod tests { use super::*; + use async_trait::async_trait; use crate::doctor::check::tests::build_run_fail_fix_succeed_action; use crate::doctor::check::{ @@ -881,8 +1080,10 @@ mod tests { use crate::prelude::{ActionReport, ActionTaskReport, CaptureError, MockExecutionProvider}; use anyhow::Result; use std::collections::{BTreeMap, BTreeSet, HashMap}; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::vec; + use tokio::sync::Notify; use tracing::field::{Field, Visit}; use tracing_subscriber::layer::{Context, Layer, SubscriberExt}; @@ -890,6 +1091,641 @@ mod tests { BTreeSet::new() } + struct TimedAction { + name: &'static str, + active: Arc, + peak: Arc, + events: Option>>>, + status: ActionRunStatus, + required: bool, + delay_ms: u64, + should_run: bool, + fail_with_error: bool, + panic_after_delay: bool, + } + + #[async_trait] + impl DoctorActionRun for TimedAction { + async fn run_action( + &self, + _: for<'a> fn(&'a str, &'a Option) -> bool, + ) -> Result { + assert!(self.should_run, "{} must not run", self.name); + if let Some(events) = &self.events { + events + .lock() + .unwrap() + .push(format!("{} started", self.name)); + } + let now = self.active.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(now, Ordering::SeqCst); + tokio::time::sleep(std::time::Duration::from_millis(self.delay_ms)).await; + self.active.fetch_sub(1, Ordering::SeqCst); + if let Some(events) = &self.events { + events + .lock() + .unwrap() + .push(format!("{} finished", self.name)); + } + if self.fail_with_error { + anyhow::bail!("{} action error", self.name); + } + assert!(!self.panic_after_delay, "{} action panicked", self.name); + Ok(ActionRunResult::new( + self.name, + self.status.clone(), + None, + None, + None, + )) + } + fn required(&self) -> bool { + self.required + } + fn name(&self) -> String { + self.name.to_string() + } + fn description(&self) -> String { + self.name.to_string() + } + fn help_text(&self) -> Option { + None + } + fn help_url(&self) -> Option { + None + } + } + + struct TerminalAction { + name: &'static str, + started: Arc, + events: Arc>>, + } + + #[async_trait] + impl DoctorActionRun for TerminalAction { + async fn run_action( + &self, + _: for<'a> fn(&'a str, &'a Option) -> bool, + ) -> Result { + if self.name == "command" { + let _terminal = crate::shared::TERMINAL_ACCESS.read().await; + self.started.notify_one(); + tokio::time::sleep(std::time::Duration::from_millis(40)).await; + self.events.lock().unwrap().push("command finished"); + } else { + self.started.notified().await; + with_terminal_prompt(|| self.events.lock().unwrap().push("prompt shown")); + } + Ok(ActionRunResult::new( + self.name, + ActionRunStatus::CheckSucceeded, + None, + None, + None, + )) + } + fn required(&self) -> bool { + true + } + fn name(&self) -> String { + self.name.to_string() + } + fn description(&self) -> String { + self.name.to_string() + } + fn help_text(&self) -> Option { + None + } + fn help_url(&self) -> Option { + None + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_prompt_waits_for_other_command_without_deadlock() -> Result<()> { + let started = Arc::new(Notify::new()); + let events = Arc::new(Mutex::new(Vec::new())); + let action = |name| TerminalAction { + name, + started: started.clone(), + events: events.clone(), + }; + let group_actions = BTreeMap::from([ + make_group_action_requiring("command", vec![], vec![action("command")]), + make_group_action_requiring("prompt", vec![], vec![action("prompt")]), + ]); + let order = ["command", "prompt"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.map(str::to_string).to_vec(), + skipped_groups: BTreeSet::new(), + yolo: false, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = tokio::time::timeout( + std::time::Duration::from_secs(2), + run_groups.execute_parallel(), + ) + .await??; + assert!(result.did_succeed); + assert_eq!( + *events.lock().unwrap(), + vec!["command finished", "prompt shown"] + ); + Ok(()) + } + + #[tokio::test] + async fn parallel_requires_multithread_runtime() { + let run_groups = RunGroups:: { + graph: RunGraph::new(&[], &BTreeMap::new()), + group_actions: BTreeMap::new(), + full_order: Vec::new(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let error = run_groups.execute_parallel().await.unwrap_err(); + assert!(error.to_string().contains("multi-thread Tokio runtime")); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_independent_groups_overlap_but_child_waits() -> Result<()> { + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let events = Arc::new(Mutex::new(Vec::new())); + let action = |name| TimedAction { + name, + active: active.clone(), + peak: peak.clone(), + events: Some(events.clone()), + status: ActionRunStatus::CheckSucceeded, + required: true, + delay_ms: 40, + should_run: true, + fail_with_error: false, + panic_after_delay: false, + }; + let group_actions = BTreeMap::from([ + make_group_action_requiring("a", vec![], vec![action("a")]), + make_group_action_requiring("b", vec![], vec![action("b")]), + make_group_action_requiring("child", vec!["a", "b"], vec![action("child")]), + ]); + let order = ["a", "b", "child"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.iter().map(|name| name.to_string()).collect(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = run_groups.execute_parallel().await?; + assert!(result.did_succeed); + assert_eq!(peak.load(Ordering::SeqCst), 2); + assert_eq!(result.group_reports.len(), 3); + let events = events.lock().unwrap(); + let child_start = events + .iter() + .position(|event| event == "child started") + .unwrap(); + assert!(events[..child_start].contains(&"a finished".to_string())); + assert!(events[..child_start].contains(&"b finished".to_string())); + assert_eq!( + result + .group_reports + .iter() + .map(GroupReport::group_name) + .collect::>(), + vec!["a", "b", "child"] + ); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_diamond_runs_shared_group_once_and_orders_actions() -> Result<()> { + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let events = Arc::new(Mutex::new(Vec::new())); + let action = |name| TimedAction { + name, + active: active.clone(), + peak: peak.clone(), + events: Some(events.clone()), + status: ActionRunStatus::CheckSucceeded, + required: true, + delay_ms: 10, + should_run: true, + fail_with_error: false, + panic_after_delay: false, + }; + let group_actions = BTreeMap::from([ + make_group_action_requiring("shared", vec![], vec![action("shared")]), + make_group_action_requiring("left", vec!["shared"], vec![action("left")]), + make_group_action_requiring("right", vec!["shared"], vec![action("right")]), + make_group_action_requiring( + "root", + vec!["left", "right"], + vec![action("root first"), action("root second")], + ), + ]); + let order = ["shared", "left", "right", "root"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.map(str::to_string).to_vec(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = run_groups.execute_parallel().await?; + assert!(result.did_succeed); + assert_eq!(peak.load(Ordering::SeqCst), 2); + let events = events.lock().unwrap(); + assert_eq!( + events + .iter() + .filter(|event| event.as_str() == "shared started") + .count(), + 1 + ); + let position = |event: &str| { + events + .iter() + .position(|item| item.as_str() == event) + .unwrap() + }; + assert!(position("shared finished") < position("left started")); + assert!(position("shared finished") < position("right started")); + assert!(position("left finished") < position("root first started")); + assert!(position("right finished") < position("root first started")); + assert!(position("root first finished") < position("root second started")); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_exclusive_resource_serializes_groups() -> Result<()> { + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let action = |name| TimedAction { + name, + active: active.clone(), + peak: peak.clone(), + events: None, + status: ActionRunStatus::CheckSucceeded, + required: true, + delay_ms: 40, + should_run: true, + fail_with_error: false, + panic_after_delay: false, + }; + let mut group_actions = BTreeMap::from([ + make_group_action_requiring("a", vec![], vec![action("a")]), + make_group_action_requiring("b", vec![], vec![action("b")]), + ]); + for container in group_actions.values_mut() { + container.group.exclusive_resources = vec!["homebrew".to_string()]; + } + let order = ["a", "b"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.iter().map(|name| name.to_string()).collect(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = run_groups.execute_parallel().await?; + assert!(result.did_succeed); + assert_eq!(peak.load(Ordering::SeqCst), 1); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_optional_failure_blocks_dependent_and_drains_independent() -> Result<()> { + let group_actions = BTreeMap::from([ + make_group_action_requiring( + "failed", + vec![], + vec![make_action_run( + ActionRunStatus::CheckFailedFixSucceedVerifyFailed, + false, + )], + ), + make_group_action_requiring( + "independent", + vec![], + make_action_runs(ActionRunStatus::CheckSucceeded), + ), + make_group_action_requiring("child", vec!["failed"], will_not_run()), + ]); + let order = ["failed", "independent", "child"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.iter().map(|name| name.to_string()).collect(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = run_groups.execute_parallel().await?; + assert!(!result.did_succeed); + assert_eq!(result.failed_group, BTreeSet::from(["failed".to_string()])); + assert_eq!( + result.succeeded_groups, + BTreeSet::from(["independent".to_string()]) + ); + assert!(result.skipped_group.contains("child")); + assert_eq!(result.group_reports.len(), 2); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_fail_fast_drains_running_and_does_not_start_queued() -> Result<()> { + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let action = |name, status, delay_ms, should_run| TimedAction { + name, + active: active.clone(), + peak: peak.clone(), + events: None, + status, + required: true, + delay_ms, + should_run, + fail_with_error: false, + panic_after_delay: false, + }; + let group_actions = BTreeMap::from([ + make_group_action_requiring( + "failed", + vec![], + vec![action( + "failed", + ActionRunStatus::CheckFailedFixFailedStop, + 10, + true, + )], + ), + make_group_action_requiring( + "running", + vec![], + vec![action("running", ActionRunStatus::CheckSucceeded, 40, true)], + ), + make_group_action_requiring( + "queued", + vec![], + vec![action("queued", ActionRunStatus::CheckSucceeded, 0, false)], + ), + ]); + let order = ["failed", "running", "queued"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.map(str::to_string).to_vec(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = run_groups.execute_parallel().await?; + assert_eq!(active.load(Ordering::SeqCst), 0); + assert_eq!(peak.load(Ordering::SeqCst), 2); + assert!(result.succeeded_groups.contains("running")); + assert!(result.failed_group.contains("failed")); + assert!(result.skipped_group.contains("queued")); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_action_error_drains_running_group() -> Result<()> { + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let events = Arc::new(Mutex::new(Vec::new())); + let action = |name, delay_ms, fail_with_error| TimedAction { + name, + active: active.clone(), + peak: peak.clone(), + events: Some(events.clone()), + status: ActionRunStatus::CheckSucceeded, + required: true, + delay_ms, + should_run: true, + fail_with_error, + panic_after_delay: false, + }; + let group_actions = BTreeMap::from([ + make_group_action_requiring("error", vec![], vec![action("error", 10, true)]), + make_group_action_requiring("running", vec![], vec![action("running", 40, false)]), + ]); + let order = ["error", "running"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.map(str::to_string).to_vec(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let error = run_groups.execute_parallel().await.unwrap_err(); + assert!(error.to_string().contains("error action error")); + assert_eq!(active.load(Ordering::SeqCst), 0); + assert!( + events + .lock() + .unwrap() + .contains(&"running finished".to_string()) + ); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_action_panic_drains_running_group() { + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let events = Arc::new(Mutex::new(Vec::new())); + let action = |name, delay_ms, panic_after_delay| TimedAction { + name, + active: active.clone(), + peak: peak.clone(), + events: Some(events.clone()), + status: ActionRunStatus::CheckSucceeded, + required: true, + delay_ms, + should_run: true, + fail_with_error: false, + panic_after_delay, + }; + let group_actions = BTreeMap::from([ + make_group_action_requiring("panic", vec![], vec![action("panic", 10, true)]), + make_group_action_requiring("running", vec![], vec![action("running", 40, false)]), + ]); + let order = ["panic", "running"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.map(str::to_string).to_vec(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + assert!(run_groups.execute_parallel().await.is_err()); + assert!( + events + .lock() + .unwrap() + .contains(&"running finished".to_string()) + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelled_parallel_run_aborts_running_group() -> Result<()> { + let events = Arc::new(Mutex::new(Vec::new())); + let group_actions = BTreeMap::from([make_group_action_requiring( + "long", + vec![], + vec![TimedAction { + name: "long", + active: Arc::new(AtomicUsize::new(0)), + peak: Arc::new(AtomicUsize::new(0)), + events: Some(events.clone()), + status: ActionRunStatus::CheckSucceeded, + required: true, + delay_ms: 100, + should_run: true, + fail_with_error: false, + panic_after_delay: false, + }], + )]); + let order = ["long"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.map(str::to_string).to_vec(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let run = tokio::spawn(run_groups.execute_parallel()); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while !events.lock().unwrap().contains(&"long started".to_string()) { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await?; + run.abort(); + assert!(run.await.unwrap_err().is_cancelled()); + tokio::time::sleep(std::time::Duration::from_millis(120)).await; + assert!( + !events + .lock() + .unwrap() + .contains(&"long finished".to_string()) + ); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_user_denied_prerequisite_blocks_child() -> Result<()> { + let group_actions = BTreeMap::from([ + make_group_action_requiring( + "denied", + vec![], + make_action_runs(ActionRunStatus::CheckFailedFixUserDenied), + ), + make_group_action_requiring("child", vec!["denied"], will_not_run()), + ]); + let order = ["denied", "child"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.map(str::to_string).to_vec(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = run_groups.execute_parallel().await?; + assert!(!result.did_succeed); + assert_eq!( + result.skipped_group, + BTreeSet::from(["denied".to_string(), "child".to_string()]) + ); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_fix_false_failure_blocks_dependent_check() -> Result<()> { + let group_actions = BTreeMap::from([ + make_group_action_requiring( + "unresolved", + vec![], + make_action_runs(ActionRunStatus::CheckFailedNoRunFix), + ), + make_group_action_requiring("child", vec!["unresolved"], will_not_run()), + ]); + let order = ["unresolved", "child"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.map(str::to_string).to_vec(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = run_groups.execute_parallel().await?; + assert!(!result.did_succeed); + assert!(result.failed_group.contains("unresolved")); + assert!(result.skipped_group.contains("child")); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_cycle_fails_instead_of_silently_dropping_groups() -> Result<()> { + let group_actions = BTreeMap::from([ + make_group_action_requiring("a", vec!["b"], will_not_run()), + make_group_action_requiring("b", vec!["a"], will_not_run()), + make_group_action_requiring("unrelated", vec![], will_not_run()), + ]); + let order = ["unrelated", "a", "b"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: order.iter().map(|name| name.to_string()).collect(), + skipped_groups: BTreeSet::new(), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let error = run_groups.execute_parallel().await.unwrap_err(); + assert!(error.to_string().contains("cyclic or unsatisfied")); + assert!( + error.to_string().contains("involving a") || error.to_string().contains("involving b") + ); + Ok(()) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parallel_planned_skip_does_not_block_consumer() -> Result<()> { + let group_actions = BTreeMap::from([make_group_action_requiring( + "consumer", + vec!["conditional"], + make_action_runs(ActionRunStatus::CheckSucceeded), + )]); + let order = ["consumer"]; + let run_groups = RunGroups { + graph: graph_from(&order, &group_actions), + group_actions, + full_order: ["conditional", "consumer"].map(str::to_string).to_vec(), + skipped_groups: BTreeSet::from(["conditional".to_string()]), + yolo: true, + }; + let _serialize = SPAN_CALLSITE_LOCK.lock().await; + let result = run_groups.execute_parallel().await?; + assert!(result.did_succeed); + assert!(result.succeeded_groups.contains("consumer")); + assert!(result.skipped_group.contains("conditional")); + Ok(()) + } + #[tokio::test] async fn test_compute_group_order_with_no_dep_will_have_no_tasks() -> Result<()> { let action = build_run_fail_fix_succeed_action(); diff --git a/src/models/v1alpha/doctor_group.rs b/src/models/v1alpha/doctor_group.rs index 990c393..787776c 100644 --- a/src/models/v1alpha/doctor_group.rs +++ b/src/models/v1alpha/doctor_group.rs @@ -107,6 +107,10 @@ pub struct DoctorGroupSpec { #[serde(default)] pub needs: Vec, + /// Resources that this group must hold exclusively while its actions run. + #[serde(default)] + pub exclusive_resources: Vec, + /// A series of steps to check and fix for the group. pub actions: Vec, diff --git a/src/shared/analyze/mod.rs b/src/shared/analyze/mod.rs index 2140e59..046c056 100644 --- a/src/shared/analyze/mod.rs +++ b/src/shared/analyze/mod.rs @@ -71,11 +71,8 @@ where &captures, )?; - tracing_indicatif::suspend_tracing_indicatif(|| { - let exec_path = ke.metadata.exec_path(); - prompt_and_run_fix(working_dir, exec_path, &fix, yolo) - }) - .await? + let exec_path = ke.metadata.exec_path(); + prompt_and_run_fix(working_dir, exec_path, &fix, yolo).await? } None => AnalyzeStatus::KnownErrorFoundNoFixFound, }; @@ -126,7 +123,10 @@ async fn prompt_and_run_fix( } }; - match prompt.prompt() { + let response = crate::shared::with_terminal_prompt(|| { + tracing_indicatif::suspend_tracing_indicatif(|| prompt.prompt()) + }); + match response { Ok(accepted) => Ok::(accepted), Err(InquireError::NotTTY) => { warn!(target: "user", "Prompting user for fix, but input device is not a TTY. Skipping fix."); diff --git a/src/shared/capture.rs b/src/shared/capture.rs index 0f572fb..3b5d75f 100644 --- a/src/shared/capture.rs +++ b/src/shared/capture.rs @@ -10,13 +10,33 @@ use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; use std::process::Stdio; use std::sync::Arc; +use std::sync::LazyLock; use thiserror::Error; use tokio::io; use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::sync::RwLock; -use tracing::{Level, debug, error, info, instrument}; +use tracing::{Instrument, Level, debug, error, info, instrument}; use which::which_in; +// Child commands inherit the terminal. A Scope confirmation must wait for +// running commands to release it, while independent commands may run together. +pub(crate) static TERMINAL_ACCESS: LazyLock> = LazyLock::new(RwLock::default); + +pub(crate) fn with_terminal_prompt(prompt: impl FnOnce() -> R) -> R { + // The CLI uses a multi-thread runtime. Keep the old prompt behavior for + // library callers using a current-thread runtime. + if tokio::runtime::Handle::try_current() + .is_ok_and(|handle| handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread) + { + tokio::task::block_in_place(|| { + let _terminal_access = TERMINAL_ACCESS.blocking_write(); + prompt() + }) + } else { + prompt() + } +} + #[derive(Debug, Default)] struct RwLockOutput { output: RwLock, String)>>, @@ -166,6 +186,7 @@ impl OutputCapture { #[instrument(skip_all)] pub async fn capture_output(opts: CaptureOpts<'_>) -> Result { check_pre_exec(&opts)?; + let _terminal_access = TERMINAL_ACCESS.read().await; let args = opts.args.to_vec(); debug!("Executing PATH={} {:?}", &opts.path, &args); @@ -180,8 +201,13 @@ impl OutputCapture { .stderr(Stdio::piped()) .stdout(Stdio::piped()) .current_dir(opts.working_dir) + .kill_on_drop(true) .spawn()?; + // Both pipes must be read while the child is running. JoinSet aborts the + // readers if the capture is cancelled; kill_on_drop stops the child. + let mut readers = tokio::task::JoinSet::new(); + // capture stdout let stdout = child.stdout.take().expect("stdout to be available"); let stdout_stream = StreamCapture { @@ -190,7 +216,8 @@ impl OutputCapture { level: Level::INFO, dest: opts.output_dest.clone(), }; - let stdout = stdout_stream.capture_output(); + readers + .spawn(async move { (true, stdout_stream.capture_output().await) }.in_current_span()); // capture stderr let stderr = child.stderr.take().expect("stderr to be available"); @@ -200,16 +227,23 @@ impl OutputCapture { level: Level::ERROR, dest: opts.output_dest.clone(), }; - let stderr = stderr_stream.capture_output(); + readers + .spawn(async move { (false, stderr_stream.capture_output().await) }.in_current_span()); // wait for app to exit - let (command_result, wait_stdout, wait_stderr) = tokio::join!(child.wait(), stdout, stderr); + let command_result = child.wait().await; + let mut captured_stdout = Vec::new(); + let mut captured_stderr = Vec::new(); + while let Some(read_result) = readers.join_next().await { + match read_result { + Ok((true, Ok(lines))) => captured_stdout = lines, + Ok((false, Ok(lines))) => captured_stderr = lines, + _ => {} + } + } let end_time = Utc::now(); debug!("join result {:?}", command_result); - let captured_stdout = wait_stdout.unwrap_or_default(); - let captured_stderr = wait_stderr.unwrap_or_default(); - Ok(Self { working_dir: opts.working_dir.to_path_buf(), stdout: captured_stdout, @@ -316,3 +350,78 @@ fn check_pre_exec(opts: &CaptureOpts) -> Result<(), CaptureError> { Ok(()) } + +#[cfg(test)] +mod terminal_tests { + use super::*; + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn child_waits_for_terminal_and_drains_large_output() { + let prompt = TERMINAL_ACCESS.write().await; + let child = tokio::spawn(async { + let workdir = std::env::temp_dir(); + let args = vec![ + "sh".to_string(), + "-c".to_string(), + "yes output | head -n 20000".to_string(), + ]; + OutputCapture::capture_output(CaptureOpts { + working_dir: &workdir, + env_vars: BTreeMap::new(), + path: "/usr/bin:/bin", + args: &args, + output_dest: OutputDisplay::Silent, + }) + .await + .unwrap() + }); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert!(!child.is_finished()); + drop(prompt); + let output = tokio::time::timeout(std::time::Duration::from_secs(5), child) + .await + .expect("child should finish after prompt releases terminal") + .unwrap(); + assert_eq!(output.exit_code, Some(0)); + assert_eq!(output.get_stdout().lines().count(), 20000); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelled_capture_stops_child_command() { + let directory = tempfile::tempdir().unwrap(); + let ready = directory.path().join("ready"); + let marker = directory.path().join("marker"); + let workdir = directory.path().to_path_buf(); + let ready_value = ready.to_string_lossy().into_owned(); + let marker_value = marker.to_string_lossy().into_owned(); + let capture = tokio::spawn(async move { + let args = vec![ + "sh".to_string(), + "-c".to_string(), + "touch \"$READY\"; sleep 1; touch \"$MARKER\"".to_string(), + ]; + OutputCapture::capture_output(CaptureOpts { + working_dir: &workdir, + env_vars: BTreeMap::from([ + ("READY".to_string(), ready_value), + ("MARKER".to_string(), marker_value), + ]), + path: "/usr/bin:/bin", + args: &args, + output_dest: OutputDisplay::Silent, + }) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while !ready.exists() { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("child should start"); + capture.abort(); + assert!(capture.await.unwrap_err().is_cancelled()); + tokio::time::sleep(std::time::Duration::from_millis(1100)).await; + assert!(!marker.exists()); + } +} diff --git a/src/shared/mod.rs b/src/shared/mod.rs index 43a8ef5..f96519d 100644 --- a/src/shared/mod.rs +++ b/src/shared/mod.rs @@ -6,6 +6,9 @@ use std::cmp::max; use std::path::Path; mod capture; +#[cfg(test)] +pub(crate) use capture::TERMINAL_ACCESS; +pub(crate) use capture::with_terminal_prompt; mod config_load; mod logging; diff --git a/src/shared/models/internal/doctor_group.rs b/src/shared/models/internal/doctor_group.rs index 9240f2e..ba81d44 100644 --- a/src/shared/models/internal/doctor_group.rs +++ b/src/shared/models/internal/doctor_group.rs @@ -53,6 +53,8 @@ pub struct DoctorGroup { pub full_name: String, pub metadata: ModelMetadata, pub requires: Vec, + #[builder(default)] + pub exclusive_resources: Vec, pub run_by_default: bool, pub actions: Vec, pub extra_report_args: BTreeMap, @@ -84,6 +86,7 @@ impl TryFrom for DoctorGroup { metadata: model.metadata, actions, requires: model.spec.needs, + exclusive_resources: model.spec.exclusive_resources, run_by_default: model.spec.include == DoctorInclude::ByDefault, extra_report_args: model.spec.report_extra_details, skip: model.spec.skip, @@ -232,6 +235,28 @@ mod tests { ); } + #[test] + fn parses_exclusive_resources() { + let text = r#" +apiVersion: scope.github.com/v1alpha +kind: ScopeDoctorGroup +metadata: + name: package-install +spec: + exclusiveResources: + - homebrew + actions: [] +"#; + let configs = parse_models_from_string( + Path::new("/foo/bar"), + Path::new("/foo/bar/.scope/file.yaml"), + text, + ) + .unwrap(); + let group = configs[0].get_doctor_group().unwrap(); + assert_eq!(group.exclusive_resources, vec!["homebrew"]); + } + /// Regression test for https://github.com/Gusto/scope/issues/153: `check` becoming /// `Option` must not change how an explicit, empty `check: {}` (the /// pre-existing "always run the fix" idiom) is parsed. diff --git a/src/shared/report.rs b/src/shared/report.rs index c492142..4fcd75d 100644 --- a/src/shared/report.rs +++ b/src/shared/report.rs @@ -225,6 +225,10 @@ pub struct GroupReport { } impl GroupReport { + pub fn group_name(&self) -> &str { + &self.group_name + } + pub fn add_action(&mut self, action_report: &ActionReport) { self.action_result.push(action_report.clone()); }