Sign packages concurrently when adding to repo

Assisted by: GPT-5.2-Codex

fixes #4357

Author: daviddavis

CHANGES/4357.feature

@@ -0,0 +1,3 @@
+Added code and a setting (MAX_PACKAGE_SIGNING_WORKERS) to sign packages concurrently when adding
+them to a repo. Signing packages concurrently will speed up the time it takes to sign and add
+packages to a repo.

docs/admin/reference/settings.md

@@ -64,3 +64,8 @@ When publishing RPM metadata, if this is true, Pulp will use the timestamp that
 added to the repo rather than the timestamp that the package first appeared in Pulp. This timestamp
 appears in the "file" field of the time element for each package in primary.xml. Defaults to
 `False`.
+
+
+## MAX_PACKAGE_SIGNING_WORKERS
+
+Sets the number of workers that pulp_rpm uses when concurrently signing packages. Defaults to 5.

pulp_rpm/app/models/content.py

@@ -45,6 +45,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_figerprint.
+
+        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 a Rpm Package signature.

pulp_rpm/app/settings.py

@@ -19,3 +19,4 @@
 PRUNE_WORKERS_MAX = 5
 # workaround for: https://github.com/pulp/pulp_rpm/issues/4125
 SPECTACULAR_SETTINGS__OAS_VERSION = "3.0.1"
+MAX_PACKAGE_SIGNING_WORKERS = 5

pulp_rpm/app/tasks/signing.py

@@ -1,9 +1,12 @@
+import asyncio
 import logging
 import re
 import subprocess
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 
+from django.conf import settings
+
 from pulpcore.plugin.models import (
     Artifact,
     ContentArtifact,
@@ -65,9 +68,7 @@ def _verify_package_fingerprint(package_file, signing_fingerprint):
     return False
 
 
-def _sign_file(package_file, signing_service, signing_fingerprint):
-    result = signing_service.sign(package_file.name, pubkey_fingerprint=signing_fingerprint)
-    signed_package_path = Path(result["rpm_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))
@@ -77,6 +78,73 @@ def _sign_file(package_file, signing_service, signing_fingerprint):
     return artifact
 
 
+def _sign_file(package_file, signing_service, signing_fingerprint):
+    result = signing_service.sign(package_file.name, pubkey_fingerprint=signing_fingerprint)
+    signed_package_path = Path(result["rpm_package"])
+    return _create_signed_artifact(signed_package_path, result)
+
+
+async def _asign_file(package_file, signing_service, signing_fingerprint):
+    result = await signing_service.asign(package_file.name, pubkey_fingerprint=signing_fingerprint)
+    signed_package_path = Path(result["rpm_package"])
+    return await asyncio.to_thread(_create_signed_artifact, signed_package_path, result)
+
+
+def _sign_package(package, signing_service, signing_fingerprint):
+    """
+    Sign a package or reuse an existing signed result.
+
+    Returns None if already signed with the fingerprint, otherwise a
+    tuple of (original_package_id, new_package_id).
+    """
+    # the viewset is currently already checking (and rejecting) on demand content
+    # but in the future we could just download it instead
+    content_artifact = package.contentartifact_set.first()
+    artifact_obj = content_artifact.artifact
+    package_id = str(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 _verify_package_fingerprint(final_package, signing_fingerprint):
+            return None
+
+        # check if the package has been signed in the past with our fingerprint and replace
+        # it with the previously-created signed package if so
+        if existing_result := RpmPackageSigningResult.objects.filter(
+            original_package_sha256=content_artifact.artifact.sha256,
+            package_signing_fingerprint=signing_fingerprint,
+        ).first():
+            return (package_id, str(existing_result.result_package.pk))
+
+        # create a new signed version of the package
+        log.info(f"Signing package {package.filename}.")
+        artifact = _sign_file(final_package, signing_service, signing_fingerprint)
+        signed_package = package
+        signed_package.pk = None
+        signed_package.pulp_id = None
+        signed_package.pkgId = artifact.sha256
+        signed_package.checksum_type = CHECKSUM_TYPES.SHA256
+        signed_package.save()
+        ContentArtifact.objects.create(
+            artifact=artifact,
+            content=signed_package,
+            relative_path=content_artifact.relative_path,
+        )
+        RpmPackageSigningResult.objects.create(
+            original_package_sha256=artifact_obj.sha256,
+            package_signing_fingerprint=signing_fingerprint,
+            result_package=signed_package,
+        )
+
+        resource = CreatedResource(content_object=signed_package)
+        resource.save()
+        log.info(f"Signed package {package.filename}.")
+        return (package_id, str(signed_package.pk))
+
+
 def sign_and_create(
     app_label,
     serializer_name,
@@ -120,58 +188,30 @@ def signed_add_and_remove(
     repo = RpmRepository.objects.get(pk=repository_pk)
 
     if repo.package_signing_service:
-        # sign each package and replace it in the add_content_units list
-        for package in Package.objects.filter(pk__in=add_content_units).iterator():
-            # the viewset is currently already checking (and rejecting) on demand content
-            # but in the future we could just download it instead
-            content_artifact = package.contentartifact_set.first()
-            artifact_obj = content_artifact.artifact
-            package_id = str(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 _verify_package_fingerprint(final_package, repo.package_signing_fingerprint):
-                    continue
-
-                # check if the package has been signed in the past with our fingerprint and replace
-                # it with the previously-created signed package if so
-                if existing_result := RpmPackageSigningResult.objects.filter(
-                    original_package_sha256=content_artifact.artifact.sha256,
-                    package_signing_fingerprint=repo.package_signing_fingerprint,
-                ).first():
-                    while package_id in add_content_units:
-                        add_content_units.remove(package_id)
-                    add_content_units.append(str(existing_result.result_package.pk))
-                    continue
-
-                # create a new signed version of the package
-                artifact = _sign_file(
-                    final_package, repo.package_signing_service, repo.package_signing_fingerprint
-                )
-                signed_package = package
-                signed_package.pk = None
-                signed_package.pulp_id = None
-                signed_package.pkgId = artifact.sha256
-                signed_package.checksum_type = CHECKSUM_TYPES.SHA256
-                signed_package.save()
-                ContentArtifact.objects.create(
-                    artifact=artifact,
-                    content=signed_package,
-                    relative_path=content_artifact.relative_path,
-                )
-                RpmPackageSigningResult.objects.create(
-                    original_package_sha256=artifact_obj.sha256,
-                    package_signing_fingerprint=repo.package_signing_fingerprint,
-                    result_package=signed_package,
-                )
-
-                resource = CreatedResource(content_object=signed_package)
-                resource.save()
-                while package_id in add_content_units:
-                    add_content_units.remove(package_id)
-                add_content_units.append(str(signed_package.pk))
+        add_content_units = set(add_content_units)
+        packages = list(Package.objects.filter(pk__in=add_content_units).all())
+
+        async def _sign_packages():
+            semaphore = asyncio.Semaphore(settings.MAX_PACKAGE_SIGNING_WORKERS)
+
+            async def _bounded_sign(pkg):
+                async with semaphore:
+                    return await asyncio.to_thread(
+                        _sign_package,
+                        pkg,
+                        repo.package_signing_service,
+                        repo.package_signing_fingerprint,
+                    )
+
+            return await asyncio.gather(*(_bounded_sign(pkg) for pkg in packages))
+
+        for result in asyncio.run(_sign_packages()):
+            if not result:
+                continue
+            old_id, new_id = result
+            add_content_units.discard(old_id)
+            add_content_units.add(new_id)
+
+        add_content_units = list(add_content_units)
 
     return add_and_remove(repository_pk, add_content_units, remove_content_units, base_version_pk)