[pbs-devel] [PATCH vma-to-pbs 4/9] add support for streaming the VMA file via stdin

[Filip Schauer <f.schauer@proxmox.com>]

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>
---
 src/main.rs    | 241 ++-----------------------------
 src/vma.rs     | 330 +++++++++++++++++++-----------------------
 src/vma2pbs.rs | 383 +++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 547 insertions(+), 407 deletions(-)
 create mode 100644 src/vma2pbs.rs

diff --git a/src/main.rs b/src/main.rs
index 8f0c9b2..aa76ce1 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::{Context, 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 {
@@ -344,18 +127,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)?;
 
     Ok(())
 }
diff --git a/src/vma.rs b/src/vma.rs
index d30cb09..447e8db 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 anyhow::{anyhow, bail, 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,
+pub struct VmaReader<T> {
+    vma_file: T,
     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();
+impl<T: Read> VmaReader<T> {
+    pub fn new(mut vma_file: T) -> Result<Self> {
+        let vma_header = Self::read_header(&mut vma_file)?;
+        let configs = Self::read_configs(&vma_header)?;
 
         let instance = Self {
             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 T) -> 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;
+            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;
+            bail!("device_name_offset cannot be 0");
         }
 
-        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..5b9e20f
--- /dev/null
+++ b/src/vma2pbs.rs
@@ -0,0 +1,383 @@
+use std::cell::RefCell;
+use std::collections::hash_map::Entry;
+use std::collections::HashMap;
+use std::ffi::{c_char, CStr, CString};
+use std::fs::File;
+use std::io::{stdin, BufRead, BufReader, Read};
+use std::ptr;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use anyhow::{anyhow, bail, 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> {
+    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: {}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
+        1970 + backup_time / (365 * 24 * 3600),
+        (backup_time / (30 * 24 * 3600)) % 12 + 1,
+        (backup_time / (24 * 3600)) % 30 + 1,
+        (backup_time / 3600) % 24,
+        (backup_time / 60) % 60,
+        backup_time % 60
+    );
+
+    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) };
+        bail!("proxmox_backup_new_ns failed: {pbs_err_cstr:?}");
+    }
+
+    Ok(pbs)
+}
+
+fn upload_configs<T>(vma_reader: &VmaReader<T>, pbs: *mut ProxmoxBackupHandle) -> Result<()>
+where
+    T: Read,
+{
+    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) };
+            bail!("proxmox_backup_add_config failed: {pbs_err_cstr:?}");
+        }
+    }
+
+    Ok(())
+}
+
+fn register_block_devices<T>(
+    vma_reader: &VmaReader<T>,
+    pbs: *mut ProxmoxBackupHandle,
+) -> Result<[Option<BlockDeviceInfo>; 256]>
+where
+    T: Read,
+{
+    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) };
+            bail!("proxmox_backup_register_image failed: {pbs_err_cstr:?}");
+        }
+
+        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<T>(
+    mut vma_reader: VmaReader<T>,
+    pbs: *mut ProxmoxBackupHandle,
+) -> Result<()>
+where
+    T: Read,
+{
+    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());
+
+    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 => 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 = match images_chunks.entry(dev_id) {
+            Entry::Occupied(image_chunks) => image_chunks.into_mut(),
+            Entry::Vacant(image_chunks) => image_chunks.insert(HashMap::new()),
+        };
+        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 prepare_pbs_chunk = |sub_chunk_count: u8, image_chunk: &ImageChunk| {
+            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_chunk_buffer
+        };
+
+        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) };
+                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);
+        };
+
+        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 pbs_chunk_buffer = prepare_pbs_chunk(sub_chunk_count, image_chunk);
+                        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);
+                }
+            }
+        }
+
+        Ok(())
+    })?;
+
+    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) };
+            bail!("proxmox_backup_close_image failed: {pbs_err_cstr:?}");
+        }
+    }
+
+    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())),
+    };
+    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) };
+        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) };
+        bail!("proxmox_backup_finish failed: {pbs_err_cstr:?}");
+    }
+
+    let elapsed_ms = SystemTime::now()
+        .duration_since(start_transfer_time)?
+        .as_millis();
+    println!(
+        "Backup finished within {} minutes, {} seconds and {} ms",
+        elapsed_ms / 1000000,
+        (elapsed_ms / 1000) % 60,
+        elapsed_ms % 1000
+    );
+
+    Ok(())
+}
-- 
2.39.2