Re: [PATCH v5 proxmox-backup 09/16] fix #3847: client: support fifo pipe inputs for images

[Christian Ebner <c.ebner@proxmox.com>]

On 1/30/26 5:45 PM, Robert Obkircher wrote:
> Accept fifo files as inputs for image backups. The unknown size is
> represented in the UploadOptions using a new IndexType enum where the
> total size for the Fixed variant is now optional.
> 
> Signed-off-by: Robert Obkircher <r.obkircher@proxmox.com>
> ---
>   pbs-client/src/backup_writer.rs   | 38 ++++++++++++++++++++++---------
>   proxmox-backup-client/src/main.rs | 30 +++++++++++++-----------
>   src/server/push.rs                | 11 +++++----
>   3 files changed, 50 insertions(+), 29 deletions(-)
> 
> diff --git a/pbs-client/src/backup_writer.rs b/pbs-client/src/backup_writer.rs
> index dbd177d8..f33f1063 100644
> --- a/pbs-client/src/backup_writer.rs
> +++ b/pbs-client/src/backup_writer.rs
> @@ -52,7 +52,17 @@ pub struct UploadOptions {
>       pub previous_manifest: Option<Arc<BackupManifest>>,
>       pub compress: bool,
>       pub encrypt: bool,
> -    pub fixed_size: Option<u64>,
> +    pub index_type: IndexType,
> +}
> +
> +/// Index type for upload options.
> +#[derive(Default, Clone)]
> +pub enum IndexType {
> +    /// Dynamic chunking.
> +    #[default]
> +    Dynamic,
> +    /// Fixed size chunking with optional image file size.
> +    Fixed(Option<u64>),
>   }
>   
>   struct ChunkUploadResponse {
> @@ -292,11 +302,14 @@ impl BackupWriter {
>           options: UploadOptions,
>       ) -> Result<BackupStats, Error> {
>           let mut param = json!({ "archive-name": archive_name });
> -        let prefix = if let Some(size) = options.fixed_size {
> -            param["size"] = size.into();
> -            "fixed"
> -        } else {
> -            "dynamic"
> +        let prefix = match options.index_type {
> +            IndexType::Fixed(image_file_size) => {
> +                if let Some(size) = image_file_size {
> +                    param["size"] = size.into();
> +                }
> +                "fixed"
> +            }
> +            IndexType::Dynamic => "dynamic",

this and the identical hunk below could be made a bit more readable by 
introducing a helper for the prefix and size on IndexType. Which would
also bundle the prefix generation into a single location. E.g.
```
diff --git a/pbs-client/src/backup_writer.rs 
b/pbs-client/src/backup_writer.rs
index f33f1063e..835e02ac2 100644
--- a/pbs-client/src/backup_writer.rs
+++ b/pbs-client/src/backup_writer.rs
@@ -65,6 +65,15 @@ pub enum IndexType {
      Fixed(Option<u64>),
  }

+impl IndexType {
+    fn to_prefix_and_size(&self) -> (&'static str, Option<u64>) {
+        match self {
+            IndexType::Fixed(size) => ("fixed", *size),
+            IndexType::Dynamic => ("dynamic", None),
+        }
+    }
+}
+
  struct ChunkUploadResponse {
      future: h2::client::ResponseFuture,
      size: usize,
@@ -302,15 +311,10 @@ impl BackupWriter {
          options: UploadOptions,
      ) -> Result<BackupStats, Error> {
          let mut param = json!({ "archive-name": archive_name });
-        let prefix = match options.index_type {
-            IndexType::Fixed(image_file_size) => {
-                if let Some(size) = image_file_size {
-                    param["size"] = size.into();
-                }
-                "fixed"
-            }
-            IndexType::Dynamic => "dynamic",
-        };
+        let (prefix, archive_size) = 
options.index_type.to_prefix_and_size();
+        if let Some(size) = archive_size {
+            param["size"] = size.into();
+        }

          if options.encrypt && self.crypt_config.is_none() {
              bail!("requested encryption without a crypt config");
```

>           };
>   
>           if options.encrypt && self.crypt_config.is_none() {
> @@ -387,11 +400,14 @@ impl BackupWriter {
>           let known_chunks = Arc::new(Mutex::new(HashSet::new()));
>   
>           let mut param = json!({ "archive-name": archive_name });
> -        let prefix = if let Some(size) = options.fixed_size {
> -            param["size"] = size.into();
> -            "fixed"
> -        } else {
> -            "dynamic"
> +        let prefix = match options.index_type {
> +            IndexType::Fixed(image_file_size) => {
> +                if let Some(size) = image_file_size {
> +                    param["size"] = size.into();
> +                }
> +                "fixed"
> +            }
> +            IndexType::Dynamic => "dynamic",
>           };
>   
>           if options.encrypt && self.crypt_config.is_none() {
> diff --git a/proxmox-backup-client/src/main.rs b/proxmox-backup-client/src/main.rs
> index 999e5020..7fc711fd 100644
> --- a/proxmox-backup-client/src/main.rs
> +++ b/proxmox-backup-client/src/main.rs
> @@ -46,7 +46,7 @@ use pbs_client::tools::{
>   use pbs_client::{
>       delete_ticket_info, parse_backup_specification, view_task_result, BackupDetectionMode,
>       BackupReader, BackupRepository, BackupSpecificationType, BackupStats, BackupWriter,
> -    BackupWriterOptions, ChunkStream, FixedChunkStream, HttpClient, InjectionData,
> +    BackupWriterOptions, ChunkStream, FixedChunkStream, HttpClient, IndexType, InjectionData,
>       PxarBackupStream, RemoteChunkReader, UploadOptions, BACKUP_SOURCE_SCHEMA,
>   };
>   use pbs_datastore::catalog::{BackupCatalogWriter, CatalogReader, CatalogWriter};
> @@ -205,7 +205,7 @@ async fn backup_directory<P: AsRef<Path>>(
>       pxar_create_options: pbs_client::pxar::PxarCreateOptions,
>       upload_options: UploadOptions,
>   ) -> Result<(BackupStats, Option<BackupStats>), Error> {
> -    if upload_options.fixed_size.is_some() {
> +    if let IndexType::Fixed(_) = upload_options.index_type {
>           bail!("cannot backup directory with fixed chunk size!");
>       }
>   
> @@ -295,7 +295,7 @@ async fn backup_image<P: AsRef<Path>>(
>   
>       let stream = FixedChunkStream::new(stream, chunk_size.unwrap_or(4 * 1024 * 1024));
>   
> -    if upload_options.fixed_size.is_none() {
> +    if let IndexType::Dynamic = upload_options.index_type {
>           bail!("cannot backup image with dynamic chunk size!");
>       }
>   
> @@ -859,15 +859,17 @@ async fn create_backup(
>                   upload_list.push((BackupSpecificationType::PXAR, filename, target, "didx", 0));
>               }
>               BackupSpecificationType::IMAGE => {
> -                if !(file_type.is_file() || file_type.is_block_device()) {
> -                    bail!("got unexpected file type (expected file or block device)");
> -                }
> -
> -                let size = image_size(&PathBuf::from(&filename))?;
> -
> -                if size == 0 {
> -                    bail!("got zero-sized file '{}'", filename);
> -                }
> +                let size = if file_type.is_file() || file_type.is_block_device() {
> +                    let size = image_size(&PathBuf::from(&filename))?;
> +                    if size == 0 {
> +                        bail!("got zero-sized file '{}'", filename);

nit: since touching this, the filename should be inlined into the format 
string.

> +                    }
> +                    size
> +                } else if file_type.is_fifo() {
> +                    0
> +                } else {
> +                    bail!("got unexpected file type (expected file, block device, or fifo");
> +                };
>   
>                   upload_list.push((
>                       BackupSpecificationType::IMAGE,
> @@ -1191,9 +1193,11 @@ async fn create_backup(
>               (BackupSpecificationType::IMAGE, false) => {
>                   log_file("image", &filename, target.as_ref());
>   
> +                // 0 means fifo pipe with unknown size
> +                let image_file_size = (size != 0).then_some(size);
>                   let upload_options = UploadOptions {
>                       previous_manifest: previous_manifest.clone(),
> -                    fixed_size: Some(size),
> +                    index_type: IndexType::Fixed(image_file_size),
>                       compress: true,
>                       encrypt: crypto.mode == CryptMode::Encrypt,
>                   };
> diff --git a/src/server/push.rs b/src/server/push.rs
> index d7884fce..b1b41297 100644
> --- a/src/server/push.rs
> +++ b/src/server/push.rs
> @@ -17,7 +17,8 @@ use pbs_api_types::{
>       PRIV_REMOTE_DATASTORE_MODIFY, PRIV_REMOTE_DATASTORE_PRUNE,
>   };
>   use pbs_client::{
> -    BackupRepository, BackupWriter, BackupWriterOptions, HttpClient, MergedChunkInfo, UploadOptions,
> +    BackupRepository, BackupWriter, BackupWriterOptions, HttpClient, IndexType, MergedChunkInfo,
> +    UploadOptions,
>   };
>   use pbs_config::CachedUserInfo;
>   use pbs_datastore::data_blob::ChunkInfo;
> @@ -917,7 +918,7 @@ pub(crate) async fn push_snapshot(
>                           index,
>                           chunk_reader,
>                           &backup_writer,
> -                        None,
> +                        IndexType::Dynamic,
>                           known_chunks.clone(),
>                       )
>                       .await?;
> @@ -944,7 +945,7 @@ pub(crate) async fn push_snapshot(
>                           index,
>                           chunk_reader,
>                           &backup_writer,
> -                        Some(size),
> +                        IndexType::Fixed(Some(size)),
>                           known_chunks.clone(),
>                       )
>                       .await?;
> @@ -1002,7 +1003,7 @@ async fn push_index(
>       index: impl IndexFile + Send + 'static,
>       chunk_reader: Arc<dyn AsyncReadChunk>,
>       backup_writer: &BackupWriter,
> -    size: Option<u64>,
> +    index_type: IndexType,
>       known_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
>   ) -> Result<SyncStats, Error> {
>       let (upload_channel_tx, upload_channel_rx) = mpsc::channel(20);
> @@ -1048,7 +1049,7 @@ async fn push_index(
>       let upload_options = UploadOptions {
>           compress: true,
>           encrypt: false,
> -        fixed_size: size,
> +        index_type,
>           ..UploadOptions::default()
>       };
>