fix(retain): match chunk-delete link endpoints through indexable joins (#3387)

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

@@ -155,21 +155,39 @@ async def delete_chunks_by_ids(conn, chunk_ids: list[str], bank_id: str | None =
     # memory_links in opposite orders and deadlock. Delete links explicitly in a
     # total order before deleting chunks so every writer takes row locks the same
     # way; the FK cascade still handles anything inserted later in this txn.
+    #
+    # ``matched_links`` collects the endpoints as a UNION of two single-column joins
+    # rather than the one ``tu.id = ml.from_unit_id OR tu.id = ml.to_unit_id`` predicate
+    # it replaces. An OR spanning two columns of ``ml`` is not indexable: the planner
+    # cannot drive it from either endpoint index, so it made memory_links the outer
+    # relation of a nested-loop semi join and sequentially scanned the whole table once
+    # per delete — O(rows_in_memory_links x target_units). Past a few million links that
+    # exceeded the asyncpg command timeout and delta retain failed with a bare
+    # TimeoutError (issue #3387). Split in two, each half is an index scan on
+    # idx_memory_links_from_type_weight / idx_memory_links_to_type_weight.
+    # The UNION yields the identical row set; the deterministic ORDER BY and
+    # FOR UPDATE that #2570 added stay in ``ordered_links``, which locks the rows in
+    # that order after the endpoints have been found.
     await conn.execute(
         f"""
         WITH target_units AS MATERIALIZED (
             SELECT id
             FROM {fq_table("memory_units")}
             WHERE chunk_id = ANY($1::text[])
         ),
+        matched_links AS MATERIALIZED (
+            SELECT ml.ctid AS link_ctid
+            FROM {fq_table("memory_links")} ml
+            JOIN target_units tu ON tu.id = ml.from_unit_id
+            UNION
+            SELECT ml.ctid AS link_ctid
+            FROM {fq_table("memory_links")} ml
+            JOIN target_units tu ON tu.id = ml.to_unit_id
+        ),
         ordered_links AS MATERIALIZED (
             SELECT ml.ctid
             FROM {fq_table("memory_links")} ml
-            WHERE EXISTS (
-                SELECT 1
-                FROM target_units tu
-                WHERE tu.id = ml.from_unit_id OR tu.id = ml.to_unit_id
-            )
+            JOIN matched_links ON ml.ctid = matched_links.link_ctid
             ORDER BY
                 LEAST(ml.from_unit_id, ml.to_unit_id),
                 GREATEST(ml.from_unit_id, ml.to_unit_id),

hindsight-api-slim/tests/test_chunk_storage_delete_ordering.py

@@ -1,8 +1,13 @@
 """Regression coverage for deterministic chunk deletion ordering."""
 
+import uuid
+
 import pytest
 
+from hindsight_api.engine.memories import get_memories
+from hindsight_api.engine.memory_engine import MemoryEngine
 from hindsight_api.engine.retain import chunk_storage
+from hindsight_api.models import RequestContext
 
 
 class RecordingConn:
@@ -41,10 +46,127 @@ async def test_delete_chunks_by_ids_predeletes_links_before_chunks():
     assert "FOR UPDATE" in chunk_sql
 
 
+@pytest.mark.asyncio
+async def test_delete_chunks_by_ids_matches_link_endpoints_through_indexable_joins():
+    """The endpoint match must stay two single-column joins, never an OR across columns.
+
+    ``tu.id = ml.from_unit_id OR tu.id = ml.to_unit_id`` cannot be driven from either
+    endpoint index, so PostgreSQL made memory_links the outer relation of a semi join
+    and sequentially scanned the entire table on every delete — 20s at 1.5M links,
+    over the asyncpg command timeout at production sizes (issue #3387).
+    """
+    conn = RecordingConn()
+
+    await chunk_storage.delete_chunks_by_ids(conn, ["chunk-a"])
+
+    link_sql = conn.calls[0][0]
+    normalized = " ".join(link_sql.split())
+    assert "tu.id = ml.from_unit_id OR tu.id = ml.to_unit_id" not in normalized, (
+        "the endpoint match must not reintroduce an OR across from_unit_id/to_unit_id"
+    )
+    assert "JOIN target_units tu ON tu.id = ml.from_unit_id" in link_sql
+    assert "JOIN target_units tu ON tu.id = ml.to_unit_id" in link_sql
+    assert "UNION" in link_sql
+
+
 @pytest.mark.asyncio
 async def test_delete_chunks_by_ids_noops_without_chunks():
     conn = RecordingConn()
 
     await chunk_storage.delete_chunks_by_ids(conn, [])
 
     assert conn.calls == []
+
+
+@pytest.mark.asyncio
+async def test_delete_chunks_by_ids_sweeps_links_on_both_endpoints(
+    memory: MemoryEngine, request_context: RequestContext
+):
+    """Both endpoints of the deleted chunk's units lose their links; other links survive.
+
+    Equivalence guard for the #3387 rewrite: the OR-across-columns predicate became a
+    UNION of two single-column joins, so a link is now matched twice — once from each
+    side — and a rewrite that dropped one arm would silently leave half the links behind
+    for the FK cascade to delete in executor-chosen order, reopening the #2570 deadlock.
+    """
+    bank_id = f"test-link-sweep-{uuid.uuid4().hex[:8]}"
+    if not get_memories().writes_memory_rows_in_sql_for(bank_id):
+        pytest.skip("memory_links reference memory_units rows, which this store keeps outside SQL")
+
+    await memory.get_bank_profile(bank_id=bank_id, request_context=request_context)
+    pool = await memory._get_pool()
+
+    doc_id = str(uuid.uuid4())
+    doomed_chunk = f"{doc_id}_0"
+    kept_chunk = f"{doc_id}_1"
+
+    async def _unit(conn, chunk_id: str | None) -> uuid.UUID:
+        unit_id = uuid.uuid4()
+        await conn.execute(
+            """
+            INSERT INTO memory_units (
+                id, bank_id, text, fact_type, event_date, document_id, chunk_id, created_at, updated_at
+            ) VALUES ($1, $2, $3, 'experience', NOW(), $4, $5, NOW(), NOW())
+            """,
+            unit_id,
+            bank_id,
+            f"fact {unit_id}",
+            doc_id if chunk_id else None,
+            chunk_id,
+        )
+        return unit_id
+
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            INSERT INTO documents (id, bank_id, original_text, content_hash, created_at, updated_at)
+            VALUES ($1, $2, 'doomed text\n\nkept text', 'hash-old', NOW(), NOW())
+            """,
+            doc_id,
+            bank_id,
+        )
+        for idx, chunk_id in enumerate((doomed_chunk, kept_chunk)):
+            await conn.execute(
+                """
+                INSERT INTO chunks (chunk_id, document_id, bank_id, chunk_index, chunk_text, content_hash)
+                VALUES ($1, $2, $3, $4, $5, $6)
+                """,
+                chunk_id,
+                doc_id,
+                bank_id,
+                idx,
+                f"text {idx}",
+                f"hash-{idx}",
+            )
+
+        doomed = await _unit(conn, doomed_chunk)
+        kept = await _unit(conn, kept_chunk)
+        unrelated = await _unit(conn, None)
+
+        # One link per direction on the doomed unit, plus one that must survive.
+        for from_id, to_id in ((doomed, kept), (unrelated, doomed), (kept, unrelated)):
+            await conn.execute(
+                """
+                INSERT INTO memory_links (from_unit_id, to_unit_id, link_type, weight, bank_id)
+                VALUES ($1, $2, 'semantic', 0.8, $3)
+                """,
+                from_id,
+                to_id,
+                bank_id,
+            )
+
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            await chunk_storage.delete_chunks_by_ids(conn, [doomed_chunk], bank_id, ops=memory._backend.ops)
+
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            "SELECT from_unit_id, to_unit_id FROM memory_links WHERE bank_id = $1",
+            bank_id,
+        )
+
+    assert {(str(r["from_unit_id"]), str(r["to_unit_id"])) for r in rows} == {(str(kept), str(unrelated))}, (
+        "links on both endpoints of the deleted chunk's unit must go; the survivors' link must stay"
+    )
+
+    await memory.delete_bank(bank_id, request_context=request_context)

hindsight-api-slim/tests/test_integrity_violation_not_retried.py

@@ -217,7 +217,9 @@ def _patch_update_action_deps(consolidator, conn, source_ids, append_mock) -> Ex
     capability flag, config, and the history append — leaving the UPDATE
     rowcount as the single variable under test.
     """
-    store = SimpleNamespace(writes_memory_rows_in_sql=True)
+    # The capability is consulted per bank (#3388), so the stub answers the bank-scoped
+    # form rather than carrying the bare class attribute it replaced.
+    store = SimpleNamespace(writes_memory_rows_in_sql_for=lambda bank_id: True)
     stack = ExitStack()
     stack.enter_context(patch("hindsight_api.config.get_config", _fake_config))
     stack.enter_context(patch.object(consolidator, "acquire_with_retry", MagicMock(return_value=_AsyncNullCtx(conn))))

hindsight-api-slim/tests/test_list_banks_non_sql_store.py

@@ -40,6 +40,10 @@ async def count_memories(self, *, conn, fq_table, bank_id: str) -> dict:
         self.count_calls.append(bank_id)
         return {"world": 7}
 
+    async def drop_bank_storage(self, bank_id: str) -> None:
+        """``delete_bank`` routes the drop through the store for a non-SQL bank, so the
+        teardown below reaches this. Nothing to drop — the counts above are synthetic."""
+
 
 @pytest.mark.asyncio
 async def test_list_banks_counts_via_store_for_non_sql_bank(memory, monkeypatch):