diff --git a/workshop-baker/Cargo.lock b/workshop-baker/Cargo.lock index 4e3bacc..24173d7 100644 --- a/workshop-baker/Cargo.lock +++ b/workshop-baker/Cargo.lock @@ -5302,6 +5302,7 @@ dependencies = [ "chrono", "duration-str", "lazy_static", + "libc", "log", "os_info", "serde", diff --git a/workshop-baker/Cargo.toml b/workshop-baker/Cargo.toml index 3cfd082..4d75ff4 100644 --- a/workshop-baker/Cargo.toml +++ b/workshop-baker/Cargo.toml @@ -15,10 +15,6 @@ path = "src/main.rs" default = [] integration-tests = [] -[[test]] -name = "engine_integration" -path = "tests/engine_integration.rs" - [dependencies] workshop-engine = { path = "../workshop-engine" } anyhow = "1.0.100" diff --git a/workshop-baker/tests/finalize_hook_integration.rs b/workshop-baker/tests/finalize_hook_integration.rs index 7e0b25c..fd243fc 100644 --- a/workshop-baker/tests/finalize_hook_integration.rs +++ b/workshop-baker/tests/finalize_hook_integration.rs @@ -5,10 +5,10 @@ use std::collections::HashMap; use std::time::Duration; -use workshop_baker::engine::{ExecutionContext, ExecutionEvent}; +use workshop_baker::ExecutionContext; use workshop_baker::finalize::plugin::PluginMap; use workshop_baker::finalize::stage::hook::hook; -use workshop_baker::types::command::CustomCommand; +use workshop_engine::{ExecutionEvent, CustomCommand}; fn create_test_context() -> ExecutionContext { ExecutionContext { @@ -18,7 +18,6 @@ fn create_test_context() -> ExecutionContext { working_dir: std::env::temp_dir(), env_vars: HashMap::new(), timeout: Some(Duration::from_secs(30)), - cgroup_path: None, shell: "bash".to_string(), dry_run: false, privileged: false, diff --git a/workshop-baker/tests/finalize_pipeline_integration.rs b/workshop-baker/tests/finalize_pipeline_integration.rs index db72e23..a8a5fb7 100644 --- a/workshop-baker/tests/finalize_pipeline_integration.rs +++ b/workshop-baker/tests/finalize_pipeline_integration.rs @@ -6,7 +6,8 @@ use std::collections::HashMap; use std::time::Duration; use workshop_baker::cli::Cli; -use workshop_baker::engine::{ExecutionContext, ExecutionEvent}; +use workshop_baker::ExecutionContext; +use workshop_engine::ExecutionEvent; use workshop_baker::finalize::finalize; fn create_test_context() -> ExecutionContext { @@ -17,7 +18,6 @@ fn create_test_context() -> ExecutionContext { working_dir: std::env::temp_dir(), env_vars: HashMap::new(), timeout: Some(Duration::from_secs(30)), - cgroup_path: None, shell: "bash".to_string(), dry_run: false, privileged: false, diff --git a/workshop-baker/tests/prebake_config_integration.rs b/workshop-baker/tests/prebake_config_integration.rs index 48b9f4b..0344c71 100644 --- a/workshop-baker/tests/prebake_config_integration.rs +++ b/workshop-baker/tests/prebake_config_integration.rs @@ -1,5 +1,5 @@ use workshop_baker::prebake::parse; -use workshop_baker::types::builderconfig::BuilderType; +use workshop_baker::prebake::types::builderconfig::BuilderType; #[test] fn test_prebake_config_minimal_yaml() { diff --git a/workshop-baker/tests/types_config_integration.rs b/workshop-baker/tests/types_config_integration.rs index 0ce94ce..4e5227a 100644 --- a/workshop-baker/tests/types_config_integration.rs +++ b/workshop-baker/tests/types_config_integration.rs @@ -1,9 +1,8 @@ -use workshop_baker::types::architecture::Architecture; -use workshop_baker::types::builderconfig::BuilderConfig; -use workshop_baker::types::cache::{CacheDirectory, CacheMode, CacheStrategyType}; -use workshop_baker::types::command::CustomCommand; -use workshop_baker::types::compression::CompressionMethod; -use workshop_baker::types::repology::RepologyEndpoint; +use workshop_baker::prebake::types::architecture::Architecture; +use workshop_baker::prebake::types::builderconfig::BuilderConfig; +use workshop_baker::prebake::types::cache::{CacheDirectory, CacheMode, CacheStrategyType}; +use workshop_baker::finalize::types::compression::CompressionMethod; +use workshop_engine::{CustomCommand, RepologyEndpoint}; #[test] fn test_architecture_from_str() { diff --git a/workshop-engine/Cargo.lock b/workshop-engine/Cargo.lock index c4c70db..fe81128 100644 --- a/workshop-engine/Cargo.lock +++ b/workshop-engine/Cargo.lock @@ -869,6 +869,7 @@ dependencies = [ "chrono", "duration-str", "lazy_static", + "libc", "log", "os_info", "serde", diff --git a/workshop-engine/Cargo.toml b/workshop-engine/Cargo.toml index d91d3f5..2bb9028 100644 --- a/workshop-engine/Cargo.toml +++ b/workshop-engine/Cargo.toml @@ -19,3 +19,7 @@ lazy_static = "1.5.0" log = "0.4.28" os_info = "3.14.0" duration-str = "0.21.0" +libc = "0.2" + +[dev-dependencies] +tokio = { version = "1.48.0", features = ["full"] } diff --git a/workshop-engine/src/child.rs b/workshop-engine/src/child.rs new file mode 100644 index 0000000..4d7fc10 --- /dev/null +++ b/workshop-engine/src/child.rs @@ -0,0 +1,205 @@ +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::io::AsyncRead; + +/// A managed child process, abstracted over the real OS process for testability. +/// +/// This trait represents a spawned process that we can: +/// - Read its stdout/stderr +/// - Wait for it to finish (getting its exit code) +/// - Kill it +/// +/// The real implementation wraps `tokio::process::Child`. +/// Mock implementations are used in unit tests. +pub trait ManagedChild: Send { + /// Returns the OS-assigned process ID, if available. + fn pid(&self) -> Option; + + /// Takes the stdout pipe, if available. + fn take_stdout(&mut self) -> Option>; + + /// Takes the stderr pipe, if available. + fn take_stderr(&mut self) -> Option>; + + /// Waits for the process to exit and returns its exit code. + fn wait(&mut self) -> impl std::future::Future> + Send; + + /// Kills the process. + fn kill(&mut self) -> impl std::future::Future> + Send; +} + +// --------------------------------------------------------------------------- +// Real implementation: wraps tokio::process::Child +// --------------------------------------------------------------------------- + +pub struct TokioChild(pub tokio::process::Child); + +impl ManagedChild for TokioChild { + fn pid(&self) -> Option { + self.0.id() + } + + fn take_stdout(&mut self) -> Option> { + self.0.stdout.take().map(|s| Box::new(s) as Box) + } + + fn take_stderr(&mut self) -> Option> { + self.0.stderr.take().map(|s| Box::new(s) as Box) + } + + async fn wait(&mut self) -> io::Result { + self.0.wait().await.map(|s| s.code().unwrap_or(-1)) + } + + async fn kill(&mut self) -> io::Result<()> { + // Kill the entire process group. When combined with process_group(0) + // in executor.rs, the spawned command is the process group leader, + // so PGID = PID. Sending SIGKILL to -PGID terminates the whole tree. + #[cfg(unix)] + if let Some(pid) = self.0.id() { + let pgid = pid as i32; + let ret = unsafe { libc::kill(-pgid, libc::SIGKILL) }; + if ret != 0 { + let err = io::Error::last_os_error(); + // ESRCH = process group already gone, which is fine + if err.raw_os_error() != Some(libc::ESRCH) { + log::warn!("kill(-{}, SIGKILL) failed: {}", pgid, err); + } + } + } + + // Fallback: also try tokio's per-process kill + let _ = self.0.kill().await; + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// Mock implementation: for testing Engine orchestration logic +// --------------------------------------------------------------------------- + +/// A mock child that returns pre-configured results without spawning a real process. +/// +/// Use `MockChildHandle` for advanced scenarios (timeout simulation, etc.). +pub struct MockChild { + /// The exit code that `wait()` returns. + pub exit_code: i32, + /// Whether `kill()` was called. + pub killed: bool, + /// Simulated stdout content. + pub stdout_data: Vec, + /// Simulated stderr content. + pub stderr_data: Vec, + /// Simulated PID. + pub pid: u32, + /// When true, `wait()` never returns (simulates a hanging process for timeout testing). + pub hang_forever: bool, +} + +impl Default for MockChild { + fn default() -> Self { + Self { + exit_code: 0, + killed: false, + stdout_data: Vec::new(), + stderr_data: Vec::new(), + pid: 12345, + hang_forever: false, + } + } +} + +impl MockChild { + pub fn with_exit_code(code: i32) -> Self { + Self { + exit_code: code, + ..Default::default() + } + } + + pub fn with_output(stdout: &str, stderr: &str) -> Self { + Self { + stdout_data: stdout.as_bytes().to_vec(), + stderr_data: stderr.as_bytes().to_vec(), + ..Default::default() + } + } + + pub fn hanging() -> Self { + Self { + hang_forever: true, + ..Default::default() + } + } +} + +impl ManagedChild for MockChild { + fn pid(&self) -> Option { + Some(self.pid) + } + + fn take_stdout(&mut self) -> Option> { + if self.stdout_data.is_empty() { + None + } else { + let data = std::mem::take(&mut self.stdout_data); + Some(Box::new(VecReader::new(data))) + } + } + + fn take_stderr(&mut self) -> Option> { + if self.stderr_data.is_empty() { + None + } else { + let data = std::mem::take(&mut self.stderr_data); + Some(Box::new(VecReader::new(data))) + } + } + + async fn wait(&mut self) -> io::Result { + if self.hang_forever { + // Park this future indefinitely to simulate a timeout + std::future::pending::<()>().await; + unreachable!() + } + Ok(self.exit_code) + } + + async fn kill(&mut self) -> io::Result<()> { + self.killed = true; + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// VecReader: wraps Vec into tokio::io::AsyncRead +// --------------------------------------------------------------------------- + +pub struct VecReader { + data: Vec, + pos: usize, +} + +impl VecReader { + pub fn new(data: Vec) -> Self { + Self { data, pos: 0 } + } +} + +impl AsyncRead for VecReader { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + if self.pos >= self.data.len() { + return Poll::Ready(Ok(())); + } + let available = &self.data[self.pos..]; + let len = available.len().min(buf.remaining()); + buf.put_slice(&available[..len]); + self.pos += len; + Poll::Ready(Ok(())) + } +} diff --git a/workshop-engine/src/executor.rs b/workshop-engine/src/executor.rs index 795ad61..3ff3af5 100644 --- a/workshop-engine/src/executor.rs +++ b/workshop-engine/src/executor.rs @@ -6,53 +6,16 @@ use tokio::process::Command; use tokio::sync::mpsc; use tokio::time::timeout; +use crate::child::{ManagedChild, TokioChild}; use crate::pm::PackageManager; use crate::types::{ - CgroupManager, DeltaExecutionContext, EventSender, ExecutionContext, ExecutionError, - ExecutionEvent, ExecutionResult, ResourceLimits, StreamType, + DeltaExecutionContext, EventSender, ExecutionContext, ExecutionError, + ExecutionEvent, ExecutionResult, StreamType, }; impl crate::types::Engine { pub fn new() -> Self { - Self { - cgroup_manager: None, - } - } - - pub fn with_cgroup(limits: ResourceLimits, build_id: &str) -> Result { - let cgroup_manager = - CgroupManager::new(build_id).map_err(|e| format!("Failed to create cgroup: {}", e))?; - - if let Some(mem) = limits.memory_bytes { - cgroup_manager - .set_memory_limit(mem) - .map_err(|e| format!("Failed to set memory limit: {}", e))?; - } - - if let Some(weight) = limits.cpu_weight { - cgroup_manager - .set_cpu_weight(weight) - .map_err(|e| format!("Failed to set cpu weight: {}", e))?; - } - - if let Some(max) = limits.cpu_max_us { - cgroup_manager - .set_cpu_max(max, limits.cpu_period_us) - .map_err(|e| format!("Failed to set cpu max: {}", e))?; - } - - Ok(Self { - cgroup_manager: Some(cgroup_manager), - }) - } - - fn apply_cgroup(&self, pid: u32) -> Result<(), ExecutionError> { - if let Some(ref cgroup) = self.cgroup_manager { - cgroup.add_process(pid).map_err(|e| { - ExecutionError::IoError(format!("Failed to add process to cgroup: {}", e)) - })?; - } - Ok(()) + Self } pub async fn execute_script( @@ -129,9 +92,49 @@ impl crate::types::Engine { .stderr(Stdio::piped()); } - pub async fn execute( +/// Execute a pre-configured command, managing the full lifecycle. +/// +/// This method: +/// 1. Configures the command with context (env vars, working dir, stdio) +/// 2. Creates a new process group on Unix for process-tree isolation +/// 3. Returns early in dry_run mode +/// 4. Spawns the OS process +/// 5. Delegates to `execute_child` for wait/event/output management +pub async fn execute( + &self, + command: Command, + ctx: &ExecutionContext, + event_tx: &EventSender, +) -> Result { + let mut command = command; + self.construct_command(&mut command, ctx); + + if ctx.dry_run { + log::info!("[DRY_RUN] {:?}", command); + return Ok(ExecutionResult { + ..Default::default() + }); + } + + // Create a new process group so that kill() can terminate the + // entire process tree (bash + its children) on timeout. + #[cfg(unix)] + command.process_group(0); + + let child = command.spawn().map_err(|e| { + ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e)) + })?; + + self.execute_child(TokioChild(child), ctx, event_tx).await + } + + /// Core execution logic, abstracted over [`ManagedChild`] for testability. + /// + /// Handles: timeout, event emission, stdout/stderr capture, result construction. + /// Does NOT spawn processes itself — that is done by [`execute`]. + pub async fn execute_child( &self, - command: Command, + mut child: impl ManagedChild, ctx: &ExecutionContext, event_tx: &EventSender, ) -> Result { @@ -145,25 +148,6 @@ impl crate::types::Engine { }) .await; } - let mut command = command; - - self.construct_command(&mut command, ctx); - - if ctx.dry_run { - log::info!("[DRY_RUN] {:?}", command); - return Ok(ExecutionResult { - ..Default::default() - }); - } - - let mut child = command.spawn().map_err(|e| { - ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e)) - })?; - - let pid = child - .id() - .ok_or_else(|| ExecutionError::IoError("Failed to get PID".to_string()))?; - self.apply_cgroup(pid)?; let task_id = ctx.task_id.clone(); let (stdout_tx, mut stdout_rx) = mpsc::channel::(1024); @@ -171,11 +155,10 @@ impl crate::types::Engine { let mut stdout_buf = String::new(); let mut stderr_buf = String::new(); - let mut child_stdout = child.stdout.take(); - let mut child_stderr = child.stderr.take(); + let mut child_stdout = child.take_stdout(); + let mut child_stderr = child.take_stderr(); let tx_clone = stdout_tx.clone(); - let _task_id_clone = task_id.clone(); let stdout_handle = tokio::spawn(async move { if let Some(mut stdout) = child_stdout.take() { let mut reader = tokio::io::BufReader::new(&mut stdout); @@ -215,9 +198,9 @@ impl crate::types::Engine { } }); - let status = if let Some(timeout_duration) = ctx.timeout { + let exit_code = if let Some(timeout_duration) = ctx.timeout { match timeout(timeout_duration, child.wait()).await { - Ok(Ok(status)) => status, + Ok(Ok(code)) => code, Ok(Err(e)) => { return Err(ExecutionError::IoError(format!( "Failed to wait for process: {}", @@ -289,7 +272,6 @@ impl crate::types::Engine { } } - let exit_code = status.code().unwrap_or(-1); let duration = start_time.elapsed(); let result = ExecutionResult { @@ -299,7 +281,6 @@ impl crate::types::Engine { duration, stdout: stdout_buf, stderr: stderr_buf, - resource_usage: None, }; if let Some(tx) = event_tx { @@ -342,13 +323,17 @@ impl crate::types::Engine { duration: std::time::Duration::default(), stdout: String::new(), stderr: String::new(), - resource_usage: None, }; - for cmd in cmds { + for (i, cmd) in cmds.into_iter().enumerate() { log::debug!("Installing dependency with command: {:?}", cmd); let result = self.execute(cmd, ctx, event_tx).await?; - // TODO: Aggregate results properly instead of just taking the last one - combined_result.exit_code = result.exit_code; + // `execute()` returns Err for non-zero exit codes, so this code only + // runs for successful commands (exit_code == 0). We keep the first + // command's exit_code as it's semantically the "primary" result; + // duration and output are accumulated across all commands. + if i == 0 { + combined_result.exit_code = result.exit_code; + } combined_result.success &= result.success; combined_result.duration += result.duration; combined_result.stdout.push_str(&result.stdout); @@ -356,15 +341,6 @@ impl crate::types::Engine { } Ok(combined_result) } - - pub async fn cleanup(&self) -> Result<(), String> { - if let Some(ref cgroup) = self.cgroup_manager { - cgroup - .destroy() - .map_err(|e| format!("Failed to destroy cgroup: {}", e))?; - } - Ok(()) - } } impl Default for crate::types::Engine { @@ -376,131 +352,19 @@ impl Default for crate::types::Engine { #[cfg(test)] mod tests { use super::*; - use std::collections::VecDeque; + use crate::child::MockChild; + use crate::Engine; use std::path::PathBuf; - use std::sync::{Arc, Mutex}; use std::time::Duration; - #[derive(Clone, Debug)] - struct CapturedCall { - program: String, - args: Vec, - env_vars: std::collections::HashMap, - working_dir: PathBuf, - } - - #[derive(Clone)] - struct MockExecutor { - calls: Arc>>, - } - - impl MockExecutor { - fn new() -> Self { - Self { - calls: Arc::new(Mutex::new(VecDeque::new())), - } - } - - fn capture( - &self, - program: String, - args: Vec, - env_vars: std::collections::HashMap, - working_dir: PathBuf, - ) { - self.calls.lock().unwrap().push_back(CapturedCall { - program, - args, - env_vars, - working_dir, - }); - } - - fn pop(&self) -> Option { - self.calls.lock().unwrap().pop_front() - } - } - - struct TestExecutor { - mock: MockExecutor, - } - - impl TestExecutor { - fn new() -> Self { - Self { - mock: MockExecutor::new(), - } - } - - async fn execute_command( - &self, - cmd: String, - ctx: &ExecutionContext, - _event_tx: EventSender, - ) -> Result { - let mut env_vars = ctx.env_vars.clone(); - env_vars.insert("WS_TASKID".to_string(), ctx.task_id.clone()); - env_vars.insert("WS_PIPELINE".to_string(), ctx.pipeline_name.clone()); - env_vars.insert("WS_USERNAME".to_string(), ctx.username.clone()); - - self.mock.capture( - ctx.shell.clone(), - vec!["-c".to_string(), cmd.clone()], - env_vars, - ctx.working_dir.clone(), - ); - - Ok(ExecutionResult { - task_id: ctx.task_id.clone(), - exit_code: 0, - success: true, - duration: Duration::from_millis(10), - stdout: String::new(), - stderr: String::new(), - resource_usage: None, - }) - } - - async fn execute_script( - &self, - script_path: &PathBuf, - ctx: &ExecutionContext, - _event_tx: EventSender, - ) -> Result { - let abs_path = if script_path.is_absolute() { - script_path.clone() - } else { - ctx.working_dir.join(script_path) - }; - - self.mock.capture( - ctx.shell.clone(), - vec!["-c".to_string(), abs_path.to_string_lossy().into_owned()], - ctx.env_vars.clone(), - ctx.working_dir.clone(), - ); - - Ok(ExecutionResult { - task_id: ctx.task_id.clone(), - exit_code: 0, - success: true, - duration: Duration::from_millis(10), - stdout: String::new(), - stderr: String::new(), - resource_usage: None, - }) - } - } - - fn create_test_context() -> ExecutionContext { + fn test_ctx() -> ExecutionContext { ExecutionContext { - task_id: "test-task-1".to_string(), + task_id: "test-task".to_string(), pipeline_name: "test-pipeline".to_string(), username: "testuser".to_string(), working_dir: PathBuf::from("/tmp"), env_vars: std::collections::HashMap::new(), - timeout: Some(Duration::from_secs(30)), - cgroup_path: None, + timeout: None, shell: "bash".to_string(), dry_run: false, privileged: false, @@ -508,152 +372,245 @@ mod tests { } } - #[tokio::test] - async fn test_execute_command_captures_args() { - let executor = TestExecutor::new(); - let ctx = create_test_context(); + // ------------------------------------------------------------------ + // TaskStarted event + // ------------------------------------------------------------------ - let result = executor - .execute_command("echo hello".to_string(), &ctx, None) - .await; + #[tokio::test] + async fn send_task_started_event_when_channel_provided() { + let engine = Engine::new(); + let (tx, mut rx) = mpsc::channel(64); + let child = MockChild::with_exit_code(0); + + let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await; + + assert!(matches!( + rx.try_recv(), + Ok(ExecutionEvent::TaskStarted { .. }) + )); + } + + #[tokio::test] + async fn skip_task_started_event_when_channel_is_none() { + let engine = Engine::new(); + let child = MockChild::with_exit_code(0); + + let result = engine.execute_child(child, &test_ctx(), &None).await; assert!(result.is_ok()); - - let call = executor.mock.pop(); - assert!(call.is_some()); - - let call = call.unwrap(); - assert_eq!(call.program, "bash"); - assert!(call.args.contains(&"-c".to_string())); - assert!(call.args.contains(&"echo hello".to_string())); } - #[tokio::test] - async fn test_execute_script_captures_path() { - let executor = TestExecutor::new(); - let ctx = create_test_context(); + // ------------------------------------------------------------------ + // Exit code → success / failure mapping + // ------------------------------------------------------------------ - let script_path = PathBuf::from("/tmp/test_script.sh"); - let result = executor.execute_script(&script_path, &ctx, None).await; + #[tokio::test] + async fn exit_zero_returns_ok() { + let engine = Engine::new(); + let child = MockChild::with_exit_code(0); + + let result = engine.execute_child(child, &test_ctx(), &None).await; assert!(result.is_ok()); - - let call = executor.mock.pop().unwrap(); - assert!(call.args.contains(&"-c".to_string())); - assert!(call.args.iter().any(|a| a.contains("test_script.sh"))); + assert_eq!(result.unwrap().exit_code, 0); } #[tokio::test] - async fn test_execute_command_with_env_vars() { - let executor = TestExecutor::new(); - let mut ctx = create_test_context(); - ctx.env_vars - .insert("TEST_VAR".to_string(), "test_value".to_string()); + async fn exit_nonzero_returns_err() { + let engine = Engine::new(); + let child = MockChild::with_exit_code(1); - executor - .execute_command("echo $TEST_VAR".to_string(), &ctx, None) - .await - .unwrap(); + let result = engine.execute_child(child, &test_ctx(), &None).await; - let call = executor.mock.pop().unwrap(); - assert_eq!( - call.env_vars.get("TEST_VAR"), - Some(&"test_value".to_string()) - ); + assert!(result.is_err()); + assert!(matches!(result.unwrap_err(), ExecutionError::ExecutionFailed(_))); } #[tokio::test] - async fn test_execute_includes_ws_environment() { - let executor = TestExecutor::new(); - let ctx = create_test_context(); + async fn exit_nonzero_sends_task_failed_event() { + let engine = Engine::new(); + let (tx, mut rx) = mpsc::channel(64); + let child = MockChild::with_exit_code(42); - executor - .execute_command("ls".to_string(), &ctx, None) - .await - .unwrap(); + let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await; - let call = executor.mock.pop().unwrap(); - - assert!(call.env_vars.contains_key("WS_TASKID")); - assert!(call.env_vars.contains_key("WS_PIPELINE")); - assert!(call.env_vars.contains_key("WS_USERNAME")); + // Skip TaskStarted + let _ = rx.recv().await; + let event = rx.recv().await; + assert!(matches!(event, Some(ExecutionEvent::TaskFailed { .. }))); + if let Some(ExecutionEvent::TaskFailed { error, .. }) = event { + assert!(error.contains("42"), "error should mention exit code 42"); + } } #[tokio::test] - async fn test_execute_command_preserves_working_dir() { - let executor = TestExecutor::new(); - let mut ctx = create_test_context(); - ctx.working_dir = PathBuf::from("/custom/path"); + async fn exit_zero_sends_task_completed_event() { + let engine = Engine::new(); + let (tx, mut rx) = mpsc::channel(64); + let child = MockChild::with_exit_code(0); - executor - .execute_command("pwd".to_string(), &ctx, None) - .await - .unwrap(); + let _ = engine.execute_child(child, &test_ctx(), &Some(tx)).await; - let call = executor.mock.pop().unwrap(); - assert_eq!(call.working_dir, PathBuf::from("/custom/path")); + let _ = rx.recv().await; // skip TaskStarted + let event = rx.recv().await; + assert!(matches!(event, Some(ExecutionEvent::TaskCompleted { .. }))); + } + + // ------------------------------------------------------------------ + // Stdout / stderr capture + // ------------------------------------------------------------------ + + #[tokio::test] + async fn captures_stdout() { + let engine = Engine::new(); + let child = MockChild::with_output("hello world\n", ""); + let ctx = test_ctx(); + + let result = engine.execute_child(child, &ctx, &None).await.unwrap(); + + assert_eq!(result.stdout, "hello world\n"); } #[tokio::test] - async fn test_local_executor_new_creates_default() { - let executor = super::super::types::Engine::new(); - assert!(executor.cgroup_manager.is_none()); + async fn captures_stderr() { + let engine = Engine::new(); + let child = MockChild::with_output("", "error msg\n"); + let ctx = test_ctx(); + + let result = engine.execute_child(child, &ctx, &None).await.unwrap(); + + assert_eq!(result.stderr, "error msg\n"); } #[tokio::test] - async fn test_execution_context_default() { - let ctx = ExecutionContext::default(); + async fn captures_both_streams() { + let engine = Engine::new(); + let child = MockChild::with_output("out\n", "err\n"); + let ctx = test_ctx(); - assert!(ctx.task_id.is_empty()); - assert_eq!(ctx.shell, "bash"); + let result = engine.execute_child(child, &ctx, &None).await.unwrap(); + + assert_eq!(result.stdout, "out\n"); + assert_eq!(result.stderr, "err\n"); + } + + // ------------------------------------------------------------------ + // OutputChunk events + // ------------------------------------------------------------------ + + #[tokio::test] + async fn sends_output_chunk_events() { + let engine = Engine::new(); + let (tx, mut rx) = mpsc::channel(64); + let child = MockChild::with_output("chunk1\nchunk2\n", ""); + let ctx = test_ctx(); + + let _ = engine.execute_child(child, &ctx, &Some(tx)).await; + + let _ = rx.recv().await; // skip TaskStarted + + let mut found_stdout = false; + while let Ok(event) = rx.try_recv() { + if matches!(&event, ExecutionEvent::OutputChunk { stream: StreamType::Stdout, .. }) { + found_stdout = true; + } + } + assert!(found_stdout, "Should have received at least one OutputChunk for stdout"); + } + + // ------------------------------------------------------------------ + // Timeout handling + // ------------------------------------------------------------------ + + #[tokio::test] + async fn timeout_kills_child_and_returns_err() { + let engine = Engine::new(); + let (tx, mut rx) = mpsc::channel(64); + let child = MockChild::hanging(); + let mut ctx = test_ctx(); + ctx.timeout = Some(Duration::from_millis(10)); + + let result = engine.execute_child(child, &ctx, &Some(tx)).await; + + assert!(result.is_err()); + assert!(matches!(result.unwrap_err(), ExecutionError::Timeout)); + + // Should have received TaskFailed for timeout + let _ = rx.recv().await; // skip TaskStarted + let event = rx.recv().await; + assert!(matches!(event, Some(ExecutionEvent::TaskFailed { error, .. }) if error == "Timeout")); + } + + // ------------------------------------------------------------------ + // Result fields + // ------------------------------------------------------------------ + + #[tokio::test] + async fn result_contains_correct_task_id() { + let engine = Engine::new(); + let child = MockChild::with_exit_code(0); + let ctx = test_ctx(); + + let result = engine.execute_child(child, &ctx, &None).await.unwrap(); + + assert_eq!(result.task_id, "test-task"); } #[tokio::test] - async fn test_execution_result_success() { - let result = ExecutionResult { - task_id: "test".to_string(), - exit_code: 0, - success: true, - duration: Duration::from_secs(1), - stdout: "output".to_string(), - stderr: "".to_string(), - resource_usage: None, - }; + async fn result_contains_duration() { + let engine = Engine::new(); + let child = MockChild::with_exit_code(0); + let ctx = test_ctx(); - assert!(result.success); - assert_eq!(result.exit_code, 0); + let result = engine.execute_child(child, &ctx, &None).await.unwrap(); + + assert!(result.duration.as_nanos() > 0); } - #[tokio::test] - async fn test_execution_result_failure() { - let result = ExecutionResult { - task_id: "test".to_string(), - exit_code: 1, - success: false, - duration: Duration::from_secs(1), - stdout: "".to_string(), - stderr: "error".to_string(), - resource_usage: None, - }; + // ------------------------------------------------------------------ + // dry_run does not execute_child + // ------------------------------------------------------------------ - assert!(!result.success); - assert_eq!(result.exit_code, 1); + #[tokio::test] + async fn dry_run_returns_ok_without_spawning() { + // This tests the `execute()` path: with dry_run=true it should + // return before reaching spawn / execute_child. + let engine = Engine::new(); + let mut ctx = test_ctx(); + ctx.dry_run = true; + + let cmd = Command::new("true"); + let result = engine.execute(cmd, &ctx, &None).await; + + assert!(result.is_ok()); + let res = result.unwrap(); + assert!(res.success); + assert_eq!(res.stdout, ""); + assert_eq!(res.stderr, ""); } - #[tokio::test] - async fn test_execution_error_variants() { - let _err1 = ExecutionError::InvalidCommand("test".to_string()); - let _err2 = ExecutionError::ScriptNotFound(PathBuf::from("/nonexistent")); - let _err3 = ExecutionError::PackageManagerNotFound; - let _err4 = ExecutionError::Timeout; - } + // ------------------------------------------------------------------ + // EventSender=None does not panic + // ------------------------------------------------------------------ #[tokio::test] - async fn test_resource_limits_default() { - let limits = ResourceLimits::default(); + async fn no_event_channel_does_not_panic() { + let engine = Engine::new(); - assert!(limits.memory_bytes.is_none()); - assert!(limits.cpu_weight.is_none()); - assert_eq!(limits.cpu_period_us, 0); + // Various exit codes + for code in [0, 1, 127] { + let child = MockChild::with_exit_code(code); + let _ = engine.execute_child(child, &test_ctx(), &None).await; + } + + // With output + let child = MockChild::with_output("data", "err"); + let _ = engine.execute_child(child, &test_ctx(), &None).await; + + // Timeout + let mut ctx = test_ctx(); + ctx.timeout = Some(Duration::from_millis(10)); + let child = MockChild::hanging(); + let _ = engine.execute_child(child, &ctx, &None).await; } } diff --git a/workshop-engine/src/lib.rs b/workshop-engine/src/lib.rs index 97b00c5..267760b 100644 --- a/workshop-engine/src/lib.rs +++ b/workshop-engine/src/lib.rs @@ -1,4 +1,4 @@ -pub mod cgroups; +pub mod child; pub mod command; pub mod error; pub mod executor; @@ -8,13 +8,16 @@ pub mod time; pub mod types; pub mod upm; +// Re-export child types for convenience +pub use child::{ManagedChild, TokioChild}; + // Convenience re-exports pub use error::{ExecutionError, HasExitCode, EXITCODE_OK, EXITCODE_GENERAL_ERROR, EXITCODE_INVALID_ARGS, EXITCODE_IO_ERROR, EXITCODE_PARSE_ERROR, EXITCODE_EXECUTION_ERROR, EXITCODE_TIMEOUT, EXITCODE_PRIV_DROP_FAILED}; pub use types::{Engine, EventSender, ExecutionContext, ExecutionEvent, ExecutionResult, - StreamType, ResourceLimits, ResourceUsage, DeltaExecutionContext}; + StreamType, DeltaExecutionContext}; pub use command::CustomCommand; pub use repology::RepologyEndpoint; pub use time::{parse_duration_to_ms, deserialize_duration_ms, deserialize_option_duration_ms}; diff --git a/workshop-engine/src/types.rs b/workshop-engine/src/types.rs index 9a0e468..8ce26c3 100644 --- a/workshop-engine/src/types.rs +++ b/workshop-engine/src/types.rs @@ -1,20 +1,11 @@ -use serde::{Deserialize, Serialize}; +use serde::Serialize; use std::collections::HashMap; use std::path::PathBuf; use std::time::Duration; use tokio::sync::mpsc; -pub use crate::cgroups::CgroupManager; pub use crate::error::{EXITCODE_EXECUTION_ERROR, EXITCODE_IO_ERROR, ExecutionError}; -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct ResourceUsage { - pub cpu_time_secs: f64, - pub max_memory_kb: u64, - pub io_read_kb: u64, - pub io_write_kb: u64, -} - #[derive(Debug, Clone)] pub struct ExecutionContext { pub task_id: String, @@ -23,7 +14,6 @@ pub struct ExecutionContext { pub working_dir: PathBuf, pub env_vars: HashMap, pub timeout: Option, - pub cgroup_path: Option, pub shell: String, pub dry_run: bool, pub privileged: bool, @@ -46,7 +36,6 @@ impl Default for ExecutionContext { working_dir: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")), env_vars: HashMap::new(), timeout: None, - cgroup_path: None, shell: String::from("bash"), dry_run: false, privileged: false, @@ -55,14 +44,6 @@ impl Default for ExecutionContext { } } -#[derive(Debug, Clone, Default)] -pub struct ResourceLimits { - pub memory_bytes: Option, - pub cpu_weight: Option, - pub cpu_max_us: Option, - pub cpu_period_us: u64, -} - #[derive(Debug, Clone, Serialize)] pub struct ExecutionResult { pub task_id: String, @@ -71,7 +52,6 @@ pub struct ExecutionResult { pub duration: Duration, pub stdout: String, pub stderr: String, - pub resource_usage: Option, } impl Default for ExecutionResult { @@ -83,7 +63,6 @@ impl Default for ExecutionResult { duration: Duration::default(), stdout: String::new(), stderr: String::new(), - resource_usage: None, } } } @@ -117,6 +96,4 @@ pub enum ExecutionEvent { pub type EventSender = Option>; -pub struct Engine { - pub cgroup_manager: Option, -} +pub struct Engine; diff --git a/workshop-baker/tests/engine_integration.rs b/workshop-engine/tests/engine_integration.rs similarity index 68% rename from workshop-baker/tests/engine_integration.rs rename to workshop-engine/tests/engine_integration.rs index 7dfaa45..aa7e417 100644 --- a/workshop-baker/tests/engine_integration.rs +++ b/workshop-engine/tests/engine_integration.rs @@ -1,7 +1,7 @@ -//! Integration tests for workshop-executor engine +//! Integration tests for workshop-engine //! //! These tests execute real commands and verify results. -//! Run with: cargo test --test engine_integration +//! Run with: cargo test --test engine_integration -p workshop-engine //! //! Note: These are TRUE integration tests - they run in a separate test binary //! and execute actual commands on the system. @@ -9,7 +9,7 @@ use std::collections::HashMap; use std::path::PathBuf; use std::time::Duration; -use workshop_baker::engine::{Engine, ExecutionContext}; +use workshop_engine::{Engine, ExecutionContext}; fn create_executor() -> Engine { Engine::new() @@ -23,7 +23,6 @@ fn create_context() -> ExecutionContext { working_dir: std::env::temp_dir(), env_vars: HashMap::new(), timeout: Some(Duration::from_secs(30)), - cgroup_path: None, shell: "bash".to_string(), dry_run: false, privileged: false, @@ -173,7 +172,6 @@ async fn test_execute_respects_working_directory() { working_dir: PathBuf::from("/tmp"), env_vars: HashMap::new(), timeout: Some(Duration::from_secs(30)), - cgroup_path: None, shell: "bash".to_string(), standalone: false, ..Default::default() @@ -204,3 +202,66 @@ async fn test_execute_captures_stderr() { assert!(result.success); assert!(result.stderr.contains("error")); } + +/// Test: with `process_group(0)`, kill() propagates to the entire process tree. +/// +/// We spawn a bash that creates a background sleep. With process_group(0), +/// bash is the process group leader. When we kill(-PGID), the sleep should die too. +/// +/// This test uses tokio::process::Command + process_group(0) directly +/// (bypassing Engine) to verify the kernel-level behavior. +#[tokio::test] +async fn test_process_group_kill_kills_subprocesses() { + use std::os::unix::process::CommandExt; + use tokio::io::AsyncBufReadExt; + use tokio::process::Command; + use std::process::Stdio; + + let mut cmd = Command::new("bash"); + cmd.args(["-c", "sleep 999 & echo $!"]); + cmd.process_group(0); + cmd.stdin(Stdio::null()); + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + + let mut child = cmd.spawn().expect("spawn failed"); + let bash_pid = child.id().expect("no pid"); + eprintln!("[test] bash PID: {}", bash_pid); + + // Read the background sleep's PID from stdout + let stdout = child.stdout.take().expect("stdout pipe"); + let mut reader = tokio::io::BufReader::new(stdout); + let mut pid_line = String::new(); + reader.read_line(&mut pid_line).await.expect("read pid line"); + let sleep_pid: u32 = pid_line.trim().parse().expect("valid sleep PID"); + eprintln!("[test] sleep PID: {}", sleep_pid); + + // Verify sleep is alive before kill + let alive_before = std::path::Path::new(&format!("/proc/{}", sleep_pid)).exists(); + eprintln!("[test] sleep alive before kill: {}", alive_before); + + // Kill by sending SIGKILL to the NEGATIVE PGID (entire process group) + unsafe { + libc::kill(-(bash_pid as i32), libc::SIGKILL); + } + child.wait().await.expect("wait failed"); + eprintln!("[test] kill+wait done"); + + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + + // Check if sleep survived — it SHOULD be dead now + let alive_after = std::path::Path::new(&format!("/proc/{}", sleep_pid)).exists(); + eprintln!("[test] sleep alive after kill: {}", alive_after); + + // Cleanup just in case + let _ = std::process::Command::new("kill") + .arg("-9") + .arg(sleep_pid.to_string()) + .spawn(); + + assert!(alive_before, "Test setup: sleep was never created"); + assert!( + !alive_after, + "sleep survived process-group kill! process_group(0) fix did not work." + ); +}