use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::process::ExitCode; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, mpsc}; use std::time::{Duration, Instant}; use avril_runtime::config::{Error as ConfigError, ErrorKind, Span}; use avril_runtime::package_config::{PackageConfig, Task}; use avril_subprocess::Scope; use avril_task::{ Relevance, WatchContext, WatchEvent, WatchSet, WatchState, external_roots, forget_directories, known_directories, remember_directories, resolution_paths, resolved_input_paths, shell_quote, }; use crate::cli::ParseError; type Result = std::result::Result>; type TaskResult = std::result::Result; #[derive(Debug, thiserror::Error)] enum TaskError { #[error("{0}")] Message(String), #[error(transparent)] Io(#[from] std::io::Error), } impl From for TaskError { fn from(message: String) -> Self { Self::Message(message) } } struct Loaded { config: PackageConfig, graph: avril_task::Graph, source: String, path: PathBuf, root: PathBuf, } fn load(cwd: &Path) -> Result { let path = cwd .ancestors() .map(|directory| directory.join("avril.conf")) .find(|candidate| candidate.is_file()) .ok_or_else(|| { format!( "No avril.conf found in {} or any parent directory", cwd.display() ) })?; load_file(path) } fn load_file(path: PathBuf) -> Result { let source = std::fs::read_to_string(&path)?; let base = url::Url::from_file_path(&path).map_err(|_| "Config paths must be absolute")?; let config = PackageConfig::parse(&source, &base) .map_err(|error| error.render(&path.to_string_lossy(), &source))?; let root = path.parent().unwrap().to_path_buf(); let graph = to_graph(&config.tasks); Ok(Loaded { config, graph, source, path, root, }) } fn list(tasks: &BTreeMap) -> String { if tasks.is_empty() { return "No tasks are defined in avril.conf".into(); } let mut text = String::from("Available tasks:"); for (name, task) in tasks { text.push_str(&format!("\n- {name}")); if let Some(description) = &task.description { text.push_str(&format!("\n {description}")); } else if let Some(command) = &task.command { text.push_str(&format!("\n {}", command.display())); } else { let dependencies: Vec<_> = task .dependencies .iter() .map(|dependency| dependency.name.as_str()) .collect(); text.push_str(&format!("\n dependencies: {}", dependencies.join(", "))); } } text } pub struct Args { watch: bool, name: Option, extra: Vec, } /// `avril task` argument parsing failed. #[derive(Debug, thiserror::Error)] pub enum TaskParseError { #[error(transparent)] Generic(#[from] ParseError), } fn task_usage() -> TaskParseError { ParseError::Usage( crate::cli::command("task") .map(|command| command.usage()) .unwrap_or_else(crate::cli::usage), ) .into() } pub fn parse(args: &[String]) -> std::result::Result { let (watch, rest) = match args.first().map(String::as_str) { Some("--watch") => (true, &args[1..]), _ => (false, args), }; let Some(name) = rest.first() else { if watch { return Err(task_usage()); } return Ok(Args { watch, name: None, extra: Vec::new(), }); }; if name.starts_with('-') { return Err(task_usage()); } Ok(Args { watch, name: Some(name.clone()), extra: rest[1..].to_vec(), }) } pub fn run(args: Args) -> Result { let cwd = std::env::current_dir()?.canonicalize()?; let Args { watch, name, extra } = args; let Some(name) = name else { let loaded = load(&cwd)?; println!("{}", list(&loaded.config.tasks)); return Ok(ExitCode::SUCCESS); }; let loaded = load(&cwd)?; if !loaded.config.tasks.contains_key(&name) { return Err(format!("Unknown task {name:?}\n\n{}", list(&loaded.config.tasks)).into()); } let mut extra = extra.as_slice(); if extra.first().is_some_and(|arg| arg == "--") { extra = &extra[1..]; } let host = avril_host::Host::new()?; let handle = host.handle(); let _enter = handle.enter(); if watch { return watch_run(&cwd, loaded, &name, extra); } match execute_graph(&loaded, &cwd, &name, extra, None, false) { Ok(code) => Ok(ExitCode::from(code)), Err(error) => Err(error.into()), } } const DEBOUNCE: Duration = Duration::from_millis(50); static SHUTDOWN: AtomicBool = AtomicBool::new(false); #[cfg(unix)] fn install_shutdown_handler() -> Result<()> { extern "C" fn handler(_signal: libc::c_int) { SHUTDOWN.store(true, Ordering::SeqCst); } for signal in [libc::SIGINT, libc::SIGTERM] { if unsafe { libc::signal(signal, handler as *const () as libc::sighandler_t) } == libc::SIG_ERR { return Err(std::io::Error::last_os_error().into()); } } Ok(()) } #[cfg(not(unix))] fn install_shutdown_handler() -> Result<()> { ctrlc::set_handler(|| SHUTDOWN.store(true, Ordering::SeqCst))?; Ok(()) } fn create_watcher( loaded: &Loaded, watch_set: &WatchSet, extra: &[PathBuf], sender: mpsc::Sender>, ) -> Result { use notify::Watcher; let config = notify::Config::default().with_follow_symlinks(false); let mut watcher = notify::RecommendedWatcher::new( move |result| { let _ = sender.send(result); }, config, )?; watcher.watch(&loaded.root, notify::RecursiveMode::Recursive)?; for parent in external_roots(&loaded.root, watch_set, extra) { watcher.watch(&parent, notify::RecursiveMode::NonRecursive)?; } Ok(watcher) } /// One watched run: its thread (while still tracked) and the control /// owning its process groups. A generation whose thread cannot be joined /// yet is retired with its handle rather than joined unconditionally. struct Generation { handle: Option>>, control: Arc, forced: bool, } impl Generation { /// Escalates INT, TERM, KILL against a generation's groups. Returns whether /// everything settled: the run thread (if still tracked) finished and no /// group has live members left. Settling is bounded even after KILL, so a /// member stuck in uninterruptible sleep cannot hang the watcher. fn stop(&mut self, name: &str) -> bool { if !self.control.terminate(|| { self.handle .as_ref() .is_none_or(|handle| handle.is_finished()) }) { eprintln!( "Task {name}: some processes did not exit after SIGKILL; leaving them behind" ); return false; } if let Some(handle) = self.handle.take() { let _ = handle.join(); } true } } fn watch_run(cwd: &Path, loaded: Loaded, name: &str, extra: &[String]) -> Result { if cfg!(not(any( target_os = "linux", target_os = "android", target_os = "macos", windows ))) { return Err("Task --watch is not supported on this platform".into()); } let watch_set = loaded .graph .watch_set(name, &loaded.root) .map_err(|error| render_task_error(&loaded, error))?; if watch_set.patterns.is_empty() { return Err(format!( "Task {name:?} cannot be watched: neither it nor its dependencies declare 'files'" ) .into()); } install_shutdown_handler()?; let mut loaded = Arc::new(loaded); let mut watch_set = watch_set; let (sender, receiver) = mpsc::channel(); let mut config_paths = resolution_paths(&loaded.path); let mut input_paths = resolved_input_paths(&loaded.root, &watch_set); let linked: Vec = config_paths.iter().chain(&input_paths).cloned().collect(); let mut _watcher = create_watcher(&loaded, &watch_set, &linked, sender.clone())?; eprintln!("Task {name}: watching for changes"); let mut baseline = WatchState::capture( &loaded.graph, &loaded.root, &loaded.source, name, extra, &watch_set, ); macro_rules! start_run { ($force:expr) => {{ let control = Arc::new(Scope::new( &std::env::current_exe()?, vec![ ("SIGINT".into(), Duration::from_secs(1)), ("SIGTERM".into(), Duration::from_secs(1)), ], )?); let run_runtime = tokio::runtime::Handle::current(); let run_loaded = loaded.clone(); let run_cwd = cwd.to_path_buf(); let run_name = name.to_owned(); let run_extra = extra.to_vec(); let run_control = control.clone(); let force = $force; let handle = std::thread::spawn(move || { let _enter = run_runtime.enter(); execute_graph( &run_loaded, &run_cwd, &run_name, &run_extra, Some(run_control), force, ) }); Some(Generation { handle: Some(handle), control, forced: force, }) }}; } let mut active = start_run!(false); // Generations whose thread finished but whose process groups may // still hold background processes; cleaned up before the next run // and at shutdown, not silently dropped. let mut retired: Vec = Vec::new(); // A forced (cache-bypassing) refresh was launched but not yet seen to // complete uncancelled; the obligation carries over to the next run. let mut force_pending = false; let mut pending = false; let mut rewatch_needed = false; // When the next watcher-repair retry is due, if one is scheduled. // Set only by failed installations, consumed when it fires, and // cleared by any successful one: it cannot bypass a fresh event's // debounce or spin against a permanently broken config. let mut rewatch_retry_at: Option = None; let mut installed_external = external_roots(&loaded.root, &watch_set, &linked); let mut known_dirs = known_directories(&loaded.root); let mut last_event = Instant::now(); // Coverage repair (config symlink target, external topology) is // independent of whether the config parses and of content debounce: // it only (re)installs watches by path. macro_rules! repair_coverage { () => {{ // One installation, against the candidate config paths: a // successful rebuild with stale paths must not be able to // erase a failed candidate repair's obligation. let candidate_config = resolution_paths(&loaded.path); if candidate_config != config_paths || rewatch_needed { let linked: Vec = candidate_config .iter() .chain(&input_paths) .cloned() .collect(); match create_watcher(&loaded, &watch_set, &linked, sender.clone()) { Ok(fresh_watcher) => { _watcher = fresh_watcher; config_paths = candidate_config; installed_external = external_roots(&loaded.root, &watch_set, &linked); rewatch_needed = false; rewatch_retry_at = None; } Err(error) => { rewatch_retry_at = Some(Instant::now() + Duration::from_secs(1)); eprintln!("Task {name}: cannot rewatch: {error}") } } } }}; } loop { if SHUTDOWN.load(Ordering::SeqCst) { if let Some(mut generation) = active.take() && !generation.stop(name) { retired.push(generation); } retired.retain_mut(|generation| !generation.stop(name)); for generation in &retired { generation.control.force_close(); } return Ok(ExitCode::SUCCESS); } // Drain in bounded batches so an event flood cannot starve the // shutdown check above. for _ in 0..1024 { match receiver.try_recv() { Ok(Ok(event)) => { remember_directories(&loaded.root, &mut known_dirs, &event.paths); let relevance = if event.need_rescan() { // Content may have been missed and directory watches may be // gone with it: rebuild the watcher, not just the state. Some(Relevance { evaluate: true, rewatch: true }) } else if matches!(event.kind, notify::EventKind::Access(_)) { None } else { watch_set.classify( WatchContext { root: &loaded.root, config_path: &loaded.path, config_paths: &config_paths, input_paths: &input_paths, known_dirs: &known_dirs, }, WatchEvent { paths: &event.paths, structural: matches!( event.kind, notify::EventKind::Create(_) | notify::EventKind::Remove(_) | notify::EventKind::Modify( notify::event::ModifyKind::Name(_) ) | notify::EventKind::Any | notify::EventKind::Other ), has_directory: event.paths.iter().any(|path| path.is_dir()), is_folder_event: matches!( event.kind, notify::EventKind::Create(notify::event::CreateKind::Folder) | notify::EventKind::Remove( notify::event::RemoveKind::Folder ) ), }, ) }; if let Some(relevance) = relevance { pending |= relevance.evaluate; rewatch_needed |= relevance.rewatch; last_event = Instant::now(); } let gone: &[PathBuf] = match &event.kind { notify::EventKind::Remove(_) => &event.paths, notify::EventKind::Modify(notify::event::ModifyKind::Name( notify::event::RenameMode::From, )) => &event.paths, notify::EventKind::Modify(notify::event::ModifyKind::Name( notify::event::RenameMode::Both, )) => &event.paths[..1.min(event.paths.len())], _ => &[], }; forget_directories(&loaded.root, &mut known_dirs, gone); } Ok(Err(error)) => { eprintln!("Task {name}: filesystem watcher error: {error}"); pending = true; rewatch_needed = true; last_event = Instant::now(); } Err(mpsc::TryRecvError::Empty) => break, Err(mpsc::TryRecvError::Disconnected) => { return Err("Filesystem watcher stopped".into()); } } } if active.as_ref().is_some_and(|generation| { generation .handle .as_ref() .is_some_and(|handle| handle.is_finished()) }) { let mut generation = active.take().unwrap(); let forced = generation.forced; let handle = generation.handle.take().unwrap(); match handle.join() { Ok(Ok(0)) => { if forced { force_pending = false; } } Ok(Ok(code)) => { eprintln!("Task {name} exited with status {code}; watching for changes") } Ok(Err(error)) => eprintln!("{error}"), Err(_) => eprintln!("Task {name}: task thread panicked"), } retired.push(generation); } let retry_due = rewatch_retry_at.is_some_and(|deadline| Instant::now() >= deadline); let debouncing = pending && last_event.elapsed() < DEBOUNCE; if retry_due { rewatch_retry_at = None; } if retry_due && debouncing { // A due repair must not let unfinished edits skip their quiet // period: repair coverage now, evaluate content later. if rewatch_needed { repair_coverage!(); } } else if (pending && !debouncing) || retry_due { pending = false; repair_coverage!(); // Evaluate against a freshly loaded config before touching a // running task: an unchanged state (a touch, a same-content // save) leaves it and the cache exactly as they were. let mut committed = true; match load_file(loaded.path.clone()) { Ok(fresh) => match fresh .graph .watch_set(name, &fresh.root) .map_err(|error| render_task_error(&fresh, error)) { Ok(fresh_set) if !fresh_set.patterns.is_empty() => { if fresh.source != loaded.source || fresh_set != watch_set { // Commit nothing until the new set is really // being watched. let fresh_config = resolution_paths(&fresh.path); let fresh_inputs = resolved_input_paths(&fresh.root, &fresh_set); let fresh_extra: Vec = fresh_config.iter().chain(&fresh_inputs).cloned().collect(); match create_watcher(&fresh, &fresh_set, &fresh_extra, sender.clone()) { Ok(fresh_watcher) => { loaded = Arc::new(fresh); watch_set = fresh_set; config_paths = fresh_config; input_paths = fresh_inputs; _watcher = fresh_watcher; installed_external = external_roots(&loaded.root, &watch_set, &fresh_extra); rewatch_retry_at = None; } Err(error) => { committed = false; rewatch_retry_at = Some(Instant::now() + Duration::from_secs(1)); eprintln!("Task {name}: cannot rewatch: {error}") } } } } Ok(_) => { committed = false; eprintln!( "Task {name:?} cannot be watched: neither it nor its dependencies declare 'files'" ); } Err(error) => { committed = false; eprintln!("{error}") } }, Err(error) => { committed = false; eprintln!("Task {name}: cannot reload avril.conf: {error}") } } if committed { known_dirs = known_directories(&loaded.root); config_paths = resolution_paths(&loaded.path); input_paths = resolved_input_paths(&loaded.root, &watch_set); let linked: Vec = config_paths.iter().chain(&input_paths).cloned().collect(); // Directory topology outside the package (an env file's // parent appearing or being replaced) changes coverage // without changing the config; reconcile it here. let desired_external = external_roots(&loaded.root, &watch_set, &linked); if rewatch_needed || desired_external != installed_external { match create_watcher(&loaded, &watch_set, &linked, sender.clone()) { Ok(fresh_watcher) => { _watcher = fresh_watcher; installed_external = desired_external; rewatch_needed = false; rewatch_retry_at = None; } Err(error) => { rewatch_needed = true; rewatch_retry_at = Some(Instant::now() + Duration::from_secs(1)); eprintln!("Task {name}: cannot rewatch: {error}") } } } let state = WatchState::capture( &loaded.graph, &loaded.root, &loaded.source, name, extra, &watch_set, ); if state != baseline { let force = force_pending || state.metadata_changed(&baseline); baseline = state; if let Some(mut generation) = active.take() { eprintln!("Task {name}: restarting"); if !generation.stop(name) { retired.push(generation); } } retired.retain_mut(|generation| !generation.stop(name)); if SHUTDOWN.load(Ordering::SeqCst) { continue; } if force { force_pending = true; } active = start_run!(force); } } } std::thread::sleep(Duration::from_millis(5)); } } // ------------------------------------------------------------- lib driver struct CliReporter; impl avril_task::Reporter for CliReporter { fn cached(&self, name: &str) { eprintln!("Task {name} (cached)"); } fn unmatched_output(&self, name: &str, pattern: &str) { eprintln!("Task {name}: output pattern {pattern:?} matched nothing"); } fn cache_error(&self, name: &str, message: &str) { eprintln!("Task {name}: cannot record cached outputs: {message}"); } } fn to_graph(tasks: &BTreeMap) -> avril_task::Graph { avril_task::Graph { tasks: tasks .iter() .map(|(name, task)| { ( name.clone(), avril_task::Task { description: task.description.clone(), action: match &task.command { None => avril_task::Action::None, Some(avril_runtime::package_config::TaskCommand::Shell(command)) => { avril_task::Action::Shell(command.clone()) } Some(avril_runtime::package_config::TaskCommand::Argv(argv)) => { avril_task::Action::Argv(argv.clone()) } }, dependencies: task .dependencies .iter() .map(|dependency| dependency.name.clone()) .collect(), parallel: task.parallel, env_file: task.env_file.as_ref().map(|file| PathBuf::from(&file.path)), files: task.files.as_ref().map(|patterns| { patterns .iter() .map(|pattern| pattern.pattern.clone()) .collect() }), output: task .output .iter() .map(|pattern| pattern.pattern.clone()) .collect(), env: task.env.clone(), }, ) }) .collect(), } } /// Renders a lib/task error against the config, using the span of the /// task, dependency, or pattern it concerns when one is known. fn render_task_error(loaded: &Loaded, error: avril_task::Error) -> String { let message = error.to_string(); let span = match &error { avril_task::Error::UnknownTask(name) => loaded .config .tasks .values() .flat_map(|task| &task.dependencies) .find(|dependency| &dependency.name == name) .map(|dependency| dependency.span.clone()) .or_else(|| loaded.config.tasks.get(name).map(|task| task.span.clone())), avril_task::Error::Cycle(cycle) => { let names: Vec<_> = cycle.split(" -> ").collect(); names .iter() .rev() .nth(1) .and_then(|name| loaded.config.tasks.get(*name)) .and_then(|task| { task.dependencies .iter() .find(|dependency| Some(dependency.name.as_str()) == names.last().copied()) }) .map(|dependency| dependency.span.clone()) } avril_task::Error::Env(avril_task::EnvError::Load { path, .. }) => loaded .config .tasks .values() .filter_map(|task| task.env_file.as_ref()) .find(|file| loaded.root.join(&file.path) == *path) .map(|file| file.span.clone()), avril_task::Error::UnmatchedFiles { task, pattern } | avril_task::Error::Pattern { task, pattern, .. } => loaded .config .tasks .get(task) .and_then(|task| task.files.as_ref()) .and_then(|patterns| { patterns .iter() .find(|candidate| &candidate.pattern == pattern) .map(|candidate| candidate.span.clone()) }), _ => None, }; match span { Some(span) => render_error(loaded, span, message), None => message, } } fn execute_graph( loaded: &Loaded, cwd: &Path, name: &str, extra: &[String], control: Option>, force: bool, ) -> TaskResult { let cancelled = avril_task::CancelToken::new(); let mut options = avril_task::RunOptions::new(loaded.root.clone()); options.cache = Some(avril_task::Cache::new( crate::cli::default_cache(cwd).join("task-cache"), )); options.force = force; options.extra_args = extra.to_vec(); options.reporter = Box::new(CliReporter); options.cancelled = cancelled.clone(); let runtime = tokio::runtime::Handle::current(); let mut execution = avril_task::Execution::new(loaded.graph.clone(), name, options) .map_err(|error| render_task_error(loaded, error))?; loop { if control .as_ref() .is_some_and(|control| control.is_cancelled()) { cancelled.cancel(); } match execution .step() .map_err(|error| render_task_error(loaded, error))? { avril_task::Step::Done(report) => return Ok(report.code), avril_task::Step::Ready(batch) => { if batch.is_empty() { return Err(TaskError::Message("Task execution stalled".into())); } let results = std::thread::scope(|scope| { let handles: Vec<_> = batch .iter() .map(|ready| { let control = control.clone(); let runtime = runtime.clone(); scope.spawn(move || run_ready(loaded, ready, cwd, runtime, control)) }) .collect(); handles .into_iter() .map(|handle| handle.join().map_err(|_| "Task thread panicked".to_owned())) .collect::>() }); for (ready, result) in batch.iter().zip(results) { let code = result??; execution.finish(&ready.name, code); } } } } } fn run_ready( loaded: &Loaded, ready: &avril_task::ReadyTask, cwd: &Path, runtime: tokio::runtime::Handle, control: Option>, ) -> TaskResult { let name = &ready.name; let directory = loaded.root.as_path(); let mut options = match &ready.action { avril_task::Action::Shell(command) => { let mut full = command.clone(); for arg in &ready.args { full.push(' '); full.push_str(&shell_quote(arg)); } let (shell, flag) = if cfg!(windows) { ("cmd", "/C") } else { ("sh", "-c") }; let mut options = avril_subprocess::Options::new(shell, vec![flag.into(), full]); #[cfg(unix)] { options.command = shell.into(); } options } avril_task::Action::Argv(argv) => { let Some((program, args)) = argv.split_first() else { return Err(format!("Task {name:?}: command array cannot be empty").into()); }; let mut args: Vec = args.to_vec(); args.extend(ready.args.iter().cloned()); avril_subprocess::Options::new(program, args) } avril_task::Action::Callback(_) => { return Err( format!("Task {name:?}: function tasks require the avril:task module").into(), ); } avril_task::Action::None => return Ok(0), }; eprintln!("Task {name} {}", ready.display()); options.cwd = Some(directory.to_owned()); options.termination.scope = "child".into(); // Like `--env-file`, file values are defaults: the inherited process // environment wins when it already defines a name. options.env = ready .env .iter() .filter(|(key, _)| std::env::var_os(key).is_none()) .map(|(key, value)| (key.clone(), value.clone())) .collect(); let mut inherited: BTreeMap<_, _> = std::env::vars_os().collect(); if !inherited.contains_key(std::ffi::OsStr::new("INIT_CWD")) && !ready.env.contains_key("INIT_CWD") { inherited.insert("INIT_CWD".into(), cwd.as_os_str().to_owned()); } #[cfg(unix)] if matches!(ready.action, avril_task::Action::Shell(_)) && !inherited.contains_key(std::ffi::OsStr::new("PATH")) && !ready.env.contains_key("PATH") { options.command = "/bin/sh".into(); } if SHUTDOWN.load(Ordering::SeqCst) && let Some(control) = &control { control.cancel(); } let process = avril_subprocess::Launcher { runtime: runtime.clone(), caps: avril_caps::Capabilities::new(avril_caps::Initial::new(false), directory.to_owned()), owners: Vec::new(), } .spawn(options, inherited, control.as_deref(), |_, _, _| { unreachable!("task stdio is inherited") }) .map_err(|error| format!("Task {name:?}: {error}"))?; let status = runtime .block_on(process.wait(true, None)) .map_err(|error| format!("Task {name:?}: {error}"))? .ok_or_else(|| format!("Task {name:?}: Task exited without a status"))?; Ok(status.code.unwrap_or(1) as u8) } fn render_error(loaded: &Loaded, span: Span, message: String) -> String { ConfigError { kind: ErrorKind::Syntax, message, span, previous: None, } .render(&loaded.path.to_string_lossy(), &loaded.source) }