From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [IPv6:2a0f:8001:1:32::40]) by lore.proxmox.com (Postfix) with ESMTPS id ACC311FF0E1 for ; Mon, 13 Jul 2026 11:18:14 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id EC70121344; Mon, 13 Jul 2026 11:18:13 +0200 (CEST) Mime-Version: 1.0 Content-Transfer-Encoding: quoted-printable Content-Type: text/plain; charset=UTF-8 Date: Mon, 13 Jul 2026 11:18:05 +0200 Message-Id: Subject: Re: [PATCH datacenter-manager 15/15] task cache: handle potentially duplicated archive files after 'compress_archive_file' From: "Lukas Wagner" To: "Dominik Csapak" , "Lukas Wagner" , X-Mailer: aerc 0.21.0-0-g5549850facc2-dirty References: <20260702092258.174740-1-l.wagner@proxmox.com> <20260702092258.174740-16-l.wagner@proxmox.com> In-Reply-To: X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1783934270341 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.333 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_LOW -0.7 Sender listed at https://www.dnswl.org/, low trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: IVI67TXJ6D4BVTOXPROURURZHN6MFUOS X-Message-ID-Hash: IVI67TXJ6D4BVTOXPROURURZHN6MFUOS X-MailFrom: l.wagner@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox Datacenter Manager development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: 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. >>=20 >> This is handled by ignoring duplicates files when reading/writing >> from/into the task archive, and by actively cleaning up the duplicates >> during `rotate`. >>=20 >> Signed-off-by: Lukas Wagner >> --- >> server/src/remote_tasks/task_cache.rs | 97 +++++++++++++++++++++++++-- >> 1 file changed, 93 insertions(+), 4 deletions(-) >>=20 >> diff --git a/server/src/remote_tasks/task_cache.rs b/server/src/remote_t= asks/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) >> } >> =20 >> - /// Rotate task archive if the the newest archive file is older tha= n `rotate_after`. >> + /// Rotate task archive if the newest archive file is older than `r= otate_after`. >> /// >> /// The oldest archive files are removed if the total number of ar= chive files exceeds >> /// `max_files`. `now` is supposed to be a UNIX timestamp (seconds= ). >> + /// >> + /// If there are any duplicate archive files (can happen if `compre= ss` is interrupted at the >> + /// wrong time), the uncompressed version is deleted. >> pub fn rotate(&self, now: i64) -> Result { >> let mut did_rotate =3D false; >> - let mut archive_files =3D self.cache.archive_files(&self.lock)?= ; >> + let archive_files_with_dupes =3D self.cache.archive_files_with_= dupes(&self.lock)?; >> + >> + let mut deduped: Vec =3D Vec::new(); >> + >> + for file in archive_files_with_dupes { >> + if deduped.iter().any(|e| e.starttime =3D=3D file.starttime= ) { >> + // Found dupe, remove it. >> + if let Err(err) =3D std::fs::remove_file(&file.path) { >> + log::error!( >> + "could not clean up duplicate archive file '{pa= th}': {err}", >> + path =3D 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 =3D deduped; >> =20 >> let mut start_new_file =3D |files: &mut Vec| -> R= esult<(), Error> { >> let new_file =3D self.new_file(now, self.cache.uncompresse= d_files =3D=3D 0)?; >> @@ -848,6 +869,11 @@ impl<'a> WritableTaskCache<'a> { >> return Err(err); >> } >> =20 >> + // 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. `arch= ive_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 unc= ompressed archive file")?; >> =20 >> Ok(file_state) >> @@ -970,7 +996,7 @@ impl TaskCache { >> /// cut-off timestamp. The result is sorted ascending by cut-off t= imestamp (most recent one >> /// first). >> /// The task archive should be locked for reading when calling thi= s function. >> - fn archive_files(&self, _lock: &TaskCacheLock) -> Result, Error> { >> + fn archive_files_with_dupes(&self, _lock: &TaskCacheLock) -> Result= , Error> { >> let mut names =3D Vec::new(); >> =20 >> for entry in std::fs::read_dir(&self.base_path)? { >> @@ -983,7 +1009,23 @@ impl TaskCache { >> } >> } >> =20 >> - 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 r= espective >> + /// cut-off timestamp. The result is sorted ascending by cut-off ti= mestamp (most recent one >> + /// first). Any duplicates (equal starttime but different compressi= on status) are removed. >> + /// >> + /// The task archive should be locked for reading when calling this= function. >> + fn archive_files(&self, lock: &TaskCacheLock) -> Result, Error> { >> + let mut names =3D self.archive_files_with_dupes(lock)?; >> + names.dedup_by_key(|e| e.starttime); >> =20 >> Ok(names) >> } >> @@ -2104,4 +2146,51 @@ mod tests { >> "cutoff timestamp should be reset to the lower bound of th= e corrupted archive file" >> ); >> } >> + >> + /// Ensure that if for any reason there exist multiple files with t= he same start time (only >> + /// possible if one of them is compressed and the other one uncompr= essed), we only return one >> + /// of them. Also verify that compressed files are preferred. >> + #[test] >> + fn dedup_duplicate_archive_files() { >> + let (_tmp_dir, mut cache) =3D make_cache().unwrap(); >> + cache.rotate_after =3D 100; >> + cache.uncompressed_files =3D 1; >> + cache.max_files =3D 3; >> + >> + let cache =3D 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 =3D cache.cache.archive_files(&cache.lock).unwrap(); >> + >> + assert_eq!(files.len(), 2); >> + let first =3D files.get(0).unwrap(); >> + let second =3D files.get(1).unwrap(); >> + >> + assert!(first.compressed); >> + assert_eq!(first.starttime, 2000); >> + >> + assert!(second.compressed); >> + assert_eq!(second.starttime, 1000); >> + >> + let files =3D cache.cache.archive_files_with_dupes(&cache.lock)= .unwrap(); >> + assert_eq!(files.len(), 4); >> + >> + cache.rotate(2050).unwrap(); >> + >> + let files =3D cache.cache.archive_files_with_dupes(&cache.lock)= .unwrap(); >> + assert_eq!(files.len(), 2); >> + >> + let first =3D files.get(0).unwrap(); >> + let second =3D files.get(1).unwrap(); >> + >> + assert!(first.compressed); >> + assert_eq!(first.starttime, 2000); >> + >> + assert!(second.compressed); >> + assert_eq!(second.starttime, 1000); >> + } >> }