refactor: process never unfurled version ranges first

- re-unfurl version ranges every 2 days
- run unfurl pipeline every 2 hours

Signed-off-by: Keshav Priyadarshi <git@keshav.space>

Author: keshav-space

vulnerabilities/migrations/0120_impactedpackage_last_range_unfurl_at_and_more.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.2.11 on 2026-04-06 20:51
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("vulnerabilities", "0119_remove_advisoryset_identifiers_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="impactedpackage",
+            name="last_range_unfurl_at",
+            field=models.DateTimeField(
+                blank=True,
+                db_index=True,
+                help_text="Timestamp of the last vers range unfurl.",
+                null=True,
+            ),
+        ),
+        migrations.AddField(
+            model_name="pipelineschedule",
+            name="run_priority",
+            field=models.IntegerField(
+                choices=[(1, "high"), (2, "default")],
+                default=2,
+                help_text="Select the pipeline execution priority",
+            ),
+        ),
+    ]

vulnerabilities/models.py

@@ -3252,6 +3252,13 @@ class ImpactedPackage(models.Model):
         help_text="Timestamp indicating when this impact was added.",
     )
 
+    last_range_unfurl_at = models.DateTimeField(
+        blank=True,
+        null=True,
+        db_index=True,
+        help_text="Timestamp of the last vers range unfurl.",
+    )
+
     def to_dict(self):
         from vulnerabilities.utils import purl_to_dict
 

vulnerabilities/pipelines/__init__.py

@@ -24,6 +24,7 @@
 from vulnerabilities.improver import MAX_CONFIDENCE
 from vulnerabilities.models import Advisory
 from vulnerabilities.models import PipelineRun
+from vulnerabilities.models import PipelineSchedule
 from vulnerabilities.pipes.advisory import import_advisory
 from vulnerabilities.pipes.advisory import insert_advisory
 from vulnerabilities.pipes.advisory import insert_advisory_v2
@@ -144,6 +145,9 @@ class VulnerableCodePipeline(PipelineDefinition, BasePipelineRun):
     # When set to true pipeline is run only once.
     # To rerun onetime pipeline reset is_active field to True via migration.
     run_once = False
+    # Interval between runs in hour.
+    run_interval = 24
+    run_priority = PipelineSchedule.ExecutionPriority.DEFAULT
 
     def on_failure(self):
         """
@@ -176,6 +180,9 @@ class VulnerableCodeBaseImporterPipeline(VulnerableCodePipeline):
     # When set to true pipeline is run only once.
     # To rerun onetime pipeline reset is_active field to True via migration.
     run_once = False
+    # Interval between runs in hour.
+    run_interval = 24
+    run_priority = PipelineSchedule.ExecutionPriority.DEFAULT
 
     @classmethod
     def steps(cls):
@@ -277,6 +284,9 @@ class VulnerableCodeBaseImporterPipelineV2(VulnerableCodePipeline):
     # When set to true pipeline is run only once.
     # To rerun onetime pipeline reset is_active field to True via migration.
     run_once = False
+    # Interval between runs in hour.
+    run_interval = 24
+    run_priority = PipelineSchedule.ExecutionPriority.DEFAULT
 
     @classmethod
     def steps(cls):

vulnerabilities/pipelines/v2_improvers/unfurl_version_range.py

@@ -8,9 +8,13 @@
 #
 
 import logging
+from datetime import timedelta
 from traceback import format_exc as traceback_format_exc
 
 from aboutcode.pipeline import LoopProgress
+from django.db.models import F
+from django.db.models import Q
+from django.utils import timezone
 from fetchcode.package_versions import SUPPORTED_ECOSYSTEMS as FETCHCODE_SUPPORTED_ECOSYSTEMS
 from packageurl import PackageURL
 from univers.version_range import RANGE_CLASS_BY_SCHEMES
@@ -19,29 +23,45 @@
 from vulnerabilities.models import ImpactedPackage
 from vulnerabilities.models import ImpactedPackageAffecting
 from vulnerabilities.models import PackageV2
+from vulnerabilities.models import PipelineSchedule
 from vulnerabilities.pipelines import VulnerableCodePipeline
 from vulnerabilities.pipes.fetchcode_utils import get_versions
 from vulnerabilities.utils import update_purl_version
 
 
 class UnfurlVersionRangePipeline(VulnerableCodePipeline):
+    """
+    Unfurl affected version ranges by first processing those that have
+    never been unfurled and then handling ranges that were last unfurled
+    two or more days ago.
+    """
 
     pipeline_id = "unfurl_version_range_v2"
 
+    run_interval = 2
+    run_priority = PipelineSchedule.ExecutionPriority.HIGH
+
+    # Days elapsed before version range is re-unfurled
+    reunfurl_after_days = 2
+
     @classmethod
     def steps(cls):
         return (cls.unfurl_version_range,)
 
     def unfurl_version_range(self):
-        impacted_packages = ImpactedPackage.objects.all().order_by("-created_at")
-        impacted_packages_count = impacted_packages.count()
-
         processed_impacted_packages_count = 0
         processed_affected_packages_count = 0
         cached_versions = {}
+        impacts_to_update = []
+        update_batch_size = 5
+
+        impacted_packages = impacted_package_qs(cutoff_day=self.reunfurl_after_days)
+        impacted_packages_count = impacted_packages.count()
         self.log(f"Unfurl affected vers range for {impacted_packages_count:,d} ImpactedPackage.")
+
         progress = LoopProgress(total_iterations=impacted_packages_count, logger=self.log)
-        for impact in progress.iter(impacted_packages):
+        for impact in progress.iter(impacted_packages.iterator(chunk_size=5000)):
+            impacts_to_update.append(impact.pk)
             purl = PackageURL.from_string(impact.base_purl)
             if not impact.affecting_vers or not any(
                 c in impact.affecting_vers for c in ("<", ">", "!")
@@ -52,11 +72,10 @@ def unfurl_version_range(self):
             if purl.type not in RANGE_CLASS_BY_SCHEMES:
                 continue
 
-            versions = get_purl_versions(purl, cached_versions) or []
+            versions = get_purl_versions(purl, cached_versions, self.log) or []
             affected_purls = get_affected_purls(
                 versions=versions,
-                affecting_vers=impact.affecting_vers,
-                base_purl=purl,
+                impact=impact,
                 logger=self.log,
             )
             if not affected_purls:
@@ -70,12 +89,21 @@ def unfurl_version_range(self):
             )
             processed_impacted_packages_count += 1
 
+            if len(impacts_to_update) > update_batch_size:
+                ImpactedPackage.objects.filter(pk__in=impacts_to_update).update(
+                    last_range_unfurl_at=timezone.now()
+                )
+                impacts_to_update.clear()
+
+        ImpactedPackage.objects.filter(pk__in=impacts_to_update).update(
+            last_range_unfurl_at=timezone.now()
+        )
         self.log(f"Successfully processed {processed_impacted_packages_count:,d} ImpactedPackage.")
         self.log(f"{processed_affected_packages_count:,d} new Impact-Package relation created.")
 
 
-def get_affected_purls(versions, affecting_vers, base_purl, logger):
-    affecting_version_range = VersionRange.from_string(affecting_vers)
+def get_affected_purls(versions, impact, logger):
+    affecting_version_range = VersionRange.from_string(impact.affecting_vers)
     version_class = affecting_version_range.version_class
 
     try:
@@ -84,7 +112,7 @@ def get_affected_purls(versions, affecting_vers, base_purl, logger):
         versions = [version_class(v) for v in versions]
     except Exception as e:
         logger(
-            f"Error while parsing versions for {base_purl!s}: {e!r} \n {traceback_format_exc()}",
+            f"Error while parsing versions for {impact.base_purl!s}: {e!r} \n {traceback_format_exc()}",
             level=logging.ERROR,
         )
         return
@@ -95,21 +123,24 @@ def get_affected_purls(versions, affecting_vers, base_purl, logger):
             if version in affecting_version_range:
                 affected_purls.append(
                     update_purl_version(
-                        purl=base_purl,
+                        purl=impact.base_purl,
                         version=str(version),
                     )
                 )
         except Exception as e:
             logger(
-                f"Error while checking {version!s} in {affecting_version_range!s}: {e!r} \n {traceback_format_exc()}",
+                (
+                    f"Error while checking {version!s} in {affecting_version_range!s} for "
+                    f"advisory {impact.advisory.avid}: {e!r} \n {traceback_format_exc()}"
+                ),
                 level=logging.ERROR,
             )
     return affected_purls
 
 
-def get_purl_versions(purl, cached_versions):
+def get_purl_versions(purl, cached_versions, logger):
     if not purl in cached_versions:
-        purls = get_versions(purl)
+        purls = get_versions(purl, logger)
         if purls is not None:
             cached_versions[purl] = purls
     return cached_versions.get(purl) or []
@@ -135,3 +166,16 @@ def bulk_create_with_m2m(purls, impact, relation, logger):
         return 0
 
     return len(relations)
+
+
+def impacted_package_qs(cutoff_day=2):
+    cutoff = timezone.now() - timedelta(days=cutoff_day)
+    return (
+        ImpactedPackage.objects.filter(
+            (Q(last_range_unfurl_at__isnull=True) | Q(last_range_unfurl_at__lte=cutoff))
+            & Q(affecting_vers__isnull=False)
+            & ~Q(affecting_vers="")
+        )
+        .order_by(F("last_range_unfurl_at").asc(nulls_first=True))
+        .only("pk", "affecting_vers", "advisory", "base_purl")
+    )