fix(retain): globalise memory_links lock order on the insert path (#3396)

The deadlock #2570 targeted still fired a few times a day because the
insert-side lock ordering was only partial:

1. _bulk_insert_links sorted on (from, to) — two of the four columns in
   the unique index (from, to, link_type, COALESCE(entity_id, nil)). A
   temporal and a semantic edge on the same pair compared equal, so a
   stable sort left them in input order and concurrent inserts could take
   the two index entries in opposite orders.

2. That order also disagreed with chunk_storage.delete_chunks_by_ids,
   which normalises direction via LEAST/GREATEST. Two different total
   orders can still cycle.

Sort both paths on one canonical key — the full, direction-normalised
unique key — by extracting _lock_order_key and pointing the insert sort
at it. The delete side already uses exactly this order and is unchanged.

hindsight-api-slim/hindsight_api/engine/retain/link_utils.py

@@ -79,6 +79,26 @@ def _cap_links_per_unit(links: list[tuple], max_per_unit: int = MAX_TEMPORAL_LIN
     return result
 
 
+def _lock_order_key(lnk: tuple) -> tuple[str, str, str, str]:
+    """Canonical lock-order key for a link row, shared by every writer.
+
+    Mirrors the total order that ``chunk_storage.delete_chunks_by_ids`` uses when
+    it locks ``memory_links`` before a cascade delete:
+
+        (LEAST(from, to), GREATEST(from, to), link_type, COALESCE(entity_id, nil))
+
+    Direction is normalised so ``(A, B)`` and ``(B, A)`` sort adjacent, and the
+    key covers the full unique index — including ``link_type`` and ``entity_id``
+    — so two edges sharing a ``(from, to)`` pair can't be locked in opposite
+    orders by concurrent inserts. UUID string ordering matches the DB's ``uuid``
+    ordering because the ids are canonical lowercase-hex form.
+    """
+    a, b = str(lnk[0]), str(lnk[1])
+    low, high = (a, b) if a <= b else (b, a)
+    entity = str(lnk[4]) if lnk[4] is not None else _NIL_ENTITY_UUID
+    return (low, high, str(lnk[2]), entity)
+
+
 async def _bulk_insert_links(
     conn,
     links: list[tuple],
@@ -89,8 +109,9 @@ async def _bulk_insert_links(
 ) -> None:
     """Bulk-insert links using sorted INSERT FROM unnest().
 
-    Sorting by (from_unit_id, to_unit_id) ensures all concurrent transactions
-    acquire index locks in the same order, eliminating circular-wait deadlocks.
+    Sorting on the full, direction-normalised unique key ensures all concurrent
+    writers — inserts and deletes alike — acquire index locks in the same order,
+    eliminating circular-wait deadlocks. See :func:`_lock_order_key`.
 
     Args:
         conn: Database connection (must be inside a transaction).
@@ -106,9 +127,9 @@ async def _bulk_insert_links(
     if not links:
         return
 
-    # Sort by (from_unit_id, to_unit_id) to guarantee consistent lock ordering
-    # across concurrent transactions — prevents deadlocks.
-    sorted_links = sorted(links, key=lambda lnk: (str(lnk[0]), str(lnk[1])))
+    # Sort on the canonical lock-order key so every concurrent writer takes the
+    # index locks in the same order — prevents circular-wait deadlocks.
+    sorted_links = sorted(links, key=_lock_order_key)
 
     exists_clause = ""
     if not skip_exists_check:

hindsight-api-slim/tests/test_link_utils.py

@@ -1,17 +1,20 @@
 """Tests for link_utils datetime handling, temporal link computation, and semantic link splitting."""
 
-import numpy as np
-import pytest
 from datetime import datetime, timezone
 from unittest.mock import AsyncMock, MagicMock
 
+import numpy as np
+import pytest
+
 from hindsight_api.config import DEFAULT_SEMANTIC_LINK_MIN_SIMILARITY, clear_config_cache
 from hindsight_api.engine.retain.link_utils import (
-    _normalize_datetime,
+    _NIL_ENTITY_UUID,
+    MAX_TEMPORAL_LINKS_PER_UNIT,
     _cap_links_per_unit,
+    _lock_order_key,
+    _normalize_datetime,
     compute_semantic_links_ann,
     compute_semantic_links_within_batch,
-    MAX_TEMPORAL_LINKS_PER_UNIT,
 )
 
 
@@ -228,6 +231,60 @@ class TestComputeSemanticLinksWithinBatch:
             assert entity_id is None
 
 
+class TestLockOrderKey:
+    """The insert-side sort key must reproduce the same total order that
+    ``chunk_storage.delete_chunks_by_ids`` locks ``memory_links`` in, so every
+    concurrent writer takes index locks in one global order (issue #3396).
+
+    Delete-side order:
+        (LEAST(from, to), GREATEST(from, to), link_type, COALESCE(entity_id, nil))
+    """
+
+    A = "00000000-0000-0000-0000-00000000000a"
+    B = "00000000-0000-0000-0000-00000000000b"
+
+    def test_direction_is_normalised(self):
+        """(A, B) and (B, A) collapse to the same first two components so
+        opposite-direction edges sort adjacent, matching LEAST/GREATEST."""
+        fwd = _lock_order_key((self.A, self.B, "temporal", 1.0, None))
+        rev = _lock_order_key((self.B, self.A, "temporal", 1.0, None))
+        assert fwd[:2] == rev[:2] == (self.A, self.B)
+
+    def test_link_type_disambiguates_same_pair(self):
+        """Two edges sharing a (from, to) pair but differing in link_type must
+        get distinct, deterministic keys — the gap that let insert-vs-insert
+        deadlock (mechanism 1 in the issue)."""
+        semantic = _lock_order_key((self.A, self.B, "semantic", 0.9, None))
+        temporal = _lock_order_key((self.A, self.B, "temporal", 1.0, None))
+        assert semantic != temporal
+        assert semantic < temporal  # "semantic" < "temporal"
+
+    def test_none_entity_id_uses_nil_uuid(self):
+        """COALESCE(entity_id, nil) on the delete side ⇒ None maps to the nil
+        UUID here, not the string 'None'."""
+        key = _lock_order_key((self.A, self.B, "temporal", 1.0, None))
+        assert key[3] == _NIL_ENTITY_UUID
+
+    def test_matches_delete_total_order(self):
+        """Sorting a mixed batch by the key reproduces the delete's ORDER BY."""
+        c = "00000000-0000-0000-0000-00000000000c"
+        links = [
+            (self.B, self.A, "temporal", 1.0, None),
+            (self.A, self.B, "semantic", 0.9, None),
+            (self.A, c, "temporal", 1.0, None),
+            (self.A, self.B, "temporal", 1.0, None),
+        ]
+        ordered = sorted(links, key=_lock_order_key)
+
+        def canonical(lnk):
+            a, b = str(lnk[0]), str(lnk[1])
+            low, high = (a, b) if a <= b else (b, a)
+            entity = str(lnk[4]) if lnk[4] is not None else _NIL_ENTITY_UUID
+            return (low, high, str(lnk[2]), entity)
+
+        assert [canonical(lnk) for lnk in ordered] == sorted(canonical(lnk) for lnk in links)
+
+
 class TestComputeSemanticLinksAnnPgBouncerSafety:
     """Regression tests ensuring compute_semantic_links_ann stays in a single
     transaction so that the `_ann_seeds` temp table remains visible when the