From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [45.144.208.40]) by lore.proxmox.com (Postfix) with ESMTPS id C66671FF0B2 for ; Mon, 24 Aug 2026 14:33:43 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id A3748215D3; Mon, 24 Aug 2026 14:33:43 +0200 (CEST) Message-ID: Date: Mon, 24 Aug 2026 14:33:39 +0200 MIME-Version: 1.0 User-Agent: Mozilla Thunderbird Beta Subject: Re: [PATCH datacenter-manager v2 12/20] parallel fetcher: support a custom client factory To: Lukas Wagner , pdm-devel@lists.proxmox.com References: <20260820145220.418032-1-l.wagner@proxmox.com> <20260820145220.418032-13-l.wagner@proxmox.com> Content-Language: en-US From: Dominik Csapak In-Reply-To: <20260820145220.418032-13-l.wagner@proxmox.com> Content-Type: text/plain; charset=UTF-8; format=flowed Content-Transfer-Encoding: 7bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1787574789800 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.749 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: A5C5RT2IOEMNXLVHM6F2BZMVRE2PJGYG X-Message-ID-Hash: A5C5RT2IOEMNXLVHM6F2BZMVRE2PJGYG X-MailFrom: d.csapak@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox Datacenter Manager development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: just a question: wouldn't it be more future-proof if we always give a a PdmApplication to the parallel fetcher and extract the client factory from there? no way to forget it that way... would probably mean that the commit comes later, after we introduce the pdmapplication state in every api call that uses a parallel fetcher.. On 8/20/26 4:52 PM, Lukas Wagner wrote: > Add a client_factory() builder method and thread the resulting handle > through fetch_remote/fetch_node, defaulting to the client factory of > the current PdmApplication when none is set explicitly. This lets > callers, and later tests, supply a fake client factory instead of > always going through the real one. > > Signed-off-by: Lukas Wagner > --- > server/src/parallel_fetcher.rs | 34 ++++++++++++++++++++++++++++++++-- > 1 file changed, 32 insertions(+), 2 deletions(-) > > diff --git a/server/src/parallel_fetcher.rs b/server/src/parallel_fetcher.rs > index 0819eb86..74ba91b3 100644 > --- a/server/src/parallel_fetcher.rs > +++ b/server/src/parallel_fetcher.rs > @@ -73,7 +73,8 @@ use pve_api_types::ClusterNodeIndexResponse; > > use pdm_api_types::remotes::{Remote, RemoteType}; > > -use crate::connection; > +use crate::connection::ClientFactory; > +use crate::context; > > /// Maximum number of parallel outgoing API requests. > pub const DEFAULT_MAX_CONNECTIONS: usize = 20; > @@ -229,6 +230,7 @@ impl NodeResponse { > pub struct ParallelFetcherBuilder { > max_connections: Option, > max_connections_per_remote: Option, > + client_factory: Option>, > context: C, > } > > @@ -238,6 +240,7 @@ impl ParallelFetcherBuilder { > context, > max_connections: None, > max_connections_per_remote: None, > + client_factory: None, > } > } > > @@ -255,6 +258,12 @@ impl ParallelFetcherBuilder { > self > } > > + /// Set the client factory that should be used. > + pub fn client_factory(mut self, client_factory: Arc) -> Self { > + self.client_factory = Some(client_factory); > + self > + } > + > /// Build the [`ParallelFetcher`] instance. > pub fn build(self) -> ParallelFetcher { > ParallelFetcher { > @@ -262,6 +271,9 @@ impl ParallelFetcherBuilder { > max_connections_per_remote: self > .max_connections_per_remote > .unwrap_or(DEFAULT_MAX_CONNECTIONS_PER_REMOTE), > + client_factory: self > + .client_factory > + .unwrap_or_else(|| context::pdm_application().client_factory_shared()), > context: self.context, > } > } > @@ -279,12 +291,15 @@ pub struct ParallelFetcherArgs { > /// The node. This may be 'localhost' for PBS remotes or if using > /// [`ParallelFetcher::do_for_all_remotes`]. > pub node: String, > + /// A handle to the client factory. > + pub client_factory: Arc, > } > > /// Helper for parallelizing API requests to multiple remotes/nodes. > pub struct ParallelFetcher { > max_connections: usize, > max_connections_per_remote: usize, > + client_factory: Arc, > context: C, > } > > @@ -318,13 +333,16 @@ impl ParallelFetcher { > for remote in remotes { > let semaphore = Arc::clone(&total_connections_semaphore); > > + let client_factory = Arc::clone(&self.client_factory); > let f = func.clone(); > + > let future = Self::fetch_remote( > remote, > self.context.clone(), > semaphore, > f, > self.max_connections_per_remote, > + client_factory, > ); > > if let Some(log_context) = LogContext::current() { > @@ -356,6 +374,7 @@ impl ParallelFetcher { > semaphore: Arc, > func: F, > max_connections_per_remote: usize, > + client_factory: Arc, > ) -> RemoteResponse> > where > F: Fn(ParallelFetcherArgs) -> Ft + Clone + Send + 'static, > @@ -371,8 +390,10 @@ impl ParallelFetcher { > RemoteType::Pve => { > let remote_clone = remote.clone(); > > + let cf = Arc::clone(&client_factory); > + > let nodes = match async move { > - let client = connection::make_pve_client(&remote_clone)?; > + let client = cf.make_pve_client(&remote_clone)?; > let nodes = client.list_nodes().await?; > > Ok::, Error>(nodes) > @@ -407,12 +428,14 @@ impl ParallelFetcher { > let remote_clone = remote.clone(); > let node_name = node.node.clone(); > let context_clone = context.clone(); > + let client_factory = Arc::clone(&client_factory); > > let future = Self::fetch_node( > func_clone, > context_clone, > remote_clone, > node_name, > + client_factory, > permit, > Some(per_remote_connections_permit), > ); > @@ -441,6 +464,7 @@ impl ParallelFetcher { > context, > remote.clone(), > "localhost".into(), > + client_factory, > permit.unwrap(), // Always set to `Some` at this point > None, > ) > @@ -464,6 +488,7 @@ impl ParallelFetcher { > context: C, > remote: Remote, > node: String, > + client_factory: Arc, > _permit: OwnedSemaphorePermit, > _per_remote_connections_permit: Option, > ) -> NodeResponse > @@ -478,6 +503,7 @@ impl ParallelFetcher { > context, > remote, > node: node.clone(), > + client_factory, > }; > > let result = func(parallel_fetcher_context).await; > @@ -514,6 +540,9 @@ impl ParallelFetcher { > > let context = self.context.clone(); > let func = func.clone(); > + > + let client_factory = Arc::clone(&self.client_factory); > + > let future = async move { > let permit = total_connections_semaphore.acquire_owned().await.unwrap(); > > @@ -525,6 +554,7 @@ impl ParallelFetcher { > context, > remote, > "localhost".into(), > + client_factory, > permit, > None, > )