cargo_reclaim/background/

service.rs

mod model;
mod request;

use std::fs::{self, File, OpenOptions};
use std::path::{Path, PathBuf};
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use crate::active_process::platform_active_observation_provider;
use crate::config::ReclaimConfig;
use crate::persistence::PersistedTimestamp;
use fs2::FileExt;

use super::{BackgroundRunReport, BackgroundRunRequest, run_background_cleanup_cycle};
use request::{BackgroundCycleRequestContext, scheduler_mode_from_config};

pub use model::{
    BACKGROUND_SERVICE_STATE_SCHEMA_VERSION, BackgroundServiceError, BackgroundServiceOptions,
    BackgroundServicePaths, BackgroundServiceResult, BackgroundServiceRunSummary,
    BackgroundServiceState, BackgroundServiceStatus, DEFAULT_BACKGROUND_CHECK_EVERY,
};

pub trait BackgroundServiceClock {
    fn now(&mut self) -> SystemTime;
}

pub trait BackgroundServiceSleeper {
    fn sleep(&mut self, duration: Duration);
}

pub trait BackgroundServiceCycleRunner {
    fn run_cycle(
        &mut self,
        request: BackgroundRunRequest,
    ) -> BackgroundServiceResult<BackgroundRunReport>;
}

#[derive(Debug, Default)]
pub struct SystemBackgroundServiceClock;

impl BackgroundServiceClock for SystemBackgroundServiceClock {
    fn now(&mut self) -> SystemTime {
        SystemTime::now()
    }
}

#[derive(Debug, Default)]
pub struct ThreadBackgroundServiceSleeper;

impl BackgroundServiceSleeper for ThreadBackgroundServiceSleeper {
    fn sleep(&mut self, duration: Duration) {
        thread::sleep(duration);
    }
}

#[derive(Debug, Default)]
pub struct PlatformBackgroundServiceCycleRunner;

impl BackgroundServiceCycleRunner for PlatformBackgroundServiceCycleRunner {
    fn run_cycle(
        &mut self,
        request: BackgroundRunRequest,
    ) -> BackgroundServiceResult<BackgroundRunReport> {
        let provider = platform_active_observation_provider();
        run_background_cleanup_cycle(request, &provider).map_err(BackgroundServiceError::from)
    }
}

pub fn run_background_service(
    options: BackgroundServiceOptions,
    config: &ReclaimConfig,
) -> BackgroundServiceResult<BackgroundServiceRunSummary> {
    let mut clock = SystemBackgroundServiceClock;
    let mut sleeper = ThreadBackgroundServiceSleeper;
    let mut runner = PlatformBackgroundServiceCycleRunner;
    run_background_service_with_runtime(options, config, &mut clock, &mut sleeper, &mut runner)
}

pub fn run_background_service_with_runtime(
    options: BackgroundServiceOptions,
    config: &ReclaimConfig,
    clock: &mut impl BackgroundServiceClock,
    sleeper: &mut impl BackgroundServiceSleeper,
    runner: &mut impl BackgroundServiceCycleRunner,
) -> BackgroundServiceResult<BackgroundServiceRunSummary> {
    let paths = BackgroundServicePaths::new(&options.state_dir, &options.log_dir);
    let _lock = ServiceLock::acquire(&paths.lock_path)?;
    ensure_service_dirs(&paths)?;

    let started_at = persisted_timestamp(clock.now())?;
    let mut state = BackgroundServiceState::running(started_at);
    write_background_service_state(&paths.state_path, &state)?;

    let interval = config
        .background
        .check_every
        .unwrap_or(DEFAULT_BACKGROUND_CHECK_EVERY);
    let request_context = BackgroundCycleRequestContext::from_config(
        config,
        &options.config_path,
        match options.mode {
            Some(mode) => mode,
            None => scheduler_mode_from_config(config)?,
        },
    )?;
    let mut cycles_completed = 0;

    loop {
        let now = clock.now();
        let stamp = build_service_timestamp(now);
        let run_id = format!("scheduler-{stamp}");
        let plan_path = paths.plans_dir.join(format!("cargo-reclaim-{stamp}.json"));
        let request =
            request_context.request(run_id.clone(), paths.runs_log_path.clone(), plan_path, now)?;

        match runner.run_cycle(request) {
            Ok(_) => {
                state.last_problem = None;
                state.consecutive_failures = 0;
            }
            Err(error) => {
                state.consecutive_failures = state.consecutive_failures.saturating_add(1);
                state.last_problem = Some(error.to_string());
            }
        }
        cycles_completed += 1;

        state.last_run_id = Some(run_id);
        state.last_run_at = Some(persisted_timestamp(now)?);
        let next_run_at = now + interval;
        state.next_run_at = Some(persisted_timestamp(next_run_at)?);
        write_background_service_state(&paths.state_path, &state)?;

        if options
            .max_cycles
            .is_some_and(|max_cycles| cycles_completed >= max_cycles)
        {
            state.status = BackgroundServiceStatus::Stopped;
            state.pid = None;
            state.next_run_at = None;
            write_background_service_state(&paths.state_path, &state)?;
            return Ok(BackgroundServiceRunSummary {
                state,
                cycles_completed,
            });
        }

        sleeper.sleep(interval);
    }
}

pub fn read_background_service_state(
    path: impl AsRef<Path>,
) -> BackgroundServiceResult<Option<BackgroundServiceState>> {
    let path = path.as_ref();
    match fs::read(path) {
        Ok(contents) => serde_json::from_slice(&contents)
            .map(Some)
            .map_err(|source| BackgroundServiceError::Json {
                path: path.to_path_buf(),
                source,
            }),
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(source) => Err(BackgroundServiceError::Io {
            path: path.to_path_buf(),
            source,
        }),
    }
}

pub fn refresh_background_service_state(state: BackgroundServiceState) -> BackgroundServiceState {
    if state.status != BackgroundServiceStatus::Running {
        return state;
    }

    let Some(pid) = state.pid else {
        return stale_service_state(state, "service state is running without a pid");
    };

    match process_liveness(pid) {
        ProcessLiveness::Alive => state,
        ProcessLiveness::Dead => stale_service_state(state, "service pid is not running"),
        ProcessLiveness::Unknown => state,
    }
}

pub fn write_background_service_state(
    path: impl AsRef<Path>,
    state: &BackgroundServiceState,
) -> BackgroundServiceResult<()> {
    let path = path.as_ref();
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).map_err(|source| BackgroundServiceError::Io {
            path: parent.to_path_buf(),
            source,
        })?;
    }
    let contents =
        serde_json::to_vec_pretty(state).map_err(|source| BackgroundServiceError::Serialize {
            path: path.to_path_buf(),
            source,
        })?;
    fs::write(path, contents).map_err(|source| BackgroundServiceError::Io {
        path: path.to_path_buf(),
        source,
    })
}

struct ServiceLock {
    path: PathBuf,
    _file: File,
}

impl ServiceLock {
    fn acquire(path: &Path) -> BackgroundServiceResult<Self> {
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent).map_err(|source| BackgroundServiceError::Io {
                path: parent.to_path_buf(),
                source,
            })?;
        }
        if path.exists() && !path.is_file() {
            return Err(BackgroundServiceError::StaleLock {
                lock_path: path.to_path_buf(),
            });
        }

        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(path)
            .map_err(|source| BackgroundServiceError::Io {
                path: path.to_path_buf(),
                source,
            })?;
        match file.try_lock_exclusive() {
            Ok(()) => {
                write_lock_file(&file, path)?;
                Ok(Self {
                    path: path.to_path_buf(),
                    _file: file,
                })
            }
            Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => {
                Err(BackgroundServiceError::AlreadyRunning {
                    lock_path: path.to_path_buf(),
                })
            }
            Err(source) => Err(BackgroundServiceError::Io {
                path: path.to_path_buf(),
                source,
            }),
        }
    }
}

impl Drop for ServiceLock {
    fn drop(&mut self) {
        let _ = fs::remove_file(&self.path);
    }
}

fn write_lock_file(mut file: &File, path: &Path) -> BackgroundServiceResult<()> {
    let state = serde_json::json!({
        "schema_version": BACKGROUND_SERVICE_STATE_SCHEMA_VERSION,
        "pid": std::process::id(),
    });
    file.set_len(0)
        .map_err(|source| BackgroundServiceError::Io {
            path: path.to_path_buf(),
            source,
        })?;
    use std::io::{Seek, Write};
    file.rewind().map_err(|source| BackgroundServiceError::Io {
        path: path.to_path_buf(),
        source,
    })?;
    serde_json::to_writer(&mut file, &state).map_err(|source| {
        BackgroundServiceError::Serialize {
            path: path.to_path_buf(),
            source,
        }
    })?;
    file.write_all(b"\n")
        .map_err(|source| BackgroundServiceError::Io {
            path: path.to_path_buf(),
            source,
        })?;
    file.sync_all()
        .map_err(|source| BackgroundServiceError::Io {
            path: path.to_path_buf(),
            source,
        })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(dead_code)]
enum ProcessLiveness {
    Alive,
    Dead,
    Unknown,
}

fn process_liveness(pid: u32) -> ProcessLiveness {
    if pid == 0 {
        return ProcessLiveness::Dead;
    }

    #[cfg(target_os = "linux")]
    {
        if Path::new("/proc").join(pid.to_string()).exists() {
            ProcessLiveness::Alive
        } else {
            ProcessLiveness::Unknown
        }
    }

    #[cfg(not(target_os = "linux"))]
    {
        let _ = pid;
        ProcessLiveness::Unknown
    }
}

fn stale_service_state(mut state: BackgroundServiceState, problem: &str) -> BackgroundServiceState {
    state.status = BackgroundServiceStatus::Stale;
    state.pid = None;
    state.next_run_at = None;
    state.last_problem = Some(problem.to_owned());
    state
}

fn ensure_service_dirs(paths: &BackgroundServicePaths) -> BackgroundServiceResult<()> {
    for path in [&paths.state_dir, &paths.log_dir, &paths.plans_dir] {
        fs::create_dir_all(path).map_err(|source| BackgroundServiceError::Io {
            path: path.clone(),
            source,
        })?;
    }
    Ok(())
}

fn persisted_timestamp(time: SystemTime) -> BackgroundServiceResult<PersistedTimestamp> {
    PersistedTimestamp::from_system_time(time).map_err(BackgroundServiceError::Timestamp)
}

fn build_service_timestamp(now: SystemTime) -> String {
    let duration = now.duration_since(UNIX_EPOCH).unwrap_or_default();
    let total_seconds = duration.as_secs();
    let days = (total_seconds / 86_400) as i64;
    let seconds_of_day = total_seconds % 86_400;
    let (year, month, day) = civil_from_days(days);
    let hour = seconds_of_day / 3_600;
    let minute = (seconds_of_day % 3_600) / 60;
    let second = seconds_of_day % 60;
    format!("{year:04}{month:02}{day:02}T{hour:02}{minute:02}{second:02}Z")
}

fn civil_from_days(days_since_unix_epoch: i64) -> (i64, u32, u32) {
    let shifted = days_since_unix_epoch + 719_468;
    let era = if shifted >= 0 {
        shifted
    } else {
        shifted - 146_096
    } / 146_097;
    let day_of_era = shifted - era * 146_097;
    let year_of_era =
        (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365;
    let mut year = year_of_era + era * 400;
    let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100);
    let month_prime = (5 * day_of_year + 2) / 153;
    let day = day_of_year - (153 * month_prime + 2) / 5 + 1;
    let month = month_prime + if month_prime < 10 { 3 } else { -9 };
    if month <= 2 {
        year += 1;
    }
    (year, month as u32, day as u32)
}