diff --git a/lib/apps/fabro-cli/Cargo.toml b/lib/apps/fabro-cli/Cargo.toml index cb86d79c4..15ae0c779 100644 --- a/lib/apps/fabro-cli/Cargo.toml +++ b/lib/apps/fabro-cli/Cargo.toml @@ -98,7 +98,7 @@ object_store.workspace = true bytes.workspace = true tokio-util.workspace = true libc = "0.2" -nix = { version = "0.30", features = ["fs"] } +nix = { version = "0.30", features = ["fs", "signal"] } [target.'cfg(target_os = "macos")'.dependencies] core-foundation = { version = "0.9", optional = true } diff --git a/lib/apps/fabro-cli/src/commands/run/command.rs b/lib/apps/fabro-cli/src/commands/run/command.rs index 2ec25f636..2dfdd235a 100644 --- a/lib/apps/fabro-cli/src/commands/run/command.rs +++ b/lib/apps/fabro-cli/src/commands/run/command.rs @@ -1,6 +1,7 @@ use anyhow::Result; use fabro_util::terminal::Styles; +use super::remote_workflow::Interruption; use crate::args::RunArgs; use crate::command_context::CommandContext; use crate::shared::print_json_pretty; @@ -15,16 +16,32 @@ pub(crate) async fn execute(mut args: RunArgs, base_ctx: &CommandContext) -> Res let quiet = args.detach; let prevent_idle_sleep = ctx.user_settings().cli.exec.prevent_idle_sleep; - let created_run = Box::pin(super::create::create_run(&ctx, &args, styles)).await?; + // Ctrl-C stays owned here through start; `attach` installs its own listener. + let interruption = Interruption::for_run_args(&args); + let (created_run, client) = interruption + .guard(async { + let created_run = Box::pin(super::create::create_run( + &ctx, + &args, + styles, + &interruption, + )) + .await?; - if !quiet { - fabro_util::printerr!( - printer, - " {} {}", - styles.dim.apply_to("Run:"), - styles.dim.apply_to(&created_run.run_id), - ); - } + if !quiet { + fabro_util::printerr!( + printer, + " {} {}", + styles.dim.apply_to("Run:"), + styles.dim.apply_to(&created_run.run_id), + ); + } + + let client = ctx.server().await?; + super::start::start_run_with_client(&client, &created_run.run_id, false).await?; + Ok((created_run, client)) + }) + .await?; #[cfg(feature = "sleep_inhibitor")] let _sleep_guard = sleep_inhibitor::guard(prevent_idle_sleep); @@ -32,9 +49,6 @@ pub(crate) async fn execute(mut args: RunArgs, base_ctx: &CommandContext) -> Res #[cfg(not(feature = "sleep_inhibitor"))] let _ = prevent_idle_sleep; - let client = ctx.server().await?; - super::start::start_run_with_client(&client, &created_run.run_id, false).await?; - let json = ctx.json_output(); if args.detach { if json { diff --git a/lib/apps/fabro-cli/src/commands/run/create.rs b/lib/apps/fabro-cli/src/commands/run/create.rs index c7145ef73..ed69e165c 100644 --- a/lib/apps/fabro-cli/src/commands/run/create.rs +++ b/lib/apps/fabro-cli/src/commands/run/create.rs @@ -7,6 +7,7 @@ use fabro_types::{RunId, RunIntent}; use fabro_util::terminal::Styles; use super::overrides::prepare_intent_overrides; +use super::remote_workflow::Interruption; use super::resolution::ResolvedWorkflow; use super::selection::WorkflowSelection; use super::{resolution, selection}; @@ -23,10 +24,14 @@ pub(crate) struct CreatedRun { /// run from an immutable workflow intent, leaving it in the submitted state. /// /// This does NOT start the workflow — starting is a separate request. +/// +/// Native Git acquisition runs under `interruption`; the caller guards this +/// call (and any later phase before `attach`) with the same handle. pub(crate) async fn create_run( ctx: &CommandContext, args: &RunArgs, styles: &Styles, + interruption: &Interruption, ) -> anyhow::Result { let (workflow_selection, target_selection) = selection::parse(args)?; let canonical_cwd = ctx.cwd().canonicalize().with_context(|| { @@ -41,6 +46,7 @@ pub(crate) async fn create_run( &workflow_selection, &canonical_cwd, Some(&user_workflows_root), + interruption, ) }; // Preserve local lookup diagnostics before contacting the server. Remote @@ -94,6 +100,7 @@ pub(crate) async fn create_run( &target_selection, environment.settings.provider, &canonical_cwd, + interruption, ) .await?; if dirty_worktree { diff --git a/lib/apps/fabro-cli/src/commands/run/mod.rs b/lib/apps/fabro-cli/src/commands/run/mod.rs index 6b9177673..b904549b4 100644 --- a/lib/apps/fabro-cli/src/commands/run/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/mod.rs @@ -47,7 +47,15 @@ pub(crate) async fn dispatch( RunCommands::Create(args) => { let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr())); let ctx = base_ctx.with_target(&args.target)?; - let created_run = Box::pin(create::create_run(&ctx, &args, styles)).await?; + let interruption = remote_workflow::Interruption::for_run_args(&args); + let created_run = interruption + .guard(Box::pin(create::create_run( + &ctx, + &args, + styles, + &interruption, + ))) + .await?; if ctx.json_output() { print_json_pretty(&serde_json::json!({ "run_id": created_run.run_id }))?; } else { diff --git a/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs b/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs index 6a7944c92..015353490 100644 --- a/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs +++ b/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs @@ -11,8 +11,10 @@ use tokio::io::{AsyncRead, AsyncReadExt as _}; use tokio::process::Command; use tokio::{fs, signal as tokio_signal, task, time}; use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; use super::selection::RemoteWorkflowRevision; +use crate::args::RunArgs; const OUTPUT_LIMIT: usize = 64 * 1024; @@ -457,27 +459,76 @@ fn resolve_name(records: &str, branch: &str, tag: &str) -> anyhow::Result( - work: impl FnOnce(CancellationToken) -> TFuture + Send + 'static, -) -> anyhow::Result -where - TFuture: Future> + Send + 'static, -{ - let cancel = CancellationToken::new(); - let _cancel_on_drop = cancel.clone().drop_guard(); - let mut task = tokio::spawn(work(cancel.clone())); - tokio::select! { - result = &mut task => result.context("local Git task failed")?, - signal = tokio_signal::ctrl_c() => { - cancel.cancel(); - let _result = task.await.context("local Git cleanup task failed")?; - signal.context("failed to listen for interruption")?; - Err(RemoteWorkflowError::Cancelled.into()) +/// Cooperative Ctrl-C handling for commands that acquire sources with native +/// Git. +/// +/// Tokio's Ctrl-C listener permanently replaces the default SIGINT disposition +/// for the process, so it is installed only when native Git is in play, and the +/// command keeps it armed for every phase up to the point where `attach` +/// installs its own listener or the process exits. Interruption cancels owned +/// Git tasks and waits for their cleanup; an in-progress blocking collection +/// must finish first. +#[derive(Clone)] +pub(crate) struct Interruption { + cancel: CancellationToken, + tasks: TaskTracker, + listens: bool, +} + +impl Interruption { + /// `native_git` reports whether any selection runs native Git. Without it + /// the default SIGINT disposition is left untouched and `guard` is a + /// pass-through. + pub(crate) fn new(native_git: bool) -> Self { + Self { + cancel: CancellationToken::new(), + tasks: TaskTracker::new(), + listens: native_git, } } + + pub(crate) fn for_run_args(args: &RunArgs) -> Self { + Self::new(args.workflow_git.is_some() || args.target_git.is_some()) + } + + /// Run `work` to completion, or until Ctrl-C cancels it and every owned + /// Git task has cleaned up. + pub(crate) async fn guard( + &self, + work: impl Future>, + ) -> anyhow::Result { + if !self.listens { + return work.await; + } + tokio::select! { + result = work => result, + signal = tokio_signal::ctrl_c() => { + self.cancel.cancel(); + self.tasks.close(); + self.tasks.wait().await; + signal.context("failed to listen for interruption")?; + Err(RemoteWorkflowError::Cancelled.into()) + } + } + } + + /// The task owns its child processes and temporary checkout. Dropping the + /// waiter requests cooperative cleanup, not task abortion; the task keeps + /// running until its Git children are reaped and its files are removed. + pub(super) async fn owned( + &self, + work: impl FnOnce(CancellationToken) -> TFuture + Send + 'static, + ) -> anyhow::Result + where + TFuture: Future> + Send + 'static, + { + let cancel = self.cancel.child_token(); + let _cancel_on_drop = cancel.clone().drop_guard(); + self.tasks + .spawn(work(cancel.clone())) + .await + .context("local Git task failed")? + } } #[cfg(test)] @@ -487,7 +538,10 @@ where )] mod tests { use std::os::unix::fs::PermissionsExt as _; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use nix::sys::signal::{self, Signal}; use nix::sys::stat::Mode; use nix::unistd; @@ -706,16 +760,20 @@ mod tests { let (fake, git) = fake_git("printf '%s' $$ > pid; exec /bin/sleep 60"); let checkout = tempfile::tempdir().unwrap(); let path = checkout.path().to_path_buf(); - let worker = tokio::spawn(owned(move |cancel| async move { - git.collect_checkout( - "acme/workflows".parse().unwrap(), - "review".into(), - "1111111111111111111111111111111111111111".into(), - checkout, - cancel, - ) - .await - })); + let worker = tokio::spawn(async move { + Interruption::new(true) + .owned(move |cancel| async move { + git.collect_checkout( + "acme/workflows".parse().unwrap(), + "review".into(), + "1111111111111111111111111111111111111111".into(), + checkout, + cancel, + ) + .await + }) + .await + }); time::timeout(Duration::from_secs(5), async { while !path.join("pid").exists() { time::sleep(Duration::from_millis(10)).await; @@ -740,6 +798,45 @@ mod tests { drop(fake); } + /// Each nextest test runs in its own process, so raising SIGINT here only + /// reaches the listener `guard` installed before polling `work`. + #[tokio::test] + async fn remote_workflow_guard_cancels_owned_tasks_and_waits_for_cleanup_on_ctrl_c() { + let interruption = Interruption::new(true); + let cleaned = Arc::new(AtomicBool::new(false)); + let result: anyhow::Result<()> = interruption + .guard({ + let interruption = interruption.clone(); + let cleaned = Arc::clone(&cleaned); + async move { + interruption + .owned(move |cancel| async move { + signal::raise(Signal::SIGINT).unwrap(); + cancel.cancelled().await; + // Cleanup after cancellation must finish before + // `guard` reports the interruption. + time::sleep(Duration::from_millis(200)).await; + cleaned.store(true, Ordering::SeqCst); + Ok(()) + }) + .await + } + }) + .await; + assert!(matches!( + result.unwrap_err().downcast_ref::(), + Some(RemoteWorkflowError::Cancelled) + )); + assert!(cleaned.load(Ordering::SeqCst)); + assert_eq!( + Interruption::new(false) + .guard(async { Ok::<_, anyhow::Error>(7) }) + .await + .unwrap(), + 7 + ); + } + #[tokio::test] async fn remote_workflow_resolves_exact_default_branches_tags_and_commits() { let fixture = Fixture::new(); diff --git a/lib/apps/fabro-cli/src/commands/run/resolution.rs b/lib/apps/fabro-cli/src/commands/run/resolution.rs index a6fa1daa3..556e92d1d 100644 --- a/lib/apps/fabro-cli/src/commands/run/resolution.rs +++ b/lib/apps/fabro-cli/src/commands/run/resolution.rs @@ -6,7 +6,7 @@ use fabro_types::settings::run::EnvironmentProvider; use fabro_types::{DirtyStatus, RunTarget}; use tokio::task; -use super::remote_workflow::{self, NativeGit}; +use super::remote_workflow::{Interruption, NativeGit}; use super::selection::{TargetSelection, WorkflowSelection}; /// Owns the canonical collector result without copying its contents. Local @@ -29,6 +29,7 @@ pub(super) async fn workflow( selection: &WorkflowSelection, cwd: &Path, user_workflows: Option<&Path>, + interruption: &Interruption, ) -> anyhow::Result { match selection { WorkflowSelection::Local(path) => { @@ -57,10 +58,11 @@ pub(super) async fn workflow( let git = NativeGit::new(); let (repository, selector, revision) = (repository.clone(), selector.clone(), revision.clone()); - let closure = remote_workflow::owned(move |cancel| async move { - git.collect(repository, selector, revision, cancel).await - }) - .await?; + let closure = interruption + .owned(move |cancel| async move { + git.collect(repository, selector, revision, cancel).await + }) + .await?; Ok(ResolvedWorkflow::Git(closure)) } } @@ -70,6 +72,7 @@ pub(super) async fn target( selection: &TargetSelection, provider: EnvironmentProvider, cwd: &Path, + interruption: &Interruption, ) -> anyhow::Result<(RunTarget, bool)> { let path = match selection { TargetSelection::Path(path) => cwd @@ -82,10 +85,11 @@ pub(super) async fn target( } let git = NativeGit::new(); let (repository, branch) = (repository.clone(), branch.clone()); - let target = remote_workflow::owned(move |cancel| async move { - git.resolve_target(repository, branch, &cancel).await - }) - .await?; + let target = interruption + .owned(move |cancel| async move { + git.resolve_target(repository, branch, &cancel).await + }) + .await?; // Canonical admission retains ownership of provider capabilities. return Ok((RunTarget::Git(target), false)); } @@ -199,8 +203,9 @@ mod tests { write_workflow(&root, ".fabro/workflows/review"); std::fs::create_dir(root.join("target")).unwrap(); let selected = TargetSelection::Path("target".into()); + let interruption = Interruption::new(false); assert_eq!( - target(&selected, EnvironmentProvider::Local, &root) + target(&selected, EnvironmentProvider::Local, &root, &interruption) .await .unwrap() .0, @@ -210,7 +215,10 @@ mod tests { ); for provider in [EnvironmentProvider::Docker, EnvironmentProvider::Daytona] { assert_eq!( - target(&selected, provider, &root).await.unwrap().0, + target(&selected, provider, &root, &interruption) + .await + .unwrap() + .0, RunTarget::None {} ); } @@ -218,7 +226,8 @@ mod tests { target( &TargetSelection::Path(".".into()), EnvironmentProvider::Local, - &root + &root, + &interruption ) .await .unwrap() @@ -234,7 +243,8 @@ mod tests { branch: None, }, EnvironmentProvider::Local, - &root + &root, + &interruption ) .await .unwrap_err() @@ -246,7 +256,8 @@ mod tests { target( &TargetSelection::Path(path.into()), EnvironmentProvider::Local, - &root + &root, + &interruption ) .await .is_err(), @@ -267,13 +278,16 @@ mod tests { std::fs::write(project.join(".fabro/project.toml"), "_version = 1\n").unwrap(); git2::Repository::init(&checkout).unwrap(); let selected = WorkflowSelection::Local("review".into()); + let interruption = Interruption::new(false); for (cwd, expected_root) in [ (checkout.as_path(), checkout.as_path()), (project.as_path(), project.as_path()), (root.path(), user.as_path()), ] { let ResolvedWorkflow::Local(package) = - workflow(&selected, cwd, Some(&user)).await.unwrap() + workflow(&selected, cwd, Some(&user), &interruption) + .await + .unwrap() else { panic!("local package"); }; @@ -283,7 +297,8 @@ mod tests { workflow( &WorkflowSelection::Local("missing.toml".into()), &checkout, - Some(&user) + Some(&user), + &interruption ) .await .is_err()