Sign packages concurrently when adding to repo

daviddavis · daviddavis · commit 771677f901f6 · 2026-02-18T15:03:28.000Z
Assisted By: Claude Sonnet 4.5
diff --git a/pulp_deb/app/models/signing_service.py b/pulp_deb/app/models/signing_service.py
@@ -198,6 +198,26 @@ def sign(
         _env_vars["PULP_SIGNING_KEY_FINGERPRINT"] = pubkey_fingerprint
         return super().sign(filename, _env_vars)
 
+    async def asign(
+        self,
+        filename: str,
+        env_vars: Optional[dict] = None,
+        pubkey_fingerprint: Optional[str] = None,
+    ):
+        """
+        Asynchronously sign a package @filename using @pubkey_fingerprint.
+
+        Args:
+            filename: The absolute path to the package to be signed.
+            env_vars: (optional) Dict of env_vars to be passed to the signing script.
+            pubkey_fingerprint: The V4 fingerprint that correlates with the private key to use.
+        """
+        if not pubkey_fingerprint:
+            raise ValueError("A pubkey_fingerprint must be provided.")
+        _env_vars = env_vars or {}
+        _env_vars["PULP_SIGNING_KEY_FINGERPRINT"] = pubkey_fingerprint
+        return await super().asign(filename, _env_vars)
+
     def validate(self):
         """
         Validate a signing service for an Apt package signature.
diff --git a/pulp_deb/app/settings.py b/pulp_deb/app/settings.py
@@ -9,3 +9,4 @@
 FORBIDDEN_CHECKSUM_WARNINGS = True
 FORCE_IGNORE_MISSING_PACKAGE_INDICES = False
 PERMISSIVE_SYNC = False
+MAX_PACKAGE_SIGNING_WORKERS = 5
diff --git a/pulp_deb/app/tasks/signing.py b/pulp_deb/app/tasks/signing.py
@@ -1,6 +1,8 @@
+import asyncio
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 
+from django.conf import settings
 from django.db.models import Q
 
 from pulpcore.plugin.models import (
@@ -43,12 +45,7 @@ def _save_upload(uploadobj, final_package):
     final_package.flush()
 
 
-def _sign_file(package_file, signing_service, signing_fingerprint):
-    logging.info(
-        _("Signing package %s with fingerprint %s"), package_file.name, signing_fingerprint
-    )
-    result = signing_service.sign(package_file.name, pubkey_fingerprint=signing_fingerprint)
-    signed_package_path = Path(result["deb_package"])
+def _create_signed_artifact(signed_package_path, result):
     if not signed_package_path.exists():
         raise Exception(f"Signing script did not create the signed package: {result}")
     artifact = Artifact.init_and_validate(str(signed_package_path))
@@ -58,6 +55,15 @@ def _sign_file(package_file, signing_service, signing_fingerprint):
     return artifact
 
 
+async def _sign_file(package_file, signing_service, signing_fingerprint):
+    logging.info(
+        _("Signing package %s with fingerprint %s"), package_file.name, signing_fingerprint
+    )
+    result = await signing_service.asign(package_file.name, pubkey_fingerprint=signing_fingerprint)
+    signed_package_path = Path(result["deb_package"])
+    return await asyncio.to_thread(_create_signed_artifact, signed_package_path, result)
+
+
 def sign_and_create(
     app_label,
     serializer_name,
@@ -79,7 +85,9 @@ def sign_and_create(
             uploaded_package = Upload.objects.get(pk=temporary_file_pk)
             _save_upload(uploaded_package, final_package)
 
-        artifact = _sign_file(final_package, package_signing_service, signing_fingerprint)
+        artifact = asyncio.run(
+            _sign_file(final_package, package_signing_service, signing_fingerprint)
+        )
     uploaded_package.delete()
     # Create Package content
     data["artifact"] = get_url(artifact)
@@ -93,35 +101,66 @@ def sign_and_create(
     general_create(app_label, serializer_name, data=data, context=context, *args, **kwargs)
 
 
-def _update_content_units(content_units, old_pk, new_pk):
-    while str(old_pk) in content_units:
-        content_units.remove(str(old_pk))
+def _sign_package(package, signing_service, signing_fingerprint, package_release_map):
+    """
+    Sign a package or reuse an existing signed result.
 
-    if str(new_pk) not in content_units:
-        content_units.append(str(new_pk))
+    Returns None if already signed with the fingerprint, otherwise a
+    tuple of (original_package_id, new_package_id, prcs_to_update).
+    """
+    content_artifact = package.contentartifact_set.first()
+    artifact_obj = content_artifact.artifact
+    package_id = str(package.pk)
 
-    # Repoint PackageReleaseComponents included in this transaction to the new package.
-    for prc in PackageReleaseComponent.objects.filter(pk__in=content_units, package_id=old_pk):
-        new_prc, _ = PackageReleaseComponent.objects.get_or_create(
-            release_component=prc.release_component,
-            package_id=new_pk,
-            _pulp_domain=prc._pulp_domain,
-        )
-
-        while str(prc.pk) in content_units:
-            content_units.remove(str(prc.pk))
+    with NamedTemporaryFile(mode="wb", dir=".", delete=False) as final_package:
+        artifact_file = artifact_obj.file
+        _save_file(artifact_file, final_package)
 
-        if str(new_prc.pk) not in content_units:
-            content_units.append(str(new_prc.pk))
+        # check if the package is already signed with our fingerprint
+        try:
+            signing_service.validate_signature(final_package.name)
+            return None
+        except (UnsignedPackage, InvalidSignature, FingerprintMismatch):
+            pass
+
+        # Collect PackageReleaseComponents that need to be updated
+        prcs_to_update = list(
+            PackageReleaseComponent.objects.filter(
+                package_id=package_id, _pulp_domain=package._pulp_domain
+            )
+        )
 
+        # check if the package has been signed in the past with our fingerprint
+        if existing_result := DebPackageSigningResult.objects.filter(
+            sha256=content_artifact.artifact.sha256,
+            package_signing_fingerprint=signing_fingerprint,
+        ).first():
+            return (package_id, str(existing_result.result.pk), prcs_to_update)
+
+        # create a new signed version of the package
+        log.info(f"Signing package {package.name}.")
+        artifact = asyncio.run(_sign_file(final_package, signing_service, signing_fingerprint))
+        signed_package = package
+        signed_package.pk = None
+        signed_package.pulp_id = None
+        signed_package.sha256 = artifact.sha256
+        signed_package.save()
+        ContentArtifact.objects.create(
+            artifact=artifact,
+            content=signed_package,
+            relative_path=content_artifact.relative_path,
+        )
+        DebPackageSigningResult.objects.create(
+            sha256=artifact_obj.sha256,
+            package_signing_fingerprint=signing_fingerprint,
+            result=signed_package,
+        )
 
-def _check_package_signature(repository, package_path):
-    try:
-        repository.package_signing_service.validate_signature(package_path)
-    except (UnsignedPackage, InvalidSignature, FingerprintMismatch):
-        return False
+        resource = CreatedResource(content_object=signed_package)
+        resource.save()
+        log.info(f"Signed package {package.name}.")
 
-    return True
+        return (package_id, str(signed_package.pk), prcs_to_update)
 
 
 def signed_add_and_remove(
@@ -136,53 +175,54 @@ def signed_add_and_remove(
         ).select_related("package", "release_component")
         package_release_map = {prc.package_id: prc.release_component.distribution for prc in prcs}
 
-        # sign each package and replace it in the add_content_units list
+        # Prepare package list with their fingerprints
+        packages = []
         for package in Package.objects.filter(pk__in=add_content_units):
-            content_artifact = package.contentartifact_set.first()
-            artifact_obj = content_artifact.artifact
-            package_id = package.pk
-
             # match the package's release to a fingerprint override if one exists
             fingerprint = repo.release_package_signing_fingerprint(
-                package_release_map.get(package_id)
+                package_release_map.get(package.pk)
             )
-
-            with NamedTemporaryFile(mode="wb", dir=".", delete=False) as final_package:
-                artifact_file = artifact_obj.file
-                _save_file(artifact_file, final_package)
-
-                # check if the package is already signed with our fingerprint
-                if _check_package_signature(repo, final_package.name):
-                    continue
-
-                # check if the package has been signed in the past with our fingerprint
-                if existing_result := DebPackageSigningResult.objects.filter(
-                    sha256=content_artifact.artifact.sha256,
-                    package_signing_fingerprint=fingerprint,
-                ).first():
-                    _update_content_units(add_content_units, package_id, existing_result.result.pk)
-                    continue
-
-                # create a new signed version of the package
-                artifact = _sign_file(final_package, repo.package_signing_service, fingerprint)
-                signed_package = package
-                signed_package.pk = None
-                signed_package.pulp_id = None
-                signed_package.sha256 = artifact.sha256
-                signed_package.save()
-                ContentArtifact.objects.create(
-                    artifact=artifact,
-                    content=signed_package,
-                    relative_path=content_artifact.relative_path,
-                )
-                DebPackageSigningResult.objects.create(
-                    sha256=artifact_obj.sha256,
-                    package_signing_fingerprint=fingerprint,
-                    result=signed_package,
+            packages.append((package, fingerprint))
+
+        async def _sign_packages():
+            semaphore = asyncio.Semaphore(settings.MAX_PACKAGE_SIGNING_WORKERS)
+
+            async def _bounded_sign(pkg_tuple):
+                pkg, fingerprint = pkg_tuple
+                async with semaphore:
+                    return await asyncio.to_thread(
+                        _sign_package,
+                        pkg,
+                        repo.package_signing_service,
+                        fingerprint,
+                        package_release_map,
+                    )
+
+            return await asyncio.gather(*(_bounded_sign(pkg_tuple) for pkg_tuple in packages))
+
+        for result in asyncio.run(_sign_packages()):
+            if not result:
+                continue
+            old_id, new_id, prcs_to_update = result
+
+            # Update the add_content_units list with the new package
+            while old_id in add_content_units:
+                add_content_units.remove(old_id)
+            if new_id not in add_content_units:
+                add_content_units.append(new_id)
+
+            # Repoint PackageReleaseComponents that were collected during signing
+            for prc in prcs_to_update:
+                new_prc, _ = PackageReleaseComponent.objects.get_or_create(
+                    release_component=prc.release_component,
+                    package_id=new_id,
+                    _pulp_domain=prc._pulp_domain,
                 )
 
-                _update_content_units(add_content_units, package_id, signed_package.pk)
-                resource = CreatedResource(content_object=signed_package)
-                resource.save()
+                while str(prc.pk) in add_content_units:
+                    add_content_units.remove(str(prc.pk))
+
+                if str(new_prc.pk) not in add_content_units:
+                    add_content_units.append(str(new_prc.pk))
 
     return add_and_remove(repository_pk, add_content_units, remove_content_units, base_version_pk)
