Base path db constaint

pulpcore/app/migrations/0157_distribution_base_path_constraint.py

@@ -0,0 +1,104 @@
+# Generated by Django 5.2.17 on 2026-08-11 10:34
+
+from django.db import migrations
+
+
+ADD_TRIGGER = r"""
+CREATE OR REPLACE FUNCTION path_prefixes("path" text)
+RETURNS text[] AS $$
+DECLARE
+        segment text;
+        segments text[] := string_to_array(path, '/');
+        prefix text := '';
+        prefixes text[];
+BEGIN
+        -- Return directory prefixes of a path.
+        -- 'a/b/c' -> {'a', 'a/b'}
+        prefix := segments[1];
+        FOREACH segment IN ARRAY segments[2:]
+        LOOP
+                prefixes := prefixes || prefix;
+                prefix := prefix || '/' || segment;
+        END LOOP;
+        RETURN prefixes;
+END;
+$$ LANGUAGE plpgsql
+IMMUTABLE
+RETURNS NULL ON NULL INPUT
+PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION "check_core_distribution_base_path_prefix_free" ()
+RETURNS TRIGGER AS $$
+DECLARE
+        base_path_slash text := new.base_path || '/';
+BEGIN
+        -- Check that no base_path is a prefix of another.
+        -- The normalization of "/" ensures that a simple string comparison actually suffice.
+        -- WARNING: This check alone does not ensure uniqueness.
+
+        -- ^@ seems to only use an index when the indexed expression is on the left (pg18).
+        PERFORM 1 FROM "core_distribution"
+                WHERE
+                        "pulp_id" != new."pulp_id"
+                        AND
+                        "pulp_domain_id" = new."pulp_domain_id"
+                        AND
+                        "base_path" ^@ base_path_slash
+                LIMIT 1;
+        IF FOUND THEN
+                RAISE EXCEPTION '"%" is the prefix of an existing base_path.', new."base_path"
+                    USING ERRCODE = 'exclusion_violation';
+        END IF;
+
+        -- This variant however uses the existing uniqueness index.
+        PERFORM 1 FROM "core_distribution"
+                WHERE
+                        "pulp_id" != new."pulp_id"
+                        AND
+                        "pulp_domain_id" = new."pulp_domain_id"
+                        AND
+                        "base_path" = ANY(path_prefixes(new."base_path"))
+                LIMIT 1;
+        IF FOUND THEN
+                RAISE EXCEPTION '"%" is prefixed by an existing base_path.', new."base_path"
+                    USING ERRCODE = 'exclusion_violation';
+        END IF;
+
+        RETURN new;
+END
+$$ LANGUAGE plpgsql;
+
+CREATE CONSTRAINT TRIGGER "core_distribution_insert_base_path_overlap_constraint"
+        AFTER INSERT
+        ON "core_distribution"
+        DEFERRABLE INITIALLY DEFERRED
+        FOR EACH ROW
+        WHEN (new."base_path" IS NOT NULL)
+        EXECUTE FUNCTION "check_core_distribution_base_path_prefix_free" ();
+
+CREATE CONSTRAINT TRIGGER "core_distribution_update_base_path_overlap_constraint"
+        AFTER UPDATE
+        ON "core_distribution"
+        DEFERRABLE INITIALLY DEFERRED
+        FOR EACH ROW
+        WHEN (new."base_path" IS NOT NULL AND new."base_path" != old."base_path")
+        EXECUTE FUNCTION "check_core_distribution_base_path_prefix_free" ();
+"""
+
+REMOVE_TRIGGER = r"""
+DROP TRIGGER IF EXISTS "core_distribution_update_base_path_overlap_constraint" ON "core_distribution";
+DROP TRIGGER IF EXISTS "core_distribution_insert_base_path_overlap_constraint" ON "core_distribution";
+DROP FUNCTION IF EXISTS "check_core_distribution_base_path_prefix_free";
+DROP FUNCTION IF EXISTS "path_prefixes";
+"""
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0156_alter_contentartifact_relative_path_and_more'),
+    ]
+
+    operations = [
+        migrations.RunSQL(sql=ADD_TRIGGER, reverse_sql=REMOVE_TRIGGER, elidable=False),
+    ]

pulpcore/app/serializers/publication.py

@@ -274,18 +274,15 @@ def validate_base_path(self, path):
             q |= Q(base_path=search)
 
         # look for any base paths that nest path
-        q |= Q(base_path__startswith="{}/".format(path))
+        q |= Q(base_path__startswith=f"{path}/")
         qs = models.Distribution.objects.filter(q & Q(pulp_domain=get_domain()))
 
         if self.instance is not None:
             qs = qs.exclude(pk=self.instance.pk)
 
-        match = qs.first()
-        if match:
+        if qs.exists():
             raise serializers.ValidationError(
-                detail={
-                    "base_path": _("Overlaps with existing distribution '{}'").format(match.name)
-                },
+                detail={"base_path": _("Overlaps with an existing distribution.")},
             )
 
         return path

pulpcore/app/viewsets/base.py

@@ -429,27 +429,20 @@ def async_reserved_resources(self, instance, **kwargs):
 
         This default implementation locks the instance being worked on.
 
-        .. note::
-
-          This does not work for [pulpcore.app.viewsets.AsyncCreateMixin][]
-          (as there is no instance). Classes using [pulpcore.app.viewsets.AsyncCreateMixin][]
-          must override this method.
-
         Args:
-            instance (django.models.Model): The instance that will be worked
+            instance (django.models.Model | None): The instance that will be worked
                 on by the task.
 
         Returns:
-            list/str: The resources to put in the task's reservation
+            list[django.models.Model | str]: The resources to put in the task's reservation
 
         Raises:
             AssertionError if instance is None (which happens for creation)
 
         """
-        assert instance is not None, (
-            "'{}' must not use the default `async_reserved_resources` method when using create."
-        ).format(self.__class__.__name__)
-        return [instance]
+        if instance is not None:
+            return [instance]
+        return []
 
     def async_shared_resources(self, instance, **kwargs):
         """
@@ -460,7 +453,7 @@ def async_shared_resources(self, instance, **kwargs):
         return []
 
 
-class AsyncCreateMixin:
+class AsyncCreateMixin(AsyncReservedObjectMixin):
     """
     Provides a create method that dispatches a task with reservation.
     """

pulpcore/app/viewsets/publication.py

@@ -25,7 +25,6 @@
     RBACContentGuardSerializer,
 )
 from pulpcore.app.serializers.publication import CompositeContentGuardSerializer
-from pulpcore.app.util import get_domain
 from pulpcore.app.viewsets import (
     AsyncCreateMixin,
     AsyncRemoveMixin,
@@ -526,34 +525,6 @@ def get_queryset(self):
             )
         return qs
 
-    def async_reserved_resources(self, instance):
-        """
-        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"]
-
 
 class ListDistributionViewSet(BaseDistributionViewSet, mixins.ListModelMixin):
     DEFAULT_ACCESS_POLICY = {

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

@@ -167,81 +167,6 @@ 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,

pulpcore/tests/unit/models/test_distribution.py

@@ -0,0 +1,40 @@
+import pytest
+from django.db import connection
+from django.db.utils import IntegrityError
+
+from pulpcore.app.models import Distribution
+
+# "SET CONSTRAINTS ALL IMMEDIATE" is automatically called by the fixtures teardown.
+# We only need to do it manually when we need the exception during the test.
+
+
+@pytest.mark.django_db
+class TestDistributionBasePathConstraint:
+    def test_must_be_unique(self):
+        Distribution(name="0", base_path="a").save()
+        with pytest.raises(IntegrityError, match="unique constraint"):
+            Distribution(name="1", base_path="a").save()
+
+    def test_can_share_a_prefix_with_another_base_path(self):
+        Distribution(name="0", base_path="a/a").save()
+        Distribution(name="1", base_path="a/b").save()
+
+    def test_cannot_be_the_prefix_of_another_base_path(self):
+        Distribution(name="0", base_path="a/a").save()
+        Distribution(name="1", base_path="a").save()
+        with pytest.raises(IntegrityError, match="prefix"):
+            with connection.cursor() as cursor:
+                cursor.execute("SET CONSTRAINTS ALL IMMEDIATE")
+
+    def test_cannot_contain_another_base_path_as_prefix(self):
+        Distribution(name="0", base_path="a").save()
+        Distribution(name="1", base_path="a/a").save()
+        with pytest.raises(IntegrityError, match="prefix"):
+            with connection.cursor() as cursor:
+                cursor.execute("SET CONSTRAINTS ALL IMMEDIATE")
+
+    def test_prefixes_are_checke_at_slash_boundaries(self):
+        Distribution(name="0", base_path="abc").save()
+        Distribution(name="1", base_path="ab").save()
+        Distribution(name="2", base_path="abcd").save()
+        Distribution(name="3", base_path="abcde/a").save()