api: admin/config: introduce sync direction as job config parameter

Add the sync direction for the sync job as optional config parameter
and refrain from using the config section type for conditional
direction check, as they are now the same (see previous commit).

Use the configured sync job parameter instead of passing it to the
various methods as function parameter and only filter based on sync
direction if an optional api parameter to distingush/filter based on
direction is given.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
Reviewed-by: Dominik Csapak <d.csapak@proxmox.com>
Tested-by: Dominik Csapak <d.csapak@proxmox.com>

pbs-api-types/src/jobs.rs

@@ -597,7 +597,11 @@ pub const RESYNC_CORRUPT_SCHEMA: Schema =
         "resync-corrupt": {
             schema: RESYNC_CORRUPT_SCHEMA,
             optional: true,
-        }
+        },
+        "sync-direction": {
+            type: SyncDirection,
+            optional: true,
+        },
     }
 )]
 #[derive(Serialize, Deserialize, Clone, Updater, PartialEq)]
@@ -633,6 +637,8 @@ pub struct SyncJobConfig {
     pub transfer_last: Option<usize>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub resync_corrupt: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sync_direction: Option<SyncDirection>,
 }
 
 impl SyncJobConfig {

src/api2/admin/sync.rs

@@ -85,9 +85,9 @@ pub fn list_config_sync_jobs(
     let sync_direction = sync_direction.unwrap_or_default();
 
     let mut list = Vec::with_capacity(config.sections.len());
-    for (_, (sync_type, job)) in config.sections.into_iter() {
+    for (_, (_, job)) in config.sections.into_iter() {
         let job: SyncJobConfig = serde_json::from_value(job)?;
-        let direction = SyncDirection::from_config_type_str(&sync_type)?;
+        let direction = job.sync_direction.unwrap_or_default();
 
         match &store {
             Some(store) if &job.store != store => continue,
@@ -100,7 +100,7 @@ pub fn list_config_sync_jobs(
             _ => {}
         }
 
-        if !check_sync_job_read_access(&user_info, &auth_id, &job, direction) {
+        if !check_sync_job_read_access(&user_info, &auth_id, &job) {
             continue;
         }
 
@@ -144,15 +144,9 @@ pub fn run_sync_job(
     let user_info = CachedUserInfo::new()?;
 
     let (config, _digest) = sync::config()?;
-    let (config_type, config_section) = config
+    let sync_job: SyncJobConfig = config.lookup("sync", &id)?;
-        .sections
-        .get(&id)
-        .ok_or_else(|| format_err!("No sync job with id '{id}' found in config"))?;
 
-    let sync_direction = SyncDirection::from_config_type_str(config_type)?;
+    if !check_sync_job_modify_access(&user_info, &auth_id, &sync_job) {
-    let sync_job = SyncJobConfig::deserialize(config_section)?;
-
-    if !check_sync_job_modify_access(&user_info, &auth_id, &sync_job, sync_direction) {
         bail!("permission check failed, '{auth_id}' is missing access");
     }
 
@@ -160,7 +154,7 @@ pub fn run_sync_job(
 
     let to_stdout = rpcenv.env_type() == RpcEnvironmentType::CLI;
 
-    let upid_str = do_sync_job(job, sync_job, &auth_id, None, sync_direction, to_stdout)?;
+    let upid_str = do_sync_job(job, sync_job, &auth_id, None, to_stdout)?;
 
     Ok(upid_str)
 }

src/api2/config/datastore.rs

@@ -13,9 +13,8 @@ use proxmox_uuid::Uuid;
 
 use pbs_api_types::{
     Authid, DataStoreConfig, DataStoreConfigUpdater, DatastoreNotify, DatastoreTuning, KeepOptions,
-    MaintenanceMode, PruneJobConfig, PruneJobOptions, SyncDirection, DATASTORE_SCHEMA,
+    MaintenanceMode, PruneJobConfig, PruneJobOptions, DATASTORE_SCHEMA, PRIV_DATASTORE_ALLOCATE,
-    PRIV_DATASTORE_ALLOCATE, PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_MODIFY,
+    PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_MODIFY, PROXMOX_CONFIG_DIGEST_SCHEMA, UPID_SCHEMA,
-    PROXMOX_CONFIG_DIGEST_SCHEMA, UPID_SCHEMA,
 };
 use pbs_config::BackupLockGuard;
 use pbs_datastore::chunk_store::ChunkStore;
@@ -591,16 +590,9 @@ pub async fn delete_datastore(
         for job in list_verification_jobs(Some(name.clone()), Value::Null, rpcenv)? {
             delete_verification_job(job.config.id, None, rpcenv)?
         }
-        for direction in [SyncDirection::Pull, SyncDirection::Push] {
+        for job in list_config_sync_jobs(Some(name.clone()), None, Value::Null, rpcenv)? {
-            for job in list_config_sync_jobs(
-                Some(name.clone()),
-                Some(direction.into()),
-                Value::Null,
-                rpcenv,
-            )? {
             delete_sync_job(job.config.id, None, rpcenv)?
         }
-        }
         for job in list_prune_jobs(Some(name.clone()), Value::Null, rpcenv)? {
             delete_prune_job(job.config.id, None, rpcenv)?
         }

src/api2/config/notifications/mod.rs

@@ -9,7 +9,7 @@ use proxmox_schema::api;
 use proxmox_sortable_macro::sortable;
 
 use crate::api2::admin::datastore::get_datastore_list;
-use pbs_api_types::{SyncDirection, PRIV_SYS_AUDIT};
+use pbs_api_types::PRIV_SYS_AUDIT;
 
 use crate::api2::admin::prune::list_prune_jobs;
 use crate::api2::admin::sync::list_config_sync_jobs;
@@ -156,8 +156,7 @@ pub fn get_values(
         });
     }
 
-    for direction in [SyncDirection::Pull, SyncDirection::Push] {
+    let sync_jobs = list_config_sync_jobs(None, None, param.clone(), rpcenv)?;
-        let sync_jobs = list_config_sync_jobs(None, Some(direction.into()), param.clone(), rpcenv)?;
     for job in sync_jobs {
         values.push(MatchableValue {
             field: "job-id".into(),
@@ -165,7 +164,6 @@ pub fn get_values(
             comment: job.config.comment,
         });
     }
-    }
 
     let verify_jobs = list_verification_jobs(None, param.clone(), rpcenv)?;
     for job in verify_jobs {
@@ -188,7 +186,6 @@ pub fn get_values(
         "package-updates",
         "prune",
         "sync",
-        "sync-push",
         "system-mail",
         "tape-backup",
         "tape-load",

src/api2/config/sync.rs

@@ -17,12 +17,12 @@ use pbs_config::sync;
 
 use pbs_config::CachedUserInfo;
 use pbs_datastore::check_backup_owner;
+use crate::api2::admin::sync::ListSyncDirection;
 
 pub fn check_sync_job_read_access(
     user_info: &CachedUserInfo,
     auth_id: &Authid,
     job: &SyncJobConfig,
-    sync_direction: SyncDirection,
 ) -> bool {
     // check for audit access on datastore/namespace, applies for pull and push direction
     let ns_anchor_privs = user_info.lookup_privs(auth_id, &job.acl_path());
@@ -30,6 +30,7 @@ pub fn check_sync_job_read_access(
         return false;
     }
 
+    let sync_direction = job.sync_direction.unwrap_or_default();
     match sync_direction {
         SyncDirection::Pull => {
             if let Some(remote) = &job.remote {
@@ -71,8 +72,8 @@ pub fn check_sync_job_modify_access(
     user_info: &CachedUserInfo,
     auth_id: &Authid,
     job: &SyncJobConfig,
-    sync_direction: SyncDirection,
 ) -> bool {
+    let sync_direction = job.sync_direction.unwrap_or_default();
     match sync_direction {
         SyncDirection::Pull => {
             let ns_anchor_privs = user_info.lookup_privs(auth_id, &job.acl_path());
@@ -150,7 +151,7 @@ pub fn check_sync_job_modify_access(
     input: {
         properties: {
             "sync-direction": {
-                type: SyncDirection,
+                type: ListSyncDirection,
                 optional: true,
             },
         },
@@ -168,23 +169,29 @@ pub fn check_sync_job_modify_access(
 /// List all sync jobs
 pub fn list_sync_jobs(
     _param: Value,
-    sync_direction: Option<SyncDirection>,
+    sync_direction: Option<ListSyncDirection>,
     rpcenv: &mut dyn RpcEnvironment,
 ) -> Result<Vec<SyncJobConfig>, Error> {
     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
     let user_info = CachedUserInfo::new()?;
+    let sync_direction = sync_direction.unwrap_or_default();
 
     let (config, digest) = sync::config()?;
 
-    let sync_direction = sync_direction.unwrap_or_default();
+    let list: Vec<SyncJobConfig> = config.convert_to_typed_array("sync")?;
-    let list = config.convert_to_typed_array(sync_direction.as_config_type_str())?;
 
     rpcenv["digest"] = hex::encode(digest).into();
 
     let list = list
         .into_iter()
         .filter(|sync_job| {
-            check_sync_job_read_access(&user_info, &auth_id, sync_job, sync_direction)
+            let direction = sync_job.sync_direction.unwrap_or_default();
+            match &sync_direction {
+                ListSyncDirection::Pull if direction != SyncDirection::Pull => return false,
+                ListSyncDirection::Push if direction != SyncDirection::Push => return false,
+                _ => {}
+            }
+            check_sync_job_read_access(&user_info, &auth_id, sync_job)
         })
         .collect();
     Ok(list)
@@ -198,10 +205,6 @@ pub fn list_sync_jobs(
                 type: SyncJobConfig,
                 flatten: true,
             },
-            "sync-direction": {
-                type: SyncDirection,
-                optional: true,
-            },
         },
     },
     access: {
@@ -212,16 +215,14 @@ pub fn list_sync_jobs(
 /// Create a new sync job.
 pub fn create_sync_job(
     config: SyncJobConfig,
-    sync_direction: Option<SyncDirection>,
     rpcenv: &mut dyn RpcEnvironment,
 ) -> Result<(), Error> {
     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
     let user_info = CachedUserInfo::new()?;
-    let sync_direction = sync_direction.unwrap_or_default();
 
     let _lock = sync::lock_config()?;
 
-    if !check_sync_job_modify_access(&user_info, &auth_id, &config, sync_direction) {
+    if !check_sync_job_modify_access(&user_info, &auth_id, &config) {
         bail!("permission check failed");
     }
 
@@ -229,6 +230,7 @@ pub fn create_sync_job(
         bail!("source and target datastore can't be the same");
     }
 
+    let sync_direction = config.sync_direction.unwrap_or_default();
     if sync_direction == SyncDirection::Push && config.resync_corrupt.is_some() {
         bail!("push jobs do not support resync-corrupt option");
     }
@@ -248,7 +250,7 @@ pub fn create_sync_job(
         param_bail!("id", "job '{}' already exists.", config.id);
     }
 
-    section_config.set_data(&config.id, sync_direction.as_config_type_str(), &config)?;
+    section_config.set_data(&config.id, "sync", &config)?;
 
     sync::save_config(&section_config)?;
 
@@ -278,17 +280,9 @@ pub fn read_sync_job(id: String, rpcenv: &mut dyn RpcEnvironment) -> Result<Sync
 
     let (config, digest) = sync::config()?;
 
-    let (sync_job, sync_direction) =
+    let sync_job = config.lookup("sync", &id)?;
-        if let Some((config_type, config_section)) = config.sections.get(&id) {
-            (
-                SyncJobConfig::deserialize(config_section)?,
-                SyncDirection::from_config_type_str(config_type)?,
-            )
-        } else {
-            http_bail!(NOT_FOUND, "job '{id}' does not exist.")
-        };
 
-    if !check_sync_job_read_access(&user_info, &auth_id, &sync_job, sync_direction) {
+    if !check_sync_job_read_access(&user_info, &auth_id, &sync_job) {
         bail!("permission check failed");
     }
 
@@ -330,6 +324,8 @@ pub enum DeletableProperty {
     MaxDepth,
     /// Delete the transfer_last property,
     TransferLast,
+    /// Delete the sync_direction property,
+    SyncDirection,
 }
 
 #[api(
@@ -383,17 +379,11 @@ pub fn update_sync_job(
         crate::tools::detect_modified_configuration_file(&digest, &expected_digest)?;
     }
 
-    let (mut data, sync_direction) =
+    let mut data: SyncJobConfig = config.lookup("sync", &id)?;
-        if let Some((config_type, config_section)) = config.sections.get(&id) {
-            (
-                SyncJobConfig::deserialize(config_section)?,
-                SyncDirection::from_config_type_str(config_type)?,
-            )
-        } else {
-            http_bail!(NOT_FOUND, "job '{id}' does not exist.")
-        };
 
-    if sync_direction == SyncDirection::Push && update.resync_corrupt.is_some() {
+    if update.sync_direction.unwrap_or_default() == SyncDirection::Push
+        && update.resync_corrupt.is_some()
+    {
         bail!("push jobs do not support resync-corrupt option");
     }
 
@@ -442,6 +432,9 @@ pub fn update_sync_job(
                 DeletableProperty::TransferLast => {
                     data.transfer_last = None;
                 }
+                DeletableProperty::SyncDirection => {
+                    data.sync_direction = None;
+                }
             }
         }
     }
@@ -482,6 +475,9 @@ pub fn update_sync_job(
     if let Some(resync_corrupt) = update.resync_corrupt {
         data.resync_corrupt = Some(resync_corrupt);
     }
+    if let Some(sync_direction) = update.sync_direction {
+        data.sync_direction = Some(sync_direction);
+    }
 
     if update.limit.rate_in.is_some() {
         data.limit.rate_in = update.limit.rate_in;
@@ -519,11 +515,11 @@ pub fn update_sync_job(
         }
     }
 
-    if !check_sync_job_modify_access(&user_info, &auth_id, &data, sync_direction) {
+    if !check_sync_job_modify_access(&user_info, &auth_id, &data) {
         bail!("permission check failed");
     }
 
-    config.set_data(&id, sync_direction.as_config_type_str(), &data)?;
+    config.set_data(&id, "sync", &data)?;
 
     sync::save_config(&config)?;
 
@@ -570,16 +566,17 @@ pub fn delete_sync_job(
         crate::tools::detect_modified_configuration_file(&digest, &expected_digest)?;
     }
 
-    if let Some((config_type, config_section)) = config.sections.get(&id) {
+    match config.lookup("sync", &id) {
-        let sync_direction = SyncDirection::from_config_type_str(config_type)?;
+        Ok(job) => {
-        let job = SyncJobConfig::deserialize(config_section)?;
+            if !check_sync_job_modify_access(&user_info, &auth_id, &job) {
-        if !check_sync_job_modify_access(&user_info, &auth_id, &job, sync_direction) {
                 bail!("permission check failed");
             }
             config.sections.remove(&id);
-    } else {
+        }
+        Err(_) => {
             http_bail!(NOT_FOUND, "job '{}' does not exist.", id)
         }
+    }
 
     sync::save_config(&config)?;
 
@@ -647,62 +644,36 @@ acl:1:/remote/remote1/remotestore1:write@pbs:RemoteSyncOperator
         schedule: None,
         limit: pbs_api_types::RateLimitConfig::default(), // no limit
         transfer_last: None,
+        sync_direction: None, // use default
     };
 
     // should work without ACLs
-    assert!(check_sync_job_read_access(
+    assert!(check_sync_job_read_access(&user_info, root_auth_id, &job));
-        &user_info,
+    assert!(check_sync_job_modify_access(&user_info, root_auth_id, &job,));
-        root_auth_id,
-        &job,
-        SyncDirection::Pull,
-    ));
-    assert!(check_sync_job_modify_access(
-        &user_info,
-        root_auth_id,
-        &job,
-        SyncDirection::Pull,
-    ));
 
     // user without permissions must fail
     assert!(!check_sync_job_read_access(
         &user_info,
         &no_perm_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
     assert!(!check_sync_job_modify_access(
         &user_info,
         &no_perm_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // reading without proper read permissions on either remote or local must fail
-    assert!(!check_sync_job_read_access(
+    assert!(!check_sync_job_read_access(&user_info, &read_auth_id, &job,));
-        &user_info,
-        &read_auth_id,
-        &job,
-        SyncDirection::Pull,
-    ));
 
     // reading without proper read permissions on local end must fail
     job.remote = Some("remote1".to_string());
-    assert!(!check_sync_job_read_access(
+    assert!(!check_sync_job_read_access(&user_info, &read_auth_id, &job,));
-        &user_info,
-        &read_auth_id,
-        &job,
-        SyncDirection::Pull,
-    ));
 
     // reading without proper read permissions on remote end must fail
     job.remote = Some("remote0".to_string());
     job.store = "localstore1".to_string();
-    assert!(!check_sync_job_read_access(
+    assert!(!check_sync_job_read_access(&user_info, &read_auth_id, &job,));
-        &user_info,
-        &read_auth_id,
-        &job,
-        SyncDirection::Pull,
-    ));
 
     // writing without proper write permissions on either end must fail
     job.store = "localstore0".to_string();
@@ -710,7 +681,6 @@ acl:1:/remote/remote1/remotestore1:write@pbs:RemoteSyncOperator
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // writing without proper write permissions on local end must fail
@@ -723,53 +693,38 @@ acl:1:/remote/remote1/remotestore1:write@pbs:RemoteSyncOperator
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // reset remote to one where users have access
     job.remote = Some("remote1".to_string());
 
     // user with read permission can only read, but not modify/run
-    assert!(check_sync_job_read_access(
+    assert!(check_sync_job_read_access(&user_info, &read_auth_id, &job,));
-        &user_info,
-        &read_auth_id,
-        &job,
-        SyncDirection::Pull,
-    ));
     job.owner = Some(read_auth_id.clone());
     assert!(!check_sync_job_modify_access(
         &user_info,
         &read_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
     job.owner = None;
     assert!(!check_sync_job_modify_access(
         &user_info,
         &read_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
     job.owner = Some(write_auth_id.clone());
     assert!(!check_sync_job_modify_access(
         &user_info,
         &read_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // user with simple write permission can modify/run
-    assert!(check_sync_job_read_access(
+    assert!(check_sync_job_read_access(&user_info, &write_auth_id, &job,));
-        &user_info,
-        &write_auth_id,
-        &job,
-        SyncDirection::Pull,
-    ));
     assert!(check_sync_job_modify_access(
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // but can't modify/run with deletion
@@ -778,7 +733,6 @@ acl:1:/remote/remote1/remotestore1:write@pbs:RemoteSyncOperator
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // unless they have Datastore.Prune as well
@@ -787,7 +741,6 @@ acl:1:/remote/remote1/remotestore1:write@pbs:RemoteSyncOperator
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // changing owner is not possible
@@ -796,7 +749,6 @@ acl:1:/remote/remote1/remotestore1:write@pbs:RemoteSyncOperator
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // also not to the default 'root@pam'
@@ -805,7 +757,6 @@ acl:1:/remote/remote1/remotestore1:write@pbs:RemoteSyncOperator
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     // unless they have Datastore.Modify as well
@@ -815,14 +766,12 @@ acl:1:/remote/remote1/remotestore1:write@pbs:RemoteSyncOperator
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
     job.owner = None;
     assert!(check_sync_job_modify_access(
         &user_info,
         &write_auth_id,
         &job,
-        SyncDirection::Pull,
     ));
 
     Ok(())

src/bin/proxmox-backup-proxy.rs

@@ -40,8 +40,8 @@ use pbs_buildcfg::configdir;
 use proxmox_time::CalendarEvent;
 
 use pbs_api_types::{
-    Authid, DataStoreConfig, Operation, PruneJobConfig, SyncDirection, SyncJobConfig,
+    Authid, DataStoreConfig, Operation, PruneJobConfig, SyncJobConfig, TapeBackupJobConfig,
-    TapeBackupJobConfig, VerificationJobConfig,
+    VerificationJobConfig,
 };
 
 use proxmox_backup::auth_helpers::*;
@@ -595,14 +595,7 @@ async fn schedule_datastore_sync_jobs() {
         Ok((config, _digest)) => config,
     };
 
-    for (job_id, (job_type, job_config)) in config.sections {
+    for (job_id, (_, job_config)) in config.sections {
-        let sync_direction = match SyncDirection::from_config_type_str(&job_type) {
-            Ok(direction) => direction,
-            Err(err) => {
-                eprintln!("unexpected config type in sync job config - {err}");
-                continue;
-            }
-        };
         let job_config: SyncJobConfig = match serde_json::from_value(job_config) {
             Ok(c) => c,
             Err(err) => {
@@ -624,14 +617,7 @@ async fn schedule_datastore_sync_jobs() {
             };
 
             let auth_id = Authid::root_auth_id().clone();
-            if let Err(err) = do_sync_job(
+            if let Err(err) = do_sync_job(job, job_config, &auth_id, Some(event_str), false) {
-                job,
-                job_config,
-                &auth_id,
-                Some(event_str),
-                sync_direction,
-                false,
-            ) {
                 eprintln!("unable to start datastore sync job {job_id} - {err}");
             }
         };

src/bin/proxmox_backup_manager/sync.rs

@@ -4,9 +4,10 @@ use serde_json::Value;
 use proxmox_router::{cli::*, ApiHandler, RpcEnvironment};
 use proxmox_schema::api;
 
-use pbs_api_types::{SyncDirection, JOB_ID_SCHEMA};
+use pbs_api_types::JOB_ID_SCHEMA;
 
 use proxmox_backup::api2;
+use crate::api2::admin::sync::ListSyncDirection;
 
 fn render_group_filter(value: &Value, _record: &Value) -> Result<String, Error> {
     if let Some(group_filters) = value.as_array() {
@@ -21,7 +22,7 @@ fn render_group_filter(value: &Value, _record: &Value) -> Result<String, Error>
     input: {
         properties: {
             "sync-direction": {
-                type: SyncDirection,
+                type: ListSyncDirection,
                 optional: true,
             },
             "output-format": {

src/server/sync.rs

@@ -597,7 +597,6 @@ pub fn do_sync_job(
     sync_job: SyncJobConfig,
     auth_id: &Authid,
     schedule: Option<String>,
-    sync_direction: SyncDirection,
     to_stdout: bool,
 ) -> Result<String, Error> {
     let job_id = format!(
@@ -609,6 +608,7 @@ pub fn do_sync_job(
         job.jobname(),
     );
     let worker_type = job.jobtype().to_string();
+    let sync_direction = sync_job.sync_direction.unwrap_or_default();
 
     if sync_job.remote.is_none() && sync_job.store == sync_job.remote_store {
         bail!("can't sync to same datastore");