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

[Dominik Csapak <d.csapak@proxmox.com>]

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?

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?

> +
> +        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);
> +    }
>   }