diff --git a/bin/src/commands/job.rs b/bin/src/commands/job.rs new file mode 100644 index 0000000..b51daed --- /dev/null +++ b/bin/src/commands/job.rs @@ -0,0 +1,212 @@ +//! Job management commands (pause, resume, kill, clean). + +use anyhow::{Context, Result}; +use paracas_daemon::{DaemonSpawner, DownloadJob, JobStatus, StateManager}; + +/// Pause a running job by sending SIGSTOP to its process. +pub(crate) fn pause_job(state: &StateManager, job_id: &str) -> Result<()> { + let id = job_id.parse().context("Invalid job ID format")?; + + let mut job: DownloadJob = state.load_job(id).context("Job not found")?; + + if job.status != JobStatus::Running { + anyhow::bail!("Job is not running (status: {})", job.status); + } + + let Some(pid) = job.pid else { + anyhow::bail!("Job has no associated process"); + }; + + // Send SIGSTOP to pause the process + #[cfg(unix)] + { + use std::process::Command; + let status = Command::new("kill") + .args(["-STOP", &pid.to_string()]) + .status() + .context("Failed to send SIGSTOP")?; + + if !status.success() { + anyhow::bail!("Failed to pause process {}", pid); + } + } + + #[cfg(windows)] + { + // Windows doesn't have SIGSTOP equivalent, we'll just update the state + eprintln!("Warning: Pause is not fully supported on Windows. Job state updated but process continues."); + } + + job.mark_paused(); + state.save_job(&job)?; + + println!("Job {} paused.", id); + Ok(()) +} + +/// Resume a paused job by sending SIGCONT to its process. +pub(crate) fn resume_job(state: &StateManager, job_id: &str) -> Result<()> { + let id = job_id.parse().context("Invalid job ID format")?; + + let mut job: DownloadJob = state.load_job(id).context("Job not found")?; + + if job.status != JobStatus::Paused { + anyhow::bail!("Job is not paused (status: {})", job.status); + } + + let Some(pid) = job.pid else { + anyhow::bail!("Job has no associated process"); + }; + + // Check if the process is still alive + if !StateManager::is_process_running(pid) { + // Process is dead, need to respawn + println!("Process {} is no longer running. Respawning daemon...", pid); + return respawn_job(state, &mut job); + } + + // Send SIGCONT to resume the process + #[cfg(unix)] + { + use std::process::Command; + let status = Command::new("kill") + .args(["-CONT", &pid.to_string()]) + .status() + .context("Failed to send SIGCONT")?; + + if !status.success() { + anyhow::bail!("Failed to resume process {}", pid); + } + } + + #[cfg(windows)] + { + eprintln!("Warning: Resume is not fully supported on Windows."); + } + + job.mark_resumed(pid); + state.save_job(&job)?; + + println!("Job {} resumed.", id); + Ok(()) +} + +/// Respawn a job that needs to be resumed but whose process is dead. +fn respawn_job(state: &StateManager, job: &mut DownloadJob) -> Result<()> { + let spawner = DaemonSpawner::new(state.clone()).context("Failed to create daemon spawner")?; + + // Reset job status to pending so it can be picked up + job.status = JobStatus::Pending; + job.pid = None; + + spawner.spawn(job).context("Failed to respawn daemon")?; + + println!("Job {} respawned with PID {:?}.", job.id, job.pid); + Ok(()) +} + +/// Kill a running or paused job by sending SIGKILL to its process. +pub(crate) fn kill_job(state: &StateManager, job_id: &str) -> Result<()> { + let id = job_id.parse().context("Invalid job ID format")?; + + let mut job: DownloadJob = state.load_job(id).context("Job not found")?; + + if !matches!( + job.status, + JobStatus::Running | JobStatus::Pending | JobStatus::Paused + ) { + anyhow::bail!("Job is not active (status: {})", job.status); + } + + // Send SIGKILL to the process if it exists + if let Some(pid) = job.pid { + #[cfg(unix)] + { + use std::process::Command; + // First try SIGTERM for graceful shutdown + let _ = Command::new("kill") + .args(["-TERM", &pid.to_string()]) + .status(); + + // Wait briefly then force kill if still running + std::thread::sleep(std::time::Duration::from_millis(500)); + + if StateManager::is_process_running(pid) { + let _ = Command::new("kill") + .args(["-KILL", &pid.to_string()]) + .status(); + } + } + + #[cfg(windows)] + { + use std::process::Command; + let _ = Command::new("taskkill") + .args(["/F", "/PID", &pid.to_string()]) + .status(); + } + } + + job.mark_cancelled(); + state.save_job(&job)?; + + println!("Job {} killed.", id); + Ok(()) +} + +/// Clean up completed, failed, or cancelled jobs from storage. +pub(crate) fn clean_jobs(state: &StateManager, all: bool) -> Result<()> { + let jobs = state.list_jobs()?; + + let mut cleaned_count = 0; + + for job in jobs { + let should_clean = if all { + job.is_finished() + } else { + // By default, only clean jobs older than 24 hours that are finished + let is_old = job.created_at < chrono::Utc::now() - chrono::Duration::hours(24); + is_old && job.is_finished() + }; + + if should_clean { + state.delete_job(job.id)?; + cleaned_count += 1; + } + } + + if cleaned_count == 0 { + println!("No jobs to clean."); + } else { + println!("Cleaned {} job(s).", cleaned_count); + } + + Ok(()) +} + +/// Execute the job management command. +pub(crate) fn job_command( + action: &str, + job_id: Option<&str>, + all: bool, +) -> Result<()> { + let state_manager = + StateManager::with_default_path().context("Failed to initialize state manager")?; + + match action { + "pause" => { + let id = job_id.context("Job ID required for pause")?; + pause_job(&state_manager, id) + } + "resume" => { + let id = job_id.context("Job ID required for resume")?; + resume_job(&state_manager, id) + } + "kill" => { + let id = job_id.context("Job ID required for kill")?; + kill_job(&state_manager, id) + } + "clean" => clean_jobs(&state_manager, all), + _ => anyhow::bail!("Unknown action: {}", action), + } +} diff --git a/bin/src/commands/mod.rs b/bin/src/commands/mod.rs index d9a552f..f41cc56 100644 --- a/bin/src/commands/mod.rs +++ b/bin/src/commands/mod.rs @@ -4,5 +4,6 @@ pub(crate) mod daemon_run; pub(crate) mod download; pub(crate) mod download_all; pub(crate) mod info; +pub(crate) mod job; pub(crate) mod list; pub(crate) mod status; diff --git a/bin/src/main.rs b/bin/src/main.rs index 56045a2..517e55b 100644 --- a/bin/src/main.rs +++ b/bin/src/main.rs @@ -1,7 +1,7 @@ //! paracas CLI - High-performance Dukascopy tick data downloader. -use anyhow::{Context, Result}; -use clap::{Parser, Subcommand}; +use anyhow::Result; +use clap::{CommandFactory, Parser, Subcommand}; use std::path::PathBuf; mod commands; @@ -151,6 +151,41 @@ enum Commands { #[arg(long)] yes: bool, }, + + /// Manage background jobs (pause, resume, kill, clean) + Job { + #[command(subcommand)] + action: JobAction, + }, +} + +/// Actions for managing background jobs. +#[derive(Subcommand)] +enum JobAction { + /// Pause a running job + Pause { + /// Job ID to pause + job_id: String, + }, + + /// Resume a paused job + Resume { + /// Job ID to resume + job_id: String, + }, + + /// Kill a running or paused job + Kill { + /// Job ID to kill + job_id: String, + }, + + /// Clean up finished jobs from storage + Clean { + /// Clean all finished jobs (not just old ones) + #[arg(long)] + all: bool, + }, } #[tokio::main] @@ -162,10 +197,11 @@ async fn main() -> Result<()> { return commands::daemon_run::daemon_run(&job_id).await; } - // Require a command otherwise - let command = cli - .command - .context("No command provided. Use --help for usage.")?; + // Show help if no command provided + let Some(command) = cli.command else { + Cli::command().print_help()?; + return Ok(()); + }; match command { Commands::Download { @@ -231,5 +267,17 @@ async fn main() -> Result<()> { ) .await } + Commands::Job { action } => match action { + JobAction::Pause { job_id } => { + commands::job::job_command("pause", Some(&job_id), false) + } + JobAction::Resume { job_id } => { + commands::job::job_command("resume", Some(&job_id), false) + } + JobAction::Kill { job_id } => { + commands::job::job_command("kill", Some(&job_id), false) + } + JobAction::Clean { all } => commands::job::job_command("clean", None, all), + }, } } diff --git a/crates/paracas-daemon/src/job.rs b/crates/paracas-daemon/src/job.rs index 4d30e12..e44ea14 100644 --- a/crates/paracas-daemon/src/job.rs +++ b/crates/paracas-daemon/src/job.rs @@ -17,6 +17,8 @@ pub enum JobStatus { Pending, /// Job is currently running. Running, + /// Job is paused by the user. + Paused, /// Job completed successfully. Completed, /// Job failed with an error. @@ -38,6 +40,7 @@ impl JobStatus { match self { Self::Pending => "pending", Self::Running => "running", + Self::Paused => "paused", Self::Completed => "completed", Self::Failed => "failed", Self::Cancelled => "cancelled", @@ -217,6 +220,31 @@ impl DownloadJob { } } } + + /// Marks the job as paused. + pub fn mark_paused(&mut self) { + self.status = JobStatus::Paused; + + // Pause any running tasks + for task in &mut self.tasks { + if task.status == JobStatus::Running { + task.status = JobStatus::Paused; + } + } + } + + /// Marks the job as resumed (back to running). + pub fn mark_resumed(&mut self, pid: u32) { + self.status = JobStatus::Running; + self.pid = Some(pid); + + // Resume any paused tasks + for task in &mut self.tasks { + if task.status == JobStatus::Paused { + task.status = JobStatus::Running; + } + } + } } #[cfg(test)]