From 68acceb9a3e4914bbc5ee915341660bcb21940f0 Mon Sep 17 00:00:00 2001 From: Diego Tavares Date: Mon, 10 Aug 2026 15:01:23 -0700 Subject: [PATCH] [rqd] Fix rqd recovery mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recover mode was never working properly on Linux as when running as under systemctl service, spawned frames were being killed with the rqd process for both belong to the same cgroup. This change not only changes the .service default `KillMode=control-group` to `KillMode=process` but also fixes a bug on the frames hardness: - frame_cmd.rs — enabled the exit-file harness on Linux and rewrote the wrapper script, which had real bugs: command_pid=$! was captured after a foreground eval (so it was always empty), the trap wrote the raw signal number as the exit code, and bash defers traps while a foreground child runs (kills couldn't be forwarded). The new script runs the command in the background + wait, forwards signals, re-waits until the command truly exits, and writes the exit file atomically (tmp + mv). --- rust/crates/rqd/resources/openrqd.service | 6 + rust/crates/rqd/src/frame/frame_cmd.rs | 238 ++++++++-- rust/crates/rqd/src/frame/running_frame.rs | 406 ++++++++++++++---- rust/crates/rqd/src/main.rs | 14 +- .../crates/rqd/tests/rqd_integration_tests.rs | 240 +++++++++-- 5 files changed, 756 insertions(+), 148 deletions(-) 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 b7b3a1fa98..ccbdc9b442 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 { @@ -2579,4 +2589,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 8a821b6257..b7aa1b6864 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, @@ -84,9 +84,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!"); +}