diff --git a/.github/workflows/build_and_unit_test.yml b/.github/workflows/build_and_unit_test.yml index 716ffcd4..b788c925 100644 --- a/.github/workflows/build_and_unit_test.yml +++ b/.github/workflows/build_and_unit_test.yml @@ -10,9 +10,13 @@ on: - labeled paths: - src/** + - Cargo.toml + - Cargo.lock push: paths: - src/** + - Cargo.toml + - Cargo.lock branches: - master - unstable @@ -50,3 +54,37 @@ jobs: printf '\e[1;33m\tPLEASE, SOLVE THEM LOCALLY W/ `cargo test`\e[0m\n' printf '\e[1;33m\t==========================================\n\e[0m' exit 1 + + # The kw actor and the process infrastructure it uses are unix-only + # (#[cfg(unix)]). This job exists so a cfg leak — non-gated code + # referencing them — fails mechanically instead of in review. + check-non-unix: + runs-on: ubuntu-latest + timeout-minutes: 4 + if: '!github.event.pull_request.draft' + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.sha }} + + - name: Update rustup and install rustc and cargo + shell: bash + run: | + rustup update + rustup install stable + + # rustls/ring needs a C toolchain for the target; windows-gnu builds + # with mingw on Ubuntu, unlike the MSVC target. + - name: Install the mingw cross toolchain + shell: bash + run: | + sudo apt-get update + sudo apt-get install -y gcc-mingw-w64-x86-64 + + - name: Check non-unix compilation + shell: bash + env: + CARGO_TARGET_X86_64_PC_WINDOWS_GNU_LINKER: x86_64-w64-mingw32-gcc + run: | + rustup target add x86_64-pc-windows-gnu + cargo check --target x86_64-pc-windows-gnu --verbose diff --git a/Cargo.toml b/Cargo.toml index d9ffd029..008a7053 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,7 @@ chrono = "0.4.41" ansi-to-tui = "7.0.0" which = "8.0.0" ureq = { version = "3.0.12", features = ["rustls"] } -tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync", "process", "time"] } +tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync", "process", "time", "test-util"] } async-trait = "0.1" nix = { version = "0.31", features = ["signal"] } base64 = "0.22" diff --git a/src/app/actor.rs b/src/app/actor.rs index 497666a9..c0ee9e22 100644 --- a/src/app/actor.rs +++ b/src/app/actor.rs @@ -234,6 +234,7 @@ mod tests { fs: Box::new(MockFileSystemTrait::new()), config: dummy_config_handle(), kw_history: Arc::new(MockKwHistoryStore::new()), + kw: None, }, } } @@ -336,6 +337,7 @@ mod tests { lore_api.clone(), render.clone(), Arc::new(MockKwHistoryStore::new()), + None, ) .expect("App::new must succeed"); diff --git a/src/app/integration_tests/helpers/app_harness.rs b/src/app/integration_tests/helpers/app_harness.rs index ba7531fa..6969eac1 100644 --- a/src/app/integration_tests/helpers/app_harness.rs +++ b/src/app/integration_tests/helpers/app_harness.rs @@ -56,6 +56,7 @@ pub(crate) fn app_with_bootstrap_and_handles( lore_api, render, Arc::new(MockKwHistoryStore::new()), + None, ) .expect("minimal app should build") } diff --git a/src/app/integration_tests/patchset_actions.rs b/src/app/integration_tests/patchset_actions.rs index 1eadbc7a..72b0d780 100644 --- a/src/app/integration_tests/patchset_actions.rs +++ b/src/app/integration_tests/patchset_actions.rs @@ -1,5 +1,6 @@ use std::{ collections::{HashMap, HashSet, VecDeque}, + path::PathBuf, sync::{Arc, Mutex}, }; @@ -16,10 +17,12 @@ use crate::{ }, config::{ConfigSnapshot, ConfigState}, infrastructure::{ + env::MockEnvTrait, file_system::{FileSystemError, MockFileSystemTrait}, + process::FakeProcess, shell::{MockShellTrait, ShellCommand, ShellOutput}, }, - kw::history::MockKwHistoryStore, + kw::{actor::KwActor, history::MockKwHistoryStore}, lore::application::{ cache::BootstrapLoreData, handle::LoreApiHandle, messages::LoreApiMessage, }, @@ -64,6 +67,7 @@ async fn apply_success_sets_success_popup_and_resets_apply_action() { "Current branch: 'patchset-", ], ); + shutdown_kw(&app).await; } #[tokio::test] @@ -94,11 +98,14 @@ async fn apply_success_switches_back_when_stay_disabled() { "Patchset Apply Success", &["Current branch: 'feature'"], ); - let calls = calls.lock().unwrap(); - assert_eq!( - command(&["git", "-C", KERNEL_TREE_PATH, "switch", "feature"]), - calls[6] - ); + { + let calls = calls.lock().unwrap(); + assert_eq!( + command(&["git", "-C", KERNEL_TREE_PATH, "switch", "feature"]), + calls[6] + ); + } + shutdown_kw(&app).await; } #[tokio::test] @@ -123,11 +130,14 @@ async fn apply_failure_sets_failure_popup_and_resets_apply_action() { "Patchset Apply Fail", &["`git am` failed", "feature", "apply failed"], ); - let calls = calls.lock().unwrap(); - assert_eq!( - command(&["git", "-C", KERNEL_TREE_PATH, "am", "--abort"]), - calls[6] - ); + { + let calls = calls.lock().unwrap(); + assert_eq!( + command(&["git", "-C", KERNEL_TREE_PATH, "am", "--abort"]), + calls[6] + ); + } + shutdown_kw(&app).await; } #[tokio::test] @@ -169,6 +179,7 @@ async fn apply_success_records_apply_history() { "Patchset Apply Success", &["applied successfully"], ); + shutdown_kw(&app).await; } #[tokio::test] @@ -208,6 +219,7 @@ async fn apply_success_with_history_write_failure_keeps_success_popup() { "inspect or delete that file", ], ); + shutdown_kw(&app).await; } #[tokio::test] @@ -236,6 +248,7 @@ async fn apply_failure_does_not_record_history() { app.consolidate_patchset_actions().await.unwrap(); assert_info_popup_contains(app.state.popup.as_ref(), "Patchset Apply Fail", &[]); + shutdown_kw(&app).await; } #[tokio::test] @@ -268,6 +281,7 @@ async fn reviewed_reply_success_records_persists_and_resets_reply_action() { .clone() .expect("reviewed state should be persisted"); assert_eq!(HashSet::from([0]), saved[&message_id]); + shutdown_kw(&app).await; } #[tokio::test] @@ -297,6 +311,7 @@ async fn reviewed_reply_failure_does_not_record_failed_index() { .clone() .expect("reviewed state should be persisted"); assert!(saved[&message_id].is_empty()); + shutdown_kw(&app).await; } fn app_with_apply_details(fs: MockFileSystemTrait, shell: MockShellTrait) -> App { @@ -310,6 +325,14 @@ fn app_with_apply_details(fs: MockFileSystemTrait, shell: MockShellTrait) -> App ) } +/// Shuts down the KwActor the app was wired with, instead of relying on +/// the test runtime aborting it at drop. +async fn shutdown_kw(app: &App) { + if let Some(kw) = &app.services.kw { + kw.shutdown().await; + } +} + fn app_with_reviewed_reply_details(shell: MockShellTrait, lore_api: LoreApiHandle) -> App { app_with_details( MockFileSystemTrait::new(), @@ -329,6 +352,16 @@ fn app_with_details( config: ConfigSnapshot, kw_history: MockKwHistoryStore, ) -> App { + // Apply history is recorded through the real actor wrapping the mock + // store, mirroring production wiring. + let kw = KwActor::spawn( + Arc::new(kw_history), + Arc::new(FakeProcess::new()), + Arc::new(MockShellTrait::new()), + Arc::new(MockFileSystemTrait::new()), + Arc::new(MockEnvTrait::new()), + PathBuf::from("/tmp/patch-hub-test-kw-logs"), + ); let mut app = App::new( config, dummy_config_handle(), @@ -341,7 +374,8 @@ fn app_with_details( Box::new(shell), lore_api, dummy_render_handle(), - Arc::new(kw_history), + Arc::new(MockKwHistoryStore::new()), + Some(kw), ) .expect("app should build"); diff --git a/src/app/mod.rs b/src/app/mod.rs index 58b418c1..c10819cb 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -47,7 +47,10 @@ use crate::{ file_system::FileSystemTrait, monitoring::logging::garbage_collector::collect_garbage, shell::ShellTrait, }, - kw::history::{KwApplyRecord, KwHistoryStore}, + kw::{ + handle::KwHandle, + history::{KwApplyRecord, KwHistoryStore}, + }, lore::{ application::{ cache::{BootstrapLoreData, CacheMode}, @@ -76,8 +79,12 @@ pub struct AppServices { pub shell: Box, pub fs: Box, pub config: ConfigHandle, - /// Shared with KwActor once it exists (the actor adopts the same store). + /// Direct access to the store, for the non-unix fallback below. pub kw_history: Arc, + /// `None` on non-unix builds, where ProcessTrait (and thus KwActor) + /// does not exist; apply-history writes then go to `kw_history` + /// directly, as they did before the actor landed. + pub kw: Option, } /// Result type signalling whether a patchset was successfully loaded. @@ -110,6 +117,7 @@ impl App { lore_api: LoreApiHandle, render: RenderHandle, kw_history: Arc, + kw: Option, ) -> Result { event!(Level::INFO, "patch-hub started"); collect_garbage(&config); @@ -147,6 +155,7 @@ impl App { fs, config: config_handle, kw_history, + kw, }, }) } @@ -310,7 +319,7 @@ impl App { debug!("consolidating patchset actions"); self.sync_patchset_bookmark().await?; self.execute_reviewed_reply().await?; - self.execute_apply_patchset(); + self.execute_apply_patchset().await; debug!("patchset actions consolidated"); Ok(()) } @@ -408,7 +417,7 @@ impl App { Ok(()) } - fn execute_apply_patchset(&mut self) { + async fn execute_apply_patchset(&mut self) { let details = self .state .lore @@ -437,18 +446,33 @@ impl App { applied.message ) } - // The git apply itself succeeded; a history-write - // failure must not turn it into a reported failure. - Some(record) => match self.services.kw_history.record_apply(record) { - Ok(()) => applied.message, - Err(e) => { - warn!(error = %e, "failed to record kw apply history"); - format!( - "{}\n\nWarning: the apply was not recorded in the kw history: {e}\nIf this warning keeps appearing, inspect or delete that file.", - applied.message - ) + Some(record) => { + // History writes go through KwActor so apply + // recording serializes with job state; without an + // actor (non-unix), write the store directly. + let recorded = match &self.services.kw { + Some(kw) => { + kw.record_apply(record).await.map_err(|e| e.to_string()) + } + None => self + .services + .kw_history + .record_apply(record) + .map_err(|e| e.to_string()), + }; + // The git apply itself succeeded; a history-write + // failure must not turn it into a reported failure. + match recorded { + Ok(()) => applied.message, + Err(e) => { + warn!(error = %e, "failed to record kw apply history"); + format!( + "{}\n\nWarning: the apply was not recorded in the kw history: {e}\nIf this warning keeps appearing, inspect or delete that file.", + applied.message + ) + } } - }, + } }; popup::AppPopup::info("Patchset Apply Success", popup_body) } diff --git a/src/infrastructure/process/fake.rs b/src/infrastructure/process/fake.rs index 4d6c1af2..1cb84ebe 100644 --- a/src/infrastructure/process/fake.rs +++ b/src/infrastructure/process/fake.rs @@ -41,6 +41,11 @@ pub struct FakeControl { struct FakeState { raw_status: Option, killed: bool, + force_killed: bool, + /// When set, `kill()` (SIGTERM) is recorded but the process keeps + /// "running": the model for a process group that ignores SIGTERM, so + /// tests can exercise the SIGKILL escalation. + ignores_sigterm: bool, } impl FakeControl { @@ -66,6 +71,10 @@ impl FakeControl { pub fn was_killed(&self) -> bool { self.state.lock().unwrap().killed } + + pub fn was_force_killed(&self) -> bool { + self.state.lock().unwrap().force_killed + } } struct FakeSpawn { @@ -80,6 +89,7 @@ struct FakeSpawn { pub struct FakeProcess { spawns: Mutex>, refuse_spawns: AtomicBool, + ignore_sigterm: AtomicBool, } impl FakeProcess { @@ -94,6 +104,12 @@ impl FakeProcess { self.refuse_spawns.store(refuse, Ordering::Relaxed); } + /// Make subsequently spawned processes ignore `kill()` (SIGTERM): the + /// kill is recorded but they keep "running" until `force_kill()`. + pub fn ignore_sigterm(&self, ignore: bool) { + self.ignore_sigterm.store(ignore, Ordering::Relaxed); + } + pub fn spawned(&self) -> Vec { self.spawns .lock() @@ -135,6 +151,8 @@ impl ProcessTrait for FakeProcess { state: Mutex::new(FakeState { raw_status: None, killed: false, + force_killed: false, + ignores_sigterm: self.ignore_sigterm.load(Ordering::Relaxed), }), notify: Notify::new(), log_path: log_path.to_path_buf(), @@ -177,7 +195,21 @@ impl RunningProcess for FakeRunningProcess { // process is a successful no-op, not a kill. if state.raw_status.is_none() { state.killed = true; - state.raw_status = Some(Signal::SIGTERM as i32); + if !state.ignores_sigterm { + state.raw_status = Some(Signal::SIGTERM as i32); + drop(state); + self.control.notify.notify_one(); + } + } + Ok(()) + } + + fn force_kill(&mut self) -> Result<(), ProcessError> { + let mut state = self.control.state.lock().unwrap(); + // SIGKILL cannot be ignored: even a SIGTERM-stubborn process dies. + if state.raw_status.is_none() { + state.force_killed = true; + state.raw_status = Some(Signal::SIGKILL as i32); drop(state); self.control.notify.notify_one(); } diff --git a/src/infrastructure/process/mod.rs b/src/infrastructure/process/mod.rs index b97a88e2..f1e6d3ad 100644 --- a/src/infrastructure/process/mod.rs +++ b/src/infrastructure/process/mod.rs @@ -33,7 +33,6 @@ use tokio::process::{Child, Command}; use crate::infrastructure::shell::ShellCommand; -#[allow(dead_code)] pub struct OsProcess; impl ProcessTrait for OsProcess { @@ -88,7 +87,19 @@ impl RunningProcess for OsRunningProcess { } fn kill(&mut self) -> Result<(), ProcessError> { - match killpg(self.pgid, Signal::SIGTERM) { + self.signal_group(Signal::SIGTERM) + } + + fn force_kill(&mut self) -> Result<(), ProcessError> { + self.signal_group(Signal::SIGKILL) + } +} + +impl OsRunningProcess { + /// ESRCH tolerance: signaling an already-gone group is a successful + /// no-op, not an error. + fn signal_group(&self, signal: Signal) -> Result<(), ProcessError> { + match killpg(self.pgid, signal) { Ok(()) | Err(Errno::ESRCH) => Ok(()), Err(errno) => Err(ProcessError::IoError(io::Error::from(errno))), } diff --git a/src/infrastructure/process/trait.rs b/src/infrastructure/process/trait.rs index c9204c63..5ce80709 100644 --- a/src/infrastructure/process/trait.rs +++ b/src/infrastructure/process/trait.rs @@ -12,7 +12,6 @@ pub enum ProcessError { IoError(#[from] io::Error), } -#[allow(dead_code)] #[automock] pub trait ProcessTrait: Send + Sync { /// Spawn `cmd` with `cwd` as its working directory, redirecting stdout and @@ -29,7 +28,6 @@ pub trait ProcessTrait: Send + Sync { // `automock` must stay the outermost attribute: with `async_trait` listed // first, the generated mock's async methods return an unusable type. -#[allow(dead_code)] #[automock] #[async_trait] pub trait RunningProcess: Send { @@ -47,4 +45,8 @@ pub trait RunningProcess: Send { /// to cancel should `tokio::select!` between `wait()` and its cancel /// signal, then call `kill()`. fn kill(&mut self) -> Result<(), ProcessError>; + + /// Send SIGKILL to the whole process group. The escalation rung for a + /// group that ignored `kill()`'s SIGTERM; same ESRCH tolerance. + fn force_kill(&mut self) -> Result<(), ProcessError>; } diff --git a/src/kw/actor.rs b/src/kw/actor.rs new file mode 100644 index 00000000..3907f709 --- /dev/null +++ b/src/kw/actor.rs @@ -0,0 +1,1119 @@ +//! kw actor: owns kw build/deploy job state and the kw history store. +//! +//! All kw operations go through [`KwHandle`](crate::kw::handle::KwHandle) as +//! typed request/reply messages. `Start*` messages reply immediately with an +//! accept/refuse verdict — a job accepted by the actor keeps running after +//! the caller has been answered, so the AppActor loop never blocks on a +//! kernel build. The actor composes the readiness probes from +//! [`crate::kw::readiness`] and records applies through the shared +//! [`KwHistoryStore`](crate::kw::history::KwHistoryStore). +//! +//! `create_dir_all`, the HEAD-probe `git` call, and history writes run +//! inline on the actor task. + +use std::{ops::ControlFlow, path::PathBuf, process::ExitStatus, sync::Arc, time::Duration}; + +use tokio::{ + spawn, + sync::{mpsc, oneshot, watch}, +}; + +use crate::{ + config::KernelTree, + infrastructure::{ + env::EnvTrait, + file_system::FileSystemTrait, + process::{ProcessError, ProcessTrait, RunningProcess}, + shell::{ShellCommand, ShellTrait}, + }, + kw::{ + errors::{KwError, KwStartError}, + handle::KwHandle, + history::KwHistoryStore, + messages::{KwMessage, StartRequest}, + readiness::{self, KwReadiness}, + status::{KwJobKind, KwJobStatus, KwPhase, KwStatusSnapshot}, + }, +}; + +pub const DEFAULT_KW_CHANNEL_SIZE: usize = 16; + +/// Grace periods for the cancel escalation ladder: SIGTERM the process +/// group, wait, SIGKILL, wait, then give up. Giving up still terminates the +/// job from the actor's point of view — a group that ignores both signals +/// must not wedge the actor into refusing every later Start with +/// JobAlreadyRunning for the rest of the session. +const TERM_GRACE: Duration = Duration::from_secs(3); +const KILL_GRACE: Duration = Duration::from_secs(2); + +/// How a job's process ended, as observed by the detached task that owns +/// the process handle. +enum JobOutcome { + Exited(ExitStatus), + WaitFailed(ProcessError), + Cancelled, +} + +/// Internal report from a job task back to the actor loop. Kept off the +/// public [`KwMessage`] protocol: no caller can fake a completion. +enum JobEvent { + Finished(JobOutcome), +} + +/// What the actor remembers about the running job while the detached task +/// owns the process itself (see [`run_job`]). +struct JobState { + kind: KwJobKind, + phase: KwPhase, + kernel_tree_id: String, + branch: String, + log_path: PathBuf, + /// HEAD at accept time, so RestorePreviousBranch can offer to switch + /// back after the job. Read once the checkout policy lands. + #[allow(dead_code)] + pre_job_branch: Option, + /// `None` once a cancel has been requested; a second `Cancel` is an + /// idempotent ack. + cancel_tx: Option>, +} + +pub struct KwActor { + rx: mpsc::Receiver, + job_event_rx: mpsc::Receiver, + job_event_tx: mpsc::Sender, + status_tx: watch::Sender, + history: Arc, + shell: Arc, + fs: Arc, + env: Arc, + process: Arc, + kw_log_dir: PathBuf, + job: Option, +} + +impl KwActor { + pub fn new( + rx: mpsc::Receiver, + history: Arc, + process: Arc, + shell: Arc, + fs: Arc, + env: Arc, + kw_log_dir: PathBuf, + ) -> Self { + let (status_tx, _) = watch::channel(KwStatusSnapshot::idle()); + let (job_event_tx, job_event_rx) = mpsc::channel(DEFAULT_KW_CHANNEL_SIZE); + Self { + rx, + job_event_rx, + job_event_tx, + status_tx, + history, + shell, + fs, + env, + process, + kw_log_dir, + job: None, + } + } + + pub fn spawn( + history: Arc, + process: Arc, + shell: Arc, + fs: Arc, + env: Arc, + kw_log_dir: PathBuf, + ) -> KwHandle { + let (tx, rx) = mpsc::channel(DEFAULT_KW_CHANNEL_SIZE); + tracing::debug!(channel_size = DEFAULT_KW_CHANNEL_SIZE, "spawning kw actor"); + spawn(Self::new(rx, history, process, shell, fs, env, kw_log_dir).run()); + KwHandle::new(tx) + } + + pub async fn run(mut self) { + tracing::info!("kw actor started"); + // The job-event sender is held by the actor itself, so that arm of + // the select never closes while the actor is alive. + loop { + tokio::select! { + message = self.rx.recv() => { + let Some(message) = message else { break }; + if let ControlFlow::Break(()) = self.handle_message(message).await { + break; + } + } + Some(event) = self.job_event_rx.recv() => { + self.handle_job_event(event); + } + } + } + tracing::info!("kw actor stopped"); + } + + async fn handle_message(&mut self, message: KwMessage) -> ControlFlow<()> { + let message_name = message.name(); + tracing::debug!(message = message_name, "kw request received"); + + match message { + KwMessage::RecordApply { record, reply } => { + send_kw_reply( + message_name, + reply, + self.history.record_apply(record).map_err(KwError::from), + ); + ControlFlow::Continue(()) + } + KwMessage::StartBuild { request, reply } => { + send_start_reply( + message_name, + reply, + self.start_job(KwJobKind::Build, request), + ); + ControlFlow::Continue(()) + } + // Not implemented: reply immediately with NotImplemented. + KwMessage::StartDeploy { reply, .. } + | KwMessage::StartBuildThenDeploy { reply, .. } => { + send_start_reply(message_name, reply, Err(KwStartError::NotImplemented)); + ControlFlow::Continue(()) + } + KwMessage::Cancel { reply } => { + send_kw_reply(message_name, reply, self.request_cancel()); + ControlFlow::Continue(()) + } + KwMessage::GetStatus { reply } => { + send_value_reply(message_name, reply, self.status_tx.borrow().clone()); + ControlFlow::Continue(()) + } + KwMessage::WatchStatus { reply } => { + send_value_reply(message_name, reply, self.status_tx.subscribe()); + ControlFlow::Continue(()) + } + KwMessage::GetReadiness { + kernel_tree_id, + tree, + reply, + } => { + send_kw_reply( + message_name, + reply, + self.evaluate_readiness(&kernel_tree_id, &tree), + ); + ControlFlow::Continue(()) + } + // Restoring needs the dirty-worktree check and `git switch`, + // which arrive with the checkout policy in the build step. + KwMessage::RestorePreviousBranch { reply } => { + send_kw_reply(message_name, reply, Err(KwError::NoRecordedBranch)); + ControlFlow::Continue(()) + } + KwMessage::Shutdown { reply } => { + // Quit-prompt cooperation: the app asks whether a job is + // running before quitting; if it quits anyway, the job's + // process group is killed here. The log may capture partial + // output and the tree a partial build. The reply is held + // until the kill escalation has run its (bounded) course, + // so a caller tearing down the runtime afterwards cannot + // leave orphaned kw processes behind. + if self.job.is_some() { + tracing::info!("killing kw job process group during shutdown"); + self.request_cancel().ok(); + while self.job.is_some() { + // `None` is unreachable — the actor holds + // `job_event_tx`, so the channel never closes — + // but break defensively rather than spin. + match self.job_event_rx.recv().await { + Some(event) => self.handle_job_event(event), + None => break, + } + } + } + reply.send(()).ok(); + tracing::debug!("kw actor shutting down"); + ControlFlow::Break(()) + } + } + } + + /// Accepts and starts a build job, or refuses. The reply is sent by the + /// caller right after this returns: the job itself keeps running in a + /// detached task and is observed via the status snapshot. + /// + /// The argv is `kw build --alert=n`. The `branch` on the Running status + /// is the requested branch; this skeleton does not switch the tree. + fn start_job(&mut self, kind: KwJobKind, request: StartRequest) -> Result<(), KwStartError> { + if self.job.is_some() { + return Err(KwStartError::JobAlreadyRunning); + } + + // Recorded at accept so RestorePreviousBranch can switch back. + // An unprobed HEAD (detached, or not a git repo) records nothing. + let pre_job_branch = match self.head_branch(&request.tree) { + branch if branch.is_empty() => None, + branch => Some(branch), + }; + + self.fs.create_dir_all(&self.kw_log_dir)?; + // Millisecond suffix: two jobs started within the same second must + // not share a log file — spawn truncates it. + let log_path = self.kw_log_dir.join(format!( + "build-{}.log", + chrono::Utc::now().format("%Y%m%d-%H%M%S-%3f") + )); + let cmd = ShellCommand::new("kw").args(["build", "--alert=n"]); + let cwd = PathBuf::from(request.tree.path()); + let process = self.process.spawn(&cmd, &cwd, &log_path)?; + + let (cancel_tx, cancel_rx) = oneshot::channel(); + spawn(run_job(process, cancel_rx, self.job_event_tx.clone())); + + let phase = KwPhase::Building; + tracing::info!( + kernel_tree_id = request.kernel_tree_id, + branch = request.branch, + log_path = %log_path.display(), + "kw build job started" + ); + self.job = Some(JobState { + kind, + phase, + kernel_tree_id: request.kernel_tree_id.clone(), + branch: request.branch.clone(), + log_path: log_path.clone(), + pre_job_branch, + cancel_tx: Some(cancel_tx), + }); + self.set_status(KwJobStatus::Running { + kind, + phase, + kernel_tree_id: request.kernel_tree_id, + branch: request.branch, + log_path, + }); + Ok(()) + } + + fn request_cancel(&mut self) -> Result<(), KwError> { + match self.job.as_mut() { + Some(job) => { + if let Some(cancel) = job.cancel_tx.take() { + let _ = cancel.send(()); + } + Ok(()) + } + None => Err(KwError::NoJobRunning), + } + } + + fn handle_job_event(&mut self, event: JobEvent) { + match event { + JobEvent::Finished(outcome) => { + let Some(job) = self.job.take() else { + tracing::warn!("kw job finished with no job state recorded"); + return; + }; + let status = match outcome { + JobOutcome::Exited(exit) if exit.success() => { + tracing::info!( + kernel_tree_id = job.kernel_tree_id, + branch = job.branch, + "kw job succeeded" + ); + KwJobStatus::Succeeded { + kind: job.kind, + kernel_tree_id: job.kernel_tree_id, + branch: job.branch, + log_path: job.log_path, + } + } + JobOutcome::Exited(exit) => { + tracing::warn!( + kernel_tree_id = job.kernel_tree_id, + branch = job.branch, + exit_code = exit.code(), + "kw job failed" + ); + KwJobStatus::Failed { + kind: job.kind, + phase: job.phase, + exit_code: exit.code(), + log_path: job.log_path, + } + } + JobOutcome::WaitFailed(error) => { + tracing::warn!(branch = job.branch, %error, "failed to wait on kw job"); + KwJobStatus::Failed { + kind: job.kind, + phase: job.phase, + exit_code: None, + log_path: job.log_path, + } + } + JobOutcome::Cancelled => { + tracing::info!(branch = job.branch, "kw job cancelled"); + KwJobStatus::Cancelled { + kind: job.kind, + phase: job.phase, + } + } + }; + self.set_status(status); + } + } + } + + fn set_status(&mut self, job: KwJobStatus) { + // send_replace, not send: no receiver (nobody called WatchStatus + // yet) is a normal state, not an error. + self.status_tx.send_replace(KwStatusSnapshot { job }); + } + + fn evaluate_readiness( + &self, + kernel_tree_id: &str, + tree: &KernelTree, + ) -> Result { + let head = self.head_branch(tree); + Ok(readiness::evaluate_readiness( + &*self.fs, + &*self.env, + &*self.shell, + &*self.history, + kernel_tree_id, + tree, + &head, + )?) + } + + /// The tree's current branch, probed via git. An unresolvable HEAD + /// (not a git repo, or a detached HEAD, for which `branch + /// --show-current` prints nothing) yields an empty string: no build + /// record can match it, so deploy-alone readiness refuses — the safe + /// direction for an unknown HEAD. + fn head_branch(&self, tree: &KernelTree) -> String { + let cmd = ShellCommand::new("git").args(["-C", tree.path(), "branch", "--show-current"]); + match self.shell.execute(&cmd) { + Ok(output) if output.success => { + String::from_utf8_lossy(&output.stdout).trim().to_string() + } + Ok(output) => { + tracing::warn!( + tree = tree.path(), + stderr = %String::from_utf8_lossy(&output.stderr), + "failed to probe the kernel tree's HEAD branch" + ); + String::new() + } + Err(error) => { + tracing::warn!(tree = tree.path(), %error, "failed to probe the kernel tree's HEAD branch"); + String::new() + } + } + } +} + +/// Owns the spawned process until it ends: waits on it, or — when the +/// cancel signal fires — runs the kill escalation and reaps it. The +/// `wait()` future is dropped before the cancel arm's body runs, releasing +/// the mutable borrow so `kill()` can be called. Reports the outcome back +/// to the actor over the internal event channel. +async fn run_job( + mut process: Box, + mut cancel_rx: oneshot::Receiver<()>, + events: mpsc::Sender, +) { + let outcome = tokio::select! { + status = process.wait() => match status { + Ok(status) => JobOutcome::Exited(status), + Err(error) => JobOutcome::WaitFailed(error), + }, + // A dropped sender (actor shutting down) cancels the job too. + _ = &mut cancel_rx => cancel_job(&mut *process).await, + }; + events.send(JobEvent::Finished(outcome)).await.ok(); +} + +/// Cancel escalation ladder: SIGTERM the group, wait a grace period, +/// SIGKILL, wait again, then give up. Giving up still reports Cancelled and +/// lets the actor clear its job state — a process group that ignores both +/// signals must not wedge the actor into refusing every later Start with +/// `JobAlreadyRunning` for the rest of the session. +async fn cancel_job(process: &mut dyn RunningProcess) -> JobOutcome { + if let Err(error) = process.kill() { + tracing::warn!(%error, "failed to SIGTERM kw job process group"); + } + match tokio::time::timeout(TERM_GRACE, process.wait()).await { + Ok(outcome) => outcome_after_cancel(outcome), + Err(_) => { + tracing::warn!("kw job ignored SIGTERM; escalating to SIGKILL"); + if let Err(error) = process.force_kill() { + tracing::warn!(%error, "failed to SIGKILL kw job process group"); + } + match tokio::time::timeout(KILL_GRACE, process.wait()).await { + Ok(outcome) => outcome_after_cancel(outcome), + Err(_) => { + tracing::warn!( + "kw job process group could not be reaped after SIGKILL; giving up" + ); + JobOutcome::Cancelled + } + } + } + } +} + +/// Maps a reap result observed after a cancel request. A signal-terminated +/// process means our SIGTERM/SIGKILL landed — the job was really cancelled. +/// A plain exit means the process finished on its own before the signal: +/// report the real outcome, because a cancel must not mask a failure the +/// build history (and deploy-alone readiness) needs to see. +/// +/// Known, accepted edges: a process that *traps* our SIGTERM and exits 0 +/// counts as success (kw is bash, so this is possible in principle), and an +/// external signal racing a cancel (e.g. the OOM killer) reads as +/// Cancelled. Both are indistinguishable from the honest cases without +/// comparing who signaled first, and both favor showing the user real +/// output over inventing failures. +fn outcome_after_cancel(result: Result) -> JobOutcome { + use std::os::unix::process::ExitStatusExt; + + match result { + Ok(status) => match status.signal() { + Some(_) => JobOutcome::Cancelled, + None => JobOutcome::Exited(status), + }, + Err(error) => JobOutcome::WaitFailed(error), + } +} + +fn send_kw_reply( + message_name: &'static str, + reply: oneshot::Sender>, + result: Result, +) { + if let Err(error) = &result { + tracing::warn!( + message = message_name, + error = %error, + "kw request failed" + ); + } + + if reply.send(result).is_err() { + tracing::warn!( + message = message_name, + "kw reply receiver dropped before response" + ); + } +} + +fn send_start_reply( + message_name: &'static str, + reply: oneshot::Sender>, + result: Result<(), KwStartError>, +) { + if let Err(error) = &result { + tracing::warn!( + message = message_name, + error = %error, + "kw start request refused" + ); + } + + if reply.send(result).is_err() { + tracing::warn!( + message = message_name, + "kw reply receiver dropped before response" + ); + } +} + +fn send_value_reply(message_name: &'static str, reply: oneshot::Sender, value: T) { + if reply.send(value).is_err() { + tracing::warn!( + message = message_name, + "kw reply receiver dropped before response" + ); + } +} + +#[cfg(test)] +mod tests { + use std::{ + io, + path::Path, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, + }; + + use crate::{ + infrastructure::{ + env::MockEnvTrait, + file_system::{FileSystemError, MockFileSystemTrait}, + process::FakeProcess, + shell::{MockShellTrait, ShellOutput}, + }, + kw::{ + errors::KwStartError, + history::{KwApplyRecord, MockKwHistoryStore}, + messages::StartRequest, + readiness::{DeployAloneRefusal, TreeReadiness}, + status::{KwJobKind, KwJobStatus, KwPhase}, + }, + }; + + use super::*; + + static TEST_SEQ: AtomicU64 = AtomicU64::new(0); + + /// A real directory: FakeProcess creates the log file on spawn, so the + /// parent must exist even though the fs trait is mocked. + fn tmp_log_dir(test_name: &str) -> PathBuf { + let n = TEST_SEQ.fetch_add(1, Ordering::SeqCst); + let dir = std::env::temp_dir().join(format!( + "patch-hub-kw-actor-{}-{test_name}-{n}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + fn kernel_tree(path: &Path) -> KernelTree { + serde_json::from_value(serde_json::json!({ + "path": path.to_str().unwrap(), + "branch": "master" + })) + .unwrap() + } + + fn start_request() -> StartRequest { + StartRequest { + kernel_tree_id: "mainline".to_string(), + tree: kernel_tree(Path::new("/home/user/linux")), + branch: "patchset-2026-08-01-17-30-00".to_string(), + } + } + + /// Spawns the actor with the already-configured mocks (mockall + /// expectations need `&mut`, so they are set before the mocks move + /// behind `Arc`s). The log dir is a real unique temp dir even though + /// these tests never start a job, so a future test that accidentally + /// does cannot share a fixed path with the job tests. + fn spawn_test_actor( + test_name: &str, + history: MockKwHistoryStore, + shell: MockShellTrait, + fs: MockFileSystemTrait, + env: MockEnvTrait, + ) -> KwHandle { + KwActor::spawn( + Arc::new(history), + Arc::new(FakeProcess::new()), + Arc::new(shell), + Arc::new(fs), + Arc::new(env), + tmp_log_dir(test_name), + ) + } + + /// Spawns the actor with a real temp log dir and exposes the + /// [`FakeProcess`] so tests drive the "running" process. The shell mock + /// answers the pre-job HEAD probe. + fn spawn_job_actor_with_fs( + test_name: &str, + fs: MockFileSystemTrait, + ) -> (KwHandle, Arc, PathBuf) { + let process = Arc::new(FakeProcess::new()); + let log_dir = tmp_log_dir(test_name); + let mut shell = MockShellTrait::new(); + shell.expect_execute().returning(|_| { + Ok(ShellOutput { + stdout: b"master\n".to_vec(), + stderr: Vec::new(), + success: true, + }) + }); + let handle = KwActor::spawn( + Arc::new(MockKwHistoryStore::new()), + process.clone(), + Arc::new(shell), + Arc::new(fs), + Arc::new(MockEnvTrait::new()), + log_dir.clone(), + ); + (handle, process, log_dir) + } + + fn spawn_job_actor(test_name: &str) -> (KwHandle, Arc, PathBuf) { + let mut fs = MockFileSystemTrait::new(); + fs.expect_create_dir_all().returning(|_| Ok(())); + spawn_job_actor_with_fs(test_name, fs) + } + + fn apply_record() -> KwApplyRecord { + KwApplyRecord { + message_id: "msg-1".to_string(), + kernel_tree_id: "mainline".to_string(), + tree_path: "/home/user/linux".to_string(), + applied_branch: "patchset-2026-08-01-17-30-00".to_string(), + base_branch: "master".to_string(), + applied_at: "2026-08-01T17:30:00Z".to_string(), + } + } + + #[tokio::test] + async fn record_apply_writes_through_history_store() { + let expected = apply_record(); + let mut history = MockKwHistoryStore::new(); + history + .expect_record_apply() + .withf(move |record| *record == expected) + .times(1) + .returning(|_| Ok(())); + let handle = spawn_test_actor( + "record-apply", + history, + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + handle.record_apply(apply_record()).await.unwrap(); + handle.shutdown().await; + } + + #[tokio::test] + async fn record_apply_surfaces_store_errors() { + let mut history = MockKwHistoryStore::new(); + history + .expect_record_apply() + .times(1) + .returning(|_| Err(FileSystemError::IoError(io::Error::other("disk full")))); + let handle = spawn_test_actor( + "record-apply-error", + history, + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let err = handle.record_apply(apply_record()).await.unwrap_err(); + + assert!(matches!(err, KwError::History(_))); + handle.shutdown().await; + } + + #[tokio::test] + async fn get_status_reports_idle_before_any_job() { + let handle = spawn_test_actor( + "idle-status", + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let snapshot = handle.get_status().await.unwrap(); + + assert_eq!(KwJobStatus::Idle, snapshot.job); + handle.shutdown().await; + } + + #[tokio::test] + async fn watch_status_receiver_sees_current_snapshot() { + let handle = spawn_test_actor( + "watch-idle", + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let receiver = handle.watch_status().await.unwrap(); + + assert_eq!(KwJobStatus::Idle, receiver.borrow().job); + handle.shutdown().await; + } + + #[tokio::test] + async fn get_readiness_composes_probes_and_head_branch() { + let tree = kernel_tree(Path::new("/home/user/linux")); + + let mut env = MockEnvTrait::new(); + env.expect_which() + .withf(|name| name == "kw") + .returning(|_| false); + let mut fs = MockFileSystemTrait::new(); + fs.expect_is_file().returning(|_| false); + fs.expect_is_dir().returning(|_| false); + fs.expect_read_dir().returning(|_| { + Err(FileSystemError::IoError(io::Error::new( + io::ErrorKind::NotFound, + "missing", + ))) + }); + let mut history = MockKwHistoryStore::new(); + history + .expect_build_records() + .withf(|kernel_tree_id, branch| kernel_tree_id == "mainline" && branch == "for-next") + .times(1) + .returning(|_, _| Ok((None, None))); + let mut shell = MockShellTrait::new(); + shell + .expect_execute() + .withf(|cmd| { + cmd.program == "git" + && cmd.args == ["-C", "/home/user/linux", "branch", "--show-current"] + }) + .times(1) + .returning(|_| { + Ok(ShellOutput { + stdout: b"for-next\n".to_vec(), + stderr: Vec::new(), + success: true, + }) + }); + let handle = spawn_test_actor("readiness", history, shell, fs, env); + + let readiness = handle.get_readiness("mainline", &tree).await.unwrap(); + + assert!(!readiness.kw_binary.available); + assert_eq!(TreeReadiness::Missing, readiness.tree); + assert_eq!( + Err(DeployAloneRefusal::TreeNotReady(TreeReadiness::Missing)), + readiness.deploy_alone + ); + handle.shutdown().await; + } + + /// Waits until the status leaves `Idle`/`Running` and returns the + /// terminal status. The receiver may have observed the `Running` + /// transition first, so a single `changed()` is not enough. The timeout + /// backstops against a wedged actor; tests that exercise the grace + /// periods run with paused time instead of waiting them out. + async fn wait_for_terminal_status( + watch: &mut watch::Receiver, + ) -> KwJobStatus { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let status = watch.borrow().job.clone(); + if !matches!(status, KwJobStatus::Idle | KwJobStatus::Running { .. }) { + return status; + } + watch.changed().await.unwrap(); + } + }) + .await + .expect("status must reach a terminal state") + } + + #[tokio::test] + async fn start_build_replies_immediately_and_runs_in_background() { + let (handle, process, log_dir) = spawn_job_actor("start-immediate"); + + // start_build resolves while the spawned process is still running + // (no finish() was ever signaled). + let result = + tokio::time::timeout(Duration::from_secs(1), handle.start_build(start_request())) + .await + .expect("start_build must reply immediately"); + result.unwrap(); + + let spawned = process.spawned(); + assert_eq!(1, spawned.len()); + assert_eq!("kw", spawned[0].program); + assert_eq!(["build", "--alert=n"], spawned[0].args.as_slice()); + assert_eq!(Path::new("/home/user/linux"), spawned[0].cwd); + assert!(spawned[0].log_path.starts_with(&log_dir)); + + let snapshot = handle.get_status().await.unwrap(); + assert!( + matches!( + snapshot.job, + KwJobStatus::Running { + kind: KwJobKind::Build, + phase: KwPhase::Building, + .. + } + ), + "unexpected status: {:?}", + snapshot.job + ); + + process.last_child().finish(0); + let mut watch = handle.watch_status().await.unwrap(); + let status = wait_for_terminal_status(&mut watch).await; + assert!( + matches!( + status, + KwJobStatus::Succeeded { + kind: KwJobKind::Build, + .. + } + ), + "unexpected status: {status:?}" + ); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn second_start_while_running_is_refused() { + let (handle, process, log_dir) = spawn_job_actor("busy"); + + handle.start_build(start_request()).await.unwrap(); + let second = handle.start_build(start_request()).await; + + assert!(matches!(second, Err(KwStartError::JobAlreadyRunning))); + + process.last_child().finish(0); + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn failed_build_reports_exit_code_and_log_path() { + let (handle, process, log_dir) = spawn_job_actor("failed"); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + process.last_child().finish(2); + + let status = wait_for_terminal_status(&mut watch).await; + + match status { + KwJobStatus::Failed { + kind, + phase, + exit_code, + log_path, + } => { + assert_eq!(KwJobKind::Build, kind); + assert_eq!(KwPhase::Building, phase); + assert_eq!(Some(2), exit_code); + assert!(log_path.starts_with(&log_dir)); + } + other => panic!("expected Failed, got {other:?}"), + } + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn cancel_kills_process_group_and_reports_cancelled() { + let (handle, process, log_dir) = spawn_job_actor("cancel"); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + // The ack is immediate: process death is observed via the status, + // not the reply. + tokio::time::timeout(Duration::from_secs(1), handle.cancel()) + .await + .expect("cancel must ack immediately") + .unwrap(); + + assert!(process.last_child().was_killed()); + let status = wait_for_terminal_status(&mut watch).await; + assert_eq!( + KwJobStatus::Cancelled { + kind: KwJobKind::Build, + phase: KwPhase::Building, + }, + status + ); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn shutdown_kills_running_job() { + let (handle, process, log_dir) = spawn_job_actor("shutdown-kill"); + + handle.start_build(start_request()).await.unwrap(); + // shutdown() returns only after the kill escalation has completed. + handle.shutdown().await; + + assert!(process.last_child().was_killed()); + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn spawn_failure_refuses_start_and_stays_idle() { + let (handle, process, log_dir) = spawn_job_actor("spawn-fail"); + process.refuse_spawns(true); + + let err = handle.start_build(start_request()).await.unwrap_err(); + + assert!(matches!(err, KwStartError::Spawn(_))); + assert_eq!(KwJobStatus::Idle, handle.get_status().await.unwrap().job); + // A refused start must leave the actor able to accept a later one. + process.refuse_spawns(false); + handle.start_build(start_request()).await.unwrap(); + + process.last_child().finish(0); + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn start_deploy_still_refused_until_the_deploy_step() { + let handle = spawn_test_actor( + "deploy-refused", + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let result = + tokio::time::timeout(Duration::from_secs(1), handle.start_deploy(start_request())) + .await + .expect("start_deploy must reply immediately"); + + assert!(matches!(result, Err(KwStartError::NotImplemented))); + handle.shutdown().await; + } + + #[tokio::test] + async fn cancel_and_restore_without_job_are_immediate_errors() { + let handle = spawn_test_actor( + "no-job", + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + let cancel = tokio::time::timeout(Duration::from_secs(1), handle.cancel()) + .await + .expect("cancel must reply immediately"); + let restore = + tokio::time::timeout(Duration::from_secs(1), handle.restore_previous_branch()) + .await + .expect("restore must reply immediately"); + + assert!(matches!(cancel, Err(KwError::NoJobRunning))); + assert!(matches!(restore, Err(KwError::NoRecordedBranch))); + handle.shutdown().await; + } + + #[tokio::test] + async fn shutdown_stops_actor() { + let handle = spawn_test_actor( + "shutdown", + MockKwHistoryStore::new(), + MockShellTrait::new(), + MockFileSystemTrait::new(), + MockEnvTrait::new(), + ); + + handle.shutdown().await; + let err = handle.get_status().await.unwrap_err(); + + assert!(matches!(err, KwError::ActorUnavailable(_))); + } + + #[tokio::test] + async fn log_dir_creation_failure_refuses_start_and_stays_idle() { + let mut fs = MockFileSystemTrait::new(); + fs.expect_create_dir_all().returning(|_| { + Err(FileSystemError::IoError(io::Error::other( + "read-only filesystem", + ))) + }); + let (handle, process, log_dir) = spawn_job_actor_with_fs("log-dir-fail", fs); + + let err = handle.start_build(start_request()).await.unwrap_err(); + + assert!(matches!(err, KwStartError::Fs(_))); + assert_eq!(KwJobStatus::Idle, handle.get_status().await.unwrap().job); + assert!(process.spawned().is_empty()); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn cancel_racing_a_successful_exit_reports_success() { + let (handle, process, log_dir) = spawn_job_actor("cancel-race"); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + process.last_child().finish(0); + // Whether the actor processes the exit or the cancel first is + // timing-dependent, but the terminal status must be Succeeded + // either way: the process exited before any signal landed. + let _ = handle.cancel().await; + + let status = wait_for_terminal_status(&mut watch).await; + assert!( + matches!(status, KwJobStatus::Succeeded { .. }), + "expected Succeeded, got {status:?}" + ); + // Killing an already-finished process is a no-op, not a kill. + assert!(!process.last_child().was_killed()); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + #[tokio::test] + async fn cancel_racing_a_failed_exit_reports_failure() { + let (handle, process, log_dir) = spawn_job_actor("cancel-race-fail"); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + process.last_child().finish(2); + // Whether the actor processes the exit or the cancel first is + // timing-dependent, but a plain exit (not signal-terminated) means + // the process failed on its own before the signal landed: the + // terminal status must be Failed either way, so the build history + // records a failure rather than a cancel. + let _ = handle.cancel().await; + + let status = wait_for_terminal_status(&mut watch).await; + assert!( + matches!( + status, + KwJobStatus::Failed { + exit_code: Some(2), + .. + } + ), + "expected Failed(2), got {status:?}" + ); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } + + // Paused time: the runtime auto-advances through the grace-period + // timers instead of burning wall-clock seconds on them. + #[tokio::test(start_paused = true)] + async fn cancel_escalates_to_sigkill_when_sigterm_is_ignored() { + let (handle, process, log_dir) = spawn_job_actor("sigkill-escalation"); + process.ignore_sigterm(true); + let mut watch = handle.watch_status().await.unwrap(); + + handle.start_build(start_request()).await.unwrap(); + handle.cancel().await.unwrap(); + + let status = wait_for_terminal_status(&mut watch).await; + assert!( + matches!(status, KwJobStatus::Cancelled { .. }), + "expected Cancelled, got {status:?}" + ); + let child = process.last_child(); + assert!(child.was_killed()); + assert!(child.was_force_killed()); + + handle.shutdown().await; + std::fs::remove_dir_all(&log_dir).unwrap(); + } +} diff --git a/src/kw/errors.rs b/src/kw/errors.rs new file mode 100644 index 00000000..30179b49 --- /dev/null +++ b/src/kw/errors.rs @@ -0,0 +1,46 @@ +// The actor that constructs/reads these is unix-only. +#![cfg_attr(not(unix), allow(dead_code))] + +use thiserror::Error; + +#[cfg(unix)] +use crate::infrastructure::process::ProcessError; +use crate::{ + infrastructure::{file_system::FileSystemError, shell::ShellError}, + kw::readiness::KwReadinessError, +}; + +#[derive(Debug, Error)] +pub enum KwError { + #[error("kw actor unavailable: {0}")] + ActorUnavailable(String), + #[error("no kw job is running")] + NoJobRunning, + #[error("no pre-job branch was recorded")] + NoRecordedBranch, + #[error("history error: {0}")] + History(#[from] FileSystemError), + #[error("readiness error: {0}")] + Readiness(#[from] KwReadinessError), + #[error("shell error: {0}")] + Shell(#[from] ShellError), +} + +/// Accept/refuse verdict for `Start*` messages. The reply is always +/// immediate: an accepted job keeps running inside the actor after the +/// caller has been answered. +#[derive(Debug, Error)] +pub enum KwStartError { + #[error("kw actor unavailable: {0}")] + ActorUnavailable(String), + #[error("a kw job is already running")] + JobAlreadyRunning, + #[error("kw jobs are not supported yet")] + NotImplemented, + // Spawning a process is unix-only (ProcessTrait is cfg(unix)). + #[cfg(unix)] + #[error("failed to spawn the kw process: {0}")] + Spawn(#[from] ProcessError), + #[error("filesystem error: {0}")] + Fs(#[from] FileSystemError), +} diff --git a/src/kw/handle.rs b/src/kw/handle.rs new file mode 100644 index 00000000..9530f5ee --- /dev/null +++ b/src/kw/handle.rs @@ -0,0 +1,121 @@ +use tokio::sync::{mpsc, oneshot, watch}; + +use crate::{ + config::KernelTree, + kw::{ + errors::{KwError, KwStartError}, + history::KwApplyRecord, + messages::{KwMessage, StartRequest}, + readiness::KwReadiness, + status::KwStatusSnapshot, + }, +}; + +#[derive(Clone)] +pub struct KwHandle { + tx: mpsc::Sender, +} + +#[allow(dead_code)] +impl KwHandle { + pub fn new(tx: mpsc::Sender) -> Self { + Self { tx } + } + + pub async fn record_apply(&self, record: KwApplyRecord) -> Result<(), KwError> { + self.request_result(|reply| KwMessage::RecordApply { record, reply }) + .await + } + + /// Resolves as soon as the actor accepts or refuses the job — never + /// when the job finishes. + pub async fn start_build(&self, request: StartRequest) -> Result<(), KwStartError> { + self.start(|reply| KwMessage::StartBuild { request, reply }) + .await + } + + pub async fn start_deploy(&self, request: StartRequest) -> Result<(), KwStartError> { + self.start(|reply| KwMessage::StartDeploy { request, reply }) + .await + } + + pub async fn start_build_then_deploy(&self, request: StartRequest) -> Result<(), KwStartError> { + self.start(|reply| KwMessage::StartBuildThenDeploy { request, reply }) + .await + } + + pub async fn cancel(&self) -> Result<(), KwError> { + self.request_result(|reply| KwMessage::Cancel { reply }) + .await + } + + pub async fn get_status(&self) -> Result { + self.request(|reply| KwMessage::GetStatus { reply }).await + } + + pub async fn watch_status(&self) -> Result, KwError> { + self.request(|reply| KwMessage::WatchStatus { reply }).await + } + + pub async fn get_readiness( + &self, + kernel_tree_id: &str, + tree: &KernelTree, + ) -> Result { + self.request_result(|reply| KwMessage::GetReadiness { + kernel_tree_id: kernel_tree_id.to_string(), + tree: tree.clone(), + reply, + }) + .await + } + + pub async fn restore_previous_branch(&self) -> Result<(), KwError> { + self.request_result(|reply| KwMessage::RestorePreviousBranch { reply }) + .await + } + + /// Signals the actor to stop processing messages and exit its run loop. + /// A running job's process group is killed first; this returns only + /// after the kill escalation has completed, so teardown can drop the + /// runtime without orphaning kw's child processes. + pub async fn shutdown(&self) { + let (reply, rx) = oneshot::channel(); + if self.tx.send(KwMessage::Shutdown { reply }).await.is_ok() { + rx.await.ok(); + } + } + + async fn request_result( + &self, + build_message: impl FnOnce(oneshot::Sender>) -> KwMessage, + ) -> Result { + self.request(build_message).await? + } + + async fn request( + &self, + build_message: impl FnOnce(oneshot::Sender) -> KwMessage, + ) -> Result { + let (reply, rx) = oneshot::channel(); + self.tx + .send(build_message(reply)) + .await + .map_err(|_| KwError::ActorUnavailable("request channel closed".to_string()))?; + rx.await + .map_err(|_| KwError::ActorUnavailable("reply channel closed".to_string())) + } + + async fn start( + &self, + build_message: impl FnOnce(oneshot::Sender>) -> KwMessage, + ) -> Result<(), KwStartError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(build_message(reply)) + .await + .map_err(|_| KwStartError::ActorUnavailable("request channel closed".to_string()))?; + rx.await + .map_err(|_| KwStartError::ActorUnavailable("reply channel closed".to_string()))? + } +} diff --git a/src/kw/messages.rs b/src/kw/messages.rs new file mode 100644 index 00000000..067b6da3 --- /dev/null +++ b/src/kw/messages.rs @@ -0,0 +1,89 @@ +// The actor that reads these fields is unix-only. +#![cfg_attr(not(unix), allow(dead_code))] + +use tokio::sync::{oneshot, watch}; + +use crate::{ + config::KernelTree, + kw::{ + errors::{KwError, KwStartError}, + history::KwApplyRecord, + readiness::KwReadiness, + status::KwStatusSnapshot, + }, +}; + +/// Everything the actor needs to start a job. The tree context is resolved +/// by the caller from its config snapshot, keeping KwActor decoupled from +/// ConfigActor. +#[derive(Debug, Clone)] +pub struct StartRequest { + pub kernel_tree_id: String, + pub tree: KernelTree, + /// Branch the job must run on. + pub branch: String, +} + +pub enum KwMessage { + RecordApply { + record: KwApplyRecord, + reply: oneshot::Sender>, + }, + StartBuild { + request: StartRequest, + reply: oneshot::Sender>, + }, + StartDeploy { + #[allow(dead_code)] + request: StartRequest, + reply: oneshot::Sender>, + }, + StartBuildThenDeploy { + #[allow(dead_code)] + request: StartRequest, + reply: oneshot::Sender>, + }, + /// Acknowledges that kill was requested; the actual process death is + /// observed via the status snapshot, not this reply. + Cancel { + reply: oneshot::Sender>, + }, + GetStatus { + reply: oneshot::Sender, + }, + /// Called once by the AppActor when it attaches, not per frame: the + /// returned receiver is its wake source for status changes. + WatchStatus { + reply: oneshot::Sender>, + }, + GetReadiness { + kernel_tree_id: String, + tree: KernelTree, + reply: oneshot::Sender>, + }, + RestorePreviousBranch { + reply: oneshot::Sender>, + }, + /// Unlike the other actors' bare `Shutdown`, this one replies — the + /// `TerminalMessage::Shutdown` convention. Teardown must know the + /// running job's process group was actually killed before the runtime + /// is dropped; a fire-and-forget message leaves that to scheduler luck. + Shutdown { reply: oneshot::Sender<()> }, +} + +impl KwMessage { + pub fn name(&self) -> &'static str { + match self { + KwMessage::RecordApply { .. } => "RecordApply", + KwMessage::StartBuild { .. } => "StartBuild", + KwMessage::StartDeploy { .. } => "StartDeploy", + KwMessage::StartBuildThenDeploy { .. } => "StartBuildThenDeploy", + KwMessage::Cancel { .. } => "Cancel", + KwMessage::GetStatus { .. } => "GetStatus", + KwMessage::WatchStatus { .. } => "WatchStatus", + KwMessage::GetReadiness { .. } => "GetReadiness", + KwMessage::RestorePreviousBranch { .. } => "RestorePreviousBranch", + KwMessage::Shutdown { .. } => "Shutdown", + } + } +} diff --git a/src/kw/mod.rs b/src/kw/mod.rs index 2e5b1b99..24f19ea7 100644 --- a/src/kw/mod.rs +++ b/src/kw/mod.rs @@ -1,5 +1,11 @@ -//! kw integration: persistence and readiness probes for `kw build` / -//! `kw deploy` jobs. +//! kw integration: persistence, readiness probes, and the actor +//! orchestrating `kw build` / `kw deploy` jobs. +#[cfg(unix)] +pub mod actor; +pub mod errors; +pub mod handle; pub mod history; +pub mod messages; pub mod readiness; +pub mod status; diff --git a/src/kw/status.rs b/src/kw/status.rs new file mode 100644 index 00000000..3f638aa9 --- /dev/null +++ b/src/kw/status.rs @@ -0,0 +1,67 @@ +//! Status snapshot projected out of the kw actor. +//! +//! `AppState` never owns job state; it polls (`GetStatus`) or watches +//! (`WatchStatus`) these snapshots and projects them into the view model. + +// The actor that constructs/reads these is unix-only. +#![cfg_attr(not(unix), allow(dead_code))] + +use std::path::PathBuf; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum KwJobKind { + Build, + #[allow(dead_code)] + Deploy, + #[allow(dead_code)] + BuildThenDeploy, +} + +/// Running phase of a job. `BuildThenDeploy` jobs pass through both. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum KwPhase { + Building, + #[allow(dead_code)] + Deploying, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum KwJobStatus { + Idle, + Running { + kind: KwJobKind, + phase: KwPhase, + kernel_tree_id: String, + branch: String, + log_path: PathBuf, + }, + Succeeded { + kind: KwJobKind, + kernel_tree_id: String, + branch: String, + log_path: PathBuf, + }, + Failed { + kind: KwJobKind, + phase: KwPhase, + exit_code: Option, + log_path: PathBuf, + }, + Cancelled { + kind: KwJobKind, + phase: KwPhase, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct KwStatusSnapshot { + pub job: KwJobStatus, +} + +impl KwStatusSnapshot { + pub fn idle() -> Self { + Self { + job: KwJobStatus::Idle, + } + } +} diff --git a/src/main.rs b/src/main.rs index fb3e07b4..32f67777 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,6 +16,8 @@ use clap::Parser; use cli::Cli; use color_eyre::{eyre::eyre, Result}; use config::{bootstrap_parts, ConfigActor}; +#[cfg(unix)] +use infrastructure::process::OsProcess; use infrastructure::{ env::OsEnv, file_system::{FileSystemTrait, OsFileSystem}, @@ -25,6 +27,8 @@ use infrastructure::{ terminal::init, }; use input::{actor::InputActor, event::InputEvent}; +#[cfg(unix)] +use kw::actor::KwActor; use kw::history::{FileKwHistoryStore, KwHistoryStore}; use lore::{ application::{actor::LoreApiActor, cache::CacheTtl, service::LoreService}, @@ -36,7 +40,7 @@ use lore::{ }, }; use render::{actor::RenderActor, ShellRenderService}; -use std::{ops::ControlFlow, sync::Arc}; +use std::{ops::ControlFlow, path::Path, sync::Arc}; use terminal::{actor::TerminalActor, session::CrosstermTerminalSession}; use tokio::sync::mpsc; use tracing::{event, Level}; @@ -101,6 +105,21 @@ async fn main() -> Result<()> { config.data_dir().to_string(), )); + // The kw actor is unix-only because ProcessTrait (process-group kill) + // is; everywhere else there is no handle and App falls back to writing + // apply history directly. + #[cfg(unix)] + let kw_handle = Some(KwActor::spawn( + kw_history.clone(), + Arc::new(OsProcess), + shell_arc.clone(), + fs_arc.clone(), + Arc::new(OsEnv), + Path::new(config.cache_dir()).join("kw_logs"), + )); + #[cfg(not(unix))] + let kw_handle = None; + let render = RenderActor::spawn(Box::new(ShellRenderService::new(shell_arc.clone()))); let lore_api = LoreApiActor::spawn(LoreService::new( @@ -132,6 +151,7 @@ async fn main() -> Result<()> { lore_api.clone(), render.clone(), kw_history.clone(), + kw_handle.clone(), )?; let (app_input_tx, app_input_rx) = mpsc::channel::(64); let input_handle = InputActor::spawn(terminal_handle.clone(), app.input_context()); @@ -144,11 +164,13 @@ async fn main() -> Result<()> { // Shutdown ordering: // 1. AppActor — exits when the user quits (input channel closes) // 2. InputActor — no further terminal input is needed once App is gone - // 3. ConfigActor — no further configuration requests once App is gone - // 4. LoreApiActor — no further requests once App is gone - // 5. RenderActor — no further requests once App is gone - // 6. UiActor — no further scene builds once App is gone - // 7. TerminalActor — restores the terminal last so the screen stays usable + // 3. KwActor — no further kw requests once App is gone; kills any + // running job's process group before stopping + // 4. ConfigActor — no further configuration requests once App is gone + // 5. LoreApiActor — no further requests once App is gone + // 6. RenderActor — no further requests once App is gone + // 7. UiActor — no further scene builds once App is gone + // 8. TerminalActor — restores the terminal last so the screen stays usable // during the steps above AppActor::spawn( app, @@ -163,6 +185,9 @@ async fn main() -> Result<()> { .shutdown() .await .map_err(|e| eyre!("{e}"))?; + if let Some(kw_handle) = kw_handle { + kw_handle.shutdown().await; + } config_handle.shutdown().await; lore_api.shutdown().await; render.shutdown().await;