From b4d28e30c6356bbc96505ab4a71dc5fa5497ac8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20M=C3=A9ndez=20Hern=C3=A1ndez?= Date: Wed, 22 Jul 2026 21:27:22 +0200 Subject: [PATCH 1/3] Fixes: #3322 - Narrow distribution task locks for unchanged base_path Only reserve the domain-wide distributions resource for operations that can affect base_path overlap validation. Distribution creates, deletes, and updates that change base_path continue to reserve `pdrn::distributions`, while updates that leave base_path unchanged now reserve only the distribution instance itself. This reduces unnecessary serialization of `ageneral_update` tasks for ordinary distribution updates, including partial PATCH requests that omit base_path or send the existing base_path unchanged. Add a functional test covering the reserved resources used for create, partial update without base_path, partial update with unchanged base_path, partial update with changed base_path, and delete. Co-authored-by: Cursor --- CHANGES/7896.bugfix | 1 + pulpcore/app/viewsets/publication.py | 28 +++++++-- .../api/using_plugin/test_distributions.py | 60 +++++++++++++++++++ 3 files changed, 84 insertions(+), 5 deletions(-) create mode 100644 CHANGES/7896.bugfix diff --git a/CHANGES/7896.bugfix b/CHANGES/7896.bugfix new file mode 100644 index 00000000000..a6f1876cf2e --- /dev/null +++ b/CHANGES/7896.bugfix @@ -0,0 +1 @@ +Reduced lock contention for distribution updates that leave `base_path` unchanged. diff --git a/pulpcore/app/viewsets/publication.py b/pulpcore/app/viewsets/publication.py index ae7e38a8e6d..e6031607abc 100644 --- a/pulpcore/app/viewsets/publication.py +++ b/pulpcore/app/viewsets/publication.py @@ -527,8 +527,26 @@ def get_queryset(self): return qs def async_reserved_resources(self, instance): - """Return resource that locks all Distributions.""" - return [f"pdrn:{get_domain().pulp_id}:distributions"] + """ + Reserve the narrowest safe lock for async distribution operations. + + Creates, deletes, and base_path changes still lock the domain-wide distributions resource + because base_path overlap validation is domain scoped. Other updates only need to lock the + specific distribution instance. + """ + domain_distributions = f"pdrn:{get_domain().pulp_id}:distributions" + if instance is None: + return [domain_distributions] + + if getattr(self, "action", "") == "destroy": + return [instance, domain_distributions] + + request_data = getattr(getattr(self, "request", None), "data", {}) + requested_base_path = request_data.get("base_path", instance.base_path) + if requested_base_path == instance.base_path: + return [instance] + + return [instance, domain_distributions] class ListDistributionViewSet(BaseDistributionViewSet, mixins.ListModelMixin): @@ -567,9 +585,9 @@ class DistributionViewSet( LabelsMixin, ): """ - Provides read and list methods and also provides asynchronous CUD methods to dispatch tasks - with reservation that lock all Distributions preventing race conditions during base_path - checking. + Provides read and list methods plus asynchronous CUD methods that reserve the narrowest safe + distribution locks, only taking the domain-wide lock when base_path overlap validation or + base_path release needs it. """ diff --git a/pulpcore/tests/functional/api/using_plugin/test_distributions.py b/pulpcore/tests/functional/api/using_plugin/test_distributions.py index 1870c86e462..9e724638c8c 100644 --- a/pulpcore/tests/functional/api/using_plugin/test_distributions.py +++ b/pulpcore/tests/functional/api/using_plugin/test_distributions.py @@ -167,6 +167,66 @@ def test_distribution_base_path( assert json.loads(exc.value.body)["base_path"] is not None +@pytest.mark.parallel +def test_distribution_update_task_reservations( + file_bindings, + monitor_task, +): + create_task = monitor_task( + file_bindings.DistributionsFileApi.create( + {"name": str(uuid4()), "base_path": str(uuid4())} + ).task + ) + assert any( + resource.endswith(":distributions") for resource in create_task.reserved_resources_record + ) + distribution = file_bindings.DistributionsFileApi.read(create_task.created_resources[0]) + assert distribution.prn not in create_task.reserved_resources_record + + no_base_path_update_task = monitor_task( + file_bindings.DistributionsFileApi.partial_update( + distribution.pulp_href, + {"name": str(uuid4())}, + ).task + ) + assert distribution.prn in no_base_path_update_task.reserved_resources_record + assert not any( + resource.endswith(":distributions") + for resource in no_base_path_update_task.reserved_resources_record + ) + + unchanged_base_path_update_task = monitor_task( + file_bindings.DistributionsFileApi.partial_update( + distribution.pulp_href, + {"name": str(uuid4()), "base_path": distribution.base_path}, + ).task + ) + assert distribution.prn in unchanged_base_path_update_task.reserved_resources_record + assert not any( + resource.endswith(":distributions") + for resource in unchanged_base_path_update_task.reserved_resources_record + ) + + base_path_update_task = monitor_task( + file_bindings.DistributionsFileApi.partial_update( + distribution.pulp_href, {"base_path": str(uuid4())} + ).task + ) + assert distribution.prn in base_path_update_task.reserved_resources_record + assert any( + resource.endswith(":distributions") + for resource in base_path_update_task.reserved_resources_record + ) + + delete_task = monitor_task( + file_bindings.DistributionsFileApi.delete(distribution.pulp_href).task + ) + assert distribution.prn in delete_task.reserved_resources_record + assert any( + resource.endswith(":distributions") for resource in delete_task.reserved_resources_record + ) + + @pytest.mark.parallel def test_distribution_filtering( file_bindings, From 2abdfecdefdaa4b0e5c5455d6ab8931341e5aa57 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20M=C3=A9ndez=20Hern=C3=A1ndez?= Date: Thu, 6 Aug 2026 11:41:51 +0200 Subject: [PATCH 2/3] Refine distribution locks for upgrade-safe base_path protection Rework distribution async CUD to keep the legacy domain-scoped distributions lock shared for upgrade compatibility while introducing an explicit global base_path lock for operations that create or change paths. Ordinary updates and deletes now keep instance-scoped serialization without relying on the broad distributions lock. Fixes: #3322 Co-authored-by: Cursor --- pulpcore/app/viewsets/publication.py | 99 ++++++++++++++++--- .../api/using_plugin/test_distributions.py | 51 ++++++---- 2 files changed, 121 insertions(+), 29 deletions(-) diff --git a/pulpcore/app/viewsets/publication.py b/pulpcore/app/viewsets/publication.py index e6031607abc..02ba8772855 100644 --- a/pulpcore/app/viewsets/publication.py +++ b/pulpcore/app/viewsets/publication.py @@ -2,8 +2,11 @@ from django.db.models import Prefetch from django_filters import Filter +from drf_spectacular.utils import extend_schema from rest_framework import mixins, serializers +from rest_framework.response import Response +from pulpcore.app import tasks from pulpcore.app.models import ( ArtifactDistribution, ContentGuard, @@ -15,7 +18,9 @@ Repository, ) from pulpcore.app.models.publication import CompositeContentGuard +from pulpcore.app.response import OperationPostponedResponse from pulpcore.app.serializers import ( + AsyncOperationResponseSerializer, ArtifactDistributionSerializer, ContentGuardSerializer, ContentRedirectContentGuardSerializer, @@ -43,6 +48,8 @@ WithContentInFilter, ) from pulpcore.filters import BaseFilterSet +from pulpcore.openapi import InheritSerializer +from pulpcore.tasking.tasks import dispatch class RepositoryThroughVersionFilter(Filter): @@ -528,25 +535,31 @@ def get_queryset(self): def async_reserved_resources(self, instance): """ - Reserve the narrowest safe lock for async distribution operations. + Reserve safe distribution locks for async operations. - Creates, deletes, and base_path changes still lock the domain-wide distributions resource - because base_path overlap validation is domain scoped. Other updates only need to lock the - specific distribution instance. + The explicit distribution.base_path lock protects the domain-wide base_path invariant. + The older domain-scoped distributions lock remains shared so tasks queued before an upgrade + still overlap safely with new tasks. """ - domain_distributions = f"pdrn:{get_domain().pulp_id}:distributions" + distribution_base_path = f"pdrn:{get_domain().pulp_id}:distribution.base_path" if instance is None: - return [domain_distributions] + return [distribution_base_path] if getattr(self, "action", "") == "destroy": - return [instance, domain_distributions] + return [instance] request_data = getattr(getattr(self, "request", None), "data", {}) requested_base_path = request_data.get("base_path", instance.base_path) if requested_base_path == instance.base_path: return [instance] - return [instance, domain_distributions] + return [instance, distribution_base_path] + + def async_shared_resources(self, instance): + """ + Keep the legacy domain-scoped distribution lock shared for upgrade compatibility. + """ + return [f"pdrn:{get_domain().pulp_id}:distributions"] class ListDistributionViewSet(BaseDistributionViewSet, mixins.ListModelMixin): @@ -585,11 +598,75 @@ class DistributionViewSet( LabelsMixin, ): """ - Provides read and list methods plus asynchronous CUD methods that reserve the narrowest safe - distribution locks, only taking the domain-wide lock when base_path overlap validation or - base_path release needs it. + Provides read and list methods plus asynchronous CUD methods that reserve per-distribution + locks, an explicit base_path lock when needed, and the legacy domain-wide distributions lock in + shared mode for upgrade compatibility. """ + @extend_schema( + description="Trigger an asynchronous create task", + responses={202: AsyncOperationResponseSerializer}, + ) + def create(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + app_label = self.queryset.model._meta.app_label + task = dispatch( + tasks.base.general_create, + exclusive_resources=self.async_reserved_resources(None), + shared_resources=self.async_shared_resources(None), + args=(app_label, serializer.__class__.__name__), + kwargs={"data": request.data}, + ) + return OperationPostponedResponse(task, request) + + @extend_schema( + description="Update the entity and trigger an asynchronous task if necessary", + responses={200: InheritSerializer, 202: AsyncOperationResponseSerializer}, + ) + def update(self, request, pk, **kwargs): + partial = kwargs.pop("partial", False) + instance = self.get_object() + serializer = self.get_serializer(instance, data=request.data, partial=partial) + serializer.is_valid(raise_exception=True) + + if all(getattr(instance, key) == value for key, value in serializer.validated_data.items()): + return Response(serializer.data) + + task = dispatch( + tasks.base.ageneral_update, + exclusive_resources=self.async_reserved_resources(instance), + shared_resources=self.async_shared_resources(instance), + args=(pk, instance._meta.app_label, serializer.__class__.__name__), + kwargs={"data": request.data, "partial": partial}, + immediate=self.ALLOW_NON_BLOCKING_UPDATE, + ) + return OperationPostponedResponse(task, request) + + @extend_schema( + description="Update the entity partially and trigger an asynchronous task if necessary", + responses={200: InheritSerializer, 202: AsyncOperationResponseSerializer}, + ) + def partial_update(self, request, *args, **kwargs): + kwargs["partial"] = True + return self.update(request, *args, **kwargs) + + @extend_schema( + description="Trigger an asynchronous delete task", + responses={202: AsyncOperationResponseSerializer}, + ) + def destroy(self, request, pk, **kwargs): + instance = self.get_object() + serializer = self.get_serializer(instance) + task = dispatch( + tasks.base.ageneral_delete, + exclusive_resources=self.async_reserved_resources(instance), + shared_resources=self.async_shared_resources(instance), + args=(pk, instance._meta.app_label, serializer.__class__.__name__), + immediate=self.ALLOW_NON_BLOCKING_DELETE, + ) + return OperationPostponedResponse(task, request) + class ArtifactDistributionViewSet(ReadOnlyDistributionViewSet): """ diff --git a/pulpcore/tests/functional/api/using_plugin/test_distributions.py b/pulpcore/tests/functional/api/using_plugin/test_distributions.py index 9e724638c8c..5d7f57c1e59 100644 --- a/pulpcore/tests/functional/api/using_plugin/test_distributions.py +++ b/pulpcore/tests/functional/api/using_plugin/test_distributions.py @@ -172,14 +172,32 @@ def test_distribution_update_task_reservations( file_bindings, monitor_task, ): + def has_shared_distributions_lock(task): + return any( + resource.startswith("shared:") and resource.endswith(":distributions") + for resource in task.reserved_resources_record + ) + + def has_exclusive_distributions_lock(task): + return any( + not resource.startswith("shared:") and resource.endswith(":distributions") + for resource in task.reserved_resources_record + ) + + def has_base_path_lock(task): + return any( + not resource.startswith("shared:") and resource.endswith(":distribution.base_path") + for resource in task.reserved_resources_record + ) + create_task = monitor_task( file_bindings.DistributionsFileApi.create( {"name": str(uuid4()), "base_path": str(uuid4())} ).task ) - assert any( - resource.endswith(":distributions") for resource in create_task.reserved_resources_record - ) + assert has_base_path_lock(create_task) + assert has_shared_distributions_lock(create_task) + assert not has_exclusive_distributions_lock(create_task) distribution = file_bindings.DistributionsFileApi.read(create_task.created_resources[0]) assert distribution.prn not in create_task.reserved_resources_record @@ -190,10 +208,9 @@ def test_distribution_update_task_reservations( ).task ) assert distribution.prn in no_base_path_update_task.reserved_resources_record - assert not any( - resource.endswith(":distributions") - for resource in no_base_path_update_task.reserved_resources_record - ) + assert not has_base_path_lock(no_base_path_update_task) + assert has_shared_distributions_lock(no_base_path_update_task) + assert not has_exclusive_distributions_lock(no_base_path_update_task) unchanged_base_path_update_task = monitor_task( file_bindings.DistributionsFileApi.partial_update( @@ -202,10 +219,9 @@ def test_distribution_update_task_reservations( ).task ) assert distribution.prn in unchanged_base_path_update_task.reserved_resources_record - assert not any( - resource.endswith(":distributions") - for resource in unchanged_base_path_update_task.reserved_resources_record - ) + assert not has_base_path_lock(unchanged_base_path_update_task) + assert has_shared_distributions_lock(unchanged_base_path_update_task) + assert not has_exclusive_distributions_lock(unchanged_base_path_update_task) base_path_update_task = monitor_task( file_bindings.DistributionsFileApi.partial_update( @@ -213,18 +229,17 @@ def test_distribution_update_task_reservations( ).task ) assert distribution.prn in base_path_update_task.reserved_resources_record - assert any( - resource.endswith(":distributions") - for resource in base_path_update_task.reserved_resources_record - ) + assert has_base_path_lock(base_path_update_task) + assert has_shared_distributions_lock(base_path_update_task) + assert not has_exclusive_distributions_lock(base_path_update_task) delete_task = monitor_task( file_bindings.DistributionsFileApi.delete(distribution.pulp_href).task ) assert distribution.prn in delete_task.reserved_resources_record - assert any( - resource.endswith(":distributions") for resource in delete_task.reserved_resources_record - ) + assert not has_base_path_lock(delete_task) + assert has_shared_distributions_lock(delete_task) + assert not has_exclusive_distributions_lock(delete_task) @pytest.mark.parallel From af924fc5c4837f123d44eca59b8553bdd0a2d528 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20M=C3=A9ndez=20Hern=C3=A1ndez?= Date: Thu, 6 Aug 2026 19:32:40 +0200 Subject: [PATCH 3/3] Fix CI for distribution lock rework Add the issue-numbered changelog fragment required by the existing Fixes trailers on this branch and fix the publication viewset import ordering expected by lint. Fixes: #3322 Co-authored-by: Cursor --- CHANGES/3322.bugfix | 1 + pulpcore/app/viewsets/publication.py | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) create mode 100644 CHANGES/3322.bugfix diff --git a/CHANGES/3322.bugfix b/CHANGES/3322.bugfix new file mode 100644 index 00000000000..a6f1876cf2e --- /dev/null +++ b/CHANGES/3322.bugfix @@ -0,0 +1 @@ +Reduced lock contention for distribution updates that leave `base_path` unchanged. diff --git a/pulpcore/app/viewsets/publication.py b/pulpcore/app/viewsets/publication.py index 02ba8772855..d2bc7d79063 100644 --- a/pulpcore/app/viewsets/publication.py +++ b/pulpcore/app/viewsets/publication.py @@ -20,8 +20,8 @@ from pulpcore.app.models.publication import CompositeContentGuard from pulpcore.app.response import OperationPostponedResponse from pulpcore.app.serializers import ( - AsyncOperationResponseSerializer, ArtifactDistributionSerializer, + AsyncOperationResponseSerializer, ContentGuardSerializer, ContentRedirectContentGuardSerializer, DistributionSerializer,