From a2f18efde0f80008fa838f6303e6d2edbfaab8fc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Tue, 11 Aug 2026 16:59:30 +0200 Subject: [PATCH 1/2] fix(retain): match chunk-delete link endpoints through indexable joins (#3387) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ordered memory_links pre-delete in delete_chunks_by_ids matched link endpoints with `tu.id = ml.from_unit_id OR tu.id = ml.to_unit_id`. An OR spanning two columns of ml cannot be driven from either endpoint index, so PostgreSQL made memory_links the outer relation of a nested-loop semi join and sequentially scanned the whole table on every delete — O(links x units), with no bank_id predicate, so every bank scanned every other bank's rows. Past a few million links that exceeded the asyncpg command timeout and delta retain failed with a bare TimeoutError (str(TimeoutError()) is empty, so it logged as "Task execution failed: batch_retain, error:" with nothing after). Splitting the OR into a UNION of two single-column joins makes each half an index scan. Measured on PG18 with the current index set, 300k units / 1.5M links, deleting the links of 3 chunks (150 units): before 20,420 ms Seq Scan, 224.9M rows removed by join filter after 31 ms two index scans, same 1,464 rows 10 chunks took 49.8s before — already over the 60s default at a table size well below production. The row set is identical (EXCEPT in both directions returns nothing), and the deterministic ORDER BY + FOR UPDATE that #2570 added stay in ordered_links, which still locks the rows in that order. --- .../engine/retain/chunk_storage.py | 28 +++- .../test_chunk_storage_delete_ordering.py | 122 ++++++++++++++++++ 2 files changed, 145 insertions(+), 5 deletions(-) diff --git a/hindsight-api-slim/hindsight_api/engine/retain/chunk_storage.py b/hindsight-api-slim/hindsight_api/engine/retain/chunk_storage.py index 4621462322..031684326c 100644 --- a/hindsight-api-slim/hindsight_api/engine/retain/chunk_storage.py +++ b/hindsight-api-slim/hindsight_api/engine/retain/chunk_storage.py @@ -155,6 +155,19 @@ 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 ( @@ -162,14 +175,19 @@ async def delete_chunks_by_ids(conn, chunk_ids: list[str], bank_id: str | None = 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), diff --git a/hindsight-api-slim/tests/test_chunk_storage_delete_ordering.py b/hindsight-api-slim/tests/test_chunk_storage_delete_ordering.py index b25d535644..5830dbdbb2 100644 --- a/hindsight-api-slim/tests/test_chunk_storage_delete_ordering.py +++ b/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,6 +46,29 @@ 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() @@ -48,3 +76,97 @@ async def test_delete_chunks_by_ids_noops_without_chunks(): 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) From 0ecbbd5d2533eead14a87adef80cdac0bcd5b1d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Tue, 11 Aug 2026 17:19:15 +0200 Subject: [PATCH 2/2] test(memories): update store doubles for the per-bank capability API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #3388 moved the store capability to writes_memory_rows_in_sql_for(bank_id) and added drop_bank_storage, but two duck-typed test doubles still carried the old surface, so test-api shard 2 has been red on main since it merged: test_integrity_violation_not_retried — SimpleNamespace(writes_memory_rows_in_sql=True) -> AttributeError: no attribute 'writes_memory_rows_in_sql_for' test_list_banks_non_sql_store._NonSqlStore — teardown's delete_bank routes the drop through the store for a non-SQL bank -> AttributeError: no attribute 'drop_bank_storage' Both doubles are hand-rolled rather than subclasses of the store base, which is why the refactor could not update them mechanically. --- .../tests/test_integrity_violation_not_retried.py | 4 +++- hindsight-api-slim/tests/test_list_banks_non_sql_store.py | 4 ++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/hindsight-api-slim/tests/test_integrity_violation_not_retried.py b/hindsight-api-slim/tests/test_integrity_violation_not_retried.py index 752a8a600a..fc0cf1cdd4 100644 --- a/hindsight-api-slim/tests/test_integrity_violation_not_retried.py +++ b/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)))) diff --git a/hindsight-api-slim/tests/test_list_banks_non_sql_store.py b/hindsight-api-slim/tests/test_list_banks_non_sql_store.py index bd175af9c0..37f633d5eb 100644 --- a/hindsight-api-slim/tests/test_list_banks_non_sql_store.py +++ b/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):