use std::ffi::OsString;
use std::fs;
use std::io::Write;
use std::io::{BufRead, BufReader};
use std::path::PathBuf;
use std::process::ExitCode;
use cargo_reclaim::{
BackgroundRunEventKind, BackgroundRunLogRecord, BackgroundServiceOptions,
BackgroundServicePaths, BackgroundServiceState, BackgroundServiceStatus,
DEFAULT_SCHEDULER_INSTANCE_NAME, ReclaimConfig, SchedulerPlatform, default_instance_log_dir,
default_instance_state_dir, default_log_dir, default_state_dir, load_config_from_path,
read_background_service_state, refresh_background_service_state, run_background_service,
scheduler_instance_name_from_config,
};
use super::super::target_report::human_bytes;
use super::super::{CliError, OutputFormat, inline_config_path, next_path, next_value};
use super::scheduler_subcommand_usage;
#[derive(Debug)]
pub(in crate::cli) enum SchedulerServiceCommand {
Run(SchedulerServiceRunCommand),
Status(SchedulerServiceStatusCommand),
}
#[derive(Debug)]
pub(in crate::cli) struct SchedulerServiceRunCommand {
config_path: PathBuf,
max_cycles: Option<usize>,
output_format: OutputFormat,
}
#[derive(Debug)]
pub(in crate::cli) struct SchedulerServiceStatusCommand {
config_path: PathBuf,
output_format: OutputFormat,
}
pub(super) fn parse_scheduler_service(
args: impl IntoIterator<Item = OsString>,
) -> Result<SchedulerServiceCommand, CliError> {
let mut args = args.into_iter();
let Some(subcommand) = args.next() else {
return Err(CliError::Usage(
"scheduler service requires `run` or `status`".to_string(),
));
};
match subcommand.to_string_lossy().as_ref() {
"run" => parse_service_run(args).map(SchedulerServiceCommand::Run),
"status" => parse_service_status(args).map(SchedulerServiceCommand::Status),
"-h" | "--help" | "help" => Err(CliError::Help(scheduler_subcommand_usage("service"))),
value => Err(CliError::Usage(format!(
"unknown scheduler service command `{value}`; expected `run` or `status`"
))),
}
}
pub(super) fn run_scheduler_service(
command: &SchedulerServiceCommand,
output: &mut impl Write,
) -> Result<ExitCode, CliError> {
match command {
SchedulerServiceCommand::Run(command) => run_service(command, output),
SchedulerServiceCommand::Status(command) => run_status(command, output),
}
}
fn parse_service_run(
args: impl IntoIterator<Item = OsString>,
) -> Result<SchedulerServiceRunCommand, CliError> {
let mut config_path = None;
let mut max_cycles = None;
let mut output_format = OutputFormat::Terminal;
let mut args = args.into_iter();
while let Some(arg) = args.next() {
if let Some(path) = inline_config_path(&arg)? {
config_path = Some(path);
continue;
}
let Some(arg_text) = arg.as_os_str().to_str() else {
return Err(CliError::Usage(
"scheduler service run options must be valid UTF-8".to_string(),
));
};
match arg_text {
"--config" => config_path = Some(next_path(&mut args, "--config")?),
"--max-cycles" => {
max_cycles = Some(parse_max_cycles(&next_value(&mut args, "--max-cycles")?)?)
}
value if value.starts_with("--max-cycles=") => {
max_cycles = Some(parse_max_cycles(&value["--max-cycles=".len()..])?)
}
"--json" => output_format = OutputFormat::Json,
"-h" | "--help" => {
return Err(CliError::Help(scheduler_subcommand_usage("service run")));
}
value if value.starts_with('-') => {
return Err(CliError::Usage(format!(
"unknown scheduler service run option `{value}`"
)));
}
value => {
return Err(CliError::Usage(format!(
"unexpected scheduler service run argument `{value}`"
)));
}
}
}
Ok(SchedulerServiceRunCommand {
config_path: config_path.ok_or_else(|| {
CliError::Usage("scheduler service run requires --config".to_string())
})?,
max_cycles,
output_format,
})
}
fn parse_service_status(
args: impl IntoIterator<Item = OsString>,
) -> Result<SchedulerServiceStatusCommand, CliError> {
let mut config_path = None;
let mut output_format = OutputFormat::Terminal;
let mut args = args.into_iter();
while let Some(arg) = args.next() {
if let Some(path) = inline_config_path(&arg)? {
config_path = Some(path);
continue;
}
let Some(arg_text) = arg.as_os_str().to_str() else {
return Err(CliError::Usage(
"scheduler service status options must be valid UTF-8".to_string(),
));
};
match arg_text {
"--config" => config_path = Some(next_path(&mut args, "--config")?),
"--json" => output_format = OutputFormat::Json,
"-h" | "--help" => {
return Err(CliError::Help(scheduler_subcommand_usage("service status")));
}
value if value.starts_with('-') => {
return Err(CliError::Usage(format!(
"unknown scheduler service status option `{value}`"
)));
}
value => {
return Err(CliError::Usage(format!(
"unexpected scheduler service status argument `{value}`"
)));
}
}
}
Ok(SchedulerServiceStatusCommand {
config_path: config_path.ok_or_else(|| {
CliError::Usage("scheduler service status requires --config".to_string())
})?,
output_format,
})
}
fn run_service(
command: &SchedulerServiceRunCommand,
output: &mut impl Write,
) -> Result<ExitCode, CliError> {
let config = load_config_from_path(&command.config_path)?;
let config_path = canonical_config_path(command.config_path.clone());
let paths = service_paths(&config_path, &config)?;
let options = service_options(&config_path, paths.clone(), command.max_cycles);
let summary = run_background_service(options, &config)?;
match command.output_format {
OutputFormat::Terminal => {
writeln!(output, "cargo-reclaim scheduler service")?;
writeln!(output, "status: {}", status_label(summary.state.status))?;
writeln!(output, "state file: {}", paths.state_path.display())?;
writeln!(output, "log dir: {}", paths.log_dir.display())?;
writeln!(output, "run log: {}", paths.runs_log_path.display())?;
writeln!(output, "cycles: {}", summary.cycles_completed)?;
if let Some(run_id) = &summary.state.last_run_id {
writeln!(output, "last run: {run_id}")?;
}
}
OutputFormat::Json => write_status_json(
output,
"scheduler-service-run",
&summary.state,
Some(summary.cycles_completed),
Some(&paths),
None,
)?,
}
Ok(ExitCode::SUCCESS)
}
fn run_status(
command: &SchedulerServiceStatusCommand,
output: &mut impl Write,
) -> Result<ExitCode, CliError> {
let config = load_config_from_path(&command.config_path)?;
let config_path = canonical_config_path(command.config_path.clone());
let paths = service_paths(&config_path, &config)?;
let state = read_background_service_state(&paths.state_path)?
.map(refresh_background_service_state)
.unwrap_or_else(BackgroundServiceState::missing);
let run_summary = read_run_summary(&paths.runs_log_path)?;
match command.output_format {
OutputFormat::Terminal => {
write_status_terminal(output, &paths, &state, run_summary.as_ref())?
}
OutputFormat::Json => write_status_json(
output,
"scheduler-service-status",
&state,
None,
Some(&paths),
run_summary.as_ref(),
)?,
}
Ok(ExitCode::SUCCESS)
}
fn service_options(
config_path: &std::path::Path,
paths: BackgroundServicePaths,
max_cycles: Option<usize>,
) -> BackgroundServiceOptions {
BackgroundServiceOptions {
config_path: config_path.to_path_buf(),
state_dir: paths.state_dir,
log_dir: paths.log_dir,
mode: None,
max_cycles,
}
}
fn service_paths(
config_path: &std::path::Path,
config: &ReclaimConfig,
) -> Result<BackgroundServicePaths, CliError> {
let instance_name =
scheduler_instance_name_from_config(config.scheduler.name.as_deref(), config_path)?;
Ok(BackgroundServicePaths::new(
config
.scheduler
.state_dir
.clone()
.unwrap_or_else(|| default_service_state_dir(&instance_name)),
config
.scheduler
.log_dir
.clone()
.unwrap_or_else(|| default_service_log_dir(&instance_name)),
))
}
fn default_service_state_dir(instance_name: &str) -> PathBuf {
if instance_name == DEFAULT_SCHEDULER_INSTANCE_NAME {
default_state_dir(SchedulerPlatform::SystemdUser)
} else {
default_instance_state_dir(SchedulerPlatform::SystemdUser, instance_name)
}
}
fn default_service_log_dir(instance_name: &str) -> PathBuf {
if instance_name == DEFAULT_SCHEDULER_INSTANCE_NAME {
default_log_dir(SchedulerPlatform::SystemdUser)
} else {
default_instance_log_dir(SchedulerPlatform::SystemdUser, instance_name)
}
}
fn write_status_terminal(
output: &mut impl Write,
paths: &BackgroundServicePaths,
state: &BackgroundServiceState,
run_summary: Option<&RunSummary>,
) -> Result<(), CliError> {
writeln!(
output,
"cargo-reclaim scheduler service: {}",
status_label(state.status)
)?;
writeln!(output, "state file: {}", paths.state_path.display())?;
writeln!(output, "log dir: {}", paths.log_dir.display())?;
writeln!(output, "run log: {}", paths.runs_log_path.display())?;
if let Some(pid) = state.pid {
writeln!(output, "pid: {pid}")?;
}
if let Some(run_id) = &state.last_run_id {
writeln!(output, "last run: {run_id}")?;
}
if let Some(problem) = &state.last_problem {
writeln!(output, "problem: {problem}")?;
}
if let Some(summary) = run_summary {
writeln!(output, "run log records: {}", summary.record_count)?;
if summary.corrupt_record_count > 0 {
writeln!(
output,
"corrupt run log records: {}",
summary.corrupt_record_count
)?;
}
writeln!(output, "started cycles: {}", summary.started_count)?;
writeln!(
output,
"apply completed cycles: {}",
summary.apply_completed_count
)?;
writeln!(output, "failed cycles: {}", summary.failed_count)?;
writeln!(
output,
"applied bytes: {} ({})",
summary.applied_bytes,
human_bytes(summary.applied_bytes)
)?;
if let Some(run_id) = &summary.last_event_run_id {
writeln!(output, "last event run: {run_id}")?;
}
if let Some(kind) = summary.last_event_kind {
writeln!(output, "last event: {}", run_event_kind_label(kind))?;
}
if !summary.recent_runs.is_empty() {
writeln!(output, "recent runs:")?;
for run in &summary.recent_runs {
writeln!(
output,
" {}: last_event={} applied_bytes={} ({}) failed={}",
run.run_id,
run_event_kind_label(run.last_event_kind),
run.applied_bytes,
human_bytes(run.applied_bytes),
run.failed
)?;
}
}
}
Ok(())
}
fn write_status_json(
output: &mut impl Write,
command: &'static str,
state: &BackgroundServiceState,
cycles_completed: Option<usize>,
paths: Option<&BackgroundServicePaths>,
run_summary: Option<&RunSummary>,
) -> Result<(), CliError> {
let document = serde_json::json!({
"command": command,
"schema_version": state.schema_version,
"status": status_label(state.status),
"pid": state.pid,
"started_at": state.started_at,
"last_run_id": state.last_run_id,
"last_run_at": state.last_run_at,
"next_run_at": state.next_run_at,
"consecutive_failures": state.consecutive_failures,
"last_problem": state.last_problem,
"cycles_completed": cycles_completed,
"paths": paths.map(JsonServicePaths::from),
"run_log": run_summary.map(JsonRunSummary::from),
});
serde_json::to_writer(&mut *output, &document)?;
writeln!(output)?;
Ok(())
}
#[derive(serde::Serialize)]
struct JsonServicePaths {
state_dir: String,
state_path: String,
log_dir: String,
run_log_path: String,
}
impl From<&BackgroundServicePaths> for JsonServicePaths {
fn from(paths: &BackgroundServicePaths) -> Self {
Self {
state_dir: paths.state_dir.display().to_string(),
state_path: paths.state_path.display().to_string(),
log_dir: paths.log_dir.display().to_string(),
run_log_path: paths.runs_log_path.display().to_string(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct RunSummary {
record_count: usize,
corrupt_record_count: usize,
started_count: usize,
apply_completed_count: usize,
failed_count: usize,
applied_bytes: u64,
last_event_run_id: Option<String>,
last_event_kind: Option<BackgroundRunEventKind>,
recent_runs: Vec<RecentRunSummary>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct RecentRunSummary {
run_id: String,
last_event_kind: BackgroundRunEventKind,
applied_bytes: u64,
failed: bool,
}
#[derive(serde::Serialize)]
struct JsonRunSummary {
record_count: usize,
corrupt_record_count: usize,
started_count: usize,
apply_completed_count: usize,
failed_count: usize,
applied_bytes: u64,
last_event_run_id: Option<String>,
last_event: Option<&'static str>,
recent_runs: Vec<JsonRecentRunSummary>,
}
impl From<&RunSummary> for JsonRunSummary {
fn from(summary: &RunSummary) -> Self {
Self {
record_count: summary.record_count,
corrupt_record_count: summary.corrupt_record_count,
started_count: summary.started_count,
apply_completed_count: summary.apply_completed_count,
failed_count: summary.failed_count,
applied_bytes: summary.applied_bytes,
last_event_run_id: summary.last_event_run_id.clone(),
last_event: summary.last_event_kind.map(run_event_kind_label),
recent_runs: summary
.recent_runs
.iter()
.map(JsonRecentRunSummary::from)
.collect(),
}
}
}
#[derive(serde::Serialize)]
struct JsonRecentRunSummary {
run_id: String,
last_event: &'static str,
applied_bytes: u64,
failed: bool,
}
impl From<&RecentRunSummary> for JsonRecentRunSummary {
fn from(summary: &RecentRunSummary) -> Self {
Self {
run_id: summary.run_id.clone(),
last_event: run_event_kind_label(summary.last_event_kind),
applied_bytes: summary.applied_bytes,
failed: summary.failed,
}
}
}
fn read_run_summary(path: &std::path::Path) -> Result<Option<RunSummary>, CliError> {
if !path.exists() {
return Ok(None);
}
Ok(Some(read_run_summary_lossy(path)?))
}
fn read_run_summary_lossy(path: &std::path::Path) -> Result<RunSummary, CliError> {
let file = fs::File::open(path)?;
let reader = BufReader::new(file);
let mut records = Vec::new();
let mut corrupt_record_count = 0;
for line in reader.lines() {
let line = line?;
if line.trim().is_empty() {
continue;
}
match serde_json::from_str::<BackgroundRunLogRecord>(&line) {
Ok(record)
if record.schema_version == cargo_reclaim::BACKGROUND_RUN_LOG_SCHEMA_VERSION =>
{
records.push(record);
}
Ok(_) | Err(_) => corrupt_record_count += 1,
}
}
let mut summary = summarize_run_log(&records);
summary.corrupt_record_count = corrupt_record_count;
Ok(summary)
}
fn summarize_run_log(records: &[BackgroundRunLogRecord]) -> RunSummary {
let mut summary = RunSummary {
record_count: records.len(),
corrupt_record_count: 0,
started_count: 0,
apply_completed_count: 0,
failed_count: 0,
applied_bytes: 0,
last_event_run_id: None,
last_event_kind: None,
recent_runs: Vec::new(),
};
let mut run_summaries = Vec::<RecentRunSummary>::new();
for record in records {
let run_index = match run_summaries
.iter()
.position(|summary| summary.run_id == record.run_id)
{
Some(index) => index,
None => {
run_summaries.push(RecentRunSummary {
run_id: record.run_id.clone(),
last_event_kind: record.kind,
applied_bytes: 0,
failed: false,
});
run_summaries.len() - 1
}
};
let run_summary = &mut run_summaries[run_index];
run_summary.last_event_kind = record.kind;
match record.kind {
BackgroundRunEventKind::Started => summary.started_count += 1,
BackgroundRunEventKind::ApplyCompleted => {
summary.apply_completed_count += 1;
if let Some(apply) = &record.apply {
run_summary.applied_bytes = run_summary
.applied_bytes
.saturating_add(apply.totals.applied_bytes);
summary.applied_bytes = summary
.applied_bytes
.saturating_add(apply.totals.applied_bytes);
}
}
BackgroundRunEventKind::Failed => {
summary.failed_count += 1;
run_summary.failed = true;
}
BackgroundRunEventKind::Triggered
| BackgroundRunEventKind::PlanBuilt
| BackgroundRunEventKind::Skipped => {}
}
summary.last_event_run_id = Some(record.run_id.clone());
summary.last_event_kind = Some(record.kind);
}
let recent_start = run_summaries.len().saturating_sub(5);
summary.recent_runs = run_summaries.split_off(recent_start);
summary
}
fn run_event_kind_label(kind: BackgroundRunEventKind) -> &'static str {
match kind {
BackgroundRunEventKind::Started => "started",
BackgroundRunEventKind::Triggered => "triggered",
BackgroundRunEventKind::PlanBuilt => "plan_built",
BackgroundRunEventKind::ApplyCompleted => "apply_completed",
BackgroundRunEventKind::Skipped => "skipped",
BackgroundRunEventKind::Failed => "failed",
}
}
fn status_label(status: BackgroundServiceStatus) -> &'static str {
match status {
BackgroundServiceStatus::Running => "running",
BackgroundServiceStatus::Stopped => "stopped",
BackgroundServiceStatus::Unknown => "unknown",
BackgroundServiceStatus::Stale => "stale",
BackgroundServiceStatus::Error => "error",
}
}
fn parse_max_cycles(value: &str) -> Result<usize, CliError> {
let max_cycles = value.parse::<usize>().map_err(|_| {
CliError::Usage(format!(
"invalid --max-cycles `{value}`; expected a positive integer"
))
})?;
if max_cycles == 0 {
return Err(CliError::Usage(
"--max-cycles must be greater than zero".to_string(),
));
}
Ok(max_cycles)
}
fn canonical_config_path(path: PathBuf) -> PathBuf {
std::fs::canonicalize(&path).unwrap_or(path)
}
#[cfg(test)]
mod tests {
use std::path::Path;
use cargo_reclaim::parse_config;
use super::service_paths;
#[test]
fn service_paths_use_generic_defaults_without_name() -> Result<(), Box<dyn std::error::Error>> {
let config = parse_config("version = 1\n")?;
let paths = service_paths(Path::new("/tmp/projects/nodedb.toml"), &config)?;
assert!(paths.state_dir.ends_with("cargo-reclaim"));
assert!(
paths
.log_dir
.ends_with(Path::new("cargo-reclaim").join("logs"))
);
Ok(())
}
}