diff --git a/rust/crates/rqd/resources/openrqd.service b/rust/crates/rqd/resources/openrqd.service index 6f7d766cab..3a0e22ebe6 100644 --- a/rust/crates/rqd/resources/openrqd.service +++ b/rust/crates/rqd/resources/openrqd.service @@ -14,6 +14,12 @@ SyslogIdentifier=openrqd TimeoutSec=5 Restart=on-failure RestartSec=30 +# Only signal the main rqd process on stop/restart. With the default +# KillMode=control-group, systemd would SIGTERM (and after TimeoutSec, SIGKILL) +# every process in the unit's cgroup - including running frames - which defeats +# rqd's live-restart frame recovery. Frames run in their own session and are +# reattached by the new rqd instance through their on-disk snapshots. +KillMode=process [Install] WantedBy=multi-user.target diff --git a/rust/crates/rqd/src/frame/frame_cmd.rs b/rust/crates/rqd/src/frame/frame_cmd.rs index a1d28117e6..ffa7d64fc9 100644 --- a/rust/crates/rqd/src/frame/frame_cmd.rs +++ b/rust/crates/rqd/src/frame/frame_cmd.rs @@ -93,44 +93,57 @@ impl FrameCmdBuilder { }; // If an exit_file_path is passed, build a script that traps the inner command and write its - // output to the exit_file_path + // output to the exit_file_path. + // + // The wrapper runs the command in the background and `wait`s for it: bash only runs trap + // handlers between foreground commands, so a foreground child would make the wrapper deaf + // to signals until the command finished. The background+wait pattern lets a trapped signal + // interrupt `wait` immediately, get forwarded to the command, after which the wrapper + // resumes waiting for the command's real exit status. + // + // The exit status is written to the exit file with a write-then-rename so a recovering RQD + // can never observe a partially written file: either the file does not exist yet, or it + // holds the complete status. let script = match &self.exit_file_path { Some(exit_file_path) => format!( - r#"#!{} - wait_for_output() {{ - # Wait for the command to complete - wait $command_pid - exit_code=$1 - - # Write the exit code to the specified file - echo $exit_code > {} - exit $exit_code - }} - - # Function to handle signals - handle_signal() {{ - local signal=$1 - # Forward the signal to the child process if it exists - if [ -n "$command_pid" ] && kill -0 $command_pid 2>/dev/null; then - kill -$signal $command_pid - wait_for_output $signal - fi - }} - - # Set up signal handling - trap 'handle_signal TERM' SIGTERM - trap 'handle_signal INT' SIGINT - trap 'handle_signal HUP' SIGHUP - {} - - # Start the command and get its PID - eval '{}' + r#"#!{shell} +# Forward a trapped signal to the command +handle_signal() {{ + local signal=$1 + if [ -n "$command_pid" ] && kill -0 $command_pid 2>/dev/null; then + kill -$signal $command_pid 2>/dev/null + fi +}} + +# Set up signal handling +trap 'handle_signal TERM' SIGTERM +trap 'handle_signal INT' SIGINT +trap 'handle_signal HUP' SIGHUP +{add_user} + +# Start the command in the background and wait for it, so traps fire promptly +eval '{cmd_str}' & +command_pid=$! +wait $command_pid +exit_code=$? + +# `wait` returns 128+signal when interrupted by a trapped signal while the command +# is still alive; keep waiting until the command has really exited. `kill -0` also +# succeeds while the command is an unreaped zombie, in which case the extra `wait` +# reaps it and returns its real exit status. +while [ $exit_code -gt 128 ] && kill -0 $command_pid 2>/dev/null; do + wait $command_pid exit_code=$? - command_pid=$! +done - wait_for_output $exit_code - "#, - self.shell, exit_file_path, add_user, cmd_str +# Atomically write the exit code to the exit file +echo $exit_code > {exit_file_path}.tmp && mv {exit_file_path}.tmp {exit_file_path} +exit $exit_code +"#, + shell = self.shell, + exit_file_path = exit_file_path, + add_user = add_user, + cmd_str = cmd_str ), None => format!( r#"#!{} @@ -266,8 +279,7 @@ impl FrameCmdBuilder { #[cfg(target_os = "linux")] pub fn with_exit_file(&mut self, exit_file_path: String) -> &mut Self { - // Meant for the recovery mode feature. Which is disabled on linux for not being stable - // self.exit_file_path = Some(exit_file_path); + self.exit_file_path = Some(exit_file_path); self } @@ -284,3 +296,157 @@ impl FrameCmdBuilder { self } } + +#[cfg(test)] +#[cfg(any(target_os = "linux", target_os = "macos"))] +mod tests { + use super::FrameCmdBuilder; + use std::process::Command as StdCommand; + + struct BuiltScript { + _temp: tempfile::TempDir, + entrypoint: String, + exit_file: String, + script: String, + } + + fn build_script(frame_cmd: &str) -> BuiltScript { + let temp = tempfile::tempdir().unwrap(); + let entrypoint = temp + .path() + .join("entrypoint.sh") + .to_string_lossy() + .to_string(); + let exit_file = temp + .path() + .join("exit_status") + .to_string_lossy() + .to_string(); + let shell = "/bin/bash".to_string(); + let mut builder = FrameCmdBuilder::new(&shell, entrypoint.clone()); + builder + .with_frame_cmd(frame_cmd.to_string()) + .with_exit_file(exit_file.clone()); + let (_cmd, script) = builder.build().unwrap(); + BuiltScript { + _temp: temp, + entrypoint, + exit_file, + script, + } + } + + /// The exit-file harness must be active on every unix platform. This is the regression + /// guard for the era when `with_exit_file` was a no-op on Linux, which silently disabled + /// frame recovery there. + #[test] + fn test_exit_file_enabled_on_this_platform() { + let built = build_script("echo hello"); + assert!( + built.script.contains(&built.exit_file), + "generated script must reference the exit file: {}", + built.script + ); + } + + #[test] + fn test_script_structure() { + let built = build_script("echo hello"); + let script = &built.script; + + // The command must run in the background: bash defers trap handlers while a + // foreground child runs, so a foreground command would make the wrapper deaf to + // kill requests until the frame finished on its own. + assert!( + script.contains("eval 'echo hello' &"), + "command must run in the background: {script}" + ); + assert!( + script.contains("wait $command_pid"), + "wrapper must wait for the background command: {script}" + ); + // Signals must be forwarded to the frame process. + for trap in [ + "trap 'handle_signal TERM' SIGTERM", + "trap 'handle_signal INT' SIGINT", + "trap 'handle_signal HUP' SIGHUP", + ] { + assert!(script.contains(trap), "missing {trap}: {script}"); + } + // The exit status must be written atomically (write to temp + rename) so a + // recovering RQD can never read a partially written status. + assert!( + script.contains(&format!( + "echo $exit_code > {exit}.tmp && mv {exit}.tmp {exit}", + exit = built.exit_file + )), + "exit file must be written atomically: {script}" + ); + + // The entrypoint file must be executable. + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(&built.entrypoint) + .unwrap() + .permissions() + .mode(); + assert_eq!(mode & 0o111, 0o111, "entrypoint must be executable"); + } + + /// Executes the generated wrapper and checks that a plain exit code is both propagated + /// as the wrapper's own exit status and persisted in the exit file. + #[test] + fn test_script_propagates_and_persists_exit_code() { + let built = build_script("exit 7"); + + let status = StdCommand::new(&built.entrypoint) + .status() + .expect("entrypoint should execute"); + assert_eq!(status.code(), Some(7)); + + let persisted = std::fs::read_to_string(&built.exit_file).unwrap(); + assert_eq!(persisted.trim(), "7"); + } + + /// SIGTERM delivered to the wrapper alone (not the whole process group) must be + /// forwarded to the frame command, and the resulting 128+15 status must be persisted. + /// This is what keeps kill requests working for frames that survived an RQD restart. + #[test] + fn test_script_forwards_sigterm_and_persists_status() { + let built = build_script("sleep 30"); + + let mut child = StdCommand::new(&built.entrypoint) + .spawn() + .expect("entrypoint should spawn"); + + // Wait until the wrapper's background command exists: traps are installed before + // the command is started, so its presence proves the wrapper is ready for signals. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + let listed = StdCommand::new("pgrep") + .args(["-P", &child.id().to_string()]) + .output() + .expect("pgrep should run"); + if !String::from_utf8_lossy(&listed.stdout).trim().is_empty() { + break; + } + assert!( + std::time::Instant::now() < deadline, + "wrapper never started its background command" + ); + std::thread::sleep(std::time::Duration::from_millis(50)); + } + + let pid = nix::unistd::Pid::from_raw(child.id() as i32); + nix::sys::signal::kill(pid, nix::sys::signal::Signal::SIGTERM).unwrap(); + + let status = child.wait().unwrap(); + assert_eq!( + status.code(), + Some(143), + "wrapper should exit with 128+SIGTERM after forwarding the signal" + ); + + let persisted = std::fs::read_to_string(&built.exit_file).unwrap(); + assert_eq!(persisted.trim(), "143"); + } +} diff --git a/rust/crates/rqd/src/frame/running_frame.rs b/rust/crates/rqd/src/frame/running_frame.rs index 8475269c06..f83ff4f8e6 100644 --- a/rust/crates/rqd/src/frame/running_frame.rs +++ b/rust/crates/rqd/src/frame/running_frame.rs @@ -18,6 +18,7 @@ use std::os::fd::IntoRawFd; use std::os::fd::{FromRawFd, RawFd}; #[cfg(unix)] use std::os::unix::process::ExitStatusExt; +use std::process::Stdio; use std::time::SystemTime; use std::{ collections::HashMap, @@ -28,7 +29,6 @@ use std::{ sync::atomic::{AtomicBool, Ordering}, sync::{Arc, RwLock}, }; -use std::{process::Stdio, thread}; use tokio::time::{self, Duration}; use bytesize::KIB; @@ -189,6 +189,12 @@ pub struct CreatedState { pub struct RunningState { pub pid: u32, start_time: SystemTime, + /// OS-reported start time (seconds since epoch) of the spawned process, captured right + /// after spawn. Persisted in the snapshot so a recovering RQD can verify the pid still + /// belongs to the frame's process and not to an unrelated process that reused the pid. + /// `None` when the start time could not be captured; identity checks then fall back to + /// pid existence alone. + proc_start_epoch: Option, // Attention: Recovered frames will never have a joinHandle #[serde(skip_serializing)] #[serde(skip_deserializing)] @@ -327,6 +333,7 @@ impl RunningFrame { start_time: SystemTime::now() .checked_sub(duration) .unwrap_or(SystemTime::now()), + proc_start_epoch: None, launch_thread_handle: created_state.launch_thread_handle.take(), kill_reason: None, }); @@ -383,6 +390,7 @@ impl RunningFrame { FrameState::Running(ref r) => FrameState::Running(RunningState { pid: r.pid, start_time: r.start_time, + proc_start_epoch: r.proc_start_epoch, launch_thread_handle: None, kill_reason: r.kill_reason.clone(), }), @@ -505,6 +513,10 @@ impl RunningFrame { /// Returning an error is pointless as we want the frame that trigger this transition to finish /// regardless pub(super) fn start(&self, pid: u32) { + // Capture the OS-reported start time of the process before taking the state lock. + // Best effort: a process that exits immediately may already be gone, in which case + // recovery-time identity checks fall back to pid existence alone. + let proc_start_epoch = Self::process_start_epoch(pid); let mut state = self.state.write().unwrap_or_else(|err| err.into_inner()); match &mut *state { @@ -512,6 +524,7 @@ impl RunningFrame { *state = FrameState::Running(RunningState { pid, start_time: SystemTime::now(), + proc_start_epoch, launch_thread_handle: created_state.launch_thread_handle.take(), kill_reason: None, }); @@ -1006,7 +1019,7 @@ impl RunningFrame { let (log_pipe_handle, logger_signal) = self.spawn_logger(logger).await; info!("Frame {self} recovered with pid {pid}"); - self.wait()?; + self.wait().await?; // Send a signal to the logger thread if logger_signal.send(()).await.is_err() { @@ -1048,6 +1061,16 @@ impl RunningFrame { } } + /// Returns the OS-reported start time recorded when the frame process was spawned. + /// Only available while the frame is in the Running state. + fn recorded_proc_start_epoch(&self) -> Option { + let state = self.state.read().unwrap_or_else(|err| err.into_inner()); + match *state { + FrameState::Running(ref running_state) => running_state.proc_start_epoch, + _ => None, + } + } + /// Reads the exit status from the exit file written by the frame process /// /// # Returns @@ -1114,72 +1137,61 @@ impl RunningFrame { } } - /// Waits for a process to exit by checking its status periodically + /// Waits for the frame process to exit by checking its status periodically. /// /// # Returns - /// Returns `Ok(())` if the process successfully exits or is already gone. + /// Returns `Ok(())` if the process exits or is already gone. /// /// # Errors - /// Returns an error if: - /// - There's no valid PID for the frame - /// - There's an error when checking the process status (not including ESRCH) + /// Returns an error if there's no valid PID for the frame. /// /// # Details - /// This function polls the process status every 500ms using the kill(2) syscall - /// with a null signal. When the process exits, the syscall will return ESRCH - /// (No such process) error, indicating the process has terminated. - #[cfg(any(target_os = "linux", target_os = "macos"))] - pub fn wait(&self) -> Result<()> { - use nix::sys::signal; - use nix::unistd::Pid; - + /// Used on the recovery path, where the process is not a child of this RQD instance and + /// therefore cannot be `wait(2)`ed on directly. The poll is async (tokio sleep) so a host + /// recovering many frames does not pin one runtime worker thread per frame for the frames' + /// entire lifetime. + /// + /// Each poll also verifies process identity: if the recorded process start time no longer + /// matches the process currently owning the pid, the pid was recycled by the OS for an + /// unrelated process and the frame's process is considered exited. Without this check a + /// recovered frame could latch onto a stranger process and "run" for hours. + pub async fn wait(&self) -> Result<()> { let pid = self.pid().ok_or(miette!( "Failed to wait for frame. Process have never started: {}", self ))?; + let recorded_start_epoch = self.recorded_proc_start_epoch(); - // Convert to nix Pid - let nix_pid = Pid::from_raw(pid as i32); - - // Poll process status periodically - loop { - // Check if process is still running - match signal::kill(nix_pid, None) { - Ok(_) => { - // Process still running, wait a bit and check again - thread::sleep(Duration::from_millis(1500)); - } - Err(nix::Error::ESRCH) => { - // Process has exited - break; - } - Err(e) => { - return Err(miette!("Error checking process status: {}", e)); - } - } + while Self::process_matches(pid, recorded_start_epoch) { + time::sleep(Duration::from_millis(1500)).await; } Ok(()) } - #[cfg(target_os = "windows")] - pub fn wait(&self) -> Result<()> { - let pid = self.pid().ok_or(miette!( - "Failed to wait for frame. Process have never started: {}", - self - ))?; + /// Returns the OS-reported start time (seconds since epoch) of `pid`, or `None` when no + /// such process is currently running. + fn process_start_epoch(pid: u32) -> Option { + let mut system = System::new(); + system.refresh_processes( + sysinfo::ProcessesToUpdate::Some(&[Pid::from_u32(pid)]), + true, + ); + system.process(Pid::from_u32(pid)).map(|p| p.start_time()) + } - let mut sysinfo = System::new(); - loop { - sysinfo.refresh_processes( - sysinfo::ProcessesToUpdate::Some(&[Pid::from_u32(pid)]), - true, - ); - if sysinfo.process(Pid::from_u32(pid)).is_none() { - break; - } - thread::sleep(Duration::from_millis(1500)); + /// Checks that `pid` is alive and still refers to the process the frame spawned. + /// + /// When `recorded_start_epoch` is known, the process currently owning the pid must have + /// started within 2 seconds of it (tolerance for clock/rounding differences between + /// capture points). When unknown, falls back to pid existence alone. + fn process_matches(pid: u32, recorded_start_epoch: Option) -> bool { + match Self::process_start_epoch(pid) { + None => false, + Some(actual_start) => match recorded_start_epoch { + None => true, + Some(recorded_start) => actual_start.abs_diff(recorded_start) <= 2, + }, } - Ok(()) } /// Retrieves the process ID (PID) that should be killed when terminating this frame @@ -1405,20 +1417,19 @@ impl RunningFrame { /// Returns an error if: /// - The snapshot file cannot be opened or read /// - The snapshot data cannot be deserialized - /// - The frame's process is no longer running /// - The snapshot doesn't contain a valid PID /// /// # Details - /// This function loads a previously saved frame state from a snapshot file, - /// updates it with the provided configuration, and verifies that the process - /// is still running before returning the frame. This is primarily used for - /// recovering frames after RQD restarts. - /// - /// # Known issues: - /// This function relies on pid uniqueness, which is not ensured at the OS level. - /// TODO: Consider discarding old snapshots, or add additional checks to ensures - /// the snapshot is binding to the correct process - /// + /// This function loads a previously saved frame state from a snapshot file and updates it + /// with the provided configuration. The frame is returned for recovery in all of these + /// scenarios, so a frame's outcome is never silently lost across an RQD restart: + /// - The frame process is still running (verified by pid **and** recorded process start + /// time, so a pid recycled by the OS is not mistaken for the frame). + /// - The process finished while RQD was down but left its exit status in the exit file; + /// recovery reports the real exit status to Cuebot. + /// - The process is gone without a trace; recovery reports a failure (exit 1/SIGTERM) + /// so Cuebot can immediately reschedule the frame instead of waiting for a stuck-frame + /// timeout. pub async fn from_snapshot(path: &str, config: RunnerConfig) -> Result { let buff = tokio::fs::read(path).await.into_diagnostic()?; @@ -1433,27 +1444,26 @@ impl RunningFrame { // Initialize host mem snapshot (skipped during deserialization) frame.latest_host_mem_snapshot = RwLock::new(None); - let pid = frame.pid(); - - // Check if pid is still active - match pid { - Some(pid) => Self::is_process_running(pid).then_some(pid).ok_or(miette!( - "Frame pid {} not found for this snapshot. {}", - pid, - frame.to_string() - )), - None => Err(miette!("Invalid snapshot. Pid not present. {}", frame)), + let pid = frame + .pid() + .ok_or(miette!("Invalid snapshot. Pid not present. {}", frame))?; + + if Self::process_matches(pid, frame.recorded_proc_start_epoch()) { + info!("Snapshot {}: process {} is still running", frame, pid); + } else if Path::new(&frame.exit_file_path).exists() { + info!( + "Snapshot {}: process {} finished while RQD was down, recovering its exit \ + status from {}", + frame, pid, frame.exit_file_path + ); + } else { + warn!( + "Snapshot {}: process {} is gone and left no exit status. Frame will be \ + reported as terminated so it can be rescheduled", + frame, pid + ); } - .map(|_| frame) - } - - fn is_process_running(pid: u32) -> bool { - let mut system = System::new_all(); - system.refresh_processes( - sysinfo::ProcessesToUpdate::Some(&[Pid::from_u32(pid)]), - true, - ); - system.process(Pid::from_u32(pid)).is_some() + Ok(frame) } pub(super) fn write_header(&self) -> String { @@ -2612,4 +2622,236 @@ mod tests { // assert!(status.is_ok()); // assert_eq!((0, None), status.unwrap()); // } + + // === Recovery tests === + // + // These tests simulate an RQD restart: the task driving `run_inner` is aborted (standing in + // for the old RQD process dying) while the frame process itself keeps running in its own + // session, and a fresh `RunningFrame` deserialized from the on-disk snapshot takes over. + + #[cfg(any(target_os = "linux", target_os = "macos"))] + mod recovery { + use super::super::RunningFrame; + use super::create_running_frame; + use crate::frame::logging::{FrameLoggerT, TestLogger}; + use std::collections::HashMap; + use std::sync::Arc; + use tokio::time::{sleep, timeout, Duration}; + + /// Spawns `run_inner` on a background task and waits until the frame has a pid and its + /// snapshot exists on disk. Returns the launch task handle to abort ("kill rqd") later. + async fn launch_and_snapshot(frame: &Arc) -> tokio::task::JoinHandle<()> { + let launch_frame = Arc::clone(frame); + let handle = tokio::spawn(async move { + let logger = + Arc::new(TestLogger::init()) as Arc; + let _ = launch_frame.run_inner(logger).await; + }); + + timeout(Duration::from_secs(10), async { + while frame.pid().is_none() { + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("frame should reach the Running state"); + + frame + .create_snapshot() + .await + .expect("snapshot should be written"); + handle + } + + async fn recover_from_snapshot(frame: &RunningFrame) -> RunningFrame { + let snapshot_path = frame.snapshot_path().expect("snapshot path"); + RunningFrame::from_snapshot(&snapshot_path, frame.config.clone()) + .await + .expect("snapshot should be recoverable") + } + + fn frame_pgid(frame: &RunningFrame) -> nix::unistd::Pid { + nix::unistd::Pid::from_raw(frame.pid().expect("pid") as i32) + } + + async fn cleanup(frame: &RunningFrame) { + let _ = frame.clear_snapshot().await; + let _ = tokio::fs::remove_file(&frame.exit_file_path).await; + } + + /// Waits until the wrapper's background command is visible in the frame's process + /// group. The wrapper installs its signal traps before starting the command, so once + /// the command's process exists it is safe to signal the group and expect the traps + /// to forward and record the exit status. + async fn wait_for_frame_child(pgid: nix::unistd::Pid) { + timeout(Duration::from_secs(10), async { + loop { + let listed = std::process::Command::new("pgrep") + .args(["-g", &pgid.to_string()]) + .output() + .expect("pgrep should run"); + let count = String::from_utf8_lossy(&listed.stdout) + .lines() + .filter(|l| !l.trim().is_empty()) + .count(); + // wrapper + at least the background command + if count >= 2 { + break; + } + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("frame command should start within the timeout"); + } + + /// A frame still running across the restart must be re-attached and report the exit + /// code the process eventually returns, not a synthetic failure. + #[tokio::test] + async fn test_recover_running_frame_preserves_exit_code() { + let frame = Arc::new(create_running_frame( + "sleep 2 && exit 7", + 1, + 1, + HashMap::new(), + )); + let launch_handle = launch_and_snapshot(&frame).await; + + let recovered = recover_from_snapshot(&frame).await; + // Identity data must survive the snapshot round-trip, otherwise the pid-reuse + // guard silently degrades to pid-existence checks. + assert_eq!( + recovered.recorded_proc_start_epoch(), + frame.recorded_proc_start_epoch(), + "proc start epoch must survive the snapshot round-trip" + ); + + // Old RQD dies. The frame process survives in its own session. + launch_handle.abort(); + + let logger = + Arc::new(TestLogger::init()) as Arc; + let status = timeout(Duration::from_secs(30), recovered.recover_inner(logger)) + .await + .expect("recovery should not hang") + .expect("recovery should succeed"); + assert_eq!((7, None), status); + + cleanup(&frame).await; + } + + /// A frame whose process finished while RQD was down must report its real exit status, + /// read from the exit file, instead of being dropped or reported as killed. + #[tokio::test] + async fn test_recover_frame_finished_while_rqd_down() { + let frame = Arc::new(create_running_frame("exit 7", 1, 1, HashMap::new())); + + // Run the frame to completion: state stays Running (only `run` finalizes it), + // the process is gone and the exit file holds the status - exactly the state a + // restarted RQD finds when the frame ended during the downtime window. + let logger = + Arc::new(TestLogger::init()) as Arc; + let status = frame.run_inner(logger).await.expect("frame should run"); + assert_eq!((7, None), status); + frame + .create_snapshot() + .await + .expect("snapshot should be written"); + + let recovered = recover_from_snapshot(&frame).await; + let logger = + Arc::new(TestLogger::init()) as Arc; + let status = timeout(Duration::from_secs(30), recovered.recover_inner(logger)) + .await + .expect("recovery should not hang") + .expect("recovery should succeed"); + assert_eq!((7, None), status); + + cleanup(&frame).await; + } + + /// A frame whose whole session was SIGKILLed (no exit file written) must be reported + /// as terminated (exit 1, SIGTERM) so Cuebot can reschedule it, instead of hanging or + /// being silently dropped. + #[tokio::test] + async fn test_recover_frame_died_without_trace() { + let frame = Arc::new(create_running_frame("sleep 30", 1, 1, HashMap::new())); + let launch_handle = launch_and_snapshot(&frame).await; + let pgid = frame_pgid(&frame); + + launch_handle.abort(); + // SIGKILL the whole session: nothing gets the chance to write the exit file. + nix::sys::signal::killpg(pgid, nix::sys::signal::Signal::SIGKILL) + .expect("session should be killable"); + + let recovered = recover_from_snapshot(&frame).await; + let logger = + Arc::new(TestLogger::init()) as Arc; + // No exit file exists, so recovery falls back to the "assume terminated" status + // (exit 1, code 143 = 128+SIGTERM) instead of hanging or dropping the frame. + let status = timeout(Duration::from_secs(30), recovered.recover_inner(logger)) + .await + .expect("recovery should not hang") + .expect("recovery should succeed"); + assert_eq!((1, Some(143)), status); + + cleanup(&frame).await; + } + + /// Killing a recovered frame (as `kill_session` does: SIGTERM to the process group) + /// must surface as exit 1 / signal 15 through the exit file. + #[tokio::test] + async fn test_recover_then_kill_reports_sigterm() { + let frame = Arc::new(create_running_frame("sleep 30", 1, 1, HashMap::new())); + let launch_handle = launch_and_snapshot(&frame).await; + let pgid = frame_pgid(&frame); + + let recovered = recover_from_snapshot(&frame).await; + launch_handle.abort(); + + // Only signal the session once the wrapper's traps are provably in place. + wait_for_frame_child(pgid).await; + nix::sys::signal::killpg(pgid, nix::sys::signal::Signal::SIGTERM) + .expect("session should be killable"); + + let logger = + Arc::new(TestLogger::init()) as Arc; + let status = timeout(Duration::from_secs(30), recovered.recover_inner(logger)) + .await + .expect("recovery should not hang") + .expect("recovery should succeed"); + assert_eq!((1, Some(15)), status); + + cleanup(&frame).await; + } + + /// The identity check must accept the frame's own process and reject a pid whose + /// current owner started at a different time (pid recycled by the OS). + #[tokio::test] + async fn test_process_identity_guard() { + let my_pid = std::process::id(); + let my_start = + RunningFrame::process_start_epoch(my_pid).expect("own process should be visible"); + + assert!(RunningFrame::process_matches(my_pid, Some(my_start))); + // Unknown recorded start time falls back to pid existence. + assert!(RunningFrame::process_matches(my_pid, None)); + // Same pid, but owned by a process started at a very different time: recycled. + assert!(!RunningFrame::process_matches( + my_pid, + Some(my_start.saturating_sub(1000)) + )); + + // A reaped process must not match at all. + let dead_pid = { + let mut child = std::process::Command::new("true") + .spawn() + .expect("spawn true"); + let pid = child.id(); + child.wait().expect("wait true"); + pid + }; + assert!(!RunningFrame::process_matches(dead_pid, None)); + } + } } diff --git a/rust/crates/rqd/src/main.rs b/rust/crates/rqd/src/main.rs index d8528a983e..e28455c002 100644 --- a/rust/crates/rqd/src/main.rs +++ b/rust/crates/rqd/src/main.rs @@ -17,7 +17,7 @@ use tokio::{select, sync::oneshot}; use tracing::{error, warn}; use tracing_rolling_file::{RollingConditionBase, RollingFileAppenderBase}; -#[cfg(target_os = "macos")] +#[cfg(unix)] use crate::frame::manager; use crate::{ config::CONFIG, @@ -89,9 +89,15 @@ async fn async_main() -> miette::Result<()> { // Await for the confirmation machine_monitor has fully initialized let _machine_monitor_started = rx.await; - // Recovering frames is unstable on linux. Launched frames are somehow still bound - // to the rqd process and receive a kill signal when rqd stops - #[cfg(target_os = "macos")] + // Recover frames that survived an RQD restart. Frames run in their own session + // (setsid at spawn) so they are not killed alongside RQD; their exit status is + // recovered from the exit file written by the frame's entrypoint wrapper. + // + // Note for Linux deployments under systemd: the unit must set `KillMode=process` + // (see resources/openrqd.service). With the default `control-group` kill mode, + // systemd kills every process in the unit's cgroup on stop/restart — including + // the frames — regardless of their session or process group. + #[cfg(unix)] if let Err(err) = manager::instance().await?.recover_snapshots().await { warn!("Failed to recover frames from snapshot: {}", err); }; diff --git a/rust/crates/rqd/tests/rqd_integration_tests.rs b/rust/crates/rqd/tests/rqd_integration_tests.rs index 7f304f22c3..423bf0fe24 100644 --- a/rust/crates/rqd/tests/rqd_integration_tests.rs +++ b/rust/crates/rqd/tests/rqd_integration_tests.rs @@ -136,11 +136,19 @@ fn memory_fork_script_path() -> &'static str { } fn timeout_secs() -> u64 { - if cfg!(windows) { 20 } else { 15 } + if cfg!(windows) { + 20 + } else { + 15 + } } fn wait_millis() -> u64 { - if cfg!(windows) { 10000 } else { 8000 } + if cfg!(windows) { + 10000 + } else { + 8000 + } } /// Helper function to monitor server output for frame completion @@ -303,10 +311,11 @@ fn integration_test_lock() -> std::sync::MutexGuard<'static, ()> { // --- Test environment setup --- struct TestEnv { - _temp_dir: TempDir, + temp_dir: TempDir, dummy_server: std::process::Child, openrqd: std::process::Child, rqd_port: u16, + config_path: std::path::PathBuf, } fn make_test_config( @@ -370,7 +379,13 @@ fn setup_test_env(monitor_interval: &str, worker_threads: u32) -> TestEnv { let config_path = temp_dir.path().join("test_config.yaml"); let (rqd_port, cuebot_port) = get_two_free_ports(); - let test_config = make_test_config(&temp_dir, rqd_port, cuebot_port, monitor_interval, worker_threads); + let test_config = make_test_config( + &temp_dir, + rqd_port, + cuebot_port, + monitor_interval, + worker_threads, + ); std::fs::write(&config_path, test_config).unwrap(); let mut dummy_server = Command::new(get_binary_path("dummy-cuebot")) @@ -382,30 +397,53 @@ fn setup_test_env(monitor_interval: &str, worker_threads: u32) -> TestEnv { wait_for_port_open(&mut dummy_server, cuebot_port, "dummy-cuebot", 10); - let mut openrqd = Command::new(get_binary_path("openrqd")) - .env("OPENCUE_RQD_CONFIG", config_path.to_str().unwrap()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .expect("Failed to start openrqd"); + let mut openrqd = start_openrqd(&config_path); wait_for_port_open(&mut openrqd, rqd_port, "openrqd", 20); TestEnv { - _temp_dir: temp_dir, + temp_dir, dummy_server, openrqd, rqd_port, + config_path, } } +fn start_openrqd(config_path: &Path) -> std::process::Child { + Command::new(get_binary_path("openrqd")) + .env("OPENCUE_RQD_CONFIG", config_path.to_str().unwrap()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("Failed to start openrqd") +} + +/// Returns true when a frame entrypoint wrapper spawned from `temp_dir` is still running. +/// The wrapper's command line contains the entrypoint script path, which lives under the +/// test's private temp dir, so matching on that path only sees this test's frame. +#[cfg(unix)] +fn frame_wrapper_running(temp_dir: &Path) -> bool { + let output = Command::new("pgrep") + .args(["-f", temp_dir.to_str().unwrap()]) + .output() + .expect("pgrep should run"); + !String::from_utf8_lossy(&output.stdout).trim().is_empty() +} + +#[cfg(unix)] +fn snapshot_count(temp_dir: &Path) -> usize { + std::fs::read_dir(temp_dir.join("snapshots")) + .map(|dir| { + dir.filter_map(|entry| entry.ok()) + .filter(|entry| entry.file_name().to_string_lossy().ends_with(".bin")) + .count() + }) + .unwrap_or(0) +} + fn launch_frame(rqd_port: u16, command: &str, extra_args: &[&str]) -> std::process::Output { - let mut args = vec![ - "rqd-client", - "--hostname", - "127.0.0.1", - "--port", - ]; + let mut args = vec!["rqd-client", "--hostname", "127.0.0.1", "--port"]; let port_str = rqd_port.to_string(); args.push(&port_str); args.push("launch-frame"); @@ -425,7 +463,12 @@ fn launch_frame(rqd_port: u16, command: &str, extra_args: &[&str]) -> std::proce async fn test_openrqd_frame_execution_with_completion() { let _guard = integration_test_lock(); let env = setup_test_env("2s", 2); - let TestEnv { dummy_server, mut openrqd, rqd_port, .. } = env; + let TestEnv { + dummy_server, + mut openrqd, + rqd_port, + .. + } = env; let frame_output = launch_frame(rqd_port, &sleep_and_echo_cmd("Test frame execution"), &[]); @@ -435,7 +478,8 @@ async fn test_openrqd_frame_execution_with_completion() { String::from_utf8_lossy(&frame_output.stderr) ); - let server_handle = thread::spawn(move || monitor_server_output(dummy_server, 1, timeout_secs())); + let server_handle = + thread::spawn(move || monitor_server_output(dummy_server, 1, timeout_secs())); sleep(Duration::from_millis(wait_millis())).await; let _ = openrqd.kill(); @@ -456,7 +500,12 @@ async fn test_openrqd_frame_execution_with_completion() { async fn test_frame_with_environment_variables_and_completion() { let _guard = integration_test_lock(); let env = setup_test_env("2s", 2); - let TestEnv { dummy_server, mut openrqd, rqd_port, .. } = env; + let TestEnv { + dummy_server, + mut openrqd, + rqd_port, + .. + } = env; let frame_output = launch_frame( rqd_port, @@ -470,7 +519,8 @@ async fn test_frame_with_environment_variables_and_completion() { String::from_utf8_lossy(&frame_output.stderr) ); - let server_handle = thread::spawn(move || monitor_server_output(dummy_server, 1, timeout_secs())); + let server_handle = + thread::spawn(move || monitor_server_output(dummy_server, 1, timeout_secs())); sleep(Duration::from_millis(wait_millis())).await; let _ = openrqd.kill(); @@ -492,7 +542,12 @@ async fn test_frame_with_environment_variables_and_completion() { async fn test_frame_run_as_user() { let _guard = integration_test_lock(); let env = setup_test_env("5s", 2); - let TestEnv { mut dummy_server, mut openrqd, rqd_port, .. } = env; + let TestEnv { + mut dummy_server, + mut openrqd, + rqd_port, + .. + } = env; let frame_output = launch_frame(rqd_port, "whoami", &["--run-as-user"]); @@ -515,7 +570,12 @@ async fn test_frame_run_as_user() { async fn test_memory_fork_script() { let _guard = integration_test_lock(); let env = setup_test_env("5s", 2); - let TestEnv { mut dummy_server, mut openrqd, rqd_port, .. } = env; + let TestEnv { + mut dummy_server, + mut openrqd, + rqd_port, + .. + } = env; let script_path = memory_fork_script_path(); let frame_output = launch_frame(rqd_port, "ed_if_needed(script_path), &[]); @@ -572,10 +632,19 @@ async fn test_connection_error_handling() { async fn test_multiple_frames_sequential_with_completion() { let _guard = integration_test_lock(); let env = setup_test_env("2s", 4); - let TestEnv { dummy_server, mut openrqd, rqd_port, .. } = env; + let TestEnv { + dummy_server, + mut openrqd, + rqd_port, + .. + } = env; const NUM_FRAMES: usize = 3; - let frame_delay = if cfg!(windows) { Duration::from_secs(10) } else { Duration::from_millis(500) }; + let frame_delay = if cfg!(windows) { + Duration::from_secs(10) + } else { + Duration::from_millis(500) + }; for i in 1..=NUM_FRAMES { let frame_output = launch_frame(rqd_port, &echo_cmd(&format!("Frame {}", i)), &[]); @@ -591,7 +660,8 @@ async fn test_multiple_frames_sequential_with_completion() { } let monitor_timeout = if cfg!(windows) { 25 } else { 20 }; - let server_handle = thread::spawn(move || monitor_server_output(dummy_server, NUM_FRAMES, monitor_timeout)); + let server_handle = + thread::spawn(move || monitor_server_output(dummy_server, NUM_FRAMES, monitor_timeout)); let post_launch_wait = if cfg!(windows) { 12000 } else { 10000 }; sleep(Duration::from_millis(post_launch_wait)).await; @@ -608,3 +678,121 @@ async fn test_multiple_frames_sequential_with_completion() { println!("All {} frames completed successfully!", NUM_FRAMES); } + +/// Live-restart test: SIGKILL rqd while a frame is running, verify the frame process +/// survives, then restart rqd with the same config and verify it recovers the frame and +/// reports its successful completion to Cuebot. +#[cfg(unix)] +#[tokio::test] +async fn test_rqd_restart_recovers_running_frame() { + let _guard = integration_test_lock(); + let mut env = setup_test_env("2s", 2); + + let frame_output = launch_frame(env.rqd_port, "sleep 6 && echo 'survived rqd restart'", &[]); + assert!( + frame_output.status.success(), + "Frame launch failed: {}", + String::from_utf8_lossy(&frame_output.stderr) + ); + + // Let the frame start and its snapshot land on disk. + sleep(Duration::from_millis(2000)).await; + assert!( + snapshot_count(env.temp_dir.path()) > 0, + "expected a frame snapshot on disk before killing rqd" + ); + assert!( + frame_wrapper_running(env.temp_dir.path()), + "expected the frame process to be running before killing rqd" + ); + + // Hard-kill rqd: no graceful shutdown, exactly like a crash or a live upgrade. + env.openrqd.kill().expect("openrqd should be killable"); + let _ = env.openrqd.wait(); + + // The frame must not die with rqd. + assert!( + frame_wrapper_running(env.temp_dir.path()), + "frame process should survive rqd being SIGKILLed" + ); + + // Restart rqd with the same config; it must recover the frame from its snapshot. + let mut openrqd2 = start_openrqd(&env.config_path); + wait_for_port_open(&mut openrqd2, env.rqd_port, "openrqd (restarted)", 20); + + let dummy_server = env.dummy_server; + let server_handle = thread::spawn(move || monitor_server_output(dummy_server, 1, 40)); + let (success, output) = server_handle.join().unwrap(); + + let _ = openrqd2.kill(); + let _ = openrqd2.wait(); + + if !success { + println!("Server output:\n{}", output); + panic!("Recovered frame completion was not reported after rqd restart"); + } + assert!( + output.contains("exit_status: 0"), + "recovered frame should complete successfully, server output:\n{}", + output + ); + + println!("Frame survived rqd restart and completed successfully!"); +} + +/// Downtime-completion test: the frame finishes while rqd is down. On restart, rqd must +/// recover the frame's real exit status from the exit file and report it to Cuebot instead +/// of losing the frame or reporting a synthetic failure. +#[cfg(unix)] +#[tokio::test] +async fn test_rqd_restart_reports_frame_finished_while_down() { + let _guard = integration_test_lock(); + let mut env = setup_test_env("2s", 2); + + // Distinctive non-zero exit code proves the status came from the exit file and not + // from a default success/failure path. + let frame_output = launch_frame(env.rqd_port, "sleep 3 && exit 3", &[]); + assert!( + frame_output.status.success(), + "Frame launch failed: {}", + String::from_utf8_lossy(&frame_output.stderr) + ); + + // Wait for the snapshot, then hard-kill rqd while the frame is still running. + sleep(Duration::from_millis(1500)).await; + assert!( + snapshot_count(env.temp_dir.path()) > 0, + "expected a frame snapshot on disk before killing rqd" + ); + env.openrqd.kill().expect("openrqd should be killable"); + let _ = env.openrqd.wait(); + + // Let the frame finish during the downtime window. + sleep(Duration::from_millis(4000)).await; + assert!( + !frame_wrapper_running(env.temp_dir.path()), + "frame should have finished while rqd was down" + ); + + let mut openrqd2 = start_openrqd(&env.config_path); + wait_for_port_open(&mut openrqd2, env.rqd_port, "openrqd (restarted)", 20); + + let dummy_server = env.dummy_server; + let server_handle = thread::spawn(move || monitor_server_output(dummy_server, 1, 40)); + let (success, output) = server_handle.join().unwrap(); + + let _ = openrqd2.kill(); + let _ = openrqd2.wait(); + + if !success { + println!("Server output:\n{}", output); + panic!("Completion of a frame that finished during rqd downtime was not reported"); + } + assert!( + output.contains("exit_status: 3"), + "the frame's real exit code (3) should be recovered from the exit file, server output:\n{}", + output + ); + + println!("Frame that finished during rqd downtime was reported correctly!"); +}