fix(hummingbird): go back to stateless download

Previous work attempted to make the hummingbird provider a little bit
faster by more caching in the input directory, but this has possible
failure modes and raises the disk space requirements. Instead, just
always recreate the input directory by downloading the archive and then
following changes.csv and deletions.txt for information about what needs
to change after the archive.

Additionally, unlink the archive extraction dir before re-extracting to
prevent disk space exhaustion, and tag vunnel as "large" to give it a
bit more space.

Signed-off-by: Will Murphy <willmurphyscode@users.noreply.github.com>

Author: willmurphyscode

src/vunnel/providers/hummingbird/__init__.py

@@ -57,7 +57,7 @@ def name(cls) -> str:
 
     @classmethod
     def tags(cls) -> list[str]:
-        return ["vulnerability", "os"]
+        return ["vulnerability", "os", "large"]
 
     def update(self, last_updated: datetime.datetime | None) -> tuple[list[str], int]:
         with timer(self.name(), self.logger):

src/vunnel/providers/hummingbird/csaf_client.py

@@ -9,8 +9,6 @@
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING
 
-import requests
-
 from vunnel.utils import http_wrapper as http
 from vunnel.utils.archive import extract
 
@@ -21,11 +19,13 @@
 
 VEX_FEED_LATEST_URL = "https://security.access.redhat.com/data/csaf/v2/vex-feed/archive_latest.txt"
 
-TIMESTAMP_FILE = "archive_timestamp.txt"
-CHANGES_TIMESTAMP_FILE = "changes_timestamp.txt"
-
 ARCHIVE_SUFFIXES = (".tar", ".tar.zst", ".tar.gz", ".tar.xz", ".tar.bz2")
 
+# state files written by a previous incremental version of this client; they are
+# meaningless now that every run downloads the archive, but would otherwise
+# persist in the workspace (and any cached copies of it) forever
+LEGACY_STATE_FILES = ("archive_timestamp.txt", "changes_timestamp.txt")
+
 
 class CSAFVEXClient:
     def __init__(
@@ -47,7 +47,6 @@ def __init__(
         if not skip_download:
             self._sync()
         else:
-            self._load_timestamp()
             self.logger.info("skipping downloads in hummingbird CSAF VEX client")
 
     @staticmethod
@@ -79,47 +78,12 @@ def _archive_url(self) -> str:
 
     # ── local paths ───────────────────────────────────────────────────
 
-    def _timestamp_path(self) -> str:
-        return os.path.join(self.workspace.input_path, TIMESTAMP_FILE)
-
-    def _changes_timestamp_path(self) -> str:
-        return os.path.join(self.workspace.input_path, CHANGES_TIMESTAMP_FILE)
-
     def _local_changes_path(self) -> str:
         return os.path.join(self.workspace.input_path, "changes.csv")
 
     def _local_deletions_path(self) -> str:
         return os.path.join(self.workspace.input_path, "deletions.csv")
 
-    # ── timestamp persistence ─────────────────────────────────────────
-
-    def _load_timestamp(self) -> None:
-        ts_path = self._timestamp_path()
-        if os.path.exists(ts_path):
-            with open(ts_path) as fh:
-                self.archive_mod_time = datetime.fromisoformat(fh.read().strip())
-            self.logger.debug(f"loaded archive timestamp: {self.archive_mod_time}")
-
-    def _save_timestamp(self, mod_time: datetime) -> None:
-        self.archive_mod_time = mod_time
-        with open(self._timestamp_path(), "w") as fh:
-            fh.write(mod_time.isoformat())
-
-    def _load_changes_timestamp(self) -> datetime | None:
-        ts_path = self._changes_timestamp_path()
-        if os.path.exists(ts_path):
-            with open(ts_path) as fh:
-                return datetime.fromisoformat(fh.read().strip())
-        return None
-
-    def _save_changes_timestamp(self, mod_time: datetime) -> None:
-        with open(self._changes_timestamp_path(), "w") as fh:
-            fh.write(mod_time.isoformat())
-
-    def _clear_changes_timestamp(self) -> None:
-        with contextlib.suppress(FileNotFoundError):
-            os.remove(self._changes_timestamp_path())
-
     # ── download helpers ──────────────────────────────────────────────
 
     def _download_stream(self, url: str, path: str) -> datetime | None:
@@ -133,41 +97,18 @@ def _download_stream(self, url: str, path: str) -> datetime | None:
             return email.utils.parsedate_to_datetime(lm)
         return None
 
-    def _head_last_modified(self, url: str) -> datetime | None:
-        """HEAD the URL and return Last-Modified as a tz-aware datetime, or None."""
-        resp = requests.head(url, timeout=30)
-        resp.raise_for_status()
-        lm = resp.headers.get("Last-Modified")
-        if lm:
-            return email.utils.parsedate_to_datetime(lm)
-        return None
-
     # ── core sync logic ───────────────────────────────────────────────
 
     def _sync(self) -> None:
         os.makedirs(self.advisories_path, exist_ok=True)
         self._remove_stray_files()
 
-        archive_url = self._archive_url()
-        self._load_timestamp()
+        # stateless flow: every run downloads and extracts the archive, then
+        # re-applies everything that changed since the archive was baked. this
+        # trades some redundant downloading for having no persisted state that
+        # can disagree with what is actually on disk.
+        self._download_archive(self._archive_url())
 
-        # decide whether we need to (re-)download the archive
-        need_download = False
-        if self.archive_mod_time is None:
-            self.logger.info("no local timestamp found - downloading archive")
-            need_download = True
-        else:
-            remote_mod = self._head_last_modified(archive_url)
-            if remote_mod and remote_mod > self.archive_mod_time:
-                self.logger.info(f"remote archive is newer ({remote_mod}) than local ({self.archive_mod_time}) - re-downloading")
-                need_download = True
-            else:
-                self.logger.info("archive is up to date")
-
-        if need_download:
-            self._download_archive(archive_url)
-
-        # always apply incremental updates
         self._download_stream(self._changes_url(), self._local_changes_path())
         self._download_stream(self._deletions_url(), self._local_deletions_path())
         self._process_changes_and_deletions()
@@ -186,6 +127,10 @@ def _remove_stray_files(self) -> None:
                 with contextlib.suppress(OSError):
                     os.remove(os.path.join(self.workspace.input_path, name))
 
+        for name in LEGACY_STATE_FILES:
+            with contextlib.suppress(FileNotFoundError):
+                os.remove(os.path.join(self.workspace.input_path, name))
+
     def _advisories_tmp_path(self) -> str:
         return self.advisories_path + ".tmp"
 
@@ -197,19 +142,18 @@ def _download_archive(self, archive_url: str) -> None:
         try:
             remote_mod = self._download_stream(archive_url, archive_path)
 
-            # extract to a temp dir and swap so a failed extraction can't
-            # destroy the existing advisories tree
+            # the downloaded archive supersedes the existing advisories tree, so
+            # remove the tree before extracting to keep peak disk usage at one
+            # tree plus the archive (instead of two trees plus the archive)
+            if os.path.isdir(self.advisories_path):
+                shutil.rmtree(self.advisories_path)
+
             self.logger.info("extracting archive")
             tmp_dir = self._advisories_tmp_path()
             extract(archive_path, tmp_dir)
-
-            if os.path.isdir(self.advisories_path):
-                shutil.rmtree(self.advisories_path)
             os.rename(tmp_dir, self.advisories_path)
 
-            self._save_timestamp(remote_mod or datetime.now(tz=UTC))
-            # the new tree supersedes any previously applied changes
-            self._clear_changes_timestamp()
+            self.archive_mod_time = remote_mod or datetime.now(tz=UTC)
         finally:
             # always clean up the archive file to save disk space
             with contextlib.suppress(OSError):
@@ -218,7 +162,7 @@ def _download_archive(self, archive_url: str) -> None:
     def _process_changes_and_deletions(self) -> None:
         self._apply_deletions()
 
-        seen_files, years, newest_change = self._collect_pending_changes()
+        seen_files, years = self._collect_pending_changes()
 
         if not seen_files:
             self.logger.info("no changed files newer than archive")
@@ -229,11 +173,7 @@ def _process_changes_and_deletions(self) -> None:
         for year in years:
             os.makedirs(os.path.join(self.advisories_path, year), exist_ok=True)
 
-        any_failed = self._download_changed_files(seen_files)
-
-        # only advance the watermark when everything landed, so failed files are retried next run
-        if not any_failed and newest_change:
-            self._save_changes_timestamp(newest_change)
+        self._download_changed_files(seen_files)
 
     def _apply_deletions(self) -> None:
         with open(self._local_deletions_path(), newline="") as fh:
@@ -243,37 +183,24 @@ def _apply_deletions(self) -> None:
                 with contextlib.suppress(FileNotFoundError):
                     os.remove(os.path.join(self.advisories_path, deleted_fragment))
 
-    def _collect_pending_changes(self) -> tuple[set[str], set[str], datetime | None]:
-        """Read changes.csv (newest first) and return (files, years, newest change date) not yet applied."""
-        # skip changes already applied by a previous run (the watermark), falling
-        # back to the archive timestamp when no changes have been applied yet
-        watermark = self.archive_mod_time
-        changes_applied = self._load_changes_timestamp()
-        if changes_applied and (watermark is None or changes_applied > watermark):
-            watermark = changes_applied
-
+    def _collect_pending_changes(self) -> tuple[set[str], set[str]]:
+        """Read changes.csv (newest first) and return (files, years) for entries newer than the archive."""
         seen_files: set[str] = set()
         years: set[str] = set()
-        newest_change: datetime | None = None
         with open(self._local_changes_path(), newline="") as fh:
             reader = csv.reader(fh)
             for row in reader:
                 changed_file = row[0]
-                date_str = row[1]
-                change_date = datetime.fromisoformat(date_str)
-                if watermark and change_date < watermark:
+                change_date = datetime.fromisoformat(row[1])
+                if self.archive_mod_time and change_date < self.archive_mod_time:
                     break
-                if newest_change is None or change_date > newest_change:
-                    newest_change = change_date
                 seen_files.add(changed_file)
-                year = changed_file.split("/")[0]
-                years.add(year)
+                years.add(changed_file.split("/")[0])
 
-        return seen_files, years, newest_change
+        return seen_files, years
 
-    def _download_changed_files(self, seen_files: set[str]) -> bool:
-        """Download the given advisory fragments in parallel, returning True if any failed."""
-        any_failed = False
+    def _download_changed_files(self, seen_files: set[str]) -> None:
+        """Download the given advisory fragments in parallel; failures are logged and retried next run."""
         with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
             futures = {
                 executor.submit(
@@ -286,6 +213,4 @@ def _download_changed_files(self, seen_files: set[str]) -> bool:
             concurrent.futures.wait(futures.keys())
             for future, changed_file in futures.items():
                 if future.exception() is not None:
-                    any_failed = True
                     self.logger.warning(f"failed to download {changed_file}: {future.exception()}")
-        return any_failed