Re: [PATCH datacenter-manager 15/15] task cache: handle potentially duplicated archive files after 'compress_archive_file'

["Lukas Wagner" <l.wagner@proxmox.com>]

On Fri Jul 10, 2026 at 1:36 PM CEST, Dominik Csapak wrote:
> comment inline
>
> On 7/2/26 11:23 AM, Lukas Wagner wrote:
>> `compress_archive_file` cannot be made fully atomic, therefore we must
>> consider the case where we have both, the old, uncompressed file and the
>> new, compressed file in the archive directory.
>> 
>> This is handled by ignoring duplicates files when reading/writing
>> from/into the task archive, and by actively cleaning up the duplicates
>> during `rotate`.
>> 
>> Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
>> ---
>>   server/src/remote_tasks/task_cache.rs | 97 +++++++++++++++++++++++++--
>>   1 file changed, 93 insertions(+), 4 deletions(-)
>> 
>> diff --git a/server/src/remote_tasks/task_cache.rs b/server/src/remote_tasks/task_cache.rs
>> index 23238cca..2110037b 100644
>> --- a/server/src/remote_tasks/task_cache.rs
>> +++ b/server/src/remote_tasks/task_cache.rs
>> @@ -289,13 +289,34 @@ impl<'a> WritableTaskCache<'a> {
>>           Ok(file)
>>       }
>>   
>> -    /// Rotate task archive if the the newest archive file is older than `rotate_after`.
>> +    /// Rotate task archive if the newest archive file is older than `rotate_after`.
>>       ///
>>       /// The oldest archive files are removed if the total number of archive files exceeds
>>       /// `max_files`. `now` is supposed to be a UNIX timestamp (seconds).
>> +    ///
>> +    /// If there are any duplicate archive files (can happen if `compress` is interrupted at the
>> +    /// wrong time), the uncompressed version is deleted.
>>       pub fn rotate(&self, now: i64) -> Result<bool, Error> {
>>           let mut did_rotate = false;
>> -        let mut archive_files = self.cache.archive_files(&self.lock)?;
>> +        let archive_files_with_dupes = self.cache.archive_files_with_dupes(&self.lock)?;
>> +
>> +        let mut deduped: Vec<ArchiveFile> = Vec::new();
>> +
>> +        for file in archive_files_with_dupes {
>> +            if deduped.iter().any(|e| e.starttime == file.starttime) {
>> +                // Found dupe, remove it.
>> +                if let Err(err) = std::fs::remove_file(&file.path) {
>> +                    log::error!(
>> +                        "could not clean up duplicate archive file '{path}': {err}",
>> +                        path = file.path.display()
>> +                    )
>> +                }
>> +            } else {
>> +                deduped.push(file);
>> +            }
>> +        }
>
> i think we could avoid this double loop here if we use a hashset?

Changed it to a HashSet in v2.
Given the low number of archive files (7 at the moment, we might raise
them later I guess), the old approach with the vec was probably more
efficient, which was the reason why I implemented it that way.
However, I do agree that a HashSet captures the intent better, and also
it scales better if we ever to decide to allow having a considerably
larger number of archive files.

>
> also we only detect duplicates by the starttime?
> sound a bit strange to me, wouldn't the more natural way
> by looking at e.g. the filenames?
>

Yes, looking at the starttimes is sufficient here. All archive files are
in the same base directory, and the only kind of dupes that we care
about is if there is are two files with the same starttime, but on
compressed, the other one uncompressed (so say, archive.1000 and
archive.1000.zst) -- which can happen if `rotate` is interrupted at just
the right moment.

I've added a comment in v2 that tries to explain this better.

Thanks!

>> +
>> +        let mut archive_files = deduped;
>>   
>>           let mut start_new_file = |files: &mut Vec<ArchiveFile>| -> Result<(), Error> {
>>               let new_file = self.new_file(now, self.cache.uncompressed_files == 0)?;
>> @@ -848,6 +869,11 @@ impl<'a> WritableTaskCache<'a> {
>>               return Err(err);
>>           }
>>   
>> +        // If we crash here or if `remove_file` fails, we might end up with the original,
>> +        // uncompressed file as well as the new, compressed file. `archive_files` filters
>> +        // duplicates out, so we should never read from the duplicated file. Any left-over
>> +        // duplicates are cleaned during `rotate`.
>> +
>>           std::fs::remove_file(&file.path).context("failed to remove uncompressed archive file")?;
>>   
>>           Ok(file_state)
>> @@ -970,7 +996,7 @@ impl TaskCache {
>>       /// cut-off timestamp. The result is sorted ascending by cut-off timestamp (most recent one
>>       /// first).
>>       /// The task archive should be locked for reading when calling this function.
>> -    fn archive_files(&self, _lock: &TaskCacheLock) -> Result<Vec<ArchiveFile>, Error> {
>> +    fn archive_files_with_dupes(&self, _lock: &TaskCacheLock) -> Result<Vec<ArchiveFile>, Error> {
>>           let mut names = Vec::new();
>>   
>>           for entry in std::fs::read_dir(&self.base_path)? {
>> @@ -983,7 +1009,23 @@ impl TaskCache {
>>               }
>>           }
>>   
>> -        names.sort_by_key(|e| -e.starttime);
>> +        names.sort_by(|a, b| {
>> +            b.starttime
>> +                .cmp(&a.starttime)
>> +                .then(b.compressed.cmp(&a.compressed))
>> +        });
>> +
>> +        Ok(names)
>> +    }
>> +
>> +    /// Returns a list of existing archive files, together with their respective
>> +    /// cut-off timestamp. The result is sorted ascending by cut-off timestamp (most recent one
>> +    /// first). Any duplicates (equal starttime but different compression status) are removed.
>> +    ///
>> +    /// The task archive should be locked for reading when calling this function.
>> +    fn archive_files(&self, lock: &TaskCacheLock) -> Result<Vec<ArchiveFile>, Error> {
>> +        let mut names = self.archive_files_with_dupes(lock)?;
>> +        names.dedup_by_key(|e| e.starttime);
>>   
>>           Ok(names)
>>       }
>> @@ -2104,4 +2146,51 @@ mod tests {
>>               "cutoff timestamp should be reset to the lower bound of the corrupted archive file"
>>           );
>>       }
>> +
>> +    /// Ensure that if for any reason there exist multiple files with the same start time (only
>> +    /// possible if one of them is compressed and the other one uncompressed), we only return one
>> +    /// of them. Also verify that compressed files are preferred.
>> +    #[test]
>> +    fn dedup_duplicate_archive_files() {
>> +        let (_tmp_dir, mut cache) = make_cache().unwrap();
>> +        cache.rotate_after = 100;
>> +        cache.uncompressed_files = 1;
>> +        cache.max_files = 3;
>> +
>> +        let cache = cache.write().unwrap();
>> +
>> +        cache.new_file(1000, false).unwrap();
>> +        cache.new_file(1000, true).unwrap();
>> +        cache.new_file(2000, true).unwrap();
>> +        cache.new_file(2000, false).unwrap();
>> +
>> +        let files = cache.cache.archive_files(&cache.lock).unwrap();
>> +
>> +        assert_eq!(files.len(), 2);
>> +        let first = files.get(0).unwrap();
>> +        let second = files.get(1).unwrap();
>> +
>> +        assert!(first.compressed);
>> +        assert_eq!(first.starttime, 2000);
>> +
>> +        assert!(second.compressed);
>> +        assert_eq!(second.starttime, 1000);
>> +
>> +        let files = cache.cache.archive_files_with_dupes(&cache.lock).unwrap();
>> +        assert_eq!(files.len(), 4);
>> +
>> +        cache.rotate(2050).unwrap();
>> +
>> +        let files = cache.cache.archive_files_with_dupes(&cache.lock).unwrap();
>> +        assert_eq!(files.len(), 2);
>> +
>> +        let first = files.get(0).unwrap();
>> +        let second = files.get(1).unwrap();
>> +
>> +        assert!(first.compressed);
>> +        assert_eq!(first.starttime, 2000);
>> +
>> +        assert!(second.compressed);
>> +        assert_eq!(second.starttime, 1000);
>> +    }
>>   }