[PATCH 2/7] client: job: add backup job config and state types

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

Add a per-user jobs.cfg under $XDG_CONFIG_HOME/proxmox-backup/ for
scheduled backups, with the job state under $XDG_STATE_HOME. A job
holds a schedule, repository, sources and the options of a one-shot
backup, plus where its secrets come from and an optional pre-run hook.

Every write is validated, so that mistakes fail at once instead of at
03:00. Paths must be absolute, as a scheduled run does not share the
working directory or $PATH of an interactive shell.

Unknown section types are rejected, so that a misspelled header fails
the parse instead of silently dropping that job.

A per-job flock keeps a job from running twice, and tells a live run
from one that was killed.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
 pbs-client/Cargo.toml        |    3 +-
 pbs-client/src/client_job.rs | 1808 ++++++++++++++++++++++++++++++++++
 pbs-client/src/lib.rs        |    1 +
 3 files changed, 1811 insertions(+), 1 deletion(-)
 create mode 100644 pbs-client/src/client_job.rs

diff --git a/pbs-client/Cargo.toml b/pbs-client/Cargo.toml
index 233969f65..912c44956 100644
--- a/pbs-client/Cargo.toml
+++ b/pbs-client/Cargo.toml
@@ -43,7 +43,8 @@ proxmox-log = { workspace = true }
 proxmox-rate-limiter = { workspace = true, features = [ "rate-limiter" ] }
 proxmox-router = { workspace = true, features = [ "cli", "server" ] }
 proxmox-schema.workspace = true
-proxmox-sys.workspace = true
+proxmox-section-config.workspace = true
+proxmox-sys = { workspace = true, features = [ "timer" ] }
 proxmox-time.workspace = true
 
 pxar.workspace = true
diff --git a/pbs-client/src/client_job.rs b/pbs-client/src/client_job.rs
new file mode 100644
index 000000000..0c89bd107
--- /dev/null
+++ b/pbs-client/src/client_job.rs
@@ -0,0 +1,1808 @@
+//! Client-side backup jobs: their config in `$XDG_CONFIG_HOME/proxmox-backup/jobs.cfg`, their
+//! state and locks in `$XDG_STATE_HOME/proxmox-backup/jobstates/`.
+
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::LazyLock;
+
+use anyhow::{Error, bail, format_err};
+use nix::fcntl::{Flock, FlockArg};
+use nix::sys::stat::Mode;
+use serde::{Deserialize, Serialize};
+
+use proxmox_human_byte::HumanByte;
+use proxmox_schema::{ApiStringFormat, ApiType, Schema, StringSchema, Updater, api};
+use proxmox_section_config::{SectionConfig, SectionConfigData, SectionConfigPlugin};
+use proxmox_sys::fs::{
+    CreateOptions, atomic_open_or_create_file, create_path, file_read_optional_string, lock_file,
+    open_file_locked, replace_file,
+};
+
+use pbs_api_types::{
+    Authid, BACKUP_ID_SCHEMA, BACKUP_TYPE_SCHEMA, BackupNamespace, CERT_FINGERPRINT_SHA256_SCHEMA,
+    CryptMode, DATASTORE_SCHEMA, JOB_ID_SCHEMA, SINGLE_LINE_COMMENT_SCHEMA,
+};
+
+use crate::tools::CHUNK_SIZE_SCHEMA;
+use crate::tools::key_source::{KEYFILE_SCHEMA, MASTER_PUBKEY_FILE_SCHEMA};
+use crate::{
+    BACKUP_REPO_PORT_SCHEMA, BACKUP_REPO_SERVER_SCHEMA, BACKUP_SOURCE_SCHEMA, BackupDetectionMode,
+    BackupRepository, BackupRepositoryArgs, REPO_URL_SCHEMA,
+};
+
+/// Config and state files name where secrets come from, so keep them private.
+const FILE_MODE: Mode = Mode::from_bits_truncate(0o600);
+const DIR_MODE: Mode = Mode::from_bits_truncate(0o700);
+
+fn file_create_options() -> CreateOptions {
+    CreateOptions::new().perm(FILE_MODE)
+}
+
+const JOB_RATE_SCHEMA: Schema = HumanByte::API_SCHEMA
+    .unwrap_string_schema_cloned()
+    .description("Rate limit in bytes per second, with optional unit, for example '10MB'.")
+    .schema();
+
+const JOB_BURST_SCHEMA: Schema = HumanByte::API_SCHEMA
+    .unwrap_string_schema_cloned()
+    .description("Largest burst the rate limit allows, in bytes with optional unit.")
+    .schema();
+
+/// An empty comment would be written back as a bare `comment` key, so drop it instead.
+fn normalize_comment(comment: &str) -> Option<String> {
+    let comment = comment.trim();
+    (!comment.is_empty()).then(|| comment.to_string())
+}
+
+/// The scheduler shares neither the working directory nor `$PATH` with an interactive shell.
+fn check_absolute(name: &str, path: &str) -> Result<(), Error> {
+    if !Path::new(path).is_absolute() {
+        bail!("{name} must be an absolute path: '{path}'");
+    }
+    Ok(())
+}
+
+/// The config format trims values and reads one per line, so reject what would not read back as is.
+fn check_round_trip(name: &str, value: &str) -> Result<(), Error> {
+    if value.trim() != value {
+        bail!("{name} must not start or end with whitespace: '{value}'");
+    }
+    if value.contains(['\n', '\r']) {
+        bail!("{name} must not contain a line break: '{value}'");
+    }
+    Ok(())
+}
+
+fn check_absolute_opt(name: &str, path: Option<&str>) -> Result<(), Error> {
+    match path {
+        Some(path) => check_absolute(name, path),
+        None => Ok(()),
+    }
+}
+
+const CLIENT_BACKUP_SCHEDULE_SCHEMA: Schema = StringSchema::new(
+    "Backup schedule as a calendar event, for example 'daily' or 'mon..fri 02:30'.",
+)
+.format(&ApiStringFormat::VerifyFn(
+    proxmox_time::verify_calendar_event,
+))
+.type_text("<calendar-event>")
+.schema();
+
+// The repository fields mirror `BackupRepositoryArgs` but are not flattened, so that the updater
+// and `DeletableProperty` can handle them like any other field.
+#[api(
+    properties: {
+        id: {
+            schema: JOB_ID_SCHEMA,
+        },
+        schedule: {
+            schema: CLIENT_BACKUP_SCHEDULE_SCHEMA,
+        },
+        repository: {
+            schema: REPO_URL_SCHEMA,
+            optional: true,
+        },
+        server: {
+            schema: BACKUP_REPO_SERVER_SCHEMA,
+            optional: true,
+        },
+        port: {
+            schema: BACKUP_REPO_PORT_SCHEMA,
+            optional: true,
+        },
+        datastore: {
+            schema: DATASTORE_SCHEMA,
+            optional: true,
+        },
+        "auth-id": {
+            type: Authid,
+            optional: true,
+        },
+        ns: {
+            type: BackupNamespace,
+            optional: true,
+        },
+        // Backup sources
+        backupspec: {
+            type: Array,
+            description: "Backup source as <archive-name>.<type>:<source-path>, for example \
+                'root.pxar:/'.",
+            items: {
+                schema: BACKUP_SOURCE_SCHEMA,
+            },
+        },
+        // Auth secrets and TLS pinning
+        "password-file": {
+            description: "File whose first line is the password or API token secret.",
+            optional: true,
+            type: String,
+            min_length: 1,
+        },
+        "password-cmd": {
+            description: "Command that prints the password or API token secret on its first \
+                line. It runs without a shell, so use '/bin/sh -c' for shell syntax.",
+            optional: true,
+            type: String,
+            min_length: 1,
+        },
+        "encryption-password-file": {
+            description: "File whose first line is the passphrase of the key file.",
+            optional: true,
+            type: String,
+            min_length: 1,
+        },
+        "encryption-password-cmd": {
+            description: "Command that prints the passphrase of the key file on its first \
+                line. It runs without a shell, so use '/bin/sh -c' for shell syntax.",
+            optional: true,
+            type: String,
+            min_length: 1,
+        },
+        fingerprint: {
+            schema: CERT_FINGERPRINT_SHA256_SCHEMA,
+            optional: true,
+        },
+        // Backup options
+        "backup-id": {
+            schema: BACKUP_ID_SCHEMA,
+            optional: true,
+        },
+        "backup-type": {
+            schema: BACKUP_TYPE_SCHEMA,
+            optional: true,
+        },
+        keyfile: {
+            schema: KEYFILE_SCHEMA,
+            optional: true,
+        },
+        "master-pubkey-file": {
+            schema: MASTER_PUBKEY_FILE_SCHEMA,
+            optional: true,
+        },
+        "crypt-mode": {
+            type: CryptMode,
+            optional: true,
+        },
+        exclude: {
+            type: Array,
+            description: "List of paths or patterns for matching files to exclude.",
+            optional: true,
+            items: {
+                type: String,
+                description: "Path or match pattern.",
+            },
+        },
+        "include-dev": {
+            type: Array,
+            description: "Include mountpoints with same st_dev number.",
+            optional: true,
+            items: {
+                type: String,
+                description: "Path to file.",
+            },
+        },
+        "all-file-systems": {
+            type: Boolean,
+            description: "Include all mounted subdirectories.",
+            optional: true,
+            default: false,
+        },
+        "skip-lost-and-found": {
+            type: Boolean,
+            description: "Skip lost+found directory.",
+            optional: true,
+            default: false,
+        },
+        "change-detection-mode": {
+            type: BackupDetectionMode,
+            optional: true,
+        },
+        "chunk-size": {
+            schema: CHUNK_SIZE_SCHEMA,
+            optional: true,
+        },
+        rate: {
+            schema: JOB_RATE_SCHEMA,
+            optional: true,
+        },
+        burst: {
+            schema: JOB_BURST_SCHEMA,
+            optional: true,
+        },
+        "entries-max": {
+            type: Integer,
+            description: "Max number of entries to hold in memory.",
+            optional: true,
+        },
+        "skip-e2big-xattr": {
+            type: Boolean,
+            description: "Ignore the E2BIG error when retrieving xattrs.",
+            optional: true,
+            default: false,
+        },
+        "no-cache": {
+            type: Boolean,
+            description: "Bypass local datastore cache for network storages.",
+            optional: true,
+            default: false,
+        },
+        // Job control
+        disable: {
+            type: Boolean,
+            description: "Disable this job.",
+            optional: true,
+            default: false,
+        },
+        comment: {
+            schema: SINGLE_LINE_COMMENT_SCHEMA,
+            optional: true,
+        },
+        "pre-run-hook": {
+            description: "Executable run before each backup. Exit status 0 proceeds, 1 skips \
+                the run, and anything else fails it.",
+            optional: true,
+            type: String,
+            min_length: 1,
+        },
+    },
+)]
+#[derive(Serialize, Deserialize, Clone, Updater)]
+#[serde(rename_all = "kebab-case")]
+/// Client backup job configuration.
+pub struct ClientBackupJobConfig {
+    #[updater(skip)]
+    pub id: String,
+    pub schedule: String,
+
+    // Repository location, either URL or atoms, see `repo_args`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub repository: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub server: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub port: Option<u16>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub datastore: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub auth_id: Option<Authid>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub ns: Option<BackupNamespace>,
+
+    pub backupspec: Vec<String>,
+
+    // Auth secrets and TLS pinning
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub password_file: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub password_cmd: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub encryption_password_file: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub encryption_password_cmd: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub fingerprint: Option<String>,
+
+    // Backup options
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub backup_id: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub backup_type: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub keyfile: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub master_pubkey_file: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub crypt_mode: Option<CryptMode>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub exclude: Option<Vec<String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub include_dev: Option<Vec<String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub all_file_systems: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub skip_lost_and_found: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub change_detection_mode: Option<BackupDetectionMode>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub chunk_size: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub rate: Option<HumanByte>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub burst: Option<HumanByte>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub entries_max: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub skip_e2big_xattr: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub no_cache: Option<bool>,
+
+    // Job control
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub disable: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub comment: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub pre_run_hook: Option<String>,
+}
+
+#[api]
+#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+/// Optional properties of [`ClientBackupJobConfig`] that `job update --delete`
+/// can clear.
+pub enum DeletableProperty {
+    /// Clear an explicit repository URL.
+    Repository,
+    /// Unset the server atom (reverts to the URL form or default).
+    Server,
+    /// Unset the port atom.
+    Port,
+    /// Unset the datastore atom.
+    Datastore,
+    /// Unset the auth-id atom.
+    AuthId,
+    /// Reset the namespace to the root namespace.
+    Ns,
+    /// Clear the password-file path.
+    PasswordFile,
+    /// Clear the password-cmd string.
+    PasswordCmd,
+    /// Clear the encryption-password-file path.
+    EncryptionPasswordFile,
+    /// Clear the encryption-password-cmd string.
+    EncryptionPasswordCmd,
+    /// Clear the pinned server fingerprint.
+    Fingerprint,
+    /// Clear an explicit backup-id (reverts to the host name).
+    BackupId,
+    /// Clear an explicit backup-type (reverts to "host").
+    BackupType,
+    /// Clear the keyfile path.
+    Keyfile,
+    /// Clear the master public key path.
+    MasterPubkeyFile,
+    /// Reset the crypt-mode to the default.
+    CryptMode,
+    /// Drop any configured exclude patterns.
+    Exclude,
+    /// Drop any configured include-dev paths.
+    IncludeDev,
+    /// Reset all-file-systems to the default.
+    AllFileSystems,
+    /// Reset skip-lost-and-found to the default.
+    SkipLostAndFound,
+    /// Reset the change detection mode to the default.
+    ChangeDetectionMode,
+    /// Reset chunk size to the default.
+    ChunkSize,
+    /// Drop the upload rate limit.
+    Rate,
+    /// Drop the upload rate limit's burst size.
+    Burst,
+    /// Reset the entries-max limit to the default.
+    EntriesMax,
+    /// Reset skip-e2big-xattr to the default.
+    SkipE2bigXattr,
+    /// Reset no-cache to the default.
+    NoCache,
+    /// Reset the disable flag.
+    Disable,
+    /// Delete the comment.
+    Comment,
+    /// Clear the pre-run hook.
+    PreRunHook,
+}
+
+impl ClientBackupJobConfig {
+    /// Collect the repository fields into a [`BackupRepositoryArgs`].
+    fn repo_args(&self) -> BackupRepositoryArgs {
+        BackupRepositoryArgs {
+            repository: self.repository.clone(),
+            server: self.server.clone(),
+            port: self.port,
+            datastore: self.datastore.clone(),
+            auth_id: self.auth_id.clone(),
+        }
+    }
+
+    /// Resolve the job's repository, without falling back to `PBS_*` environment variables.
+    pub fn resolve_repository(&self) -> Result<BackupRepository, Error> {
+        BackupRepository::try_from(self.repo_args())
+    }
+
+    /// Store values that have more than one spelling in one form, as every write path must.
+    pub fn normalize(&mut self) {
+        if let Some(comment) = self.comment.take() {
+            self.comment = normalize_comment(&comment);
+        }
+        // the same as `job enable`, which drops the flag
+        if self.disable == Some(false) {
+            self.disable = None;
+        }
+    }
+
+    /// Check a job at create and update time, rather than letting it fail when it runs.
+    pub fn validate(&self) -> Result<(), Error> {
+        proxmox_time::verify_calendar_event(&self.schedule)?;
+        self.resolve_repository()?;
+
+        if matches!(self.crypt_mode, Some(CryptMode::None)) {
+            for (name, path) in [
+                ("keyfile", &self.keyfile),
+                ("master-pubkey-file", &self.master_pubkey_file),
+            ] {
+                if path.is_some() {
+                    bail!("{name} cannot be used with crypt-mode 'none'");
+                }
+            }
+        }
+
+        if self.password_file.is_some() && self.password_cmd.is_some() {
+            bail!("password-file and password-cmd are mutually exclusive");
+        }
+        if self.encryption_password_file.is_some() && self.encryption_password_cmd.is_some() {
+            bail!("encryption-password-file and encryption-password-cmd are mutually exclusive");
+        }
+
+        for (name, cmd) in [
+            ("password-cmd", &self.password_cmd),
+            ("encryption-password-cmd", &self.encryption_password_cmd),
+        ] {
+            let Some(cmd) = cmd else { continue };
+            let args = proxmox_router::cli::shellword_split(cmd)
+                .map_err(|err| format_err!("invalid {name} '{cmd}': {err}"))?;
+            // an empty command would panic in the shared secret lookup
+            let program = args
+                .first()
+                .ok_or_else(|| format_err!("{name} must not be empty"))?;
+            check_absolute(name, program)?;
+        }
+
+        check_absolute_opt("pre-run-hook", self.pre_run_hook.as_deref())?;
+        check_absolute_opt("keyfile", self.keyfile.as_deref())?;
+        check_absolute_opt("master-pubkey-file", self.master_pubkey_file.as_deref())?;
+        check_absolute_opt("password-file", self.password_file.as_deref())?;
+        check_absolute_opt(
+            "encryption-password-file",
+            self.encryption_password_file.as_deref(),
+        )?;
+        for path in self.include_dev.iter().flatten() {
+            check_absolute("include-dev", path)?;
+        }
+        let mut archives = std::collections::HashSet::new();
+        for spec in &self.backupspec {
+            let spec = crate::parse_backup_specification(spec)
+                .map_err(|err| format_err!("invalid backup source '{spec}': {err}"))?;
+            check_absolute("backup source", &spec.config_string)?;
+            if !archives.insert(spec.archive_name.clone()) {
+                bail!(
+                    "duplicate backup source archive name '{}'",
+                    spec.archive_name
+                );
+            }
+        }
+        if self.all_file_systems.unwrap_or(false) && self.include_dev.is_some() {
+            bail!("all-file-systems and include-dev are mutually exclusive");
+        }
+
+        if let Some(size) = self.chunk_size {
+            if !size.is_power_of_two() {
+                bail!("chunk-size must be a power of 2, got {size}");
+            }
+        }
+        if self.burst.is_some() && self.rate.is_none() {
+            bail!("burst requires rate to be set");
+        }
+
+        // Last, so that more specific checks report first. The serialized form covers new fields.
+        let serialized = serde_json::to_value(self)?;
+        for (name, value) in serialized.as_object().into_iter().flatten() {
+            match value {
+                serde_json::Value::String(text) => check_round_trip(name, text)?,
+                serde_json::Value::Array(items) => {
+                    for text in items.iter().filter_map(serde_json::Value::as_str) {
+                        check_round_trip(name, text)?;
+                    }
+                }
+                _ => (),
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Apply deletions, then updates. Setting the repository URL clears the separate repository
+    /// fields and the other way around, so a job can switch forms in one call.
+    pub fn apply_update(
+        &mut self,
+        update: ClientBackupJobConfigUpdater,
+        delete: &[DeletableProperty],
+    ) {
+        for prop in delete {
+            match prop {
+                DeletableProperty::Repository => self.repository = None,
+                DeletableProperty::Server => self.server = None,
+                DeletableProperty::Port => self.port = None,
+                DeletableProperty::Datastore => self.datastore = None,
+                DeletableProperty::AuthId => self.auth_id = None,
+                DeletableProperty::Ns => self.ns = None,
+                DeletableProperty::PasswordFile => self.password_file = None,
+                DeletableProperty::PasswordCmd => self.password_cmd = None,
+                DeletableProperty::EncryptionPasswordFile => self.encryption_password_file = None,
+                DeletableProperty::EncryptionPasswordCmd => self.encryption_password_cmd = None,
+                DeletableProperty::Fingerprint => self.fingerprint = None,
+                DeletableProperty::BackupId => self.backup_id = None,
+                DeletableProperty::BackupType => self.backup_type = None,
+                DeletableProperty::Keyfile => self.keyfile = None,
+                DeletableProperty::MasterPubkeyFile => self.master_pubkey_file = None,
+                DeletableProperty::CryptMode => self.crypt_mode = None,
+                DeletableProperty::Exclude => self.exclude = None,
+                DeletableProperty::IncludeDev => self.include_dev = None,
+                DeletableProperty::AllFileSystems => self.all_file_systems = None,
+                DeletableProperty::SkipLostAndFound => self.skip_lost_and_found = None,
+                DeletableProperty::ChangeDetectionMode => self.change_detection_mode = None,
+                DeletableProperty::ChunkSize => self.chunk_size = None,
+                DeletableProperty::Rate => self.rate = None,
+                DeletableProperty::Burst => self.burst = None,
+                DeletableProperty::EntriesMax => self.entries_max = None,
+                DeletableProperty::SkipE2bigXattr => self.skip_e2big_xattr = None,
+                DeletableProperty::NoCache => self.no_cache = None,
+                DeletableProperty::Disable => self.disable = None,
+                DeletableProperty::Comment => self.comment = None,
+                DeletableProperty::PreRunHook => self.pre_run_hook = None,
+            }
+        }
+
+        let switching_to_url = update.repository.is_some();
+        let switching_to_atoms = update.server.is_some()
+            || update.port.is_some()
+            || update.datastore.is_some()
+            || update.auth_id.is_some();
+        if switching_to_url {
+            self.server = None;
+            self.port = None;
+            self.datastore = None;
+            self.auth_id = None;
+        } else if switching_to_atoms {
+            self.repository = None;
+        }
+
+        if let Some(schedule) = update.schedule {
+            self.schedule = schedule;
+        }
+        if let Some(backupspec) = update.backupspec {
+            self.backupspec = backupspec;
+        }
+        if let Some(repository) = update.repository {
+            self.repository = Some(repository);
+        }
+        if let Some(server) = update.server {
+            self.server = Some(server);
+        }
+        if let Some(port) = update.port {
+            self.port = Some(port);
+        }
+        if let Some(datastore) = update.datastore {
+            self.datastore = Some(datastore);
+        }
+        if let Some(auth_id) = update.auth_id {
+            self.auth_id = Some(auth_id);
+        }
+        if let Some(ns) = update.ns {
+            self.ns = Some(ns);
+        }
+        if let Some(password_file) = update.password_file {
+            self.password_file = Some(password_file);
+        }
+        if let Some(password_cmd) = update.password_cmd {
+            self.password_cmd = Some(password_cmd);
+        }
+        if let Some(encryption_password_file) = update.encryption_password_file {
+            self.encryption_password_file = Some(encryption_password_file);
+        }
+        if let Some(encryption_password_cmd) = update.encryption_password_cmd {
+            self.encryption_password_cmd = Some(encryption_password_cmd);
+        }
+        if let Some(fingerprint) = update.fingerprint {
+            self.fingerprint = Some(fingerprint);
+        }
+        if let Some(backup_id) = update.backup_id {
+            self.backup_id = Some(backup_id);
+        }
+        if let Some(backup_type) = update.backup_type {
+            self.backup_type = Some(backup_type);
+        }
+        if let Some(keyfile) = update.keyfile {
+            self.keyfile = Some(keyfile);
+        }
+        if let Some(master_pubkey_file) = update.master_pubkey_file {
+            self.master_pubkey_file = Some(master_pubkey_file);
+        }
+        if let Some(crypt_mode) = update.crypt_mode {
+            self.crypt_mode = Some(crypt_mode);
+        }
+        if let Some(exclude) = update.exclude {
+            self.exclude = Some(exclude);
+        }
+        if let Some(include_dev) = update.include_dev {
+            self.include_dev = Some(include_dev);
+        }
+        if let Some(all_file_systems) = update.all_file_systems {
+            self.all_file_systems = Some(all_file_systems);
+        }
+        if let Some(skip_lost_and_found) = update.skip_lost_and_found {
+            self.skip_lost_and_found = Some(skip_lost_and_found);
+        }
+        if let Some(change_detection_mode) = update.change_detection_mode {
+            self.change_detection_mode = Some(change_detection_mode);
+        }
+        if let Some(chunk_size) = update.chunk_size {
+            self.chunk_size = Some(chunk_size);
+        }
+        if let Some(rate) = update.rate {
+            self.rate = Some(rate);
+        }
+        if let Some(burst) = update.burst {
+            self.burst = Some(burst);
+        }
+        if let Some(entries_max) = update.entries_max {
+            self.entries_max = Some(entries_max);
+        }
+        if let Some(skip_e2big_xattr) = update.skip_e2big_xattr {
+            self.skip_e2big_xattr = Some(skip_e2big_xattr);
+        }
+        if let Some(no_cache) = update.no_cache {
+            self.no_cache = Some(no_cache);
+        }
+        if let Some(disable) = update.disable {
+            self.disable = Some(disable);
+        }
+        if let Some(comment) = update.comment {
+            self.comment = Some(comment);
+        }
+        if let Some(pre_run_hook) = update.pre_run_hook {
+            self.pre_run_hook = Some(pre_run_hook);
+        }
+
+        self.normalize();
+    }
+}
+
+pub static CONFIG: LazyLock<SectionConfig> = LazyLock::new(init);
+
+/// Section type of a backup job, and the type key used in `jobs.cfg`.
+pub const JOB_SECTION_TYPE: &str = "backup-job";
+
+fn init() -> SectionConfig {
+    let obj_schema = ClientBackupJobConfig::API_SCHEMA.unwrap_object_schema();
+
+    let plugin = SectionConfigPlugin::new(
+        JOB_SECTION_TYPE.to_string(),
+        Some(String::from("id")),
+        obj_schema,
+    );
+    // no unknown section types, so a misspelled header fails instead of dropping that job
+    let mut config = SectionConfig::new(&JOB_ID_SCHEMA);
+    config.register_plugin(plugin);
+    config
+}
+
+/// The job IDs of a config, sorted for a stable `job list` and a deterministic scheduler.
+pub fn job_ids(data: &SectionConfigData) -> impl Iterator<Item = &String> {
+    let mut ids: Vec<&String> = data.sections.keys().collect();
+    ids.sort_unstable();
+    ids.into_iter()
+}
+
+/// Find a backup job by ID, if the config has one.
+pub fn find_job(
+    config: &SectionConfigData,
+    id: &str,
+) -> Result<Option<ClientBackupJobConfig>, Error> {
+    if !config.sections.contains_key(id) {
+        return Ok(None);
+    }
+    config.lookup(JOB_SECTION_TYPE, id).map(Some)
+}
+
+/// Look up a backup job by ID, failing if there is none.
+pub fn lookup_job(config: &SectionConfigData, id: &str) -> Result<ClientBackupJobConfig, Error> {
+    find_job(config, id)?.ok_or_else(|| format_err!("job '{id}' does not exist"))
+}
+
+const JOBS_CFG_FILENAME: &str = "jobs.cfg";
+const JOBS_CFG_LOCKFILE: &str = ".jobs.lck";
+
+fn config_path() -> Result<PathBuf, Error> {
+    crate::tools::place_xdg_file(JOBS_CFG_FILENAME, "backup job config")
+}
+
+fn lockfile_path() -> Result<PathBuf, Error> {
+    crate::tools::place_xdg_file(JOBS_CFG_LOCKFILE, "backup job lock")
+}
+
+/// Acquire an exclusive lock on the job config file.
+pub fn lock_config() -> Result<std::fs::File, Error> {
+    let path = lockfile_path()?;
+    open_file_locked(
+        &path,
+        std::time::Duration::from_secs(10),
+        true,
+        file_create_options(),
+    )
+    .map_err(|err| format_err!("cannot lock the job config: {err}"))
+}
+
+/// Read the job config, with a digest for change detection.
+pub fn config() -> Result<(SectionConfigData, [u8; 32]), Error> {
+    let path = config_path()?;
+    let content = file_read_optional_string(&path)?.unwrap_or_default();
+    let digest = openssl::sha::sha256(content.as_bytes());
+    let data = CONFIG.parse(&path, &content)?;
+    Ok((data, digest))
+}
+
+/// Write the job config file.
+pub fn save_config(data: &SectionConfigData) -> Result<(), Error> {
+    let path = config_path()?;
+    let raw = CONFIG.write(&path, data)?;
+    replace_file(&path, raw.as_bytes(), file_create_options(), true)
+}
+
+/// Shell completion helper for job IDs.
+pub fn complete_job_id(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
+    match config() {
+        Ok((data, _)) => job_ids(&data).map(|id| id.to_string()).collect(),
+        Err(_) => Vec::new(),
+    }
+}
+
+const JOBSTATE_DIRNAME: &str = "jobstates";
+
+fn state_dir() -> Result<PathBuf, Error> {
+    Ok(crate::tools::base_directories()?
+        .get_state_home()
+        .join(JOBSTATE_DIRNAME))
+}
+
+/// Create the state directory, only when about to write, so that `job list` creates nothing.
+fn create_state_dir() -> Result<PathBuf, Error> {
+    let dir = state_dir()?;
+    create_path(&dir, None, Some(CreateOptions::new().perm(DIR_MODE)))?;
+    Ok(dir)
+}
+
+fn state_path(id: &str) -> Result<PathBuf, Error> {
+    Ok(state_dir()?.join(format!("backup-job-{id}.json")))
+}
+
+fn state_lock_path(id: &str) -> Result<PathBuf, Error> {
+    Ok(state_dir()?.join(format!(".backup-job-{id}.lck")))
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+/// On-disk state of a client backup job, as written to its state file.
+enum ClientJobState {
+    /// Job was created but never started.
+    Created { time: i64 },
+    /// Job is running. The PID is informational, liveness comes from the job lock.
+    Running { pid: u32, start_time: i64 },
+    /// Job finished (successfully or not).
+    Finished {
+        start_time: i64,
+        end_time: i64,
+        success: bool,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        message: Option<String>,
+    },
+    /// The pre-run hook gated this run, so no backup was made.
+    Skipped {
+        start_time: i64,
+        end_time: i64,
+        code: i32,
+    },
+}
+
+impl ClientJobState {
+    /// Timestamp of this state transition, used for schedule computation.
+    fn time(&self) -> i64 {
+        match *self {
+            ClientJobState::Created { time } => time,
+            ClientJobState::Running { start_time, .. } => start_time,
+            ClientJobState::Finished { end_time, .. }
+            | ClientJobState::Skipped { end_time, .. } => end_time,
+        }
+    }
+}
+
+/// Read a job's state. A broken state file counts as none, so that it cannot wedge the job.
+fn read_state(id: &str) -> Option<ClientJobState> {
+    let path = state_path(id).ok()?;
+    let raw = match file_read_optional_string(&path) {
+        Ok(Some(raw)) => raw,
+        Ok(None) => return None,
+        Err(err) => {
+            proxmox_log::warn!("job '{id}': cannot read state file: {err}");
+            return None;
+        }
+    };
+    match serde_json::from_str(&raw) {
+        Ok(state) => Some(state),
+        Err(err) => {
+            proxmox_log::warn!("job '{id}': ignoring unparsable state file: {err}");
+            None
+        }
+    }
+}
+
+#[api]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+/// Outcome of a backup job's most recent run.
+pub enum JobResult {
+    /// The job was created but has not run yet.
+    Created,
+    /// A run is currently in progress.
+    Running,
+    /// The last run completed successfully.
+    Ok,
+    /// The last run failed.
+    Failed,
+    /// The last run was skipped by the pre-run hook.
+    Skipped,
+    /// A run was started but never completed, for example because it was killed.
+    Interrupted,
+}
+
+/// When `event` next fires after the job's last run. A job without state is due at once.
+pub fn next_run_after(event: &proxmox_time::CalendarEvent, id: &str) -> Result<Option<i64>, Error> {
+    event.compute_next_event(read_state(id).map_or(0, |state| state.time()))
+}
+
+#[api(
+    properties: {
+        config: {
+            type: ClientBackupJobConfig,
+            flatten: true,
+        },
+        "last-run": {
+            description: "Time of the last run, in seconds since the epoch.",
+            type: Integer,
+            optional: true,
+        },
+        "last-result": {
+            type: JobResult,
+            optional: true,
+        },
+        message: {
+            description: "Error of the last run, or why it was skipped.",
+            type: String,
+            optional: true,
+        },
+        "next-run": {
+            description: "When the job is due next, in seconds since the epoch. Unset if the \
+                job is disabled or its schedule has no future event.",
+            type: Integer,
+            optional: true,
+        },
+    },
+)]
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+/// A configured backup job together with the summary of its most recent run.
+pub struct ClientBackupJobStatus {
+    #[serde(flatten)]
+    pub config: ClientBackupJobConfig,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub last_run: Option<i64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub last_result: Option<JobResult>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub message: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub next_run: Option<i64>,
+}
+
+/// When a job is due next, counting from `last`. An overdue job is due now, not in the past.
+fn next_run(config: &ClientBackupJobConfig, last: i64) -> Option<i64> {
+    if config.disable.unwrap_or(false) {
+        return None;
+    }
+    let event: proxmox_time::CalendarEvent = config.schedule.parse().ok()?;
+    let next = event.compute_next_event(last).ok()??;
+    Some(next.max(proxmox_time::epoch_i64()))
+}
+
+impl ClientBackupJobStatus {
+    /// Pair a job config with its state. A `Running` state without a lock holder is interrupted.
+    pub fn with_state(config: ClientBackupJobConfig) -> Self {
+        let state = read_state(&config.id);
+        let next_run = next_run(&config, state.as_ref().map_or(0, ClientJobState::time));
+        let Some(state) = state else {
+            return Self {
+                config,
+                last_run: None,
+                last_result: None,
+                message: None,
+                next_run,
+            };
+        };
+
+        let last_run = match state {
+            ClientJobState::Created { .. } => None,
+            _ => Some(state.time()),
+        };
+        let (last_result, message) = match state {
+            ClientJobState::Created { .. } => (JobResult::Created, None),
+            ClientJobState::Running { .. } if is_running(&config.id) => (JobResult::Running, None),
+            ClientJobState::Running { .. } => (JobResult::Interrupted, None),
+            ClientJobState::Finished {
+                success: true,
+                message,
+                ..
+            } => (JobResult::Ok, message),
+            ClientJobState::Finished { message, .. } => (JobResult::Failed, message),
+            ClientJobState::Skipped { code, .. } => (
+                JobResult::Skipped,
+                Some(format!("pre-run hook exited with status {code}")),
+            ),
+        };
+
+        Self {
+            config,
+            last_run,
+            last_result: Some(last_result),
+            message,
+            next_run,
+        }
+    }
+}
+
+/// Not zero, so that a liveness probe holding the lock for a moment cannot fail a run.
+const JOB_LOCK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
+
+/// Whether someone holds an exclusive lock on `path`.
+///
+/// The probe takes a shared lock, so that two probes do not see each other as the holder.
+pub fn is_locked(path: &Path) -> bool {
+    let Ok(file) = std::fs::File::open(path) else {
+        return false;
+    };
+    matches!(
+        Flock::lock(file, FlockArg::LockSharedNonblock),
+        Err((_, nix::errno::Errno::EWOULDBLOCK))
+    )
+}
+
+/// Whether a run of the job holds its lock.
+pub fn is_running(id: &str) -> bool {
+    state_lock_path(id).is_ok_and(|path| is_locked(&path))
+}
+
+/// A running job, holding the lock that keeps it from running twice.
+pub struct ClientJob {
+    id: String,
+    start_time: i64,
+    _lock: std::fs::File,
+}
+
+impl ClientJob {
+    /// Take the job lock, failing if the job is already running. Callers must hold the config lock,
+    /// so that `job remove` cannot unlink the lock file under a run waiting for it.
+    pub fn new(id: &str) -> Result<Self, Error> {
+        create_state_dir()?;
+        let lock_path = state_lock_path(id)?;
+        // opened separately, so that an open error is not reported as a running job
+        let mut lock = atomic_open_or_create_file(
+            &lock_path,
+            nix::fcntl::OFlag::O_RDWR | nix::fcntl::OFlag::O_CLOEXEC | nix::fcntl::OFlag::O_APPEND,
+            &[],
+            file_create_options(),
+            false,
+        )
+        .map_err(|err| format_err!("cannot open the lock file of job '{id}': {err}"))?;
+        lock_file(&mut lock, true, Some(JOB_LOCK_TIMEOUT)).map_err(|err| {
+            // the timeout interrupts the wait
+            if err.kind() == std::io::ErrorKind::Interrupted {
+                format_err!("job '{id}' is already running")
+            } else {
+                format_err!("cannot lock job '{id}': {err}")
+            }
+        })?;
+
+        Ok(Self {
+            id: id.to_string(),
+            start_time: 0,
+            _lock: lock,
+        })
+    }
+
+    /// Mark the job as running.
+    pub fn start(&mut self) -> Result<(), Error> {
+        self.start_time = proxmox_time::epoch_i64();
+        let state = ClientJobState::Running {
+            pid: std::process::id(),
+            start_time: self.start_time,
+        };
+        self.write_state(&state)
+    }
+
+    /// Mark the job as finished.
+    pub fn finish(&self, success: bool, message: Option<String>) -> Result<(), Error> {
+        let state = ClientJobState::Finished {
+            start_time: self.start_time,
+            end_time: proxmox_time::epoch_i64(),
+            success,
+            message,
+        };
+        self.write_state(&state)
+    }
+
+    /// Record that the pre-run hook gated this run, so no backup was made.
+    pub fn skip(&self, code: i32) -> Result<(), Error> {
+        let state = ClientJobState::Skipped {
+            start_time: self.start_time,
+            end_time: proxmox_time::epoch_i64(),
+            code,
+        };
+        self.write_state(&state)
+    }
+
+    fn write_state(&self, state: &ClientJobState) -> Result<(), Error> {
+        let path = state_path(&self.id)?;
+        let data = serde_json::to_string_pretty(state)?;
+        replace_file(&path, data.as_bytes(), file_create_options(), true)
+    }
+}
+
+/// Record a new job as created, so that it does not count as overdue.
+pub fn create_state(id: &str) -> Result<(), Error> {
+    create_state_dir()?;
+    let state = ClientJobState::Created {
+        time: proxmox_time::epoch_i64(),
+    };
+    let path = state_path(id)?;
+    let data = serde_json::to_string_pretty(&state)?;
+    replace_file(&path, data.as_bytes(), file_create_options(), true)
+}
+
+/// Remove a job's state and lock file. Takes the held lock, as unlinking it is only safe then.
+pub fn remove_state(job: &ClientJob) -> Result<(), Error> {
+    for path in [state_path(&job.id)?, state_lock_path(&job.id)?] {
+        match std::fs::remove_file(&path) {
+            Ok(()) => (),
+            Err(err) if err.kind() == std::io::ErrorKind::NotFound => (),
+            Err(err) => return Err(err.into()),
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pbs_api_types::Authid;
+    use proxmox_section_config::SectionConfigData;
+
+    fn minimal_job(id: &str) -> ClientBackupJobConfig {
+        ClientBackupJobConfig {
+            id: id.to_string(),
+            schedule: "daily".to_string(),
+            repository: None,
+            server: None,
+            port: None,
+            datastore: Some("store1".to_string()),
+            auth_id: None,
+            ns: None,
+            backupspec: vec!["root.pxar:/".to_string()],
+            password_file: None,
+            password_cmd: None,
+            encryption_password_file: None,
+            encryption_password_cmd: None,
+            fingerprint: None,
+            backup_id: None,
+            backup_type: None,
+            keyfile: None,
+            master_pubkey_file: None,
+            crypt_mode: None,
+            exclude: None,
+            include_dev: None,
+            all_file_systems: None,
+            skip_lost_and_found: None,
+            change_detection_mode: None,
+            chunk_size: None,
+            rate: None,
+            burst: None,
+            entries_max: None,
+            skip_e2big_xattr: None,
+            no_cache: None,
+            disable: None,
+            comment: None,
+            pre_run_hook: None,
+        }
+    }
+
+    fn round_trip(job: &ClientBackupJobConfig) -> ClientBackupJobConfig {
+        let mut data = SectionConfigData::new();
+        data.set_data(&job.id, JOB_SECTION_TYPE, job).unwrap();
+        let raw = CONFIG.write("<test>", &data).unwrap();
+        let parsed = CONFIG.parse("<test>", &raw).unwrap();
+        parsed.lookup(JOB_SECTION_TYPE, &job.id).unwrap()
+    }
+
+    #[test]
+    fn section_config_round_trip_minimal_atoms() {
+        let job = minimal_job("minimal-atoms");
+        let got = round_trip(&job);
+        assert_eq!(got.schedule, "daily");
+        assert_eq!(got.datastore.as_deref(), Some("store1"));
+        assert!(got.repository.is_none());
+        assert!(got.server.is_none());
+        assert_eq!(got.backupspec, vec!["root.pxar:/"]);
+    }
+
+    /// A job can also be configured with a plain repository URL instead of atoms.
+    #[test]
+    fn section_config_round_trip_repository_url() {
+        let mut job = minimal_job("url-job");
+        job.datastore = None;
+        job.repository = Some("user@pbs@backup.example.com:8008:store1".to_string());
+        let got = round_trip(&job);
+        assert_eq!(
+            got.repository.as_deref(),
+            Some("user@pbs@backup.example.com:8008:store1"),
+        );
+        assert!(got.datastore.is_none());
+        let repo = got.resolve_repository().unwrap();
+        assert_eq!(repo.host(), "backup.example.com");
+        assert_eq!(repo.port(), 8008);
+        assert_eq!(repo.store(), "store1");
+    }
+
+    #[test]
+    fn section_config_round_trip_full() {
+        let auth_id: Authid = "testuser@pam".parse().unwrap();
+        let ns: BackupNamespace = "prod/db".parse().unwrap();
+
+        let job = ClientBackupJobConfig {
+            id: "full-job".to_string(),
+            schedule: "mon..fri 02:00".to_string(),
+            repository: None,
+            server: Some("backup.example.com".to_string()),
+            port: Some(8008),
+            datastore: Some("bigstore".to_string()),
+            auth_id: Some(auth_id.clone()),
+            ns: Some(ns.clone()),
+            backupspec: vec!["root.pxar:/".to_string()],
+            password_file: Some("/run/secrets/pw".to_string()),
+            password_cmd: None,
+            encryption_password_file: Some("/run/secrets/enc-pw".to_string()),
+            encryption_password_cmd: None,
+            fingerprint: Some(
+                "aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99:\
+                 aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99"
+                    .to_string(),
+            ),
+            backup_id: Some("myhost".to_string()),
+            backup_type: Some("host".to_string()),
+            keyfile: Some("/etc/pbs/key.json".to_string()),
+            master_pubkey_file: Some("/etc/pbs/master.pem".to_string()),
+            crypt_mode: Some(CryptMode::Encrypt),
+            exclude: Some(vec!["/tmp".to_string(), "/proc".to_string()]),
+            include_dev: Some(vec!["/mnt/data".to_string()]),
+            all_file_systems: Some(true),
+            skip_lost_and_found: Some(true),
+            change_detection_mode: Some(BackupDetectionMode::Metadata),
+            chunk_size: Some(4096),
+            rate: Some(HumanByte::from(10 * 1024 * 1024_usize)),
+            burst: Some(HumanByte::from(20 * 1024 * 1024_usize)),
+            entries_max: Some(65536),
+            skip_e2big_xattr: Some(true),
+            no_cache: Some(true),
+            disable: Some(false),
+            comment: Some("nightly backup".to_string()),
+            pre_run_hook: Some("/usr/local/bin/pre-backup.sh".to_string()),
+        };
+
+        let got = round_trip(&job);
+        assert_eq!(got.schedule, "mon..fri 02:00");
+        assert_eq!(got.server.as_deref(), Some("backup.example.com"));
+        assert_eq!(got.port, Some(8008));
+        assert_eq!(got.datastore.as_deref(), Some("bigstore"));
+        assert_eq!(
+            got.auth_id.as_ref().unwrap().to_string(),
+            auth_id.to_string(),
+        );
+        assert_eq!(got.ns.as_ref().unwrap().to_string(), ns.to_string());
+        assert_eq!(got.password_file.as_deref(), Some("/run/secrets/pw"));
+        assert_eq!(
+            got.encryption_password_file.as_deref(),
+            Some("/run/secrets/enc-pw"),
+        );
+        assert_eq!(
+            got.fingerprint.as_deref(),
+            Some(
+                "aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99:\
+                 aa:bb:cc:dd:ee:ff:00:11:22:33:44:55:66:77:88:99",
+            ),
+        );
+        assert_eq!(got.backup_id.as_deref(), Some("myhost"));
+        assert_eq!(got.backup_type.as_deref(), Some("host"));
+        assert_eq!(got.keyfile.as_deref(), Some("/etc/pbs/key.json"));
+        assert_eq!(got.crypt_mode, Some(CryptMode::Encrypt));
+        assert_eq!(got.exclude.as_ref().unwrap(), &["/tmp", "/proc"]);
+        assert_eq!(got.include_dev.as_ref().unwrap(), &["/mnt/data"]);
+        assert_eq!(got.all_file_systems, Some(true));
+        assert_eq!(got.skip_lost_and_found, Some(true));
+        assert_eq!(
+            got.change_detection_mode,
+            Some(BackupDetectionMode::Metadata)
+        );
+        assert_eq!(got.chunk_size, Some(4096));
+        assert_eq!(got.entries_max, Some(65536));
+        assert_eq!(got.skip_e2big_xattr, Some(true));
+        assert_eq!(got.disable, Some(false));
+        assert_eq!(got.comment.as_deref(), Some("nightly backup"));
+        assert_eq!(
+            got.pre_run_hook.as_deref(),
+            Some("/usr/local/bin/pre-backup.sh"),
+        );
+    }
+
+    #[test]
+    fn section_config_round_trip_repeated_keys() {
+        let mut job = minimal_job("multi-spec");
+        job.schedule = "hourly".to_string();
+        job.backupspec = vec![
+            "root.pxar:/".to_string(),
+            "home.pxar:/home".to_string(),
+            "var.pxar:/var".to_string(),
+        ];
+        job.exclude = Some(vec![
+            "/tmp".to_string(),
+            "/dev".to_string(),
+            "/sys".to_string(),
+        ]);
+
+        let got = round_trip(&job);
+        assert_eq!(
+            got.backupspec,
+            vec!["root.pxar:/", "home.pxar:/home", "var.pxar:/var"]
+        );
+        assert_eq!(got.exclude.as_ref().unwrap(), &["/tmp", "/dev", "/sys"]);
+    }
+
+    #[test]
+    fn resolve_repository_requires_datastore() {
+        let mut job = minimal_job("bad");
+        job.datastore = None;
+        job.server = Some("pbs.local".to_string());
+        let err = job.resolve_repository().unwrap_err().to_string();
+        assert!(err.contains("--datastore is required"), "got: {err}");
+    }
+
+    #[test]
+    fn validate_rejects_bad_schedule() {
+        let mut job = minimal_job("bad-schedule");
+        job.schedule = "definitely not a calendar event".to_string();
+        assert!(job.validate().is_err());
+    }
+
+    #[test]
+    fn validate_rejects_relative_pre_run_hook() {
+        let mut job = minimal_job("rel-hook");
+        job.pre_run_hook = Some("./local-hook.sh".into());
+        let err = job.validate().unwrap_err().to_string();
+        assert!(
+            err.contains("pre-run-hook must be an absolute path"),
+            "got: {err}"
+        );
+    }
+
+    #[test]
+    fn validate_accepts_absolute_pre_run_hook() {
+        let mut job = minimal_job("abs-hook");
+        job.pre_run_hook = Some("/usr/local/bin/hook.sh".into());
+        job.validate().unwrap();
+    }
+
+    #[test]
+    fn validate_rejects_unparseable_password_cmd() {
+        let mut job = minimal_job("bad-cmd");
+        // unmatched quote
+        job.password_cmd = Some("pass show 'pbs/nightly".to_string());
+        let err = job.validate().unwrap_err().to_string();
+        assert!(err.contains("password-cmd"), "got: {err}");
+    }
+
+    /// An empty command has no program to execute and used to panic in the shared secret lookup.
+    #[test]
+    fn validate_rejects_empty_password_cmd() {
+        for cmd in ["", "   ", "\t"] {
+            let mut job = minimal_job("empty-cmd");
+            job.password_cmd = Some(cmd.to_string());
+            let err = job.validate().unwrap_err().to_string();
+            assert!(err.contains("password-cmd must not be empty"), "got: {err}");
+        }
+    }
+
+    #[test]
+    fn validate_rejects_relative_secret_cmd() {
+        let mut job = minimal_job("rel-cmd");
+        job.encryption_password_cmd = Some("get-pw --job nightly".to_string());
+        let err = job.validate().unwrap_err().to_string();
+        assert!(
+            err.contains("encryption-password-cmd must be an absolute path"),
+            "got: {err}"
+        );
+    }
+
+    #[test]
+    fn validate_rejects_conflicting_secret_sources() {
+        let mut job = minimal_job("two-sources");
+        job.password_file = Some("/run/secrets/pw".to_string());
+        job.password_cmd = Some("/usr/bin/pass show pbs".to_string());
+        let err = job.validate().unwrap_err().to_string();
+        assert!(err.contains("mutually exclusive"), "got: {err}");
+    }
+
+    /// The name a relative path is reported under, plus a setter that puts one into a config.
+    type RelativePathCase = (&'static str, fn(&mut ClientBackupJobConfig));
+
+    /// The scheduler and `job run` resolve relative paths differently, so none are accepted.
+    #[test]
+    fn validate_rejects_relative_paths() {
+        let cases: &[RelativePathCase] = &[
+            ("keyfile", |j| j.keyfile = Some("key.json".into())),
+            ("password-file", |j| j.password_file = Some("pw".into())),
+            ("encryption-password-file", |j| {
+                j.encryption_password_file = Some("enc-pw".into())
+            }),
+            ("include-dev", |j| {
+                j.include_dev = Some(vec!["mnt/data".into()])
+            }),
+            ("backup source", |j| {
+                j.backupspec = vec!["root.pxar:Documents".into()]
+            }),
+        ];
+
+        for (name, setup) in cases {
+            let mut job = minimal_job("rel");
+            setup(&mut job);
+            let err = job.validate().unwrap_err().to_string();
+            assert!(
+                err.contains(name) && err.contains("must be an absolute path"),
+                "{name}: got: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn validate_rejects_unparseable_backupspec() {
+        let mut job = minimal_job("bad-spec");
+        job.backupspec = vec!["not-a-spec".to_string()];
+        let err = job.validate().unwrap_err().to_string();
+        assert!(err.contains("invalid backup source"), "got: {err}");
+    }
+
+    /// The backup code enforces the power of 2 only deep inside, so `validate` checks it up front.
+    #[test]
+    fn validate_rejects_non_power_of_two_chunk_size() {
+        let mut job = minimal_job("chunk");
+        job.chunk_size = Some(100);
+        let err = job.validate().unwrap_err().to_string();
+        assert!(err.contains("power of 2"), "got: {err}");
+
+        job.chunk_size = Some(1024);
+        job.validate().unwrap();
+    }
+
+    #[test]
+    fn validate_rejects_burst_without_rate() {
+        let mut job = minimal_job("burst");
+        job.burst = Some(HumanByte::from(1024_usize));
+        let err = job.validate().unwrap_err().to_string();
+        assert!(err.contains("burst requires rate"), "got: {err}");
+
+        job.rate = Some(HumanByte::from(1024_usize));
+        job.validate().unwrap();
+    }
+
+    // -- apply_update / DeletableProperty tests ----------------------------
+
+    fn updater() -> ClientBackupJobConfigUpdater {
+        // all fields unset, a no-op update
+        serde_json::from_value(serde_json::json!({})).unwrap()
+    }
+
+    #[test]
+    fn apply_update_patches_optional_field() {
+        let mut job = minimal_job("p");
+        job.comment = None;
+        let mut up = updater();
+        up.comment = Some("hello".to_string());
+        job.apply_update(up, &[]);
+        assert_eq!(job.comment.as_deref(), Some("hello"));
+    }
+
+    /// Both write paths have to store the same comment.
+    #[test]
+    fn comment_is_trimmed_and_empty_becomes_absent() {
+        for (input, expected) in [
+            ("  spaced  ", Some("spaced")),
+            ("   ", None),
+            ("", None),
+            ("kept", Some("kept")),
+        ] {
+            let mut job = minimal_job("cmt");
+            job.comment = Some(input.to_string());
+            job.normalize();
+            assert_eq!(job.comment.as_deref(), expected, "normalize({input:?})");
+
+            let mut job = minimal_job("cmt");
+            let mut up = updater();
+            up.comment = Some(input.to_string());
+            job.apply_update(up, &[]);
+            assert_eq!(job.comment.as_deref(), expected, "apply_update({input:?})");
+        }
+    }
+
+    #[test]
+    fn apply_update_no_op_for_missing_fields() {
+        let mut job = minimal_job("p");
+        job.comment = Some("keep me".to_string());
+        job.apply_update(updater(), &[]);
+        assert_eq!(job.comment.as_deref(), Some("keep me"));
+    }
+
+    #[test]
+    fn apply_update_setting_repository_clears_atoms() {
+        let mut job = minimal_job("switch-to-url");
+        // start with atoms set
+        assert!(job.datastore.is_some());
+        let mut up = updater();
+        up.repository = Some("user@pbs@host:store".to_string());
+        job.apply_update(up, &[]);
+        assert_eq!(job.repository.as_deref(), Some("user@pbs@host:store"));
+        assert!(job.server.is_none());
+        assert!(job.port.is_none());
+        assert!(job.datastore.is_none());
+        assert!(job.auth_id.is_none());
+        // and the resulting job validates
+        job.validate().unwrap();
+    }
+
+    #[test]
+    fn apply_update_setting_atom_clears_repository() {
+        let mut job = minimal_job("switch-to-atoms");
+        job.datastore = None;
+        job.repository = Some("user@pbs@host:store".to_string());
+        let mut up = updater();
+        up.server = Some("new.example.com".to_string());
+        up.datastore = Some("new-store".to_string());
+        job.apply_update(up, &[]);
+        assert!(job.repository.is_none());
+        assert_eq!(job.server.as_deref(), Some("new.example.com"));
+        assert_eq!(job.datastore.as_deref(), Some("new-store"));
+        job.validate().unwrap();
+    }
+
+    #[test]
+    fn apply_update_delete_then_set_in_one_call() {
+        let mut job = minimal_job("combined");
+        job.comment = Some("old".to_string());
+        let mut up = updater();
+        up.comment = Some("new".to_string());
+        // deletions apply first, so the new value wins
+        job.apply_update(up, &[DeletableProperty::Comment]);
+        assert_eq!(job.comment.as_deref(), Some("new"));
+    }
+
+    /// One `DeletableProperty` plus a predicate that confirms the field it names is cleared.
+    type DeleteCase = (DeletableProperty, fn(&ClientBackupJobConfig) -> bool);
+
+    /// Every `DeletableProperty` must clear the field it names.
+    #[test]
+    fn delete_each_property_clears_field() {
+        let auth_id: Authid = "u@pam".parse().unwrap();
+        let mut job = minimal_job("all-set");
+        job.repository = Some("host:store".to_string());
+        job.server = Some("h".to_string());
+        job.port = Some(8007);
+        job.datastore = Some("s".to_string());
+        job.auth_id = Some(auth_id);
+        job.ns = Some(BackupNamespace::root());
+        job.password_file = Some("pf".into());
+        job.password_cmd = Some("pc".into());
+        job.encryption_password_file = Some("epf".into());
+        job.encryption_password_cmd = Some("epc".into());
+        job.fingerprint = Some("fp".into());
+        job.backup_id = Some("bi".into());
+        job.backup_type = Some("bt".into());
+        job.keyfile = Some("kf".into());
+        job.master_pubkey_file = Some("mpk".into());
+        job.crypt_mode = Some(CryptMode::Encrypt);
+        job.exclude = Some(vec!["x".into()]);
+        job.include_dev = Some(vec!["i".into()]);
+        job.all_file_systems = Some(true);
+        job.skip_lost_and_found = Some(true);
+        job.change_detection_mode = Some(BackupDetectionMode::Metadata);
+        job.chunk_size = Some(1024);
+        job.rate = Some(HumanByte::from(1024_usize));
+        job.burst = Some(HumanByte::from(2048_usize));
+        job.entries_max = Some(10);
+        job.skip_e2big_xattr = Some(true);
+        job.no_cache = Some(true);
+        job.disable = Some(true);
+        job.comment = Some("c".into());
+        job.pre_run_hook = Some("h".into());
+
+        let cases: &[DeleteCase] = &[
+            (DeletableProperty::Repository, |j| j.repository.is_none()),
+            (DeletableProperty::Server, |j| j.server.is_none()),
+            (DeletableProperty::Port, |j| j.port.is_none()),
+            (DeletableProperty::Datastore, |j| j.datastore.is_none()),
+            (DeletableProperty::AuthId, |j| j.auth_id.is_none()),
+            (DeletableProperty::Ns, |j| j.ns.is_none()),
+            (DeletableProperty::PasswordFile, |j| {
+                j.password_file.is_none()
+            }),
+            (DeletableProperty::PasswordCmd, |j| j.password_cmd.is_none()),
+            (DeletableProperty::EncryptionPasswordFile, |j| {
+                j.encryption_password_file.is_none()
+            }),
+            (DeletableProperty::EncryptionPasswordCmd, |j| {
+                j.encryption_password_cmd.is_none()
+            }),
+            (DeletableProperty::Fingerprint, |j| j.fingerprint.is_none()),
+            (DeletableProperty::BackupId, |j| j.backup_id.is_none()),
+            (DeletableProperty::BackupType, |j| j.backup_type.is_none()),
+            (DeletableProperty::Keyfile, |j| j.keyfile.is_none()),
+            (DeletableProperty::MasterPubkeyFile, |j| {
+                j.master_pubkey_file.is_none()
+            }),
+            (DeletableProperty::CryptMode, |j| j.crypt_mode.is_none()),
+            (DeletableProperty::Exclude, |j| j.exclude.is_none()),
+            (DeletableProperty::IncludeDev, |j| j.include_dev.is_none()),
+            (DeletableProperty::AllFileSystems, |j| {
+                j.all_file_systems.is_none()
+            }),
+            (DeletableProperty::SkipLostAndFound, |j| {
+                j.skip_lost_and_found.is_none()
+            }),
+            (DeletableProperty::ChangeDetectionMode, |j| {
+                j.change_detection_mode.is_none()
+            }),
+            (DeletableProperty::ChunkSize, |j| j.chunk_size.is_none()),
+            (DeletableProperty::Rate, |j| j.rate.is_none()),
+            (DeletableProperty::Burst, |j| j.burst.is_none()),
+            (DeletableProperty::EntriesMax, |j| j.entries_max.is_none()),
+            (DeletableProperty::SkipE2bigXattr, |j| {
+                j.skip_e2big_xattr.is_none()
+            }),
+            (DeletableProperty::NoCache, |j| j.no_cache.is_none()),
+            (DeletableProperty::Disable, |j| j.disable.is_none()),
+            (DeletableProperty::Comment, |j| j.comment.is_none()),
+            (DeletableProperty::PreRunHook, |j| j.pre_run_hook.is_none()),
+        ];
+
+        for (prop, check) in cases {
+            let mut copy = job.clone();
+            copy.apply_update(updater(), std::slice::from_ref(prop));
+            assert!(
+                check(&copy),
+                "DeletableProperty::{prop:?} did not clear its field"
+            );
+        }
+    }
+
+    /// Hand-editing mistakes must fail the parse, rather than quietly drop a job or an option.
+    #[test]
+    fn hand_editing_mistakes_fail_the_parse() {
+        let job = "\
+backup-job: laptop
+\tbackupspec home.pxar:/home
+\tdatastore store1
+\tschedule daily
+";
+        CONFIG.parse("<test>", job).expect("the job itself is fine");
+        for broken in [
+            format!("{job}\n\texclude /home/me/secrets\n"),
+            format!("{job}\nkeyfile: /etc/pbs/key.json\n"),
+            job.replace("backup-job:", "backup-jbo:"),
+        ] {
+            assert!(
+                CONFIG.parse("<test>", &broken).is_err(),
+                "parsed:\n{broken}"
+            );
+        }
+    }
+
+    /// The config format trims a property, so a value that needs its whitespace cannot be stored.
+    #[test]
+    fn validate_rejects_values_the_format_cannot_store() {
+        let base = minimal_job("roundtrip");
+
+        let mut job = base.clone();
+        job.backupspec = vec!["files.pxar:/srv/data ".to_string()];
+        let err = job
+            .validate()
+            .expect_err("a trailing space changes the path");
+        assert!(err.to_string().contains("whitespace"), "got: {err}");
+
+        let mut job = base.clone();
+        job.keyfile = Some(" /etc/pbs/key.json".to_string());
+        assert!(job.validate().is_err(), "a leading space changes the path");
+
+        let mut job = base.clone();
+        job.comment = Some("one\ntwo".to_string());
+        let err = job
+            .validate()
+            .expect_err("a line break would read back as a property");
+        assert!(err.to_string().contains("line break"), "got: {err}");
+
+        let mut job = base;
+        job.password_cmd = Some("/usr/bin/pass show my/secret".to_string());
+        job.validate()
+            .expect("whitespace inside a value survives the round trip");
+    }
+
+    /// A job that the backup itself always refuses must not be storable.
+    #[test]
+    fn validate_rejects_a_key_together_with_crypt_mode_none() {
+        let mut job = minimal_job("cryptmode");
+        job.crypt_mode = Some(CryptMode::None);
+
+        job.keyfile = Some("/etc/pbs/key.json".to_string());
+        let err = job
+            .validate()
+            .expect_err("crypt-mode none cannot use a key file");
+        assert!(err.to_string().contains("crypt-mode 'none'"), "got: {err}");
+
+        job.keyfile = None;
+        job.master_pubkey_file = Some("/etc/pbs/master.pem".to_string());
+        assert!(
+            job.validate().is_err(),
+            "crypt-mode none cannot use a master public key"
+        );
+
+        job.master_pubkey_file = None;
+        job.validate().expect("crypt-mode none on its own is fine");
+    }
+
+    /// The next run shown for a job must match what the scheduler does.
+    #[test]
+    fn next_run_follows_the_schedule() {
+        let now = proxmox_time::epoch_i64();
+        let mut job = minimal_job("next");
+
+        // counted from the last run
+        let next = next_run(&job, now).expect("daily always has a next event");
+        assert!(next > now && next <= now + 86400, "{next} vs {now}");
+
+        // a job that never ran is due now, not in 1970
+        let next = next_run(&job, 0).unwrap();
+        assert!(next >= now, "{next} vs {now}");
+
+        // nothing for a schedule that has run out
+        job.schedule = "2020-01-01 00:00".to_string();
+        assert_eq!(next_run(&job, now), None);
+
+        // nor for a disabled job
+        job.schedule = "daily".to_string();
+        job.disable = Some(true);
+        assert_eq!(next_run(&job, now), None);
+    }
+
+    // -- ClientJobState serde round-trip tests -------------------------------
+
+    #[test]
+    fn job_state_created_round_trip() {
+        let state = ClientJobState::Created { time: 1700000000 };
+        let json = serde_json::to_string(&state).unwrap();
+        let parsed: ClientJobState = serde_json::from_str(&json).unwrap();
+        match parsed {
+            ClientJobState::Created { time } => assert_eq!(time, 1700000000),
+            other => panic!(
+                "expected Created, got {}",
+                serde_json::to_string(&other).unwrap()
+            ),
+        }
+    }
+
+    #[test]
+    fn job_state_running_round_trip() {
+        let state = ClientJobState::Running {
+            pid: 12345,
+            start_time: 1700001000,
+        };
+        let json = serde_json::to_string(&state).unwrap();
+        let parsed: ClientJobState = serde_json::from_str(&json).unwrap();
+        match parsed {
+            ClientJobState::Running { pid, start_time } => {
+                assert_eq!(pid, 12345);
+                assert_eq!(start_time, 1700001000);
+            }
+            other => panic!(
+                "expected Running, got {}",
+                serde_json::to_string(&other).unwrap()
+            ),
+        }
+    }
+
+    #[test]
+    fn job_state_finished_round_trip() {
+        let state = ClientJobState::Finished {
+            start_time: 1700001000,
+            end_time: 1700002000,
+            success: true,
+            message: Some("all good".to_string()),
+        };
+        let json = serde_json::to_string(&state).unwrap();
+        let parsed: ClientJobState = serde_json::from_str(&json).unwrap();
+        match parsed {
+            ClientJobState::Finished {
+                start_time,
+                end_time,
+                success,
+                message,
+            } => {
+                assert_eq!(start_time, 1700001000);
+                assert_eq!(end_time, 1700002000);
+                assert!(success);
+                assert_eq!(message.as_deref(), Some("all good"));
+            }
+            other => panic!(
+                "expected Finished, got {}",
+                serde_json::to_string(&other).unwrap()
+            ),
+        }
+    }
+
+    #[test]
+    fn job_state_finished_without_message_round_trip() {
+        let state = ClientJobState::Finished {
+            start_time: 1700001000,
+            end_time: 1700002000,
+            success: false,
+            message: None,
+        };
+        let json = serde_json::to_string(&state).unwrap();
+        assert!(!json.contains("message"));
+        let parsed: ClientJobState = serde_json::from_str(&json).unwrap();
+        match parsed {
+            ClientJobState::Finished {
+                success, message, ..
+            } => {
+                assert!(!success);
+                assert!(message.is_none());
+            }
+            other => panic!(
+                "expected Finished, got {}",
+                serde_json::to_string(&other).unwrap()
+            ),
+        }
+    }
+}
diff --git a/pbs-client/src/lib.rs b/pbs-client/src/lib.rs
index 139f4f4b5..453b0be00 100644
--- a/pbs-client/src/lib.rs
+++ b/pbs-client/src/lib.rs
@@ -4,6 +4,7 @@
 //! server using https.
 
 pub mod catalog_shell;
+pub mod client_job;
 pub mod pxar;
 pub mod tools;
 
-- 
2.47.3