Reduce likelihood of collection deadlocks

Fix an issue where concurrent calls to transfer_from() in Prompt Processing were causing Postgres to abort with a deadlock error.

We now register collection names in alphabetical order when multiple collections are registered in a single transaction.  Because `ButlerCollections.register()` inserts rows into a table with a unique index, Postgres takes a lock when the row is inserted and holds it until the end of the transaction.  If there are two processes with concurrent transactions attempting to create the same collections, it is possible for the locks to be acquired in different orders, causing a deadlock.

python/lsst/daf/butler/_butler_collections.py

@@ -390,8 +390,11 @@ def register(self, name: str, type: CollectionType = CollectionType.RUN, doc: st
 
         Notes
         -----
-        This method cannot be called within transactions, as it needs to be
-        able to perform its own transaction to be concurrent
+        Avoid calling this method multiple times within a `Butler.transaction`.
+        If concurrent processes register the same collection names, they may
+        block each other until the end of the transaction and in some cases the
+        database will be required to abort one of the transactions to prevent
+        deadlock.
         """
         raise NotImplementedError()
 

python/lsst/daf/butler/direct_butler/_direct_butler.py

@@ -46,7 +46,7 @@
 from collections import Counter, defaultdict
 from collections.abc import Iterable, Iterator, MutableMapping, Sequence
 from types import EllipsisType
-from typing import TYPE_CHECKING, Any, ClassVar, TextIO, cast
+from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple, TextIO, cast
 
 from deprecated.sphinx import deprecated
 from lsst.resources import ResourcePath, ResourcePathExpression
@@ -1557,8 +1557,13 @@ def ingest_zip(self, zip_file: ResourcePathExpression, transfer: str = "auto") -
 
         # Ingest doesn't create the RUN collections so we have to do that
         # here.
+        #
+        # Sort by run collection name to ensure Postgres takes locks in the
+        # same order between different processes, to mitigate an issue
+        # where Postgres can deadlock due to the unique index on collection
+        # name. (See DM-47543).
         runs = {ref.run for ref in refs}
-        for run in runs:
+        for run in sorted(runs):
             registered = self.collections.register(run)
             if registered:
                 _LOG.verbose("Created RUN collection %s as part of zip ingest", run)
@@ -1967,9 +1972,9 @@ def transfer_from(
         # Importing requires that we group the refs by dataset type and run
         # before doing the import.
         source_dataset_types = set()
-        grouped_refs = defaultdict(list)
+        grouped_refs: defaultdict[_RefGroup, list[DatasetRef]] = defaultdict(list)
         for ref in source_refs:
-            grouped_refs[ref.datasetType, ref.run].append(ref)
+            grouped_refs[_RefGroup(ref.datasetType.name, ref.run)].append(ref)
             source_dataset_types.add(ref.datasetType)
 
         # Check to see if the dataset type in the source butler has
@@ -2079,8 +2084,14 @@ def transfer_from(
                     self._registry.insertDimensionData(element, *records, skip_existing=True)
 
             n_imported = 0
+
+            # Sort by run collection name to ensure Postgres takes locks in the
+            # same order between different processes, to mitigate an issue
+            # where Postgres can deadlock due to the unique index on collection
+            # name. (See DM-47543).
+            groups = sorted(grouped_refs.items(), key=lambda item: item[0].run)
             for (datasetType, run), refs_to_import in progress.iter_item_chunks(
-                grouped_refs.items(), desc="Importing to registry by run and dataset type"
+                groups, desc="Importing to registry by run and dataset type"
             ):
                 if run not in handled_collections:
                     # May need to create output collection. If source butler
@@ -2101,7 +2112,7 @@ def transfer_from(
                     "Importing %d ref%s of dataset type %s into run %s",
                     n_refs,
                     "" if n_refs == 1 else "s",
-                    datasetType.name,
+                    datasetType,
                     run,
                 )
 
@@ -2351,3 +2362,12 @@ def _preload_cache(self) -> None:
     """Shim object to provide a legacy public interface for querying via the
     the ``registry`` property.
     """
+
+
+class _RefGroup(NamedTuple):
+    """Key identifying a batch of DatasetRefs to be inserted in
+    `Butler.transfer_from`.
+    """
+
+    dataset_type: str
+    run: str