perf+fix(graph-maintenance): catch the #2529 sweep deadlock in continuous perf, and drive dropped passes to zero

hindsight-api-slim/hindsight_api/engine/db/ops_oracle.py

@@ -329,6 +329,12 @@ async def prune_stale_cooccurrences(
         entities_table: str,
         bank_id: str,
     ) -> int:
+        # NB: the Postgres path additionally selects victims FOR UPDATE in sorted
+        # (entity_id_1, entity_id_2) order to prevent the #2529 deadlock against
+        # retain's sorted cooccurrence upsert. Oracle's DELETE can't carry that
+        # ordered-lock CTE the same way, so here we rely on the Pass 2/3 retry
+        # wrap in run_graph_maintenance_job (retry_with_backoff is ORA-00060
+        # deadlock-aware) to recover instead. Deliberate dialect asymmetry.
         deleted = await conn.execute(
             f"""
             DELETE FROM {ec_table}

hindsight-api-slim/hindsight_api/engine/db/ops_postgresql.py

@@ -390,19 +390,41 @@ async def prune_stale_cooccurrences(
         # Scope by joining through entities.bank_id (entity_cooccurrences itself
         # has no bank_id column — entities don't span banks, so scoping via
         # entity_id_1 is sufficient).
+        #
+        # Ordered locking (deadlock avoidance, #2529): retain's concurrent
+        # cooccurrence upsert (entity_resolver._flush_pending) locks rows in
+        # sorted (entity_id_1, entity_id_2) order — sorted specifically to give
+        # every writer one consistent lock-acquisition order. A plain
+        # `DELETE ... USING` scans/locks in whatever order the join plan picks,
+        # so it could lock the same rows in the opposite order and cycle. We
+        # instead select the victims in that same sorted order `FOR UPDATE`
+        # first — the locking clause materialises the CTE and places LockRows
+        # above the Sort, so locks are acquired ascending, matching the upsert —
+        # then delete the already-locked rows. Same order on both sides ⇒ no
+        # cycle (the deadlock is prevented, not merely retried). The Pass 2/3
+        # retry wrap in run_graph_maintenance_job stays as a backstop for the
+        # residual paths (FK cascade from prune_orphan_entities, Oracle).
         result = await conn.execute(
             f"""
+            WITH victims AS (
+                SELECT c.entity_id_1, c.entity_id_2
+                FROM {ec_table} c
+                JOIN {entities_table} e ON e.id = c.entity_id_1
+                WHERE e.bank_id = $1
+                  AND NOT EXISTS (
+                      SELECT 1
+                      FROM {ue_table} u1
+                      JOIN {ue_table} u2 ON u1.unit_id = u2.unit_id
+                      WHERE u1.entity_id = c.entity_id_1
+                        AND u2.entity_id = c.entity_id_2
+                  )
+                ORDER BY c.entity_id_1, c.entity_id_2
+                FOR UPDATE OF c
+            )
             DELETE FROM {ec_table} c
-            USING {entities_table} e
-            WHERE e.id = c.entity_id_1
-              AND e.bank_id = $1
-              AND NOT EXISTS (
-                  SELECT 1
-                  FROM {ue_table} u1
-                  JOIN {ue_table} u2 ON u1.unit_id = u2.unit_id
-                  WHERE u1.entity_id = c.entity_id_1
-                    AND u2.entity_id = c.entity_id_2
-              )
+            USING victims v
+            WHERE c.entity_id_1 = v.entity_id_1
+              AND c.entity_id_2 = v.entity_id_2
             """,
             bank_id,
         )

hindsight-api-slim/hindsight_api/engine/db_utils.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import logging
+import random
 import time
 from collections.abc import AsyncIterator
 from contextlib import AsyncExitStack, asynccontextmanager
@@ -16,6 +17,20 @@
 DEFAULT_BASE_DELAY = 0.5  # seconds
 DEFAULT_MAX_DELAY = 5.0  # seconds
 
+
+def _backoff_delay(attempt: int, base_delay: float, max_delay: float) -> float:
+    """Exponential backoff with equal jitter.
+
+    Deterministic backoff makes concurrent retriers wake in lock-step and
+    re-collide on the very same rows, re-triggering the deadlock they just
+    backed off from. "Equal jitter" — half the window fixed, half random —
+    keeps a floor (so we don't hot-spin) while decorrelating the wake-ups, so
+    two contenders that deadlocked together are very unlikely to retry in sync.
+    """
+    ceil = min(base_delay * (2**attempt), max_delay)
+    return ceil / 2 + random.uniform(0, ceil / 2)
+
+
 # Retryable exception types (checked by class name to avoid hard imports)
 _RETRYABLE_EXCEPTION_NAMES = frozenset(
     {
@@ -78,7 +93,7 @@ async def retry_with_backoff(
                 raise
             last_exception = e
             if attempt < max_retries:
-                delay = min(base_delay * (2**attempt), max_delay)
+                delay = _backoff_delay(attempt, base_delay, max_delay)
                 if type(e).__name__ == "DeadlockDetectedError" or _is_oracle_deadlock(e):
                     logger.warning(
                         "Deadlock detected during parallel document processing — "
@@ -136,7 +151,7 @@ async def acquire_with_retry(backend_or_pool: Any, max_retries: int = DEFAULT_MA
                     if not _is_retryable(e):
                         raise
                     if attempt < max_retries:
-                        delay = min(DEFAULT_BASE_DELAY * (2**attempt), DEFAULT_MAX_DELAY)
+                        delay = _backoff_delay(attempt, DEFAULT_BASE_DELAY, DEFAULT_MAX_DELAY)
                         logger.warning(
                             f"Database acquire failed (attempt {attempt + 1}/{max_retries + 1}): {e}. "
                             f"Retrying in {delay:.1f}s..."

hindsight-api-slim/hindsight_api/engine/graph_maintenance.py

@@ -66,6 +66,20 @@
 # under 1s.
 _DRAIN_BATCH_SIZE = 50
 
+# Retry budget for the idempotent Pass 2/3 entity/cooccurrence sweep. Higher
+# than db_utils' default (3) because the sweep has no client waiting on it and
+# is safe to rerun, so we'd rather spend a longer jittered-backoff tail than
+# drop a maintenance pass and leak stale graph rows (see run_graph_maintenance_job).
+_SWEEP_MAX_RETRIES = 8
+
+
+@dataclass
+class _SweepCounts:
+    """Prune counts returned by the Pass 2/3 sweep (avoids a bare tuple return)."""
+
+    orphan_entities_pruned: int
+    stale_cooccurrences_pruned: int
+
 
 @dataclass
 class JobResult:
@@ -203,27 +217,51 @@ async def run_graph_maintenance_job(
 
     # --- Pass 2 & 3: entity / cooccurrence sweeps ---
     # Bank-wide single-statement deletes. Cheap when there's nothing to do.
+    #
+    # Unlike Pass 1's queue claim, these DELETEs aren't protected by any
+    # consistent lock-ordering guarantee: prune_stale_cooccurrences scans
+    # entity_cooccurrences via a join/NOT EXISTS plan, while retain's
+    # concurrent cooccurrence upserts (entity_resolver._flush_pending) lock
+    # the same rows in sorted (entity_id_1, entity_id_2) order. When a sweep
+    # and a concurrent upsert touch overlapping rows in opposite orders,
+    # Postgres detects a genuine circular wait and aborts one side with
+    # DeadlockDetectedError. Both prunes are idempotent bank-wide sweeps —
+    # rerunning only deletes what's still stale — so retrying the whole
+    # transaction on deadlock is safe.
+    from .db_utils import retry_with_backoff
     from .memory_engine import acquire_with_retry
 
-    async with acquire_with_retry(backend) as conn:
-        async with conn.transaction():
-            result.orphan_entities_pruned = await ops.prune_orphan_entities(
-                conn,
-                fq_table("entities"),
-                fq_table("unit_entities"),
-                bank_id,
-            )
-            # The orphan prune above cascades cooccurrences via FK. The
-            # explicit cooccurrence pass below catches the *stale-count*
-            # case: both entities still exist but no current unit witnesses
-            # them together.
-            result.stale_cooccurrences_pruned = await ops.prune_stale_cooccurrences(
-                conn,
-                fq_table("entity_cooccurrences"),
-                fq_table("unit_entities"),
-                fq_table("entities"),
-                bank_id,
-            )
+    async def _run_sweep() -> _SweepCounts:
+        async with acquire_with_retry(backend) as conn:
+            async with conn.transaction():
+                orphan_pruned = await ops.prune_orphan_entities(
+                    conn,
+                    fq_table("entities"),
+                    fq_table("unit_entities"),
+                    bank_id,
+                )
+                # The orphan prune above cascades cooccurrences via FK. The
+                # explicit cooccurrence pass below catches the *stale-count*
+                # case: both entities still exist but no current unit
+                # witnesses them together.
+                stale_pruned = await ops.prune_stale_cooccurrences(
+                    conn,
+                    fq_table("entity_cooccurrences"),
+                    fq_table("unit_entities"),
+                    fq_table("entities"),
+                    bank_id,
+                )
+                return _SweepCounts(orphan_entities_pruned=orphan_pruned, stale_cooccurrences_pruned=stale_pruned)
+
+    # A larger retry budget than the default (3): this is idempotent background
+    # maintenance with no client waiting on it, so a longer retry tail costs
+    # nothing, whereas a dropped sweep silently leaks orphan entities / stale
+    # cooccurrences until the next run. With jittered backoff a single sweep
+    # contending against continuous retain upserts effectively never exhausts
+    # this budget (each retry independently clears with high probability).
+    sweep = await retry_with_backoff(_run_sweep, max_retries=_SWEEP_MAX_RETRIES)
+    result.orphan_entities_pruned = sweep.orphan_entities_pruned
+    result.stale_cooccurrences_pruned = sweep.stale_cooccurrences_pruned
 
     elapsed = time.time() - job_start
     logger.info(

hindsight-api-slim/tests/test_db_utils.py

@@ -15,7 +15,7 @@
 
 import pytest
 
-from hindsight_api.engine.db_utils import acquire_with_retry
+from hindsight_api.engine.db_utils import _backoff_delay, acquire_with_retry
 
 
 class _FakeConnection:
@@ -77,3 +77,26 @@ async def test_retryable_user_code_exception_propagates_unchanged():
     assert backend.acquired == 1
     assert backend.last_conn is not None
     assert backend.last_conn.released == 1, "connection must be released exactly once"
+
+
+def test_backoff_delay_is_jittered_and_bounded():
+    """Equal-jitter backoff stays in [ceil/2, ceil] and never exceeds max_delay.
+
+    The jitter exists so concurrent deadlock retriers don't wake in lock-step
+    and re-collide (see run_graph_maintenance_job's Pass 2/3 sweep). It must
+    still keep a floor (no hot-spin) and honour the max_delay cap once the
+    exponential term saturates.
+    """
+    base, max_delay = 0.5, 5.0
+
+    # Below saturation: ceil = base * 2**attempt, jitter within [ceil/2, ceil].
+    for attempt in range(3):
+        ceil = base * (2**attempt)
+        samples = [_backoff_delay(attempt, base, max_delay) for _ in range(200)]
+        assert all(ceil / 2 <= d <= ceil for d in samples)
+        # Actually jittered — not a constant.
+        assert len(set(samples)) > 1
+
+    # At/after saturation the cap holds: every sample <= max_delay.
+    saturated = [_backoff_delay(20, base, max_delay) for _ in range(200)]
+    assert all(max_delay / 2 <= d <= max_delay for d in saturated)