[PATCH proxmox-backup v3 27/30] sync: pull: decrypt chunks and rewrite index file for matching key

Christian Ebner
2026-04-14T13:07:55Z

Once the matching decryptioin key will be provided, use it to decrypt
the chunks on pull and rewrite the index file based on the decrypted
chunk digests and offsets.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
changes since version 2:
- no changes

src/server/pull.rs | 135 ++++++++++++++++++++++++++++++++++++++-------
1 file changed, 114 insertions(+), 21 deletions(-)

diff --git a/src/server/pull.rs b/src/server/pull.rs
index 75958d625..90cb99777 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -4,7 +4,7 @@ use std::collections::hash_map::Entry;
use std::collections::{HashMap, HashSet};
use std::io::{BufReader, Read, Seek, Write};
use std::os::fd::AsRawFd;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::SystemTime;

@@ -21,7 +21,7 @@ use pbs_api_types::{
};
use pbs_client::BackupRepository;
use pbs_config::CachedUserInfo;
-use pbs_datastore::data_blob::DataBlob;
+use pbs_datastore::data_blob::{DataBlob, DataChunkBuilder};
use pbs_datastore::dynamic_index::{DynamicIndexReader, DynamicIndexWriter};
use pbs_datastore::fixed_index::{FixedIndexReader, FixedIndexWriter};
use pbs_datastore::index::IndexFile;
@@ -181,7 +181,16 @@ async fn pull_index_chunks<I: IndexFile>(
.filter(|info| {
let guard = encountered_chunks.lock().unwrap();
match guard.check_reusable(&info.digest) {
- Some((touched, _decrypted_chunk)) => !touched, // reusable and already touched, can always skip
+ Some((touched, mapped_digest)) => {
+ if mapped_digest.is_some() {
+ // if there is a mapping, then the chunk digest must be rewritten to
+ // the index, cannot skip here but optimized when processing the stream
+ true
+ } else {
+ // reusable and already touched, can always skip
+ !touched
+ }
+ }
None => true,
}
}),
@@ -203,6 +212,7 @@ async fn pull_index_chunks<I: IndexFile>(
let verify_and_write_channel = verify_pool.channel();

let bytes = Arc::new(AtomicUsize::new(0));
+ let offset = Arc::new(AtomicU64::new(0));
let chunk_count = Arc::new(AtomicUsize::new(0));

let stream = stream.map(|info| {
@@ -212,36 +222,119 @@ async fn pull_index_chunks<I: IndexFile>(
let chunk_count = Arc::clone(&chunk_count);
let verify_and_write_channel = verify_and_write_channel.clone();
let encountered_chunks = Arc::clone(&encountered_chunks);
+ let offset = Arc::clone(&offset);
+ let decrypted_index_writer = decrypted_index_writer.clone();

Ok::<_, Error>(async move {
- {
- // limit guard scope
- let mut guard = encountered_chunks.lock().unwrap();
- if let Some((touched, _decrypted_digest)) = guard.check_reusable(&info.digest) {
- if touched {
+ //info!("sync {} chunk {}", pos, hex::encode(digest));
+ let (chunk, digest, size) = match decrypted_index_writer {
+ DecryptedIndexWriter::Fixed(index) => {
+ if let Some((_touched, Some(decrypted_digest))) = encountered_chunks
+ .lock()
+ .unwrap()
+ .check_reusable(&info.digest)
+ {
+ // already got the decrypted digest and chunk has been written,
+ // no need to process again
+ let size = info.size();
+ let start_offset = offset.fetch_add(size, Ordering::SeqCst);
+
+ index.lock().unwrap().add_chunk(
+ start_offset,
+ size as u32,
+ decrypted_digest,
+ )?;
+
return Ok::<_, Error>(());
}
- let chunk_exists = proxmox_async::runtime::block_in_place(|| {
- target.cond_touch_chunk(&info.digest, false)
- })?;
- if chunk_exists {
- guard.mark_touched(&info.digest, None);
- //info!("chunk {} exists {}", pos, hex::encode(digest));
+
+ let chunk_data = chunk_reader.read_chunk(&info.digest).await?;
+ let (chunk, digest) =
+ DataChunkBuilder::new(&chunk_data).compress(true).build()?;
+
+ let size = chunk_data.len() as u64;
+ let start_offset = offset.fetch_add(size, Ordering::SeqCst);
+
+ index
+ .lock()
+ .unwrap()
+ .add_chunk(start_offset, size as u32, &digest)?;
+
+ encountered_chunks
+ .lock()
+ .unwrap()
+ .mark_reusable(&info.digest, Some(digest));
+
+ (chunk, digest, size)
+ }
+ DecryptedIndexWriter::Dynamic(index) => {
+ if let Some((_touched, Some(decrypted_digest))) = encountered_chunks
+ .lock()
+ .unwrap()
+ .check_reusable(&info.digest)
+ {
+ // already got the decrypted digest and chunk has been written,
+ // no need to process again
+ let size = info.size();
+ let start_offset = offset.fetch_add(size, Ordering::SeqCst);
+ let end_offset = start_offset + size;
+
+ index
+ .lock()
+ .unwrap()
+ .add_chunk(end_offset, decrypted_digest)?;
+
return Ok::<_, Error>(());
}
+
+ let chunk_data = chunk_reader.read_chunk(&info.digest).await?;
+ let (chunk, digest) =
+ DataChunkBuilder::new(&chunk_data).compress(true).build()?;
+
+ let size = chunk_data.len() as u64;
+ let start_offset = offset.fetch_add(size, Ordering::SeqCst);
+ let end_offset = start_offset + size;
+
+ index.lock().unwrap().add_chunk(end_offset, &digest)?;
+
+ encountered_chunks
+ .lock()
+ .unwrap()
+ .mark_reusable(&info.digest, Some(digest));
+
+ (chunk, digest, size)
}
- // mark before actually downloading the chunk, so this happens only once
- guard.mark_reusable(&info.digest, None);
- guard.mark_touched(&info.digest, None);
- }
+ DecryptedIndexWriter::None => {
+ {
+ // limit guard scope
+ let mut guard = encountered_chunks.lock().unwrap();
+ if let Some((touched, _mapped)) = guard.check_reusable(&info.digest) {
+ if touched {
+ return Ok::<_, Error>(());
+ }
+ let chunk_exists = proxmox_async::runtime::block_in_place(|| {
+ target.cond_touch_chunk(&info.digest, false)
+ })?;
+ if chunk_exists {
+ guard.mark_touched(&info.digest, None);
+ //info!("chunk {} exists {}", pos, hex::encode(digest));
+ return Ok::<_, Error>(());
+ }
+ }
+ // mark before actually downloading the chunk, so this happens only once
+ guard.mark_reusable(&info.digest, None);
+ guard.mark_touched(&info.digest, None);
+ }

- //info!("sync {} chunk {}", pos, hex::encode(digest));
- let chunk = chunk_reader.read_raw_chunk(&info.digest).await?;
+ let chunk = chunk_reader.read_raw_chunk(&info.digest).await?;
+ (chunk, info.digest, info.size())
+ }
+ };
let raw_size = chunk.raw_size() as usize;

// decode, verify and write in a separate threads to maximize throughput
proxmox_async::runtime::block_in_place(|| {
- verify_and_write_channel.send((chunk, info.digest, info.size()))
+ verify_and_write_channel.send((chunk, digest, size))
})?;

bytes.fetch_add(raw_size, Ordering::SeqCst);
--
2.47.3