diff --git a/hindsight-api-slim/hindsight_api/engine/db/ops_oracle.py b/hindsight-api-slim/hindsight_api/engine/db/ops_oracle.py index f79fa317c..b80801ccf 100644 --- a/hindsight-api-slim/hindsight_api/engine/db/ops_oracle.py +++ b/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} diff --git a/hindsight-api-slim/hindsight_api/engine/db/ops_postgresql.py b/hindsight-api-slim/hindsight_api/engine/db/ops_postgresql.py index 4bd1652f6..e81412b6c 100644 --- a/hindsight-api-slim/hindsight_api/engine/db/ops_postgresql.py +++ b/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, ) diff --git a/hindsight-api-slim/hindsight_api/engine/db_utils.py b/hindsight-api-slim/hindsight_api/engine/db_utils.py index 8d0b77839..b67263447 100644 --- a/hindsight-api-slim/hindsight_api/engine/db_utils.py +++ b/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..." diff --git a/hindsight-api-slim/hindsight_api/engine/graph_maintenance.py b/hindsight-api-slim/hindsight_api/engine/graph_maintenance.py index bdceb7931..a7af62527 100644 --- a/hindsight-api-slim/hindsight_api/engine/graph_maintenance.py +++ b/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( diff --git a/hindsight-api-slim/tests/test_db_utils.py b/hindsight-api-slim/tests/test_db_utils.py index 69f1c2462..3cbc76aab 100644 --- a/hindsight-api-slim/tests/test_db_utils.py +++ b/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) diff --git a/hindsight-api-slim/tests/test_graph_maintenance_deadlock.py b/hindsight-api-slim/tests/test_graph_maintenance_deadlock.py index 514845f5a..ea8bdcb7f 100644 --- a/hindsight-api-slim/tests/test_graph_maintenance_deadlock.py +++ b/hindsight-api-slim/tests/test_graph_maintenance_deadlock.py @@ -1,6 +1,12 @@ """Reproduces the concurrent-insert deadlock on ``graph_maintenance_queue`` that PR #2353 targets, and demonstrates that a shared insertion order cures it. +Also covers the ``entity_cooccurrences`` deadlock between the Pass 2/3 sweep +(``prune_stale_cooccurrences``) and retain's concurrent cooccurrence upserts — +the dominant ``DeadlockDetectedError`` cause in production logs (39 of 41 +occurrences in one week) — and the ``retry_with_backoff`` fix around that +sweep in ``run_graph_maintenance_job``. + A deadlock is a *database*-level phenomenon, so unlike the PR's own tests (which only assert that the Python list handed to ``conn.execute`` is sorted) these run against the real Postgres test DB and drive two genuinely-concurrent @@ -24,10 +30,14 @@ import asyncio import uuid +from datetime import UTC, datetime +from unittest.mock import AsyncMock import pytest from asyncpg.exceptions import DeadlockDetectedError +from hindsight_api import RequestContext +from hindsight_api.engine.graph_maintenance import run_graph_maintenance_job from hindsight_api.engine.memory_engine import MemoryEngine # Two keys with an unambiguous sort order (low < high as UUIDs / as text). @@ -84,6 +94,181 @@ async def worker(order: list[uuid.UUID]) -> None: assert deadlocks, f"expected one transaction aborted with DeadlockDetectedError, got {results!r}" +async def _insert_entity(conn, bank_id: str, entity_id: uuid.UUID, name: str) -> None: + await conn.execute( + """ + INSERT INTO entities (id, bank_id, canonical_name, first_seen, last_seen, mention_count) + VALUES ($1, $2, $3, NOW(), NOW(), 1) + """, + entity_id, + bank_id, + name, + ) + + +async def _insert_cooccurrence(conn, entity_a: uuid.UUID, entity_b: uuid.UUID) -> None: + first, second = sorted([entity_a, entity_b]) + await conn.execute( + """ + INSERT INTO entity_cooccurrences (entity_id_1, entity_id_2, cooccurrence_count, last_cooccurred) + VALUES ($1, $2, 1, NOW()) + """, + first, + second, + ) + + +@pytest.mark.asyncio +async def test_unordered_concurrent_sweep_and_upsert_deadlocks(memory: MemoryEngine): + """Reproduces the ``prune_stale_cooccurrences`` deadlock seen in production + (``asyncpg.exceptions.DeadlockDetectedError`` on ``entity_cooccurrences``, + 39 of 41 deadlock occurrences in one week of logs). + + ``prune_stale_cooccurrences`` (graph maintenance) scans/locks + ``entity_cooccurrences`` rows for a bank in whatever order its join/NOT + EXISTS plan visits them. Concurrently, retain's cooccurrence upserts + (``entity_resolver._flush_pending``) lock rows in sorted + ``(entity_id_1, entity_id_2)`` order — sorted specifically to avoid + deadlocking against *other* concurrent upserts, but that guarantee says + nothing about the maintenance sweep's scan order. Two transactions + touching the same two cooccurrence rows in opposite orders cycle, and + Postgres aborts one with ``DeadlockDetectedError``. + + This models each side as a sequence of single-row statements with a + barrier between them (same technique as the queue-insert test above) so + the interleaving that produces the cycle is deterministic rather than + racy. + """ + pool = await memory._get_pool() + bank_id = f"dl-sweep-{uuid.uuid4().hex[:8]}" + + # Two cooccurrence rows. entity_cooccurrence_order_check pins each row's + # own (entity_id_1, entity_id_2) internally, but the two ROWS themselves + # sort as pair_low < pair_high by their first UUID. + pair_low = (uuid.UUID("00000000-0000-4000-8000-000000000001"), uuid.UUID("00000000-0000-4000-8000-000000000002")) + pair_high = (uuid.UUID("ffffffff-ffff-4fff-8fff-fffffffffffe"), uuid.UUID("ffffffff-ffff-4fff-8fff-ffffffffffff")) + + async with pool.acquire() as conn: + for entity_id, name in [ + (pair_low[0], "low-a"), + (pair_low[1], "low-b"), + (pair_high[0], "high-a"), + (pair_high[1], "high-b"), + ]: + await _insert_entity(conn, bank_id, entity_id, name) + await _insert_cooccurrence(conn, *pair_low) + await _insert_cooccurrence(conn, *pair_high) + + barrier = asyncio.Barrier(2) + + async def touch(conn, entity_a: uuid.UUID, entity_b: uuid.UUID) -> None: + first, second = sorted([entity_a, entity_b]) + await conn.execute( + "UPDATE entity_cooccurrences SET cooccurrence_count = cooccurrence_count + 1 " + "WHERE entity_id_1 = $1 AND entity_id_2 = $2", + first, + second, + ) + + async def sweep_worker() -> None: + """Mimics the maintenance scan visiting pair_high before pair_low.""" + async with pool.acquire() as conn: + async with conn.transaction(): + await touch(conn, *pair_high) + await barrier.wait() + await touch(conn, *pair_low) + + async def upsert_worker() -> None: + """Mimics retain's sorted upsert — always ascending order.""" + async with pool.acquire() as conn: + async with conn.transaction(): + await touch(conn, *pair_low) + await barrier.wait() + await touch(conn, *pair_high) + + results = await asyncio.wait_for( + asyncio.gather(sweep_worker(), upsert_worker(), return_exceptions=True), + timeout=30, + ) + + deadlocks = [r for r in results if isinstance(r, DeadlockDetectedError)] + assert deadlocks, f"expected one transaction aborted with DeadlockDetectedError, got {results!r}" + + +@pytest.mark.asyncio +async def test_graph_maintenance_sweep_retries_on_deadlock(memory: MemoryEngine, request_context: RequestContext): + """``run_graph_maintenance_job``'s Pass 2/3 sweep must survive a deadlock. + + Fix for the production bug reproduced above: the sweep + (``prune_orphan_entities`` + ``prune_stale_cooccurrences``) now runs + inside ``retry_with_backoff``, which already special-cases + ``DeadlockDetectedError``. Both prunes are idempotent bank-wide deletes, + so retrying the whole transaction after a deadlock is safe. Simulates the + deadlock via a mock (real two-connection reproduction is covered above; + this asserts the *application* behaviour — one transient + DeadlockDetectedError must not fail the job) and asserts the retry + actually happens and the job still returns correct prune counts. + """ + bank_id = f"dl-fix-sweep-{uuid.uuid4().hex[:8]}" + await memory.get_bank_profile(bank_id=bank_id, request_context=request_context) + + pool = await memory._get_pool() + ent_a = uuid.uuid4() + ent_b = uuid.uuid4() + async with pool.acquire() as conn: + await _insert_entity(conn, bank_id, ent_a, "alice") + await _insert_entity(conn, bank_id, ent_b, "bob") + # Both entities are referenced by units (so prune_orphan_entities + # leaves them alone), but by DIFFERENT units — no unit references + # both, so the cooccurrence itself is stale and eligible for + # prune_stale_cooccurrences. + for entity_id, text in [(ent_a, "with_alice"), (ent_b, "with_bob")]: + unit_id = uuid.uuid4() + await conn.execute( + """ + INSERT INTO memory_units (id, bank_id, text, fact_type, event_date, created_at, updated_at) + VALUES ($1, $2, $3, 'experience', NOW(), NOW(), NOW()) + """, + unit_id, + bank_id, + text, + ) + await conn.execute( + "INSERT INTO unit_entities (unit_id, entity_id) VALUES ($1, $2)", + unit_id, + entity_id, + ) + await conn.execute( + """ + INSERT INTO entity_cooccurrences (entity_id_1, entity_id_2, cooccurrence_count, last_cooccurred) + VALUES ($1, $2, 1, $3) + """, + *sorted([ent_a, ent_b]), + datetime.now(UTC), + ) + + backend = await memory._get_backend() + real_prune = backend.ops.prune_stale_cooccurrences + call_count = 0 + + async def flaky_prune(*args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise DeadlockDetectedError("deadlock detected") + return await real_prune(*args, **kwargs) + + backend.ops.prune_stale_cooccurrences = AsyncMock(side_effect=flaky_prune) + try: + result = await run_graph_maintenance_job(memory, bank_id, request_context) + finally: + backend.ops.prune_stale_cooccurrences = real_prune + + assert call_count == 2, "expected exactly one deadlock retry" + assert result["orphan_entities_pruned"] == 0 + assert result["stale_cooccurrences_pruned"] == 1 + + @pytest.mark.asyncio async def test_ordered_concurrent_enqueue_does_not_deadlock(memory: MemoryEngine): """With both transactions inserting in the SAME (sorted) order — exactly what diff --git a/hindsight-dev/benchmarks/perf/README.md b/hindsight-dev/benchmarks/perf/README.md index 6657df928..cec2b75a7 100644 --- a/hindsight-dev/benchmarks/perf/README.md +++ b/hindsight-dev/benchmarks/perf/README.md @@ -18,8 +18,36 @@ uv run perf-test --output results.json # save JSON results |-------|-----------------| | `retain` | Full retain pipeline with mock LLM: fact extraction callback, embedding generation, DB writes, entity linking | | `recall` | Pre-populated bank recall: 4-way parallel retrieval (semantic, BM25, graph, temporal), RRF fusion, percentile latency | +| `graph-maintenance` | `run_graph_maintenance_job` in isolation after a batch of deletes: relink pass wall-clock broken down by probe (semantic ANN vs temporal), plus entity/cooccurrence prune counts (#1919) | +| `graph-maintenance-contention` | The maintenance cooccurrence sweep **under concurrent retain load** — drives `prune_stale_cooccurrences` against retain-shaped sorted cooccurrence upserts and counts how many maintenance passes the resulting deadlock silently drops (#2529). See below | | `stats` | `/stats` endpoint (`get_bank_stats`): uncached aggregation latency (node/link counts + entity rollup join) vs. cached latency, plus cache speedup. Runs with the result cache **disabled** (TTL=0) so the headline numbers are the real per-poll cost | +#### `graph-maintenance-contention` (#2529) + +The `graph-maintenance` suite runs the maintenance job **in isolation** — no +concurrent writer — so its Pass 2/3 cooccurrence sweep can never overlap a +retain upsert and never deadlocks. That is exactly why continuous perf never +caught #2529: the `DeadlockDetectedError` is a *contention* phenomenon between +`prune_stale_cooccurrences`' unordered `DELETE` scan and retain's sorted +`(entity_id_1, entity_id_2)` cooccurrence upsert (`entity_resolver._flush_pending`), +and no suite drove both at once. In production a losing deadlock silently drops +a maintenance pass — visible only as slow graph bloat, never as a latency number. + +This suite closes the gap: it seeds a hot set of stale cooccurrence pairs, then +runs `upsert_workers` retain-shaped upsert loops against `sweep_workers` loops +calling `run_graph_maintenance_job`. The gate metric is the **deadlock escape +rate** — of the deadlocks the sweep hits, what fraction escaped the job and +dropped a pass: + +- **Without the fix:** every deadlock escapes → escape rate ≈ **100%** → suite **fails**. +- **With the fix** (the sweep wrapped in a jittered `retry_with_backoff` with a larger retry budget): the sweep retries → escape rate ≈ **0%** → suite **passes**. A vanishingly small tail can still slip through under the deliberately brutal multi-sweep synthetic load here; a realistic single-sweep-per-bank load drops nothing. + +The suite fails if the escape rate exceeds `GRAPH_CONTENTION_ESCAPE_RATE_THRESHOLD` +(0.5), or if no deadlock reproduced at all (a hollow run that proves nothing). + +Measured (`--scale small`): **main** ≈ 100% escape (every deadlock drops a pass) +→ FAIL; **fixed** 0% escape (deadlocks still occur, none escape) → PASS. + ### Scale Configurations | Scale | Retain items | Recall bank size | Recall iterations | Recall concurrency | diff --git a/hindsight-dev/benchmarks/perf/system_perf.py b/hindsight-dev/benchmarks/perf/system_perf.py index f5007a010..62ab7ff22 100644 --- a/hindsight-dev/benchmarks/perf/system_perf.py +++ b/hindsight-dev/benchmarks/perf/system_perf.py @@ -14,6 +14,7 @@ uv run perf-test --suite retain uv run perf-test --suite recall uv run perf-test --suite graph-maintenance + uv run perf-test --suite graph-maintenance-contention # deadlock gate (#2529) uv run perf-test --suite stats # Configurable scale @@ -32,6 +33,7 @@ import argparse import asyncio import json +import random import statistics import time import uuid @@ -69,6 +71,11 @@ "recall_concurrency": 1, "consolidation_items": 20, "graph_maintenance_bank_size": 20, + "graph_contention_entities": 20, + "graph_contention_pairs": 60, + "graph_contention_upsert_workers": 4, + "graph_contention_sweep_workers": 2, + "graph_contention_rounds": 15, "stats_bank_size": 20, }, "small": { @@ -78,6 +85,11 @@ "recall_concurrency": 4, "consolidation_items": 200, "graph_maintenance_bank_size": 200, + "graph_contention_entities": 60, + "graph_contention_pairs": 400, + "graph_contention_upsert_workers": 8, + "graph_contention_sweep_workers": 2, + "graph_contention_rounds": 25, "stats_bank_size": 200, }, "medium": { @@ -87,6 +99,11 @@ "recall_concurrency": 8, "consolidation_items": 1_000, "graph_maintenance_bank_size": 1_000, + "graph_contention_entities": 120, + "graph_contention_pairs": 1_500, + "graph_contention_upsert_workers": 10, + "graph_contention_sweep_workers": 2, + "graph_contention_rounds": 35, "stats_bank_size": 1_000, }, "large": { @@ -101,6 +118,14 @@ # as an Index Scan on idx_mu_emb_*. medium (1k) stays in the exact-scan # regime, so the two scales cover both planner paths. "graph_maintenance_bank_size": 15_000, + # Hot cooccurrence set big enough that both the sweep DELETE and the + # sorted upserts lock many rows at once — wide overlap → reliable + # opposite-order deadlocks under the higher worker fan-out. + "graph_contention_entities": 200, + "graph_contention_pairs": 4_000, + "graph_contention_upsert_workers": 16, + "graph_contention_sweep_workers": 2, + "graph_contention_rounds": 45, # Large, entity-dense bank so the unit_entities→memory_units rollup join # in _compute_bank_stats is exercised at a size where its cost shows. "stats_bank_size": 15_000, @@ -119,6 +144,12 @@ "recall_concurrency": 4, "consolidation_items": 5_000, "graph_maintenance_bank_size": 15_000, + # Contention keys kept for parity; `huge` targets the stats suite only. + "graph_contention_entities": 120, + "graph_contention_pairs": 1_500, + "graph_contention_upsert_workers": 10, + "graph_contention_sweep_workers": 2, + "graph_contention_rounds": 35, "stats_bank_size": 500_000, # unused by the bulk path; kept for key parity "stats_units": 500_000, "stats_semantic_links": 9_460_147, @@ -135,6 +166,17 @@ # top up the surviving units' links" workload (see #1919). GRAPH_MAINTENANCE_DELETE_PCT = 0.1 +# graph-maintenance-contention pass/fail gate (#2529). The discriminating metric +# is the *escape rate*: of the DeadlockDetectedErrors the sweep hits, what +# fraction escaped run_graph_maintenance_job and dropped a maintenance pass. +# Without the retry wrap EVERY deadlock escapes (rate ≈ 1.0); with the jittered +# retry_with_backoff the sweep effectively never drops a pass under a realistic +# single-sweep load (rate ≈ 0). A vanishingly small tail can still slip through +# under deliberately brutal multi-sweep synthetic load if the retry budget is +# exhausted, so we gate on the rate rather than a hard zero. A run that regresses +# the fix jumps straight back to ≈1.0, so 0.5 cleanly separates the two. +GRAPH_CONTENTION_ESCAPE_RATE_THRESHOLD = 0.5 + # Recall queries that exercise different retrieval strategies RECALL_QUERIES = [ "database migration", @@ -232,6 +274,47 @@ class GraphMaintenanceResult: temporal_calls: int +@dataclass +class _ContentionCounters: + """Mutable tallies shared across the contention suite's upsert/sweep workers.""" + + upserts: int = 0 + upsert_deadlocks: int = 0 + attempted: int = 0 + deadlocks: int = 0 + dropped: int = 0 + succeeded: int = 0 + + +@dataclass +class GraphContentionResult: + """Result of the graph-maintenance-contention suite (#2529). + + The headline metric is ``sweep_passes_dropped``: maintenance passes that + raised ``DeadlockDetectedError`` out of ``run_graph_maintenance_job`` and + were silently lost. ``sweep_deadlocks_observed`` counts raw deadlocks at the + ``prune_stale_cooccurrences`` call — it stays > 0 on *both* main and the + fix (the contention is identical), but ``sweep_passes_dropped`` collapses + from > 0 to 0 once the sweep is wrapped in ``retry_with_backoff``. That + delta is the fix's measurable effect. + """ + + contention_entities: int + contention_pairs: int + upsert_workers: int + sweep_workers: int + total_upserts: int + upsert_deadlocks: int + sweep_passes_attempted: int + sweep_deadlocks_observed: int + sweep_passes_dropped: int + sweep_passes_succeeded: int + # dropped / deadlocks_observed — the gate metric. ≈1.0 unprotected, ≈0 fixed. + sweep_deadlock_escape_rate: float + sweep_latency: PercentileStats + total_duration_seconds: float + + @dataclass class StatsResult: bank_size: int @@ -259,6 +342,7 @@ class SuiteResult: recall: RecallResult | None = None consolidation: ConsolidationResult | None = None graph_maintenance: GraphMaintenanceResult | None = None + graph_contention: GraphContentionResult | None = None stats: StatsResult | None = None @@ -1424,6 +1508,280 @@ def _row(label: str, secs: float, calls: int) -> None: ) +# --------------------------------------------------------------------------- +# Suite: graph-maintenance-contention (#2529) +# --------------------------------------------------------------------------- +# +# The graph-maintenance suite above runs run_graph_maintenance_job in ISOLATION +# — populate, delete, one pass, no concurrent writer. That is exactly why +# continuous perf never caught #2529: its Pass 2/3 cooccurrence sweep +# (prune_stale_cooccurrences, an unordered DELETE scan) only deadlocks when it +# overlaps retain's concurrent cooccurrence upsert (entity_resolver +# ._flush_pending, which locks rows in sorted (entity_id_1, entity_id_2) order). +# With no concurrent upsert there is no lock cycle, so the isolated suite can't +# see it, and a dropped maintenance pass shows up in prod only as slow graph +# bloat over time — never as a latency number. +# +# This suite closes that gap: it drives both sides at once and measures how many +# maintenance passes the deadlock silently drops. On main that number is > 0; +# with #2529's retry_with_backoff wrap it is 0 while the deadlocks still happen +# (and get absorbed) — which is the whole point. + + +async def _seed_contention_fixture(engine: Any, bank_id: str, n_entities: int, n_pairs: int) -> list[tuple]: + """Seed ``n_entities`` entities plus ``n_pairs`` *stale* cooccurrences among them. + + Every entity is pinned by its own dedicated keeper unit (a unit_entities row), + so ``prune_orphan_entities`` leaves them alone — but because no single unit + references two contention entities, every seeded cooccurrence is the exact + "both entities exist, no current unit witnesses them together" stale case + that ``prune_stale_cooccurrences`` targets. Returns the sorted pair list. + """ + from hindsight_api.engine.schema import fq_table + + pool = await engine._get_pool() + ent_ids = [uuid.uuid4() for _ in range(n_entities)] + unit_ids = [uuid.uuid4() for _ in range(n_entities)] + + await pool.executemany( + f""" + INSERT INTO {fq_table("entities")} (id, bank_id, canonical_name, first_seen, last_seen, mention_count) + VALUES ($1, $2, $3, NOW(), NOW(), 1) + """, + [(eid, bank_id, f"contention-entity-{i}") for i, eid in enumerate(ent_ids)], + ) + await pool.executemany( + f""" + INSERT INTO {fq_table("memory_units")} (id, bank_id, text, fact_type, event_date, created_at, updated_at) + VALUES ($1, $2, $3, 'experience', NOW(), NOW(), NOW()) + """, + [(uid, bank_id, f"contention keeper unit {i}") for i, uid in enumerate(unit_ids)], + ) + await pool.executemany( + f"INSERT INTO {fq_table('unit_entities')} (unit_id, entity_id) VALUES ($1, $2)", + list(zip(unit_ids, ent_ids, strict=True)), + ) + + # Distinct sorted pairs (entity_cooccurrence_order_check pins id_1 < id_2). + rng = random.Random(1234) + target = min(n_pairs, n_entities * (n_entities - 1) // 2) + pairs: set[tuple] = set() + while len(pairs) < target: + a, b = rng.sample(ent_ids, 2) + pairs.add((a, b) if a < b else (b, a)) + pair_list = sorted(pairs) + + await pool.executemany( + f""" + INSERT INTO {fq_table("entity_cooccurrences")} (entity_id_1, entity_id_2, cooccurrence_count, last_cooccurred) + VALUES ($1, $2, 1, NOW()) + ON CONFLICT (entity_id_1, entity_id_2) DO NOTHING + """, + pair_list, + ) + return pair_list + + +async def run_graph_maintenance_contention_suite(scale_cfg: dict[str, int]) -> SuiteResult: + """Provoke the #2529 cooccurrence-sweep deadlock under concurrent retain load. + + Seeds a hot set of stale cooccurrence pairs, then runs two workloads at once: + + * ``upsert_workers`` tasks re-upserting those pairs in sorted + ``(entity_id_1, entity_id_2)`` order — a faithful copy of retain's + ``entity_resolver._flush_pending`` cooccurrence path (same SQL, same lock + order). + * ``sweep_workers`` tasks repeatedly calling ``run_graph_maintenance_job``, + whose Pass 2/3 ``prune_stale_cooccurrences`` DELETEs those same rows in + an unordered join-scan order. + + Overlapping rows locked in opposite orders → Postgres aborts one side with + ``DeadlockDetectedError``. When the sweep is the victim, the pass is dropped. + The suite counts dropped passes: > 0 on main, 0 with the fix. + """ + from asyncpg.exceptions import DeadlockDetectedError + from hindsight_api.engine.graph_maintenance import run_graph_maintenance_job + from hindsight_api.engine.schema import fq_table + from hindsight_api.models import RequestContext + + n_entities = scale_cfg["graph_contention_entities"] + n_pairs = scale_cfg["graph_contention_pairs"] + upsert_workers = scale_cfg["graph_contention_upsert_workers"] + sweep_workers = scale_cfg["graph_contention_sweep_workers"] + rounds = scale_cfg["graph_contention_rounds"] + bank_id = f"perf-graphcont-{uuid.uuid4().hex[:8]}" + + console.print( + f"\n[bold cyan]Suite: graph-maintenance-contention[/bold cyan] " + f"entities={n_entities} pairs={n_pairs} upsert_workers={upsert_workers} " + f"sweep_workers={sweep_workers} bank={bank_id}" + ) + + engine = _build_engine(disable_observations=True) + await engine.initialize() + await engine.get_bank_profile(bank_id=bank_id, request_context=RequestContext()) + + pair_list = await _seed_contention_fixture(engine, bank_id, n_entities, n_pairs) + console.print(f" Seeded {n_entities:,} entities + {len(pair_list):,} stale cooccurrence pairs") + + pool = await engine._get_pool() + request_context = RequestContext() + + # Mirror entity_resolver._flush_pending's cooccurrence upsert exactly. + upsert_sql = f""" + INSERT INTO {fq_table("entity_cooccurrences")} (entity_id_1, entity_id_2, cooccurrence_count, last_cooccurred) + VALUES ($1, $2, $3, $4) + ON CONFLICT (entity_id_1, entity_id_2) + DO UPDATE SET + cooccurrence_count = {fq_table("entity_cooccurrences")}.cooccurrence_count + EXCLUDED.cooccurrence_count, + last_cooccurred = GREATEST({fq_table("entity_cooccurrences")}.last_cooccurred, EXCLUDED.last_cooccurred) + """ + + counters = _ContentionCounters() + sweep_latencies: list[float] = [] + + # Count raw deadlocks at the prune call — fires on BOTH branches (the fix + # retries around it, so it still passes through here on every attempt), + # proving the contention is real rather than that the load vanished. + backend = await engine._get_backend() + ops = backend.ops + orig_prune = ops.prune_stale_cooccurrences + + async def counting_prune(*args: Any, **kwargs: Any) -> Any: + try: + return await orig_prune(*args, **kwargs) + except DeadlockDetectedError: + counters.deadlocks += 1 + raise + + async def _upsert_worker(seed: int) -> None: + rng = random.Random(seed) + subset_k = max(1, int(len(pair_list) * 0.7)) + for _ in range(rounds): + subset = sorted(rng.sample(pair_list, subset_k)) # sorted → retain's lock order + now = datetime.now(timezone.utc) + rows = [(a, b, 1, now) for (a, b) in subset] + while True: + try: + async with pool.acquire() as conn: + async with conn.transaction(): + await conn.executemany(upsert_sql, rows) + counters.upserts += 1 + break + except DeadlockDetectedError: + # retain retries at a higher level; keep the load flowing. + counters.upsert_deadlocks += 1 + + async def _sweep_worker(stop_event: asyncio.Event) -> None: + while not stop_event.is_set(): + counters.attempted += 1 + t0 = time.perf_counter() + try: + await run_graph_maintenance_job(engine, bank_id, request_context) + except DeadlockDetectedError: + # The escaping deadlock #2529 fixes — the maintenance pass is lost. + counters.dropped += 1 + else: + counters.succeeded += 1 + sweep_latencies.append(time.perf_counter() - t0) + + ops.prune_stale_cooccurrences = counting_prune + stop_event = asyncio.Event() + t0 = time.perf_counter() + try: + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + TimeElapsedColumn(), + console=console, + ) as progress: + progress.add_task(f"Contending ({upsert_workers} upsert × {sweep_workers} sweep)…") + sweep_tasks = [asyncio.create_task(_sweep_worker(stop_event)) for _ in range(sweep_workers)] + await asyncio.gather(*(_upsert_worker(1000 + i) for i in range(upsert_workers))) + # Let sweeps drain any final overlap, then wind them down. + await asyncio.sleep(0.5) + stop_event.set() + await asyncio.gather(*sweep_tasks) + finally: + ops.prune_stale_cooccurrences = orig_prune + duration = time.perf_counter() - t0 + + await engine.delete_bank(bank_id=bank_id, request_context=request_context) + await engine.close() + + # Escape rate = fraction of contention events that cost a maintenance pass. + # Denominator is max(observed, dropped) so a drop that bypasses the prune + # counter (e.g. a deadlock in the orphan-prune FK cascade) still registers + # as an escape instead of dividing by zero into a false 0%. + denom = max(counters.deadlocks, counters.dropped) + escape_rate = counters.dropped / denom if denom else 0.0 + + result = GraphContentionResult( + contention_entities=n_entities, + contention_pairs=len(pair_list), + upsert_workers=upsert_workers, + sweep_workers=sweep_workers, + total_upserts=counters.upserts, + upsert_deadlocks=counters.upsert_deadlocks, + sweep_passes_attempted=counters.attempted, + sweep_deadlocks_observed=counters.deadlocks, + sweep_passes_dropped=counters.dropped, + sweep_passes_succeeded=counters.succeeded, + sweep_deadlock_escape_rate=round(escape_rate, 3), + sweep_latency=PercentileStats.from_samples(sweep_latencies), + total_duration_seconds=round(duration, 3), + ) + + # Hollow-run guard: did the contention workload actually run? Guard on the + # workloads executing (upserts committed, sweeps attempted) — NOT on seeing + # deadlocks. A correct source-level fix (sorted FOR UPDATE lock ordering in + # prune_stale_cooccurrences) *eliminates* the deadlock, so 0 observed is a + # healthy pass, not a hollow one; gating on deadlocks>0 would wrongly fail + # the better fix. A genuinely hollow run (broken seed / crashed workers) + # shows up as no upserts or no sweeps. + ran_contention = counters.upserts > 0 and counters.attempted > 0 + if not ran_contention: + console.print( + " [yellow]WARNING: contention workload did not run (no upserts/sweeps) — " + "seed or worker failure, result inconclusive.[/yellow]" + ) + elif counters.deadlocks == 0: + console.print( + " [green]No deadlocks reproduced — sweep lock ordering prevented the cycle " + "at the source (not merely retried).[/green]" + ) + + # Healthy either way — deadlocks eliminated (0 observed) or retried away + # (observed > 0, ~none escape). The gate is the same: no pass was dropped. + success = ran_contention and escape_rate <= GRAPH_CONTENTION_ESCAPE_RATE_THRESHOLD + + table = Table(title="Graph Maintenance — Contention (#2529)") + table.add_column("Metric", style="cyan") + table.add_column("Value", style="green", justify="right") + table.add_row("Contention entities", f"{n_entities:,}") + table.add_row("Stale pairs", f"{len(pair_list):,}") + table.add_row("Upsert / sweep workers", f"{upsert_workers} / {sweep_workers}") + table.add_row("Upserts committed", f"{counters.upserts:,}") + table.add_row("Upsert-side deadlocks", f"{counters.upsert_deadlocks:,}") + table.add_row("Sweep passes attempted", f"{counters.attempted:,}") + table.add_row("Deadlocks observed (at prune)", f"{counters.deadlocks:,}") + dropped_style = "red" if counters.dropped else "green" + table.add_row("[bold]Sweep passes DROPPED[/bold]", f"[{dropped_style}]{counters.dropped:,}[/{dropped_style}]") + rate_style = "red" if escape_rate > GRAPH_CONTENTION_ESCAPE_RATE_THRESHOLD else "green" + table.add_row("[bold]Deadlock escape rate[/bold]", f"[{rate_style}]{escape_rate:.0%}[/{rate_style}]") + table.add_row("Sweep passes succeeded", f"{counters.succeeded:,}") + table.add_row("Sweep latency p50/p95", f"{result.sweep_latency.p50:.3f}s / {result.sweep_latency.p95:.3f}s") + table.add_row("Duration", f"{duration:.2f}s") + console.print(table) + + return SuiteResult( + name="graph-maintenance-contention", + duration_seconds=round(duration, 3), + success=success, + graph_contention=result, + ) + + # --------------------------------------------------------------------------- # Suite: stats # --------------------------------------------------------------------------- @@ -1630,6 +1988,7 @@ async def time_one() -> float: "recall-temporal": run_recall_temporal_suite, "consolidation": run_consolidation_suite, "graph-maintenance": run_graph_maintenance_suite, + "graph-maintenance-contention": run_graph_maintenance_contention_suite, "stats": run_stats_suite, } @@ -1904,6 +2263,36 @@ def _print_summary(results: PerfTestResults) -> None: "", ) + if suite.graph_contention: + gc = suite.graph_contention + drop_val = f"[red]{gc.sweep_passes_dropped}[/red]" if gc.sweep_passes_dropped else "[green]0[/green]" + table.add_row(suite.name, status, "sweep passes dropped", drop_val, "", "", "") + rate_red = gc.sweep_deadlock_escape_rate > GRAPH_CONTENTION_ESCAPE_RATE_THRESHOLD + rate_val = f"[{'red' if rate_red else 'green'}]{gc.sweep_deadlock_escape_rate:.0%}[/]" + table.add_row("", "", "deadlock escape rate", rate_val, "", "", "") + table.add_row("", "", "deadlocks observed", f"{gc.sweep_deadlocks_observed:,}", "", "", "") + table.add_row( + "", + "", + "passes (ok/attempted)", + f"{gc.sweep_passes_succeeded:,} / {gc.sweep_passes_attempted:,}", + "", + "", + "", + ) + table.add_row( + "", + "", + "sweep latency", + f"mean={gc.sweep_latency.mean:.3f}s", + f"{gc.sweep_latency.p50:.3f}s", + f"{gc.sweep_latency.p95:.3f}s", + f"{gc.sweep_latency.p99:.3f}s", + ) + table.add_row( + "", "", "upserts (ok/deadlocked)", f"{gc.total_upserts:,} / {gc.upsert_deadlocks:,}", "", "", "" + ) + if suite.stats: st = suite.stats table.add_row(