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/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/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 a2cf5569..8ecf9d9f 100644 --- a/pulp_maven/app/maven_deploy_api.py +++ b/pulp_maven/app/maven_deploy_api.py @@ -1,9 +1,10 @@ import hashlib import re +import time 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 @@ -11,36 +12,62 @@ from rest_framework.response import Response from rest_framework.viewsets import ViewSet -from pulpcore.plugin.models import Artifact, ContentArtifact +from pulpcore.plugin.constants import TASK_FINAL_STATES, TASK_STATES +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): """ - 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 @@ -206,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(), @@ -221,22 +263,11 @@ 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], immediate=True, - deferred=False, + deferred=True, kwargs={ "repository_pk": str(repo.pk), "add_content_units": add_content_units, @@ -247,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 36ae85c5..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 @@ -567,6 +631,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 +732,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 +762,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 +854,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 +864,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 +994,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 @@ -925,26 +1032,47 @@ def _generate_metadata(self, new_version): 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) - ) - if stale_pks: - new_version.remove_content(MavenMetadata.objects.filter(pk__in=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 = [] @@ -969,8 +1097,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 +1116,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 +1126,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 +1150,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 +1240,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/settings.py b/pulp_maven/app/settings.py index 952e91b8..e7ba8e7b 100644 --- a/pulp_maven/app/settings.py +++ b/pulp_maven/app/settings.py @@ -8,3 +8,15 @@ 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 + +# 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 09bf3b4c..002a373c 100644 --- a/pulp_maven/app/tasks/__init__.py +++ b/pulp_maven/app/tasks/__init__.py @@ -1,15 +1,20 @@ 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.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 ( @@ -34,7 +39,9 @@ MavenArtifact, MavenMetadata, MavenPackage, + MavenPendingDeploy, MavenRepository, + _deploy_batch_ctx, ) from pulp_maven.app.util import PulpListingParser @@ -133,36 +140,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: - 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. + 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. + + 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 +208,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): @@ -260,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 @@ -494,16 +599,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: 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 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(