From ad90bd017bc862c3cafadb3d7a3e318267874783 Mon Sep 17 00:00:00 2001 From: Danilo Lakovic Date: Tue, 27 Oct 2026 23:37:47 +0100 Subject: [PATCH 1/3] Let concurrent deploys to one repository wait for their turn Each uploaded file is added by a task that holds the repository exclusively. The task was dispatched as immediate only, so an upload arriving while another one held the repository was canceled and answered with 429. Maven clients do not retry that, so two builds publishing into the same repository at once failed. The task may now wait in the queue for a worker, and the request waits for it for up to MAVEN_DEPLOY_TASK_TIMEOUT seconds (30 by default) before answering 429, as the container registry API does for its uploads. --- CHANGES/+concurrent-deploys.bugfix | 3 ++ pulp_maven/app/maven_deploy_api.py | 28 ++++++++-- pulp_maven/app/settings.py | 4 ++ .../functional/api/test_concurrent_deploy.py | 54 +++++++++++++++++++ 4 files changed, 84 insertions(+), 5 deletions(-) create mode 100644 CHANGES/+concurrent-deploys.bugfix create mode 100644 pulp_maven/tests/functional/api/test_concurrent_deploy.py diff --git a/CHANGES/+concurrent-deploys.bugfix b/CHANGES/+concurrent-deploys.bugfix new file mode 100644 index 00000000..0cf2cb78 --- /dev/null +++ b/CHANGES/+concurrent-deploys.bugfix @@ -0,0 +1,3 @@ +Concurrent deploys to the same repository no longer fail with HTTP 429. An upload that arrives while +another one holds the repository waits for its turn, for up to `MAVEN_DEPLOY_TASK_TIMEOUT` seconds +(30 by default). diff --git a/pulp_maven/app/maven_deploy_api.py b/pulp_maven/app/maven_deploy_api.py index a2cf5569..2521b7b4 100644 --- a/pulp_maven/app/maven_deploy_api.py +++ b/pulp_maven/app/maven_deploy_api.py @@ -1,5 +1,6 @@ import hashlib import re +import time from tempfile import NamedTemporaryFile from django.conf import settings @@ -11,6 +12,7 @@ from rest_framework.response import Response from rest_framework.viewsets import ViewSet +from pulpcore.plugin.constants import TASK_FINAL_STATES, TASK_STATES from pulpcore.plugin.models import Artifact, ContentArtifact from pulpcore.plugin.tasking import dispatch from pulpcore.plugin.util import get_domain @@ -27,20 +29,36 @@ def has_task_completed(task): """ - Verify whether an immediate task ran properly. + Wait for a dispatched task to finish and report how it ended. + + The task runs right away when no other task holds the repository. Otherwise it waits in the + queue for a worker, and the upload waits with it for up to MAVEN_DEPLOY_TASK_TIMEOUT + seconds, so concurrent deploys to one repository take turns instead of being refused. Returns: bool: True if the task ended successfully. Raises: Exception: If an error occured during the task's runtime. - Throttled: If the task did not run due to resource constraints. + Throttled: If the task was canceled or did not finish within the timeout. """ - if task.state == "completed": + timeout = settings.MAVEN_DEPLOY_TASK_TIMEOUT + deadline = time.monotonic() + timeout + delay = 0.05 + while task.state not in TASK_FINAL_STATES: + remaining = deadline - time.monotonic() + if remaining <= 0: + # The task stays queued and still adds the content; a retried upload of the same + # file replaces it. + raise Throttled(wait=timeout) + time.sleep(min(delay, remaining)) + delay = min(delay * 2, 1) + task.refresh_from_db() + if task.state == TASK_STATES.COMPLETED: task.delete() return True - elif task.state == "canceled": + elif task.state == TASK_STATES.CANCELED: raise Throttled() else: error = task.error @@ -236,7 +254,7 @@ def create(self, request, name, path): aadd_and_remove, exclusive_resources=[repo], immediate=True, - deferred=False, + deferred=True, kwargs={ "repository_pk": str(repo.pk), "add_content_units": add_content_units, diff --git a/pulp_maven/app/settings.py b/pulp_maven/app/settings.py index 952e91b8..2368939a 100644 --- a/pulp_maven/app/settings.py +++ b/pulp_maven/app/settings.py @@ -8,3 +8,7 @@ MAVEN_PATH_INDEX_REFRESH_SECONDS = 2 # Match a hosted deployment's existing large-file redirect policy when configured. MAVEN_PATH_INDEX_REDIRECT_THRESHOLD = None + +# How long a deploy request waits, in seconds, for its turn to add the uploaded file to the +# repository while other deploys to the same repository go first. After that it answers 429. +MAVEN_DEPLOY_TASK_TIMEOUT = 30 diff --git a/pulp_maven/tests/functional/api/test_concurrent_deploy.py b/pulp_maven/tests/functional/api/test_concurrent_deploy.py new file mode 100644 index 00000000..c433f726 --- /dev/null +++ b/pulp_maven/tests/functional/api/test_concurrent_deploy.py @@ -0,0 +1,54 @@ +"""Concurrent deploys into one repository.""" + +import uuid +from concurrent.futures import ThreadPoolExecutor +from urllib.parse import urlsplit + +import requests + +UPLOADS = 8 + + +def _deploy_url(bindings_cfg, pulp_settings, repo_name, path): + parts = urlsplit(bindings_cfg.host) + prefix = "default/" if pulp_settings.DOMAIN_ENABLED else "" + return f"{parts.scheme}://{parts.netloc}/pulp/maven/{prefix}{repo_name}/{path}" + + +def test_concurrent_deploys_to_one_repository_all_succeed( + maven_repo_api_client, + maven_artifact_api_client, + maven_repo_factory, + maven_distribution_factory, + pulp_settings, + bindings_cfg, +): + """Uploads that arrive while another one holds the repository wait for their turn. + + Builds that publish into one repository at the same time, such as two CI pipelines or + a build that deploys its modules in parallel, overlap their uploads, and Maven clients + do not retry a refused upload. Refusing every upload that overlaps another one would + fail those builds outright. + """ + repo = maven_repo_factory() + maven_distribution_factory(repository=repo.pulp_href) + artifact_id = f"concurrent-{uuid.uuid4()}" + paths = [f"com/example/{artifact_id}/1.0.{i}/{artifact_id}-1.0.{i}.jar" for i in range(UPLOADS)] + + def deploy(path): + return requests.put( + _deploy_url(bindings_cfg, pulp_settings, repo.name, path), + data=f"jar {path}".encode(), + auth=(bindings_cfg.username, bindings_cfg.password), + verify=False, + ).status_code + + with ThreadPoolExecutor(max_workers=UPLOADS) as pool: + statuses = list(pool.map(deploy, paths)) + + assert statuses == [201] * UPLOADS + repo = maven_repo_api_client.read(repo.pulp_href) + deployed = maven_artifact_api_client.list( + repository_version=repo.latest_version_href, artifact_id=artifact_id + ) + assert deployed.count == UPLOADS From 1a32362478c6c199529f57072b008af6ef40ee12 Mon Sep 17 00:00:00 2001 From: Danilo Lakovic Date: Wed, 28 Oct 2026 14:26:29 +0100 Subject: [PATCH 2/3] Make finalizing a repository version cheaper Every repository version regenerates maven-metadata.xml, prefixes.txt and the directory index pages of what it changed. Most of that cost did not depend on the size of the change: - Packages and metadata were added and removed in up to seven add_content / remove_content calls, and each rewrites the version's whole list of content ids. They are now collected and applied in one add and one remove. - Deciding whether prefixes.txt changed scanned the group ids of every artifact in the repository, twice. Only the affected prefixes are looked up now. - The index pages of the affected directories were found through the version's whole content list; they are now checked for membership one by one. - Metadata files and index pages were written to storage one after another. They are now written in parallel, by a pool of threads that lives as long as the process so its storage connections stay open. A forked child, where pulpcore runs most tasks, starts its own pool, since it inherits the pool but not its threads. The resulting repository content is unchanged. --- CHANGES/+cheaper-finalize.feature | 3 + pulp_maven/app/models.py | 129 ++++++++++++------- pulp_maven/app/tasks/__init__.py | 200 ++++++++++++++++++------------ 3 files changed, 209 insertions(+), 123 deletions(-) create mode 100644 CHANGES/+cheaper-finalize.feature diff --git a/CHANGES/+cheaper-finalize.feature b/CHANGES/+cheaper-finalize.feature new file mode 100644 index 00000000..c75156f2 --- /dev/null +++ b/CHANGES/+cheaper-finalize.feature @@ -0,0 +1,3 @@ +Finalizing a repository version is cheaper: packages and metadata are added in one step instead of +rewriting the version's content ids for each, `prefixes.txt` no longer scans every artifact, index +pages are looked up by membership, and generated files are written to storage in parallel. diff --git a/pulp_maven/app/models.py b/pulp_maven/app/models.py index 36ae85c5..3dfed8fa 100644 --- a/pulp_maven/app/models.py +++ b/pulp_maven/app/models.py @@ -567,6 +567,39 @@ class Meta: ] +class _PendingContentChanges: + """Content to add to and remove from a repository version, applied in one step. + + pulpcore rewrites a version's whole list of content ids on every ``add_content`` and + ``remove_content`` call, so each call costs time in proportion to the size of the + repository. Steps that do not read each other's changes collect them here and apply + them together. + """ + + def __init__(self): + self.to_add = set() + self.to_remove = set() + + def apply(self, new_version): + # Something both removed and added stays in the version, as it would if the removal + # and the addition were applied in that order. + to_remove = self.to_remove - self.to_add + if to_remove: + new_version.remove_content(Content.objects.filter(pk__in=to_remove)) + if self.to_add: + new_version.add_content(Content.objects.filter(pk__in=self.to_add)) + self.to_add.clear() + self.to_remove.clear() + + +def _group_ids_with_prefix(prefix): + """Match the group ids that ``_compute_prefix`` maps to ``prefix`` (``/a/b`` or ``/a``).""" + group_id = prefix.lstrip("/").replace("/", ".") + if "." in group_id: + return Q(group_id=group_id) | Q(group_id__startswith=f"{group_id}.") + return Q(group_id=group_id) + + class MavenRepository(Repository, AutoAddObjPermsMixin): """ Repository for "maven" content. @@ -635,8 +668,13 @@ def finalize_new_version(self, new_version): """Remove duplicates, ensure packages, and generate metadata and index pages.""" remove_duplicates(new_version) if not getattr(_pull_through_ctx, "active", False): - self._ensure_packages(new_version) - self._generate_metadata(new_version) + # Packages and metadata only read artifacts, which they do not change, so their + # changes are applied together. The index pages list the metadata files, so they + # are generated after that. + changes = _PendingContentChanges() + self._ensure_packages(new_version, changes) + self._generate_metadata(new_version, changes) + changes.apply(new_version) self._generate_index_pages(new_version) self._generate_bloom_filter(new_version) @@ -660,8 +698,11 @@ def delete_redis_bloom_filter(self): """Remove the repository's Bloom filter from Redis.""" delete_bloom_filter(self) - def _ensure_packages(self, new_version): - """Manage MavenPackage version membership. Creates missing packages when a POM is available.""" + def _ensure_packages(self, new_version, changes): + """Manage MavenPackage version membership. Creates missing packages when a POM is available. + + The packages to add and remove are collected in ``changes`` (a _PendingContentChanges). + """ from django.db.models import Q from pulpcore.plugin.models import ContentArtifact @@ -749,8 +790,7 @@ def _ensure_packages(self, new_version): pkg.save() package_pks_to_add.append(pkg.pk) - if package_pks_to_add: - new_version.add_content(MavenPackage.objects.filter(pk__in=package_pks_to_add)) + changes.to_add.update(package_pks_to_add) dead_gavs = affected_gavs - live_gavs if dead_gavs: @@ -760,7 +800,7 @@ def _ensure_packages(self, new_version): dead_pkgs = MavenPackage.objects.filter( pk__in=new_version.content, _pulp_domain=self.pulp_domain ).filter(dead_q) - new_version.remove_content(dead_pkgs) + changes.to_remove.update(dead_pkgs.values_list("pk", flat=True)) def _ensure_packages_full_scan(self, new_version): """Reconcile MavenPackage membership across the whole version. @@ -890,8 +930,11 @@ def _parse(ca): skipped_unreadable, ) - def _generate_metadata(self, new_version): - """Generate maven-metadata.xml and checksums for affected (group_id, artifact_id) pairs.""" + def _generate_metadata(self, new_version, changes): + """Generate maven-metadata.xml and checksums for affected (group_id, artifact_id) pairs. + + The metadata to add and remove is collected in ``changes`` (a _PendingContentChanges). + """ from collections import defaultdict from django.db.models import Q @@ -934,8 +977,7 @@ def _generate_metadata(self, new_version): .filter(pairs_q) .values_list("pk", flat=True) ) - if stale_pks: - new_version.remove_content(MavenMetadata.objects.filter(pk__in=stale_pks)) + changes.to_remove.update(stale_pks) versions_by_pair = defaultdict(set) for row in ( @@ -969,8 +1011,7 @@ def _generate_metadata(self, new_version): .filter(triples_q) .values_list("pk", flat=True) ) - if stale_version_pks: - new_version.remove_content(MavenMetadata.objects.filter(pk__in=stale_version_pks)) + changes.to_remove.update(stale_version_pks) for group_id, artifact_id, version in affected_snapshot_triples: filenames = list( @@ -989,8 +1030,7 @@ def _generate_metadata(self, new_version): ) ) - if new_metadata_pks: - new_version.add_content(MavenMetadata.objects.filter(pk__in=new_metadata_pks)) + changes.to_add.update(new_metadata_pks) # --- prefixes.txt generation --- # Compute prefixes from the affected group_ids. If the set of @@ -1000,30 +1040,16 @@ def _generate_metadata(self, new_version): # Changed artifact prefixes (from removed or added artifacts) affected_prefixes = {_compute_prefix(gid) for gid, _ in affected_pairs} - # Get all group_ids currently in the version (across ALL artifacts, - # not just the affected ones) so we can check whether the affected - # prefixes are truly new or truly gone. - all_group_ids = set( - MavenArtifact.objects.filter(pk__in=new_version.content) - .values_list("group_id", flat=True) - .distinct() - ) - # All prefixes in the repo now - current_prefixes = {_compute_prefix(gid) for gid in all_group_ids} - - unaffected_group_ids = set( - MavenArtifact.objects.filter(pk__in=new_version.content) - .exclude(pairs_q) - .values_list("group_id", flat=True) - .distinct() + # The prefix set changed when an affected prefix appeared or disappeared, that is, + # when no artifact outside the affected pairs carries it. (A prefix that no artifact + # carries any more is not carried by those either.) Only the affected prefixes are + # looked up, so the cost does not grow with the size of the repository. + unaffected_artifacts = MavenArtifact.objects.filter(pk__in=new_version.content).exclude( + pairs_q ) - - unaffected_prefixes = {_compute_prefix(gid) for gid in unaffected_group_ids} - - # affected prefixes not in unaffected prefix list? -> prefixes added - # affected prefixes not in current prefix list? -> prefixes removed - prefix_set_changed = bool(affected_prefixes - unaffected_prefixes) or bool( - affected_prefixes - current_prefixes + prefix_set_changed = any( + not unaffected_artifacts.filter(_group_ids_with_prefix(prefix)).exists() + for prefix in affected_prefixes ) # Also check if this repository version has a prefixes.txt file. @@ -1038,12 +1064,16 @@ def _generate_metadata(self, new_version): ) if prefix_set_changed or not old_prefixes_pks: # Remove old prefixes.txt from the version - if old_prefixes_pks: - new_version.remove_content(MavenMetadata.objects.filter(pk__in=old_prefixes_pks)) + changes.to_remove.update(old_prefixes_pks) + current_prefixes = { + _compute_prefix(group_id) + for group_id in MavenArtifact.objects.filter(pk__in=new_version.content) + .values_list("group_id", flat=True) + .distinct() + } if current_prefixes: - prefixes_pks = _save_prefixes_txt(current_prefixes, self.pulp_domain) - new_version.add_content(MavenMetadata.objects.filter(pk__in=prefixes_pks)) + changes.to_add.update(_save_prefixes_txt(current_prefixes, self.pulp_domain)) def _generate_index_pages(self, new_version, affected_paths=None): """Generate HTML index pages for directory paths touched by this version. @@ -1124,11 +1154,16 @@ def ancestor_directory_paths(directory_path): # Pre-fetch all existing index pages so we can remove stale ones without # issuing one EXISTS query per directory. - existing_indexes_qs = MavenIndexPage.objects.filter(pk__in=new_version.content).only( - "path", "pk", "sha256", "pulp_type" - ) - if not bulk_mode: - existing_indexes_qs = existing_indexes_qs.filter(path__in=affected_paths) + if bulk_mode: + existing_indexes_qs = MavenIndexPage.objects.filter(pk__in=new_version.content) + else: + # Pages of the affected directories, each checked for membership in the version, + # instead of expanding the version's whole content list. + existing_indexes_qs = MavenIndexPage.objects.filter( + Exists(new_version._content_relationships().filter(content_id=OuterRef("pk"))), + path__in=affected_paths, + ) + existing_indexes_qs = existing_indexes_qs.only("path", "pk", "sha256", "pulp_type") existing_indexes = { index_page.path: index_page for index_page in existing_indexes_qs.iterator() } diff --git a/pulp_maven/app/tasks/__init__.py b/pulp_maven/app/tasks/__init__.py index 09bf3b4c..4b3f4904 100644 --- a/pulp_maven/app/tasks/__init__.py +++ b/pulp_maven/app/tasks/__init__.py @@ -1,13 +1,16 @@ import datetime import hashlib import logging +import os import tempfile +import threading import time from collections import defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed from xml.etree.ElementTree import Element, SubElement, tostring from asgiref.sync import sync_to_async +from django.core.files import File from django.db import IntegrityError, transaction from django.db.models import Q from redis.exceptions import RedisError @@ -133,36 +136,67 @@ def _save_artifact(content_bytes, pulp_domain): return artifact -def _save_artifacts_batch(pages, pulp_domain): - """Save a collection of (dir_path, html_bytes) pairs as Artifacts. +_storage_writers = None +_storage_writers_lock = threading.Lock() + + +def _forget_storage_writers(): + # A forked child (pulpcore runs each task in one) inherits the pool object but not its + # threads, so work submitted to it would wait forever. The child starts its own pool. + global _storage_writers, _storage_writers_lock + _storage_writers = None + _storage_writers_lock = threading.Lock() + + +os.register_at_fork(after_in_child=_forget_storage_writers) + + +def _storage_writer_pool(): + """Return this process's pool of threads that write files to storage. + + The storage backend keeps one connection per thread (an S3 client, for example), so + threads that outlive a single call keep their connections warm. A pool created per + call would open a new connection for every file and lose what writing in parallel + gains. + """ + global _storage_writers + with _storage_writers_lock: + if _storage_writers is None: + _storage_writers = ThreadPoolExecutor( + max_workers=8, thread_name_prefix="maven-storage-writer" + ) + return _storage_writers + + +def _save_blobs(blobs, pulp_domain): + """Save byte strings as Artifacts and return them keyed by sha256. + + Follows the same pattern as pulpcore's sync pipeline ArtifactSaver stage: one query + finds which digests already exist, and only genuinely new content is written, so an + existing artifact costs no file open and no upload. - Follows the same pattern as pulpcore's sync pipeline ArtifactSaver stage: - batch-query which sha256 digests already exist, then only open temp files - for genuinely new content. This avoids the file-descriptor accumulation that - occurs when checking existence one-at-a-time in a tight loop — existing - artifacts are returned with zero file opens. + The new files are written to storage in parallel, each straight to the path its + Artifact uses, so inserting the Artifact rows afterwards finds them in place and does + not upload them again. These files are small and numerous (metadata, checksums, index + pages), so the round trips to storage, not the bytes, are what would cost time if + they were written one after another. Args: - pages: iterable of (dir_path, html_bytes) pairs. + blobs: iterable of bytes. pulp_domain: the Domain instance to scope Artifacts to. Returns: - dict mapping dir_path -> Artifact. + dict mapping sha256 -> Artifact. """ - # Map each unique sha256 to one representative (dir_path, html_bytes). - # Multiple directories can share the same sha256 (identical listings). - sha256_to_repr: dict = {} - dir_to_sha256: dict = {} - for dir_path, html_bytes in pages: - digest = hashlib.sha256(html_bytes).hexdigest() - dir_to_sha256[dir_path] = digest - if digest not in sha256_to_repr: - sha256_to_repr[digest] = (dir_path, html_bytes) - - # Batch-check existence — one DB query per 1 000 sha256s to keep - # the IN clause at a manageable size. - sha256_to_artifact: dict = {} - all_digests = list(sha256_to_repr.keys()) + from pulpcore.plugin.util import get_domain, set_domain + + data_by_digest = {} + for data in blobs: + data_by_digest.setdefault(hashlib.sha256(data).hexdigest(), data) + + # One DB query per 1 000 sha256s keeps the IN clause at a manageable size. + sha256_to_artifact = {} + all_digests = list(data_by_digest) for i in range(0, len(all_digests), 1000): for artifact in Artifact.objects.filter( sha256__in=all_digests[i : i + 1000], @@ -170,62 +204,76 @@ def _save_artifacts_batch(pages, pulp_domain): ): sha256_to_artifact[artifact.sha256] = artifact - # Prepare only the artifacts that do not already exist, in parallel. Database - # insertion remains batched in the calling thread. new_items = [ - (digest, html_bytes) - for digest, (_, html_bytes) in sha256_to_repr.items() + (digest, data) + for digest, data in data_by_digest.items() if digest not in sha256_to_artifact ] - if new_items: - from pulpcore.plugin.util import get_domain, set_domain - - # ContextVar values (including the domain set by pulpcore's task runner) do - # NOT propagate to ThreadPoolExecutor worker threads. Capture the domain in - # the calling thread and set it explicitly at the start of each worker. - # Context.run() cannot be used here: a single Context object cannot be - # entered concurrently by multiple threads (raises "cannot enter context: - # already entered" when more than one worker fires at the same time). - current_domain = get_domain() + if not new_items: + return sha256_to_artifact + + # ContextVar values (including the domain set by pulpcore's task runner) do NOT + # propagate to ThreadPoolExecutor worker threads. Capture the domain in the calling + # thread and set it explicitly at the start of each worker. Context.run() cannot be + # used here: a single Context object cannot be entered concurrently by multiple + # threads (raises "cannot enter context: already entered"). + current_domain = get_domain() + + with tempfile.TemporaryDirectory() as temp_dir: + + def _store(item): + digest, data = item + set_domain(current_domain) + with tempfile.NamedTemporaryFile(dir=temp_dir, delete=False) as tmp: + tmp.write(data) + artifact = Artifact.init_and_validate(tmp.name, expected_digests={"sha256": digest}) + artifact.pulp_domain = pulp_domain + final_path = artifact.storage_path("") + with open(tmp.name, "rb") as fh: + stored_path = artifact.file.storage.save(final_path, File(fh)) + if stored_path == final_path: + # Already in place: saving the Artifact keeps the file where it is. + artifact.file = stored_path + else: + # The storage kept an existing file and chose another name. Drop the copy + # and let saving the Artifact move the file the usual way. + artifact.file.storage.delete(stored_path) + return artifact - pending_artifacts = [] + artifacts = sorted(_storage_writer_pool().map(_store, new_items), key=lambda a: a.sha256) - def flush_pending(): - pending_artifacts.sort(key=lambda artifact: artifact.sha256) + # Database insertion stays batched in the calling thread. + for i in range(0, len(artifacts), 500): + batch = artifacts[i : i + 500] try: - artifacts = Artifact.objects.bulk_get_or_create(pending_artifacts) - sha256_to_artifact.update({artifact.sha256: artifact for artifact in artifacts}) + for artifact in Artifact.objects.bulk_get_or_create(batch): + sha256_to_artifact[artifact.sha256] = artifact finally: # bulk_create bypasses Artifact.save(), including its file cleanup. - for artifact in pending_artifacts: + for artifact in batch: artifact.file.close() - pending_artifacts.clear() - - with tempfile.TemporaryDirectory() as temp_dir: - - def _upload(digest, html_bytes): - set_domain(current_domain) - with tempfile.NamedTemporaryFile(dir=temp_dir, delete=False) as tmp: - tmp.write(html_bytes) - tmp.flush() - artifact = Artifact.init_and_validate( - tmp.name, - expected_digests={"sha256": digest}, - ) - artifact.pulp_domain = pulp_domain - return artifact - with ThreadPoolExecutor(max_workers=10) as pool: - futures = {pool.submit(_upload, digest, data) for digest, data in new_items} - for future in as_completed(futures): - pending_artifacts.append(future.result()) - if len(pending_artifacts) >= 500: - flush_pending() + return sha256_to_artifact - if pending_artifacts: - flush_pending() - return {dir_path: sha256_to_artifact[digest] for dir_path, digest in dir_to_sha256.items()} +def _save_artifacts_batch(pages, pulp_domain): + """Save a collection of (dir_path, html_bytes) pairs as Artifacts. + + Multiple directories can share the same listing, and so the same Artifact. + + Args: + pages: iterable of (dir_path, html_bytes) pairs. + pulp_domain: the Domain instance to scope Artifacts to. + + Returns: + dict mapping dir_path -> Artifact. + """ + pages = list(pages) + sha256_to_artifact = _save_blobs((html_bytes for _, html_bytes in pages), pulp_domain) + return { + dir_path: sha256_to_artifact[hashlib.sha256(html_bytes).hexdigest()] + for dir_path, html_bytes in pages + } def _parse_index_pages(page_sources): @@ -494,16 +542,16 @@ def _save_metadata_content(group_id, artifact_id, version, base_path, metadata_x Returns a list of MavenMetadata primary keys. """ - xml_artifact = _save_artifact(metadata_xml, pulp_domain) - - files_to_create = [("maven-metadata.xml", base_path, xml_artifact)] + files = [("maven-metadata.xml", base_path, metadata_xml)] for ext, algo in [(".md5", "md5"), (".sha1", "sha1"), (".sha256", "sha256")]: - checksum_value = getattr(xml_artifact, algo) - if checksum_value is None: - checksum_value = hashlib.new(algo, metadata_xml).hexdigest() - checksum_bytes = checksum_value.encode("utf-8") - checksum_artifact = _save_artifact(checksum_bytes, pulp_domain) - files_to_create.append((f"maven-metadata.xml{ext}", f"{base_path}{ext}", checksum_artifact)) + checksum_bytes = hashlib.new(algo, metadata_xml).hexdigest().encode("utf-8") + files.append((f"maven-metadata.xml{ext}", f"{base_path}{ext}", checksum_bytes)) + + sha256_to_artifact = _save_blobs((data for _, _, data in files), pulp_domain) + files_to_create = [ + (filename, relative_path, sha256_to_artifact[hashlib.sha256(data).hexdigest()]) + for filename, relative_path, data in files + ] pks = [] for filename, relative_path, artifact in files_to_create: From 42f90b87c38a123829b3c7bac0bdd40e8ad71c44 Mon Sep 17 00:00:00 2001 From: Danilo Lakovic Date: Wed, 28 Oct 2026 14:26:55 +0100 Subject: [PATCH 3/3] Add deployed files to the repository in batches (opt-in) The deploy API turns every uploaded file into its own repository version. A Maven, Gradle or sbt publish uploads about fifteen files per module, the checksum files among them, so a module costs fifteen versions, each of which regenerates metadata and index pages and stores the full list of the repository's content ids. Publishing is slow and the versions table grows with the square of the number of deployed files. With MAVEN_DEPLOY_BATCH on, a deploy stores the file and its content unit, records it as pending for the repository (MavenPendingDeploy) and answers 201 at once. A commit task adds everything pending in one version once uploads pause for MAVEN_DEPLOY_BATCH_LINGER seconds, or at the latest MAVEN_DEPLOY_BATCH_MAX_WAIT seconds after the oldest pending file. This follows the pending blobs of container pushes and the grouped uploads of the Python plugin. What a client sees stays the same: - A pending file is served by the distribution until its batch is committed, and recording it invalidates the content cache, so a build can resolve what it has just published. - Within a batch the newest file deployed to a path wins. - A maven-metadata.xml the client deployed within a batch is kept for its artifact or SNAPSHOT version instead of being regenerated, which is what happens when it arrives last in a version of its own. It matters for SNAPSHOTs deployed under timestamped names, which the generated version-level metadata does not resolve to. Uploads arriving together dispatch one commit task between them, under a Postgres advisory lock per repository. Off by default; without it the deploy API behaves as before. --- CHANGES/+deploy-batching.feature | 2 + docs/user/guides/deploy.md | 32 ++++ pulp_maven/app/maven_deploy_api.py | 71 ++++++-- .../app/migrations/0015_mavenpendingdeploy.py | 63 +++++++ pulp_maven/app/models.py | 120 ++++++++++-- pulp_maven/app/settings.py | 8 + pulp_maven/app/tasks/__init__.py | 59 +++++- .../tests/functional/api/test_deploy_batch.py | 172 ++++++++++++++++++ .../tests/functional/api/test_mvn_deploy.py | 12 +- 9 files changed, 504 insertions(+), 35 deletions(-) create mode 100644 CHANGES/+deploy-batching.feature create mode 100644 pulp_maven/app/migrations/0015_mavenpendingdeploy.py create mode 100644 pulp_maven/tests/functional/api/test_deploy_batch.py diff --git a/CHANGES/+deploy-batching.feature b/CHANGES/+deploy-batching.feature new file mode 100644 index 00000000..ac7a178b --- /dev/null +++ b/CHANGES/+deploy-batching.feature @@ -0,0 +1,2 @@ +Added the opt-in `MAVEN_DEPLOY_BATCH` setting: deployed files are recorded as pending, served at once +and added to the repository in batches, one repository version per batch instead of one per file. diff --git a/docs/user/guides/deploy.md b/docs/user/guides/deploy.md index 14af4604..54cd42cf 100644 --- a/docs/user/guides/deploy.md +++ b/docs/user/guides/deploy.md @@ -141,3 +141,35 @@ Add the following stanza to the `pom.xml` in your Maven project. [INFO] Finished at: 2023-03-10T16:13:06-05:00 [INFO] ------------------------------------------------------------------------ ``` + +## Concurrent deploys + +Each deployed file is added to the repository by a task that holds the repository while it runs. +An upload that arrives while another one holds it waits for its turn, for up to +`MAVEN_DEPLOY_TASK_TIMEOUT` seconds (30 by default), and only then answers `429`. Maven, Gradle and +sbt do not retry a refused upload, so raise the timeout if many builds publish into one repository +at the same time. + +## Deploying in batches + +By default every deployed file, checksum files included, becomes its own repository version. A +build publishes about a dozen files per module, so a release of a multi-module project leaves +hundreds of versions, each of which regenerates `maven-metadata.xml` and the index pages and stores +the list of the repository's content. + +With `MAVEN_DEPLOY_BATCH = True` in the Pulp settings, a deployed file is recorded as pending and +the upload answers at once. A task then adds everything pending to the repository in one version, +once no file has been deployed for `MAVEN_DEPLOY_BATCH_LINGER` seconds (1 by default), and at the +latest `MAVEN_DEPLOY_BATCH_MAX_WAIT` seconds (10 by default) after the oldest pending file. + +What clients see does not change: + +- A pending file is served by the repository's distribution as soon as its upload is answered, so a + build can resolve what it has just published. +- When the same path is deployed twice within a batch, the newer file is kept. +- A `maven-metadata.xml` deployed by the client within a batch is kept rather than regenerated, as + it is when it arrives in a version of its own. This keeps SNAPSHOT versions deployed under + timestamped file names resolvable. + +What changes is that several files share one repository version, so versions no longer correspond +to single uploads. diff --git a/pulp_maven/app/maven_deploy_api.py b/pulp_maven/app/maven_deploy_api.py index 2521b7b4..8ecf9d9f 100644 --- a/pulp_maven/app/maven_deploy_api.py +++ b/pulp_maven/app/maven_deploy_api.py @@ -4,7 +4,7 @@ from tempfile import NamedTemporaryFile from django.conf import settings -from django.db import IntegrityError +from django.db import IntegrityError, connection, transaction from django.http import Http404 from django.shortcuts import get_object_or_404, redirect from drf_spectacular.utils import extend_schema @@ -13,18 +13,27 @@ from rest_framework.viewsets import ViewSet from pulpcore.plugin.constants import TASK_FINAL_STATES, TASK_STATES -from pulpcore.plugin.models import Artifact, ContentArtifact +from pulpcore.plugin.models import Artifact, ContentArtifact, Task from pulpcore.plugin.tasking import dispatch -from pulpcore.plugin.util import get_domain +from pulpcore.plugin.util import get_domain, get_prn from pulp_maven.app.models import ( MavenArtifact, MavenDistribution, MavenMetadata, MavenPackage, + MavenPendingDeploy, MavenRepository, ) -from pulp_maven.app.tasks import aadd_and_remove +from pulp_maven.app.tasks import aadd_and_remove, commit_pending_deploys + +# How pulpcore names the commit task, to find one that is already waiting. +COMMIT_TASK_NAME = f"{commit_pending_deploys.__module__}.{commit_pending_deploys.__name__}" + + +def _commit_lock_key(repo): + """A Postgres advisory lock key for dispatching a repository's commit task.""" + return int(hashlib.sha256(f"maven-deploy-commit:{repo.pk}".encode()).hexdigest()[:15], 16) def has_task_completed(task): @@ -224,6 +233,21 @@ def create(self, request, name, path): ca.artifact = artifact ca.save() + if not is_metadata and content.filename.endswith(".pom"): + pkg, created = MavenPackage.objects.get_or_create( + group_id=content.group_id, + artifact_id=content.artifact_id, + version=content.version, + _pulp_domain=get_domain(), + ) + if created or content.version.endswith("-SNAPSHOT"): + pkg.update_from_pom(artifact) + pkg.save() + + if settings.MAVEN_DEPLOY_BATCH: + self.add_pending(repo, content, path) + return Response(status=201) + if is_metadata: metadata_to_remove = MavenMetadata.objects.filter( pk__in=repo.latest_version().content.all(), @@ -239,17 +263,6 @@ def create(self, request, name, path): add_content_units = [str(content.pk)] - if not is_metadata and content.filename.endswith(".pom"): - pkg, created = MavenPackage.objects.get_or_create( - group_id=content.group_id, - artifact_id=content.artifact_id, - version=content.version, - _pulp_domain=get_domain(), - ) - if created or content.version.endswith("-SNAPSHOT"): - pkg.update_from_pom(artifact) - pkg.save() - dispatched_task = dispatch( aadd_and_remove, exclusive_resources=[repo], @@ -265,6 +278,34 @@ def create(self, request, name, path): if has_task_completed(dispatched_task): return Response(status=201) + @staticmethod + def add_pending(repo, content, relative_path): + """Record a deployed file for the next batch and make sure a commit task will add it.""" + MavenPendingDeploy.objects.create( + repository=repo, content=content, relative_path=relative_path + ) + # Pending files are served from the moment they are recorded, so a cached answer for + # this path (a 404 from before the upload) must not outlive it. + repo.invalidate_cache() + # A commit task still waiting to run will find this file. One that is already running + # may have taken its batch before the file arrived, so then another one is dispatched. + # The check and the dispatch hold a lock per repository, so uploads arriving together + # dispatch one task between them rather than one each. + with transaction.atomic(): + with connection.cursor() as cursor: + cursor.execute("SELECT pg_advisory_xact_lock(%s)", [_commit_lock_key(repo)]) + commit_waiting = Task.objects.filter( + name=COMMIT_TASK_NAME, + state=TASK_STATES.WAITING, + reserved_resources_record__contains=[get_prn(repo)], + ).exists() + if not commit_waiting: + dispatch( + commit_pending_deploys, + exclusive_resources=[repo], + kwargs={"repository_pk": str(repo.pk)}, + ) + @staticmethod def receive_artifact(chunk): """Handles assembling of a file as it's being uploaded.""" diff --git a/pulp_maven/app/migrations/0015_mavenpendingdeploy.py b/pulp_maven/app/migrations/0015_mavenpendingdeploy.py new file mode 100644 index 00000000..4039af45 --- /dev/null +++ b/pulp_maven/app/migrations/0015_mavenpendingdeploy.py @@ -0,0 +1,63 @@ +# Generated by Django 5.2.18 on 2026-10-28 12:02 + +import django.db.models.deletion +import django_lifecycle.mixins +import pulpcore.app.models.base +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("core", "0148_artifact_artifact_domain_size_index"), + ("maven", "0014_mavenpackage_trgm_indexes"), + ] + + operations = [ + migrations.CreateModel( + name="MavenPendingDeploy", + fields=[ + ( + "pulp_id", + models.UUIDField( + default=pulpcore.app.models.base.pulp_uuid, + editable=False, + primary_key=True, + serialize=False, + ), + ), + ("pulp_created", models.DateTimeField(auto_now_add=True)), + ("pulp_last_updated", models.DateTimeField(auto_now=True, null=True)), + ("relative_path", models.TextField()), + ( + "content", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="+", + to="core.content", + ), + ), + ( + "repository", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="pending_deploys", + to="maven.mavenrepository", + ), + ), + ], + options={ + "default_related_name": "%(app_label)s_%(model_name)s", + "indexes": [ + models.Index( + fields=["repository", "pulp_created"], name="maven_maven_reposit_10a5ab_idx" + ), + models.Index( + fields=["repository", "relative_path"], + name="maven_maven_reposit_b3b271_idx", + ), + ], + }, + bases=(django_lifecycle.mixins.LifecycleModelMixin, models.Model), + ), + ] diff --git a/pulp_maven/app/models.py b/pulp_maven/app/models.py index 3dfed8fa..c3049860 100644 --- a/pulp_maven/app/models.py +++ b/pulp_maven/app/models.py @@ -5,6 +5,7 @@ from logging import getLogger from os import path +from django.conf import settings from django.contrib.postgres.indexes import GinIndex from django.db import IntegrityError, connection, models, transaction from django.db.models import Exists, OuterRef, Q @@ -12,6 +13,7 @@ from pulpcore.plugin.models import ( AutoAddObjPermsMixin, + BaseModel, Content, Distribution, Remote, @@ -32,6 +34,8 @@ # Pull-through tasks run asynchronously from the content app; any extra work in # finalize_new_version delays version creation and races with subsequent reads. _pull_through_ctx = threading.local() +# Set while a batch of deployed files is being added (see commit_pending_deploys). +_deploy_batch_ctx = threading.local() class MavenContentMixin: @@ -459,6 +463,37 @@ class Meta: ] +class MavenPendingDeploy(BaseModel): + """A deployed file waiting to be added to its repository. + + With ``MAVEN_DEPLOY_BATCH`` on, the deploy API stores the file and its content unit and + records it here instead of creating a repository version for it. A commit task then adds + everything that is pending for the repository in one version, so a publish of many files + costs a few versions rather than one per file. + + Fields: + relative_path (models.TextField): Where the file is served from, so that a later + deploy of the same path within one batch replaces an earlier one. + + Relations: + repository (MavenRepository): The repository the file was deployed to. + content (pulpcore.plugin.models.Content): The deployed content unit. + """ + + repository = models.ForeignKey( + "MavenRepository", on_delete=models.CASCADE, related_name="pending_deploys" + ) + content = models.ForeignKey(Content, on_delete=models.CASCADE, related_name="+") + relative_path = models.TextField() + + class Meta: + default_related_name = "%(app_label)s_%(model_name)s" + indexes = [ + models.Index(fields=["repository", "pulp_created"]), + models.Index(fields=["repository", "relative_path"]), + ] + + class MavenDistribution(Distribution, AutoAddObjPermsMixin): """ Distribution for 'maven' content. @@ -466,6 +501,31 @@ class MavenDistribution(Distribution, AutoAddObjPermsMixin): TYPE = "maven" + def _pending_content_artifact(self, path): + """Return the newest deployed file at ``path`` that is still waiting for its batch. + + Serving it makes a deployed file readable as soon as its upload is answered, as when + every file got its own version, so a build that publishes and then resolves at once + does not see a 404 in between. + """ + if not settings.MAVEN_DEPLOY_BATCH or not self.repository_id or self.repository_version_id: + return None + from pulpcore.plugin.models import ContentArtifact + + content_id = ( + MavenPendingDeploy.objects.filter(repository_id=self.repository_id, relative_path=path) + .order_by("-pulp_created") + .values_list("content_id", flat=True) + .first() + ) + if content_id is None: + return None + return ( + ContentArtifact.objects.select_related("artifact") + .filter(content_id=content_id, relative_path=path) + .first() + ) + def content_handler(self, path): """Serve pre-generated HTML index pages inline, bypassing redirect-to-object-storage. @@ -485,6 +545,10 @@ def content_handler(self, path): from pulp_maven.app.path_index.content import indexed_response + pending = self._pending_content_artifact(path) + if pending is not None: + return pending + response = indexed_response(self, path) if response is not None: return response @@ -968,25 +1032,47 @@ def _generate_metadata(self, new_version, changes): for group_id, artifact_id in affected_pairs: pairs_q |= Q(group_id=group_id, artifact_id=artifact_id) - stale_pks = list( - MavenMetadata.objects.filter( - pk__in=new_version.content, - version=None, - filename__in=METADATA_FILENAMES, - ) - .filter(pairs_q) - .values_list("pk", flat=True) - ) - changes.to_remove.update(stale_pks) + # A batch of deploys that carries the client's own maven-metadata.xml for a pair, or + # for a snapshot version, keeps it. Each file in its own version would have left the + # same result: the client uploads its metadata after the files, in a version without + # artifacts, which is not regenerated. The client's metadata can also say more than + # the generated one, for example which timestamped build a SNAPSHOT resolves to. + client_pairs, client_triples = set(), set() + if getattr(_deploy_batch_ctx, "active", False): + for group_id, artifact_id, version in MavenMetadata.objects.filter( + pk__in=new_version.added(), filename="maven-metadata.xml" + ).values_list("group_id", "artifact_id", "version"): + if version is None: + client_pairs.add((group_id, artifact_id)) + else: + client_triples.add((group_id, artifact_id, version)) + metadata_pairs = affected_pairs - client_pairs + affected_snapshot_triples -= client_triples versions_by_pair = defaultdict(set) - for row in ( - MavenArtifact.objects.filter(pk__in=new_version.content) - .filter(pairs_q) - .values("group_id", "artifact_id", "version") - .distinct() - ): - versions_by_pair[(row["group_id"], row["artifact_id"])].add(row["version"]) + if metadata_pairs: + metadata_pairs_q = Q() + for group_id, artifact_id in metadata_pairs: + metadata_pairs_q |= Q(group_id=group_id, artifact_id=artifact_id) + + stale_pks = list( + MavenMetadata.objects.filter( + pk__in=new_version.content, + version=None, + filename__in=METADATA_FILENAMES, + ) + .filter(metadata_pairs_q) + .values_list("pk", flat=True) + ) + changes.to_remove.update(stale_pks) + + for row in ( + MavenArtifact.objects.filter(pk__in=new_version.content) + .filter(metadata_pairs_q) + .values("group_id", "artifact_id", "version") + .distinct() + ): + versions_by_pair[(row["group_id"], row["artifact_id"])].add(row["version"]) new_metadata_pks = [] diff --git a/pulp_maven/app/settings.py b/pulp_maven/app/settings.py index 2368939a..e7ba8e7b 100644 --- a/pulp_maven/app/settings.py +++ b/pulp_maven/app/settings.py @@ -12,3 +12,11 @@ # How long a deploy request waits, in seconds, for its turn to add the uploaded file to the # repository while other deploys to the same repository go first. After that it answers 429. MAVEN_DEPLOY_TASK_TIMEOUT = 30 + +# Opt-in: the deploy API records deployed files as pending and a commit task adds them to the +# repository in batches, one repository version per batch instead of one per file. A file is +# served once its batch is committed: about MAVEN_DEPLOY_BATCH_LINGER seconds after uploads to +# the repository pause, and at least every MAVEN_DEPLOY_BATCH_MAX_WAIT seconds while they go on. +MAVEN_DEPLOY_BATCH = False +MAVEN_DEPLOY_BATCH_LINGER = 1 +MAVEN_DEPLOY_BATCH_MAX_WAIT = 10 diff --git a/pulp_maven/app/tasks/__init__.py b/pulp_maven/app/tasks/__init__.py index 4b3f4904..002a373c 100644 --- a/pulp_maven/app/tasks/__init__.py +++ b/pulp_maven/app/tasks/__init__.py @@ -10,9 +10,11 @@ from xml.etree.ElementTree import Element, SubElement, tostring from asgiref.sync import sync_to_async +from django.conf import settings from django.core.files import File from django.db import IntegrityError, transaction -from django.db.models import Q +from django.db.models import Max, Min, Q +from django.utils import timezone from redis.exceptions import RedisError from pulpcore.plugin.models import ( @@ -37,7 +39,9 @@ MavenArtifact, MavenMetadata, MavenPackage, + MavenPendingDeploy, MavenRepository, + _deploy_batch_ctx, ) from pulp_maven.app.util import PulpListingParser @@ -308,6 +312,59 @@ async def aadd_and_remove(*args, **kwargs): return await sync_to_async(add_and_remove)(*args, **kwargs) +def commit_pending_deploys(repository_pk): + """Add the files deployed to a repository and still pending, a batch per version. + + Waits until no file has been deployed to the repository for MAVEN_DEPLOY_BATCH_LINGER + seconds, but not longer than MAVEN_DEPLOY_BATCH_MAX_WAIT seconds after the oldest + pending file, then adds everything pending in one repository version. Repeats while + files keep arriving, so nothing deployed during a commit is left behind. + + Within a batch the newest file deployed to a path wins, as it would had each file been + added in its own version. + + Args: + repository_pk (str): The repository to commit pending deploys for. + """ + repository = MavenRepository.objects.get(pk=repository_pk) + linger = datetime.timedelta(seconds=settings.MAVEN_DEPLOY_BATCH_LINGER) + max_wait = datetime.timedelta(seconds=settings.MAVEN_DEPLOY_BATCH_MAX_WAIT) + pending = MavenPendingDeploy.objects.filter(repository=repository) + + while True: + span = pending.aggregate(first=Min("pulp_created"), last=Max("pulp_created")) + if span["first"] is None: + return + wait = ( + min(span["last"] + linger, span["first"] + max_wait) - timezone.now() + ).total_seconds() + if wait > 0: + time.sleep(wait) + continue + + rows = list( + pending.order_by("pulp_created").values_list("pk", "content_id", "relative_path") + ) + newest_by_path = {} + for _, content_id, relative_path in rows: + newest_by_path[relative_path] = content_id + _deploy_batch_ctx.active = True + try: + add_and_remove( + repository_pk, + add_content_units=list(newest_by_path.values()), + remove_content_units=[], + ) + finally: + _deploy_batch_ctx.active = False + MavenPendingDeploy.objects.filter(pk__in=[pk for pk, _, _ in rows]).delete() + log.info( + "Committed pending deploys: repository=%s, files=%d", + repository.name, + len(newest_by_path), + ) + + def _add_and_remove_skip_metadata(*args, **kwargs): """Wrapper around add_and_remove that skips metadata generation.""" from pulp_maven.app.models import _pull_through_ctx diff --git a/pulp_maven/tests/functional/api/test_deploy_batch.py b/pulp_maven/tests/functional/api/test_deploy_batch.py new file mode 100644 index 00000000..19059c07 --- /dev/null +++ b/pulp_maven/tests/functional/api/test_deploy_batch.py @@ -0,0 +1,172 @@ +"""Deploys recorded as pending and added to the repository in batches (MAVEN_DEPLOY_BATCH).""" + +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from urllib.parse import urlsplit + +import pytest +import requests + +from pulp_maven.tests.functional.utils import download_file + + +@pytest.fixture(autouse=True) +def _batch_mode_only(pulp_settings): + if not pulp_settings.MAVEN_DEPLOY_BATCH: + pytest.skip("The server does not batch deploys (MAVEN_DEPLOY_BATCH is off).") + + +@pytest.fixture +def deploy(bindings_cfg, pulp_settings): + parts = urlsplit(bindings_cfg.host) + prefix = "default/" if pulp_settings.DOMAIN_ENABLED else "" + + def _deploy(repo_name, path, data): + return requests.put( + f"{parts.scheme}://{parts.netloc}/pulp/maven/{prefix}{repo_name}/{path}", + data=data, + auth=(bindings_cfg.username, bindings_cfg.password), + verify=False, + ).status_code + + return _deploy + + +def _committed(pulp_settings): + """Seconds after the last deploy within which a batch is committed, with room to spare.""" + return pulp_settings.MAVEN_DEPLOY_BATCH_LINGER + pulp_settings.MAVEN_DEPLOY_BATCH_MAX_WAIT + 20 + + +def _wait_for(check, timeout): + deadline = time.monotonic() + timeout + while not check(): + assert time.monotonic() < deadline, "the deployed files were not committed in time" + time.sleep(0.5) + + +def _version_number(maven_repo_api_client, repo): + href = maven_repo_api_client.read(repo.pulp_href).latest_version_href + return int(href.rstrip("/").rsplit("/", 1)[1]) + + +def _artifact_count(maven_repo_api_client, maven_artifact_api_client, repo, artifact_id): + href = maven_repo_api_client.read(repo.pulp_href).latest_version_href + return maven_artifact_api_client.list(repository_version=href, artifact_id=artifact_id).count + + +def test_sequential_deploys_share_versions( + deploy, + maven_repo_api_client, + maven_artifact_api_client, + maven_repo_factory, + maven_distribution_factory, + pulp_settings, +): + """A publish of many files, one after another, lands in far fewer versions than files. + + Each deploy answers at once; the files appear together once uploads pause. + """ + repo = maven_repo_factory() + maven_distribution_factory(repository=repo.pulp_href) + artifact_id = f"seq-{uuid.uuid4()}" + paths = [ + f"com/example/{artifact_id}/1.0.0/{artifact_id}-1.0.0{suffix}" + for suffix in (".jar", ".jar.md5", ".jar.sha1", ".pom", ".pom.md5", ".pom.sha1") + ] + + for path in paths: + assert deploy(repo.name, path, f"bytes of {path}".encode()) == 201 + + _wait_for( + lambda: _artifact_count(maven_repo_api_client, maven_artifact_api_client, repo, artifact_id) + == len(paths), + _committed(pulp_settings), + ) + assert _version_number(maven_repo_api_client, repo) < len(paths) + assert not maven_repo_api_client.read(repo.pulp_href).latest_version_href.endswith("/0/") + + +def test_concurrent_deploys_all_land( + deploy, + maven_repo_api_client, + maven_artifact_api_client, + maven_repo_factory, + maven_distribution_factory, + pulp_settings, +): + """Uploads arriving at the same time all answer 201 and all end up in the repository.""" + repo = maven_repo_factory() + maven_distribution_factory(repository=repo.pulp_href) + artifact_id = f"par-{uuid.uuid4()}" + paths = [f"com/example/{artifact_id}/1.0.{i}/{artifact_id}-1.0.{i}.jar" for i in range(8)] + + with ThreadPoolExecutor(max_workers=len(paths)) as pool: + statuses = list(pool.map(lambda p: deploy(repo.name, p, p.encode()), paths)) + + assert statuses == [201] * len(paths) + _wait_for( + lambda: _artifact_count(maven_repo_api_client, maven_artifact_api_client, repo, artifact_id) + == len(paths), + _committed(pulp_settings), + ) + + +def test_deployed_file_is_served_before_its_batch_is_committed( + deploy, + maven_repo_api_client, + maven_repo_factory, + maven_distribution_factory, +): + """A file is readable as soon as its upload is answered, before any version holds it. + + A build that publishes and then resolves what it published would otherwise see a 404 + for as long as the batch waits. + """ + repo = maven_repo_factory() + distribution = maven_distribution_factory(repository=repo.pulp_href) + artifact_id = f"read-{uuid.uuid4()}" + path = f"com/example/{artifact_id}/1.0.0/{artifact_id}-1.0.0.jar" + + assert deploy(repo.name, path, b"just deployed") == 201 + + served = download_file(f"{distribution.base_url.rstrip('/')}/{path}") + assert served.body == b"just deployed" + assert maven_repo_api_client.read(repo.pulp_href).latest_version_href.endswith("/0/") + + +def test_client_snapshot_metadata_in_a_batch_is_kept( + deploy, + maven_repo_api_client, + maven_repo_factory, + maven_distribution_factory, + pulp_settings, +): + """A SNAPSHOT deployed with timestamped file names keeps the client's maven-metadata.xml. + + That file is what maps the SNAPSHOT version to its timestamped build; it must not be + replaced by metadata generated from the file names when it lands in the same batch. + """ + repo = maven_repo_factory() + distribution = maven_distribution_factory(repository=repo.pulp_href) + artifact_id = f"tsnap-{uuid.uuid4()}" + base = f"com/example/{artifact_id}/1.0-SNAPSHOT" + build = "1.0-20261028.120000-1" + metadata = ( + f"com.example{artifact_id}" + "1.0-SNAPSHOT20261028.120000" + "1" + f"jar{build}" + "20261028120000" + "20261028120000" + ).encode() + + assert deploy(repo.name, f"{base}/{artifact_id}-{build}.jar", b"timestamped build") == 201 + assert deploy(repo.name, f"{base}/maven-metadata.xml", metadata) == 201 + + _wait_for( + lambda: maven_repo_api_client.read(repo.pulp_href).latest_version_href.endswith("/1/"), + _committed(pulp_settings), + ) + served = download_file(f"{distribution.base_url.rstrip('/')}/{base}/maven-metadata.xml") + assert served.body == metadata diff --git a/pulp_maven/tests/functional/api/test_mvn_deploy.py b/pulp_maven/tests/functional/api/test_mvn_deploy.py index 373333e2..a7631fa7 100644 --- a/pulp_maven/tests/functional/api/test_mvn_deploy.py +++ b/pulp_maven/tests/functional/api/test_mvn_deploy.py @@ -1,5 +1,6 @@ import os import subprocess +import time from urllib.parse import urljoin import pytest @@ -67,8 +68,15 @@ def test_mvn_deploy_workflow( pytest.fail(msg) # Assert that the latest version is 12 (6 artifacts + 6 metadata including checksums) - repo = maven_repo_api_client.read(repo.pulp_href) - assert repo.latest_version_href.endswith("/versions/12/") + if pulp_settings.MAVEN_DEPLOY_BATCH: + # The deployed files are added in batches, so they share versions; wait for them. + deadline = time.monotonic() + pulp_settings.MAVEN_DEPLOY_BATCH_MAX_WAIT + 30 + while maven_repo_api_client.read(repo.pulp_href).latest_version_href.endswith("/0/"): + assert time.monotonic() < deadline, "the deployed files were not committed in time" + time.sleep(0.5) + else: + repo = maven_repo_api_client.read(repo.pulp_href) + assert repo.latest_version_href.endswith("/versions/12/") # Assert that you can get the metadata for the simple-project pulp_unit_url = urljoin(