[PATCH datacenter-manager v3 13/21] parallel fetcher: pass arguments to closure in a single type

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

This makes it much easier to pass more data later, e.g. a client
factory.

Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
 server/src/api/pve/firewall.rs          | 31 ++++++------
 server/src/api/sdn/controllers.rs       |  6 +--
 server/src/api/sdn/vnets.rs             |  6 ++-
 server/src/api/sdn/zones.rs             |  6 ++-
 server/src/parallel_fetcher.rs          | 65 +++++++++++++++++++++----
 server/src/remote_tasks/refresh_task.rs | 23 +++++----
 server/src/remote_updates.rs            | 13 +++--
 7 files changed, 101 insertions(+), 49 deletions(-)

diff --git a/server/src/api/pve/firewall.rs b/server/src/api/pve/firewall.rs
index b381a14e..d0f2d6d1 100644
--- a/server/src/api/pve/firewall.rs
+++ b/server/src/api/pve/firewall.rs
@@ -19,7 +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::parallel_fetcher::ParallelFetcher;
+use crate::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs};
 
 // top-level firewall routers
 pub const PVE_FW_ROUTER: Router = Router::new()
@@ -112,11 +112,9 @@ struct ClusterFirewallData {
 }
 
 async fn fetch_cluster_firewall_data(
-    _context: (),
-    remote: Remote,
-    _node: String, // unused for cluster-level data
+    args: ParallelFetcherArgs<()>,
 ) -> Result<ClusterFirewallData, Error> {
-    let pve = crate::connection::make_pve_client(&remote)?;
+    let pve = crate::connection::make_pve_client(args.remote())?;
 
     let guests = match pve.cluster_resources(Some(ClusterResourceKind::Vm)).await {
         Ok(guests) => guests,
@@ -204,14 +202,12 @@ async fn load_guests_firewall_status(
 }
 
 async fn fetch_node_firewall_status(
-    context: FirewallFetchContext,
-    remote: Remote,
-    node: String,
+    args: ParallelFetcherArgs<FirewallFetchContext>,
 ) -> Result<NodeFirewallStatus, Error> {
-    let pve = crate::connection::make_pve_client(&remote)?;
+    let pve = crate::connection::make_pve_client(args.remote())?;
 
-    let options_response = pve.node_firewall_options(&node);
-    let rules_response = pve.list_node_firewall_rules(&node);
+    let options_response = pve.node_firewall_options(args.node());
+    let rules_response = pve.list_node_firewall_rules(args.node());
 
     let enabled = options_response
         .await
@@ -227,10 +223,11 @@ async fn fetch_node_firewall_status(
         _ => None,
     };
 
-    let guests_status = load_guests_firewall_status(pve, node.clone(), &context.guests).await;
+    let guests_status =
+        load_guests_firewall_status(pve, args.node().to_string(), &args.context().guests).await;
 
     Ok(NodeFirewallStatus {
-        node,
+        node: args.node().to_string(),
         status,
         guests: guests_status,
     })
@@ -280,11 +277,11 @@ pub async fn pve_firewall_status(
 
     let node_fetcher = ParallelFetcher::new(context);
     let node_results = node_fetcher
-        .do_for_all_remote_nodes(pve_remotes.iter().cloned(), move |mut ctx, remote, node| {
-            if let Some(guests) = guests_per_remote.get(&remote.id) {
-                ctx.guests = guests.clone();
+        .do_for_all_remote_nodes(pve_remotes.iter().cloned(), move |mut args| {
+            if let Some(guests) = guests_per_remote.get(&args.remote().id) {
+                args.context_mut().guests = guests.clone();
             }
-            fetch_node_firewall_status(ctx, remote, node)
+            fetch_node_firewall_status(args)
         })
         .await;
 
diff --git a/server/src/api/sdn/controllers.rs b/server/src/api/sdn/controllers.rs
index 060fef72..41108674 100644
--- a/server/src/api/sdn/controllers.rs
+++ b/server/src/api/sdn/controllers.rs
@@ -86,9 +86,9 @@ pub async fn list_controllers(
     let fetcher = ParallelFetcher::new((pending, running, ty));
 
     let results = fetcher
-        .do_for_all_remotes(iter.into_remotes(), async |ctx, r, _| {
-            Ok(pve::connect(&r)?
-                .list_controllers(ctx.0, ctx.1, ctx.2)
+        .do_for_all_remotes(iter.into_remotes(), async |args| {
+            Ok(pve::connect(args.remote())?
+                .list_controllers(args.context().0, args.context().1, args.context().2)
                 .await?)
         })
         .await;
diff --git a/server/src/api/sdn/vnets.rs b/server/src/api/sdn/vnets.rs
index 30d431bc..89017438 100644
--- a/server/src/api/sdn/vnets.rs
+++ b/server/src/api/sdn/vnets.rs
@@ -87,8 +87,10 @@ async fn list_vnets(
     let fetcher = ParallelFetcher::new((pending, running));
 
     let results = fetcher
-        .do_for_all_remotes(iter.into_remotes(), async |ctx, r, _| {
-            Ok(pve::connect(&r)?.list_vnets(ctx.0, ctx.1).await?)
+        .do_for_all_remotes(iter.into_remotes(), async |args| {
+            Ok(pve::connect(args.remote())?
+                .list_vnets(args.context().0, args.context().1)
+                .await?)
         })
         .await;
 
diff --git a/server/src/api/sdn/zones.rs b/server/src/api/sdn/zones.rs
index d37dc46c..5d8d8add 100644
--- a/server/src/api/sdn/zones.rs
+++ b/server/src/api/sdn/zones.rs
@@ -93,8 +93,10 @@ pub async fn list_zones(
     let fetcher = ParallelFetcher::new((pending, running, ty));
 
     let results = fetcher
-        .do_for_all_remotes(iter.into_remotes(), async |ctx, r, _| {
-            Ok(pve::connect(&r)?.list_zones(ctx.0, ctx.1, ctx.2).await?)
+        .do_for_all_remotes(iter.into_remotes(), async |args| {
+            Ok(pve::connect(args.remote())?
+                .list_zones(args.context().0, args.context().1, args.context().2)
+                .await?)
         })
         .await;
 
diff --git a/server/src/parallel_fetcher.rs b/server/src/parallel_fetcher.rs
index 8153f1fe..512a04d9 100644
--- a/server/src/parallel_fetcher.rs
+++ b/server/src/parallel_fetcher.rs
@@ -4,18 +4,16 @@
 //! # use anyhow::Error;
 //! #
 //! # use pdm_api_types::remotes::{RemoteType, Remote};
-//! # use server::parallel_fetcher::ParallelFetcher;
+//! # use server::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs};
 //! #
 //! # #[tokio::main]
 //! # async fn main() -> Result<(), Error> {
 //! #   let remotes: Vec<Remote> = Vec::new();
 //! #
 //!     async fn fetch_meaning(
-//!         _context: (),
-//!         remote: Remote,
-//!         node: String,
+//!         args: ParallelFetcherArgs<()>,
 //!     ) -> Result<i32, Error> {
-//!         match remote.ty {
+//!         match args.remote().ty {
 //!             RemoteType::Pve => {
 //!                 // Perform the API request here and return some result.
 //!                 Ok(42)
@@ -269,6 +267,46 @@ impl<C> ParallelFetcherBuilder<C> {
     }
 }
 
+#[non_exhaustive]
+#[derive(Clone)]
+/// The argument type that is passed when calling the closure passed to
+/// [`ParallelFetcher::do_for_all_remote_nodes`] or [`ParallelFetcher::do_for_all_remotes`].
+pub struct ParallelFetcherArgs<C> {
+    /// The context provided in [`ParallelFetcherBuilder::new`] or
+    /// [`ParallelFetcher::new`].
+    context: C,
+    /// The remote.
+    remote: Remote,
+    /// The node. This may be 'localhost' for PBS remotes or if using
+    /// [`ParallelFetcher::do_for_all_remotes`].
+    node: String,
+}
+
+impl<C> ParallelFetcherArgs<C> {
+    /// Get a reference to the remote.
+    pub fn remote(&self) -> &Remote {
+        &self.remote
+    }
+
+    /// Get the current node.
+    ///
+    /// Note: This may be 'localhost' for PBS remotes or if using
+    /// [`ParallelFetcher::do_for_all_remotes`].
+    pub fn node(&self) -> &str {
+        &self.node
+    }
+
+    /// Borrow the previously provided context.
+    pub fn context(&self) -> &C {
+        &self.context
+    }
+
+    /// Mutably borrow the previously provided context.
+    pub fn context_mut(&mut self) -> &mut C {
+        &mut self.context
+    }
+}
+
 /// Helper for parallelizing API requests to multiple remotes/nodes.
 pub struct ParallelFetcher<C> {
     max_connections: usize,
@@ -295,7 +333,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
     ) -> FetcherResponse<MultipleNodesResponse<T>>
     where
         A: Iterator<Item = Remote>,
-        F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static,
+        F: Fn(ParallelFetcherArgs<C>) -> Ft + Clone + Send + 'static,
         Ft: Future<Output = Result<T, Error>> + Send + 'static,
         T: Send + Debug + 'static,
     {
@@ -346,7 +384,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
         max_connections_per_remote: usize,
     ) -> RemoteResponse<MultipleNodesResponse<T>>
     where
-        F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static,
+        F: Fn(ParallelFetcherArgs<C>) -> Ft + Clone + Send + 'static,
         Ft: Future<Output = Result<T, Error>> + Send + 'static,
         T: Send + Debug + 'static,
     {
@@ -456,12 +494,19 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
         _per_remote_connections_permit: Option<OwnedSemaphorePermit>,
     ) -> NodeResponse<T>
     where
-        F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static,
+        F: Fn(ParallelFetcherArgs<C>) -> Ft + Clone + Send + 'static,
         Ft: Future<Output = Result<T, Error>> + Send + 'static,
         T: Send + Debug + 'static,
     {
         let now = Instant::now();
-        let result = func(context, remote.clone(), node.clone()).await;
+
+        let parallel_fetcher_context = ParallelFetcherArgs {
+            context,
+            remote,
+            node: node.clone(),
+        };
+
+        let result = func(parallel_fetcher_context).await;
         let api_response_time = now.elapsed();
 
         NodeResponse {
@@ -479,7 +524,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
     ) -> FetcherResponse<NodeResponse<T>>
     where
         A: Iterator<Item = Remote>,
-        F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static,
+        F: Fn(ParallelFetcherArgs<C>) -> Ft + Clone + Send + 'static,
         Ft: Future<Output = Result<T, Error>> + Send + 'static,
         T: Send + Debug + 'static,
     {
diff --git a/server/src/remote_tasks/refresh_task.rs b/server/src/remote_tasks/refresh_task.rs
index 668b63e4..c1753114 100644
--- a/server/src/remote_tasks/refresh_task.rs
+++ b/server/src/remote_tasks/refresh_task.rs
@@ -11,7 +11,7 @@ use proxmox_section_config::typed::SectionConfigData;
 
 use crate::api;
 use crate::connection;
-use crate::parallel_fetcher::ParallelFetcher;
+use crate::parallel_fetcher::{ParallelFetcher, ParallelFetcherArgs};
 use crate::pbs_client;
 use crate::remote_tasks::{
     KEEP_OLD_FILES, ROTATE_AFTER,
@@ -277,17 +277,16 @@ async fn fetch_remotes(
 }
 
 async fn fetch_tasks_from_single_node(
-    context: Arc<State>,
-    remote: Remote,
-    node: String,
+    args: ParallelFetcherArgs<Arc<State>>,
 ) -> Result<Vec<TaskCacheItem>, Error> {
-    let since = context
-        .cutoff_timestamp(&remote.id, &node)
+    let since = args
+        .context()
+        .cutoff_timestamp(args.remote().id.as_str(), args.node())
         .unwrap_or_else(|| {
             proxmox_time::epoch_i64() - (KEEP_OLD_FILES as u64 * ROTATE_AFTER) as i64
         });
 
-    match remote.ty {
+    match args.remote().ty {
         RemoteType::Pve => {
             let params = pve_api_types::ListTasks {
                 source: Some(pve_api_types::ListTasksSource::All),
@@ -297,13 +296,13 @@ async fn fetch_tasks_from_single_node(
                 ..Default::default()
             };
 
-            let client = connection::make_pve_client(&remote)?;
+            let client = connection::make_pve_client(args.remote())?;
 
             let task_list = client
-                .get_task_list(&node, params)
+                .get_task_list(args.node(), params)
                 .await?
                 .into_iter()
-                .map(|task| map_pve_task(task, remote.id.clone()))
+                .map(|task| map_pve_task(task, args.remote().id.clone()))
                 .collect();
 
             Ok(task_list)
@@ -315,13 +314,13 @@ async fn fetch_tasks_from_single_node(
                 limit: Some(MAX_TASKS_TO_FETCH),
             };
 
-            let client = connection::make_pbs_client(&remote)?;
+            let client = connection::make_pbs_client(args.remote())?;
 
             let task_list = client
                 .get_task_list(params)
                 .await?
                 .into_iter()
-                .map(|task| map_pbs_task(task, remote.id.clone()))
+                .map(|task| map_pbs_task(task, args.remote().id.clone()))
                 .collect();
 
             Ok(task_list)
diff --git a/server/src/remote_updates.rs b/server/src/remote_updates.rs
index 855d9507..bd648efd 100644
--- a/server/src/remote_updates.rs
+++ b/server/src/remote_updates.rs
@@ -232,14 +232,21 @@ pub async fn refresh_update_summary_cache(remotes: Vec<Remote>) -> Result<(), Er
     let fetcher = ParallelFetcher::new(());
 
     let fetch_response = fetcher
-        .do_for_all_remote_nodes(remotes.into_iter(), |context, remote, node| async move {
-            let result = fetch_available_updates(context, remote.clone(), node.clone()).await;
+        .do_for_all_remote_nodes(remotes.into_iter(), |args| async move {
+            let result =
+                fetch_available_updates((), args.remote().clone(), args.node().to_string()).await;
 
             let summary = match &result {
                 Ok(update_info) => update_info.into(),
                 Err(err) => node_error_summary(err),
             };
-            if let Err(err) = update_cached_summary_for_node(remote, node, summary).await {
+            if let Err(err) = update_cached_summary_for_node(
+                args.remote().clone(),
+                args.node().to_string(),
+                summary,
+            )
+            .await
+            {
                 log::error!("could not update 'remote-updates' API cache entry: {err}");
             }
 
-- 
2.47.3