Narrow distribution task locks for unchanged base_path

CHANGES/3322.bugfix

@@ -0,0 +1 @@
+Reduced lock contention for distribution updates that leave `base_path` unchanged.

CHANGES/7896.bugfix

@@ -0,0 +1 @@
+Reduced lock contention for distribution updates that leave `base_path` unchanged.

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,8 +18,10 @@
     Repository,
 )
 from pulpcore.app.models.publication import CompositeContentGuard
+from pulpcore.app.response import OperationPostponedResponse
 from pulpcore.app.serializers import (
     ArtifactDistributionSerializer,
+    AsyncOperationResponseSerializer,
     ContentGuardSerializer,
     ContentRedirectContentGuardSerializer,
     DistributionSerializer,
@@ -43,6 +48,8 @@
     WithContentInFilter,
 )
 from pulpcore.filters import BaseFilterSet
+from pulpcore.openapi import InheritSerializer
+from pulpcore.tasking.tasks import dispatch
 
 
 class RepositoryThroughVersionFilter(Filter):
@@ -527,7 +534,31 @@ def get_queryset(self):
         return qs
 
     def async_reserved_resources(self, instance):
-        """Return resource that locks all Distributions."""
+        """
+        Reserve safe distribution locks for async operations.
+
+        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.
+        """
+        distribution_base_path = f"pdrn:{get_domain().pulp_id}:distribution.base_path"
+        if instance is None:
+            return [distribution_base_path]
+
+        if getattr(self, "action", "") == "destroy":
+            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, 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"]
 
 
@@ -567,11 +598,75 @@ 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 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):
     """

pulpcore/tests/functional/api/using_plugin/test_distributions.py

@@ -167,6 +167,81 @@ 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,
+):
+    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 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
+
+    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 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(
+            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 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(
+            distribution.pulp_href, {"base_path": str(uuid4())}
+        ).task
+    )
+    assert distribution.prn 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 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
 def test_distribution_filtering(
     file_bindings,