fix(graph-maintenance): drive the entity prune off a queue, not a bank-wide sweep (#3222)

[nicoloboschi]

Fixes #3222.

The problem

graph_maintenance's Pass 2/3 were two bank-wide statements re-evaluated on every invocation, whether or not anything had changed:

• orphan-entity prune — one index probe per entity in the bank;
• stale-cooccurrence prune — an INTERSECT per cooccurrence row in the bank, to answer "does any unit still witness both endpoints".

Their cost tracked the size of the bank, not the size of the delete. Past a few million rows neither can finish inside asyncpg's 60s command_timeout, so the job failed on every run — forever, on exactly the banks that most needed it. str(TimeoutError()) is empty, which is why it surfaced as Task execution failed: graph_maintenance, error: with nothing after it.

It was worse than one timeout per attempt: db_utils._is_retryable treats asyncio.TimeoutError as transient, and the sweep was wrapped in retry_with_backoff(max_retries=8). Each task attempt therefore re-ran a statement that could never finish nine times, holding a worker slot for ~10 minutes, before the task-level retry did it again.

Measured

Fixture built to mirror a mid-size bank with hub entities — 100k entities, 1.5M unit_entities, 2.86M entity_cooccurrences, endpoints holding 150–400 postings each. PostgreSQL 16, statements run verbatim:

	before (bank-wide)	after (batch of 50)
orphan-entity prune	9.6 s	15 ms
stale-cooccurrence prune	did not finish in 11 min (cancelled)	2.0 s

The cooccurrence prune costs ~1.7 ms per incident pair; over 2.86M pairs that is roughly 80 minutes per run, against a 60s client timeout. This fixture is smaller than the bank in the report.

The fix

Both prunes are now driven by a new entity_maintenance_queue, filled inside the transaction that does the damage — the same pattern graph_maintenance_queue already uses for the relink pass. A run claims a bounded batch, prunes what is genuinely dead, and commits. Cost is O(delta), and work already done survives whatever stops the run.

Two producers, because there are two ways an entity becomes garbage:

1. Its last posting is deleted — captured before the cascade at every site that removes units or replaces postings: document delete, single and bulk memory delete, curation edit and invalidate, document re-ingest, and the delta-retain chunk cascade.
2. It was created and the posting never arrived — a Phase-1 resolution whose linking transaction rolled back. Such a row has no unit_entities entry to find it by and no delete will ever name it, so entity_resolver queues every entity it creates. That is one re-check per entity ever created; the drain finds the posting that did land and keeps it. Without this, queue-driven pruning would silently leak exactly what the bank-wide sweep used to collect.

Supporting changes:

• A wall-clock budget for the job. Both passes commit per batch, so exhausting it is not a failure: the run reports queues_drained: false, logs it, and chains a follow-up (only under a real queue — a synchronous backend would recurse rather than schedule). Backlogs converge over runs instead of being cancelled and retried from scratch.
• Batch size 50, deliberately. The binding cost is per incident cooccurrence pair, not per candidate (~58 pairs per candidate on the fixture). At 500 a batch took 210–300 s — still over the timeout. The constant carries the measurements so the next person doesn't have to rediscover them.
• The scoping predicate is a UNION of the two endpoint columns, not entity_id_1 = ANY(...) OR entity_id_2 = ANY(...) — that OR is the delete_chunks_by_ids: ordered memory_links pre-delete seq-scans the whole table — delta retain times out at scale #3387 shape and cannot be driven from either index.
• The migration seeds the queue with every existing entity, so garbage a bank accumulated while its sweep was failing is still reclaimed — incrementally, a bounded batch per run, instead of in one statement that cannot finish.
• Both passes now return dataclasses (RelinkPassResult, EntityPrunePassResult) rather than raw dicts.

Behaviour change worth knowing

A retain that creates entities now leaves queue rows, so it schedules a graph_maintenance job where before the empty-queue pre-check short-circuited. Dedupe-by-bank coalesces these while one is pending, and the work is one re-check per newly created entity — but on an ingest-heavy bank you will see more graph_maintenance operations than before, each doing much less.

Tests

• Prune scoping: queued orphan removed, queued-but-still-referenced kept, unqueued orphan ignored (the O(delta) contract), other banks untouched, and a pair whose only queued endpoint is entity_id_2 (guards the second UNION arm).
• One test per enqueue site, asserting the queue row rather than the prune, so a dropped call fails loudly instead of leaking quietly: delta-retain chunk delete, document re-ingest, curation invalidate, single delete, plus dedup across overlapping deletes.
• Creation-time candidacy: the resolver queues what it creates; an unlinked entity is reclaimed; a linked one survives its birth candidacy.
• Time budget: an expired deadline claims nothing, leaves the rows queued, and chains a follow-up run.
• Migration: the seed enqueues every pre-existing entity, referenced or not.

Not in this PR

• The delta-retain path still enqueues no relink victims. delete_chunks_by_ids drops links whose surviving endpoints are never queued for top-up — a real gap, but a different one from this issue's.
• retry_with_backoff still treats TimeoutError as transient. With bounded statements a timeout now means something is wrong beyond this pass; the per-batch budget is 3 rather than the sweep's 8. Changing the shared predicate belongs with whoever owns the other callers.