Deploy API: wait instead of 429, cheaper versions, and opt-in batching

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.

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).

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.

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.

pulp_maven/app/maven_deploy_api.py

@@ -1,46 +1,73 @@
 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
 from rest_framework.exceptions import Throttled
 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."""

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),
+        ),
+    ]