[PATCH datacenter-manager 08/15] task cache: archive iterator: don't yield more items if reading from file failed

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

Otherwise `get_tasks` runs into an endless loop if any of the archive
files is corrupted.

Also add a test case that verifies the absence of the issue.

Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
 server/src/remote_tasks/task_cache.rs | 48 +++++++++++++++++++++++----
 1 file changed, 42 insertions(+), 6 deletions(-)

diff --git a/server/src/remote_tasks/task_cache.rs b/server/src/remote_tasks/task_cache.rs
index f0e3533a..02be9ca1 100644
--- a/server/src/remote_tasks/task_cache.rs
+++ b/server/src/remote_tasks/task_cache.rs
@@ -1244,6 +1244,7 @@ where
 /// tasks are read line by line, without leading the entire archive file into memory.
 struct ArchiveIterator {
     iter: Lines<Box<dyn BufRead>>,
+    failed: bool,
 }
 
 impl ArchiveIterator {
@@ -1251,7 +1252,10 @@ impl ArchiveIterator {
     pub fn new(reader: Box<dyn BufRead>) -> Self {
         let lines = reader.lines();
 
-        Self { iter: lines }
+        Self {
+            iter: lines,
+            failed: false,
+        }
     }
 }
 
@@ -1259,11 +1263,18 @@ impl Iterator for ArchiveIterator {
     type Item = Result<TaskCacheItem, Error>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        self.iter.next().map(|result| {
-            result
-                .and_then(|line| Ok(serde_json::from_str(&line)?))
-                .map_err(Into::into)
-        })
+        if self.failed {
+            // Don't return any more items if we have failed reading a line once
+            return None;
+        }
+
+        self.iter.next().map(|line| match line {
+            Ok(line) => Some(serde_json::from_str(&line).context("failed to decode JSON")),
+            Err(err) => {
+                self.failed = true;
+                Some(Err(err).context("failed to read line"))
+            }
+        })?
     }
 }
 
@@ -1661,4 +1672,29 @@ mod tests {
         assert_eq!(first.iter().unwrap().unwrap().count(), 0);
         assert_eq!(second.iter().unwrap().unwrap().count(), 1);
     }
+
+    #[test]
+    fn corrupted_archive_file_does_not_lead_to_endless_loop() {
+        let (_tmp_dir, cache) = make_cache().unwrap();
+        let cache = cache.write().unwrap();
+
+        // Create compressed file
+        cache.new_file(1000, true).unwrap();
+        add_tasks(&cache, vec![task(1100, Some(1110))]).unwrap();
+        cache.apply_journal().unwrap();
+
+        assert_eq!(cache.get_tasks(GetTasks::Archived).unwrap().count(), 1);
+
+        let files = cache.cache.archive_files(&cache.lock).unwrap();
+        let file = files.get(0).expect("there is one archive file");
+
+        // truncate existing compressed file, corrupting the zstd file header
+        let _file = OpenOptions::new()
+            .write(true)
+            .truncate(true)
+            .open(&file.path)
+            .expect("file truncated");
+
+        assert_eq!(cache.get_tasks(GetTasks::Archived).unwrap().count(), 0);
+    }
 }
-- 
2.47.3