Retry the subject purge once on a Postgres deadlock

server/api/subjects.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import structlog
 from fastapi import APIRouter, Depends, Query
+from sqlalchemy.exc import DBAPIError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from server.db import repositories as repo
@@ -13,6 +15,14 @@
 
 router = APIRouter(prefix="/v1/subjects", tags=["subjects"])
 
+logger = structlog.stdlib.get_logger()
+
+
+def _is_deadlock(exc: DBAPIError) -> bool:
+    """True when the DBAPI error wraps a Postgres deadlock (SQLSTATE 40P01)."""
+    orig = getattr(exc, "orig", None)
+    return "deadlock detected" in str(orig or exc)
+
 
 @router.get("", response_model=ListSubjectsResponse, summary="List known subjects")
 async def list_subjects(
@@ -39,23 +49,40 @@ async def delete_subject(
     tenant_id: str | None = Depends(get_tenant_id),
 ):
     """Permanently delete all episodes and memories for a subject. This is irreversible."""
-    ep_count = await repo.delete_episodes_by_subject(session, subject_id, tenant_id=tenant_id)
-    mem_count = await repo.delete_memories_by_subject(session, subject_id, tenant_id=tenant_id)
-    # "Permanently delete all subject data" must also reap the subject's
-    # resolutions and health-cache row — there is no FK cascade. Leaving them
-    # behind keeps open/resolved-session logic treating the deleted subject's
-    # sessions as live and lets a stale health-cache row suppress/forge alerts
-    # if the subject id is later reused.
-    await repo.delete_resolutions_by_subject(session, subject_id, tenant_id=tenant_id)
-    await repo.delete_health_cache_by_subject(session, subject_id, tenant_id=tenant_id)
-    # Phase 2: subject_entities is OUT-OF-BAND from memories (no FK
-    # since N entities can point at M memories, neither owns the other),
-    # so the cascade has to be explicit here too. Without this, a
-    # re-ingested subject would inherit stale entity rows pointing at
-    # memory_ids that no longer exist — Phase 3 retrieval would surface
-    # boost from ghost memories.
-    await repo.delete_entities_by_subject(session, subject_id, tenant_id=tenant_id)
-    await session.commit()
+    # Retried once on a Postgres deadlock (40P01): the multi-row DELETEs can
+    # cross lock order with another multi-row writer on the same subject's
+    # rows (a still-draining compile batch, an embedding backfill). Postgres
+    # aborts exactly one side, so a rollback-and-redo of the purge converges
+    # — the competing transaction has either finished or loses the rematch.
+    for attempt in (1, 2):
+        try:
+            ep_count = await repo.delete_episodes_by_subject(
+                session, subject_id, tenant_id=tenant_id
+            )
+            mem_count = await repo.delete_memories_by_subject(
+                session, subject_id, tenant_id=tenant_id
+            )
+            # "Permanently delete all subject data" must also reap the subject's
+            # resolutions and health-cache row — there is no FK cascade. Leaving
+            # them behind keeps open/resolved-session logic treating the deleted
+            # subject's sessions as live and lets a stale health-cache row
+            # suppress/forge alerts if the subject id is later reused.
+            await repo.delete_resolutions_by_subject(session, subject_id, tenant_id=tenant_id)
+            await repo.delete_health_cache_by_subject(session, subject_id, tenant_id=tenant_id)
+            # Phase 2: subject_entities is OUT-OF-BAND from memories (no FK
+            # since N entities can point at M memories, neither owns the other),
+            # so the cascade has to be explicit here too. Without this, a
+            # re-ingested subject would inherit stale entity rows pointing at
+            # memory_ids that no longer exist — Phase 3 retrieval would surface
+            # boost from ghost memories.
+            await repo.delete_entities_by_subject(session, subject_id, tenant_id=tenant_id)
+            await session.commit()
+            break
+        except DBAPIError as exc:
+            if attempt == 2 or not _is_deadlock(exc):
+                raise
+            await session.rollback()
+            logger.warning("subject_delete_deadlock_retry", subject_id=subject_id)
     # Only fire the webhook when something was actually deleted (issue #282):
     # deleting a missing subject (or the same subject twice) is a no-op, so a
     # subject.deleted event with zero counts would be a spurious deletion

server/db/repositories.py

@@ -660,6 +660,40 @@ async def search_memories_hybrid(
 # ---------------------------------------------------------------------------
 
 
+async def _append_entity_link_atomic(
+    session: AsyncSession, row: SubjectEntityRow, memory_id: uuid.UUID
+) -> None:
+    """Append memory_id to a row's linked_memory_ids as ONE guarded SQL
+    UPDATE, never an ORM read-modify-write. Two concurrent appenders that
+    both read the same committed array would each write back their own
+    copy and silently drop the other's link (lost update — caught by the
+    #384 concurrent-convergence test on CI). The UPDATE takes the row
+    lock and re-evaluates the append against the latest committed array,
+    so concurrent appends serialize; the ANY() guard keeps repeated
+    linkage of the same memory a no-op.
+    """
+    from sqlalchemy import any_, case, literal
+
+    mid = literal(memory_id, type_=SubjectEntityRow.id.type)
+    await session.execute(
+        update(SubjectEntityRow)
+        .where(SubjectEntityRow.id == row.id)
+        .values(
+            linked_memory_ids=case(
+                (
+                    mid == any_(SubjectEntityRow.linked_memory_ids),
+                    SubjectEntityRow.linked_memory_ids,
+                ),
+                else_=func.array_append(SubjectEntityRow.linked_memory_ids, mid),
+            ),
+            updated_at=func.now(),
+        )
+    )
+    # Refresh so the caller-visible ORM row reflects the post-append array
+    # instead of the stale pre-lock snapshot.
+    await session.refresh(row)
+
+
 async def upsert_entity_with_link(
     session: AsyncSession,
     *,
@@ -703,10 +737,7 @@ async def upsert_entity_with_link(
         exact_stmt = exact_stmt.where(SubjectEntityRow.tenant_id == tenant_id)
     exact = (await session.execute(exact_stmt)).scalar_one_or_none()
     if exact is not None:
-        if memory_id not in exact.linked_memory_ids:
-            # SQLAlchemy doesn't detect mutations to ARRAY columns
-            # in-place; rebuild + reassign so the update flushes.
-            exact.linked_memory_ids = [*exact.linked_memory_ids, memory_id]
+        await _append_entity_link_atomic(session, exact, memory_id)
         return exact
 
     # Step 2: semantic dedup (only if we have an embedding to compare with)
@@ -725,11 +756,7 @@ async def upsert_entity_with_link(
             existing_row, distance = near_row
             # Cosine distance ≤ (1 - threshold) ⇒ similarity ≥ threshold.
             if float(distance) <= (1.0 - dedup_cosine_threshold):
-                if memory_id not in existing_row.linked_memory_ids:
-                    existing_row.linked_memory_ids = [
-                        *existing_row.linked_memory_ids,
-                        memory_id,
-                    ]
+                await _append_entity_link_atomic(session, existing_row, memory_id)
                 return existing_row
 
     # Step 3: insert fresh — via ON CONFLICT against the unique identity

tests/test_subject_delete_deadlock_retry.py

@@ -0,0 +1,101 @@
+"""Subject purge retries once on a Postgres deadlock (main-CI flake 09-02).
+
+The purge's multi-row DELETEs can cross lock order with another
+multi-row writer on the same subject's rows (a still-draining compile
+batch, an embedding backfill). Postgres aborts exactly one side with
+SQLSTATE 40P01; before this the purge surfaced that as a raw 500 —
+observed as `DELETE FROM memories` deadlocking in
+tests/integration/test_semantic.py on main run 33638444055. A single
+rollback-and-redo converges: the competing transaction has either
+finished or loses the rematch.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from sqlalchemy.exc import DBAPIError
+
+from server.api import subjects as api_subjects
+
+pytestmark = pytest.mark.asyncio
+
+
+def _deadlock_error() -> DBAPIError:
+    orig = Exception("deadlock detected\nDETAIL: Process 1 waits for ShareLock...")
+    return DBAPIError("DELETE FROM memories ...", params=None, orig=orig)
+
+
+def _other_db_error() -> DBAPIError:
+    return DBAPIError("DELETE ...", params=None, orig=Exception("connection reset"))
+
+
+def _wire(monkeypatch, delete_memories):
+    async def two(_s, _subj, *, tenant_id):
+        return 2
+
+    async def none(_s, _subj, *, tenant_id):
+        return 0
+
+    monkeypatch.setattr(api_subjects.repo, "delete_episodes_by_subject", two)
+    monkeypatch.setattr(api_subjects.repo, "delete_memories_by_subject", delete_memories)
+    monkeypatch.setattr(api_subjects.repo, "delete_resolutions_by_subject", none)
+    monkeypatch.setattr(api_subjects.repo, "delete_health_cache_by_subject", none)
+    monkeypatch.setattr(api_subjects.repo, "delete_entities_by_subject", none)
+    monkeypatch.setattr(api_subjects.webhooks, "fire", AsyncMock())
+
+
+def _session():
+    session = MagicMock()
+    session.commit = AsyncMock()
+    session.rollback = AsyncMock()
+    return session
+
+
+async def test_purge_retries_once_on_deadlock_and_succeeds(monkeypatch):
+    calls = {"n": 0}
+
+    async def flaky_delete(_s, _subj, *, tenant_id):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise _deadlock_error()
+        return 3
+
+    _wire(monkeypatch, flaky_delete)
+    session = _session()
+
+    resp = await api_subjects.delete_subject("subj", session=session, tenant_id=None)
+
+    assert resp.memories_deleted == 3
+    assert calls["n"] == 2
+    session.rollback.assert_awaited_once()
+    session.commit.assert_awaited_once()
+
+
+async def test_purge_gives_up_after_second_deadlock(monkeypatch):
+    async def always_deadlocks(_s, _subj, *, tenant_id):
+        raise _deadlock_error()
+
+    _wire(monkeypatch, always_deadlocks)
+    session = _session()
+
+    with pytest.raises(DBAPIError):
+        await api_subjects.delete_subject("subj", session=session, tenant_id=None)
+    session.rollback.assert_awaited_once()  # only between attempts, not after the raise
+
+
+async def test_purge_does_not_retry_non_deadlock_errors(monkeypatch):
+    calls = {"n": 0}
+
+    async def breaks(_s, _subj, *, tenant_id):
+        calls["n"] += 1
+        raise _other_db_error()
+
+    _wire(monkeypatch, breaks)
+    session = _session()
+
+    with pytest.raises(DBAPIError):
+        await api_subjects.delete_subject("subj", session=session, tenant_id=None)
+    assert calls["n"] == 1, "a non-deadlock DB error must surface immediately"
+    session.rollback.assert_not_awaited()