[PATCH 3/7] client: job: add job runner and scheduled backup daemon

[Thomas Lamprecht <t.lamprecht@proxmox.com>]

Add `job run <id>` to run one job, and `job scheduler`, a per-user
daemon that starts due jobs.

The daemon plans at least once a minute rather than sleeping until the
next event, as timers stop during suspend and wall-clock deadlines move
with NTP and DST. Each run is a separate `job run` process without
stdin, which keeps the secrets of jobs apart and avoids prompts on a
shared terminal. At most two backups run at once.

A run holds the config lock until it has the job lock, like
`job remove`, and a scheduled run then checks that the job is still
due. A run whose outcome cannot be recorded fails, as the job would
otherwise stay due and run again.

The pre-run hook proceeds on exit 0, skips on 1 and fails the run
otherwise, so that a broken hook is not mistaken for a skip. It gets a
minute, no secrets, and its own process group, so a timeout also stops
what it started.

If the config does not parse, the daemon refuses to start, and a
running one starts no new runs until it is fixed.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
 proxmox-backup-client/Cargo.toml           |   3 +-
 proxmox-backup-client/src/job/mod.rs       |  22 +
 proxmox-backup-client/src/job/runner.rs    | 502 +++++++++++++++++++++
 proxmox-backup-client/src/job/scheduler.rs | 327 ++++++++++++++
 proxmox-backup-client/src/main.rs          |   4 +-
 proxmox-backup-client/tests/job.rs         | 263 +++++++++++
 6 files changed, 1119 insertions(+), 2 deletions(-)
 create mode 100644 proxmox-backup-client/src/job/mod.rs
 create mode 100644 proxmox-backup-client/src/job/runner.rs
 create mode 100644 proxmox-backup-client/src/job/scheduler.rs
 create mode 100644 proxmox-backup-client/tests/job.rs

diff --git a/proxmox-backup-client/Cargo.toml b/proxmox-backup-client/Cargo.toml
index 5f0140e78..e3a5474d9 100644
--- a/proxmox-backup-client/Cargo.toml
+++ b/proxmox-backup-client/Cargo.toml
@@ -13,7 +13,7 @@ nix.workspace = true
 openssl.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-tokio = { workspace = true, features = [ "rt", "rt-multi-thread" ] }
+tokio = { workspace = true, features = [ "io-util", "macros", "process", "rt", "rt-multi-thread", "signal", "time" ] }
 tokio-stream.workspace = true
 tokio-util = { workspace = true, features = [ "codec" ] }
 xdg.workspace = true
@@ -29,6 +29,7 @@ proxmox-log.workspace = true
 proxmox-io.workspace = true
 proxmox-router = { workspace = true, features = [ "cli" ] }
 proxmox-schema = { workspace = true, features = [ "api-macro" ] }
+proxmox-section-config.workspace = true
 proxmox-sortable-macro.workspace = true
 proxmox-sys.workspace = true
 proxmox-systemd.workspace = true
diff --git a/proxmox-backup-client/src/job/mod.rs b/proxmox-backup-client/src/job/mod.rs
new file mode 100644
index 000000000..ffde90ad1
--- /dev/null
+++ b/proxmox-backup-client/src/job/mod.rs
@@ -0,0 +1,22 @@
+//! Scheduled backup jobs: the `job` subcommands, the [`scheduler`] daemon and the [`runner`].
+
+use proxmox_router::cli::*;
+
+use pbs_client::client_job::complete_job_id;
+
+mod runner;
+mod scheduler;
+
+pub fn cli() -> CliCommandMap {
+    CliCommandMap::new()
+        .insert(
+            "run",
+            CliCommand::new(&runner::API_METHOD_RUN_JOB)
+                .arg_param(&["id"])
+                .completion_cb("id", complete_job_id),
+        )
+        .insert(
+            "scheduler",
+            CliCommand::new(&scheduler::API_METHOD_RUN_SCHEDULER),
+        )
+}
diff --git a/proxmox-backup-client/src/job/runner.rs b/proxmox-backup-client/src/job/runner.rs
new file mode 100644
index 000000000..a3bcefeed
--- /dev/null
+++ b/proxmox-backup-client/src/job/runner.rs
@@ -0,0 +1,502 @@
+//! Executing a single backup job, as `proxmox-backup-client job run <id>` does.
+
+use anyhow::{Error, bail, format_err};
+use serde_json::Value;
+
+use proxmox_router::cli::*;
+use proxmox_schema::api;
+
+use pbs_api_types::{BackupNamespace, ClientRateLimitConfig, JOB_ID_SCHEMA};
+use pbs_client::client_job::{self, ClientBackupJobConfig};
+
+#[api(
+    input: {
+        properties: {
+            id: {
+                schema: JOB_ID_SCHEMA,
+            },
+            "dry-run": {
+                type: Boolean,
+                description: "Only show what the job would back up, without uploading \
+                    anything or recording the run.",
+                optional: true,
+                default: false,
+            },
+            scheduled: {
+                type: Boolean,
+                description: "Only run the job if it is due, as the scheduler does.",
+                optional: true,
+                default: false,
+            },
+        },
+    },
+)]
+/// Run a backup job now.
+pub(super) async fn run_job(id: String, dry_run: bool, scheduled: bool) -> Result<(), Error> {
+    // Hold the config lock until the job lock is taken, like `job remove`, so that a removal
+    // cannot slip in between.
+    let config_lock = client_job::lock_config()?;
+
+    let (config, _digest) = client_job::config()?;
+    let job_config = match client_job::find_job(&config, &id)? {
+        Some(job_config) => job_config,
+        None if scheduled => {
+            log::info!("backup job '{id}' no longer exists, not running it");
+            return Ok(());
+        }
+        None => bail!("job '{id}' does not exist"),
+    };
+
+    // even a dry run, so that it cannot overlap with a real one
+    let job = client_job::ClientJob::new(&id)?;
+    drop(config_lock);
+
+    // the daemon decided before the lock was taken, and things may have changed since
+    if scheduled && !still_due(&job_config)? {
+        return Ok(());
+    }
+
+    run_backup_job_once(job, job_config, dry_run).await
+}
+
+/// Whether a job the scheduler picked is still due.
+fn still_due(job_config: &ClientBackupJobConfig) -> Result<bool, Error> {
+    let id = &job_config.id;
+
+    if job_config.disable.unwrap_or(false) {
+        log::info!("backup job '{id}' is disabled, not running it");
+        return Ok(false);
+    }
+
+    let event: proxmox_time::CalendarEvent = job_config.schedule.parse()?;
+    match client_job::next_run_after(&event, id)? {
+        Some(next) if next <= proxmox_time::epoch_i64() => Ok(true),
+        _ => {
+            log::info!("backup job '{id}' is no longer due, not running it");
+            Ok(false)
+        }
+    }
+}
+
+/// Run a backup job and record its outcome.
+async fn run_backup_job_once(
+    mut job: client_job::ClientJob,
+    job_config: ClientBackupJobConfig,
+    dry_run: bool,
+) -> Result<(), Error> {
+    let id = &job_config.id;
+    if !dry_run {
+        job.start()?;
+    }
+
+    let result = run_backup_job_inner(&job_config, dry_run).await;
+
+    // Persist the outcome without replacing a backup error with a state-write error.
+    let record = match &result {
+        Ok(JobOutcome::Completed) if dry_run => {
+            log::info!("dry run of backup job '{id}' finished");
+            Ok(())
+        }
+        // log success only once it is recorded
+        Ok(JobOutcome::Completed) => job.finish(true, None).inspect(|()| {
+            log::info!("backup job '{id}' finished successfully");
+        }),
+        Ok(JobOutcome::SkippedByHook { code }) => {
+            if dry_run {
+                log::info!("backup job '{id}' was skipped by its pre-run hook");
+                Ok(())
+            } else {
+                job.skip(*code).inspect(|()| {
+                    log::info!("backup job '{id}' was skipped by its pre-run hook");
+                })
+            }
+        }
+        Err(_) if dry_run => Ok(()),
+        Err(err) => job.finish(false, Some(truncated_message(err))),
+    };
+    // An unrecorded outcome fails the run, as the job would otherwise stay due and run again.
+    match (result, record) {
+        (Err(err), record) => {
+            if let Err(record_err) = record {
+                log::error!("backup job '{id}': cannot record its outcome: {record_err}");
+            }
+            bail!("backup job '{id}' failed: {err:#}")
+        }
+        (Ok(_), Err(err)) => {
+            bail!("backup job '{id}' ran, but its outcome was not recorded: {err}")
+        }
+        (Ok(_), Ok(())) => Ok(()),
+    }
+}
+
+/// The error for the state file, capped, as it can carry a password helper's output.
+fn truncated_message(err: &Error) -> String {
+    const MAX_LEN: usize = 1024;
+
+    let mut text = format!("{err:#}");
+    if text.len() > MAX_LEN {
+        let end = (0..=MAX_LEN)
+            .rev()
+            .find(|&idx| text.is_char_boundary(idx))
+            .unwrap_or(0);
+        text.truncate(end);
+        text.push_str("[...]");
+    }
+    text
+}
+
+#[derive(Debug)]
+enum JobOutcome {
+    Completed,
+    SkippedByHook { code: i32 },
+}
+
+/// Everything that can fail once the run is recorded as started.
+async fn run_backup_job_inner(
+    job_config: &ClientBackupJobConfig,
+    dry_run: bool,
+) -> Result<JobOutcome, Error> {
+    // a hand-edited job never went through `job create` or `job update`
+    job_config.validate()?;
+
+    if let Some(hook) = &job_config.pre_run_hook {
+        if let Some(outcome) = run_pre_run_hook(&job_config.id, hook).await? {
+            return Ok(outcome);
+        }
+    }
+
+    populate_secret_env(job_config);
+
+    log::info!("starting backup job '{}'", job_config.id);
+    do_create_backup_from_job(job_config, dry_run).await?;
+    Ok(JobOutcome::Completed)
+}
+
+/// Environment variables that can carry a secret into the client.
+const SECRET_ENV_VARS: [&str; 8] = [
+    "PBS_PASSWORD",
+    "PBS_PASSWORD_FD",
+    "PBS_PASSWORD_FILE",
+    "PBS_PASSWORD_CMD",
+    "PBS_ENCRYPTION_PASSWORD",
+    "PBS_ENCRYPTION_PASSWORD_FD",
+    "PBS_ENCRYPTION_PASSWORD_FILE",
+    "PBS_ENCRYPTION_PASSWORD_CMD",
+];
+
+/// A hook that never returns would keep the job locked, so it gets a deadline.
+const PRE_RUN_HOOK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
+
+/// Any other non-zero exit status means a broken hook, not a skip.
+const PRE_RUN_HOOK_SKIP_CODE: i32 = 1;
+
+/// Run a job's pre-run hook: `Ok(None)` proceeds, `Ok(Some(..))` skips, an error fails the run.
+async fn run_pre_run_hook(job_id: &str, hook: &str) -> Result<Option<JobOutcome>, Error> {
+    run_pre_run_hook_timeout(job_id, hook, PRE_RUN_HOOK_TIMEOUT).await
+}
+
+async fn run_pre_run_hook_timeout(
+    job_id: &str,
+    hook: &str,
+    timeout: std::time::Duration,
+) -> Result<Option<JobOutcome>, Error> {
+    use std::os::unix::process::ExitStatusExt;
+
+    log::info!("running pre-run hook for job '{job_id}'");
+    let mut command = tokio::process::Command::new(hook);
+    command
+        .env("PBS_JOB_ID", job_id)
+        .current_dir("/")
+        .stdin(std::process::Stdio::null())
+        // its own process group, so that a timeout also stops what it started
+        .process_group(0)
+        .kill_on_drop(true);
+    // third-party code has no business seeing the backup's secrets
+    for var in SECRET_ENV_VARS {
+        command.env_remove(var);
+    }
+    command.env_remove("CREDENTIALS_DIRECTORY");
+
+    let mut child = command
+        .spawn()
+        .map_err(|err| format_err!("cannot run pre-run hook '{hook}': {err}"))?;
+
+    let status = match tokio::time::timeout(timeout, child.wait()).await {
+        Ok(status) => status.map_err(|err| format_err!("pre-run hook '{hook}' failed: {err}"))?,
+        Err(_) => {
+            kill_hook_process_group(&mut child).await;
+            bail!(
+                "pre-run hook '{hook}' did not finish within {} seconds",
+                timeout.as_secs(),
+            );
+        }
+    };
+
+    if let Some(signal) = status.signal() {
+        bail!("pre-run hook '{hook}' was killed by signal {signal}");
+    }
+    match status.code() {
+        Some(0) => Ok(None),
+        Some(PRE_RUN_HOOK_SKIP_CODE) => Ok(Some(JobOutcome::SkippedByHook {
+            code: PRE_RUN_HOOK_SKIP_CODE,
+        })),
+        Some(code) => bail!(
+            "pre-run hook '{hook}' exited with status {code}, expected 0 to proceed or \
+            {PRE_RUN_HOOK_SKIP_CODE} to skip the run"
+        ),
+        None => bail!("pre-run hook '{hook}' did not exit normally"),
+    }
+}
+
+/// Stop a hook that ran out of time, together with the processes it started.
+async fn kill_hook_process_group(child: &mut tokio::process::Child) {
+    use nix::sys::signal::{Signal, killpg};
+    use nix::unistd::Pid;
+
+    if let Some(pid) = child.id() {
+        // the hook leads its own group, so the group ID is its PID
+        if let Err(err) = killpg(Pid::from_raw(pid as i32), Signal::SIGKILL) {
+            log::warn!("cannot stop the pre-run hook's process group: {err}");
+        }
+    }
+    let _ = child.wait().await;
+}
+
+/// Hand the job's secret sources and fingerprint to the client through the usual `PBS_*` variables.
+///
+/// Inherited variables for a secret the job configures are cleared first, as they would win.
+fn populate_secret_env(job: &pbs_client::client_job::ClientBackupJobConfig) {
+    let set = |name: &str, value: Option<&String>| {
+        if let Some(value) = value {
+            std::env::set_var(name, value);
+        }
+    };
+
+    if job.password_file.is_some() || job.password_cmd.is_some() {
+        for var in SECRET_ENV_VARS
+            .iter()
+            .filter(|var| !var.contains("ENCRYPTION"))
+        {
+            std::env::remove_var(var);
+        }
+    }
+    if job.encryption_password_file.is_some() || job.encryption_password_cmd.is_some() {
+        for var in SECRET_ENV_VARS
+            .iter()
+            .filter(|var| var.contains("ENCRYPTION"))
+        {
+            std::env::remove_var(var);
+        }
+    }
+
+    set("PBS_PASSWORD_FILE", job.password_file.as_ref());
+    set("PBS_PASSWORD_CMD", job.password_cmd.as_ref());
+    set(
+        "PBS_ENCRYPTION_PASSWORD_FILE",
+        job.encryption_password_file.as_ref(),
+    );
+    set(
+        "PBS_ENCRYPTION_PASSWORD_CMD",
+        job.encryption_password_cmd.as_ref(),
+    );
+    set("PBS_FINGERPRINT", job.fingerprint.as_ref());
+}
+
+/// Insert a job option if set, so that `create_backup` applies its own default otherwise.
+fn insert_opt<T: serde::Serialize>(
+    param: &mut serde_json::Map<String, Value>,
+    key: &str,
+    value: Option<&T>,
+) -> Result<(), Error> {
+    if let Some(value) = value {
+        param.insert(key.to_string(), serde_json::to_value(value)?);
+    }
+    Ok(())
+}
+
+/// Run the backup of a job through the regular `create_backup` command.
+async fn do_create_backup_from_job(
+    job: &ClientBackupJobConfig,
+    dry_run: bool,
+) -> Result<(), Error> {
+    let repo = job.resolve_repository()?;
+
+    let mut param = serde_json::Map::new();
+    param.insert("repository".into(), repo.to_string().into());
+    // always set, or `PBS_NAMESPACE` from the environment would apply
+    let ns = job.ns.clone().unwrap_or_else(BackupNamespace::root);
+    param.insert("ns".into(), serde_json::to_value(ns)?);
+    param.insert("backupspec".into(), serde_json::to_value(&job.backupspec)?);
+
+    insert_opt(&mut param, "backup-id", job.backup_id.as_ref())?;
+    insert_opt(&mut param, "backup-type", job.backup_type.as_ref())?;
+    insert_opt(&mut param, "keyfile", job.keyfile.as_ref())?;
+    insert_opt(
+        &mut param,
+        "master-pubkey-file",
+        job.master_pubkey_file.as_ref(),
+    )?;
+    insert_opt(&mut param, "crypt-mode", job.crypt_mode.as_ref())?;
+    insert_opt(&mut param, "exclude", job.exclude.as_ref())?;
+    insert_opt(&mut param, "include-dev", job.include_dev.as_ref())?;
+    insert_opt(
+        &mut param,
+        "all-file-systems",
+        job.all_file_systems.as_ref(),
+    )?;
+    insert_opt(
+        &mut param,
+        "skip-lost-and-found",
+        job.skip_lost_and_found.as_ref(),
+    )?;
+    insert_opt(
+        &mut param,
+        "change-detection-mode",
+        job.change_detection_mode.as_ref(),
+    )?;
+    insert_opt(&mut param, "chunk-size", job.chunk_size.as_ref())?;
+    insert_opt(&mut param, "entries-max", job.entries_max.as_ref())?;
+    insert_opt(
+        &mut param,
+        "skip-e2big-xattr",
+        job.skip_e2big_xattr.as_ref(),
+    )?;
+
+    let param = Value::Object(param);
+    let all_file_systems = job.all_file_systems.unwrap_or(false);
+    let skip_lost_and_found = job.skip_lost_and_found.unwrap_or(false);
+    let skip_e2big_xattr = job.skip_e2big_xattr.unwrap_or(false);
+    let change_detection_mode = job.change_detection_mode.clone();
+    // `ClientRateLimitConfig` has private fields, so build it through serde like the CLI does
+    let mut limit = serde_json::Map::new();
+    insert_opt(&mut limit, "rate", job.rate.as_ref())?;
+    insert_opt(&mut limit, "burst", job.burst.as_ref())?;
+    let limit: ClientRateLimitConfig = serde_json::from_value(Value::Object(limit))?;
+
+    let rpcenv = &mut CliEnvironment::new() as &mut dyn proxmox_router::RpcEnvironment;
+
+    crate::create_backup(
+        param,
+        all_file_systems,
+        skip_lost_and_found,
+        change_detection_mode,
+        dry_run,
+        skip_e2big_xattr,
+        job.no_cache.unwrap_or(false),
+        limit,
+        &crate::API_METHOD_CREATE_BACKUP,
+        rpcenv,
+    )
+    .await?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use proxmox_schema::{ApiType, ObjectSchemaType};
+
+    use pbs_client::client_job::ClientBackupJobConfig;
+
+    use super::*;
+
+    /// Write an executable hook script into a fresh directory and return its path.
+    fn write_hook(name: &str, body: &str) -> std::path::PathBuf {
+        use std::os::unix::fs::PermissionsExt;
+
+        let dir = std::env::temp_dir().join(format!("pbs-hook-test-{}-{name}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let path = dir.join(format!("{name}.sh"));
+        std::fs::write(&path, body).unwrap();
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
+        path
+    }
+
+    /// A hook that runs out of time is killed together with the processes it started.
+    #[tokio::test]
+    async fn a_hook_that_times_out_takes_its_process_group_with_it() {
+        let dir = std::env::temp_dir().join(format!("pbs-hook-test-{}-group", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let pidfile = dir.join("grandchild.pid");
+        let _ = std::fs::remove_file(&pidfile);
+
+        let hook = write_hook(
+            "hangs",
+            &format!(
+                "#!/bin/sh\nsh -c 'echo $$ > {}; exec sleep 60' &\nsleep 60\n",
+                pidfile.display(),
+            ),
+        );
+        let hook = hook.to_str().unwrap();
+
+        // long enough for the helper to record itself, even on a busy machine
+        let err = run_pre_run_hook_timeout("group", hook, std::time::Duration::from_secs(2))
+            .await
+            .expect_err("a hook that sleeps past its timeout must fail the run");
+        assert!(err.to_string().contains("did not finish within"), "{err}");
+
+        let mut recorded = None;
+        for _ in 0..50 {
+            if let Some(pid) = std::fs::read_to_string(&pidfile)
+                .ok()
+                .and_then(|text| text.trim().parse::<i32>().ok())
+            {
+                recorded = Some(pid);
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        }
+        let grandchild = recorded.expect("the hook records the process it started");
+
+        // the group may take a moment to go away
+        let mut alive = true;
+        for _ in 0..50 {
+            if !std::path::Path::new(&format!("/proc/{grandchild}")).exists() {
+                alive = false;
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        }
+        assert!(
+            !alive,
+            "process {grandchild} outlived the hook that started it"
+        );
+
+        let _ = std::fs::remove_file(&pidfile);
+    }
+
+    /// Backup options a job deliberately lacks, each with the reason.
+    const NOT_IN_JOBS: [(&str, &str); 4] = [
+        (
+            "keyfd",
+            "a file descriptor cannot be stored and reopened later",
+        ),
+        ("master-pubkey-fd", "same, the job uses master-pubkey-file"),
+        ("backup-time", "each run picks its own time"),
+        ("dry-run", "per invocation, `job run` has its own flag"),
+    ];
+
+    /// A job must cover every option of a one-shot backup, or list it in `NOT_IN_JOBS`.
+    #[test]
+    fn job_config_covers_the_backup_options() {
+        let job = ClientBackupJobConfig::API_SCHEMA
+            .any_object()
+            .expect("job config schema is an object schema");
+
+        let mut missing = Vec::new();
+        for (name, _optional, _schema) in crate::API_METHOD_CREATE_BACKUP.parameters.properties() {
+            if job.lookup(name).is_some() {
+                continue;
+            }
+            if NOT_IN_JOBS.iter().any(|(excluded, _)| excluded == name) {
+                continue;
+            }
+            missing.push(*name);
+        }
+        assert!(
+            missing.is_empty(),
+            "backup options missing from the job config: {missing:?}. Add them, or list them in \
+            NOT_IN_JOBS with the reason they cannot be stored.",
+        );
+    }
+}
diff --git a/proxmox-backup-client/src/job/scheduler.rs b/proxmox-backup-client/src/job/scheduler.rs
new file mode 100644
index 000000000..e8eaa7d0f
--- /dev/null
+++ b/proxmox-backup-client/src/job/scheduler.rs
@@ -0,0 +1,327 @@
+//! The per-user scheduler daemon: it watches the job config and starts due jobs, each as its own
+//! `proxmox-backup-client job run <id>` process.
+
+use std::collections::{HashMap, HashSet};
+use std::path::{Path, PathBuf};
+
+use anyhow::{Error, bail, format_err};
+
+use proxmox_schema::api;
+use proxmox_section_config::SectionConfigData;
+
+use nix::fcntl::{Flock, FlockArg};
+use pbs_client::client_job::{self, ClientBackupJobConfig};
+
+/// Argv prefix passed to a re-exec of this binary to run a single job.
+/// Kept in sync with the CLI command tree wired up in [`cli`].
+const JOB_RUN_ARGV: [&str; 2] = ["job", "run"];
+
+#[api(
+    input: {
+        properties: {},
+    },
+)]
+/// Run the scheduler daemon that starts due backup jobs.
+pub(super) async fn run_scheduler() -> Result<(), Error> {
+    use tokio::signal::unix::{SignalKind, signal};
+
+    // before the PID file exists, as SIGHUP terminates a process without a handler
+    let mut sighup = signal(SignalKind::hangup())?;
+    let mut sigterm = signal(SignalKind::terminate())?;
+
+    // held for the daemon's lifetime, which keeps a second scheduler from starting
+    let _pid_lock = acquire_scheduler_pidfile(&scheduler_pid_path()?)?;
+    // a broken config at startup would look healthy to systemd
+    client_job::config()?;
+
+    let self_exe = std::env::current_exe()?;
+    let mut children: Vec<(String, tokio::process::Child)> = Vec::new();
+    let mut last_started: HashMap<String, i64> = HashMap::new();
+    // jobs that found no free slot, so that each wait is logged once
+    let mut waiting: HashSet<String> = HashSet::new();
+    // config problems are logged once per config, not on every pass
+    let mut reported_digest = None;
+    let mut reported_load_error = false;
+
+    log::info!("backup job scheduler started (PID {})", std::process::id());
+
+    loop {
+        reap_finished(&mut children);
+        let in_flight: HashSet<String> = children.iter().map(|(id, _)| id.clone()).collect();
+
+        // every run reads the config itself, so a broken one starts no runs at all
+        let plan = match client_job::config() {
+            Ok((config, digest)) => {
+                reported_load_error = false;
+                let report = reported_digest.replace(digest) != Some(digest);
+                plan_jobs(&config, &in_flight, &last_started, report)
+            }
+            Err(err) => {
+                if !std::mem::replace(&mut reported_load_error, true) {
+                    log::error!(
+                        "cannot read the job config, starting no runs until it is fixed: {err}"
+                    );
+                }
+                Plan {
+                    due: Vec::new(),
+                    next_wake: now_plus(MAX_SLEEP),
+                }
+            }
+        };
+        let deferred = start_due_jobs(&self_exe, plan.due, &mut children, &mut last_started);
+        for job_id in deferred.difference(&waiting) {
+            log::info!(
+                "backup job '{job_id}' is queued, {MAX_CONCURRENT_RUNS} backups are already \
+                running"
+            );
+        }
+        waiting = deferred;
+
+        let delay = (plan.next_wake - proxmox_time::epoch_i64()).clamp(0, MAX_SLEEP);
+        tokio::select! {
+            _ = tokio::time::sleep(std::time::Duration::from_secs(delay as u64)) => {}
+            // a finished run frees a slot; `Child::wait` is cancel safe
+            _ = any_child_finished(&mut children) => {}
+            _ = sighup.recv() => log::info!("received SIGHUP, reloading configuration"),
+            _ = sigterm.recv() => {
+                log::info!("received SIGTERM, shutting down");
+                break;
+            }
+        }
+    }
+
+    reap_finished(&mut children);
+    if !children.is_empty() {
+        log::info!("waiting for {} running backup(s) to finish", children.len());
+        let wait_all = async {
+            for (job_id, child) in children.iter_mut() {
+                log_child_result(job_id, child.wait().await);
+            }
+        };
+        // a second SIGTERM stops the wait
+        tokio::select! {
+            _ = wait_all => {}
+            _ = sigterm.recv() => {
+                log::warn!("received second SIGTERM, exiting without waiting for running backups");
+            }
+        }
+    }
+
+    log::info!("backup job scheduler stopped");
+    Ok(())
+}
+
+/// What one planning pass concluded.
+struct Plan {
+    /// IDs of the jobs to start now, the longest-waiting one first.
+    due: Vec<String>,
+    /// Wall-clock time by which the loop has to look again.
+    next_wake: i64,
+}
+
+/// Work out which jobs are due, skipping disabled and running ones. `report` enables the log
+/// messages about the config itself, so that they do not repeat on every pass.
+fn plan_jobs(
+    config: &SectionConfigData,
+    in_flight: &HashSet<String>,
+    last_started: &HashMap<String, i64>,
+    report: bool,
+) -> Plan {
+    use proxmox_time::CalendarEvent;
+
+    let now = proxmox_time::epoch_i64();
+    let mut next_wake = now_plus(MAX_SLEEP);
+    let mut due = Vec::new();
+
+    for job_id in client_job::job_ids(config) {
+        let job: ClientBackupJobConfig = match config.lookup(client_job::JOB_SECTION_TYPE, job_id) {
+            Ok(job) => job,
+            Err(err) => {
+                if report {
+                    log::error!("cannot parse job '{job_id}': {err}");
+                }
+                continue;
+            }
+        };
+
+        if job.disable.unwrap_or(false) {
+            continue;
+        }
+
+        let event: CalendarEvent = match job.schedule.parse() {
+            Ok(event) => event,
+            Err(err) => {
+                if report {
+                    log::error!("job '{job_id}': invalid schedule '{}': {err}", job.schedule);
+                }
+                continue;
+            }
+        };
+
+        let next = match client_job::next_run_after(&event, job_id) {
+            Ok(Some(next)) => next,
+            Ok(None) => {
+                if report {
+                    log::warn!(
+                        "job '{job_id}': schedule '{}' has no future event, the job will not run \
+                        again",
+                        job.schedule,
+                    );
+                }
+                continue;
+            }
+            Err(err) => {
+                if report {
+                    log::error!("job '{job_id}': cannot compute next run: {err}");
+                }
+                continue;
+            }
+        };
+
+        if next > now {
+            next_wake = next_wake.min(next);
+            continue;
+        }
+        if in_flight.contains(job_id) || client_job::is_running(job_id) {
+            continue;
+        }
+        // a run that records nothing leaves its job due, so do not restart it on every wakeup
+        if let Some(started) = last_started.get(job_id) {
+            let earliest = started + MIN_RESTART_INTERVAL;
+            if earliest > now {
+                next_wake = next_wake.min(earliest);
+                continue;
+            }
+        }
+        due.push((next, job_id.clone()));
+    }
+
+    // longest waiting first, rather than in config order
+    due.sort_unstable();
+
+    Plan {
+        due: due.into_iter().map(|(_next, job_id)| job_id).collect(),
+        next_wake,
+    }
+}
+
+/// Start the due jobs, up to the concurrency limit, and return the ones that have to wait.
+fn start_due_jobs(
+    self_exe: &Path,
+    due: Vec<String>,
+    children: &mut Vec<(String, tokio::process::Child)>,
+    last_started: &mut HashMap<String, i64>,
+) -> HashSet<String> {
+    let mut deferred = HashSet::new();
+    for job_id in due {
+        if children.len() >= MAX_CONCURRENT_RUNS {
+            deferred.insert(job_id);
+            continue;
+        }
+        match tokio::process::Command::new(self_exe)
+            .args(JOB_RUN_ARGV)
+            .arg(&job_id)
+            // the run checks again that the job is still due
+            .arg("--scheduled")
+            // never prompt on a terminal, the same as under systemd
+            .stdin(std::process::Stdio::null())
+            .spawn()
+        {
+            Ok(child) => {
+                match child.id() {
+                    Some(pid) => log::info!("spawned backup job '{job_id}' (PID {pid})"),
+                    None => log::info!("spawned backup job '{job_id}'"),
+                }
+                last_started.insert(job_id.clone(), proxmox_time::epoch_i64());
+                children.push((job_id, child));
+            }
+            Err(err) => log::error!("cannot spawn backup job '{job_id}': {err}"),
+        }
+    }
+    deferred
+}
+
+/// Upper bound for one sleep. Timers stop during suspend and wall-clock deadlines move with NTP
+/// and DST, so look again at least once a minute, like the proxy does.
+const MAX_SLEEP: i64 = 60;
+
+/// Shortest time between two starts of the same job, for runs that record no outcome.
+const MIN_RESTART_INTERVAL: i64 = MAX_SLEEP;
+
+/// Backups running at the same time, so that due jobs do not all compete for disk and uplink.
+const MAX_CONCURRENT_RUNS: usize = 2;
+
+fn now_plus(secs: i64) -> i64 {
+    proxmox_time::epoch_i64().saturating_add(secs)
+}
+
+/// Wait until any one of the running jobs finishes, or forever if none is running.
+async fn any_child_finished(children: &mut [(String, tokio::process::Child)]) {
+    if children.is_empty() {
+        std::future::pending::<()>().await;
+        return;
+    }
+    let waits = children
+        .iter_mut()
+        .map(|(_, child)| Box::pin(child.wait()))
+        .collect::<Vec<_>>();
+    let _ = futures::future::select_all(waits).await;
+}
+
+/// Reap the runs that exited, so that they no longer count as in flight.
+fn reap_finished(children: &mut Vec<(String, tokio::process::Child)>) {
+    children.retain_mut(|(job_id, child)| match child.try_wait() {
+        Ok(None) => true,
+        Ok(Some(status)) => {
+            log_child_result(job_id, Ok(status));
+            false
+        }
+        Err(err) => {
+            log_child_result(job_id, Err(err));
+            false
+        }
+    });
+}
+
+/// Log a failed run's exit status. Runs log their own outcome, but a killed one cannot.
+fn log_child_result(job_id: &str, result: std::io::Result<std::process::ExitStatus>) {
+    match result {
+        Ok(status) if status.success() => (),
+        Ok(status) => log::warn!("backup job '{job_id}' failed ({status})"),
+        Err(err) => log::error!("backup job '{job_id}': cannot determine outcome: {err}"),
+    }
+}
+
+fn scheduler_pid_path() -> Result<PathBuf, Error> {
+    Ok(pbs_client::tools::base_directories()?
+        .get_state_home()
+        .join("scheduler.pid"))
+}
+
+/// Lock the PID file and write our PID to it. The kernel drops the lock when the process exits.
+fn acquire_scheduler_pidfile(path: &Path) -> Result<Flock<std::fs::File>, Error> {
+    use std::io::Write;
+
+    if let Some(dir) = path.parent() {
+        proxmox_sys::fs::create_path(dir, None, None)?;
+    }
+    // not truncated before the lock is held, the file may belong to a running scheduler
+    let file = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(path)
+        .map_err(|err| format_err!("cannot open {}: {err}", path.display()))?;
+    let mut locked = match Flock::lock(file, FlockArg::LockExclusiveNonblock) {
+        Ok(locked) => locked,
+        Err((_, nix::errno::Errno::EWOULDBLOCK)) => bail!(
+            "another backup job scheduler is already running ({} is locked)",
+            path.display(),
+        ),
+        Err((_, errno)) => bail!("cannot lock {}: {errno}", path.display()),
+    };
+    locked.set_len(0)?;
+    write!(locked, "{}", std::process::id())?;
+    Ok(locked)
+}
diff --git a/proxmox-backup-client/src/main.rs b/proxmox-backup-client/src/main.rs
index 5238a75ea..5979377a6 100644
--- a/proxmox-backup-client/src/main.rs
+++ b/proxmox-backup-client/src/main.rs
@@ -65,6 +65,7 @@ use pbs_key_config::{KeyConfig, decrypt_key, rsa_encrypt_key_config};
 use pbs_tools::crypt_config::CryptConfig;
 use pbs_tools::json;
 
+pub mod job;
 pub mod key;
 pub mod namespace;
 
@@ -799,7 +800,7 @@ fn spawn_catalog_upload(
  )]
 /// Create (host) backup.
 #[allow(clippy::too_many_arguments)]
-async fn create_backup(
+pub(crate) async fn create_backup(
     param: Value,
     all_file_systems: bool,
     skip_lost_and_found: bool,
@@ -2294,6 +2295,7 @@ fn main() {
         .insert("benchmark", benchmark_cmd_def)
         .insert("change-owner", change_owner_cmd_def)
         .insert("namespace", namespace::cli_map())
+        .insert("job", job::cli())
         .insert("group", group_mgmt_cli())
         .alias(&["files"], &["snapshot", "files"])
         .alias(&["forget"], &["snapshot", "forget"])
diff --git a/proxmox-backup-client/tests/job.rs b/proxmox-backup-client/tests/job.rs
new file mode 100644
index 000000000..bb165434a
--- /dev/null
+++ b/proxmox-backup-client/tests/job.rs
@@ -0,0 +1,263 @@
+//! End-to-end checks of `job run` against the built binary, for the locks, the job state and the
+//! config on disk.
+
+use std::fs::File;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Output};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use nix::errno::Errno;
+use nix::fcntl::{Flock, FlockArg};
+
+/// A private home directory, so that the jobs of whoever runs the tests stay untouched.
+struct Home {
+    dir: PathBuf,
+}
+
+impl Home {
+    fn new(name: &str) -> Self {
+        let dir = Path::new(env!("CARGO_TARGET_TMPDIR"))
+            .join(format!("job-{name}-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).unwrap();
+        Self { dir }
+    }
+
+    fn config_dir(&self) -> PathBuf {
+        self.dir.join("config/proxmox-backup")
+    }
+
+    fn state_dir(&self) -> PathBuf {
+        self.dir.join("state/proxmox-backup/jobstates")
+    }
+
+    fn state_file(&self, id: &str) -> PathBuf {
+        self.state_dir().join(format!("backup-job-{id}.json"))
+    }
+
+    /// Write a job config holding a single job, with `extra` appended to its section.
+    fn write_job(&self, id: &str, extra: &str) {
+        std::fs::create_dir_all(self.config_dir()).unwrap();
+        std::fs::write(
+            self.config_dir().join("jobs.cfg"),
+            format!(
+                "backup-job: {id}\n\tbackupspec etc.pxar:/etc\n\tdatastore store1\n\
+                 \tschedule daily\n{extra}"
+            ),
+        )
+        .unwrap();
+    }
+
+    /// Record that the job was created just now, so that its next event lies ahead.
+    fn write_created_now(&self, id: &str) {
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs();
+        std::fs::create_dir_all(self.state_dir()).unwrap();
+        std::fs::write(
+            self.state_file(id),
+            format!("{{\"created\":{{\"time\":{now}}}}}"),
+        )
+        .unwrap();
+    }
+
+    /// Write an executable hook script and return its path.
+    fn hook(&self, name: &str, body: &str) -> PathBuf {
+        use std::os::unix::fs::PermissionsExt;
+
+        let path = self.dir.join(name);
+        std::fs::write(&path, body).unwrap();
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
+        path
+    }
+
+    fn client(&self) -> Command {
+        let mut command = Command::new(env!("CARGO_BIN_EXE_proxmox-backup-client"));
+        // no stray PBS_* variable from the caller
+        command
+            .env_clear()
+            .env("PATH", "/usr/bin:/bin")
+            .env("HOME", &self.dir)
+            .env("XDG_CONFIG_HOME", self.dir.join("config"))
+            .env("XDG_STATE_HOME", self.dir.join("state"));
+        command
+    }
+
+    fn run(&self, args: &[&str]) -> Output {
+        self.client()
+            .arg("job")
+            .arg("run")
+            .args(args)
+            .output()
+            .unwrap()
+    }
+}
+
+impl Drop for Home {
+    fn drop(&mut self) {
+        let _ = std::fs::remove_dir_all(&self.dir);
+    }
+}
+
+fn stderr(output: &Output) -> String {
+    String::from_utf8_lossy(&output.stderr).into_owned()
+}
+
+/// A run the scheduler starts must check again that the job is due, a manual one must not.
+#[test]
+fn a_scheduled_run_starts_only_a_job_that_is_still_due() {
+    let home = Home::new("scheduled");
+    let marker = home.dir.join("hook-ran");
+    let hook = home.hook(
+        "mark.sh",
+        &format!("#!/bin/sh\ntouch '{}'\nexit 1\n", marker.display()),
+    );
+    let hook_line = format!("\tpre-run-hook {}\n", hook.display());
+
+    // just created, so its next event lies ahead
+    home.write_job("nightly", &hook_line);
+    home.write_created_now("nightly");
+    let out = home.run(&["nightly", "--scheduled"]);
+    assert!(out.status.success(), "{}", stderr(&out));
+    assert!(!marker.exists(), "ran a job that is not due");
+    assert!(stderr(&out).contains("no longer due"), "{}", stderr(&out));
+
+    // due, having never run, but disabled
+    std::fs::remove_file(home.state_file("nightly")).unwrap();
+    home.write_job("nightly", &format!("{hook_line}\tdisable true\n"));
+    let out = home.run(&["nightly", "--scheduled"]);
+    assert!(out.status.success(), "{}", stderr(&out));
+    assert!(!marker.exists(), "ran a disabled job");
+    assert!(stderr(&out).contains("disabled"), "{}", stderr(&out));
+
+    // removed after the daemon picked it
+    let out = home.run(&["removed", "--scheduled"]);
+    assert!(out.status.success(), "{}", stderr(&out));
+    assert!(
+        stderr(&out).contains("no longer exists"),
+        "{}",
+        stderr(&out)
+    );
+
+    // and a job that is due does run, or the checks above would prove nothing
+    home.write_job("nightly", &hook_line);
+    let out = home.run(&["nightly", "--scheduled"]);
+    assert!(out.status.success(), "{}", stderr(&out));
+    assert!(marker.exists(), "did not run a job that is due");
+
+    // a manual run is not bound by the schedule
+    std::fs::remove_file(&marker).unwrap();
+    home.write_created_now("nightly");
+    let out = home.run(&["nightly"]);
+    assert!(out.status.success(), "{}", stderr(&out));
+    assert!(marker.exists(), "a manual run did not run the job");
+}
+
+/// A run whose outcome cannot be recorded must not report success.
+#[test]
+fn a_run_whose_outcome_cannot_be_recorded_fails() {
+    let home = Home::new("unrecorded");
+    let state = home.state_file("nightly");
+
+    // the hook puts a directory where the outcome goes, then asks to skip the run
+    let hook = home.hook(
+        "block.sh",
+        &format!(
+            "#!/bin/sh\nrm -f '{0}'\nmkdir '{0}'\nexit 1\n",
+            state.display()
+        ),
+    );
+    home.write_job("nightly", &format!("\tpre-run-hook {}\n", hook.display()));
+    let out = home.run(&["nightly"]);
+    assert!(!out.status.success(), "reported a skip it did not record");
+    assert!(
+        stderr(&out).contains("outcome was not recorded"),
+        "{}",
+        stderr(&out)
+    );
+    assert!(
+        !stderr(&out).contains("was skipped by its pre-run hook"),
+        "{}",
+        stderr(&out)
+    );
+
+    // a failed run keeps its own error, and also says that it could not record it
+    std::fs::remove_dir(&state).unwrap();
+    let hook = home.hook(
+        "block-and-fail.sh",
+        &format!(
+            "#!/bin/sh\nrm -f '{0}'\nmkdir '{0}'\nexit 2\n",
+            state.display()
+        ),
+    );
+    home.write_job("nightly", &format!("\tpre-run-hook {}\n", hook.display()));
+    let out = home.run(&["nightly"]);
+    assert!(!out.status.success(), "{}", stderr(&out));
+    assert!(
+        stderr(&out).contains("exited with status 2"),
+        "{}",
+        stderr(&out)
+    );
+    assert!(
+        stderr(&out).contains("cannot record its outcome"),
+        "{}",
+        stderr(&out)
+    );
+}
+
+/// Try to take a lock that someone else might hold, reporting whether it was free.
+fn lock_is_free(path: &Path) -> bool {
+    let file = File::options().read(true).write(true).open(path).unwrap();
+    match Flock::lock(file, FlockArg::LockExclusiveNonblock) {
+        Ok(_lock) => true,
+        Err((_, Errno::EWOULDBLOCK)) => false,
+        Err((_, errno)) => panic!("cannot probe {}: {errno}", path.display()),
+    }
+}
+
+/// A run holds the config lock until it has the job lock, so that `job remove` cannot slip in.
+#[test]
+fn a_run_holds_the_config_lock_while_it_waits_for_the_job() {
+    let home = Home::new("lock-order");
+    home.write_job("nightly", "");
+
+    // hold the job lock, as a run of the same job would
+    std::fs::create_dir_all(home.state_dir()).unwrap();
+    let job_lock_path = home.state_dir().join(".backup-job-nightly.lck");
+    let job_lock = Flock::lock(
+        File::create(&job_lock_path).unwrap(),
+        FlockArg::LockExclusive,
+    )
+    .map_err(|(_, errno)| errno)
+    .unwrap();
+
+    let config_lock = home.config_dir().join(".jobs.lck");
+    let run = home
+        .client()
+        .args(["job", "run", "nightly"])
+        .stderr(std::process::Stdio::piped())
+        .spawn()
+        .unwrap();
+
+    // still held a second later, well into the wait, not just while reading the config
+    let started = Instant::now();
+    while !config_lock.exists() || lock_is_free(&config_lock) {
+        assert!(
+            started.elapsed() < Duration::from_secs(2),
+            "the run never took the config lock"
+        );
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    std::thread::sleep(Duration::from_secs(1));
+    assert!(
+        !lock_is_free(&config_lock),
+        "the run let go of the config lock while it waited for the job"
+    );
+
+    // keep the job locked until the run gives up
+    let out = run.wait_with_output().unwrap();
+    drop(job_lock);
+    assert!(!out.status.success());
+    assert!(stderr(&out).contains("already running"), "{}", stderr(&out));
+}
-- 
2.47.3