* [PATCH 2/7] client: job: add backup job config and state types
2026-10-14 10:51 [PATCH 0/7] client: add scheduled backup jobs Thomas Lamprecht
2026-10-14 10:51 ` [PATCH 1/7] client: backup specification: widen BackupDetectionMode derives Thomas Lamprecht
@ 2026-10-14 10:51 ` Thomas Lamprecht
2026-10-14 10:51 ` [PATCH 3/7] client: job: add job runner and scheduled backup daemon Thomas Lamprecht
` (5 subsequent siblings)
7 siblings, 0 replies; 9+ messages in thread
From: Thomas Lamprecht @ 2026-10-14 10:51 UTC (permalink / raw)
To: pbs-devel
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(©),
+ "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
^ permalink raw reply related [flat|nested] 9+ messages in thread* [PATCH 3/7] client: job: add job runner and scheduled backup daemon
2026-10-14 10:51 [PATCH 0/7] client: add scheduled backup jobs Thomas Lamprecht
2026-10-14 10:51 ` [PATCH 1/7] client: backup specification: widen BackupDetectionMode derives Thomas Lamprecht
2026-10-14 10:51 ` [PATCH 2/7] client: job: add backup job config and state types Thomas Lamprecht
@ 2026-10-14 10:51 ` Thomas Lamprecht
2026-10-14 10:51 ` [PATCH 4/7] client: job: add job management CLI subcommands Thomas Lamprecht
` (4 subsequent siblings)
7 siblings, 0 replies; 9+ messages in thread
From: Thomas Lamprecht @ 2026-10-14 10:51 UTC (permalink / raw)
To: pbs-devel
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
^ permalink raw reply related [flat|nested] 9+ messages in thread