diff --git a/CHANGES/+concurrent-publishes.bugfix b/CHANGES/+concurrent-publishes.bugfix new file mode 100644 index 00000000..30b6702d --- /dev/null +++ b/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). diff --git a/docs/user/guides/npm-publish.md b/docs/user/guides/npm-publish.md index 35a51146..1618f6b7 100644 --- a/docs/user/guides/npm-publish.md +++ b/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 diff --git a/pulp_npm/app/npm_publish_api.py b/pulp_npm/app/npm_publish_api.py index b13b527d..c3d6ea8c 100644 --- a/pulp_npm/app/npm_publish_api.py +++ b/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)], diff --git a/pulp_npm/app/settings.py b/pulp_npm/app/settings.py index 00a19692..ced4e31c 100644 --- a/pulp_npm/app/settings.py +++ b/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 diff --git a/pulp_npm/tests/functional/api/test_npm_publish.py b/pulp_npm/tests/functional/api/test_npm_publish.py index 17df039c..92b86944 100644 --- a/pulp_npm/tests/functional/api/test_npm_publish.py +++ b/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,