[PATCH datacenter-manager v3 14/21] server: migrate existing ParallelFetcher users to use PdmApplication

[Lukas Wagner <l.wagner@proxmox.com>]

ParallelFetcher now requires a PdmApplication in both it's `new` and
`builder` constructions. The provided application context object is
later provided via ParallelFetcherArgs<C> to the closure.

For now, this is mostly used to give access to the client factory.

The mandatory argument should gently nudge the developer towards using
the #[state] injected handle, with the option to use context::pdm_application()
as fallback, if too much refactoring is needed on the spot.

Introducing this new mandatory parameter led to a small cascade of
changes in existing users of ParallelFetcher. These were changed to
fully use `app` everywhere they can, such as when reading remote config.
This makes the commit slightly bigger, but avoids any awkward
'in-between' states where only part of a module uses PdmApplication.

Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---

Notes:
    Changes since v2:
      - Use new #[state] attribute
      - Let ParallelFetcher have PdmApplication, not only
        ClientFactory
      - Merged two previous commits into this one:
    
        https://lore.proxmox.com/pdm-devel/20260820145220.418032-14-l.wagner@proxmox.com/T/#u
        https://lore.proxmox.com/pdm-devel/DKX7UQ5AWO68.3SIX8MRGM180I@proxmox.com/T/#mb0e73f6906c23766b6075751f2056dc86beda8fc

 server/src/api/pve/firewall.rs          | 25 +++++++++-----
 server/src/api/sdn/controllers.rs       | 13 +++++---
 server/src/api/sdn/vnets.rs             | 14 +++++---
 server/src/api/sdn/zones.rs             | 13 +++++---
 server/src/parallel_fetcher.rs          | 44 ++++++++++++++++++++-----
 server/src/remote_tasks/refresh_task.rs |  4 +--
 server/src/remote_updates.rs            |  4 +--
 7 files changed, 83 insertions(+), 34 deletions(-)

diff --git a/server/src/api/pve/firewall.rs b/server/src/api/pve/firewall.rs
index d0f2d6d1..29fb3dce 100644
--- a/server/src/api/pve/firewall.rs
+++ b/server/src/api/pve/firewall.rs
@@ -19,6 +19,7 @@ use pdm_api_types::{PRIV_RESOURCE_AUDIT, PRIV_RESOURCE_MODIFY, PRIV_SYS_MODIFY};
 use super::{connect_to_remote_by_id, find_node_for_vm};
 use crate::api::remotes::RemoteIterator;
 use crate::connection::PveClient;
+use crate::context::PdmApplication;
 use crate::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs};
 
 // top-level firewall routers
@@ -114,7 +115,10 @@ struct ClusterFirewallData {
 async fn fetch_cluster_firewall_data(
     args: ParallelFetcherArgs<()>,
 ) -> Result<ClusterFirewallData, Error> {
-    let pve = crate::connection::make_pve_client(args.remote())?;
+    let pve = args
+        .pdm_application()
+        .client_factory()
+        .make_pve_client(args.remote())?;
 
     let guests = match pve.cluster_resources(Some(ClusterResourceKind::Vm)).await {
         Ok(guests) => guests,
@@ -204,7 +208,10 @@ async fn load_guests_firewall_status(
 async fn fetch_node_firewall_status(
     args: ParallelFetcherArgs<FirewallFetchContext>,
 ) -> Result<NodeFirewallStatus, Error> {
-    let pve = crate::connection::make_pve_client(args.remote())?;
+    let pve = args
+        .pdm_application()
+        .client_factory()
+        .make_pve_client(args.remote())?;
 
     let options_response = pve.node_firewall_options(args.node());
     let rules_response = pve.list_node_firewall_rules(args.node());
@@ -246,8 +253,9 @@ async fn fetch_node_firewall_status(
 /// Get firewall status of all PVE remotes.
 pub async fn pve_firewall_status(
     _rpcenv: &mut dyn RpcEnvironment,
+    #[state] app: PdmApplication,
 ) -> Result<Vec<RemoteFirewallStatus>, Error> {
-    let pve_remotes: Vec<Remote> = RemoteIterator::new(pdm_config::remotes::instance())?
+    let pve_remotes: Vec<Remote> = RemoteIterator::new(app.remote_config())?
         .remote_type(pdm_api_types::remotes::RemoteType::Pve)
         .into_remotes()
         .collect();
@@ -257,7 +265,7 @@ pub async fn pve_firewall_status(
     }
 
     // 1: fetch cluster-level data (status + guests)
-    let cluster_fetcher = ParallelFetcher::new(());
+    let cluster_fetcher = ParallelFetcher::new(app.clone(), ());
     let cluster_results = cluster_fetcher
         .do_for_all_remotes(pve_remotes.iter().cloned(), fetch_cluster_firewall_data)
         .await;
@@ -275,7 +283,7 @@ pub async fn pve_firewall_status(
         guests: Arc::new(vec![]),
     };
 
-    let node_fetcher = ParallelFetcher::new(context);
+    let node_fetcher = ParallelFetcher::new(app, context);
     let node_results = node_fetcher
         .do_for_all_remote_nodes(pve_remotes.iter().cloned(), move |mut args| {
             if let Some(guests) = guests_per_remote.get(&args.remote().id) {
@@ -352,8 +360,9 @@ pub async fn cluster_firewall_options(
 pub async fn cluster_firewall_status(
     remote: String,
     _rpcenv: &mut dyn RpcEnvironment,
+    #[state] app: PdmApplication,
 ) -> Result<RemoteFirewallStatus, Error> {
-    let (remote_config, _) = pdm_config::remotes::config()?;
+    let (remote_config, _) = app.remote_config().read()?;
 
     let remote_obj = remote_config
         .into_iter()
@@ -362,7 +371,7 @@ pub async fn cluster_firewall_status(
         .ok_or_else(|| anyhow::format_err!("Remote '{}' not found", remote))?;
 
     // 1: fetch cluster-level data (status + guests)
-    let cluster_fetcher = ParallelFetcher::new(());
+    let cluster_fetcher = ParallelFetcher::new(app.clone(), ());
     let cluster_results = cluster_fetcher
         .do_for_all_remotes(
             std::iter::once(remote_obj.clone()),
@@ -390,7 +399,7 @@ pub async fn cluster_firewall_status(
         guests: Arc::new(guests),
     };
 
-    let node_fetcher = ParallelFetcher::new(context);
+    let node_fetcher = ParallelFetcher::new(app, context);
     let node_results = node_fetcher
         .do_for_all_remote_nodes(std::iter::once(remote_obj), fetch_node_firewall_status)
         .await;
diff --git a/server/src/api/sdn/controllers.rs b/server/src/api/sdn/controllers.rs
index 41108674..5400c189 100644
--- a/server/src/api/sdn/controllers.rs
+++ b/server/src/api/sdn/controllers.rs
@@ -9,8 +9,8 @@ use proxmox_router::{Permission, Router, RpcEnvironment, http_bail};
 use proxmox_schema::api;
 use pve_api_types::ListControllersType;
 
-use crate::api::pve;
 use crate::api::remotes::RemoteIterator;
+use crate::context::PdmApplication;
 use crate::parallel_fetcher::ParallelFetcher;
 
 pub const ROUTER: Router = Router::new().get(&API_METHOD_LIST_CONTROLLERS);
@@ -63,6 +63,7 @@ pub async fn list_controllers(
     ty: Option<ListControllersType>,
     remotes: Option<HashSet<String>>,
     rpcenv: &mut dyn RpcEnvironment,
+    #[state] app: PdmApplication,
 ) -> Result<Vec<ListController>, Error> {
     let user_info = CachedUserInfo::new()?;
 
@@ -75,7 +76,7 @@ pub async fn list_controllers(
         http_bail!(FORBIDDEN, "user has no access to resources");
     }
 
-    let mut iter = RemoteIterator::new(pdm_config::remotes::instance())?
+    let mut iter = RemoteIterator::new(app.remote_config())?
         .remote_type(RemoteType::Pve)
         .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT);
     if let Some(ref filter) = remotes {
@@ -83,11 +84,15 @@ pub async fn list_controllers(
     }
 
     let mut vnets = Vec::new();
-    let fetcher = ParallelFetcher::new((pending, running, ty));
+
+    let fetcher = ParallelFetcher::new(app, (pending, running, ty));
 
     let results = fetcher
         .do_for_all_remotes(iter.into_remotes(), async |args| {
-            Ok(pve::connect(args.remote())?
+            Ok(args
+                .pdm_application()
+                .client_factory()
+                .make_pve_client(args.remote())?
                 .list_controllers(args.context().0, args.context().1, args.context().2)
                 .await?)
         })
diff --git a/server/src/api/sdn/vnets.rs b/server/src/api/sdn/vnets.rs
index 89017438..312d46b6 100644
--- a/server/src/api/sdn/vnets.rs
+++ b/server/src/api/sdn/vnets.rs
@@ -1,6 +1,7 @@
 use std::collections::HashSet;
 
 use anyhow::{Context, Error};
+
 use pbs_api_types::REMOTE_ID_SCHEMA;
 use pdm_api_types::{
     Authid, PRIV_RESOURCE_AUDIT,
@@ -13,8 +14,7 @@ use proxmox_router::{Permission, Router, RpcEnvironment, http_bail};
 use proxmox_schema::api;
 use pve_api_types::{CreateVnet, SdnVnetType};
 
-use crate::api::pve;
-use crate::api::remotes::RemoteIterator;
+use crate::{api::remotes::RemoteIterator, context::PdmApplication};
 use crate::{parallel_fetcher::ParallelFetcher, sdn_client::LockedSdnClients};
 
 pub const ROUTER: Router = Router::new()
@@ -64,6 +64,7 @@ async fn list_vnets(
     running: Option<bool>,
     remotes: Option<HashSet<String>>,
     rpcenv: &mut dyn RpcEnvironment,
+    #[state] app: PdmApplication,
 ) -> Result<Vec<ListVnet>, Error> {
     let user_info = CachedUserInfo::new()?;
 
@@ -76,7 +77,7 @@ async fn list_vnets(
         http_bail!(FORBIDDEN, "user has no access to resources");
     }
 
-    let mut iter = RemoteIterator::new(pdm_config::remotes::instance())?
+    let mut iter = RemoteIterator::new(app.remote_config())?
         .remote_type(RemoteType::Pve)
         .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT);
     if let Some(ref filter) = remotes {
@@ -84,11 +85,14 @@ async fn list_vnets(
     }
 
     let mut vnets = Vec::new();
-    let fetcher = ParallelFetcher::new((pending, running));
+    let fetcher = ParallelFetcher::new(app, (pending, running));
 
     let results = fetcher
         .do_for_all_remotes(iter.into_remotes(), async |args| {
-            Ok(pve::connect(args.remote())?
+            Ok(args
+                .pdm_application()
+                .client_factory()
+                .make_pve_client(args.remote())?
                 .list_vnets(args.context().0, args.context().1)
                 .await?)
         })
diff --git a/server/src/api/sdn/zones.rs b/server/src/api/sdn/zones.rs
index 5d8d8add..82366e81 100644
--- a/server/src/api/sdn/zones.rs
+++ b/server/src/api/sdn/zones.rs
@@ -14,8 +14,7 @@ use proxmox_router::{Permission, Router, RpcEnvironment, http_bail};
 use proxmox_schema::api;
 use pve_api_types::{CreateZone, ListZonesType};
 
-use crate::api::pve;
-use crate::api::remotes::RemoteIterator;
+use crate::{api::remotes::RemoteIterator, context::PdmApplication};
 use crate::{parallel_fetcher::ParallelFetcher, sdn_client::LockedSdnClients};
 
 pub const ROUTER: Router = Router::new()
@@ -70,6 +69,7 @@ pub async fn list_zones(
     ty: Option<ListZonesType>,
     remotes: Option<HashSet<String>>,
     rpcenv: &mut dyn RpcEnvironment,
+    #[state] app: PdmApplication,
 ) -> Result<Vec<ListZone>, Error> {
     let user_info = CachedUserInfo::new()?;
 
@@ -82,7 +82,7 @@ pub async fn list_zones(
         http_bail!(FORBIDDEN, "user has no access to resources");
     }
 
-    let mut iter = RemoteIterator::new(pdm_config::remotes::instance())?
+    let mut iter = RemoteIterator::new(app.remote_config())?
         .remote_type(RemoteType::Pve)
         .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT);
     if let Some(ref filter) = remotes {
@@ -90,11 +90,14 @@ pub async fn list_zones(
     }
 
     let mut vnets = Vec::new();
-    let fetcher = ParallelFetcher::new((pending, running, ty));
+    let fetcher = ParallelFetcher::new(app, (pending, running, ty));
 
     let results = fetcher
         .do_for_all_remotes(iter.into_remotes(), async |args| {
-            Ok(pve::connect(args.remote())?
+            Ok(args
+                .pdm_application()
+                .client_factory()
+                .make_pve_client(args.remote())?
                 .list_zones(args.context().0, args.context().1, args.context().2)
                 .await?)
         })
diff --git a/server/src/parallel_fetcher.rs b/server/src/parallel_fetcher.rs
index 512a04d9..b90f02e3 100644
--- a/server/src/parallel_fetcher.rs
+++ b/server/src/parallel_fetcher.rs
@@ -8,6 +8,7 @@
 //! #
 //! # #[tokio::main]
 //! # async fn main() -> Result<(), Error> {
+//! #   let app = server::context::pdm_application();
 //! #   let remotes: Vec<Remote> = Vec::new();
 //! #
 //!     async fn fetch_meaning(
@@ -25,7 +26,7 @@
 //!     // This context can be passed to the function what is executed for every remote node.
 //!     let context = ();
 //!
-//!     let fetcher = ParallelFetcher::builder(context)
+//!     let fetcher = ParallelFetcher::builder(app, context)
 //!         .max_connections(10)
 //!         .max_connections_per_remote(2)
 //!         .build();
@@ -73,7 +74,7 @@ use pve_api_types::ClusterNodeIndexResponse;
 
 use pdm_api_types::remotes::{Remote, RemoteType};
 
-use crate::connection;
+use crate::context::PdmApplication;
 
 /// Maximum number of parallel outgoing API requests.
 pub const DEFAULT_MAX_CONNECTIONS: usize = 20;
@@ -229,15 +230,17 @@ impl<T> NodeResponse<T> {
 pub struct ParallelFetcherBuilder<C> {
     max_connections: Option<usize>,
     max_connections_per_remote: Option<usize>,
+    pdm_application: PdmApplication,
     context: C,
 }
 
 impl<C> ParallelFetcherBuilder<C> {
-    fn new(context: C) -> Self {
+    fn new(pdm_application: PdmApplication, context: C) -> Self {
         Self {
             context,
             max_connections: None,
             max_connections_per_remote: None,
+            pdm_application,
         }
     }
 
@@ -263,6 +266,7 @@ impl<C> ParallelFetcherBuilder<C> {
                 .max_connections_per_remote
                 .unwrap_or(DEFAULT_MAX_CONNECTIONS_PER_REMOTE),
             context: self.context,
+            pdm_application: self.pdm_application,
         }
     }
 }
@@ -280,6 +284,8 @@ pub struct ParallelFetcherArgs<C> {
     /// The node. This may be 'localhost' for PBS remotes or if using
     /// [`ParallelFetcher::do_for_all_remotes`].
     node: String,
+    /// A handle to the [`PdmApplication`] context object.
+    pdm_application: PdmApplication,
 }
 
 impl<C> ParallelFetcherArgs<C> {
@@ -305,24 +311,32 @@ impl<C> ParallelFetcherArgs<C> {
     pub fn context_mut(&mut self) -> &mut C {
         &mut self.context
     }
+
+    /// Get a reference to [`PdmApplication`].
+    ///
+    /// Use this for accessing configuration or constructing API clients.
+    pub fn pdm_application(&self) -> &PdmApplication {
+        &self.pdm_application
+    }
 }
 
 /// Helper for parallelizing API requests to multiple remotes/nodes.
 pub struct ParallelFetcher<C> {
     max_connections: usize,
     max_connections_per_remote: usize,
+    pdm_application: PdmApplication,
     context: C,
 }
 
 impl<C: Clone + Send + 'static> ParallelFetcher<C> {
     /// Create a [`ParallelFetcher`] with default settings.
-    pub fn new(context: C) -> Self {
-        Self::builder(context).build()
+    pub fn new(app: PdmApplication, context: C) -> Self {
+        Self::builder(app, context).build()
     }
 
     /// Create the builder for constructing a [`ParallelFetcher`] with custom settings.
-    pub fn builder(context: C) -> ParallelFetcherBuilder<C> {
-        ParallelFetcherBuilder::new(context)
+    pub fn builder(app: PdmApplication, context: C) -> ParallelFetcherBuilder<C> {
+        ParallelFetcherBuilder::new(app, context)
     }
 
     /// Invoke a function `func` for all nodes of a given list of remotes in parallel.
@@ -344,13 +358,16 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
         for remote in remotes {
             let semaphore = Arc::clone(&total_connections_semaphore);
 
+            let app = self.pdm_application.clone();
             let f = func.clone();
+
             let future = Self::fetch_remote(
                 remote,
                 self.context.clone(),
                 semaphore,
                 f,
                 self.max_connections_per_remote,
+                app,
             );
 
             if let Some(log_context) = LogContext::current() {
@@ -382,6 +399,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
         semaphore: Arc<Semaphore>,
         func: F,
         max_connections_per_remote: usize,
+        pdm_application: PdmApplication,
     ) -> RemoteResponse<MultipleNodesResponse<T>>
     where
         F: Fn(ParallelFetcherArgs<C>) -> Ft + Clone + Send + 'static,
@@ -397,8 +415,10 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
             RemoteType::Pve => {
                 let remote_clone = remote.clone();
 
+                let app = &pdm_application;
+
                 let nodes = match async move {
-                    let client = connection::make_pve_client(&remote_clone)?;
+                    let client = app.client_factory().make_pve_client(&remote_clone)?;
                     let nodes = client.list_nodes().await?;
 
                     Ok::<Vec<ClusterNodeIndexResponse>, Error>(nodes)
@@ -433,12 +453,14 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
                     let remote_clone = remote.clone();
                     let node_name = node.node.clone();
                     let context_clone = context.clone();
+                    let app = pdm_application.clone();
 
                     let future = Self::fetch_node(
                         func_clone,
                         context_clone,
                         remote_clone,
                         node_name,
+                        app,
                         permit,
                         Some(per_remote_connections_permit),
                     );
@@ -467,6 +489,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
                     context,
                     remote.clone(),
                     "localhost".into(),
+                    pdm_application,
                     permit.unwrap(), // Always set to `Some` at this point
                     None,
                 )
@@ -490,6 +513,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
         context: C,
         remote: Remote,
         node: String,
+        pdm_application: PdmApplication,
         _permit: OwnedSemaphorePermit,
         _per_remote_connections_permit: Option<OwnedSemaphorePermit>,
     ) -> NodeResponse<T>
@@ -504,6 +528,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
             context,
             remote,
             node: node.clone(),
+            pdm_application,
         };
 
         let result = func(parallel_fetcher_context).await;
@@ -539,7 +564,9 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
             let remote_type = remote.ty;
 
             let context = self.context.clone();
+            let app = self.pdm_application.clone();
             let func = func.clone();
+
             let future = async move {
                 let permit = total_connections_semaphore.acquire_owned().await.unwrap();
 
@@ -551,6 +578,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
                         context,
                         remote,
                         "localhost".into(),
+                        app,
                         permit,
                         None,
                     )
diff --git a/server/src/remote_tasks/refresh_task.rs b/server/src/remote_tasks/refresh_task.rs
index c1753114..7d729c22 100644
--- a/server/src/remote_tasks/refresh_task.rs
+++ b/server/src/remote_tasks/refresh_task.rs
@@ -9,7 +9,6 @@ use pdm_api_types::RemoteUpid;
 use pdm_api_types::remotes::{Remote, RemoteType};
 use proxmox_section_config::typed::SectionConfigData;
 
-use crate::api;
 use crate::connection;
 use crate::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs};
 use crate::pbs_client;
@@ -17,6 +16,7 @@ use crate::remote_tasks::{
     KEEP_OLD_FILES, ROTATE_AFTER,
     task_cache::{GetTasks, NodeFetchSuccessMap, State, TaskCache, TaskCacheItem},
 };
+use crate::{api, context};
 
 /// Interval in seconds at which to fetch the newest tasks from remotes (if there is no tracked
 /// task for this remote).
@@ -234,7 +234,7 @@ async fn fetch_remotes(
     remotes: Vec<Remote>,
     cache_state: Arc<State>,
 ) -> (Vec<TaskCacheItem>, NodeFetchSuccessMap) {
-    let fetcher = ParallelFetcher::builder(cache_state)
+    let fetcher = ParallelFetcher::builder(context::pdm_application(), cache_state)
         .max_connections(MAX_CONNECTIONS)
         .max_connections_per_remote(CONNECTIONS_PER_PVE_REMOTE)
         .build();
diff --git a/server/src/remote_updates.rs b/server/src/remote_updates.rs
index bd648efd..890bbfa9 100644
--- a/server/src/remote_updates.rs
+++ b/server/src/remote_updates.rs
@@ -12,7 +12,7 @@ use pdm_api_types::remotes::{Remote, RemoteType};
 
 use crate::namespaced_cache::CacheError;
 use crate::parallel_fetcher::ParallelFetcher;
-use crate::{api_cache, connection};
+use crate::{api_cache, connection, context};
 
 const OLD_CACHEFILE: &str = concat!(pdm_buildcfg::PDM_CACHE_DIR_M!(), "/remote-updates.json");
 
@@ -229,7 +229,7 @@ async fn update_cached_summary_for_node(
 /// delays its own entry. The final pass records whole-remote failures and prunes vanished
 /// remotes and nodes.
 pub async fn refresh_update_summary_cache(remotes: Vec<Remote>) -> Result<(), Error> {
-    let fetcher = ParallelFetcher::new(());
+    let fetcher = ParallelFetcher::new(context::pdm_application(), ());
 
     let fetch_response = fetcher
         .do_for_all_remote_nodes(remotes.into_iter(), |args| async move {
-- 
2.47.3