Narrow distribution task locks for unchanged base_path

CHANGES/7896.bugfix

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

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

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,