Re: [PATCH proxmox-backup 1/2] fix #6373: HTTP level reader heartbeat for proxy connection keepalive

["Fabian Grünbichler" <f.gruenbichler@proxmox.com>]

On January 29, 2026 1:26 pm, Christian Ebner wrote:
> Backup readers can have long periods of idle connections, e.g. if a
> backup snapshot has been mounted and all relevant chunks are locally
> cached or a backup session with previous metadata archive not needing
> to fetch new contents while the backup is ongoing.
> 
> Proxies like e.g. HAProxy might however close idle connections for
> better resource handling [0,1], even multiplexed HTTP/2 connections as
> are being used for the Proxmox Backup Sever backup/reader protocol.
> 
> This mainly affects the backup reader, while the backup writer will
> do indexing and chunk uploads anyways.

but if the storage is slow, there might not be chunk traffic for a few
seconds as well?

> Therefore, perform heartbeat traffic in the backup reader http2 client
> when no other requests are being send. To do so, an new `heartbeat`
> API endpoint is introduced as part of the backup reader protocol,
> returning empty on GET requests. By making this part of the backup
> reader protocol, this is limited to active reader HTTP/2 sessions.
> 
> On the client side the heartbeat is send out periodically whenever a
> timeout is being reached, the timeout however being reset if other
> requests are being performed via the http2 client.
> 
> Since older servers do not provide the new API path, ignore errors
> as the response is not strictly necessary for the connection to
> remain established.
> 
> The timeout is currently only being performed if the timeout value
> in seconds is given via the PBS_READER_HEARTBEAT_TIMEOUT.

s/timeout/heartbeat/ ?

> 
> Testing was performed using HAProxy with the Proxmox Backup Server
> as backend using the following 5 second connection idle timeouts
> as configuration parameters in haproxy.cfg:
> ```
> ...
> defaults
>     ...
>     timeout connect 5000
>     timeout client  5000
>     timeout server  5000
>     ..
> 
> frontend http-in
>     bind *:8007
>     mode tcp
>     default_backend pbs
> 
> backend pbs
>     mode tcp
>     http-reuse always
>     server pbs <PBS-IP>:8007 verify none
> ```
> 
> As command invocation:
> ```
> PBS_READER_HEARTBEAT_TIMEOUT=1 proxmox-backup-client mount <snapshot> <archive> \
> <mountpoint> --repository <user-and-realm>@<PROXY-IP>:<datastore> --verbose
> ```
> 
> [0] https://www.haproxy.com/documentation/haproxy-configuration-manual/latest/#4-timeout%20client
> [1] https://www.haproxy.com/documentation/haproxy-configuration-manual/latest/#4.2-timeout%20http-keep-alive
> 
> Fixes: https://bugzilla.proxmox.com/show_bug.cgi?id=6373
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
>  pbs-client/src/backup_reader.rs | 69 ++++++++++++++++++++++++++++++---
>  src/api2/reader/mod.rs          |  9 ++++-
>  2 files changed, 72 insertions(+), 6 deletions(-)
> 
> diff --git a/pbs-client/src/backup_reader.rs b/pbs-client/src/backup_reader.rs
> index 88cba599b..95236bef2 100644
> --- a/pbs-client/src/backup_reader.rs
> +++ b/pbs-client/src/backup_reader.rs
> @@ -1,10 +1,13 @@
>  use anyhow::{format_err, Error};
>  use std::fs::File;
>  use std::io::{Seek, SeekFrom, Write};
> +use std::str::FromStr;
>  use std::sync::Arc;
> +use std::time::Duration;
>  
>  use futures::future::AbortHandle;
>  use serde_json::{json, Value};
> +use tokio::sync::mpsc;
>  
>  use pbs_api_types::{BackupArchiveName, BackupDir, BackupNamespace, MANIFEST_BLOB_NAME};
>  use pbs_datastore::data_blob::DataBlob;
> @@ -18,26 +21,65 @@ use pbs_tools::sha::sha256;
>  
>  use super::{H2Client, HttpClient};
>  
> +enum HeartBeatMsg {
> +    Reset,
> +    Abort,
> +}
> +
>  /// Backup Reader
>  pub struct BackupReader {
>      h2: H2Client,
>      abort: AbortHandle,
>      crypt_config: Option<Arc<CryptConfig>>,
> +    heartbeat: Option<mpsc::Sender<HeartBeatMsg>>,
>  }
>  
>  impl Drop for BackupReader {
>      fn drop(&mut self) {
> +        self.send_msg_heartbeat(HeartBeatMsg::Abort);
>          self.abort.abort();
>      }
>  }
>  
>  impl BackupReader {
>      fn new(h2: H2Client, abort: AbortHandle, crypt_config: Option<Arc<CryptConfig>>) -> Arc<Self> {
> -        Arc::new(Self {
> -            h2,
> -            abort,
> -            crypt_config,
> -        })
> +        let timeout = match std::env::var("PBS_READER_HEARTBEAT_TIMEOUT") {
> +            Ok(val) => u64::from_str(&val).map(Duration::from_secs).ok(),

if the var is set but contains a bogus value, shouldn't that be an
error?

> +            Err(_err) => None,
> +        };
> +
> +        if let Some(timeout) = timeout {
> +            let (send, mut recv) = mpsc::channel(1);
> +            let backup_reader = Arc::new(Self {
> +                h2,
> +                abort,
> +                crypt_config,
> +                heartbeat: Some(send),
> +            });
> +            let reader_cloned = Arc::clone(&backup_reader);
> +
> +            tokio::spawn(async move {
> +                loop {
> +                    match tokio::time::timeout(timeout, recv.recv()).await {
> +                        Ok(Some(HeartBeatMsg::Reset)) => (),
> +                        Ok(Some(HeartBeatMsg::Abort)) | Ok(None) => break,
> +                        Err(_elapsed) => {
> +                            // connection idle timeout reached, send heatbeat
> +                            let _ = reader_cloned.h2.get("heartbeat", None).await;
> +                        }
> +                    }
> +                }
> +            });
> +
> +            backup_reader
> +        } else {
> +            Arc::new(Self {
> +                h2,
> +                abort,
> +                crypt_config,
> +                heartbeat: None,
> +            })
> +        }
>      }
>  
>      /// Create a new instance by upgrading the connection at '/api2/json/reader'
> @@ -79,21 +121,25 @@ impl BackupReader {
>  
>      /// Execute a GET request
>      pub async fn get(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
> +        self.send_msg_heartbeat(HeartBeatMsg::Reset);
>          self.h2.get(path, param).await
>      }
>  
>      /// Execute a PUT request
>      pub async fn put(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
> +        self.send_msg_heartbeat(HeartBeatMsg::Reset);
>          self.h2.put(path, param).await
>      }
>  
>      /// Execute a POST request
>      pub async fn post(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
> +        self.send_msg_heartbeat(HeartBeatMsg::Reset);
>          self.h2.post(path, param).await
>      }
>  
>      /// Execute a GET request and send output to a writer
>      pub async fn download<W: Write + Send>(&self, file_name: &str, output: W) -> Result<(), Error> {
> +        self.send_msg_heartbeat(HeartBeatMsg::Reset);
>          let path = "download";
>          let param = json!({ "file-name": file_name });
>          self.h2.download(path, Some(param), output).await
> @@ -103,6 +149,7 @@ impl BackupReader {
>      ///
>      /// This writes random data, and is only useful to test download speed.
>      pub async fn speedtest<W: Write + Send>(&self, output: W) -> Result<(), Error> {
> +        self.send_msg_heartbeat(HeartBeatMsg::Reset);
>          self.h2.download("speedtest", None, output).await
>      }
>  
> @@ -112,12 +159,14 @@ impl BackupReader {
>          digest: &[u8; 32],
>          output: W,
>      ) -> Result<(), Error> {
> +        self.send_msg_heartbeat(HeartBeatMsg::Reset);
>          let path = "chunk";
>          let param = json!({ "digest": hex::encode(digest) });
>          self.h2.download(path, Some(param), output).await
>      }
>  
>      pub fn force_close(self) {
> +        self.send_msg_heartbeat(HeartBeatMsg::Abort);
>          self.abort.abort();
>      }
>  
> @@ -205,4 +254,14 @@ impl BackupReader {
>  
>          Ok(index)
>      }
> +
> +    /// Send given message to the heartbeat closure.
> +    ///
> +    /// All errors are being ignored, since they cannot be handled anyways.
> +    fn send_msg_heartbeat(&self, msg: HeartBeatMsg) {
> +        let _ = self
> +            .heartbeat
> +            .as_ref()
> +            .map(|heartbeat| heartbeat.try_send(msg));
> +    }
>  }
> diff --git a/src/api2/reader/mod.rs b/src/api2/reader/mod.rs
> index f7adc366f..1c673d7a6 100644
> --- a/src/api2/reader/mod.rs
> +++ b/src/api2/reader/mod.rs
> @@ -18,7 +18,7 @@ use proxmox_router::{
>      http_err, list_subdirs_api_method, ApiHandler, ApiMethod, ApiResponseFuture, Permission,
>      Router, RpcEnvironment, SubdirMap,
>  };
> -use proxmox_schema::{BooleanSchema, ObjectSchema};
> +use proxmox_schema::{api, BooleanSchema, ObjectSchema};
>  use proxmox_sortable_macro::sortable;
>  
>  use pbs_api_types::{
> @@ -227,6 +227,7 @@ const READER_API_SUBDIRS: SubdirMap = &[
>          "download",
>          &Router::new().download(&API_METHOD_DOWNLOAD_FILE),
>      ),
> +    ("heartbeat", &Router::new().download(&API_METHOD_HEARTBEAT)),

shouldn't this be `.get`

>      ("speedtest", &Router::new().download(&API_METHOD_SPEEDTEST)),
>  ];
>  
> @@ -433,3 +434,9 @@ fn speedtest(
>  
>      future::ok(response).boxed()
>  }
> +
> +#[api()]
> +/// HTTP level heartbeat to avoid proxies closing long running idle backup reader connections.
> +pub async fn heartbeat(_rpcenv: &mut dyn RpcEnvironment) -> Result<(), Error> {

and this a regular `fn`?

> +    Ok(())
> +}
> -- 
> 2.47.3
> 
> 
> 
> 
> 
>