Re: [pbs-devel] [PATCH vma-to-pbs 07/10] add support for streaming the VMA file via stdin

["Max Carrara" <m.carrara@proxmox.com>]

On Wed Mar 20, 2024 at 3:15 PM CET, Filip Schauer wrote:
> This allows the user to stream a compressed VMA file directly to a PBS
> without the need to extract it to an intermediate .vma file.
>
> Example usage:
>
> zstd -d --stdout vzdump.vma.zst | vma-to-pbs \
>     --repository <auth_id@host:port:datastore> \
>     --vmid 123 \
>     --password_file pbs_password
>
> Signed-off-by: Filip Schauer <f.schauer@proxmox.com>
> ---
> This requires
> https://lists.proxmox.com/pipermail/pve-devel/2024-March/062182.html
> to be applied first.
>
>  src/main.rs    | 241 ++------------------------------
>  src/vma.rs     | 326 ++++++++++++++++++++-----------------------
>  src/vma2pbs.rs | 368 +++++++++++++++++++++++++++++++++++++++++++++++++
>  3 files changed, 530 insertions(+), 405 deletions(-)
>  create mode 100644 src/vma2pbs.rs
>
> diff --git a/src/main.rs b/src/main.rs
> index f619b3e..e0e1076 100644
> --- a/src/main.rs
> +++ b/src/main.rs
> @@ -1,228 +1,10 @@
> -use std::env;
> -use std::ffi::{c_char, CStr, CString};
> -use std::ptr;
> -use std::time::{SystemTime, UNIX_EPOCH};
> -
> -use anyhow::{anyhow, Context, Result};
> +use anyhow::Result;
>  use clap::{command, Arg, ArgAction};
> -use proxmox_backup_qemu::*;
>  use proxmox_sys::linux::tty;
> -use scopeguard::defer;
>  
>  mod vma;
> -use vma::*;
> -
> -fn backup_vma_to_pbs(
> -    vma_file_path: String,
> -    pbs_repository: String,
> -    backup_id: String,
> -    pbs_password: String,
> -    keyfile: Option<String>,
> -    key_password: Option<String>,
> -    master_keyfile: Option<String>,
> -    fingerprint: String,
> -    compress: bool,
> -    encrypt: bool,
> -) -> Result<()> {
> -    println!("VMA input file: {}", vma_file_path);
> -    println!("PBS repository: {}", pbs_repository);
> -    println!("PBS fingerprint: {}", fingerprint);
> -    println!("compress: {}", compress);
> -    println!("encrypt: {}", encrypt);
> -
> -    let backup_time = SystemTime::now()
> -        .duration_since(UNIX_EPOCH)
> -        .unwrap()
> -        .as_secs();
> -    println!("backup time: {}", backup_time);
> -
> -    let mut pbs_err: *mut c_char = ptr::null_mut();
> -
> -    let pbs_repository_cstr = CString::new(pbs_repository).unwrap();
> -    let backup_id_cstr = CString::new(backup_id).unwrap();
> -    let pbs_password_cstr = CString::new(pbs_password).unwrap();
> -    let fingerprint_cstr = CString::new(fingerprint).unwrap();
> -    let keyfile_cstr = keyfile.map(|v| CString::new(v).unwrap());
> -    let keyfile_ptr = keyfile_cstr.map(|v| v.as_ptr()).unwrap_or(ptr::null());
> -    let key_password_cstr = key_password.map(|v| CString::new(v).unwrap());
> -    let key_password_ptr = key_password_cstr.map(|v| v.as_ptr()).unwrap_or(ptr::null());
> -    let master_keyfile_cstr = master_keyfile.map(|v| CString::new(v).unwrap());
> -    let master_keyfile_ptr = master_keyfile_cstr
> -        .map(|v| v.as_ptr())
> -        .unwrap_or(ptr::null());
> -
> -    let pbs = proxmox_backup_new_ns(
> -        pbs_repository_cstr.as_ptr(),
> -        ptr::null(),
> -        backup_id_cstr.as_ptr(),
> -        backup_time,
> -        PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE,
> -        pbs_password_cstr.as_ptr(),
> -        keyfile_ptr,
> -        key_password_ptr,
> -        master_keyfile_ptr,
> -        true,
> -        false,
> -        fingerprint_cstr.as_ptr(),
> -        &mut pbs_err,
> -    );
> -
> -    defer! {
> -        proxmox_backup_disconnect(pbs);
> -    }
> -
> -    if pbs == ptr::null_mut() {
> -        unsafe {
> -            let pbs_err_cstr = CStr::from_ptr(pbs_err);
> -            return Err(anyhow!("proxmox_backup_new_ns failed: {pbs_err_cstr:?}"));
> -        }
> -    }
> -
> -    let connect_result = proxmox_backup_connect(pbs, &mut pbs_err);
> -
> -    if connect_result < 0 {
> -        unsafe {
> -            let pbs_err_cstr = CStr::from_ptr(pbs_err);
> -            return Err(anyhow!("proxmox_backup_connect failed: {pbs_err_cstr:?}"));
> -        }
> -    }
> -
> -    let mut vma_reader = VmaReader::new(&vma_file_path)?;
> -
> -    // Handle configs
> -    let configs = vma_reader.get_configs();
> -    for (config_name, config_data) in configs {
> -        println!("CFG: size: {} name: {}", config_data.len(), config_name);
> -
> -        let config_name_cstr = CString::new(config_name).unwrap();
> -
> -        if proxmox_backup_add_config(
> -            pbs,
> -            config_name_cstr.as_ptr(),
> -            config_data.as_ptr(),
> -            config_data.len() as u64,
> -            &mut pbs_err,
> -        ) < 0
> -        {
> -            unsafe {
> -                let pbs_err_cstr = CStr::from_ptr(pbs_err);
> -                return Err(anyhow!(
> -                    "proxmox_backup_add_config failed: {pbs_err_cstr:?}"
> -                ));
> -            }
> -        }
> -    }
> -
> -    // Handle block devices
> -    for device_id in 0..255 {
> -        let device_name = match vma_reader.get_device_name(device_id) {
> -            Some(x) => x,
> -            None => {
> -                continue;
> -            }
> -        };
> -
> -        let device_size = match vma_reader.get_device_size(device_id) {
> -            Some(x) => x,
> -            None => {
> -                continue;
> -            }
> -        };
> -
> -        println!(
> -            "DEV: dev_id={} size: {} devname: {}",
> -            device_id, device_size, device_name
> -        );
> -
> -        let device_name_cstr = CString::new(device_name).unwrap();
> -        let pbs_device_id = proxmox_backup_register_image(
> -            pbs,
> -            device_name_cstr.as_ptr(),
> -            device_size,
> -            false,
> -            &mut pbs_err,
> -        );
> -
> -        if pbs_device_id < 0 {
> -            unsafe {
> -                let pbs_err_cstr = CStr::from_ptr(pbs_err);
> -                return Err(anyhow!(
> -                    "proxmox_backup_register_image failed: {pbs_err_cstr:?}"
> -                ));
> -            }
> -        }
> -
> -        let mut image_chunk_buffer =
> -            proxmox_io::boxed::zeroed(PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE as usize);
> -        let mut bytes_transferred = 0;
> -
> -        while bytes_transferred < device_size {
> -            let bytes_left = device_size - bytes_transferred;
> -            let chunk_size = bytes_left.min(PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE);
> -            println!(
> -                "Uploading dev_id: {} offset: {:#0X} - {:#0X}",
> -                device_id,
> -                bytes_transferred,
> -                bytes_transferred + chunk_size
> -            );
> -
> -            let is_zero_chunk = vma_reader
> -                .read_device_contents(
> -                    device_id,
> -                    &mut image_chunk_buffer[0..chunk_size as usize],
> -                    bytes_transferred,
> -                )
> -                .with_context(|| {
> -                    format!(
> -                        "read {} bytes at offset {} from disk {} from VMA file",
> -                        chunk_size, bytes_transferred, device_id
> -                    )
> -                })?;
> -
> -            let write_data_result = proxmox_backup_write_data(
> -                pbs,
> -                pbs_device_id as u8,
> -                if is_zero_chunk {
> -                    ptr::null()
> -                } else {
> -                    image_chunk_buffer.as_ptr()
> -                },
> -                bytes_transferred,
> -                chunk_size,
> -                &mut pbs_err,
> -            );
> -
> -            if write_data_result < 0 {
> -                unsafe {
> -                    let pbs_err_cstr = CStr::from_ptr(pbs_err);
> -                    return Err(anyhow!(
> -                        "proxmox_backup_write_data failed: {pbs_err_cstr:?}"
> -                    ));
> -                }
> -            }
> -
> -            bytes_transferred += chunk_size;
> -        }
> -
> -        if proxmox_backup_close_image(pbs, pbs_device_id as u8, &mut pbs_err) < 0 {
> -            unsafe {
> -                let pbs_err_cstr = CStr::from_ptr(pbs_err);
> -                return Err(anyhow!(
> -                    "proxmox_backup_close_image failed: {pbs_err_cstr:?}"
> -                ));
> -            }
> -        }
> -    }
> -
> -    if proxmox_backup_finish(pbs, &mut pbs_err) < 0 {
> -        unsafe {
> -            let pbs_err_cstr = CStr::from_ptr(pbs_err);
> -            return Err(anyhow!("proxmox_backup_finish failed: {pbs_err_cstr:?}"));
> -        }
> -    }
> -
> -    Ok(())
> -}
> +mod vma2pbs;
> +use vma2pbs::{backup_vma_to_pbs, BackupVmaToPbsArgs};
>  
>  fn main() -> Result<()> {
>      let matches = command!()
> @@ -300,7 +82,8 @@ fn main() -> Result<()> {
>      let compress = matches.get_flag("compress");
>      let encrypt = matches.get_flag("encrypt");
>  
> -    let vma_file_path = matches.get_one::<String>("vma_file").unwrap().to_string();
> +    let vma_file_path = matches.get_one::<String>("vma_file");
> +
>      let password_file = matches.get_one::<String>("password_file");
>  
>      let pbs_password = match password_file {
> @@ -323,18 +106,20 @@ fn main() -> Result<()> {
>          None => None,
>      };
>  
> -    backup_vma_to_pbs(
> -        vma_file_path,
> +    let args = BackupVmaToPbsArgs {
> +        vma_file_path: vma_file_path.cloned(),
>          pbs_repository,
> -        vmid,
> +        backup_id: vmid,
>          pbs_password,
> -        keyfile.cloned(),
> +        keyfile: keyfile.cloned(),
>          key_password,
> -        master_keyfile.cloned(),
> +        master_keyfile: master_keyfile.cloned(),
>          fingerprint,
>          compress,
>          encrypt,
> -    )?;
> +    };
> +
> +    backup_vma_to_pbs(args)?;

Very glad to see an arg struct being used here now! :)

>  
>      Ok(())
>  }
> diff --git a/src/vma.rs b/src/vma.rs
> index d30cb09..d369b36 100644
> --- a/src/vma.rs
> +++ b/src/vma.rs
> @@ -1,24 +1,55 @@
> -use std::collections::HashMap;
> -use std::fs::File;
> -use std::io::{Read, Seek, SeekFrom};
> +use std::collections::HashSet;
> +use std::io::Read;
>  use std::mem::size_of;
> -use std::{cmp, str};
> +use std::str;
>  
>  use anyhow::{anyhow, Result};
>  use bincode::Options;
>  use serde::{Deserialize, Serialize};
>  use serde_big_array::BigArray;
>  
> -const VMA_BLOCKS_PER_EXTENT: usize = 59;
> +/// Maximum number of clusters in an extent
> +/// See Image Data Streams in pve-qemu.git/vma_spec.txt
> +const VMA_CLUSTERS_PER_EXTENT: usize = 59;
> +
> +/// Number of 4k blocks per cluster
> +/// See pve-qemu.git/vma_spec.txt
> +const VMA_BLOCKS_PER_CLUSTER: usize = 16;
> +
> +/// Maximum number of config files
> +/// See VMA Header in pve-qemu.git/vma_spec.txt
>  const VMA_MAX_CONFIGS: usize = 256;
> +
> +/// Maximum number of block devices
> +/// See VMA Header in pve-qemu.git/vma_spec.txt
>  const VMA_MAX_DEVICES: usize = 256;
>  
> +/// VMA magic string
> +/// See VMA Header in pve-qemu.git/vma_spec.txt
> +const VMA_HEADER_MAGIC: [u8; 4] = [b'V', b'M', b'A', 0];
> +
> +/// VMA extent magic string
> +/// See VMA Extent Header in pve-qemu.git/vma_spec.txt
> +const VMA_EXTENT_HEADER_MAGIC: [u8; 4] = [b'V', b'M', b'A', b'E'];
> +
> +/// Size of a block
> +/// See pve-qemu.git/vma_spec.txt
> +const BLOCK_SIZE: usize = 4096;
> +
> +/// Size of the VMA header without the blob buffer appended at the end
> +/// See VMA Header in pve-qemu.git/vma_spec.txt
> +const VMA_HEADER_SIZE_NO_BLOB_BUFFER: usize = 12288;
> +
> +type VmaDeviceId = u8;
> +type VmaDeviceOffset = u64;
> +type VmaDeviceSize = u64;
> +
>  #[repr(C)]
>  #[derive(Serialize, Deserialize)]
>  struct VmaDeviceInfoHeader {
>      pub device_name_offset: u32,
>      reserved: [u8; 4],
> -    pub device_size: u64,
> +    pub device_size: VmaDeviceSize,
>      reserved1: [u8; 16],
>  }
>  
> @@ -42,6 +73,8 @@ struct VmaHeader {
>      reserved1: [u8; 4],
>      #[serde(with = "BigArray")]
>      pub dev_info: [VmaDeviceInfoHeader; VMA_MAX_DEVICES],
> +    #[serde(skip_deserializing, skip_serializing)]
> +    blob_buffer: Vec<u8>,
>  }
>  
>  #[repr(C)]
> @@ -62,57 +95,46 @@ struct VmaExtentHeader {
>      pub uuid: [u8; 16],
>      pub md5sum: [u8; 16],
>      #[serde(with = "BigArray")]
> -    pub blockinfo: [VmaBlockInfo; VMA_BLOCKS_PER_EXTENT],
> +    pub blockinfo: [VmaBlockInfo; VMA_CLUSTERS_PER_EXTENT],
>  }
>  
> -#[derive(Clone)]
> -struct VmaBlockIndexEntry {
> -    pub cluster_file_offset: u64,
> -    pub mask: u16,
> +#[derive(Clone, Eq, Hash, PartialEq)]
> +pub struct VmaConfig {
> +    pub name: String,
> +    pub content: String,
>  }
>  
>  pub struct VmaReader {
> -    vma_file: File,
> +    vma_file: Box<dyn Read>,
>      vma_header: VmaHeader,
> -    configs: HashMap<String, String>,
> -    block_index: Vec<Vec<VmaBlockIndexEntry>>,
> -    blocks_are_indexed: bool,
> +    configs: HashSet<VmaConfig>,
>  }
>  
>  impl VmaReader {
> -    pub fn new(vma_file_path: &str) -> Result<Self> {
> -        let mut vma_file = match File::open(vma_file_path) {
> -            Err(why) => return Err(anyhow!("couldn't open {}: {}", vma_file_path, why)),
> -            Ok(file) => file,
> -        };
> -
> -        let vma_header = Self::read_header(&mut vma_file).unwrap();
> -        let configs = Self::read_blob_buffer(&mut vma_file, &vma_header).unwrap();
> -        let block_index: Vec<Vec<VmaBlockIndexEntry>> = (0..256).map(|_| Vec::new()).collect();
> +    pub fn new(mut vma_file: impl Read + 'static) -> Result<Self> {

Still not too sure about this here - I guess it's *fine* to stay with
`impl Read` here and `Box<dyn Read>` in the struct above, but IMO if
you're already generic here, the `VmaReader` might as well become a
`VmaReader<T>`.

The relevant `impl` blocks where `Read` is necessary could then just be
constrained.

Otherwise, the stuff given to `new()` here is being monomorphized and
_then_ made into a trait object and boxed. Sure, it's a minor thing, but
still some unnecessary work.

> +        let vma_header = Self::read_header(&mut vma_file)?;
> +        let configs = Self::read_configs(&vma_header)?;
>  
>          let instance = Self {
> -            vma_file,
> +            vma_file: Box::new(vma_file),
>              vma_header,
>              configs,
> -            block_index,
> -            blocks_are_indexed: false,
>          };
>  
>          Ok(instance)
>      }
>  
> -    fn read_header(vma_file: &mut File) -> Result<VmaHeader> {
> -        let mut buffer = Vec::with_capacity(size_of::<VmaHeader>());
> -        buffer.resize(size_of::<VmaHeader>(), 0);
> +    fn read_header(vma_file: &mut impl Read) -> Result<VmaHeader> {
> +        let mut buffer = vec![0; VMA_HEADER_SIZE_NO_BLOB_BUFFER];
>          vma_file.read_exact(&mut buffer)?;
>  
>          let bincode_options = bincode::DefaultOptions::new()
>              .with_fixint_encoding()
>              .with_big_endian();
>  
> -        let vma_header: VmaHeader = bincode_options.deserialize(&buffer)?;
> +        let mut vma_header: VmaHeader = bincode_options.deserialize(&buffer)?;
>  
> -        if vma_header.magic != [b'V', b'M', b'A', 0] {
> +        if vma_header.magic != VMA_HEADER_MAGIC {
>              return Err(anyhow!("Invalid magic number"));
>          }
>  
> @@ -121,7 +143,7 @@ impl VmaReader {
>          }
>  
>          buffer.resize(vma_header.header_size as usize, 0);
> -        vma_file.read_exact(&mut buffer[size_of::<VmaHeader>()..])?;
> +        vma_file.read_exact(&mut buffer[VMA_HEADER_SIZE_NO_BLOB_BUFFER..])?;
>  
>          // Fill the MD5 sum field with zeros to compute the MD5 sum
>          buffer[32..48].fill(0);
> @@ -131,89 +153,77 @@ impl VmaReader {
>              return Err(anyhow!("Wrong VMA header checksum"));
>          }
>  
> -        return Ok(vma_header);
> +        let blob_buffer = &buffer[VMA_HEADER_SIZE_NO_BLOB_BUFFER..vma_header.header_size as usize];
> +        vma_header.blob_buffer = blob_buffer.to_vec();
> +
> +        Ok(vma_header)
>      }
>  
> -    fn read_string_from_file(vma_file: &mut File, file_offset: u64) -> Result<String> {
> -        let mut size_bytes = [0u8; 2];
> -        vma_file.seek(SeekFrom::Start(file_offset))?;
> -        vma_file.read_exact(&mut size_bytes)?;
> +    fn read_string(buffer: &[u8]) -> Result<String> {
> +        let size_bytes: [u8; 2] = buffer[0..2].try_into()?;
>          let size = u16::from_le_bytes(size_bytes) as usize;
> -        let mut string_bytes = Vec::with_capacity(size - 1);
> -        string_bytes.resize(size - 1, 0);
> -        vma_file.read_exact(&mut string_bytes)?;
> -        let string = str::from_utf8(&string_bytes)?;
> +        let string_bytes: &[u8] = &buffer[2..1 + size];
> +        let string = str::from_utf8(string_bytes)?;
>  
> -        return Ok(string.to_string());
> +        Ok(String::from(string))
>      }
>  
> -    fn read_blob_buffer(
> -        vma_file: &mut File,
> -        vma_header: &VmaHeader,
> -    ) -> Result<HashMap<String, String>> {
> -        let mut configs = HashMap::new();
> +    fn read_configs(vma_header: &VmaHeader) -> Result<HashSet<VmaConfig>> {
> +        let mut configs = HashSet::new();
>  
>          for i in 0..VMA_MAX_CONFIGS {
> -            let config_name_offset = vma_header.config_names[i];
> -            let config_data_offset = vma_header.config_data[i];
> +            let config_name_offset = vma_header.config_names[i] as usize;
> +            let config_data_offset = vma_header.config_data[i] as usize;
>  
>              if config_name_offset == 0 || config_data_offset == 0 {
>                  continue;
>              }
>  
> -            let config_name_file_offset =
> -                (vma_header.blob_buffer_offset + config_name_offset) as u64;
> -            let config_data_file_offset =
> -                (vma_header.blob_buffer_offset + config_data_offset) as u64;
> -            let config_name = Self::read_string_from_file(vma_file, config_name_file_offset)?;
> -            let config_data = Self::read_string_from_file(vma_file, config_data_file_offset)?;
> -
> -            configs.insert(String::from(config_name), String::from(config_data));
> +            let config_name = Self::read_string(&vma_header.blob_buffer[config_name_offset..])?;
> +            let config_data = Self::read_string(&vma_header.blob_buffer[config_data_offset..])?;
> +            let vma_config = VmaConfig {
> +                name: config_name,
> +                content: config_data,
> +            };
> +            configs.insert(vma_config);
>          }
>  
> -        return Ok(configs);
> +        Ok(configs)
>      }
>  
> -    pub fn get_configs(&self) -> HashMap<String, String> {
> -        return self.configs.clone();
> +    pub fn get_configs(&self) -> HashSet<VmaConfig> {
> +        self.configs.clone()
>      }
>  
> -    pub fn get_device_name(&mut self, device_id: usize) -> Option<String> {
> -        if device_id >= VMA_MAX_DEVICES {
> -            return None;
> -        }
> +    pub fn contains_device(&self, device_id: VmaDeviceId) -> bool {
> +        self.vma_header.dev_info[device_id as usize].device_name_offset != 0
> +    }
>  
> -        let device_name_offset = self.vma_header.dev_info[device_id].device_name_offset;
> +    pub fn get_device_name(&self, device_id: VmaDeviceId) -> Result<String> {
> +        let device_name_offset =
> +            self.vma_header.dev_info[device_id as usize].device_name_offset as usize;
>  
>          if device_name_offset == 0 {
> -            return None;
> +            anyhow::bail!("device_name_offset cannot be 0");
>          }
>  
> -        let device_name_file_offset =
> -            (self.vma_header.blob_buffer_offset + device_name_offset) as u64;
> -        let device_name =
> -            Self::read_string_from_file(&mut self.vma_file, device_name_file_offset).unwrap();
> +        let device_name = Self::read_string(&self.vma_header.blob_buffer[device_name_offset..])?;
>  
> -        return Some(device_name.to_string());
> +        Ok(device_name)
>      }
>  
> -    pub fn get_device_size(&self, device_id: usize) -> Option<u64> {
> -        if device_id >= VMA_MAX_DEVICES {
> -            return None;
> -        }
> -
> -        let dev_info = &self.vma_header.dev_info[device_id];
> +    pub fn get_device_size(&self, device_id: VmaDeviceId) -> Result<VmaDeviceSize> {
> +        let dev_info = &self.vma_header.dev_info[device_id as usize];
>  
>          if dev_info.device_name_offset == 0 {
> -            return None;
> +            anyhow::bail!("device_name_offset cannot be 0");

It's okay if you `use anyhow::bail` above, no need to fully qualify it
;)

>          }
>  
> -        return Some(dev_info.device_size);
> +        Ok(dev_info.device_size)
>      }
>  
> -    fn read_extent_header(vma_file: &mut File) -> Result<VmaExtentHeader> {
> -        let mut buffer = Vec::with_capacity(size_of::<VmaExtentHeader>());
> -        buffer.resize(size_of::<VmaExtentHeader>(), 0);
> +    fn read_extent_header(mut vma_file: impl Read) -> Result<VmaExtentHeader> {
> +        let mut buffer = vec![0; size_of::<VmaExtentHeader>()];
>          vma_file.read_exact(&mut buffer)?;
>  
>          let bincode_options = bincode::DefaultOptions::new()
> @@ -222,7 +232,7 @@ impl VmaReader {
>  
>          let vma_extent_header: VmaExtentHeader = bincode_options.deserialize(&buffer)?;
>  
> -        if vma_extent_header.magic != [b'V', b'M', b'A', b'E'] {
> +        if vma_extent_header.magic != VMA_EXTENT_HEADER_MAGIC {
>              return Err(anyhow!("Invalid magic number"));
>          }
>  
> @@ -234,113 +244,75 @@ impl VmaReader {
>              return Err(anyhow!("Wrong VMA extent header checksum"));
>          }
>  
> -        return Ok(vma_extent_header);
> +        Ok(vma_extent_header)
>      }
>  
> -    fn index_device_clusters(&mut self) -> Result<()> {
> -        for device_id in 0..255 {
> -            let device_size = match self.get_device_size(device_id) {
> -                Some(x) => x,
> -                None => {
> -                    continue;
> -                }
> -            };
> -
> -            let device_cluster_count = (device_size + 4096 * 16 - 1) / (4096 * 16);
> +    fn restore_extent<F>(&mut self, callback: F) -> Result<()>
> +    where
> +        F: Fn(VmaDeviceId, VmaDeviceOffset, Option<Vec<u8>>) -> Result<()>,
> +    {
> +        let vma_extent_header = Self::read_extent_header(&mut self.vma_file)?;
>  
> -            let block_index_entry_placeholder = VmaBlockIndexEntry {
> -                cluster_file_offset: 0,
> -                mask: 0,
> -            };
> -
> -            self.block_index[device_id]
> -                .resize(device_cluster_count as usize, block_index_entry_placeholder);
> -        }
> +        for cluster_index in 0..VMA_CLUSTERS_PER_EXTENT {
> +            let blockinfo = &vma_extent_header.blockinfo[cluster_index];
>  
> -        let mut file_offset = self.vma_header.header_size as u64;
> -        let vma_file_size = self.vma_file.metadata()?.len();
> -
> -        while file_offset < vma_file_size {
> -            self.vma_file.seek(SeekFrom::Start(file_offset))?;
> -            let vma_extent_header = Self::read_extent_header(&mut self.vma_file)?;
> -            file_offset += size_of::<VmaExtentHeader>() as u64;
> -
> -            for i in 0..VMA_BLOCKS_PER_EXTENT {
> -                let blockinfo = &vma_extent_header.blockinfo[i];
> +            if blockinfo.dev_id == 0 {
> +                continue;
> +            }
>  
> -                if blockinfo.dev_id == 0 {
> -                    continue;
> +            let image_offset =
> +                (BLOCK_SIZE * VMA_BLOCKS_PER_CLUSTER * blockinfo.cluster_num as usize) as u64;
> +            let cluster_is_zero = blockinfo.mask == 0;
> +
> +            let image_chunk_buffer = if cluster_is_zero {
> +                None
> +            } else {
> +                let mut image_chunk_buffer = vec![0; BLOCK_SIZE * VMA_BLOCKS_PER_CLUSTER];
> +
> +                for block_index in 0..VMA_BLOCKS_PER_CLUSTER {
> +                    let block_is_zero = ((blockinfo.mask >> block_index) & 1) == 0;
> +                    let block_start = BLOCK_SIZE * block_index;
> +                    let block_end = block_start + BLOCK_SIZE;
> +
> +                    if block_is_zero {
> +                        image_chunk_buffer[block_start..block_end].fill(0);
> +                    } else {
> +                        self.vma_file
> +                            .read_exact(&mut image_chunk_buffer[block_start..block_end])?;
> +                    }
>                  }
>  
> -                let block_index_entry = VmaBlockIndexEntry {
> -                    cluster_file_offset: file_offset,
> -                    mask: blockinfo.mask,
> -                };
> +                Some(image_chunk_buffer)
> +            };
>  
> -                self.block_index[blockinfo.dev_id as usize][blockinfo.cluster_num as usize] =
> -                    block_index_entry;
> -                file_offset += blockinfo.mask.count_ones() as u64 * 4096;
> -            }
> +            callback(blockinfo.dev_id, image_offset, image_chunk_buffer)?;
>          }
>  
> -        self.blocks_are_indexed = true;
> -
> -        return Ok(());
> +        Ok(())
>      }
>  
> -    pub fn read_device_contents(
> -        &mut self,
> -        device_id: usize,
> -        buffer: &mut [u8],
> -        offset: u64,
> -    ) -> Result<bool> {
> -        if device_id >= VMA_MAX_DEVICES {
> -            return Err(anyhow!("invalid device id {}", device_id));
> -        }
> -
> -        if offset % (4096 * 16) != 0 {
> -            return Err(anyhow!("offset is not aligned to 65536"));
> -        }
> -
> -        // Make sure that the device clusters are already indexed
> -        if !self.blocks_are_indexed {
> -            self.index_device_clusters()?;
> -        }
> -
> -        let this_device_block_index = &self.block_index[device_id];
> -        let length = cmp::min(
> -            buffer.len(),
> -            this_device_block_index.len() * 4096 * 16 - offset as usize,
> -        );
> -        let mut buffer_offset = 0;
> -        let mut buffer_is_zero = true;
> -
> -        while buffer_offset < length {
> -            let block_index_entry =
> -                &this_device_block_index[(offset as usize + buffer_offset) / (4096 * 16)];
> -            self.vma_file
> -                .seek(SeekFrom::Start(block_index_entry.cluster_file_offset))?;
> -
> -            for i in 0..16 {
> -                if buffer_offset >= length {
> -                    break;
> -                }
> -
> -                let block_buffer_end = buffer_offset + cmp::min(length - buffer_offset, 4096);
> -                let block_mask = ((block_index_entry.mask >> i) & 1) == 1;
> -
> -                if block_mask {
> -                    self.vma_file
> -                        .read_exact(&mut buffer[buffer_offset..block_buffer_end])?;
> -                    buffer_is_zero = false;
> -                } else {
> -                    buffer[buffer_offset..block_buffer_end].fill(0);
> -                }
> -
> -                buffer_offset += 4096;
> +    pub fn restore<F>(&mut self, callback: F) -> Result<()>
> +    where
> +        F: Fn(VmaDeviceId, VmaDeviceOffset, Option<Vec<u8>>) -> Result<()>,
> +    {
> +        loop {
> +            match self.restore_extent(&callback) {
> +                Ok(()) => {}
> +                Err(e) => match e.downcast_ref::<std::io::Error>() {
> +                    Some(ioerr) => {
> +                        if ioerr.kind() == std::io::ErrorKind::UnexpectedEof {
> +                            break; // Break out of the loop since the end of the file was reached.
> +                        } else {
> +                            return Err(anyhow!("Failed to read VMA file: {}", ioerr));
> +                        }
> +                    }
> +                    _ => {
> +                        return Err(anyhow::format_err!(e));
> +                    }
> +                },
>              }
>          }
>  
> -        return Ok(buffer_is_zero);
> +        Ok(())
>      }
>  }
> diff --git a/src/vma2pbs.rs b/src/vma2pbs.rs
> new file mode 100644
> index 0000000..cef7b60
> --- /dev/null
> +++ b/src/vma2pbs.rs
> @@ -0,0 +1,368 @@
> +use std::cell::RefCell;
> +use std::collections::HashMap;
> +use std::ffi::{c_char, CStr, CString};
> +use std::fs::File;
> +use std::io::{stdin, BufRead, BufReader};
> +use std::ptr;
> +use std::time::{SystemTime, UNIX_EPOCH};
> +
> +use anyhow::{anyhow, Result};
> +use proxmox_backup_qemu::{
> +    capi_types::ProxmoxBackupHandle, proxmox_backup_add_config, proxmox_backup_close_image,
> +    proxmox_backup_connect, proxmox_backup_disconnect, proxmox_backup_finish,
> +    proxmox_backup_new_ns, proxmox_backup_register_image, proxmox_backup_write_data,
> +    PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE,
> +};
> +use scopeguard::defer;
> +
> +use crate::vma::VmaReader;
> +
> +const VMA_CLUSTER_SIZE: usize = 65536;
> +
> +pub struct BackupVmaToPbsArgs {
> +    pub vma_file_path: Option<String>,
> +    pub pbs_repository: String,
> +    pub backup_id: String,
> +    pub pbs_password: String,
> +    pub keyfile: Option<String>,
> +    pub key_password: Option<String>,
> +    pub master_keyfile: Option<String>,
> +    pub fingerprint: String,
> +    pub compress: bool,
> +    pub encrypt: bool,
> +}
> +
> +#[derive(Copy, Clone)]
> +struct BlockDeviceInfo {
> +    pub pbs_device_id: u8,
> +    pub device_size: u64,
> +}
> +
> +fn create_pbs_backup_task(args: BackupVmaToPbsArgs) -> Result<*mut ProxmoxBackupHandle> {

Ugh, I know all of the `proxmox-backup-qemu`-related things take a
`*mut` - so I guess passing that around here is fine, as it makes things
easier to read... (In an ideal world that API would take a
`NonNull<ProxmoxBackupHandle>` <.<)

> +    println!("PBS repository: {}", args.pbs_repository);
> +    println!("PBS fingerprint: {}", args.fingerprint);
> +    println!("compress: {}", args.compress);
> +    println!("encrypt: {}", args.encrypt);
> +
> +    let backup_time = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
> +    println!("backup time: {}", backup_time);

Should maybe format the time a little different; I e.g. get:

[...]
backup time: 1711359868
[...]


> +
> +    let mut pbs_err: *mut c_char = ptr::null_mut();
> +
> +    let pbs_repository_cstr = CString::new(args.pbs_repository)?;
> +    let backup_id_cstr = CString::new(args.backup_id)?;
> +    let pbs_password_cstr = CString::new(args.pbs_password)?;
> +    let fingerprint_cstr = CString::new(args.fingerprint)?;
> +    let keyfile_cstr = args.keyfile.map(|v| CString::new(v).unwrap());
> +    let keyfile_ptr = keyfile_cstr.map(|v| v.as_ptr()).unwrap_or(ptr::null());
> +    let key_password_cstr = args.key_password.map(|v| CString::new(v).unwrap());
> +    let key_password_ptr = key_password_cstr.map(|v| v.as_ptr()).unwrap_or(ptr::null());
> +    let master_keyfile_cstr = args.master_keyfile.map(|v| CString::new(v).unwrap());
> +    let master_keyfile_ptr = master_keyfile_cstr
> +        .map(|v| v.as_ptr())
> +        .unwrap_or(ptr::null());
> +
> +    let pbs = proxmox_backup_new_ns(
> +        pbs_repository_cstr.as_ptr(),
> +        ptr::null(),
> +        backup_id_cstr.as_ptr(),
> +        backup_time,
> +        PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE,
> +        pbs_password_cstr.as_ptr(),
> +        keyfile_ptr,
> +        key_password_ptr,
> +        master_keyfile_ptr,
> +        true,
> +        false,
> +        fingerprint_cstr.as_ptr(),
> +        &mut pbs_err,
> +    );
> +
> +    if pbs.is_null() {
> +        let pbs_err_cstr = unsafe { CStr::from_ptr(pbs_err) };
> +        anyhow::bail!("proxmox_backup_new_ns failed: {pbs_err_cstr:?}");
> +    }

As mentioned in my previous review, you really should put this kind of
error handling / propagation into some kind of helper function. You want
to make sure that all necessary invariants for `CStr::from_ptr`  [0] are
upheld.

I know, this is our API and it's very unlikely that we'll do anything
funky with those pointers, but we should nevertheless never rely on C
FFI stuff to always remain sound IMO, as mistakes / accidents can
happen. All it takes is one faulty commit in `proxmox-backup-qemu`.

> +
> +    Ok(pbs)
> +}
> +
> +fn upload_configs(vma_reader: &VmaReader, pbs: *mut ProxmoxBackupHandle) -> Result<()> {
> +    let mut pbs_err: *mut c_char = ptr::null_mut();
> +    let configs = vma_reader.get_configs();
> +    for config in configs {
> +        let config_name = config.name;
> +        let config_data = config.content;
> +
> +        println!("CFG: size: {} name: {}", config_data.len(), config_name);
> +
> +        let config_name_cstr = CString::new(config_name)?;
> +
> +        if proxmox_backup_add_config(
> +            pbs,
> +            config_name_cstr.as_ptr(),
> +            config_data.as_ptr(),
> +            config_data.len() as u64,
> +            &mut pbs_err,
> +        ) < 0
> +        {
> +            let pbs_err_cstr = unsafe { CStr::from_ptr(pbs_err) };
> +            anyhow::bail!("proxmox_backup_add_config failed: {pbs_err_cstr:?}");
> +        }

Same as above (error handling).

> +    }
> +
> +    Ok(())
> +}
> +
> +fn register_block_devices(
> +    vma_reader: &VmaReader,
> +    pbs: *mut ProxmoxBackupHandle,
> +) -> Result<[Option<BlockDeviceInfo>; 256]> {
> +    let mut block_device_infos: [Option<BlockDeviceInfo>; 256] = [None; 256];
> +    let mut pbs_err: *mut c_char = ptr::null_mut();
> +
> +    for device_id in 0..255 {
> +        if !vma_reader.contains_device(device_id) {
> +            continue;
> +        }
> +
> +        let device_name = vma_reader.get_device_name(device_id)?;
> +        let device_size = vma_reader.get_device_size(device_id)?;
> +
> +        println!(
> +            "DEV: dev_id={} size: {} devname: {}",
> +            device_id, device_size, device_name
> +        );
> +
> +        let device_name_cstr = CString::new(device_name)?;
> +        let pbs_device_id = proxmox_backup_register_image(
> +            pbs,
> +            device_name_cstr.as_ptr(),
> +            device_size,
> +            false,
> +            &mut pbs_err,
> +        );
> +
> +        if pbs_device_id < 0 {
> +            let pbs_err_cstr = unsafe { CStr::from_ptr(pbs_err) };
> +            anyhow::bail!("proxmox_backup_register_image failed: {pbs_err_cstr:?}");
> +        }

Same as above (error handling).

> +
> +        let block_device_info = BlockDeviceInfo {
> +            pbs_device_id: pbs_device_id as u8,
> +            device_size,
> +        };
> +
> +        block_device_infos[device_id as usize] = Some(block_device_info);
> +    }
> +
> +    Ok(block_device_infos)
> +}
> +
> +fn upload_block_devices(mut vma_reader: VmaReader, pbs: *mut ProxmoxBackupHandle) -> Result<()> {
> +    let block_device_infos = register_block_devices(&vma_reader, pbs)?;
> +
> +    struct ImageChunk {
> +        sub_chunks: HashMap<u8, Option<Vec<u8>>>,
> +        mask: u64,
> +        non_zero_mask: u64,
> +    }
> +
> +    let images_chunks: RefCell<HashMap<u8, HashMap<u64, ImageChunk>>> =
> +        RefCell::new(HashMap::new());
> +

All of this, starting from here ...

> +    vma_reader.restore(|dev_id, offset, buffer| {
> +        let block_device_info = match block_device_infos[dev_id as usize] {
> +            Some(block_device_info) => block_device_info,
> +            None => anyhow::bail!("Reference to unknown device id {} in VMA file", dev_id),
> +        };
> +
> +        let pbs_device_id = block_device_info.pbs_device_id;
> +        let device_size = block_device_info.device_size;
> +
> +        let mut images_chunks = images_chunks.borrow_mut();
> +        let image_chunks = images_chunks.get_mut(&dev_id);
> +        let pbs_chunk_offset =
> +            PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE * (offset / PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE);
> +        let sub_chunk_index =
> +            ((offset % PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE) / VMA_CLUSTER_SIZE as u64) as u8;
> +
> +        let pbs_chunk_size = PROXMOX_BACKUP_DEFAULT_CHUNK_SIZE.min(device_size - pbs_chunk_offset);
> +
> +        let pbs_upload_chunk = |pbs_chunk_buffer: Option<&[u8]>| {
> +            println!(
> +                "Uploading dev_id: {} offset: {:#0X} - {:#0X}",
> +                dev_id,
> +                pbs_chunk_offset,
> +                pbs_chunk_offset + pbs_chunk_size,
> +            );
> +
> +            let mut pbs_err: *mut c_char = ptr::null_mut();
> +
> +            let write_data_result = proxmox_backup_write_data(
> +                pbs,
> +                pbs_device_id,
> +                match pbs_chunk_buffer {
> +                    Some(pbs_chunk_buffer) => pbs_chunk_buffer.as_ptr(),
> +                    None => ptr::null(),
> +                },
> +                pbs_chunk_offset,
> +                pbs_chunk_size,
> +                &mut pbs_err,
> +            );
> +
> +            if write_data_result < 0 {
> +                let pbs_err_cstr = unsafe { CStr::from_ptr(pbs_err) };
> +                anyhow::bail!("proxmox_backup_write_data failed: {pbs_err_cstr:?}");
> +            }
> +
> +            Ok(())
> +        };
> +
> +        let insert_image_chunk = |image_chunks: &mut HashMap<u64, ImageChunk>,
> +                                  buffer: Option<Vec<u8>>| {
> +            let mut sub_chunks: HashMap<u8, Option<Vec<u8>>> = HashMap::new();
> +            let mask = 1 << sub_chunk_index;
> +            let non_zero_mask = buffer.is_some() as u64;
> +            sub_chunks.insert(sub_chunk_index, buffer);
> +
> +            let image_chunk = ImageChunk {
> +                sub_chunks,
> +                mask,
> +                non_zero_mask,
> +            };
> +
> +            image_chunks.insert(pbs_chunk_offset, image_chunk);
> +        };
> +
> +        match image_chunks {
> +            Some(image_chunks) => {
> +                let image_chunk = image_chunks.get_mut(&pbs_chunk_offset);
> +
> +                match image_chunk {
> +                    Some(image_chunk) => {
> +                        image_chunk.mask |= 1 << sub_chunk_index;
> +                        image_chunk.non_zero_mask |= (buffer.is_some() as u64) << sub_chunk_index;
> +                        image_chunk.sub_chunks.insert(sub_chunk_index, buffer);
> +
> +                        let sub_chunk_count =
> +                            ((pbs_chunk_size + 65535) / VMA_CLUSTER_SIZE as u64) as u8;
> +                        let pbs_chunk_mask = 1_u64
> +                            .checked_shl(sub_chunk_count.into())
> +                            .unwrap_or(0)
> +                            .wrapping_sub(1);
> +
> +                        if image_chunk.mask == pbs_chunk_mask {
> +                            if image_chunk.non_zero_mask == 0 {
> +                                pbs_upload_chunk(None)?;
> +                            } else {
> +                                let mut pbs_chunk_buffer =
> +                                    proxmox_io::boxed::zeroed(pbs_chunk_size as usize);
> +
> +                                for i in 0..sub_chunk_count {
> +                                    let sub_chunk = &image_chunk.sub_chunks[&i];
> +                                    let start = i as usize * VMA_CLUSTER_SIZE;
> +                                    let end =
> +                                        (start + VMA_CLUSTER_SIZE).min(pbs_chunk_size as usize);
> +
> +                                    match sub_chunk {
> +                                        Some(sub_chunk) => {
> +                                            pbs_chunk_buffer[start..end]
> +                                                .copy_from_slice(&sub_chunk[0..end - start]);
> +                                        }
> +                                        None => pbs_chunk_buffer[start..end].fill(0),
> +                                    }
> +                                }
> +
> +                                pbs_upload_chunk(Some(&*pbs_chunk_buffer))?;
> +                            }
> +
> +                            image_chunks.remove(&pbs_chunk_offset);
> +                        }
> +                    }
> +                    None => {
> +                        if pbs_chunk_size <= VMA_CLUSTER_SIZE as u64 {
> +                            pbs_upload_chunk(buffer.as_deref())?;
> +                        } else {
> +                            insert_image_chunk(image_chunks, buffer);
> +                        }
> +                    }
> +                }
> +            }
> +            None => {
> +                if pbs_chunk_size <= VMA_CLUSTER_SIZE as u64 {
> +                    pbs_upload_chunk(buffer.as_deref())?;
> +                } else {
> +                    let mut image_chunks: HashMap<u64, ImageChunk> = HashMap::new();
> +                    insert_image_chunk(&mut image_chunks, buffer);
> +                    images_chunks.insert(dev_id, image_chunks);
> +                }
> +            }
> +        }
> +
> +        Ok(())
> +    })?;

... and ending here should still be taken apart and put into a bunch of
smaller functions. If it's hard to keep track of all the things going on
here while you're disentangling this, encapsulate the state as struct.

> +
> +    let mut pbs_err: *mut c_char = ptr::null_mut();
> +
> +    for block_device_info in block_device_infos.iter().take(255) {
> +        let block_device_info = match block_device_info {
> +            Some(block_device_info) => block_device_info,
> +            None => continue,
> +        };
> +
> +        let pbs_device_id = block_device_info.pbs_device_id;
> +
> +        if proxmox_backup_close_image(pbs, pbs_device_id, &mut pbs_err) < 0 {
> +            let pbs_err_cstr = unsafe { CStr::from_ptr(pbs_err) };
> +            anyhow::bail!("proxmox_backup_close_image failed: {pbs_err_cstr:?}");
> +        }

Same as above (error handling).

> +    }
> +
> +    Ok(())
> +}
> +
> +pub fn backup_vma_to_pbs(args: BackupVmaToPbsArgs) -> Result<()> {
> +    let vma_file: Box<dyn BufRead> = match &args.vma_file_path {
> +        Some(vma_file_path) => match File::open(vma_file_path) {
> +            Err(why) => return Err(anyhow!("Couldn't open file: {}", why)),
> +            Ok(file) => Box::new(BufReader::new(file)),
> +        },
> +        None => Box::new(BufReader::new(stdin())),

IMO we should consider using a separate argument that makes the CLI read
from stin, or e.g. if the file is named "-" (without quotes). Otherwise,
weird things could happen if one forgets to add an argument:

zstd -d --stdout vzdump-qemu-300-2024_03_25-10_21_10.vma.zst | vma-to-pbs --repository root@pam@my.pbs.tld:8007:default --vmid 300
Error: stream did not contain valid UTF-8


> +    };
> +    let vma_reader = VmaReader::new(vma_file)?;
> +
> +    let pbs = create_pbs_backup_task(args)?;
> +
> +    defer! {
> +        proxmox_backup_disconnect(pbs);
> +    }
> +
> +    let mut pbs_err: *mut c_char = ptr::null_mut();
> +    let connect_result = proxmox_backup_connect(pbs, &mut pbs_err);
> +
> +    if connect_result < 0 {
> +        let pbs_err_cstr = unsafe { CStr::from_ptr(pbs_err) };
> +        anyhow::bail!("proxmox_backup_connect failed: {pbs_err_cstr:?}");
> +    }
> +
> +    println!("Connected to Proxmox Backup Server");
> +
> +    let start_transfer_time = SystemTime::now();
> +
> +    upload_configs(&vma_reader, pbs)?;
> +    upload_block_devices(vma_reader, pbs)?;
> +
> +    if proxmox_backup_finish(pbs, &mut pbs_err) < 0 {
> +        let pbs_err_cstr = unsafe { CStr::from_ptr(pbs_err) };
> +        anyhow::bail!("proxmox_backup_finish failed: {pbs_err_cstr:?}");
> +    }

Same as above (error handling).

> +
> +    let elapsed_time = SystemTime::now().duration_since(start_transfer_time)?;
> +    let total_ms = elapsed_time.as_millis();
> +    let ms = total_ms % 1000;
> +    let seconds = (total_ms / 1000) % 60;
> +    let minutes = total_ms / 1000000;
> +    println!("Backup finished within {minutes} minutes, {seconds} seconds and {ms} ms");
> +
> +    Ok(())
> +}

[0]: https://doc.rust-lang.org/std/ffi/struct.CStr.html#method.from_ptr