Let concurrent publishes to one repository wait for their turn

CHANGES/+concurrent-publishes.bugfix

@@ -0,0 +1,3 @@
+Concurrent publishes to the same repository no longer fail with HTTP 429. A publish that arrives
+while another one holds the repository waits for its turn, for up to `NPM_PUBLISH_TASK_TIMEOUT`
+seconds (30 by default).

docs/user/guides/npm-publish.md

@@ -117,6 +117,14 @@ You can also set the registry per-scope in `.npmrc`:
 //pulp.example.com/npm/default/my-npm-repo/:_auth=YWRtaW46cGFzc3dvcmQ=
 ```
 
+## Concurrent Publishes
+
+Each published package is added to the repository by a task that holds the repository while it
+runs. A publish that arrives while another one holds it waits for its turn, for up to
+`NPM_PUBLISH_TASK_TIMEOUT` seconds (30 by default), and only then answers `429`. npm, yarn and
+pnpm do not retry a refused publish, so raise the timeout if many packages are published into one
+repository at the same time, for example by a workspace publishing all its packages.
+
 ## Verify Published Packages
 
 ### Check the repository

pulp_npm/app/npm_publish_api.py

@@ -1,9 +1,11 @@
 import base64
 import hashlib
+import time
 from logging import getLogger
 from tempfile import NamedTemporaryFile
 from urllib.parse import unquote
 
+from django.conf import settings
 from django.db import IntegrityError
 from django.shortcuts import get_object_or_404
 from rest_framework.exceptions import Throttled
@@ -12,6 +14,7 @@
 from rest_framework.views import APIView
 
 from pulpcore.plugin.access_policy import AccessPolicyFromDB
+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
@@ -23,10 +26,33 @@
 
 
 def _has_task_completed(task):
-    if task.state == "completed":
+    """
+    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 publish request waits with it for up to NPM_PUBLISH_TASK_TIMEOUT
+    seconds, so concurrent publishes to one repository take turns instead of being refused.
+
+    Raises:
+        Exception: If an error occured during the task's runtime.
+        Throttled: If the task was canceled or did not finish within the timeout.
+    """
+    timeout = settings.NPM_PUBLISH_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 package; a retried publish of the same
+            # version is accepted as well.
+            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
@@ -147,7 +173,7 @@ def put(self, request, path, package_name):
             aadd_and_remove,
             exclusive_resources=[repository],
             immediate=True,
-            deferred=False,
+            deferred=True,
             kwargs={
                 "repository_pk": str(repository.pk),
                 "add_content_units": [str(content.pk)],

pulp_npm/app/settings.py

@@ -2,3 +2,7 @@
     "dynaconf_merge_unique": True,
     "reusable_conditions": ["pulp_npm.app.global_access_conditions"],
 }
+
+# How long a publish request waits, in seconds, for its turn to add the package to the
+# repository while other publishes to the same repository go first. After that it answers 429.
+NPM_PUBLISH_TASK_TIMEOUT = 30

pulp_npm/tests/functional/api/test_npm_publish.py

@@ -185,6 +185,42 @@ def test_publish_duplicate_is_idempotent(
     assert s2 == 201
 
 
+@pytest.mark.parallel
+def test_concurrent_publishes_to_one_repository_all_succeed(
+    npm_bindings,
+    npm_repository_factory,
+    npm_distribution_factory,
+    pulp_settings,
+):
+    """Publishes that arrive while another one holds the repository wait for their turn.
+
+    Workspace tools publish several packages at once and npm does not retry a refused
+    publish, so refusing every publish that overlaps another one would fail the release.
+    """
+    domain = "default" if pulp_settings.DOMAIN_ENABLED else None
+    repo = npm_repository_factory()
+    distro = npm_distribution_factory(repository=repo.pulp_href)
+    names = [f"concurrent-{uuid.uuid4().hex[:8]}" for _ in range(8)]
+
+    async def publish_all():
+        return await asyncio.gather(
+            *(
+                _put_publish(
+                    _npm_publish_url(distro.base_path, name, domain=domain),
+                    _build_publish_body(name, "1.0.0", _build_npm_tgz(name=name)),
+                )
+                for name in names
+            )
+        )
+
+    results = _run(publish_all())
+    assert [status for status, _ in results] == [201] * len(names), results
+
+    repo = npm_bindings.RepositoriesNpmApi.read(repo.pulp_href)
+    ver = npm_bindings.RepositoriesNpmVersionsApi.read(repo.latest_version_href)
+    assert ver.content_summary.present["npm.package"]["count"] == len(names)
+
+
 @pytest.mark.parallel
 def test_publish_multiple_versions(
     npm_bindings,