scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-06 02:48:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Keep Ctrl-C owned for the whole native Git command phase

Browse source 
owned() polled tokio's ctrl_c() only for the duration of one Git
acquisition. On Unix that listener permanently replaces the default
SIGINT disposition, so once acquisition finished nothing handled Ctrl-C
and fabro create, fabro run --detach, and fabro run before attach
silently ignored it while waiting on the server. Introduce an
Interruption handle that the command entry points create from the run
arguments: it installs a listener only when --workflow-git or
--target-git is in play, guards create (and start for fabro run) as one
phase, and tracks owned Git tasks so interruption waits for their
cleanup before returning. attach keeps installing its own listener.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Scott Werner 2026-09-11 09:46:30 -06:00
parent 39694ec5e8
commit f15972f8f2
6 changed files with 200 additions and 59 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/apps/fabro-cli/Cargo.toml
View file

@ -100,7 +100,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 }

38
lib/apps/fabro-cli/src/commands/run/command.rs
View file

@ -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 {

7
lib/apps/fabro-cli/src/commands/run/create.rs
View file

@ -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<CreatedRun> {
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 {

10
lib/apps/fabro-cli/src/commands/run/mod.rs
View file

@ -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 {

155
lib/apps/fabro-cli/src/commands/run/remote_workflow.rs
View file

@ -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<String
}
}
/// The task owns its child processes and temporary checkout. Dropping the
/// waiter requests cooperative cleanup, not task abortion. Ctrl-C waits for
/// cleanup; an in-progress blocking collection must finish first.
pub(super) async fn owned<T: Send + 'static, TFuture>(
work: impl FnOnce(CancellationToken) -> TFuture + Send + 'static,
) -> anyhow::Result<T>
where
TFuture: Future<Output = anyhow::Result<T>> + 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<T>(
&self,
work: impl Future<Output = anyhow::Result<T>>,
) -> anyhow::Result<T> {
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<T: Send + 'static, TFuture>(
&self,
work: impl FnOnce(CancellationToken) -> TFuture + Send + 'static,
) -> anyhow::Result<T>
where
TFuture: Future<Output = anyhow::Result<T>> + 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::<RemoteWorkflowError>(),
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();

47
lib/apps/fabro-cli/src/commands/run/resolution.rs
View file

@ -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<ResolvedWorkflow> {
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()