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/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/base.py b/pulpcore/app/viewsets/base.py
index 9f302103c5f..0dadf658778 100644
--- a/pulpcore/app/viewsets/base.py
+++ b/pulpcore/app/viewsets/base.py
@@ -451,6 +451,14 @@ def async_reserved_resources(self, instance, **kwargs):
         ).format(self.__class__.__name__)
         return [instance]
 
+    def async_shared_resources(self, instance, **kwargs):
+        """
+        Return shared resources to reserve for the task created by the Async...Mixins.
+
+        This default implementation does not add any shared reservations.
+        """
+        return []
+
 
 class AsyncCreateMixin:
     """
@@ -472,6 +480,7 @@ def create(self, request, *args, **kwargs):
         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=task_kwargs,
         )
@@ -506,6 +515,7 @@ def update(self, request, pk, **kwargs):
             task = dispatch(
                 tasks.base.ageneral_update,
                 exclusive_resources=self.async_reserved_resources(instance),
+                shared_resources=self.async_shared_resources(instance),
                 args=(pk, app_label, serializer.__class__.__name__),
                 kwargs=task_kwargs,
                 immediate=self.ALLOW_NON_BLOCKING_UPDATE,
@@ -542,6 +552,7 @@ def destroy(self, request, pk, **kwargs):
         task = dispatch(
             tasks.base.ageneral_delete,
             exclusive_resources=self.async_reserved_resources(instance),
+            shared_resources=self.async_shared_resources(instance),
             args=(pk, app_label, serializer.__class__.__name__),
             immediate=self.ALLOW_NON_BLOCKING_DELETE,
         )
diff --git a/pulpcore/app/viewsets/publication.py b/pulpcore/app/viewsets/publication.py
index ae7e38a8e6d..d98a254fd62 100644
--- a/pulpcore/app/viewsets/publication.py
+++ b/pulpcore/app/viewsets/publication.py
@@ -527,7 +527,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,9 +591,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 per-distribution
+    locks, an explicit base_path lock when needed, and the legacy domain-wide distributions lock in
+    shared mode for upgrade compatibility.
     """
 
 
diff --git a/pulpcore/tests/functional/api/using_plugin/test_distributions.py b/pulpcore/tests/functional/api/using_plugin/test_distributions.py
index 1870c86e462..5d7f57c1e59 100644
--- a/pulpcore/tests/functional/api/using_plugin/test_distributions.py
+++ b/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,