* [PATCH proxmox v3 01/21] router: introduce shared state
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH proxmox v3 02/21] rest-server: allow to inject " Lukas Wagner
` (19 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
API handlers often need access to long-lived application data such as
configuration, caches or client handles. So far the only way to get
there is a global static, which hides the actual dependencies of a
handler and makes testing awkward.
Add a type-keyed registry that a server fills once during startup,
together with an accessor on RpcEnvironment so that handlers can reach
it.
The type-keyed registry enables us to inject multiple types. For
instance, there can be one or more application specific types, but also
context types defined in one of our shared crates, which then allows us
to retrieve the crate-level context handle in an API handler defined in
the crate itself.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
Notes:
Changes since v2:
- In lookup, return a Option<&T> instead o Option<T>.
The api macro now supports taking state parameters by reference,
so returning a borrowed value makes more sense.
The `.cloned()` is moved to the code generated by the API macro, in
case the state argument is taken by-value
proxmox-router/src/cli/environment.rs | 12 ++++++-
proxmox-router/src/lib.rs | 2 ++
proxmox-router/src/rpc_environment.rs | 7 ++++
proxmox-router/src/shared_state.rs | 46 +++++++++++++++++++++++++++
4 files changed, 66 insertions(+), 1 deletion(-)
create mode 100644 proxmox-router/src/shared_state.rs
diff --git a/proxmox-router/src/cli/environment.rs b/proxmox-router/src/cli/environment.rs
index c85105a7..c9aefb75 100644
--- a/proxmox-router/src/cli/environment.rs
+++ b/proxmox-router/src/cli/environment.rs
@@ -5,7 +5,7 @@ use serde_json::Value;
use proxmox_schema::ApiType;
-use crate::{RpcEnvironment, RpcEnvironmentType};
+use crate::{RpcEnvironment, RpcEnvironmentType, SharedStateRegistry};
/// [`RpcEnvironment`] implementation for command line tools.
///
@@ -15,6 +15,7 @@ use crate::{RpcEnvironment, RpcEnvironmentType};
pub struct CliEnvironment {
result_attributes: Value,
auth_id: Option<String>,
+ shared_state_registry: Option<SharedStateRegistry>,
pub(crate) global_options: HashMap<TypeId, Box<dyn Any + Send + Sync + 'static>>,
}
@@ -23,6 +24,11 @@ impl CliEnvironment {
Default::default()
}
+ /// Set the shared state registry for this environment.
+ pub fn set_shared_state_registry(&mut self, registry: SharedStateRegistry) {
+ self.shared_state_registry = Some(registry);
+ }
+
/// Borrow a global option by type.
///
/// Returns `None` if the option type was not registered or no value was provided on the
@@ -98,4 +104,8 @@ impl RpcEnvironment for CliEnvironment {
fn get_auth_id(&self) -> Option<String> {
self.auth_id.clone()
}
+
+ fn shared_state(&self) -> Option<&SharedStateRegistry> {
+ self.shared_state_registry.as_ref()
+ }
}
diff --git a/proxmox-router/src/lib.rs b/proxmox-router/src/lib.rs
index da2f018f..8f329a17 100644
--- a/proxmox-router/src/lib.rs
+++ b/proxmox-router/src/lib.rs
@@ -16,6 +16,7 @@ mod permission;
mod router;
mod rpc_environment;
mod serializable_return;
+mod shared_state;
#[doc(inline)]
#[cfg(feature = "server")]
@@ -25,6 +26,7 @@ pub use permission::*;
pub use router::*;
pub use rpc_environment::{RpcEnvironment, RpcEnvironmentType};
pub use serializable_return::SerializableReturn;
+pub use shared_state::SharedStateRegistry;
// make list_subdirs_api_method! work without an explicit proxmox-schema dependency:
#[doc(hidden)]
diff --git a/proxmox-router/src/rpc_environment.rs b/proxmox-router/src/rpc_environment.rs
index 8ce2d99d..e065504a 100644
--- a/proxmox-router/src/rpc_environment.rs
+++ b/proxmox-router/src/rpc_environment.rs
@@ -2,6 +2,8 @@ use std::any::Any;
use serde_json::Value;
+use crate::SharedStateRegistry;
+
/// Helper to get around `RpcEnvironment: Sized`
pub trait AsAny {
fn as_any(&self) -> &(dyn Any + Send);
@@ -45,6 +47,11 @@ pub trait RpcEnvironment: Any + AsAny + Send {
fn get_client_ip(&self) -> Option<std::net::SocketAddr> {
None // dummy no-op implementation, as most environments don't need this
}
+
+ /// Return a reference to the shared state registry.
+ fn shared_state(&self) -> Option<&SharedStateRegistry> {
+ None
+ }
}
/// Environment Type
diff --git a/proxmox-router/src/shared_state.rs b/proxmox-router/src/shared_state.rs
new file mode 100644
index 00000000..e48bffc8
--- /dev/null
+++ b/proxmox-router/src/shared_state.rs
@@ -0,0 +1,46 @@
+//! Type-keyed state that API handlers can request as a parameter.
+
+use std::any::{Any, TypeId};
+use std::collections::HashMap;
+
+use anyhow::{Error, bail};
+
+/// Registry of state values, keyed by their type.
+///
+/// It holds at most one value per type. Values are registered with
+/// [`register`](SharedStateRegistry::register) before the registry is handed over to the API
+/// environment, from where handlers can access them through
+/// [`RpcEnvironment::shared_state`](crate::RpcEnvironment::shared_state). This allows passing
+/// application context to handlers without resorting to globals.
+#[derive(Default)]
+pub struct SharedStateRegistry {
+ map: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
+}
+
+impl SharedStateRegistry {
+ /// Borrow the registered value of type `T`, if there is one.
+ pub fn lookup<T: 'static + Send + Sync + Clone>(&self) -> Option<&T> {
+ self.map.get(&TypeId::of::<T>()).map(|s| {
+ s.downcast_ref()
+ .expect("mismatch between typeid and contained type")
+ })
+ }
+
+ /// Register a value so that handlers can request it in a `#[state]` parameter.
+ ///
+ /// Since the type is the key, wrap values in a newtype if their type alone is not specific
+ /// enough to identify them.
+ ///
+ /// Fails if a value of type `T` was already registered.
+ pub fn register<T: 'static + Send + Sync + Clone>(&mut self, data: T) -> Result<(), Error> {
+ let type_id = TypeId::of::<T>();
+
+ if self.map.contains_key(&type_id) {
+ bail!("type already registered");
+ }
+
+ self.map.insert(type_id, Box::new(data));
+
+ Ok(())
+ }
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH proxmox v3 02/21] rest-server: allow to inject shared state
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
2026-08-27 11:42 ` [PATCH proxmox v3 01/21] router: introduce shared state Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH proxmox v3 03/21] api-macro: support shared state extraction type Lukas Wagner
` (18 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
Let users of ApiConfig register a shared state registry while setting
up the server and pass it on to API handlers via RestEnvironment.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
proxmox-rest-server/src/api_config.rs | 14 +++++++++++++-
proxmox-rest-server/src/environment.rs | 6 +++++-
2 files changed, 18 insertions(+), 2 deletions(-)
diff --git a/proxmox-rest-server/src/api_config.rs b/proxmox-rest-server/src/api_config.rs
index 3543f8c4..2a8eec70 100644
--- a/proxmox-rest-server/src/api_config.rs
+++ b/proxmox-rest-server/src/api_config.rs
@@ -17,7 +17,7 @@ use proxmox_daemon::command_socket::CommandSocket;
use proxmox_http::Body;
use proxmox_log::{FileLogOptions, FileLogger};
use proxmox_network_types::Cidr;
-use proxmox_router::{Router, RpcEnvironmentType, UserInformation};
+use proxmox_router::{Router, RpcEnvironmentType, SharedStateRegistry, UserInformation};
use proxmox_sys::fs::{CreateOptions, create_path};
use crate::RestEnvironment;
@@ -33,6 +33,8 @@ pub struct ApiConfig {
handlers: Vec<Handler>,
auth_handler: Option<AuthHandler>,
index_handler: Option<IndexHandler>,
+ /// State that API handlers can request via a `#[state]` parameter.
+ pub(crate) shared_state: Option<SharedStateRegistry>,
pub(crate) privileged_addr: Option<PrivilegedAddr>,
// Name of the auth cookie that should be unset on 401 request. If `None` no cookie will be
// removed.
@@ -92,6 +94,7 @@ impl ApiConfig {
index_handler: None,
privileged_addr: None,
auth_cookie_name: None,
+ shared_state: None,
real_ip_header,
real_ip_allow_from: None,
@@ -366,6 +369,15 @@ impl ApiConfig {
.push(Handler::unformatted_router(prefix, router));
self
}
+
+ /// Set the [`SharedStateRegistry`] from which API handlers get their `#[state]` parameters.
+ ///
+ /// All types a handler requests must be registered using
+ /// [`SharedStateRegistry::register`], otherwise a call to the API handler will fail.
+ pub fn with_shared_state(mut self, shared_state: SharedStateRegistry) -> Self {
+ self.shared_state = Some(shared_state);
+ self
+ }
}
#[cfg(feature = "templates")]
diff --git a/proxmox-rest-server/src/environment.rs b/proxmox-rest-server/src/environment.rs
index 47ff5a4f..9133999f 100644
--- a/proxmox-rest-server/src/environment.rs
+++ b/proxmox-rest-server/src/environment.rs
@@ -3,7 +3,7 @@ use std::sync::Arc;
use serde_json::{Value, json};
-use proxmox_router::{RpcEnvironment, RpcEnvironmentType};
+use proxmox_router::{RpcEnvironment, RpcEnvironmentType, SharedStateRegistry};
use crate::ApiConfig;
@@ -89,4 +89,8 @@ impl RpcEnvironment for RestEnvironment {
fn get_client_ip(&self) -> Option<SocketAddr> {
self.client_ip
}
+
+ fn shared_state(&self) -> Option<&SharedStateRegistry> {
+ self.api.shared_state.as_ref()
+ }
}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH proxmox v3 03/21] api-macro: support shared state extraction type
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
2026-08-27 11:42 ` [PATCH proxmox v3 01/21] router: introduce shared state Lukas Wagner
2026-08-27 11:42 ` [PATCH proxmox v3 02/21] rest-server: allow to inject " Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH proxmox v3 04/21] product-config: add ProductConfig type Lukas Wagner
` (17 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
The #[state] attribute marks types that come from the registry, which
allows telling them apart from regular API parameters. The API macro
recognizes them in API handler signatures and looks them up in the
shared state registry of the RpcEnvironment, instead of deserializing
them from the request parameters. That way a handler declares its
dependencies on application state in its signature and does not have to
access the environment by hand.
State arguments can either be taken by value or by reference, the latter
avoids a clone in the method generated by the API macro. Taking the
argument by reference is mutually exclusive with a mutable
`RpcEnvironment` reference, since the state argument is borrowed from
the environment. The macro shows a distinct error message in this case,
educating the developer to use a by-value argument instead.
If we ever decide to store shared state outside the rpc-environment,
this restriction can be lifted.
State arguments may also of type Option<T> or Option<&T>. In this case,
the option is None if the type was never registered or if the shared
state registry was not set up at all.
For non-optional arguments, the API handler fails at runtime if the type
was not registered.
The error-handling in the top-level `api` macro is changed to pass the
token stream to handle_error with the the #[state] attributes removed,
otherwise this leads to bogus 'cannot find attribute `state` in this
scope' errors if there is any other error in the macro.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
Notes:
Changes since v1:
- Fix testcase, it called the wrong api handler
Changes since v2:
- Instead of using a newtype to mark state arguments, use attributes
- Support taking state arguments by reference
- Support taking optional state arguments (by reference or by value)
proxmox-api-macro/src/api/method.rs | 228 ++++++++++++++++++++++++--
proxmox-api-macro/src/api/mod.rs | 6 +
proxmox-api-macro/src/lib.rs | 56 ++++++-
proxmox-api-macro/tests/state.rs | 240 ++++++++++++++++++++++++++++
4 files changed, 520 insertions(+), 10 deletions(-)
create mode 100644 proxmox-api-macro/tests/state.rs
diff --git a/proxmox-api-macro/src/api/method.rs b/proxmox-api-macro/src/api/method.rs
index 5ef69e72..ab5a725a 100644
--- a/proxmox-api-macro/src/api/method.rs
+++ b/proxmox-api-macro/src/api/method.rs
@@ -3,8 +3,10 @@
//! This has to perform quite a few things: infer types from parameters, deal with optional types
//! and defaults, expose parameter and return value schema to the public, and finally create the
//! wrapper function converting from a json Value hash to the parameters listed in the function
-//! signature, while recognizing specially handling `RPCEnvironment` and `ApiMethod` parameters.
+//! signature, while recognizing specially handling `RPCEnvironment` and `ApiMethod` parameters as
+//! well as parameters marked with `#[state]`.
+use std::collections::HashSet;
use std::convert::{TryFrom, TryInto};
use std::mem;
@@ -349,6 +351,7 @@ enum ParameterType {
Value,
ApiMethod,
RpcEnv,
+ State(StateParameter),
Normal(NormalParameter),
}
@@ -357,6 +360,116 @@ struct NormalParameter {
entry: ObjectEntry,
}
+/// A parameter marked with `#[state]`.
+#[derive(Debug)]
+struct StateParameter {
+ /// The type to look up in the shared state registry.
+ ty: syn::Type,
+
+ /// Whether the parameter takes the state by reference.
+ by_ref: bool,
+
+ /// Whether the parameter is an `Option`, in which case an unavailable value is not an error.
+ optional: bool,
+}
+
+impl StateParameter {
+ /// The registry is keyed by type, and neither a reference nor an `Option` is the same type as
+ /// the value itself, so both are removed to get at the type to look up. A parameter taking
+ /// the state by reference borrows it straight out of the registry, and an `Option` parameter
+ /// gets `None` instead of failing if no value of the type is available.
+ fn new(pat_type: &syn::PatType) -> Self {
+ let (ty, optional) = match util::is_option_type(&pat_type.ty) {
+ Some(ty) => (ty, true),
+ None => (&*pat_type.ty, false),
+ };
+
+ let syn::Type::Reference(reference) = ty else {
+ return Self {
+ ty: ty.clone(),
+ by_ref: false,
+ optional,
+ };
+ };
+
+ if let Some(mutability) = &reference.mutability {
+ error!(mutability => "'state' parameters cannot be taken by mutable reference");
+ }
+
+ if let Some(lifetime) = &reference.lifetime
+ && lifetime.ident != "_"
+ {
+ error!(lifetime => "'state' parameters cannot have an explicit lifetime");
+ }
+
+ if matches!(&*reference.elem, syn::Type::Reference(_)) {
+ error!(&*reference.elem => "'state' parameters cannot be nested references");
+ }
+
+ Self {
+ ty: (*reference.elem).clone(),
+ by_ref: true,
+ optional,
+ }
+ }
+}
+
+fn param_attributes_mut(input: &mut syn::FnArg) -> &mut Vec<syn::Attribute> {
+ match input {
+ syn::FnArg::Receiver(receiver) => &mut receiver.attrs,
+ syn::FnArg::Typed(pat_type) => &mut pat_type.attrs,
+ }
+}
+
+fn is_state_attribute(attr: &syn::Attribute) -> bool {
+ attr.style == syn::AttrStyle::Outer && attr.path().is_ident("state")
+}
+
+/// Removes `#[state]` attributes from the function's parameters and returns the
+/// positions of the parameters which had one.
+///
+/// The attribute only has meaning to this macro, so it has to be stripped before the function is
+/// emitted again. Other attributes are preserved.
+fn take_state_attributes(sig: &mut syn::Signature) -> HashSet<usize> {
+ let mut state_param_indices = HashSet::new();
+
+ for (index, input) in sig.inputs.iter_mut().enumerate() {
+ let attrs = param_attributes_mut(input);
+
+ for attr in mem::take(attrs) {
+ if !is_state_attribute(&attr) {
+ attrs.push(attr);
+ continue;
+ }
+
+ if !matches!(attr.meta, syn::Meta::Path(_)) {
+ error!(&attr.meta => "'state' attribute does not take any data");
+ }
+
+ state_param_indices.insert(index);
+ }
+ }
+
+ state_param_indices
+}
+
+/// Removes `#[state]` attributes from a function without any diagnostics.
+///
+/// Used for the copy of the original function which is emitted alongside a fatal error, where the
+/// leftover attributes would only add confusing follow-up errors due to `state` not being in
+/// scope.
+pub(crate) fn strip_state_attributes(item: TokenStream) -> TokenStream {
+ let Ok(mut func) = syn::parse2::<syn::ItemFn>(item.clone()) else {
+ return item;
+ };
+
+ for input in func.sig.inputs.iter_mut() {
+ param_attributes_mut(input).retain(|attr| !is_state_attribute(attr));
+ }
+
+ func.into_token_stream()
+}
+
fn check_input_type(input: &syn::FnArg) -> Result<(&syn::PatType, &syn::PatIdent), syn::Error> {
// `self` types are not supported:
let pat_type = match input {
@@ -374,15 +487,22 @@ fn check_input_type(input: &syn::FnArg) -> Result<(&syn::PatType, &syn::PatIdent
}
fn handle_function_signature(method_info: &mut MethodInfo) -> Result<Ident, Error> {
+ let state_params = take_state_attributes(&mut method_info.func.sig);
+
let sig = &method_info.func.sig;
let mut api_method_param = None;
let mut rpc_env_param = None;
+ let mut rpc_env_mutable = false;
let mut value_param = None;
let mut param_list = Vec::<(FieldName, ParameterType)>::new();
- for input in sig.inputs.iter() {
+ for (index, input) in sig.inputs.iter().enumerate() {
+ if state_params.contains(&index) {
+ continue;
+ }
+
let (pat_type, pat) = match check_input_type(input) {
Ok(input) => input,
Err(err) => {
@@ -410,7 +530,7 @@ fn handle_function_signature(method_info: &mut MethodInfo) -> Result<Ident, Erro
};
}
- for input in sig.inputs.iter() {
+ for (index, input) in sig.inputs.iter().enumerate() {
let (pat_type, pat) = match check_input_type(input) {
Ok(input) => input,
Err(_err) => continue, // we already produced errors above,
@@ -421,17 +541,20 @@ fn handle_function_signature(method_info: &mut MethodInfo) -> Result<Ident, Erro
// values, and our 2 fixed function parameters: `&ApiMethod` and `&mut dyn RpcEnvironment`.
//
// Our strategy is as follows:
- // 1) See if the parameter name also appears in the input schema. In this case we
+ // 1) See if the parameter was marked with `#[state]`. Such parameters are filled in
+ // from the environment's shared state registry and are not part of the schema.
+ //
+ // 2) See if the parameter name also appears in the input schema. In this case we
// assume that we want the parameter to be extracted from the `Value` and passed
// directly to the function.
//
- // 2) Check the parameter type for `&ApiMethod` and remember its position (since we may
+ // 3) Check the parameter type for `&ApiMethod` and remember its position (since we may
// need to reorder it!)
//
- // 3) Check the parameter type for `&dyn RpcEnvironment` and remember its position
+ // 4) Check the parameter type for `&dyn RpcEnvironment` and remember its position
// (since we may need to reorder it!).
//
- // 4) Check for a `Value` or `serde_json::Value` parameter. This becomes the
+ // 5) Check for a `Value` or `serde_json::Value` parameter. This becomes the
// "catch-all" parameter and only 1 may exist.
// Note that we may still use further `Value` parameters if they have been
// explicitly named in the `input_schema`. However, only 1 unnamed `Value` parameter
@@ -439,11 +562,25 @@ fn handle_function_signature(method_info: &mut MethodInfo) -> Result<Ident, Erro
// If no such parameter exists, we automatically fail the function if the `Value` is
// not empty after extracting the parameters.
//
- // 5) Finally, if none of the above conditions are met, we do not know what to do and
+ // 6) Finally, if none of the above conditions are met, we do not know what to do and
// bail out with an error.
let pat_ident = pat.ident.unraw();
let mut param_name: FieldName = pat_ident.clone().into();
- let param_type = if let Some(entry) = method_info
+ let param_type = if state_params.contains(&index) {
+ if method_info
+ .input_schema
+ .find_obj_property_by_ident(&pat_ident.to_string())
+ .is_some()
+ {
+ error!(
+ pat_type => "'state' parameter '{}' must not appear in the input schema",
+ pat_ident,
+ );
+ continue;
+ }
+
+ ParameterType::State(StateParameter::new(pat_type))
+ } else if let Some(entry) = method_info
.input_schema
.find_obj_property_by_ident(&pat_ident.to_string())
{
@@ -469,6 +606,8 @@ fn handle_function_signature(method_info: &mut MethodInfo) -> Result<Ident, Erro
continue;
}
rpc_env_param = Some(param_list.len());
+ rpc_env_mutable =
+ matches!(&*pat_type.ty, syn::Type::Reference(r) if r.mutability.is_some());
ParameterType::RpcEnv
} else if is_value_type(&pat_type.ty) {
if value_param.is_some() {
@@ -485,6 +624,23 @@ fn handle_function_signature(method_info: &mut MethodInfo) -> Result<Ident, Erro
param_list.push((param_name, param_type));
}
+ // the borrow of a `#[state]` reference reaches into the environment, so it cannot be handed
+ // out together with a mutable borrow of the very same environment:
+ if rpc_env_mutable {
+ for (name, param) in ¶m_list {
+ if let ParameterType::State(state) = param
+ && state.by_ref
+ {
+ error!(
+ name.span(),
+ "'state' parameter '{}' cannot be borrowed in a method also taking a mutable \
+ 'RpcEnvironment' parameter, take the state by value instead",
+ name.as_str(),
+ );
+ }
+ }
+ }
+
create_wrapper_function(method_info, param_list)
}
@@ -563,6 +719,9 @@ fn create_wrapper_function(
&mut method_info.default_consts,
)?;
}
+ ParameterType::State(param) => {
+ extract_state_parameter(param, &mut body, &mut args, name, span)?;
+ }
}
}
@@ -824,6 +983,57 @@ fn extract_normal_parameter(
Ok(())
}
+fn extract_state_parameter(
+ param: StateParameter,
+ body: &mut TokenStream,
+ args: &mut TokenStream,
+ name: FieldName,
+ span: Span,
+) -> Result<(), Error> {
+ let name_str = syn::LitStr::new(name.as_str(), span);
+ let arg_name = Ident::new(&format!("state_arg_{}", name.as_ident()), span);
+ let ty = param.ty;
+
+ // SharedStateRegistry::lookup() returns a borrowed value, so a parameter taking the state
+ // by value needs the reference to be cloned.
+ let clone = (!param.by_ref).then(|| quote_spanned! { span => .cloned() });
+
+ let lookup = if param.optional {
+ // an unavailable value is not an error here, which includes an environment without any
+ // registry at all.
+ quote_spanned! { span =>
+ let #arg_name = rpc_env_param
+ .shared_state()
+ .and_then(|shared_state| shared_state.lookup::<#ty>())
+ #clone;
+ }
+ } else {
+ quote_spanned! { span =>
+ let #arg_name = rpc_env_param
+ .shared_state()
+ .ok_or_else(|| ::anyhow::format_err!(
+ "router has no shared state set up",
+ ))?
+ .lookup::<#ty>()
+ #clone
+ .ok_or_else(|| ::anyhow::format_err!(
+ "shared state type '{}' for parameter '{}' not registered",
+ ::std::any::type_name::<#ty>(),
+ #name_str,
+ ))?;
+ }
+ };
+
+ body.extend(quote_spanned! { span =>
+ #[allow(non_snake_case)]
+ #lookup
+ });
+
+ args.extend(quote_spanned! { span => #arg_name, });
+
+ Ok(())
+}
+
/// Returns a tuple containing the schema code first and the `ParameterSchema` parameter for the
/// `ApiMethod` second.
fn serialize_input_schema(
diff --git a/proxmox-api-macro/src/api/mod.rs b/proxmox-api-macro/src/api/mod.rs
index b87c49eb..be5ab9f3 100644
--- a/proxmox-api-macro/src/api/mod.rs
+++ b/proxmox-api-macro/src/api/mod.rs
@@ -790,6 +790,12 @@ impl SchemaArray {
}
}
+/// Removes the attributes `#[api]` consumes itself, so that the item can be emitted as-is when the
+/// macro fails with a fatal error.
+pub(crate) fn strip_helper_attributes(item: TokenStream) -> TokenStream {
+ method::strip_state_attributes(item)
+}
+
/// Parse `input`, `returns` and `protected` attributes out of an function annotated
/// with an `#[api]` attribute and produce a `const ApiMethod` named after the function.
///
diff --git a/proxmox-api-macro/src/lib.rs b/proxmox-api-macro/src/lib.rs
index c7aebfae..a9d62e8f 100644
--- a/proxmox-api-macro/src/lib.rs
+++ b/proxmox-api-macro/src/lib.rs
@@ -210,6 +210,59 @@ fn router_do(item: TokenStream) -> Result<TokenStream, Error> {
}
```
+ # Shared state parameters.
+
+ Parameters marked with `#[state]` are not part of the API schema. Instead of being extracted
+ from the API input, they are filled in from the `SharedStateRegistry` of the environment
+ the handler is called with. The handler receives a clone of the value registered for the
+ parameter's type and fails if no value of that type was registered:
+
+ ```no_run
+ # use proxmox_api_macro::api;
+ # use anyhow::Error;
+ #[derive(Clone)]
+ struct AuthContext {
+ realm: String,
+ }
+
+ #[api]
+ /// Show which realm the daemon was configured with.
+ fn print_realm(#[state] auth: AuthContext) -> Result<(), Error> {
+ println!("realm: {}", auth.realm);
+
+ Ok(())
+ }
+ ```
+
+ Such a parameter can also be taken by shared reference to avoid the clone, in which case it
+ borrows the value straight out of the registry. The registry is keyed by type, so the type the
+ reference points to is what has to be registered. The reference must not have an explicit
+ lifetime.
+
+ Since the state argument is borrowed from the RpcEnvironemnt, it cannot be combined with a
+ `&mut dyn RpcEnvironment` parameter.
+
+ Both forms can be wrapped in an `Option`, which turns an unavailable value from an error into a
+ `None`. This covers a type which was not registered as well as an environment without any
+ registry at all.
+
+ ```no_run
+ # use proxmox_api_macro::api;
+ # use anyhow::Error;
+ # #[derive(Clone)]
+ # struct AuthContext { realm: String }
+ #[api]
+ /// Show which realm the daemon was configured with, if any.
+ fn print_realm(#[state] auth: Option<&AuthContext>) -> Result<(), Error> {
+ match auth {
+ Some(auth) => println!("realm: {}", auth.realm),
+ None => println!("no auth context configured"),
+ }
+
+ Ok(())
+ }
+ ```
+
# Deprecated property aliases
Use this when a property has been renamed and you want to keep the old name accepted for
@@ -504,7 +557,8 @@ fn router_do(item: TokenStream) -> Result<TokenStream, Error> {
pub fn api(attr: TokenStream_1, item: TokenStream_1) -> TokenStream_1 {
let _error_guard = init_local_error();
let item: TokenStream = item.into();
- handle_error(item.clone(), api::api(attr.into(), item)).into()
+ let fallback = api::strip_helper_attributes(item.clone());
+ handle_error(fallback, api::api(attr.into(), item)).into()
}
/// *Experimental:* Transform a json-like schema definition into an expression yielding a `Schema`.
diff --git a/proxmox-api-macro/tests/state.rs b/proxmox-api-macro/tests/state.rs
new file mode 100644
index 00000000..8b9ee39d
--- /dev/null
+++ b/proxmox-api-macro/tests/state.rs
@@ -0,0 +1,240 @@
+use proxmox_api_macro::api;
+use proxmox_router::SharedStateRegistry;
+
+use anyhow::Error;
+use serde_json::{Value, json};
+
+#[derive(Clone)]
+struct Foo {
+ val: i32,
+}
+
+#[derive(Clone)]
+struct Unregistered;
+
+#[api(
+ input: {
+ properties: {
+ value: {
+ description: "Something",
+ }
+ }
+ }
+)]
+/// Test multiple state args.
+fn multiple_args(
+ #[state] state_arg1: i32,
+ value: isize,
+ #[state] state_arg2: Foo,
+) -> Result<(), Error> {
+ assert_eq!(state_arg1, 10);
+ assert_eq!(state_arg2.val, 20);
+
+ assert_eq!(value, 50);
+
+ Ok(())
+}
+
+#[api(
+ input: {
+ properties: {
+ value: {
+ description: "Something",
+ }
+ }
+ }
+)]
+/// Test state args taken by reference.
+fn by_reference(
+ #[state] state_arg1: &i32,
+ value: isize,
+ #[state] state_arg2: &Foo,
+) -> Result<(), Error> {
+ assert_eq!(*state_arg1, 10);
+ assert_eq!(state_arg2.val, 20);
+
+ assert_eq!(value, 50);
+
+ Ok(())
+}
+
+#[api]
+/// Test an owned state arg next to a mutable environment.
+fn with_rpcenv(
+ #[state] state_arg: Foo,
+ rpcenv: &mut dyn proxmox_router::RpcEnvironment,
+) -> Result<(), Error> {
+ let _ = rpcenv;
+
+ assert_eq!(state_arg.val, 20);
+
+ Ok(())
+}
+
+#[api]
+/// Test optional state args.
+fn optional_args(
+ #[state] state_arg1: Option<i32>,
+ #[state] state_arg2: Option<&Foo>,
+ #[state] state_arg3: Option<Unregistered>,
+) -> Result<(), Error> {
+ assert_eq!(state_arg1, Some(10));
+ assert_eq!(state_arg2.map(|state| state.val), Some(20));
+ assert!(state_arg3.is_none());
+
+ Ok(())
+}
+
+#[api]
+/// Test optional state args without a registry.
+fn optional_without_registry(
+ #[state] state_arg1: Option<i32>,
+ #[state] state_arg2: Option<&Foo>,
+) -> Result<(), Error> {
+ assert!(state_arg1.is_none());
+ assert!(state_arg2.is_none());
+
+ Ok(())
+}
+
+#[api]
+/// Test not registered state arg.
+fn not_registered(#[state] _state: Foo) -> Result<(), Error> {
+ panic!("should not reach this");
+}
+
+#[api(
+ input: {
+ properties: {
+ unused: {
+ description: "foo",
+ }
+ }
+ }
+)]
+/// Test whether attributes on *other* arguments are preserved.
+fn other_attributes_are_preserved(
+ #[state] _state_arg: Option<Foo>,
+ #[allow(unused)] unused: i32,
+) -> Result<(), Error> {
+ Ok(())
+}
+
+struct RpcEnv {
+ shared_state: Option<SharedStateRegistry>,
+}
+impl proxmox_router::RpcEnvironment for RpcEnv {
+ fn result_attrib_mut(&mut self) -> &mut Value {
+ panic!("result_attrib_mut called");
+ }
+
+ fn result_attrib(&self) -> &Value {
+ panic!("result_attrib called");
+ }
+
+ /// The environment type
+ fn env_type(&self) -> proxmox_router::RpcEnvironmentType {
+ panic!("env_type called");
+ }
+
+ /// Set authentication id
+ fn set_auth_id(&mut self, user: Option<String>) {
+ let _ = user;
+ panic!("set_auth_id called");
+ }
+
+ /// Get authentication id
+ fn get_auth_id(&self) -> Option<String> {
+ panic!("get_auth_id called");
+ }
+
+ fn shared_state(&self) -> Option<&SharedStateRegistry> {
+ self.shared_state.as_ref()
+ }
+}
+
+#[test]
+fn test_invocations() {
+ let mut shared_state = SharedStateRegistry::default();
+
+ shared_state.register::<i32>(10).unwrap();
+ shared_state.register::<Foo>(Foo { val: 20 }).unwrap();
+
+ let mut env = RpcEnv {
+ shared_state: Some(shared_state),
+ };
+
+ api_function_multiple_args(
+ json!({
+ "value": 50,
+ }),
+ &API_METHOD_MULTIPLE_ARGS,
+ &mut env,
+ )
+ .expect("func with multiple injected works");
+
+ api_function_by_reference(
+ json!({
+ "value": 50,
+ }),
+ &API_METHOD_BY_REFERENCE,
+ &mut env,
+ )
+ .expect("func with injected references works");
+
+ api_function_with_rpcenv(json!({}), &API_METHOD_WITH_RPCENV, &mut env)
+ .expect("func with injected state and environment works");
+
+ api_function_optional_args(json!({}), &API_METHOD_OPTIONAL_ARGS, &mut env)
+ .expect("func with optional injected state works");
+}
+
+#[test]
+fn test_not_registered() {
+ let mut env = RpcEnv {
+ shared_state: Some(SharedStateRegistry::default()),
+ };
+
+ api_function_not_registered(json!({}), &API_METHOD_NOT_REGISTERED, &mut env)
+ .expect_err("func did not fail");
+
+ api_function_optional_without_registry(
+ json!({}),
+ &API_METHOD_OPTIONAL_WITHOUT_REGISTRY,
+ &mut env,
+ )
+ .expect("optional state args are 'None' if not registered");
+}
+
+#[test]
+fn test_without_registry() {
+ let mut env = RpcEnv { shared_state: None };
+
+ api_function_not_registered(json!({}), &API_METHOD_NOT_REGISTERED, &mut env)
+ .expect_err("func did not fail");
+
+ api_function_optional_without_registry(
+ json!({}),
+ &API_METHOD_OPTIONAL_WITHOUT_REGISTRY,
+ &mut env,
+ )
+ .expect("optional state args are 'None' without a registry");
+}
+
+#[test]
+/// Test whether attributes on *other* arguments are preserved.
+///
+/// This test does not really assert anything, but the called function
+/// will emit a warning during compilation if it does not work.
+fn test_attributes_are_preserved() {
+ let mut env = RpcEnv { shared_state: None };
+
+ api_function_other_attributes_are_preserved(
+ json!({
+ "unused": 3
+ }),
+ &API_METHOD_OTHER_ATTRIBUTES_ARE_PRESERVED,
+ &mut env,
+ )
+ .unwrap();
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH proxmox v3 04/21] product-config: add ProductConfig type
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (2 preceding siblings ...)
2026-08-27 11:42 ` [PATCH proxmox v3 03/21] api-macro: support shared state extraction type Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 05/21] context: promote context to a dir-style module Lukas Wagner
` (16 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
This type bundles user information for priv/unpriv users and common
runtime paths. This type should be instantiated once and then live in
the application code base, either in a OnceLock or an application
context handle.
Some of the methods are based on the existing helpers from
filesystem_helpers.rs, with some additional ones to account for
permission differences between directories and files (execute bit is set
for directories).
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
proxmox-product-config/src/lib.rs | 3 +
proxmox-product-config/src/product_config.rs | 205 +++++++++++++++++++
2 files changed, 208 insertions(+)
create mode 100644 proxmox-product-config/src/product_config.rs
diff --git a/proxmox-product-config/src/lib.rs b/proxmox-product-config/src/lib.rs
index 9576a08d..48a6ba56 100644
--- a/proxmox-product-config/src/lib.rs
+++ b/proxmox-product-config/src/lib.rs
@@ -5,3 +5,6 @@ pub use filesystem_helpers::*;
mod init;
pub use init::*;
+
+mod product_config;
+pub use product_config::{ProductConfig, ProductConfigParams};
diff --git a/proxmox-product-config/src/product_config.rs b/proxmox-product-config/src/product_config.rs
new file mode 100644
index 00000000..d38764ec
--- /dev/null
+++ b/proxmox-product-config/src/product_config.rs
@@ -0,0 +1,205 @@
+use std::path::{Path, PathBuf};
+
+use nix::sys::stat::Mode;
+use nix::unistd::User;
+
+use proxmox_sys::fs::CreateOptions;
+
+/// Parameter type for [`ProductConfig::new`].
+///
+/// This type is used to avoid having long lists of parameters with the same type, while also
+/// keeping the members of the final type, [`ProductConfig`] private.
+#[derive(Clone, Debug)]
+pub struct ProductConfigParams {
+ /// The user the unprivileged API daemon runs as.
+ pub api_user: User,
+
+ /// The user the privileged API daemon runs as.
+ pub priv_user: User,
+
+ /// Directory for configuration files.
+ ///
+ /// This is typically in `/etc/<product>`.
+ pub config_dir: PathBuf,
+
+ /// Directory for persistent state.
+ ///
+ /// This is typically in `/var/lib/<product>`.
+ pub state_dir: PathBuf,
+
+ /// Directory for runtime data, not persisted across reboots.
+ ///
+ /// This is typically in `/run/<product>`.
+ pub run_dir: PathBuf,
+
+ /// Directory for cached data, which can be regenerated if lost.
+ ///
+ /// This is typically in `/var/cache/<product>`.
+ pub cache_dir: PathBuf,
+}
+
+/// Product-specific configuration, such as the users the product runs as and the directories it
+/// stores its files in.
+///
+/// # Examples
+///
+/// ```no_run
+/// use nix::unistd::User;
+///
+/// use proxmox_product_config::{ProductConfig, ProductConfigParams};
+///
+/// # fn main() -> Result<(), anyhow::Error> {
+/// let config = ProductConfig::new(ProductConfigParams {
+/// api_user: User::from_name("www-data")?.expect("www-data user exists"),
+/// priv_user: User::from_name("root")?.expect("root user exists"),
+/// config_dir: "/etc/proxmox-product".into(),
+/// state_dir: "/var/lib/proxmox-product".into(),
+/// run_dir: "/run/proxmox-product".into(),
+/// cache_dir: "/var/cache/proxmox-product".into(),
+/// });
+///
+/// // Write a config file as `www-data:www-data` with mode 0640.
+/// let config_file = config.config_dir().join("product.conf");
+/// let options = config.default_file_create_options();
+/// proxmox_sys::fs::replace_file(&config_file, b"key: value\n", options, true)?;
+///
+/// // Files holding secrets are only accessible to the privileged daemon: `root:root`, mode 0600.
+/// let key_file = config.config_dir().join("auth.key");
+/// let options = config.secret_file_create_options();
+/// proxmox_sys::fs::replace_file(&key_file, b"secret\n", options, true)?;
+/// # Ok(())
+/// # }
+/// ```
+#[derive(Clone, Debug)]
+pub struct ProductConfig(ProductConfigParams);
+
+impl ProductConfig {
+ /// Create a new [`ProductConfig`] from the provided
+ /// [`ProductConfigParams`].
+ pub fn new(params: ProductConfigParams) -> Self {
+ Self(params)
+ }
+
+ /// The user the unprivileged API daemon runs as.
+ pub fn api_user(&self) -> &User {
+ &self.0.api_user
+ }
+
+ /// The user the privileged API daemon runs as.
+ pub fn priv_user(&self) -> &User {
+ &self.0.priv_user
+ }
+
+ /// Directory for configuration files.
+ ///
+ /// This is typically in `/etc/<product>`.
+ pub fn config_dir(&self) -> &Path {
+ &self.0.config_dir
+ }
+
+ /// Directory for persistent state.
+ ///
+ /// This is typically in `/var/lib/<product>`.
+ pub fn state_dir(&self) -> &Path {
+ &self.0.state_dir
+ }
+
+ /// Directory for runtime data, not persisted across reboots.
+ ///
+ /// This is typically in `/run/<product>`.
+ pub fn run_dir(&self) -> &Path {
+ &self.0.run_dir
+ }
+
+ /// Directory for cached data, which can be regenerated if lost.
+ ///
+ /// This is typically in `/var/cache/<product>`.
+ pub fn cache_dir(&self) -> &Path {
+ &self.0.cache_dir
+ }
+
+ /// Default options for creating files: mode 0640, owned by the API user.
+ pub fn default_file_create_options(&self) -> CreateOptions {
+ let api_user = self.api_user();
+ let mode = Mode::from_bits_truncate(0o0640);
+
+ CreateOptions::new()
+ .perm(mode)
+ .owner(api_user.uid)
+ .group(api_user.gid)
+ }
+
+ /// Default options for creating directories: mode 0750, owned by the API user.
+ pub fn default_dir_create_options(&self) -> CreateOptions {
+ let api_user = self.api_user();
+ let mode = Mode::from_bits_truncate(0o0750);
+
+ CreateOptions::new()
+ .perm(mode)
+ .owner(api_user.uid)
+ .group(api_user.gid)
+ }
+
+ /// Return [CreateOptions] for files owned by `priv_user.uid:api_user.gid` with permission `0640`.
+ ///
+ /// Only `priv_user` can write those files, but group `api_user.gid` can read them.
+ pub fn privileged_file_create_options(&self) -> CreateOptions {
+ let api_user = self.api_user();
+ let priv_user = self.priv_user();
+ let mode = Mode::from_bits_truncate(0o0640);
+
+ CreateOptions::new()
+ .perm(mode)
+ .owner(priv_user.uid)
+ .group(api_user.gid)
+ }
+
+ /// Return [CreateOptions] for files owned by `priv_user.uid:api_user.gid` with permission `0750`.
+ ///
+ /// Only `priv_user` can write those files, but group `api_user.gid` can read them.
+ pub fn privileged_dir_create_options(&self) -> CreateOptions {
+ let api_user = self.api_user();
+ let priv_user = self.priv_user();
+ let mode = Mode::from_bits_truncate(0o0750);
+
+ CreateOptions::new()
+ .perm(mode)
+ .owner(priv_user.uid)
+ .group(api_user.gid)
+ }
+
+ /// Return [CreateOptions] for files owned by `priv_user.uid:priv_user.gid` with permission `0600`.
+ ///
+ /// Only `priv_user` can read and write those files.
+ pub fn secret_file_create_options(&self) -> CreateOptions {
+ let priv_user = self.priv_user();
+ let mode = Mode::from_bits_truncate(0o0600);
+
+ CreateOptions::new()
+ .perm(mode)
+ .owner(priv_user.uid)
+ .group(priv_user.gid)
+ }
+
+ /// Return [CreateOptions] for directories owned by `priv_user.uid:priv_user.gid` with permission `0600`.
+ ///
+ /// Only `priv_user` can read and write those files.
+ pub fn secret_dir_create_options(&self) -> CreateOptions {
+ let priv_user = self.priv_user();
+ let mode = Mode::from_bits_truncate(0o0700);
+
+ CreateOptions::new()
+ .perm(mode)
+ .owner(priv_user.uid)
+ .group(priv_user.gid)
+ }
+
+ /// Return [CreateOptions] for lock files, owner `api_user.uid/api_user.gid` and mode `0660`.
+ pub fn lockfile_create_options(&self) -> CreateOptions {
+ let api_user = self.api_user();
+ CreateOptions::new()
+ .perm(Mode::from_bits_truncate(0o660))
+ .owner(api_user.uid)
+ .group(api_user.gid)
+ }
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 05/21] context: promote context to a dir-style module
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (3 preceding siblings ...)
2026-08-27 11:42 ` [PATCH proxmox v3 04/21] product-config: add ProductConfig type Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 06/21] pdm-config: remotes: rename trait methods to read/write/lock Lukas Wagner
` (15 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
No functional changes.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
server/src/{context.rs => context/mod.rs} | 0
1 file changed, 0 insertions(+), 0 deletions(-)
rename server/src/{context.rs => context/mod.rs} (100%)
diff --git a/server/src/context.rs b/server/src/context/mod.rs
similarity index 100%
rename from server/src/context.rs
rename to server/src/context/mod.rs
--
2.47.3
^ permalink raw reply [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 06/21] pdm-config: remotes: rename trait methods to read/write/lock
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (4 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 05/21] context: promote context to a dir-style module Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 07/21] pdm-config: subscriptions: " Lukas Wagner
` (14 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
This makes the use of these less awkward when being called via the trait
object itself, e.g. on a dependency injected application context object:
app.remote_config().read()
instead of
app.remote_config().config()
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
lib/pdm-config/src/remotes.rs | 24 ++++++++++++------------
server/src/test_support/fake_remote.rs | 8 ++++----
2 files changed, 16 insertions(+), 16 deletions(-)
diff --git a/lib/pdm-config/src/remotes.rs b/lib/pdm-config/src/remotes.rs
index a1004cfe..47754e9b 100644
--- a/lib/pdm-config/src/remotes.rs
+++ b/lib/pdm-config/src/remotes.rs
@@ -35,36 +35,36 @@ fn instance() -> &'static (dyn RemoteConfig + Send + Sync) {
///
/// Will panic if the the remote config instance has not been set before.
pub fn lock_config() -> Result<ApiLockGuard, Error> {
- instance().lock_config()
+ instance().lock()
}
/// Return contents of the remotes config
///
/// Will panic if the the remote config instance has not been set before.
pub fn config() -> Result<(SectionConfigData<Remote>, ConfigDigest), Error> {
- instance().config()
+ instance().read()
}
pub fn get_secret_token(remote: &Remote) -> Result<String, Error> {
- instance().get_secret_token(remote)
+ instance().read_secret_token(remote)
}
/// Replace the currently persisted remotes config
///
/// Will panic if the the remote config instance has not been set before.
pub fn save_config(config: SectionConfigData<Remote>) -> Result<(), Error> {
- instance().save_config(config)
+ instance().write(config)
}
pub trait RemoteConfig {
/// Return contents of the remotes config
- fn config(&self) -> Result<(SectionConfigData<Remote>, ConfigDigest), Error>;
+ fn read(&self) -> Result<(SectionConfigData<Remote>, ConfigDigest), Error>;
/// Return contents of the remotes shadow config
- fn get_secret_token(&self, remote: &Remote) -> Result<String, Error>;
+ fn read_secret_token(&self, remote: &Remote) -> Result<String, Error>;
/// Lock the remotes config
- fn lock_config(&self) -> Result<ApiLockGuard, Error>;
+ fn lock(&self) -> Result<ApiLockGuard, Error>;
/// Replace the currently persisted remotes config
- fn save_config(&self, remotes: SectionConfigData<Remote>) -> Result<(), Error>;
+ fn write(&self, remotes: SectionConfigData<Remote>) -> Result<(), Error>;
}
/// Default, production implementation for reading/writing the `remotes.cfg`
@@ -72,11 +72,11 @@ pub trait RemoteConfig {
pub struct DefaultRemoteConfig;
impl RemoteConfig for DefaultRemoteConfig {
- fn lock_config(&self) -> Result<ApiLockGuard, Error> {
+ fn lock(&self) -> Result<ApiLockGuard, Error> {
open_api_lockfile(REMOTES_CFG_LOCKFILE, None, true)
}
- fn config(&self) -> Result<(SectionConfigData<Remote>, ConfigDigest), Error> {
+ fn read(&self) -> Result<(SectionConfigData<Remote>, ConfigDigest), Error> {
let content =
proxmox_sys::fs::file_read_optional_string(REMOTES_CFG_FILENAME)?.unwrap_or_default();
@@ -86,7 +86,7 @@ impl RemoteConfig for DefaultRemoteConfig {
Ok((data, digest.into()))
}
- fn save_config(&self, mut config: SectionConfigData<Remote>) -> Result<(), Error> {
+ fn write(&self, mut config: SectionConfigData<Remote>) -> Result<(), Error> {
let shadow_content = proxmox_sys::fs::file_read_optional_string(REMOTES_SHADOW_FILENAME)?
.unwrap_or_default();
@@ -135,7 +135,7 @@ impl RemoteConfig for DefaultRemoteConfig {
replace_config(REMOTES_CFG_FILENAME, raw.as_bytes())
}
- fn get_secret_token(&self, remote: &Remote) -> Result<String, Error> {
+ fn read_secret_token(&self, remote: &Remote) -> Result<String, Error> {
// not yet rewritten into shadow config
if remote.token != "-" {
return Ok(remote.token.clone());
diff --git a/server/src/test_support/fake_remote.rs b/server/src/test_support/fake_remote.rs
index e13ffa43..f387029f 100644
--- a/server/src/test_support/fake_remote.rs
+++ b/server/src/test_support/fake_remote.rs
@@ -35,7 +35,7 @@ pub struct FakeRemoteConfig {
}
impl RemoteConfig for FakeRemoteConfig {
- fn config(&self) -> Result<(SectionConfigData<Remote>, ConfigDigest), Error> {
+ fn read(&self) -> Result<(SectionConfigData<Remote>, ConfigDigest), Error> {
let mut section_config = SectionConfigData::default();
for i in 0..self.nr_of_pve_remotes {
@@ -59,15 +59,15 @@ impl RemoteConfig for FakeRemoteConfig {
Ok((section_config, digest))
}
- fn lock_config(&self) -> Result<ApiLockGuard, Error> {
+ fn lock(&self) -> Result<ApiLockGuard, Error> {
unsafe { Ok(proxmox_product_config::create_mocked_lock()) }
}
- fn save_config(&self, _remotes: SectionConfigData<Remote>) -> Result<(), Error> {
+ fn write(&self, _remotes: SectionConfigData<Remote>) -> Result<(), Error> {
Ok(())
}
- fn get_secret_token(&self, _remote: &Remote) -> Result<String, Error> {
+ fn read_secret_token(&self, _remote: &Remote) -> Result<String, Error> {
Ok(String::new())
}
}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 07/21] pdm-config: subscriptions: rename trait methods to read/write/lock
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (5 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 06/21] pdm-config: remotes: rename trait methods to read/write/lock Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 08/21] remote iterator: pass remote config reader explicitly Lukas Wagner
` (13 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
This makes the use of these less awkward when being called via the trait
object itself, e.g. on a dependency injected application context object:
app.subscription_key_config().read()
instead of
app.subscription_key_config().config()
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
lib/pdm-config/src/subscriptions.rs | 30 ++++++++++++++---------------
1 file changed, 15 insertions(+), 15 deletions(-)
diff --git a/lib/pdm-config/src/subscriptions.rs b/lib/pdm-config/src/subscriptions.rs
index a2954418..5be88a13 100644
--- a/lib/pdm-config/src/subscriptions.rs
+++ b/lib/pdm-config/src/subscriptions.rs
@@ -35,46 +35,46 @@ fn instance() -> &'static (dyn SubscriptionKeyConfig + Send + Sync) {
}
pub fn lock_config() -> Result<ApiLockGuard, Error> {
- instance().lock_config()
+ instance().lock()
}
pub fn config() -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error> {
- instance().config()
+ instance().read()
}
pub fn shadow_config() -> Result<SectionConfigData<SubscriptionKeyShadow>, Error> {
- instance().shadow_config()
+ instance().read_shadow()
}
pub fn save_config(
config: &SectionConfigData<SubscriptionKeyEntry>,
) -> Result<ConfigDigest, Error> {
- instance().save_config(config)
+ instance().write(config)
}
pub fn save_shadow(shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error> {
- instance().save_shadow(shadow)
+ instance().write_shadow(shadow)
}
pub trait SubscriptionKeyConfig {
- fn config(&self) -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error>;
- fn shadow_config(&self) -> Result<SectionConfigData<SubscriptionKeyShadow>, Error>;
- fn lock_config(&self) -> Result<ApiLockGuard, Error>;
- fn save_config(
+ fn read(&self) -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error>;
+ fn read_shadow(&self) -> Result<SectionConfigData<SubscriptionKeyShadow>, Error>;
+ fn lock(&self) -> Result<ApiLockGuard, Error>;
+ fn write(
&self,
config: &SectionConfigData<SubscriptionKeyEntry>,
) -> Result<ConfigDigest, Error>;
- fn save_shadow(&self, shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error>;
+ fn write_shadow(&self, shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error>;
}
pub struct DefaultSubscriptionKeyConfig;
impl SubscriptionKeyConfig for DefaultSubscriptionKeyConfig {
- fn lock_config(&self) -> Result<ApiLockGuard, Error> {
+ fn lock(&self) -> Result<ApiLockGuard, Error> {
open_api_lockfile(SUBSCRIPTIONS_CFG_LOCKFILE, None, true)
}
- fn config(&self) -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error> {
+ fn read(&self) -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error> {
let content = proxmox_sys::fs::file_read_optional_string(SUBSCRIPTIONS_CFG_FILENAME)?
.unwrap_or_default();
@@ -85,13 +85,13 @@ impl SubscriptionKeyConfig for DefaultSubscriptionKeyConfig {
Ok((data, digest.into()))
}
- fn shadow_config(&self) -> Result<SectionConfigData<SubscriptionKeyShadow>, Error> {
+ fn read_shadow(&self) -> Result<SectionConfigData<SubscriptionKeyShadow>, Error> {
let content = proxmox_sys::fs::file_read_optional_string(SUBSCRIPTIONS_SHADOW_FILENAME)?
.unwrap_or_default();
SubscriptionKeyShadow::parse_section_config(SUBSCRIPTIONS_SHADOW_FILENAME, &content)
}
- fn save_config(
+ fn write(
&self,
config: &SectionConfigData<SubscriptionKeyEntry>,
) -> Result<ConfigDigest, Error> {
@@ -101,7 +101,7 @@ impl SubscriptionKeyConfig for DefaultSubscriptionKeyConfig {
Ok(digest)
}
- fn save_shadow(&self, shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error> {
+ fn write_shadow(&self, shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error> {
let raw =
SubscriptionKeyShadow::write_section_config(SUBSCRIPTIONS_SHADOW_FILENAME, shadow)?;
// Signed `SubscriptionInfo` blobs are secrets - mode 0600, priv:priv, so the
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 08/21] remote iterator: pass remote config reader explicitly
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (6 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 07/21] pdm-config: subscriptions: " Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 09/21] context: introduce a ContextFactory to build application context Lukas Wagner
` (12 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
This will be useful later once we get a reference to the remote config
implementation via the application context object. Also, it gently
nudges developers to think about the previously hidden dependency.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
lib/pdm-config/src/remotes.rs | 2 +-
server/src/api/pbs/mod.rs | 10 ++++------
server/src/api/pve/firewall.rs | 3 ++-
server/src/api/pve/mod.rs | 3 ++-
server/src/api/remotes/mod.rs | 5 +++--
server/src/api/remotes/updates.rs | 3 ++-
server/src/api/sdn/controllers.rs | 2 +-
server/src/api/sdn/vnets.rs | 2 +-
server/src/api/sdn/zones.rs | 2 +-
server/src/api/subscriptions/mod.rs | 10 ++++++----
10 files changed, 23 insertions(+), 19 deletions(-)
diff --git a/lib/pdm-config/src/remotes.rs b/lib/pdm-config/src/remotes.rs
index 47754e9b..4db6be20 100644
--- a/lib/pdm-config/src/remotes.rs
+++ b/lib/pdm-config/src/remotes.rs
@@ -21,7 +21,7 @@ pub const REMOTES_CFG_LOCKFILE: &str = configdir!("/.remotes.lock");
static INSTANCE: OnceLock<Box<dyn RemoteConfig + Send + Sync>> = OnceLock::new();
-fn instance() -> &'static (dyn RemoteConfig + Send + Sync) {
+pub fn instance() -> &'static (dyn RemoteConfig + Send + Sync) {
// Not initializing the remote config instance is
// entirely in our responsibility and not something we can recover from,
// so it should be okay to panic in this case.
diff --git a/server/src/api/pbs/mod.rs b/server/src/api/pbs/mod.rs
index 1fc75c34..41ee2c08 100644
--- a/server/src/api/pbs/mod.rs
+++ b/server/src/api/pbs/mod.rs
@@ -13,11 +13,9 @@ use pdm_api_types::{
Authid, HOST_OPTIONAL_PORT_FORMAT, PRIV_RESOURCE_AUDIT, PRIV_SYS_MODIFY, RemoteUpid,
};
-use crate::{
- connection::{self, probe_tls_connection},
- pbs_client::{self, PbsClient, get_remote},
-};
-
+use crate::api::remotes::RemoteIterator;
+use crate::connection::{self, probe_tls_connection};
+use crate::pbs_client::{self, PbsClient, get_remote};
use crate::remote_tasks;
mod node;
@@ -96,7 +94,7 @@ pub async fn new_remote_upid(
)]
/// Return the list of PBS remotes
fn list_remotes() -> Result<Vec<RemoteListEntry>, Error> {
- Ok(super::remotes::RemoteIterator::new()?
+ Ok(RemoteIterator::new(pdm_config::remotes::instance())?
.remote_type(RemoteType::Pbs)
.into_names()
.map(|name| RemoteListEntry { remote: name })
diff --git a/server/src/api/pve/firewall.rs b/server/src/api/pve/firewall.rs
index 5a6a209e..b381a14e 100644
--- a/server/src/api/pve/firewall.rs
+++ b/server/src/api/pve/firewall.rs
@@ -17,6 +17,7 @@ use pdm_api_types::{NODE_SCHEMA, VMID_SCHEMA};
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;
@@ -249,7 +250,7 @@ async fn fetch_node_firewall_status(
pub async fn pve_firewall_status(
_rpcenv: &mut dyn RpcEnvironment,
) -> Result<Vec<RemoteFirewallStatus>, Error> {
- let pve_remotes: Vec<Remote> = crate::api::remotes::RemoteIterator::new()?
+ let pve_remotes: Vec<Remote> = RemoteIterator::new(pdm_config::remotes::instance())?
.remote_type(pdm_api_types::remotes::RemoteType::Pve)
.into_remotes()
.collect();
diff --git a/server/src/api/pve/mod.rs b/server/src/api/pve/mod.rs
index 0970f2ff..4f6f7215 100644
--- a/server/src/api/pve/mod.rs
+++ b/server/src/api/pve/mod.rs
@@ -30,6 +30,7 @@ use pve_api_types::{ClusterResourceKind, ClusterResourceType};
use super::resources::{map_pve_lxc, map_pve_node, map_pve_qemu, map_pve_storage};
+use crate::api::remotes::RemoteIterator;
use crate::connection::PveClient;
use crate::connection::{self, probe_tls_connection};
use crate::remote_tasks;
@@ -142,7 +143,7 @@ pub fn connect_to_remote_by_id(id: &str) -> Result<Arc<PveClient>, Error> {
)]
/// Return the list of PVE remotes
fn list_remotes() -> Result<Vec<RemoteListEntry>, Error> {
- Ok(super::remotes::RemoteIterator::new()?
+ Ok(RemoteIterator::new(pdm_config::remotes::instance())?
.remote_type(RemoteType::Pve)
.into_names()
.map(|name| RemoteListEntry { remote: name })
diff --git a/server/src/api/remotes/mod.rs b/server/src/api/remotes/mod.rs
index 5c9fdccf..d039e6da 100644
--- a/server/src/api/remotes/mod.rs
+++ b/server/src/api/remotes/mod.rs
@@ -4,6 +4,7 @@ use std::collections::HashSet;
use std::error::Error as _;
use anyhow::{Context, Error, bail, format_err};
+use pdm_config::remotes::RemoteConfig;
use serde::{Deserialize, Serialize};
use proxmox_access_control::CachedUserInfo;
@@ -129,8 +130,8 @@ pub struct RemoteIterator {
impl RemoteIterator {
/// Load the remote config and create an unfiltered iterator.
- pub fn new() -> Result<Self, Error> {
- let (config, _) = pdm_config::remotes::config()?;
+ pub fn new(config_impl: &dyn RemoteConfig) -> Result<Self, Error> {
+ let (config, _) = config_impl.read()?;
Ok(Self {
remotes: config.into_iter().collect(),
})
diff --git a/server/src/api/remotes/updates.rs b/server/src/api/remotes/updates.rs
index d5b6d2dd..0945478b 100644
--- a/server/src/api/remotes/updates.rs
+++ b/server/src/api/remotes/updates.rs
@@ -17,6 +17,7 @@ use proxmox_router::{
use proxmox_schema::api;
use proxmox_sortable_macro::sortable;
+use crate::api::remotes::RemoteIterator;
use crate::{connection, remote_updates};
use super::get_remote;
@@ -88,7 +89,7 @@ pub fn refresh_remote_update_summaries(rpcenv: &mut dyn RpcEnvironment) -> Resul
http_bail!(FORBIDDEN, "user has no access to resources");
}
- let remotes: Vec<Remote> = super::RemoteIterator::new()?
+ let remotes: Vec<Remote> = RemoteIterator::new(pdm_config::remotes::instance())?
.all_privs(&user_info, &auth_id, PRIV_RESOURCE_MODIFY)
.into_remotes()
.collect();
diff --git a/server/src/api/sdn/controllers.rs b/server/src/api/sdn/controllers.rs
index c90e44c9..060fef72 100644
--- a/server/src/api/sdn/controllers.rs
+++ b/server/src/api/sdn/controllers.rs
@@ -75,7 +75,7 @@ pub async fn list_controllers(
http_bail!(FORBIDDEN, "user has no access to resources");
}
- let mut iter = RemoteIterator::new()?
+ let mut iter = RemoteIterator::new(pdm_config::remotes::instance())?
.remote_type(RemoteType::Pve)
.any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT);
if let Some(ref filter) = remotes {
diff --git a/server/src/api/sdn/vnets.rs b/server/src/api/sdn/vnets.rs
index ede7b539..30d431bc 100644
--- a/server/src/api/sdn/vnets.rs
+++ b/server/src/api/sdn/vnets.rs
@@ -76,7 +76,7 @@ async fn list_vnets(
http_bail!(FORBIDDEN, "user has no access to resources");
}
- let mut iter = RemoteIterator::new()?
+ let mut iter = RemoteIterator::new(pdm_config::remotes::instance())?
.remote_type(RemoteType::Pve)
.any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT);
if let Some(ref filter) = remotes {
diff --git a/server/src/api/sdn/zones.rs b/server/src/api/sdn/zones.rs
index da471ba9..d37dc46c 100644
--- a/server/src/api/sdn/zones.rs
+++ b/server/src/api/sdn/zones.rs
@@ -82,7 +82,7 @@ pub async fn list_zones(
http_bail!(FORBIDDEN, "user has no access to resources");
}
- let mut iter = RemoteIterator::new()?
+ let mut iter = RemoteIterator::new(pdm_config::remotes::instance())?
.remote_type(RemoteType::Pve)
.any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT);
if let Some(ref filter) = remotes {
diff --git a/server/src/api/subscriptions/mod.rs b/server/src/api/subscriptions/mod.rs
index 48b15b98..81582945 100644
--- a/server/src/api/subscriptions/mod.rs
+++ b/server/src/api/subscriptions/mod.rs
@@ -29,6 +29,7 @@ use pdm_api_types::{
Authid, NODE_SCHEMA, PRIV_RESOURCE_AUDIT, PRIV_RESOURCE_MODIFY, PRIV_SYS_AUDIT, PRIV_SYS_MODIFY,
};
+use crate::api::remotes::RemoteIterator;
use crate::api::resources::{
get_subscription_info_for_remote, invalidate_subscription_info_for_remote,
};
@@ -1431,10 +1432,11 @@ async fn collect_node_status(
.parse()?;
let user_info = CachedUserInfo::new()?;
- let visible_remotes: Vec<(String, Remote)> = crate::api::remotes::RemoteIterator::new()?
- .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT)
- .into_iter()
- .collect();
+ let visible_remotes: Vec<(String, Remote)> =
+ RemoteIterator::new(pdm_config::remotes::instance())?
+ .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT)
+ .into_iter()
+ .collect();
let (keys_config, _) = pdm_config::subscriptions::config()?;
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 09/21] context: introduce a ContextFactory to build application context
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (7 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 08/21] remote iterator: pass remote config reader explicitly Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 10/21] context: establish PdmApplication object Lukas Wagner
` (11 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
This makes it easier to selectively override behavior for the fake
remote feature, as well as integration tests.
The change from Box to Arc in the stored client factory is merely to
maintain bisectability, a future commit needs make_client_factory to
return an Arc.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
Notes:
Changes since v2:
Use the context factory to build the client factory. This required an
intermittent change from Box<...> to Arc<...> in the stored client
factory, but this is changed anyways in the following commit.
server/src/connection.rs | 4 +-
server/src/context/default.rs | 5 ++
server/src/context/faked_remotes.rs | 42 ++++++++++
server/src/context/mod.rs | 78 ++++++++++---------
.../remote_collection_task.rs | 2 +-
5 files changed, 93 insertions(+), 38 deletions(-)
create mode 100644 server/src/context/default.rs
create mode 100644 server/src/context/faked_remotes.rs
diff --git a/server/src/connection.rs b/server/src/connection.rs
index a63ea7da..140a1902 100644
--- a/server/src/connection.rs
+++ b/server/src/connection.rs
@@ -28,7 +28,7 @@ use pve_api_types::client::PveClientImpl;
use crate::pbs_client::PbsClient;
use crate::remote_cache::ConnectionState;
-static INSTANCE: OnceLock<Box<dyn ClientFactory + Send + Sync>> = OnceLock::new();
+static INSTANCE: OnceLock<Arc<dyn ClientFactory + Send + Sync>> = OnceLock::new();
/// Connection Info returned from [`prepare_connect_client`]
struct ConnectInfo {
@@ -470,7 +470,7 @@ pub async fn make_pbs_client_and_login(remote: &Remote) -> Result<Box<PbsClient<
/// Initialize the [`ClientFactory`] instance.
///
/// Will panic if the instance has already been set.
-pub fn init(instance: Box<dyn ClientFactory + Send + Sync>) {
+pub fn init(instance: Arc<dyn ClientFactory + Send + Sync>) {
if INSTANCE.set(instance).is_err() {
panic!("connection factory instance already set");
}
diff --git a/server/src/context/default.rs b/server/src/context/default.rs
new file mode 100644
index 00000000..e9c29e53
--- /dev/null
+++ b/server/src/context/default.rs
@@ -0,0 +1,5 @@
+use crate::context::ContextFactory;
+
+pub struct DefaultContextFactory;
+
+impl ContextFactory for DefaultContextFactory {}
diff --git a/server/src/context/faked_remotes.rs b/server/src/context/faked_remotes.rs
new file mode 100644
index 00000000..b2ae1c3f
--- /dev/null
+++ b/server/src/context/faked_remotes.rs
@@ -0,0 +1,42 @@
+use std::sync::Arc;
+
+use anyhow::{Context, Error};
+use pdm_config::remotes::RemoteConfig;
+
+use crate::connection::ClientFactory;
+use crate::context::ContextFactory;
+use crate::test_support::fake_remote::{FakeClientFactory, FakeRemoteConfig};
+
+pub struct FakedRemoteContextFactory(FakeRemoteConfig);
+
+impl FakedRemoteContextFactory {
+ pub fn new() -> Result<Self, Error> {
+ let path = std::env::var("PDM_FAKED_REMOTE_CONFIG").context(
+ "compiled with remote_config = 'faked', but PDM_FAKED_REMOTE_CONFIG not set",
+ )?;
+
+ log::info!("using fake remotes from {path:?}");
+ let config = FakeRemoteConfig::from_json_config(&path)
+ .context("could not deserialize fake remote config")?;
+
+ Ok(Self(config))
+ }
+}
+
+impl ContextFactory for FakedRemoteContextFactory {
+ fn make_client_factory(&self) -> Result<Arc<dyn ClientFactory + Send + Sync>, Error> {
+ Ok(Arc::new(FakeClientFactory {
+ config: self.0.clone(),
+ }))
+ }
+
+ fn make_remote_config(&self) -> Result<Box<dyn RemoteConfig + Send + Sync>, Error> {
+ Ok(Box::new(self.0.clone()))
+ }
+
+ // No need to override subscription_key_config_impl here.
+ //
+ // The subscription key pool is product-only (PDM stores its own pool of
+ // keys regardless of how remotes are mocked or not), so initialise it on
+ // both paths.
+}
diff --git a/server/src/context/mod.rs b/server/src/context/mod.rs
index a4afcddd..77c84f98 100644
--- a/server/src/context/mod.rs
+++ b/server/src/context/mod.rs
@@ -2,48 +2,56 @@
//!
//! Make sure to call `init` *once* when starting up the API server.
+use std::sync::Arc;
+
use anyhow::Error;
+use pdm_config::{remotes::RemoteConfig, subscriptions::SubscriptionKeyConfig};
-use crate::connection;
+use crate::connection::{self, ClientFactory};
-/// Dependency-inject production remote-config implementation and remote client factory
-#[allow(dead_code)]
-fn default_remote_setup() {
- pdm_config::remotes::init(Box::new(pdm_config::remotes::DefaultRemoteConfig));
- connection::init(Box::new(connection::DefaultClientFactory));
-}
+#[cfg(remote_config = "faked")]
+mod faked_remotes;
+
+#[cfg(not(remote_config = "faked"))]
+mod default;
/// Dependency-inject concrete implementations needed at runtime.
pub fn init() -> Result<(), Error> {
- // The subscription key pool is product-only (PDM stores its own pool of
- // keys regardless of how remotes are mocked or not), so initialise it on
- // both paths.
- pdm_config::subscriptions::init(Box::new(
- pdm_config::subscriptions::DefaultSubscriptionKeyConfig,
- ));
+ let factory = context_factory()?;
- #[cfg(remote_config = "faked")]
- {
- use anyhow::bail;
-
- use crate::test_support::fake_remote;
-
- match std::env::var("PDM_FAKED_REMOTE_CONFIG") {
- Ok(path) => {
- log::info!("using fake remotes from {path:?}");
- let config = fake_remote::FakeRemoteConfig::from_json_config(&path)?;
- pdm_config::remotes::init(Box::new(config.clone()));
- connection::init(Box::new(fake_remote::FakeClientFactory { config }));
- }
- Err(_) => {
- bail!("compiled with remote_config = 'faked', but PDM_FAKED_REMOTE_CONFIG not set")
- }
- }
- }
- #[cfg(not(remote_config = "faked"))]
- {
- default_remote_setup();
- }
+ pdm_config::subscriptions::init(factory.make_subscription_key_config()?);
+ pdm_config::remotes::init(factory.make_remote_config()?);
+ // FIXME: Rather let connection use an Application context object from here
+ connection::init(factory.make_client_factory()?);
Ok(())
}
+
+pub trait ContextFactory {
+ fn make_client_factory(&self) -> Result<Arc<dyn ClientFactory + Send + Sync>, Error> {
+ Ok(Arc::new(connection::DefaultClientFactory))
+ }
+
+ fn make_remote_config(&self) -> Result<Box<dyn RemoteConfig + Send + Sync>, Error> {
+ Ok(Box::new(pdm_config::remotes::DefaultRemoteConfig))
+ }
+
+ fn make_subscription_key_config(
+ &self,
+ ) -> Result<Box<dyn SubscriptionKeyConfig + Send + Sync>, Error> {
+ Ok(Box::new(
+ pdm_config::subscriptions::DefaultSubscriptionKeyConfig,
+ ))
+ }
+}
+
+fn context_factory() -> Result<impl ContextFactory, Error> {
+ #[cfg(remote_config = "faked")]
+ {
+ faked_remotes::FakedRemoteContextFactory::new()
+ }
+ #[cfg(not(remote_config = "faked"))]
+ {
+ Ok(default::DefaultContextFactory)
+ }
+}
diff --git a/server/src/metric_collection/remote_collection_task.rs b/server/src/metric_collection/remote_collection_task.rs
index d243dcf1..9fe67371 100644
--- a/server/src/metric_collection/remote_collection_task.rs
+++ b/server/src/metric_collection/remote_collection_task.rs
@@ -558,7 +558,7 @@ pub(super) mod tests {
// TODO: the client factory is currently stored in a OnceLock -
// we can only set it from one test... Ideally we'd like to have the
// option to set it in every single test if needed - task/thread local?
- connection::init(Box::new(TestClientFactory { now }));
+ connection::init(Arc::new(TestClientFactory { now }));
});
now
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 10/21] context: establish PdmApplication object
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (8 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 09/21] context: introduce a ContextFactory to build application context Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 11/21] context: register PdmApplication in router Lukas Wagner
` (10 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
PdmApplication bundles the client factory, remote config, subscription
key config, and product config behind a single cloneable handle. This
replaces the growing set of independent global statics with one object
that can be assembled differently for production, the fake-remote
feature, and integration tests.
It is reachable through a new context::pdm_application() accessor, built
by the same context::init() call that already seeds the existing
remote/subscription config globals. Later commits in this series inject
it through the API handlers via the router's shared state instead of
this accessor.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
server/src/context/mod.rs | 102 ++++++++++++++++++++++++++++++++++++--
1 file changed, 98 insertions(+), 4 deletions(-)
diff --git a/server/src/context/mod.rs b/server/src/context/mod.rs
index 77c84f98..3e89bceb 100644
--- a/server/src/context/mod.rs
+++ b/server/src/context/mod.rs
@@ -2,9 +2,12 @@
//!
//! Make sure to call `init` *once* when starting up the API server.
-use std::sync::Arc;
+use std::sync::{Arc, OnceLock};
use anyhow::Error;
+
+use proxmox_product_config::{ProductConfig, ProductConfigParams};
+
use pdm_config::{remotes::RemoteConfig, subscriptions::SubscriptionKeyConfig};
use crate::connection::{self, ClientFactory};
@@ -15,16 +18,36 @@ mod faked_remotes;
#[cfg(not(remote_config = "faked"))]
mod default;
+static APP: OnceLock<PdmApplication> = OnceLock::new();
+
/// Dependency-inject concrete implementations needed at runtime.
-pub fn init() -> Result<(), Error> {
+pub fn init() -> Result<PdmApplication, Error> {
let factory = context_factory()?;
- pdm_config::subscriptions::init(factory.make_subscription_key_config()?);
+ let app = factory.make_pdm_application()?;
+ APP.set(app.clone())
+ .map_err(|_| anyhow::format_err!("context::init was already called"))?;
+
+ // NOTE: This is technically a second, independent instance of the remote config.
+ // Long-term, we'd like to get rid of the global instance handle in pdm_config::remotes
+ // anyway, and the implementation is stateless, so having this second
+ // instance is not an issue *currently*.
pdm_config::remotes::init(factory.make_remote_config()?);
+ pdm_config::subscriptions::init(factory.make_subscription_key_config()?);
+
// FIXME: Rather let connection use an Application context object from here
connection::init(factory.make_client_factory()?);
- Ok(())
+ Ok(app)
+}
+
+/// Retrieve a handle to [`PdmApplication`] for this server.
+///
+/// Prefer to retrieve this via the API handler using [`proxmox_router::State`].
+pub fn pdm_application() -> PdmApplication {
+ APP.get()
+ .expect("context::init was not called to set up the application context object")
+ .clone()
}
pub trait ContextFactory {
@@ -43,6 +66,30 @@ pub trait ContextFactory {
pdm_config::subscriptions::DefaultSubscriptionKeyConfig,
))
}
+
+ fn make_product_config(&self) -> Result<ProductConfig, Error> {
+ let product_config = ProductConfig::new(ProductConfigParams {
+ api_user: pdm_config::api_user()?,
+ priv_user: pdm_config::priv_user()?,
+ config_dir: pdm_buildcfg::configdir!("/").into(),
+ state_dir: pdm_buildcfg::statedir!("/").into(),
+ run_dir: pdm_buildcfg::rundir!("/").into(),
+ cache_dir: pdm_buildcfg::PDM_CACHE_DIR.into(),
+ });
+
+ Ok(product_config)
+ }
+
+ fn make_pdm_application(&self) -> Result<PdmApplication, Error> {
+ Ok(PdmApplication {
+ inner: Arc::new(PdmApplicationInner {
+ client_factory: self.make_client_factory()?,
+ remote_config: self.make_remote_config()?,
+ subscription_key_config: self.make_subscription_key_config()?,
+ product_config: self.make_product_config()?,
+ }),
+ })
+ }
}
fn context_factory() -> Result<impl ContextFactory, Error> {
@@ -55,3 +102,50 @@ fn context_factory() -> Result<impl ContextFactory, Error> {
Ok(default::DefaultContextFactory)
}
}
+
+/// Application context handle.
+///
+/// This type gives access to dependency-injected implementations and general product
+/// configuration.
+///
+/// This implements [`Clone`] and can be cheaply copied (it contains a single `Arc`).
+#[derive(Clone)]
+pub struct PdmApplication {
+ inner: Arc<PdmApplicationInner>,
+}
+
+impl PdmApplication {
+ /// Get a handle to the [`ClientFactory`] as an [`Arc`].
+ ///
+ /// Prefer to use [`Self::client_factory`] if possible.
+ pub fn client_factory_shared(&self) -> Arc<dyn ClientFactory + Send + Sync> {
+ Arc::clone(&self.inner.client_factory)
+ }
+
+ /// Get a reference to the [`ClientFactory`].
+ pub fn client_factory(&self) -> &(dyn ClientFactory + Send + Sync) {
+ self.inner.client_factory.as_ref()
+ }
+
+ /// Get a reference to the [`RemoteConfig`].
+ pub fn remote_config(&self) -> &(dyn RemoteConfig + Send + Sync) {
+ self.inner.remote_config.as_ref()
+ }
+
+ /// Get a reference to the [`SubscriptionKeyConfig`].
+ pub fn subscription_key_config(&self) -> &(dyn SubscriptionKeyConfig + Send + Sync) {
+ self.inner.subscription_key_config.as_ref()
+ }
+
+ /// Get a reference to the [`ProductConfig`].
+ pub fn product_config(&self) -> &ProductConfig {
+ &self.inner.product_config
+ }
+}
+
+struct PdmApplicationInner {
+ client_factory: Arc<dyn ClientFactory + Send + Sync>,
+ remote_config: Box<dyn RemoteConfig + Send + Sync>,
+ subscription_key_config: Box<dyn SubscriptionKeyConfig + Send + Sync>,
+ product_config: ProductConfig,
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 11/21] context: register PdmApplication in router
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (9 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 10/21] context: establish PdmApplication object Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 12/21] connection: use client factory from PdmApplication handle Lukas Wagner
` (9 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
Register the PdmApplication handle in a SharedStateRegistry and pass
it to the REST server config via with_shared_state(), so API handlers
can request it through the State<PdmApplication> extractor instead of
reaching for context::pdm_application() or other globals directly.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
cli/admin/src/main.rs | 10 ++++++++--
server/src/bin/proxmox-datacenter-api/main.rs | 15 ++++++++++-----
.../src/bin/proxmox-datacenter-privileged-api.rs | 15 ++++++++++-----
3 files changed, 28 insertions(+), 12 deletions(-)
diff --git a/cli/admin/src/main.rs b/cli/admin/src/main.rs
index a2c5c623..a514d2fc 100644
--- a/cli/admin/src/main.rs
+++ b/cli/admin/src/main.rs
@@ -3,15 +3,17 @@ use core::matches;
use anyhow::{Context, Error};
use serde_json::{Value, json};
-use proxmox_router::RpcEnvironment;
use proxmox_router::cli::{
CliCommand, CliCommandMap, CliEnvironment, ColumnConfig, OUTPUT_FORMAT,
default_table_format_options, format_and_print_result_full, get_output_format,
run_async_cli_command,
};
+use proxmox_router::{RpcEnvironment, SharedStateRegistry};
use proxmox_schema::api;
use proxmox_sys::fs::CreateOptions;
+use server::context::PdmApplication;
+
mod acme;
mod cert;
mod remotes;
@@ -36,7 +38,7 @@ async fn run() -> Result<(), Error> {
.init()
.context("failed to set up logger")?;
- server::context::init().context("could not set up server context")?;
+ let app = server::context::init().context("could not set up server context")?;
let cmd_def = CliCommandMap::new()
.insert("acme", acme::acme_mgmt_cli())
@@ -68,7 +70,11 @@ async fn run() -> Result<(), Error> {
.context("failed to activate the socket")?;
}
+ let mut shared_state = SharedStateRegistry::default();
+ shared_state.register::<PdmApplication>(app)?;
+
let mut rpcenv = CliEnvironment::new();
+ rpcenv.set_shared_state_registry(shared_state);
rpcenv.set_auth_id(Some("root@pam".into()));
run_async_cli_command(cmd_def, rpcenv).await;
diff --git a/server/src/bin/proxmox-datacenter-api/main.rs b/server/src/bin/proxmox-datacenter-api/main.rs
index 57aa6e22..64326b9b 100644
--- a/server/src/bin/proxmox-datacenter-api/main.rs
+++ b/server/src/bin/proxmox-datacenter-api/main.rs
@@ -18,7 +18,7 @@ use url::form_urlencoded;
use proxmox_lang::try_block;
use proxmox_rest_server::{ApiConfig, RestEnvironment, RestServer};
-use proxmox_router::{RpcEnvironment, RpcEnvironmentType};
+use proxmox_router::{RpcEnvironment, RpcEnvironmentType, SharedStateRegistry};
use proxmox_sys::fs::CreateOptions;
use pdm_buildcfg::configdir;
@@ -28,6 +28,7 @@ use proxmox_auth_api::api::assemble_csrf_prevention_token;
use server::auth;
use server::auth::csrf::csrf_secret;
+use server::context::PdmApplication;
use server::metric_collection;
use server::resource_cache;
use server::task_utils;
@@ -67,9 +68,9 @@ fn main() -> Result<(), Error> {
}
proxmox_product_config::init(pdm_config::api_user()?, pdm_config::priv_user()?);
- server::context::init()?;
+ let app = server::context::init()?;
- proxmox_async::runtime::main(run(debug))
+ proxmox_async::runtime::main(run(app, debug))
}
async fn get_index_future(env: RestEnvironment, parts: Parts) -> Response<proxmox_http::Body> {
@@ -144,7 +145,7 @@ async fn get_index_future(env: RestEnvironment, parts: Parts) -> Response<proxmo
resp
}
-async fn run(debug: bool) -> Result<(), Error> {
+async fn run(app: PdmApplication, debug: bool) -> Result<(), Error> {
auth::init(false);
proxmox_acme_api::init(configdir!("/acme"), false)?;
@@ -159,6 +160,9 @@ async fn run(debug: bool) -> Result<(), Error> {
let indexpath = Path::new(pdm_buildcfg::JS_DIR).join("index.hbs");
+ let mut shared_state = SharedStateRegistry::default();
+ shared_state.register::<PdmApplication>(app.clone())?;
+
let config = ApiConfig::new(pdm_buildcfg::JS_DIR, RpcEnvironmentType::PUBLIC)
.privileged_addr(
std::os::unix::net::SocketAddr::from_pathname(
@@ -194,7 +198,8 @@ async fn run(debug: bool) -> Result<(), Error> {
Some(dir_opts),
Some(file_opts),
&mut command_sock,
- )?;
+ )?
+ .with_shared_state(shared_state);
let rest_server = RestServer::new(config);
let redirector = proxmox_rest_server::Redirector::new();
diff --git a/server/src/bin/proxmox-datacenter-privileged-api.rs b/server/src/bin/proxmox-datacenter-privileged-api.rs
index 59d30513..3815ab03 100644
--- a/server/src/bin/proxmox-datacenter-privileged-api.rs
+++ b/server/src/bin/proxmox-datacenter-privileged-api.rs
@@ -11,9 +11,10 @@ use tracing::level_filters::LevelFilter;
use proxmox_lang::try_block;
use proxmox_rest_server::{ApiConfig, RestServer};
-use proxmox_router::RpcEnvironmentType;
+use proxmox_router::{RpcEnvironmentType, SharedStateRegistry};
use proxmox_sys::fs::CreateOptions;
+use server::context::PdmApplication;
use server::{api_cache, auth};
use pdm_buildcfg::configdir;
@@ -54,9 +55,9 @@ fn main() -> Result<(), Error> {
}
}
- server::context::init()?;
+ let app = server::context::init()?;
- proxmox_async::runtime::main(run())
+ proxmox_async::runtime::main(run(app))
}
fn create_directories() -> Result<(), Error> {
@@ -114,7 +115,7 @@ fn create_directories() -> Result<(), Error> {
Ok(())
}
-async fn run() -> Result<(), Error> {
+async fn run(app: PdmApplication) -> Result<(), Error> {
auth::init(true);
proxmox_acme_api::init(configdir!("/acme"), true)?;
@@ -139,6 +140,9 @@ async fn run() -> Result<(), Error> {
let dir_opts = CreateOptions::new().owner(api_user.uid).group(api_user.gid);
let file_opts = CreateOptions::new().owner(api_user.uid).group(api_user.gid);
+ let mut shared_state = SharedStateRegistry::default();
+ shared_state.register::<PdmApplication>(app)?;
+
let config = ApiConfig::new(pdm_buildcfg::JS_DIR, RpcEnvironmentType::PRIVILEGED)
.auth_handler_func(|h, m| Box::pin(auth::check_auth(h, m)))
.formatted_router(&["api2"], &server::api::ROUTER)
@@ -153,7 +157,8 @@ async fn run() -> Result<(), Error> {
Some(dir_opts),
Some(file_opts),
&mut command_sock,
- )?;
+ )?
+ .with_shared_state(shared_state);
let rest_server = RestServer::new(config);
proxmox_rest_server::init_worker_tasks(pdm_buildcfg::PDM_LOG_DIR_M!().into(), file_opts)?;
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 12/21] connection: use client factory from PdmApplication handle
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (10 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 11/21] context: register PdmApplication in router Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 13/21] parallel fetcher: pass arguments to closure in a single type Lukas Wagner
` (8 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
The `connection` module now does not store a handle to the client
factory any more, instead the free-standing helpers access the client
factory instead via context::pdm_application(). Long-term, the
free-standing helpers should be removed anyway, and the client factory
only be accessed through the app handle.
The change in `connection` required some adaptations to test cases for
the remote metric collection task.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
Notes:
Changes since v2:
- Use ProductConfig from proxmox-product-config instead of
implementing it locally
server/src/bin/proxmox-datacenter-api/main.rs | 2 +-
server/src/connection.rs | 54 +++++++++----------
server/src/context/mod.rs | 3 --
server/src/metric_collection/mod.rs | 6 ++-
.../remote_collection_task.rs | 47 ++++++++--------
5 files changed, 55 insertions(+), 57 deletions(-)
diff --git a/server/src/bin/proxmox-datacenter-api/main.rs b/server/src/bin/proxmox-datacenter-api/main.rs
index 64326b9b..eac1585b 100644
--- a/server/src/bin/proxmox-datacenter-api/main.rs
+++ b/server/src/bin/proxmox-datacenter-api/main.rs
@@ -344,7 +344,7 @@ async fn run(app: PdmApplication, debug: bool) -> Result<(), Error> {
});
start_task_scheduler();
- metric_collection::start_task()?;
+ metric_collection::start_task(app)?;
tasks::remote_node_mapping::start_task();
resource_cache::start_task();
tasks::remote_tasks::start_task()?;
diff --git a/server/src/connection.rs b/server/src/connection.rs
index 140a1902..df122836 100644
--- a/server/src/connection.rs
+++ b/server/src/connection.rs
@@ -7,9 +7,9 @@ use std::collections::HashMap;
use std::future::Future;
use std::pin::{Pin, pin};
use std::sync::Arc;
+use std::sync::LazyLock;
use std::sync::Mutex as StdMutex;
use std::sync::Once;
-use std::sync::{LazyLock, OnceLock};
use std::time::{Duration, SystemTime};
use anyhow::{Error, bail, format_err};
@@ -25,11 +25,10 @@ use proxmox_time::epoch_i64;
use pdm_api_types::remotes::{NodeUrl, Remote, RemoteType, TlsProbeOutcome};
use pve_api_types::client::PveClientImpl;
+use crate::context;
use crate::pbs_client::PbsClient;
use crate::remote_cache::ConnectionState;
-static INSTANCE: OnceLock<Arc<dyn ClientFactory + Send + Sync>> = OnceLock::new();
-
/// Connection Info returned from [`prepare_connect_client`]
struct ConnectInfo {
prefix: String,
@@ -404,19 +403,11 @@ impl ClientFactory for DefaultClientFactory {
}
}
-fn instance() -> &'static (dyn ClientFactory + Send + Sync) {
- // Not initializing the connection factory instance is
- // entirely in our responsibility and not something we can recover from,
- // so it should be okay to panic in this case.
- INSTANCE
- .get()
- .expect("client factory instance not set")
- .as_ref()
-}
-
/// Create a new API client for PVE remotes
pub fn make_pve_client(remote: &Remote) -> Result<Arc<PveClient>, Error> {
- instance().make_pve_client(remote)
+ context::pdm_application()
+ .client_factory()
+ .make_pve_client(remote)
}
/// Create a new API client for PVE remotes, but for a specific endpoint
@@ -424,21 +415,29 @@ pub fn make_pve_client_with_endpoint(
remote: &Remote,
target_endpoint: Option<&str>,
) -> Result<Arc<PveClient>, Error> {
- instance().make_pve_client_with_endpoint(remote, target_endpoint)
+ context::pdm_application()
+ .client_factory()
+ .make_pve_client_with_endpoint(remote, target_endpoint)
}
/// Create a new API client for PVE remotes and try to make it connect to a specific *node*.
pub fn make_pve_client_with_node(remote: &Remote, node: &str) -> Result<Arc<PveClient>, Error> {
- instance().make_pve_client_with_node(remote, node)
+ context::pdm_application()
+ .client_factory()
+ .make_pve_client_with_node(remote, node)
}
/// Create a new API client for PBS remotes
pub fn make_pbs_client(remote: &Remote) -> Result<Box<PbsClient>, Error> {
- instance().make_pbs_client(remote)
+ context::pdm_application()
+ .client_factory()
+ .make_pbs_client(remote)
}
pub fn make_raw_client(remote: &Remote) -> Result<Box<Client>, Error> {
- instance().make_raw_client(remote)
+ context::pdm_application()
+ .client_factory()
+ .make_raw_client(remote)
}
/// Create a new API client for PVE remotes.
@@ -451,7 +450,10 @@ pub fn make_raw_client(remote: &Remote) -> Result<Box<Client>, Error> {
///
/// Note: currently does not support two factor authentication.
pub async fn make_pve_client_and_login(remote: &Remote) -> Result<Arc<PveClient>, Error> {
- instance().make_pve_client_and_login(remote).await
+ context::pdm_application()
+ .client_factory()
+ .make_pve_client_and_login(remote)
+ .await
}
/// Create a new API client for PBS remotes.
@@ -464,16 +466,10 @@ pub async fn make_pve_client_and_login(remote: &Remote) -> Result<Arc<PveClient>
///
/// Note: currently does not support two factor authentication.
pub async fn make_pbs_client_and_login(remote: &Remote) -> Result<Box<PbsClient<Client>>, Error> {
- instance().make_pbs_client_and_login(remote).await
-}
-
-/// Initialize the [`ClientFactory`] instance.
-///
-/// Will panic if the instance has already been set.
-pub fn init(instance: Arc<dyn ClientFactory + Send + Sync>) {
- if INSTANCE.set(instance).is_err() {
- panic!("connection factory instance already set");
- }
+ context::pdm_application()
+ .client_factory()
+ .make_pbs_client_and_login(remote)
+ .await
}
/// In order to allow the [`MultiClient`] to check the cached reachability state of a client, we
diff --git a/server/src/context/mod.rs b/server/src/context/mod.rs
index 3e89bceb..f0853d65 100644
--- a/server/src/context/mod.rs
+++ b/server/src/context/mod.rs
@@ -35,9 +35,6 @@ pub fn init() -> Result<PdmApplication, Error> {
pdm_config::remotes::init(factory.make_remote_config()?);
pdm_config::subscriptions::init(factory.make_subscription_key_config()?);
- // FIXME: Rather let connection use an Application context object from here
- connection::init(factory.make_client_factory()?);
-
Ok(app)
}
diff --git a/server/src/metric_collection/mod.rs b/server/src/metric_collection/mod.rs
index 3d7a477d..15b01b9a 100644
--- a/server/src/metric_collection/mod.rs
+++ b/server/src/metric_collection/mod.rs
@@ -20,6 +20,7 @@ pub mod top_entities;
use remote_collection_task::{ControlMsg, RemoteMetricCollectionTask};
use rrd_cache::RrdCache;
+use crate::context::PdmApplication;
use crate::metric_collection::local_collection_task::LocalMetricCollectionTask;
const RRD_CACHE_BASEDIR: &str = concat!(PDM_STATE_DIR_M!(), "/rrdb");
@@ -39,7 +40,7 @@ pub fn init() -> Result<(), Error> {
}
/// Start the metric collection task.
-pub fn start_task() -> Result<(), Error> {
+pub fn start_task(app: PdmApplication) -> Result<(), Error> {
let (metric_data_tx, metric_data_rx) = mpsc::channel(128);
let cache = rrd_cache::get_cache();
@@ -57,7 +58,8 @@ pub fn start_task() -> Result<(), Error> {
let metric_data_tx_clone = metric_data_tx.clone();
tokio::spawn(async move {
let metric_collection_task_future = pin!(async move {
- match RemoteMetricCollectionTask::new(metric_data_tx_clone, trigger_collection_rx) {
+ match RemoteMetricCollectionTask::new(app, metric_data_tx_clone, trigger_collection_rx)
+ {
Ok(mut task) => task.run().await,
Err(err) => log::error!("could not start metric collection task: {err}"),
}
diff --git a/server/src/metric_collection/remote_collection_task.rs b/server/src/metric_collection/remote_collection_task.rs
index 9fe67371..57910d46 100644
--- a/server/src/metric_collection/remote_collection_task.rs
+++ b/server/src/metric_collection/remote_collection_task.rs
@@ -19,8 +19,9 @@ use proxmox_sys::fs::CreateOptions;
use pdm_api_types::remotes::{Remote, RemoteType};
+use crate::context::PdmApplication;
use crate::metric_collection::rrd_task::CollectionStats;
-use crate::{connection, task_utils};
+use crate::task_utils;
use super::{
rrd_task::{RrdStoreRequest, RrdStoreResult},
@@ -48,6 +49,7 @@ pub(super) enum ControlMsg {
/// Task which periodically collects metrics from all remotes and stores
/// them in the local metrics database.
pub(super) struct RemoteMetricCollectionTask {
+ app: PdmApplication,
state: MetricCollectionState,
metric_data_tx: Sender<RrdStoreRequest>,
control_message_rx: Receiver<ControlMsg>,
@@ -56,12 +58,14 @@ pub(super) struct RemoteMetricCollectionTask {
impl RemoteMetricCollectionTask {
/// Create a new metric collection task.
pub(super) fn new(
+ app: PdmApplication,
metric_data_tx: Sender<RrdStoreRequest>,
control_message_rx: Receiver<ControlMsg>,
) -> Result<Self, Error> {
let state = load_state()?;
Ok(Self {
+ app,
state,
metric_data_tx,
control_message_rx,
@@ -234,9 +238,12 @@ impl RemoteMetricCollectionTask {
// called on the semaphore.
let permit = Arc::clone(&semaphore).acquire_owned().await.unwrap();
+ let app_clone = self.app.clone();
+
if let Some(remote) = remote_config.get(remote_name).cloned() {
log::debug!("fetching remote '{}'", remote.id);
handles.spawn(Self::fetch_single_remote(
+ app_clone,
remote,
status,
self.metric_data_tx.clone(),
@@ -294,6 +301,7 @@ impl RemoteMetricCollectionTask {
/// Fetch a single remote.
#[tracing::instrument(skip_all, fields(remote = remote.id), name = "metric_collection_task")]
async fn fetch_single_remote(
+ app: PdmApplication,
remote: Remote,
mut status: RemoteStatus,
sender: Sender<RrdStoreRequest>,
@@ -307,7 +315,7 @@ impl RemoteMetricCollectionTask {
let res: Result<RrdStoreResult, Error> = async {
match remote.ty {
RemoteType::Pve => {
- let client = connection::make_pve_client(&remote)?;
+ let client = app.client_factory().make_pve_client(&remote)?;
let metrics = client
.cluster_metrics_export(
Some(true),
@@ -331,7 +339,7 @@ impl RemoteMetricCollectionTask {
.await?;
}
RemoteType::Pbs => {
- let client = connection::make_pbs_client(&remote)?;
+ let client = app.client_factory().make_pbs_client(&remote)?;
let metrics = client
.metrics(Some(true), Some(status.most_recent_datapoint))
.await?;
@@ -386,8 +394,6 @@ pub(super) fn load_state() -> Result<MetricCollectionState, Error> {
#[cfg(test)]
pub(super) mod tests {
- use std::sync::Once;
-
use anyhow::bail;
use http::StatusCode;
@@ -397,6 +403,7 @@ pub(super) mod tests {
use crate::{
connection::{ClientFactory, PveClient},
+ context::ContextFactory,
metric_collection::rrd_task::RrdStoreResult,
pbs_client::PbsClient,
test_support::temp::NamedTempFile,
@@ -550,25 +557,18 @@ pub(super) mod tests {
number_of_requests
}
- static START: Once = Once::new();
+ const NOW: i64 = 1000;
- fn test_init() -> i64 {
- let now = 10000;
- START.call_once(|| {
- // TODO: the client factory is currently stored in a OnceLock -
- // we can only set it from one test... Ideally we'd like to have the
- // option to set it in every single test if needed - task/thread local?
- connection::init(Arc::new(TestClientFactory { now }));
- });
+ struct TestContextFactory();
- now
+ impl ContextFactory for TestContextFactory {
+ fn make_client_factory(&self) -> Result<Arc<dyn ClientFactory + Send + Sync>, Error> {
+ Ok(Arc::new(TestClientFactory { now: NOW }))
+ }
}
#[tokio::test]
async fn test_fetch_remotes_updates_state() {
- // Arrange
- let now = test_init();
-
let (tx, rx) = tokio::sync::mpsc::channel(10);
let handle = tokio::task::spawn(fake_rrd_task(rx));
@@ -579,7 +579,10 @@ pub(super) mod tests {
let (_control_tx, control_rx) = tokio::sync::mpsc::channel(10);
+ let app = TestContextFactory().make_pdm_application().unwrap();
+
let mut task = RemoteMetricCollectionTask {
+ app,
state,
metric_data_tx: tx,
control_message_rx: control_rx,
@@ -608,7 +611,7 @@ pub(super) mod tests {
);
assert_eq!(status.last_collection, None);
} else {
- assert!(now - status.most_recent_datapoint <= 10);
+ assert!(NOW - status.most_recent_datapoint <= 10);
assert!(status.error.is_none());
}
}
@@ -619,9 +622,6 @@ pub(super) mod tests {
#[tokio::test]
async fn test_fetch_overdue() {
- // Arrange
- test_init();
-
let (tx, rx) = tokio::sync::mpsc::channel(10);
let handle = tokio::task::spawn(fake_rrd_task(rx));
@@ -630,6 +630,8 @@ pub(super) mod tests {
let state_file = NamedTempFile::new(get_create_options()).unwrap();
let mut state = MetricCollectionState::new(state_file.path().into(), get_create_options());
+ let app = TestContextFactory().make_pdm_application().unwrap();
+
let now = proxmox_time::epoch_i64();
// This one should be fetched
@@ -652,6 +654,7 @@ pub(super) mod tests {
let (_control_tx, control_rx) = tokio::sync::mpsc::channel(10);
let mut task = RemoteMetricCollectionTask {
+ app,
state,
metric_data_tx: tx,
control_message_rx: control_rx,
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 13/21] parallel fetcher: pass arguments to closure in a single type
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (11 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 12/21] connection: use client factory from PdmApplication handle Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 14/21] server: migrate existing ParallelFetcher users to use PdmApplication Lukas Wagner
` (7 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
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
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 14/21] server: migrate existing ParallelFetcher users to use PdmApplication
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (12 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 13/21] parallel fetcher: pass arguments to closure in a single type Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 15/21] tests: add helpers for building API-handler-level integration tests Lukas Wagner
` (6 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
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
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 15/21] tests: add helpers for building API-handler-level integration tests
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (13 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 14/21] server: migrate existing ParallelFetcher users to use PdmApplication Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 16/21] tests: add example tests for SDN API routes Lukas Wagner
` (5 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
Unfortunately, some subsystems still require static initialization,
namely access control and worker tasks. For those we need a setup
function that is ensure to be only called once in each test binary.
Worker tasks especially require a directory in which it can place task
logs. For this we create a temporary directory, which we clean up using
libc::atexit. Unfortunately, tempfile::TempDir does not work for this
case, since we'd need to put it in a static variable, and `drop` is
never called for those.
TestApplication is essentially a builder that can be used to produce a
PdmApplication suitable for tests.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
Notes:
Changes since v1:
- don't .unwrap() in read_captured_responses
Changes since v2:
- create temporary test directories in CARGO_TARGET_TMPDIR instead of
/tmp
server/tests/common/environment.rs | 25 +++
server/tests/common/mod.rs | 64 ++++++
server/tests/common/test_application.rs | 287 ++++++++++++++++++++++++
3 files changed, 376 insertions(+)
create mode 100644 server/tests/common/environment.rs
create mode 100644 server/tests/common/mod.rs
create mode 100644 server/tests/common/test_application.rs
diff --git a/server/tests/common/environment.rs b/server/tests/common/environment.rs
new file mode 100644
index 00000000..5dad7b21
--- /dev/null
+++ b/server/tests/common/environment.rs
@@ -0,0 +1,25 @@
+use proxmox_router::RpcEnvironment;
+
+pub struct TestRpcEnvironment;
+
+impl RpcEnvironment for TestRpcEnvironment {
+ fn result_attrib_mut(&mut self) -> &mut serde_json::Value {
+ unimplemented!()
+ }
+
+ fn result_attrib(&self) -> &serde_json::Value {
+ unimplemented!()
+ }
+
+ fn env_type(&self) -> proxmox_router::RpcEnvironmentType {
+ unimplemented!()
+ }
+
+ fn set_auth_id(&mut self, _user: Option<String>) {
+ unimplemented!()
+ }
+
+ fn get_auth_id(&self) -> Option<String> {
+ Some("root@pam".to_string())
+ }
+}
diff --git a/server/tests/common/mod.rs b/server/tests/common/mod.rs
new file mode 100644
index 00000000..38697353
--- /dev/null
+++ b/server/tests/common/mod.rs
@@ -0,0 +1,64 @@
+use std::{
+ path::{Path, PathBuf},
+ sync::{Once, OnceLock},
+};
+
+use anyhow::Context;
+use serde::de::DeserializeOwned;
+
+use proxmox_sys::fs::CreateOptions;
+
+mod environment;
+mod test_application;
+
+pub use environment::TestRpcEnvironment;
+pub use test_application::*;
+
+/// Read and deserialize a previously captured API response.
+pub async fn read_captured_response<T: DeserializeOwned, P: AsRef<Path>>(
+ path: P,
+) -> Result<T, proxmox_client::Error> {
+ let s = tokio::fs::read_to_string(path.as_ref())
+ .await
+ .with_context(|| format!("could not read from {path}", path = path.as_ref().display()))
+ .map_err(proxmox_client::Error::Anyhow)?;
+
+ serde_json::from_str(&s).map_err(|e| proxmox_client::Error::Anyhow(e.into()))
+}
+
+static STATIC_TEMP_DIR: OnceLock<PathBuf> = OnceLock::new();
+
+extern "C" fn cleanup() {
+ if let Some(dir) = STATIC_TEMP_DIR.get() {
+ let _ = std::fs::remove_dir_all(dir);
+ }
+}
+
+/// Run commonly-used setup steps that must only run *once*.
+///
+/// Be sure to call this at the start of every test.
+pub fn test_setup() {
+ static INIT: Once = Once::new();
+
+ INIT.call_once(|| {
+ let base = env!("CARGO_TARGET_TMPDIR");
+
+ let file_opts = CreateOptions::new();
+
+ let dir = proxmox_sys::fs::make_tmp_dir(base, None).unwrap();
+ STATIC_TEMP_DIR.set(dir.clone()).unwrap();
+
+ proxmox_rest_server::init_worker_tasks(dir.clone(), file_opts).unwrap();
+ proxmox_access_control::init::init(&pdm_api_types::AccessControlConfig, dir)
+ .expect("failed to setup access control config");
+
+ unsafe {
+ libc::atexit(cleanup);
+ }
+ });
+}
+
+/// Create a [`TestRpcEnvironment`] for the use in a test.
+pub fn rpcenv() -> TestRpcEnvironment {
+ TestRpcEnvironment
+}
diff --git a/server/tests/common/test_application.rs b/server/tests/common/test_application.rs
new file mode 100644
index 00000000..bba5e5f0
--- /dev/null
+++ b/server/tests/common/test_application.rs
@@ -0,0 +1,287 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::Mutex;
+
+use anyhow::{Context, Error, bail};
+use nix::unistd::User;
+use serde::{Serialize, de::DeserializeOwned};
+
+use pbs_api_types::Authid;
+use proxmox_client::Client;
+use proxmox_product_config::{ProductConfig, ProductConfigParams};
+use proxmox_section_config::typed::SectionConfigData;
+
+use pdm_api_types::ConfigDigest;
+use pdm_api_types::remotes::{Remote, RemoteType};
+use pdm_config::remotes::RemoteConfig;
+
+use server::connection::{ClientFactory, PveClient};
+use server::context::ContextFactory;
+use server::pbs_client::PbsClient;
+
+/// Shared, clonable storage for arbitrary data recorded by a [`PveTestRemote`]'s client
+/// implementation.
+#[derive(Clone, Default)]
+pub struct TestRemoteState(Arc<Mutex<HashMap<String, serde_json::Value>>>);
+
+impl TestRemoteState {
+ /// Record `value` under `key`, overwriting any previous value stored there.
+ pub fn set<T: Serialize>(&self, key: impl Into<String>, value: &T) {
+ let value = serde_json::to_value(value).expect("failed to serialize test state value");
+ self.0.lock().unwrap().insert(key.into(), value);
+ }
+
+ /// Retrieve the value previously stored under `key`, if any.
+ pub fn get<T: DeserializeOwned>(&self, key: &str) -> Option<T> {
+ self.0.lock().unwrap().get(key).cloned().map(|value| {
+ serde_json::from_value(value).expect("failed to deserialize test state value")
+ })
+ }
+}
+
+/// A PVE remote registered in a [`TestApplication`], passed to the client factory whenever a
+/// client for it is created.
+#[derive(Clone)]
+pub struct PveTestRemote {
+ pub name: String,
+ pub nodes: u32,
+ pub state: TestRemoteState,
+}
+
+#[derive(Clone)]
+struct PveTestRemoteWithClientMaker {
+ remote: PveTestRemote,
+ make_client: Arc<dyn Fn(PveTestRemote) -> Result<Arc<PveClient>, Error> + Send + Sync>,
+}
+
+impl PveTestRemote {
+ /// Build the [`Remote`] configuration entry describing this test remote.
+ pub fn to_remote(&self) -> Remote {
+ Remote {
+ ty: RemoteType::Pve,
+ id: self.name.clone(),
+ nodes: Vec::new(),
+ authid: Authid::root_auth_id().clone(),
+ token: "".into(),
+ web_url: None,
+ }
+ }
+}
+
+/// A [`ContextFactory`] implementation for tests, backed by mocked remotes and a temporary
+/// directory for config, caches and runtime directories.
+///
+/// Make sure to *not* drops this while running the test cast, otherwise the temporary directory
+/// will be dropped.
+#[derive(Clone)]
+pub struct TestApplication {
+ pve_remotes: HashMap<String, PveTestRemoteWithClientMaker>,
+ base_dir: Arc<tempfile::TempDir>,
+}
+
+impl Default for TestApplication {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl TestApplication {
+ /// Create a test application without any registered remotes.
+ pub fn new() -> Self {
+ let target_tmpdir = env!("CARGO_TARGET_TMPDIR");
+ let base_dir = Arc::new(
+ tempfile::tempdir_in(target_tmpdir).expect("could not create temporary directory"),
+ );
+
+ Self {
+ pve_remotes: HashMap::new(),
+ base_dir,
+ }
+ }
+
+ /// Register a PVE remote called `name`, using `f` to create its client implementation.
+ ///
+ /// The client is constructed for every request, so any data that should outlive a single
+ /// client has to be kept in `state`.
+ pub fn with_pve_remote<S, C, F>(mut self, name: S, state: TestRemoteState, f: F) -> Self
+ where
+ S: Into<String>,
+ C: pve_api_types::client::PveClient + Send + Sync + 'static,
+ F: Fn(PveTestRemote) -> C + Send + Sync + 'static,
+ {
+ let name = name.into();
+ let make_client = Arc::new(move |remote| Ok(Arc::new(f(remote)) as Arc<PveClient>));
+
+ self.pve_remotes.insert(
+ name.clone(),
+ PveTestRemoteWithClientMaker {
+ make_client,
+ remote: PveTestRemote {
+ name,
+ nodes: 1,
+ state,
+ },
+ },
+ );
+
+ self
+ }
+}
+
+impl ContextFactory for TestApplication {
+ fn make_client_factory(&self) -> Result<Arc<dyn ClientFactory + Send + Sync>, Error> {
+ Ok(Arc::new(self.clone()))
+ }
+
+ fn make_remote_config(&self) -> Result<Box<dyn RemoteConfig + Send + Sync>, Error> {
+ Ok(Box::new(self.clone()))
+ }
+
+ fn make_product_config(&self) -> Result<ProductConfig, Error> {
+ let user = User::from_uid(nix::unistd::getuid())
+ .ok()
+ .flatten()
+ .context("could not look up user")?;
+
+ let base = self.base_dir.path();
+
+ let config_dir = base.join("config");
+ let state_dir = base.join("state");
+ let run_dir = base.join("run");
+ let cache_dir = base.join("cache");
+
+ std::fs::create_dir(&config_dir)?;
+ std::fs::create_dir(&state_dir)?;
+ std::fs::create_dir(&run_dir)?;
+ std::fs::create_dir(&cache_dir)?;
+
+ Ok(ProductConfig::new(ProductConfigParams {
+ api_user: user.clone(),
+ priv_user: user,
+ config_dir,
+ state_dir,
+ run_dir,
+ cache_dir,
+ }))
+ }
+
+ // Override any other methods here if needed.
+}
+
+impl RemoteConfig for TestApplication {
+ fn read(&self) -> Result<(SectionConfigData<Remote>, ConfigDigest), anyhow::Error> {
+ let mut sections = SectionConfigData::default();
+
+ for pve_remote in self.pve_remotes.values() {
+ sections.insert(
+ pve_remote.remote.name.clone(),
+ pve_remote.remote.to_remote(),
+ );
+ }
+
+ Ok((sections, ConfigDigest::from_slice([])))
+ }
+
+ fn read_secret_token(
+ &self,
+ _remote: &pdm_api_types::remotes::Remote,
+ ) -> Result<String, anyhow::Error> {
+ unimplemented!()
+ }
+
+ fn lock(&self) -> Result<proxmox_product_config::ApiLockGuard, anyhow::Error> {
+ unimplemented!()
+ }
+
+ fn write(
+ &self,
+ _remotes: proxmox_section_config::typed::SectionConfigData<pdm_api_types::remotes::Remote>,
+ ) -> Result<(), anyhow::Error> {
+ unimplemented!()
+ }
+}
+
+#[async_trait::async_trait]
+impl ClientFactory for TestApplication {
+ fn make_pve_client(&self, remote: &Remote) -> Result<Arc<PveClient>, Error> {
+ if let Some(a) = self.pve_remotes.get(&remote.id) {
+ (a.make_client)(a.remote.clone())
+ } else {
+ bail!("remote not registered")
+ }
+ }
+
+ fn make_pve_client_with_endpoint(
+ &self,
+ _remote: &Remote,
+ _target_endpoint: Option<&str>,
+ ) -> Result<Arc<PveClient>, Error> {
+ bail!("not implemented")
+ }
+
+ fn make_pbs_client(&self, _remote: &Remote) -> Result<Box<PbsClient>, Error> {
+ bail!("not implemented")
+ }
+
+ fn make_raw_client(&self, _remote: &Remote) -> Result<Box<proxmox_client::Client>, Error> {
+ bail!("not implemented")
+ }
+
+ async fn make_pve_client_and_login(&self, _remote: &Remote) -> Result<Arc<PveClient>, Error> {
+ bail!("not implemented")
+ }
+
+ async fn make_pbs_client_and_login(
+ &self,
+ _remote: &Remote,
+ ) -> Result<Box<PbsClient<Client>>, Error> {
+ bail!("not implemented")
+ }
+}
+
+macro_rules! test_pve_client {
+ ($ty:ident { $($overrides:tt)* }) => {
+
+ pub struct $ty(crate::common::PveTestRemote);
+
+ #[async_trait::async_trait]
+ impl pve_api_types::client::PveClient for $ty {
+ async fn list_nodes(
+ &self,
+ ) -> Result<Vec<pve_api_types::ClusterNodeIndexResponse>, proxmox_client::Error> {
+ let mut nodes = Vec::new();
+
+ for i in 0..self.0.nodes {
+ nodes.push(pve_api_types::ClusterNodeIndexResponse {
+ cpu: Some(0.0),
+ level: None,
+ maxcpu: Some(4),
+ maxmem: Some(4096),
+ mem: Some(1000),
+ node: format!("{}-node-{i}", self.0.name),
+ ssl_fingerprint: None,
+ status: pve_api_types::ClusterNodeIndexResponseStatus::Online,
+ uptime: Some(1000),
+ });
+ }
+
+ Ok(nodes)
+ }
+ $($overrides)*
+ }
+
+ impl $ty {
+ /// Name of the remote this client was created for.
+ pub fn remote(&self) -> &str {
+ &self.0.name
+ }
+
+ /// Shared state of the remote this client was created for.
+ pub fn state(&self) -> crate::common::TestRemoteState {
+ self.0.state.clone()
+ }
+ }
+ };
+}
+
+pub(crate) use test_pve_client;
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 16/21] tests: add example tests for SDN API routes
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (14 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 15/21] tests: add helpers for building API-handler-level integration tests Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 17/21] api-cache: add wrapper type Lukas Wagner
` (4 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
Using the previously added dependency-injected abstractions for
accessing remotes and clients, and by using the test helpers, we can now
fairly trivially create test cases that call the API handler directly.
These tests also demonstrate how previously captured API responses from
real PVE nodes can be used to build test cases.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
server/src/api/sdn/vnets.rs | 2 +-
.../pve/remote-a/list_vnets.json | 8 ++
.../pve/remote-a/list_zones.json | 8 ++
.../pve/remote-b/list_vnets.json | 8 ++
.../pve/remote-b/list_zones.json | 8 ++
server/tests/test_sdn.rs | 86 +++++++++++++++++++
6 files changed, 119 insertions(+), 1 deletion(-)
create mode 100644 server/tests/api_responses/pve/remote-a/list_vnets.json
create mode 100644 server/tests/api_responses/pve/remote-a/list_zones.json
create mode 100644 server/tests/api_responses/pve/remote-b/list_vnets.json
create mode 100644 server/tests/api_responses/pve/remote-b/list_zones.json
create mode 100644 server/tests/test_sdn.rs
diff --git a/server/src/api/sdn/vnets.rs b/server/src/api/sdn/vnets.rs
index 312d46b6..e42fcab7 100644
--- a/server/src/api/sdn/vnets.rs
+++ b/server/src/api/sdn/vnets.rs
@@ -59,7 +59,7 @@ pub const ROUTER: Router = Router::new()
}
)]
/// Query VNets of PVE remotes with optional filtering options
-async fn list_vnets(
+pub async fn list_vnets(
pending: Option<bool>,
running: Option<bool>,
remotes: Option<HashSet<String>>,
diff --git a/server/tests/api_responses/pve/remote-a/list_vnets.json b/server/tests/api_responses/pve/remote-a/list_vnets.json
new file mode 100644
index 00000000..a105de13
--- /dev/null
+++ b/server/tests/api_responses/pve/remote-a/list_vnets.json
@@ -0,0 +1,8 @@
+[
+ {
+ "digest" : "3458032372eb62efb730fec12b34d36627a9e6e8",
+ "type" : "vnet",
+ "vnet" : "aaaavnet",
+ "zone" : "aaaa"
+ }
+]
diff --git a/server/tests/api_responses/pve/remote-a/list_zones.json b/server/tests/api_responses/pve/remote-a/list_zones.json
new file mode 100644
index 00000000..f7932bd9
--- /dev/null
+++ b/server/tests/api_responses/pve/remote-a/list_zones.json
@@ -0,0 +1,8 @@
+[
+ {
+ "digest" : "142f8be078a0fb61fa28a89cffdb17f64b05e3f9",
+ "ipam" : "pve",
+ "type" : "simple",
+ "zone" : "aaaa"
+ }
+]
diff --git a/server/tests/api_responses/pve/remote-b/list_vnets.json b/server/tests/api_responses/pve/remote-b/list_vnets.json
new file mode 100644
index 00000000..4a59edc7
--- /dev/null
+++ b/server/tests/api_responses/pve/remote-b/list_vnets.json
@@ -0,0 +1,8 @@
+[
+ {
+ "digest" : "56e0d24671dca1e6cc30a2cfdfab79a037e66a44",
+ "type" : "vnet",
+ "vnet" : "bbbbvnet",
+ "zone" : "bbbb"
+ }
+]
diff --git a/server/tests/api_responses/pve/remote-b/list_zones.json b/server/tests/api_responses/pve/remote-b/list_zones.json
new file mode 100644
index 00000000..75fd763a
--- /dev/null
+++ b/server/tests/api_responses/pve/remote-b/list_zones.json
@@ -0,0 +1,8 @@
+[
+ {
+ "digest" : "bd994284b5d766ab58b5cb69f7363153d42816f7",
+ "ipam" : "pve",
+ "type" : "simple",
+ "zone" : "bbbb"
+ }
+]
diff --git a/server/tests/test_sdn.rs b/server/tests/test_sdn.rs
new file mode 100644
index 00000000..204a94cd
--- /dev/null
+++ b/server/tests/test_sdn.rs
@@ -0,0 +1,86 @@
+use pve_api_types::{ListZonesType, SdnVnet, SdnVnetType, SdnZone};
+use server::context::ContextFactory;
+
+use crate::common::{TestApplication, rpcenv};
+
+pub mod common;
+
+common::test_pve_client!(SdnPveClient {
+ async fn list_zones(
+ &self,
+ _pending: Option<bool>,
+ _running: Option<bool>,
+ _ty: Option<ListZonesType>,
+ ) -> Result<Vec<SdnZone>, proxmox_client::Error> {
+ common::read_captured_response(&format!(
+ "tests/api_responses/pve/{}/list_zones.json",
+ self.remote(),
+ )).await
+ }
+
+ async fn list_vnets(
+ &self,
+ _pending: Option<bool>,
+ _running: Option<bool>,
+ ) -> Result<Vec<SdnVnet>, proxmox_client::Error> {
+ common::read_captured_response(&format!(
+ "tests/api_responses/pve/{}/list_vnets.json",
+ self.remote(),
+ )).await
+ }
+});
+
+#[tokio::test]
+async fn test_lists_zones() {
+ common::test_setup();
+
+ let test_app = TestApplication::new()
+ .with_pve_remote("remote-a", Default::default(), SdnPveClient)
+ .with_pve_remote("remote-b", Default::default(), SdnPveClient);
+
+ let app = test_app.make_pdm_application().unwrap();
+
+ let mut result =
+ server::api::sdn::zones::list_zones(None, None, None, None, &mut rpcenv(), app)
+ .await
+ .unwrap();
+
+ // We don't guarantee any ordering
+ result.sort_by(|a, b| a.remote.cmp(&b.remote));
+
+ assert_eq!(result.len(), 2);
+
+ assert_eq!(result[0].remote, "remote-a");
+ assert_eq!(result[0].zone.zone, "aaaa");
+ assert_eq!(result[0].zone.ty, ListZonesType::Simple);
+ assert_eq!(result[1].remote, "remote-b");
+ assert_eq!(result[1].zone.zone, "bbbb");
+ assert_eq!(result[1].zone.ty, ListZonesType::Simple);
+}
+
+#[tokio::test]
+async fn test_lists_vnets() {
+ common::test_setup();
+
+ let test_app = TestApplication::new()
+ .with_pve_remote("remote-a", Default::default(), SdnPveClient)
+ .with_pve_remote("remote-b", Default::default(), SdnPveClient);
+
+ let app = test_app.make_pdm_application().unwrap();
+
+ let mut result = server::api::sdn::vnets::list_vnets(None, None, None, &mut rpcenv(), app)
+ .await
+ .unwrap();
+
+ // We don't guarantee any ordering
+ result.sort_by(|a, b| a.remote.cmp(&b.remote));
+
+ assert_eq!(result.len(), 2);
+
+ assert_eq!(result[0].remote, "remote-a");
+ assert_eq!(result[0].vnet.vnet, "aaaavnet");
+ assert_eq!(result[0].vnet.ty, SdnVnetType::Vnet);
+ assert_eq!(result[1].remote, "remote-b");
+ assert_eq!(result[1].vnet.vnet, "bbbbvnet");
+ assert_eq!(result[1].vnet.ty, SdnVnetType::Vnet);
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 17/21] api-cache: add wrapper type
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (15 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 16/21] tests: add example tests for SDN API routes Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 18/21] context: provide api-cache on the app object Lukas Wagner
` (3 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
NamespacedCache is supposed to be fully generic and eventually be moved
to a shared crate. ApiCache adds PDM-specific semantics, namely having
per-remote namespaces and also one global namespace.
Before, these additional semantics were encoded in the helper functions
in api_cache.rs (e.g. read_global, read_remote), but since we want to
put move a handle to the api cache into PdmApplication, the helpers are
turned into methods of this new wrapper type.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
Notes:
Changes since v2:
- Extend the commit message to hopefully better convey why this wrapper
is needed.
server/src/api_cache.rs | 98 +++++++++++++++++++++++++++++++++++------
1 file changed, 84 insertions(+), 14 deletions(-)
diff --git a/server/src/api_cache.rs b/server/src/api_cache.rs
index b20f0535..6dc8a043 100644
--- a/server/src/api_cache.rs
+++ b/server/src/api_cache.rs
@@ -59,6 +59,8 @@ use std::time::Duration;
use nix::sys::stat::Mode;
+use proxmox_sys::fs::CreateOptions;
+
use crate::namespaced_cache::{
BlockingReadableCacheNamespace, BlockingWritableCacheNamespace, CacheError, NamespacedCache,
ReadableCacheNamespace, WritableCacheNamespace,
@@ -70,57 +72,125 @@ pub const PDM_API_CACHE_PATH: &str = concat!(pdm_buildcfg::PDM_RUN_DIR_M!(), "/a
const GLOBAL_NAMESPACE: &str = "global";
const LOCK_TIMEOUT: Duration = Duration::from_secs(10);
-static CACHE: LazyLock<NamespacedCache> = LazyLock::new(|| {
+static CACHE: LazyLock<ApiCache> = LazyLock::new(|| {
let file_options = proxmox_product_config::default_create_options();
let dir_options = file_options.perm(Mode::from_bits_truncate(0o750));
- NamespacedCache::new(PathBuf::from(PDM_API_CACHE_PATH), dir_options, file_options)
+ ApiCache::new(PathBuf::from(PDM_API_CACHE_PATH), dir_options, file_options)
});
fn format_remote_namespace(remote: &str) -> String {
format!("remote-{remote}")
}
+/// Cache for API responses from remotes.
+///
+/// Thin wrapper around a [`NamespacedCache`], providing remote-specific and
+/// global namespaces as described in the module documentation.
+pub struct ApiCache {
+ cache: NamespacedCache,
+}
+
+impl ApiCache {
+ pub fn new<P: Into<PathBuf>>(
+ base_directory: P,
+ dir_options: CreateOptions,
+ file_options: CreateOptions,
+ ) -> Self {
+ Self {
+ cache: NamespacedCache::new(base_directory, dir_options, file_options),
+ }
+ }
+
+ /// Lock the cache for reading remote-specific data (blocking interface).
+ pub fn read_remote_blocking(
+ &self,
+ remote: &str,
+ ) -> Result<BlockingReadableCacheNamespace, CacheError> {
+ self.cache
+ .read_blocking(&format_remote_namespace(remote), LOCK_TIMEOUT)
+ }
+
+ /// Lock the cache for writing remote-specific data (blocking interface).
+ pub fn write_remote_blocking(
+ &self,
+ remote: &str,
+ ) -> Result<BlockingWritableCacheNamespace, CacheError> {
+ self.cache
+ .write_blocking(&format_remote_namespace(remote), LOCK_TIMEOUT)
+ }
+
+ /// Lock the cache for reading global data (blocking interface).
+ pub fn read_global_blocking(&self) -> Result<BlockingReadableCacheNamespace, CacheError> {
+ self.cache.read_blocking(GLOBAL_NAMESPACE, LOCK_TIMEOUT)
+ }
+
+ /// Lock the cache for writing global data (blocking interface).
+ pub fn write_global_blocking(&self) -> Result<BlockingWritableCacheNamespace, CacheError> {
+ self.cache.write_blocking(GLOBAL_NAMESPACE, LOCK_TIMEOUT)
+ }
+
+ /// Lock the cache for reading remote-specific data (async interface).
+ pub async fn read_remote(&self, remote: &str) -> Result<ReadableCacheNamespace, CacheError> {
+ self.cache
+ .read(&format_remote_namespace(remote), LOCK_TIMEOUT)
+ .await
+ }
+
+ /// Lock the cache for writing remote-specific data (async interface).
+ pub async fn write_remote(&self, remote: &str) -> Result<WritableCacheNamespace, CacheError> {
+ self.cache
+ .write(&format_remote_namespace(remote), LOCK_TIMEOUT)
+ .await
+ }
+
+ /// Lock the cache for reading global data (async interface).
+ pub async fn read_global(&self) -> Result<ReadableCacheNamespace, CacheError> {
+ self.cache.read(GLOBAL_NAMESPACE, LOCK_TIMEOUT).await
+ }
+
+ /// Lock the cache for writing global data (async interface).
+ pub async fn write_global(&self) -> Result<WritableCacheNamespace, CacheError> {
+ self.cache.write(GLOBAL_NAMESPACE, LOCK_TIMEOUT).await
+ }
+}
+
/// Lock the cache for reading remote-specific data (blocking interface).
pub fn read_remote_blocking(remote: &str) -> Result<BlockingReadableCacheNamespace, CacheError> {
- CACHE.read_blocking(&format_remote_namespace(remote), LOCK_TIMEOUT)
+ CACHE.read_remote_blocking(remote)
}
/// Lock the cache for writing remote-specific data (blocking interface).
pub fn write_remote_blocking(remote: &str) -> Result<BlockingWritableCacheNamespace, CacheError> {
- CACHE.write_blocking(&format_remote_namespace(remote), LOCK_TIMEOUT)
+ CACHE.write_remote_blocking(remote)
}
/// Lock the cache for reading global data (blocking interface).
pub fn read_global_blocking() -> Result<BlockingReadableCacheNamespace, CacheError> {
- CACHE.read_blocking(GLOBAL_NAMESPACE, LOCK_TIMEOUT)
+ CACHE.read_global_blocking()
}
/// Lock the cache for writing global data (blocking interface).
pub fn write_global_blocking() -> Result<BlockingWritableCacheNamespace, CacheError> {
- CACHE.write_blocking(GLOBAL_NAMESPACE, LOCK_TIMEOUT)
+ CACHE.write_global_blocking()
}
/// Lock the cache for reading remote-specific data (async interface).
pub async fn read_remote(remote: &str) -> Result<ReadableCacheNamespace, CacheError> {
- CACHE
- .read(&format_remote_namespace(remote), LOCK_TIMEOUT)
- .await
+ CACHE.read_remote(remote).await
}
/// Lock the cache for writing remote-specific data (async interface).
pub async fn write_remote(remote: &str) -> Result<WritableCacheNamespace, CacheError> {
- CACHE
- .write(&format_remote_namespace(remote), LOCK_TIMEOUT)
- .await
+ CACHE.write_remote(remote).await
}
/// Lock the cache for reading global data (async interface).
pub async fn read_global() -> Result<ReadableCacheNamespace, CacheError> {
- CACHE.read(GLOBAL_NAMESPACE, LOCK_TIMEOUT).await
+ CACHE.read_global().await
}
/// Lock the cache for writing global data (async interface).
pub async fn write_global() -> Result<WritableCacheNamespace, CacheError> {
- CACHE.write(GLOBAL_NAMESPACE, LOCK_TIMEOUT).await
+ CACHE.write_global().await
}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 18/21] context: provide api-cache on the app object
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (16 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 17/21] api-cache: add wrapper type Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 19/21] api: subscriptions: use PdmApplication instead of globals Lukas Wagner
` (2 subsequent siblings)
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
Move the ApiCache instance from the global CACHE static onto
PdmApplication, built from the product config's cache directory
instead of the hardcoded PDM_API_CACHE_PATH. The free functions in
api_cache now fetch it via context::pdm_application() instead of the
static.
Add default_file_create_options()/default_dir_create_options() to
ProductConfig so the cache's directory and file permissions can be
derived the same way in both the real application and tests.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
server/src/api_cache.rs | 45 ++++++++++---------
.../bin/proxmox-datacenter-privileged-api.rs | 7 ++-
server/src/context/mod.rs | 20 ++++++++-
server/tests/common/test_application.rs | 2 +
4 files changed, 52 insertions(+), 22 deletions(-)
diff --git a/server/src/api_cache.rs b/server/src/api_cache.rs
index 6dc8a043..fed66483 100644
--- a/server/src/api_cache.rs
+++ b/server/src/api_cache.rs
@@ -54,31 +54,22 @@
//! ```
use std::path::PathBuf;
-use std::sync::LazyLock;
use std::time::Duration;
-use nix::sys::stat::Mode;
-
use proxmox_sys::fs::CreateOptions;
+use crate::context;
use crate::namespaced_cache::{
BlockingReadableCacheNamespace, BlockingWritableCacheNamespace, CacheError, NamespacedCache,
ReadableCacheNamespace, WritableCacheNamespace,
};
-/// Path at which API responses are cached.
-pub const PDM_API_CACHE_PATH: &str = concat!(pdm_buildcfg::PDM_RUN_DIR_M!(), "/api-cache");
+/// Subdirectory at which API responses are cached.
+pub const PDM_API_CACHE_SUBDIR: &str = "api-cache";
const GLOBAL_NAMESPACE: &str = "global";
const LOCK_TIMEOUT: Duration = Duration::from_secs(10);
-static CACHE: LazyLock<ApiCache> = LazyLock::new(|| {
- let file_options = proxmox_product_config::default_create_options();
- let dir_options = file_options.perm(Mode::from_bits_truncate(0o750));
-
- ApiCache::new(PathBuf::from(PDM_API_CACHE_PATH), dir_options, file_options)
-});
-
fn format_remote_namespace(remote: &str) -> String {
format!("remote-{remote}")
}
@@ -157,40 +148,54 @@ impl ApiCache {
/// Lock the cache for reading remote-specific data (blocking interface).
pub fn read_remote_blocking(remote: &str) -> Result<BlockingReadableCacheNamespace, CacheError> {
- CACHE.read_remote_blocking(remote)
+ context::pdm_application()
+ .api_cache()
+ .read_remote_blocking(remote)
}
/// Lock the cache for writing remote-specific data (blocking interface).
pub fn write_remote_blocking(remote: &str) -> Result<BlockingWritableCacheNamespace, CacheError> {
- CACHE.write_remote_blocking(remote)
+ context::pdm_application()
+ .api_cache()
+ .write_remote_blocking(remote)
}
/// Lock the cache for reading global data (blocking interface).
pub fn read_global_blocking() -> Result<BlockingReadableCacheNamespace, CacheError> {
- CACHE.read_global_blocking()
+ context::pdm_application()
+ .api_cache()
+ .read_global_blocking()
}
/// Lock the cache for writing global data (blocking interface).
pub fn write_global_blocking() -> Result<BlockingWritableCacheNamespace, CacheError> {
- CACHE.write_global_blocking()
+ context::pdm_application()
+ .api_cache()
+ .write_global_blocking()
}
/// Lock the cache for reading remote-specific data (async interface).
pub async fn read_remote(remote: &str) -> Result<ReadableCacheNamespace, CacheError> {
- CACHE.read_remote(remote).await
+ context::pdm_application()
+ .api_cache()
+ .read_remote(remote)
+ .await
}
/// Lock the cache for writing remote-specific data (async interface).
pub async fn write_remote(remote: &str) -> Result<WritableCacheNamespace, CacheError> {
- CACHE.write_remote(remote).await
+ context::pdm_application()
+ .api_cache()
+ .write_remote(remote)
+ .await
}
/// Lock the cache for reading global data (async interface).
pub async fn read_global() -> Result<ReadableCacheNamespace, CacheError> {
- CACHE.read_global().await
+ context::pdm_application().api_cache().read_global().await
}
/// Lock the cache for writing global data (async interface).
pub async fn write_global() -> Result<WritableCacheNamespace, CacheError> {
- CACHE.write_global().await
+ context::pdm_application().api_cache().write_global().await
}
diff --git a/server/src/bin/proxmox-datacenter-privileged-api.rs b/server/src/bin/proxmox-datacenter-privileged-api.rs
index 3815ab03..d32e10c5 100644
--- a/server/src/bin/proxmox-datacenter-privileged-api.rs
+++ b/server/src/bin/proxmox-datacenter-privileged-api.rs
@@ -2,6 +2,7 @@ use std::path::Path;
use std::pin::pin;
use anyhow::{Context as _, Error, bail, format_err};
+use const_format::concatcp;
use futures::*;
use hyper_util::server::graceful::GracefulShutdown;
use nix::fcntl::AtFlags;
@@ -104,7 +105,11 @@ fn create_directories() -> Result<(), Error> {
)?;
pdm_config::setup::mkdir_perms(
- api_cache::PDM_API_CACHE_PATH,
+ concatcp!(
+ pdm_buildcfg::PDM_RUN_DIR_M!(),
+ "/",
+ api_cache::PDM_API_CACHE_SUBDIR,
+ ),
api_user.uid,
api_user.gid,
0o750,
diff --git a/server/src/context/mod.rs b/server/src/context/mod.rs
index f0853d65..082ab8de 100644
--- a/server/src/context/mod.rs
+++ b/server/src/context/mod.rs
@@ -10,6 +10,7 @@ use proxmox_product_config::{ProductConfig, ProductConfigParams};
use pdm_config::{remotes::RemoteConfig, subscriptions::SubscriptionKeyConfig};
+use crate::api_cache::ApiCache;
use crate::connection::{self, ClientFactory};
#[cfg(remote_config = "faked")]
@@ -78,12 +79,22 @@ pub trait ContextFactory {
}
fn make_pdm_application(&self) -> Result<PdmApplication, Error> {
+ let product_config = self.make_product_config()?;
+
+ let api_cache = ApiCache::new(
+ product_config.run_dir().join("api-cache"),
+ product_config.default_dir_create_options(),
+ product_config.default_file_create_options(),
+ );
+
Ok(PdmApplication {
inner: Arc::new(PdmApplicationInner {
client_factory: self.make_client_factory()?,
remote_config: self.make_remote_config()?,
subscription_key_config: self.make_subscription_key_config()?,
- product_config: self.make_product_config()?,
+ product_config,
+
+ api_cache,
}),
})
}
@@ -138,6 +149,11 @@ impl PdmApplication {
pub fn product_config(&self) -> &ProductConfig {
&self.inner.product_config
}
+
+ /// Get a reference to the [`ApiCache`].
+ pub fn api_cache(&self) -> &ApiCache {
+ &self.inner.api_cache
+ }
}
struct PdmApplicationInner {
@@ -145,4 +161,6 @@ struct PdmApplicationInner {
remote_config: Box<dyn RemoteConfig + Send + Sync>,
subscription_key_config: Box<dyn SubscriptionKeyConfig + Send + Sync>,
product_config: ProductConfig,
+
+ api_cache: ApiCache,
}
diff --git a/server/tests/common/test_application.rs b/server/tests/common/test_application.rs
index bba5e5f0..83b35bf6 100644
--- a/server/tests/common/test_application.rs
+++ b/server/tests/common/test_application.rs
@@ -155,6 +155,8 @@ impl ContextFactory for TestApplication {
std::fs::create_dir(&run_dir)?;
std::fs::create_dir(&cache_dir)?;
+ std::fs::create_dir(run_dir.join("api-cache"))?;
+
Ok(ProductConfig::new(ProductConfigParams {
api_user: user.clone(),
priv_user: user,
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 19/21] api: subscriptions: use PdmApplication instead of globals
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (17 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 18/21] context: provide api-cache on the app object Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 20/21] pdm-config: subscriptions: drop unused accessor functions Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 21/21] tests: add example tests for remote subscription management Lukas Wagner
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
Inject PdmApplication through the subscription API handlers and
the daily-update binary, replacing direct calls to
pdm_config::subscriptions/remotes, crate::connection::make_*_client,
and the api_cache free functions with the equivalent methods on the
injected app handle.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
Notes:
Changes since v1:
- use app.remote_config().read() in two more spots where the
application handle was already availalbe,
check_subscription and get_all_subscription_infos
Changes since v2:
- use new #[state] attribute
server/src/api/nodes/subscription.rs | 23 +-
server/src/api/resources.rs | 39 ++--
server/src/api/subscriptions/mod.rs | 221 ++++++++++--------
...proxmox-datacenter-manager-daily-update.rs | 12 +-
4 files changed, 174 insertions(+), 121 deletions(-)
diff --git a/server/src/api/nodes/subscription.rs b/server/src/api/nodes/subscription.rs
index 04e4141e..044c8bd7 100644
--- a/server/src/api/nodes/subscription.rs
+++ b/server/src/api/nodes/subscription.rs
@@ -18,6 +18,7 @@ use pdm_api_types::subscription::{
use crate::api::resources::{
fetch_complete_subscription_info_for_remote, get_subscription_info_for_remote,
};
+use crate::context::PdmApplication;
const PRODUCT_URL: &str = "https://pdm.proxmox.com/faq.html";
const APT_AUTH_FN: &str = "/etc/apt/auth.conf.d/pdm.conf";
@@ -31,13 +32,14 @@ fn apt_auth_file_opts() -> CreateOptions {
CreateOptions::new().perm(mode).owner(nix::unistd::ROOT)
}
-async fn get_all_subscription_infos()
--> Result<HashMap<String, (RemoteType, HashMap<String, Option<NodeSubscriptionInfo>>)>, Error> {
- let (remotes_config, _digest) = pdm_config::remotes::config()?;
+async fn get_all_subscription_infos(
+ app: &PdmApplication,
+) -> Result<HashMap<String, (RemoteType, HashMap<String, Option<NodeSubscriptionInfo>>)>, Error> {
+ let (remotes_config, _digest) = app.remote_config().read()?;
let mut subscription_info = HashMap::new();
for (remote_name, remote) in remotes_config.iter() {
- match get_subscription_info_for_remote(remote, 24 * 60 * 60).await {
+ match get_subscription_info_for_remote(app, remote, 24 * 60 * 60).await {
Ok(info) => {
subscription_info.insert(remote_name.to_string(), (remote.ty, info));
}
@@ -118,8 +120,8 @@ fn check_counts(stats: &SubscriptionStatistics) -> Result<(), Error> {
}
)]
/// Return subscription status
-pub async fn get_subscription() -> Result<PdmSubscriptionInfo, Error> {
- let infos = get_all_subscription_infos().await?;
+pub async fn get_subscription(#[state] app: PdmApplication) -> Result<PdmSubscriptionInfo, Error> {
+ let infos = get_all_subscription_infos(&app).await?;
let statistics = count_subscriptions(&infos);
@@ -156,8 +158,8 @@ pub async fn get_subscription() -> Result<PdmSubscriptionInfo, Error> {
},
)]
/// Update subscription information
-pub async fn check_subscription() -> Result<(), Error> {
- let infos = get_all_subscription_infos().await?;
+pub async fn check_subscription(#[state] app: PdmApplication) -> Result<(), Error> {
+ let infos = get_all_subscription_infos(&app).await?;
let stats = count_subscriptions(&infos);
if let Err(err) = check_counts(&stats) {
@@ -181,7 +183,7 @@ pub async fn check_subscription() -> Result<(), Error> {
// Get fresh subscription info. The cache does not store the serverid, so we
// need to fetch it from the remote. This has the upside of always yielding
// fresh results.
- let (remote_config, _digest) = pdm_config::remotes::config()?;
+ let (remote_config, _digest) = app.remote_config().read()?;
let Some(remote) = remote_config.get(remote_name) else {
log::debug!(
"Remote vanished while updating subscription information \
@@ -189,7 +191,8 @@ pub async fn check_subscription() -> Result<(), Error> {
);
continue 'outer;
};
- let node_info = fetch_complete_subscription_info_for_remote(remote).await?;
+ let node_info =
+ fetch_complete_subscription_info_for_remote(&app, remote).await?;
let Some(info) = node_info.iter().find_map(|(node, val)| {
if let Some(info) = val.as_ref() {
if info.status == SubscriptionStatus::Active
diff --git a/server/src/api/resources.rs b/server/src/api/resources.rs
index 09d2b88d..a1343b49 100644
--- a/server/src/api/resources.rs
+++ b/server/src/api/resources.rs
@@ -31,6 +31,8 @@ use proxmox_subscription::SubscriptionStatus;
use pve_api_types::{ClusterResource, ClusterResourceNetworkType, ClusterResourceType};
use serde::{Deserialize, Serialize};
+use crate::api_cache::ApiCache;
+use crate::context::PdmApplication;
use crate::metric_collection::top_entities;
use crate::{api_cache, connection, views};
@@ -686,8 +688,9 @@ pub async fn get_subscription_status(
verbose: bool,
view: Option<String>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<Vec<RemoteSubscriptions>, Error> {
- let (remotes_config, _) = pdm_config::remotes::config()?;
+ let (remotes_config, _) = app.remote_config().read()?;
let mut futures = Vec::new();
@@ -711,10 +714,11 @@ pub async fn get_subscription_status(
}
let view = view.clone();
+ let app_clone = app.clone();
let future = async move {
let (node_status, error) =
- match get_subscription_info_for_remote(&remote, max_age).await {
+ match get_subscription_info_for_remote(&app_clone, &remote, max_age).await {
Ok(mut node_status) => {
node_status.retain(|node, _| {
if let Some(view) = &view {
@@ -846,17 +850,21 @@ struct CachedSubscriptionState {
/// If recent enough cached data is available, it is returned
/// instead of calling out to the remote.
pub async fn get_subscription_info_for_remote(
+ app: &PdmApplication,
remote: &Remote,
max_age: u64,
) -> Result<HashMap<String, Option<NodeSubscriptionInfo>>, Error> {
- if let Some(cached_subscription) = get_cached_subscription_info(&remote.id, max_age).await? {
+ if let Some(cached_subscription) =
+ get_cached_subscription_info(app.api_cache(), &remote.id, max_age).await?
+ {
Ok(cached_subscription.node_info)
} else {
- let node_info = fetch_remote_subscription_info(remote).await?;
+ let node_info = fetch_remote_subscription_info(app, remote).await?;
let now = proxmox_time::epoch_i64();
if let Some(existing_state) =
- update_cached_subscription_info(&remote.id, node_info.clone(), now).await?
+ update_cached_subscription_info(app.api_cache(), &remote.id, node_info.clone(), now)
+ .await?
{
// Somebody else updated the cache while we performed the API request,
// return the more recent data instead of the data we just fetched.
@@ -871,21 +879,24 @@ pub async fn get_subscription_info_for_remote(
/// The cache will be updated, but never read from. This guarantees that the `serverid` is set, as
/// it cannot be stored in the cache.
pub async fn fetch_complete_subscription_info_for_remote(
+ app: &PdmApplication,
remote: &Remote,
) -> Result<HashMap<String, Option<NodeSubscriptionInfo>>, Error> {
- let node_info = fetch_remote_subscription_info(remote).await?;
+ let node_info = fetch_remote_subscription_info(app, remote).await?;
let now = proxmox_time::epoch_i64();
- let _ = update_cached_subscription_info(&remote.id, node_info.clone(), now).await?;
+ let _ = update_cached_subscription_info(app.api_cache(), &remote.id, node_info.clone(), now)
+ .await?;
Ok(node_info)
}
const SUBSCRIPTION_STATE_CACHE_KEY: &str = "subscription-state";
async fn get_cached_subscription_info(
+ api_cache: &ApiCache,
remote: &str,
max_age: u64,
) -> Result<Option<CachedSubscriptionState>, Error> {
- let cache = api_cache::read_remote(remote).await?;
+ let cache = api_cache.read_remote(remote).await?;
let subscription_state = cache
.get_with_max_age(SUBSCRIPTION_STATE_CACHE_KEY, max_age as i64)
.await
@@ -897,8 +908,8 @@ async fn get_cached_subscription_info(
}
/// Drop the cached subscription state for a remote, forcing the next read to refetch.
-pub async fn invalidate_subscription_info_for_remote(remote_id: &str) {
- let cache = match api_cache::write_remote(remote_id).await {
+pub async fn invalidate_subscription_info_for_remote(api_cache: &ApiCache, remote_id: &str) {
+ let cache = match api_cache.write_remote(remote_id).await {
Ok(cache) => cache,
Err(err) => {
log::error!("could not open API cache for {remote_id}: {err}");
@@ -916,11 +927,12 @@ pub async fn invalidate_subscription_info_for_remote(remote_id: &str) {
/// stored state as `Ok(Some(state))`. If the data that was passed in replaced the cache
/// entry, `Ok(None)` is returned.
async fn update_cached_subscription_info(
+ api_cache: &ApiCache,
remote: &str,
node_info: HashMap<String, Option<NodeSubscriptionInfo>>,
now: i64,
) -> Result<Option<CachedSubscriptionState>, Error> {
- let cache = api_cache::write_remote(remote).await?;
+ let cache = api_cache.write_remote(remote).await?;
Ok(cache
.set_if_newer_with_timestamp(
@@ -967,12 +979,13 @@ fn map_node_subscription_list_to_state(
/// Fetch remote resources and map to pdm-native data types.
async fn fetch_remote_subscription_info(
+ app: &PdmApplication,
remote: &Remote,
) -> Result<HashMap<String, Option<NodeSubscriptionInfo>>, Error> {
let mut list = HashMap::new();
match remote.ty {
RemoteType::Pve => {
- let client = connection::make_pve_client(remote)?;
+ let client = app.client_factory().make_pve_client(remote)?;
let nodes = client.list_nodes().await?;
let mut futures = Vec::with_capacity(nodes.len());
@@ -1013,7 +1026,7 @@ async fn fetch_remote_subscription_info(
}
}
RemoteType::Pbs => {
- let client = connection::make_pbs_client(remote)?;
+ let client = app.client_factory().make_pbs_client(remote)?;
let info = client.get_subscription().await.ok().map(|info| {
let level = SubscriptionLevel::from_key(info.key.as_deref());
diff --git a/server/src/api/subscriptions/mod.rs b/server/src/api/subscriptions/mod.rs
index 81582945..370941a2 100644
--- a/server/src/api/subscriptions/mod.rs
+++ b/server/src/api/subscriptions/mod.rs
@@ -33,6 +33,7 @@ use crate::api::remotes::RemoteIterator;
use crate::api::resources::{
get_subscription_info_for_remote, invalidate_subscription_info_for_remote,
};
+use crate::context::PdmApplication;
pub const ROUTER: Router = Router::new()
.get(&list_subdirs_api_method!(SUBDIRS))
@@ -131,14 +132,18 @@ fn key_not_found(key: &str) -> Error {
/// additionally gated on per-remote `PRIV_RESOURCE_AUDIT` so that an operator who can audit the
/// pool but not a specific remote does not learn which keys are pinned to it (and through that,
/// the existence and rough size of that remote's deployment).
-fn list_keys(rpcenv: &mut dyn RpcEnvironment) -> Result<Vec<SubscriptionKeyEntry>, Error> {
+fn list_keys(
+ rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
+) -> Result<Vec<SubscriptionKeyEntry>, Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
.context("no authid available")?
.parse()?;
let user_info = CachedUserInfo::new()?;
- let (config, digest) = pdm_config::subscriptions::config()?;
+ let (config, digest) = app.subscription_key_config().read()?;
+
rpcenv["digest"] = digest.to_hex().into();
Ok(config
.into_iter()
@@ -193,6 +198,7 @@ async fn add_keys(
keys: Vec<String>,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<AddKeysResult, Error> {
if keys.is_empty() {
http_bail!(BAD_REQUEST, "no keys provided");
@@ -229,8 +235,8 @@ async fn add_keys(
let added = entries.len() as u32;
let new_digest = tokio::task::spawn_blocking(move || -> Result<ConfigDigest, Error> {
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
// `insert` returns the previous entry when one existed; treat that as the duplicate
@@ -244,7 +250,7 @@ async fn add_keys(
}
}
- pdm_config::subscriptions::save_config(&config)
+ app.subscription_key_config().write(&config)
})
.await??;
rpcenv["digest"] = new_digest.to_hex().into();
@@ -270,14 +276,18 @@ async fn add_keys(
/// Bound entries are hidden from operators who cannot audit the bound remote (mirrors the
/// `list_keys` filter); the response is the same 404 either way so a probe cannot distinguish
/// "key exists but you cannot see it" from "key not in pool".
-fn get_key(key: String, rpcenv: &mut dyn RpcEnvironment) -> Result<SubscriptionKeyEntry, Error> {
+fn get_key(
+ key: String,
+ rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
+) -> Result<SubscriptionKeyEntry, Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
.context("no authid available")?
.parse()?;
let user_info = CachedUserInfo::new()?;
- let (config, digest) = pdm_config::subscriptions::config()?;
+ let (config, digest) = app.subscription_key_config().read()?;
rpcenv["digest"] = digest.to_hex().into();
let mut entry = config
.get(&key)
@@ -322,6 +332,7 @@ async fn delete_key(
key: String,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<(), Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -333,7 +344,7 @@ async fn delete_key(
// operator with only PRIV_SYS_MODIFY should not be able to probe live subscription state on
// a remote they cannot audit. Read the entry once without the lock for this gate; the
// authoritative read happens under the spawn_blocking section below.
- let (pre_config, pre_digest) = pdm_config::subscriptions::config()?;
+ let (pre_config, pre_digest) = app.subscription_key_config().read()?;
let Some(pre_entry) = pre_config.get(&key) else {
return Err(key_not_found(&key));
};
@@ -352,7 +363,7 @@ async fn delete_key(
let pre_binding = pre_entry.remote.as_deref().zip(pre_entry.node.as_deref());
// Owned bool so the orphan guard inside spawn_blocking does not borrow `pre_config`.
let pre_had_binding = pre_binding.is_some();
- let synced_block = check_synced_assignment_for_unassign(&key, pre_binding).await?;
+ let synced_block = check_synced_assignment_for_unassign(&app, &key, pre_binding).await?;
drop(pre_config);
// The lock + sync IO runs on a blocking thread so the async runtime is free for other work
@@ -361,8 +372,8 @@ async fn delete_key(
// cross the boundary; reconstructing it is cheap (it just reads the shared ACL cache).
let new_digest = tokio::task::spawn_blocking(move || -> Result<ConfigDigest, Error> {
let user_info = CachedUserInfo::new()?;
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
let Some(entry) = config.get(&key) else {
@@ -404,14 +415,14 @@ async fn delete_key(
// Save the authoritative pool config first: an interrupted remove must not leave a `key`
// entry whose signed blob is gone. A stale shadow blob with no main entry is benign, as
// readers do not consult it.
- let new_digest = pdm_config::subscriptions::save_config(&config)?;
+ let new_digest = app.subscription_key_config().write(&config)?;
// Best-effort shadow cleanup. The shadow only caches signed info, so a corrupt or
// otherwise unparseable shadow must not block removing the key from the pool: drop the
// cached blob when the shadow loads, and leave the orphan entry behind otherwise.
- match pdm_config::subscriptions::shadow_config() {
+ match app.subscription_key_config().read_shadow() {
Ok(mut shadow) => {
shadow.remove(&key);
- if let Err(err) = pdm_config::subscriptions::save_shadow(&shadow) {
+ if let Err(err) = app.subscription_key_config().write_shadow(&shadow) {
warn!("key '{key}' removed from pool, but updating its shadow failed: {err}");
}
}
@@ -457,6 +468,7 @@ async fn set_assignment(
node: String,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<(), Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -475,7 +487,7 @@ async fn set_assignment(
// orphans whatever live subscription the old remote still ran. Same shape and same guard
// as delete_key / clear_assignment; only fires when the binding actually moves (re-set to
// the same target leaves the OLD binding intact and carries no orphan risk).
- let (pre_config, pre_digest) = pdm_config::subscriptions::config()?;
+ let (pre_config, pre_digest) = app.subscription_key_config().read()?;
let pre_entry = pre_config.get(&key);
let pre_binding = pre_entry.and_then(|e| e.remote.as_deref().zip(e.node.as_deref()));
let rebind_moves_binding = match pre_binding {
@@ -498,7 +510,7 @@ async fn set_assignment(
}
let pre_had_binding = pre_binding.is_some();
let synced_block = if rebind_moves_binding {
- check_synced_assignment_for_unassign(&key, pre_binding).await?
+ check_synced_assignment_for_unassign(&app, &key, pre_binding).await?
} else {
None
};
@@ -509,8 +521,8 @@ async fn set_assignment(
// under the lock.
let new_digest = tokio::task::spawn_blocking(move || -> Result<ConfigDigest, Error> {
let user_info = CachedUserInfo::new()?;
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
let Some(stored_entry) = config.get(&key).cloned() else {
@@ -559,7 +571,7 @@ async fn set_assignment(
);
}
- let (remotes_config, _) = pdm_config::remotes::config()?;
+ let (remotes_config, _) = app.remote_config().read()?;
let remote_entry = remotes_config
.get(&remote)
.ok_or_else(|| http_err!(NOT_FOUND, "remote '{remote}' not found"))?;
@@ -600,7 +612,7 @@ async fn set_assignment(
entry.pending_clear = false;
}
- pdm_config::subscriptions::save_config(&config)
+ app.subscription_key_config().write(&config)
})
.await??;
rpcenv["digest"] = new_digest.to_hex().into();
@@ -631,6 +643,7 @@ async fn clear_assignment(
key: String,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<(), Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -641,7 +654,7 @@ async fn clear_assignment(
// Authorise against the entry's bound remote BEFORE hitting the network. An operator with
// only PRIV_SYS_MODIFY should not be able to probe live subscription state on a remote
// they cannot audit. The authoritative re-check happens after the lock below.
- let (pre_config, pre_digest) = pdm_config::subscriptions::config()?;
+ let (pre_config, pre_digest) = app.subscription_key_config().read()?;
let pre_entry = pre_config.get(&key);
if let Some(pre_entry) = pre_entry {
if let Some(assigned_remote) = pre_entry.remote.as_deref() {
@@ -662,7 +675,7 @@ async fn clear_assignment(
let pre_binding = pre_entry.and_then(|e| e.remote.as_deref().zip(e.node.as_deref()));
// Owned bool so the orphan guard inside spawn_blocking does not borrow `pre_config`.
let pre_had_binding = pre_binding.is_some();
- let synced_block = check_synced_assignment_for_unassign(&key, pre_binding).await?;
+ let synced_block = check_synced_assignment_for_unassign(&app, &key, pre_binding).await?;
drop(pre_config);
// The lock + sync IO runs on a blocking thread so the async runtime is free for other work
@@ -671,8 +684,8 @@ async fn clear_assignment(
// boundary; reconstructing it is cheap (it just reads the shared ACL cache).
let new_digest = tokio::task::spawn_blocking(move || -> Result<ConfigDigest, Error> {
let user_info = CachedUserInfo::new()?;
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
let Some(stored_entry) = config.get(&key).cloned() else {
@@ -720,7 +733,7 @@ async fn clear_assignment(
// not re-trigger a stale teardown.
entry.pending_clear = false;
- pdm_config::subscriptions::save_config(&config)
+ app.subscription_key_config().write(&config)
})
.await??;
rpcenv["digest"] = new_digest.to_hex().into();
@@ -739,17 +752,19 @@ async fn clear_assignment(
/// parallel rebind between pre-read and here cannot redirect us at a remote the caller has no
/// AUDIT on.
async fn check_synced_assignment_for_unassign(
+ app: &PdmApplication,
key: &str,
binding: Option<(&str, &str)>,
) -> Result<Option<(String, String)>, Error> {
let Some((prev_remote, prev_node)) = binding else {
return Ok(None);
};
- let (remotes_config, _) = pdm_config::remotes::config()?;
+ let (remotes_config, _) = app.remote_config().read()?;
let Some(remote_entry) = remotes_config.get(prev_remote) else {
return Ok(None);
};
- let live = match get_subscription_info_for_remote(remote_entry, FRESH_NODE_STATUS_MAX_AGE).await
+ let live = match get_subscription_info_for_remote(app, remote_entry, FRESH_NODE_STATUS_MAX_AGE)
+ .await
{
Ok(v) => v,
Err(_) => return Ok(None),
@@ -767,13 +782,18 @@ async fn check_synced_assignment_for_unassign(
/// Push a single key to its assigned remote node. Operates on a borrowed `Remote` so the
/// caller can fetch the remotes-config once and reuse it.
-async fn push_key_to_remote(remote: &Remote, key: &str, node_name: &str) -> Result<(), Error> {
+async fn push_key_to_remote(
+ app: &PdmApplication,
+ remote: &Remote,
+ key: &str,
+ node_name: &str,
+) -> Result<(), Error> {
let product_type =
ProductType::from_key(key).ok_or_else(|| format_err!("unrecognised key format: {key}"))?;
match product_type {
ProductType::Pve => {
- let client = crate::connection::make_pve_client(remote)?;
+ let client = app.client_factory().make_pve_client(remote)?;
client
.set_subscription(
node_name,
@@ -784,7 +804,7 @@ async fn push_key_to_remote(remote: &Remote, key: &str, node_name: &str) -> Resu
.await?;
}
ProductType::Pbs => {
- let client = crate::connection::make_pbs_client(remote)?;
+ let client = app.client_factory().make_pbs_client(remote)?;
client
.set_subscription(proxmox_subscription::SetSubscription {
key: key.to_string(),
@@ -806,17 +826,18 @@ async fn push_key_to_remote(remote: &Remote, key: &str, node_name: &str) -> Resu
/// Tear down a node's subscription via the remote's `/nodes/{node}/subscription` endpoint.
async fn delete_subscription_on_remote(
+ app: &PdmApplication,
remote: &Remote,
product_type: ProductType,
node_name: &str,
) -> Result<(), Error> {
match product_type {
ProductType::Pve => {
- let client = crate::connection::make_pve_client(remote)?;
+ let client = app.client_factory().make_pve_client(remote)?;
client.delete_subscription(node_name).await?;
}
ProductType::Pbs => {
- let client = crate::connection::make_pbs_client(remote)?;
+ let client = app.client_factory().make_pbs_client(remote)?;
client.delete_subscription().await?;
}
ProductType::Pmg | ProductType::Pom => {
@@ -831,13 +852,14 @@ async fn delete_subscription_on_remote(
/// Trigger a fresh shop-side subscription check on `remote`/`node` and return once the remote
/// has stored the result. Equivalent to the per-product "Check" button, just driven through PDM.
async fn check_subscription_on_remote(
+ app: &PdmApplication,
remote: &Remote,
product_type: ProductType,
node_name: &str,
) -> Result<(), Error> {
match product_type {
ProductType::Pve => {
- let client = crate::connection::make_pve_client(remote)?;
+ let client = app.client_factory().make_pve_client(remote)?;
client
.update_subscription(
node_name,
@@ -846,7 +868,7 @@ async fn check_subscription_on_remote(
.await?;
}
ProductType::Pbs => {
- let client = crate::connection::make_pbs_client(remote)?;
+ let client = app.client_factory().make_pbs_client(remote)?;
client
.check_subscription(proxmox_subscription::UpdateSubscription { force: Some(true) })
.await?;
@@ -896,6 +918,7 @@ async fn queue_clear(
node: String,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<(), Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -916,8 +939,8 @@ async fn queue_clear(
// The lock + sync IO runs on a blocking thread so the async runtime stays free for other
// work even when /etc/proxmox-datacenter-manager/subscriptions is on slow storage.
let new_digest = tokio::task::spawn_blocking(move || -> Result<ConfigDigest, Error> {
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
let bound_id = config
@@ -943,7 +966,7 @@ async fn queue_clear(
}
entry.pending_clear = true;
- pdm_config::subscriptions::save_config(&config)
+ app.subscription_key_config().write(&config)
})
.await??;
rpcenv["digest"] = new_digest.to_hex().into();
@@ -975,6 +998,7 @@ async fn revert_pending_clear(
node: String,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<(), Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -989,8 +1013,8 @@ async fn revert_pending_clear(
)?;
let new_digest = tokio::task::spawn_blocking(move || -> Result<ConfigDigest, Error> {
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
let bound_id = config
@@ -1012,7 +1036,7 @@ async fn revert_pending_clear(
}
entry.pending_clear = false;
- pdm_config::subscriptions::save_config(&config)
+ app.subscription_key_config().write(&config)
})
.await??;
rpcenv["digest"] = new_digest.to_hex().into();
@@ -1046,6 +1070,7 @@ async fn check_subscription(
remote: String,
node: String,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<(), Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -1059,7 +1084,7 @@ async fn check_subscription(
false,
)?;
- let (remotes_config, _) = pdm_config::remotes::config()?;
+ let (remotes_config, _) = app.remote_config().read()?;
let remote_entry = remotes_config
.get(&remote)
.ok_or_else(|| http_err!(NOT_FOUND, "remote '{remote}' not found"))?;
@@ -1069,10 +1094,10 @@ async fn check_subscription(
pdm_api_types::remotes::RemoteType::Pbs => ProductType::Pbs,
};
- check_subscription_on_remote(remote_entry, product_type, &node)
+ check_subscription_on_remote(&app, remote_entry, product_type, &node)
.await
.map_err(|err| http_err!(BAD_REQUEST, "check failed on {remote}/{node}: {err}"))?;
- invalidate_subscription_info_for_remote(&remote).await;
+ invalidate_subscription_info_for_remote(app.api_cache(), &remote).await;
Ok(())
}
@@ -1115,6 +1140,7 @@ async fn adopt_key(
node: String,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<(), Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -1129,15 +1155,15 @@ async fn adopt_key(
)?;
// Pre-fetch digest to catch a parallel set_assignment during the live read below.
- let (_pre_config, pre_digest) = pdm_config::subscriptions::config()?;
+ let (_pre_config, pre_digest) = app.subscription_key_config().read()?;
// Fetch live state before grabbing the config lock so the network call does not pin the
// lock for the duration of a remote query.
- let (remotes_config, _) = pdm_config::remotes::config()?;
+ let (remotes_config, _) = app.remote_config().read()?;
let remote_entry = remotes_config
.get(&remote)
.ok_or_else(|| http_err!(NOT_FOUND, "remote '{remote}' not found"))?;
- let live = get_subscription_info_for_remote(remote_entry, FRESH_NODE_STATUS_MAX_AGE)
+ let live = get_subscription_info_for_remote(&app, remote_entry, FRESH_NODE_STATUS_MAX_AGE)
.await
.map_err(|err| {
http_err!(
@@ -1159,8 +1185,8 @@ async fn adopt_key(
// The lock + sync IO runs on a blocking thread so the async runtime stays free for other
// work even when /etc/proxmox-datacenter-manager/subscriptions is on slow storage.
let new_digest = tokio::task::spawn_blocking(move || -> Result<ConfigDigest, Error> {
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
if config_digest != pre_digest {
http_bail!(
@@ -1225,7 +1251,7 @@ async fn adopt_key(
config.insert(live_current_key, entry);
}
- pdm_config::subscriptions::save_config(&config)
+ app.subscription_key_config().write(&config)
})
.await??;
rpcenv["digest"] = new_digest.to_hex().into();
@@ -1269,6 +1295,7 @@ async fn adopt_key(
async fn adopt_all(
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<Vec<AdoptedEntry>, Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -1278,7 +1305,7 @@ async fn adopt_all(
// Use a fresh node-status snapshot: a cached entry from minutes ago could miss a live
// subscription that was just installed on a remote, or vice-versa, claim a subscription
// that has since been removed. Adopting bogus or already-cleared keys would be a footgun.
- let node_statuses = collect_node_status(FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
+ let node_statuses = collect_node_status(&app, FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
// Lock + sync IO under spawn_blocking. The closure re-resolves the candidate set under the
// lock: a parallel admin's Assign / Adopt between the network read above and the lock
@@ -1287,8 +1314,8 @@ async fn adopt_all(
let (adopted, new_digest_opt) = tokio::task::spawn_blocking(
move || -> Result<(Vec<AdoptedEntry>, Option<ConfigDigest>), Error> {
let user_info = CachedUserInfo::new()?;
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
let mut adopted: Vec<AdoptedEntry> = Vec::new();
@@ -1374,7 +1401,7 @@ async fn adopt_all(
let new_digest = if adopted.is_empty() {
None
} else {
- Some(pdm_config::subscriptions::save_config(&config)?)
+ Some(app.subscription_key_config().write(&config)?)
};
Ok((adopted, new_digest))
},
@@ -1415,14 +1442,16 @@ async fn adopt_all(
async fn node_status(
max_age: Option<u64>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<Vec<RemoteNodeStatus>, Error> {
- collect_node_status(max_age.unwrap_or(PANEL_NODE_STATUS_MAX_AGE), rpcenv).await
+ collect_node_status(&app, max_age.unwrap_or(PANEL_NODE_STATUS_MAX_AGE), rpcenv).await
}
/// Shared helper: fan out subscription queries to all remotes the caller has audit privilege on,
/// in parallel, reusing the per-remote API cache via `get_subscription_info_for_remote`.
/// Joins the results with the key-pool assignment table.
async fn collect_node_status(
+ app: &PdmApplication,
max_age: u64,
rpcenv: &mut dyn RpcEnvironment,
) -> Result<Vec<RemoteNodeStatus>, Error> {
@@ -1432,18 +1461,17 @@ async fn collect_node_status(
.parse()?;
let user_info = CachedUserInfo::new()?;
- let visible_remotes: Vec<(String, Remote)> =
- RemoteIterator::new(pdm_config::remotes::instance())?
- .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT)
- .into_iter()
- .collect();
+ let visible_remotes: Vec<(String, Remote)> = RemoteIterator::new(app.remote_config())?
+ .any_privs(&user_info, &auth_id, PRIV_RESOURCE_AUDIT)
+ .into_iter()
+ .collect();
- let (keys_config, _) = pdm_config::subscriptions::config()?;
+ let (keys_config, _) = app.subscription_key_config().read()?;
// `get_subscription_info_for_remote` re-uses the per-remote API cache so this
// fan-out is safe to run concurrently.
let fetch = visible_remotes.iter().map(|(name, remote)| async move {
- let res = get_subscription_info_for_remote(remote, max_age).await;
+ let res = get_subscription_info_for_remote(app, remote, max_age).await;
(name.clone(), remote.ty, res)
});
let results = join_all(fetch).await;
@@ -1528,9 +1556,12 @@ async fn collect_node_status(
/// The response carries nested `AutoAssignProposal` data; clients must submit follow-up
/// `bulk_assign` calls with an `application/json` body, the form-urlencoded path cannot encode
/// the nested structure.
-async fn auto_assign(rpcenv: &mut dyn RpcEnvironment) -> Result<AutoAssignProposal, Error> {
- let node_statuses = collect_node_status(FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
- let (config, keys_digest) = pdm_config::subscriptions::config()?;
+async fn auto_assign(
+ rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
+) -> Result<AutoAssignProposal, Error> {
+ let node_statuses = collect_node_status(&app, FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
+ let (config, keys_digest) = app.subscription_key_config().read()?;
let assignments = compute_proposals(&config, &node_statuses);
Ok(AutoAssignProposal {
assignments,
@@ -1566,13 +1597,14 @@ async fn auto_assign(rpcenv: &mut dyn RpcEnvironment) -> Result<AutoAssignPropos
async fn bulk_assign(
proposal: AutoAssignProposal,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<Vec<ProposedAssignment>, Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
.context("no authid available")?
.parse()?;
- let node_statuses = collect_node_status(FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
+ let node_statuses = collect_node_status(&app, FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
let live_digest = hash_node_status(&node_statuses);
if live_digest != proposal.node_status_digest {
http_bail!(
@@ -1587,10 +1619,10 @@ async fn bulk_assign(
let (applied, new_digest_opt) = tokio::task::spawn_blocking(
move || -> Result<(Vec<ProposedAssignment>, Option<ConfigDigest>), Error> {
let user_info = CachedUserInfo::new()?;
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, config_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(Some(&proposal.keys_digest))?;
- let (remotes_config, _) = pdm_config::remotes::config()?;
+ let (remotes_config, _) = app.remote_config().read()?;
let mut applied = Vec::with_capacity(proposal.assignments.len());
for p in &proposal.assignments {
@@ -1652,7 +1684,7 @@ async fn bulk_assign(
let new_digest = if applied.is_empty() {
None
} else {
- Some(pdm_config::subscriptions::save_config(&config)?)
+ Some(app.subscription_key_config().write(&config)?)
};
Ok((applied, new_digest))
},
@@ -1788,6 +1820,7 @@ fn compute_proposals(
async fn apply_pending(
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<Option<String>, Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -1795,11 +1828,11 @@ async fn apply_pending(
.parse()?;
let user_info = CachedUserInfo::new()?;
- let (_, config_digest) = pdm_config::subscriptions::config()?;
+ let (_, config_digest) = app.subscription_key_config().read()?;
config_digest.detect_modification(digest.as_ref())?;
- let node_statuses = collect_node_status(FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
- let pending = compute_pending(&user_info, &auth_id, &node_statuses)?;
+ let node_statuses = collect_node_status(&app, FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
+ let pending = compute_pending(&app, &user_info, &auth_id, &node_statuses)?;
if pending.is_empty() {
return Ok(None);
@@ -1811,7 +1844,7 @@ async fn apply_pending(
None,
auth_id.to_string(),
true,
- move |_worker| async move { run_apply_pending(worker_auth).await },
+ move |_worker| async move { run_apply_pending(&app, worker_auth).await },
)?;
Ok(Some(upid))
@@ -1822,12 +1855,12 @@ async fn apply_pending(
/// The worker re-reads remotes and the pool config so a reassign or removal between the API call
/// returning a UPID and the worker firing is honoured (pushing the old key to a node after the
/// operator retracted the assignment was a real footgun).
-async fn run_apply_pending(auth_id: Authid) -> Result<(), Error> {
+async fn run_apply_pending(app: &PdmApplication, auth_id: Authid) -> Result<(), Error> {
let user_info = CachedUserInfo::new()?;
- let (remotes_config, _) = pdm_config::remotes::config()?;
+ let (remotes_config, _) = app.remote_config().read()?;
- let node_statuses = collect_status_uncached(&remotes_config).await;
- let pending = compute_pending(&user_info, &auth_id, &node_statuses)?;
+ let node_statuses = collect_status_uncached(app, &remotes_config).await;
+ let pending = compute_pending(app, &user_info, &auth_id, &node_statuses)?;
if pending.is_empty() {
info!("apply-pending: nothing to do (state changed since the API call)");
@@ -1845,7 +1878,7 @@ async fn run_apply_pending(auth_id: Authid) -> Result<(), Error> {
// branch) makes the at-start snapshot stale, and a parallel admin's Discard Pending
// between worker start and this iteration must cancel a planned op rather than have us
// execute it against a flag the operator just retracted.
- let (config, _) = pdm_config::subscriptions::config()?;
+ let (config, _) = app.subscription_key_config().read()?;
if !pool_assignment_still_valid(&config, &entry) {
info!(
"skipping {}/{}: pool entry changed before worker ran",
@@ -1885,7 +1918,7 @@ async fn run_apply_pending(auth_id: Authid) -> Result<(), Error> {
continue;
};
info!("pushing {redacted} to {}/{}...", entry.remote, entry.node);
- if let Err(err) = push_key_to_remote(remote, &entry.key, &entry.node).await {
+ if let Err(err) = push_key_to_remote(app, remote, &entry.key, &entry.node).await {
warn!(
"push of {redacted} to {}/{} failed: {err}",
entry.remote, entry.node
@@ -1914,7 +1947,7 @@ async fn run_apply_pending(auth_id: Authid) -> Result<(), Error> {
entry.remote, entry.node
);
if let Err(err) =
- delete_subscription_on_remote(remote, product_type, &entry.node).await
+ delete_subscription_on_remote(app, remote, product_type, &entry.node).await
{
warn!(
"clear of {redacted} on {}/{} failed: {err}",
@@ -1937,9 +1970,10 @@ async fn run_apply_pending(auth_id: Authid) -> Result<(), Error> {
let entry_key = entry.key.clone();
let entry_remote = entry.remote.clone();
let entry_node = entry.node.clone();
+ let app = app.clone();
let pool_update = tokio::task::spawn_blocking(move || -> Result<(), Error> {
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut updated, _) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut updated, _) = app.subscription_key_config().read()?;
if let Some(stored) = updated.get_mut(&entry_key) {
if stored.remote.as_deref() == Some(entry_remote.as_str())
&& stored.node.as_deref() == Some(entry_node.as_str())
@@ -1950,7 +1984,7 @@ async fn run_apply_pending(auth_id: Authid) -> Result<(), Error> {
}
}
// Worker context: no `rpcenv` to set, post-save digest is unused here.
- let _ = pdm_config::subscriptions::save_config(&updated)?;
+ let _ = app.subscription_key_config().write(&updated)?;
Ok(())
})
.await
@@ -1970,7 +2004,7 @@ async fn run_apply_pending(auth_id: Authid) -> Result<(), Error> {
}
}
info!(" success");
- invalidate_subscription_info_for_remote(&entry.remote).await;
+ invalidate_subscription_info_for_remote(app.api_cache(), &entry.remote).await;
ok += 1;
}
@@ -2031,6 +2065,7 @@ async fn run_apply_pending(auth_id: Authid) -> Result<(), Error> {
async fn clear_pending(
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
+ #[state] app: PdmApplication,
) -> Result<ClearPendingResult, Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
@@ -2038,11 +2073,11 @@ async fn clear_pending(
.parse()?;
let user_info = CachedUserInfo::new()?;
- let (_, pre_digest) = pdm_config::subscriptions::config()?;
+ let (_, pre_digest) = app.subscription_key_config().read()?;
pre_digest.detect_modification(digest.as_ref())?;
- let node_statuses = collect_node_status(FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
- let pending = compute_pending(&user_info, &auth_id, &node_statuses)?;
+ let node_statuses = collect_node_status(&app, FRESH_NODE_STATUS_MAX_AGE, rpcenv).await?;
+ let pending = compute_pending(&app, &user_info, &auth_id, &node_statuses)?;
if pending.is_empty() {
return Ok(ClearPendingResult { cleared: 0 });
@@ -2052,8 +2087,8 @@ async fn clear_pending(
// operations.
let (cleared, new_digest_opt) =
tokio::task::spawn_blocking(move || -> Result<(u32, Option<ConfigDigest>), Error> {
- let _lock = pdm_config::subscriptions::lock_config()?;
- let (mut config, locked_digest) = pdm_config::subscriptions::config()?;
+ let _lock = app.subscription_key_config().lock()?;
+ let (mut config, locked_digest) = app.subscription_key_config().read()?;
locked_digest.detect_modification(digest.as_ref())?;
let mut cleared: u32 = 0;
@@ -2087,7 +2122,7 @@ async fn clear_pending(
}
let new_digest = if cleared > 0 {
- Some(pdm_config::subscriptions::save_config(&config)?)
+ Some(app.subscription_key_config().write(&config)?)
} else {
None
};
@@ -2120,11 +2155,12 @@ enum PendingOp {
}
fn compute_pending(
+ app: &PdmApplication,
user_info: &CachedUserInfo,
auth_id: &Authid,
node_statuses: &[RemoteNodeStatus],
) -> Result<Vec<PendingEntry>, Error> {
- let (config, _) = pdm_config::subscriptions::config()?;
+ let (config, _) = app.subscription_key_config().read()?;
Ok(config
.iter()
@@ -2182,10 +2218,11 @@ fn pool_assignment_still_valid(
/// Like [`collect_node_status`] but bypasses the auth filter, for the apply-pending worker
/// which gates each entry through its own per-remote priv check based on the persisted pool plan.
async fn collect_status_uncached(
+ app: &PdmApplication,
remotes_config: &SectionConfigData<Remote>,
) -> Vec<RemoteNodeStatus> {
let fetch = remotes_config.iter().map(|(name, remote)| async move {
- let res = get_subscription_info_for_remote(remote, FRESH_NODE_STATUS_MAX_AGE).await;
+ let res = get_subscription_info_for_remote(app, remote, FRESH_NODE_STATUS_MAX_AGE).await;
(name.to_string(), remote.ty, res)
});
let results = join_all(fetch).await;
diff --git a/server/src/bin/proxmox-datacenter-manager-daily-update.rs b/server/src/bin/proxmox-datacenter-manager-daily-update.rs
index 314b3399..289cf2a2 100644
--- a/server/src/bin/proxmox-datacenter-manager-daily-update.rs
+++ b/server/src/bin/proxmox-datacenter-manager-daily-update.rs
@@ -6,7 +6,7 @@ use proxmox_router::{ApiHandler, RpcEnvironment, cli::*};
use proxmox_subscription::SubscriptionStatus;
use proxmox_sys::fs::CreateOptions;
-use server::api;
+use server::{api, context::PdmApplication};
async fn wait_for_local_worker(upid_str: &str) -> Result<(), Error> {
let upid: pbs_api_types::UPID = upid_str.parse()?;
@@ -22,11 +22,11 @@ async fn wait_for_local_worker(upid_str: &str) -> Result<(), Error> {
}
/// Daily update
-async fn do_update(rpcenv: &mut dyn RpcEnvironment) -> Result<(), Error> {
- if let Err(err) = &api::nodes::subscription::check_subscription().await {
+async fn do_update(rpcenv: &mut dyn RpcEnvironment, app: PdmApplication) -> Result<(), Error> {
+ if let Err(err) = &api::nodes::subscription::check_subscription(app.clone()).await {
log::error!("Error checking subscription - {err}");
}
- match api::nodes::subscription::get_subscription().await {
+ match api::nodes::subscription::get_subscription(app).await {
Ok(info) if info.info.status == SubscriptionStatus::Active => {}
Ok(info) => {
log::warn!(
@@ -101,9 +101,9 @@ async fn run(rpcenv: &mut dyn RpcEnvironment) -> Result<(), Error> {
proxmox_product_config::init(pdm_config::api_user()?, pdm_config::priv_user()?);
proxmox_acme_api::init(pdm_buildcfg::configdir!("/acme"), false)?;
- server::context::init()?;
+ let app = server::context::init()?;
- do_update(rpcenv).await
+ do_update(rpcenv, app).await
}
fn main() {
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 20/21] pdm-config: subscriptions: drop unused accessor functions
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (18 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 19/21] api: subscriptions: use PdmApplication instead of globals Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
2026-08-27 11:42 ` [PATCH datacenter-manager v3 21/21] tests: add example tests for remote subscription management Lukas Wagner
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
Now that the subscription API handlers reach the config through
PdmApplication instead, the free functions built on top of the global
INSTANCE static are unused. Drop them along with the init() call that
used to populate it.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
lib/pdm-config/src/subscriptions.rs | 39 -----------------------------
server/src/context/mod.rs | 1 -
2 files changed, 40 deletions(-)
diff --git a/lib/pdm-config/src/subscriptions.rs b/lib/pdm-config/src/subscriptions.rs
index 5be88a13..c4b99d25 100644
--- a/lib/pdm-config/src/subscriptions.rs
+++ b/lib/pdm-config/src/subscriptions.rs
@@ -7,8 +7,6 @@
//! entries, which is intended as future proofing for a more automated (shop) import without having
//! to adapt the data layer.
-use std::sync::OnceLock;
-
use anyhow::Error;
use proxmox_config_digest::ConfigDigest;
@@ -25,37 +23,6 @@ pub const SUBSCRIPTIONS_CFG_FILENAME: &str = configdir!("/subscriptions/keys.cfg
const SUBSCRIPTIONS_SHADOW_FILENAME: &str = configdir!("/subscriptions/keys.shadow");
pub const SUBSCRIPTIONS_CFG_LOCKFILE: &str = configdir!("/subscriptions/.keys.lock");
-static INSTANCE: OnceLock<Box<dyn SubscriptionKeyConfig + Send + Sync>> = OnceLock::new();
-
-fn instance() -> &'static (dyn SubscriptionKeyConfig + Send + Sync) {
- INSTANCE
- .get()
- .expect("subscription key config not initialized")
- .as_ref()
-}
-
-pub fn lock_config() -> Result<ApiLockGuard, Error> {
- instance().lock()
-}
-
-pub fn config() -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error> {
- instance().read()
-}
-
-pub fn shadow_config() -> Result<SectionConfigData<SubscriptionKeyShadow>, Error> {
- instance().read_shadow()
-}
-
-pub fn save_config(
- config: &SectionConfigData<SubscriptionKeyEntry>,
-) -> Result<ConfigDigest, Error> {
- instance().write(config)
-}
-
-pub fn save_shadow(shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error> {
- instance().write_shadow(shadow)
-}
-
pub trait SubscriptionKeyConfig {
fn read(&self) -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error>;
fn read_shadow(&self) -> Result<SectionConfigData<SubscriptionKeyShadow>, Error>;
@@ -110,9 +77,3 @@ impl SubscriptionKeyConfig for DefaultSubscriptionKeyConfig {
replace_secret_config(SUBSCRIPTIONS_SHADOW_FILENAME, raw.as_bytes())
}
}
-
-pub fn init(instance: Box<dyn SubscriptionKeyConfig + Send + Sync>) {
- if INSTANCE.set(instance).is_err() {
- panic!("subscription key config instance already set");
- }
-}
diff --git a/server/src/context/mod.rs b/server/src/context/mod.rs
index 082ab8de..95238030 100644
--- a/server/src/context/mod.rs
+++ b/server/src/context/mod.rs
@@ -34,7 +34,6 @@ pub fn init() -> Result<PdmApplication, Error> {
// anyway, and the implementation is stateless, so having this second
// instance is not an issue *currently*.
pdm_config::remotes::init(factory.make_remote_config()?);
- pdm_config::subscriptions::init(factory.make_subscription_key_config()?);
Ok(app)
}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread* [PATCH datacenter-manager v3 21/21] tests: add example tests for remote subscription management
2026-08-27 11:42 [PATCH datacenter-manager/proxmox v3 00/21] inject application context via API macro for easier integration testing Lukas Wagner
` (19 preceding siblings ...)
2026-08-27 11:42 ` [PATCH datacenter-manager v3 20/21] pdm-config: subscriptions: drop unused accessor functions Lukas Wagner
@ 2026-08-27 11:42 ` Lukas Wagner
20 siblings, 0 replies; 22+ messages in thread
From: Lukas Wagner @ 2026-08-27 11:42 UTC (permalink / raw)
To: pdm-devel
This is obviously pretty incomplete, but it should demonstrate how easy
it is to write tests now, given the injected context.
Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
server/src/api/nodes/tasks.rs | 2 +-
server/src/api/subscriptions/mod.rs | 12 +--
server/tests/common/environment.rs | 9 +-
server/tests/common/mod.rs | 59 ++++++++++--
server/tests/common/test_application.rs | 52 +++++++++-
server/tests/test_subscriptions.rs | 122 ++++++++++++++++++++++++
6 files changed, 238 insertions(+), 18 deletions(-)
create mode 100644 server/tests/test_subscriptions.rs
diff --git a/server/src/api/nodes/tasks.rs b/server/src/api/nodes/tasks.rs
index 31ddb8f1..4c28880f 100644
--- a/server/src/api/nodes/tasks.rs
+++ b/server/src/api/nodes/tasks.rs
@@ -277,7 +277,7 @@ fn stop_task(upid: UPID, rpcenv: &mut dyn RpcEnvironment) -> Result<(), Error> {
},
)]
/// Get task status.
-async fn get_task_status(upid: UPID, rpcenv: &mut dyn RpcEnvironment) -> Result<Value, Error> {
+pub async fn get_task_status(upid: UPID, rpcenv: &mut dyn RpcEnvironment) -> Result<Value, Error> {
let auth_id: Authid = rpcenv
.get_auth_id()
.context("no authid available")?
diff --git a/server/src/api/subscriptions/mod.rs b/server/src/api/subscriptions/mod.rs
index 370941a2..d9dfb1a2 100644
--- a/server/src/api/subscriptions/mod.rs
+++ b/server/src/api/subscriptions/mod.rs
@@ -132,7 +132,7 @@ fn key_not_found(key: &str) -> Error {
/// additionally gated on per-remote `PRIV_RESOURCE_AUDIT` so that an operator who can audit the
/// pool but not a specific remote does not learn which keys are pinned to it (and through that,
/// the existence and rough size of that remote's deployment).
-fn list_keys(
+pub fn list_keys(
rpcenv: &mut dyn RpcEnvironment,
#[state] app: PdmApplication,
) -> Result<Vec<SubscriptionKeyEntry>, Error> {
@@ -194,7 +194,7 @@ fn list_keys(
///
/// The post-save digest is set on the response so clients can chain a follow-up mutation without
/// a refetch round-trip.
-async fn add_keys(
+pub async fn add_keys(
keys: Vec<String>,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
@@ -276,7 +276,7 @@ async fn add_keys(
/// Bound entries are hidden from operators who cannot audit the bound remote (mirrors the
/// `list_keys` filter); the response is the same 404 either way so a probe cannot distinguish
/// "key exists but you cannot see it" from "key not in pool".
-fn get_key(
+pub fn get_key(
key: String,
rpcenv: &mut dyn RpcEnvironment,
#[state] app: PdmApplication,
@@ -328,7 +328,7 @@ fn get_key(
/// another admin had pinned. Refuses if the key is currently the live active key on its bound
/// node, since dropping the pool entry would orphan that subscription on the remote: the
/// operator must run Clear Key on the Node Subscription Status panel first.
-async fn delete_key(
+pub async fn delete_key(
key: String,
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
@@ -462,7 +462,7 @@ async fn delete_key(
/// `PRIV_SYS_MODIFY` lets the caller touch the pool config; per-remote `PRIV_RESOURCE_MODIFY`
/// is enforced inside this handler so an operator cannot push a key to a remote they have no
/// other authority on.
-async fn set_assignment(
+pub async fn set_assignment(
key: String,
remote: String,
node: String,
@@ -1817,7 +1817,7 @@ fn compute_proposals(
/// no longer sees. The worker itself deliberately re-reads the pool when it fires (a worker can
/// be scheduled with delay), so a parallel admin edit between API return and worker firing is
/// still honoured - the digest only pins the at-API-call-time plan, not the executed plan.
-async fn apply_pending(
+pub async fn apply_pending(
digest: Option<ConfigDigest>,
rpcenv: &mut dyn RpcEnvironment,
#[state] app: PdmApplication,
diff --git a/server/tests/common/environment.rs b/server/tests/common/environment.rs
index 5dad7b21..dbfb2918 100644
--- a/server/tests/common/environment.rs
+++ b/server/tests/common/environment.rs
@@ -1,14 +1,17 @@
use proxmox_router::RpcEnvironment;
-pub struct TestRpcEnvironment;
+/// An [`RpcEnvironment`] that can be used in integration tests.
+pub struct TestRpcEnvironment {
+ pub attribs: serde_json::Value,
+}
impl RpcEnvironment for TestRpcEnvironment {
fn result_attrib_mut(&mut self) -> &mut serde_json::Value {
- unimplemented!()
+ &mut self.attribs
}
fn result_attrib(&self) -> &serde_json::Value {
- unimplemented!()
+ &self.attribs
}
fn env_type(&self) -> proxmox_router::RpcEnvironmentType {
diff --git a/server/tests/common/mod.rs b/server/tests/common/mod.rs
index 38697353..058f0a20 100644
--- a/server/tests/common/mod.rs
+++ b/server/tests/common/mod.rs
@@ -1,9 +1,8 @@
-use std::{
- path::{Path, PathBuf},
- sync::{Once, OnceLock},
-};
+use std::path::{Path, PathBuf};
+use std::sync::{Once, OnceLock};
+use std::time::Duration;
-use anyhow::Context;
+use anyhow::{Context, Error, bail};
use serde::de::DeserializeOwned;
use proxmox_sys::fs::CreateOptions;
@@ -60,5 +59,53 @@ pub fn test_setup() {
/// Create a [`TestRpcEnvironment`] for the use in a test.
pub fn rpcenv() -> TestRpcEnvironment {
- TestRpcEnvironment
+ TestRpcEnvironment {
+ attribs: serde_json::json!({}),
+ }
+}
+
+#[allow(unused)]
+macro_rules! assert_http_error {
+ ($result:expr, $status:expr $(, $expected:expr)? $(,)?) => {{
+ let err = $result.unwrap_err();
+ let http_err = err.downcast_ref::<proxmox_router::HttpError>().unwrap();
+
+ assert_eq!(http_err.code, $status);
+
+ $(
+ let err_text = err.to_string();
+ assert!(
+ err_text.contains($expected),
+ "'{}' not contained in '{}'",
+ $expected,
+ err_text,
+ );
+ )?
+ }};
+}
+
+#[allow(unused)]
+pub(crate) use assert_http_error;
+
+/// Wait for a task to finish.
+///
+/// At the moment, this uses a hard-coded timeout of 5 seconds.
+pub async fn wait_for_task(upid: &str) -> Result<String, Error> {
+ let upid = upid.parse::<pdm_api_types::UPID>()?;
+
+ for _ in 0..50 {
+ let response =
+ server::api::nodes::tasks::get_task_status(upid.clone(), &mut rpcenv()).await?;
+
+ if response["status"].as_str().unwrap() != "running" {
+ return Ok(response["exitstatus"]
+ .as_str()
+ .context("expected exitstatus to be a string")?
+ .into());
+ }
+
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ }
+
+ bail!("worker did not finish after timeout");
}
diff --git a/server/tests/common/test_application.rs b/server/tests/common/test_application.rs
index 83b35bf6..054cfb8e 100644
--- a/server/tests/common/test_application.rs
+++ b/server/tests/common/test_application.rs
@@ -1,6 +1,5 @@
use std::collections::HashMap;
-use std::sync::Arc;
-use std::sync::Mutex;
+use std::sync::{Arc, Mutex};
use anyhow::{Context, Error, bail};
use nix::unistd::User;
@@ -8,12 +7,15 @@ use serde::{Serialize, de::DeserializeOwned};
use pbs_api_types::Authid;
use proxmox_client::Client;
+use proxmox_product_config::create_mocked_lock;
use proxmox_product_config::{ProductConfig, ProductConfigParams};
use proxmox_section_config::typed::SectionConfigData;
use pdm_api_types::ConfigDigest;
use pdm_api_types::remotes::{Remote, RemoteType};
+use pdm_api_types::subscription::{SubscriptionKeyEntry, SubscriptionKeyShadow};
use pdm_config::remotes::RemoteConfig;
+use pdm_config::subscriptions::SubscriptionKeyConfig;
use server::connection::{ClientFactory, PveClient};
use server::context::ContextFactory;
@@ -137,6 +139,12 @@ impl ContextFactory for TestApplication {
Ok(Box::new(self.clone()))
}
+ fn make_subscription_key_config(
+ &self,
+ ) -> Result<Box<dyn SubscriptionKeyConfig + Send + Sync>, Error> {
+ Ok(Box::new(TestSubscriptionKeyConfig::default()))
+ }
+
fn make_product_config(&self) -> Result<ProductConfig, Error> {
let user = User::from_uid(nix::unistd::getuid())
.ok()
@@ -241,6 +249,46 @@ impl ClientFactory for TestApplication {
}
}
+#[derive(Default)]
+struct TestSubscriptionKeyConfig {
+ shadow: Mutex<SectionConfigData<SubscriptionKeyShadow>>,
+ config: Mutex<SectionConfigData<SubscriptionKeyEntry>>,
+}
+
+impl SubscriptionKeyConfig for TestSubscriptionKeyConfig {
+ fn read(&self) -> Result<(SectionConfigData<SubscriptionKeyEntry>, ConfigDigest), Error> {
+ Ok((
+ self.config.lock().unwrap().clone(),
+ ConfigDigest::from_slice([]),
+ ))
+ }
+
+ fn read_shadow(&self) -> Result<SectionConfigData<SubscriptionKeyShadow>, Error> {
+ Ok(self.shadow.lock().unwrap().clone())
+ }
+
+ fn lock(&self) -> Result<proxmox_product_config::ApiLockGuard, Error> {
+ Ok(unsafe { create_mocked_lock() })
+ }
+
+ fn write(
+ &self,
+ config: &SectionConfigData<SubscriptionKeyEntry>,
+ ) -> Result<ConfigDigest, Error> {
+ let mut guard = self.config.lock().unwrap();
+ *guard = config.clone();
+
+ Ok(ConfigDigest::from_slice([]))
+ }
+
+ fn write_shadow(&self, shadow: &SectionConfigData<SubscriptionKeyShadow>) -> Result<(), Error> {
+ let mut guard = self.shadow.lock().unwrap();
+ *guard = shadow.clone();
+
+ Ok(())
+ }
+}
+
macro_rules! test_pve_client {
($ty:ident { $($overrides:tt)* }) => {
diff --git a/server/tests/test_subscriptions.rs b/server/tests/test_subscriptions.rs
new file mode 100644
index 00000000..58dcb546
--- /dev/null
+++ b/server/tests/test_subscriptions.rs
@@ -0,0 +1,122 @@
+use http::StatusCode;
+use pdm_api_types::subscription::{ProductType, SubscriptionLevel};
+use pve_api_types::SetSubscription;
+
+use crate::common::{TestApplication, TestRemoteState, rpcenv};
+
+use server::{api, context::ContextFactory};
+
+pub mod common;
+
+#[tokio::test]
+async fn test_manage_inventory() {
+ common::test_setup();
+
+ let test_app = TestApplication::new();
+ let app = test_app.make_pdm_application().unwrap();
+
+ let existing_keys = api::subscriptions::list_keys(&mut rpcenv(), app.clone()).unwrap();
+ assert!(existing_keys.is_empty());
+
+ let keys = vec![
+ "pve4c-aaaaaaaaaa".into(),
+ "pve4c-bbbbbbbbbb".into(),
+ "pve4c-bbbbbbbbbb".into(),
+ ];
+
+ let add_result = api::subscriptions::add_keys(keys, None, &mut rpcenv(), app.clone())
+ .await
+ .unwrap();
+
+ assert_eq!(add_result.added, 2);
+ assert_eq!(add_result.deduplicated, 1);
+
+ let existing_keys = api::subscriptions::list_keys(&mut rpcenv(), app.clone()).unwrap();
+ assert_eq!(existing_keys.len(), 2);
+
+ let key =
+ api::subscriptions::get_key("pve4c-aaaaaaaaaa".into(), &mut rpcenv(), app.clone()).unwrap();
+
+ assert_eq!(key.key, "pve4c-aaaaaaaaaa");
+ assert_eq!(key.product_type, ProductType::Pve);
+ assert_eq!(key.level, SubscriptionLevel::Community);
+ assert_eq!(key.remote, None);
+
+ api::subscriptions::delete_key("pve4c-aaaaaaaaaa".into(), None, &mut rpcenv(), app.clone())
+ .await
+ .unwrap();
+
+ let result = api::subscriptions::get_key("pve4c-aaaaaaaaaa".into(), &mut rpcenv(), app.clone());
+
+ common::assert_http_error!(result, StatusCode::NOT_FOUND, "not found in pool");
+}
+
+#[tokio::test]
+async fn test_invalid_key_format() {
+ common::test_setup();
+
+ let test_app = TestApplication::new();
+ let app = test_app.make_pdm_application().unwrap();
+
+ let keys = vec!["aaaaa".into()];
+
+ let result = api::subscriptions::add_keys(keys, None, &mut rpcenv(), app.clone()).await;
+
+ common::assert_http_error!(result, StatusCode::BAD_REQUEST, "unrecognised key format");
+}
+
+common::test_pve_client!(SubscriptionClient {
+ async fn set_subscription(&self, node: &str, params: SetSubscription) -> Result<(), proxmox_client::Error> {
+ self.state().set(format!("set_subscription-{node}"), ¶ms);
+
+ Ok(())
+ }
+});
+
+#[tokio::test]
+async fn test_apply_key() {
+ common::test_setup();
+
+ let remote_state = TestRemoteState::default();
+
+ let test_app = TestApplication::new().with_pve_remote(
+ "remote-a",
+ remote_state.clone(),
+ SubscriptionClient,
+ );
+
+ let app = test_app.make_pdm_application().unwrap();
+
+ let keys = vec![
+ "pve4c-aaaaaaaaaa".into(),
+ "pve4c-bbbbbbbbbb".into(),
+ "pve4c-bbbbbbbbbb".into(),
+ ];
+ let _ = api::subscriptions::add_keys(keys.clone(), None, &mut rpcenv(), app.clone())
+ .await
+ .unwrap();
+
+ api::subscriptions::set_assignment(
+ keys[0].clone(),
+ "remote-a".into(),
+ "remote-a-node-0".into(),
+ None,
+ &mut rpcenv(),
+ app.clone(),
+ )
+ .await
+ .unwrap();
+
+ let upid = api::subscriptions::apply_pending(None, &mut rpcenv(), app.clone())
+ .await
+ .unwrap()
+ .unwrap();
+
+ assert_eq!(common::wait_for_task(&upid).await.unwrap(), "OK");
+
+ let set_subscription: SetSubscription = remote_state
+ .get("set_subscription-remote-a-node-0")
+ .expect("subscription was set on remote-a-node-0");
+
+ assert_eq!(set_subscription.key, keys[0]);
+}
--
2.47.3
^ permalink raw reply related [flat|nested] 22+ messages in thread