fix(worker): stop wedged retains from holding worker slots forever

[nicoloboschi]

Fixes #3002.

What the investigation found

The chain in the issue doesn't hold up, and the correction changes the fix set.

A deadlock in _db_consumer cannot park the gather. asyncio.gather(_llm_producer(), _db_consumer()) runs with return_exceptions=False, so the consumer's DeadlockDetectedError propagates immediately, and nothing in the retain path retries deadlocks (retry_with_backoff is never wrapped around phase 1/2). The task would fail and free its slot. Verified with a probe reproducing the exact shape.

But that probe exposed a real defect: when the consumer raises, the producer and every extraction task under it are never cancelled. Nine tasks stayed pending forever, blocked on chunk_queue.put() into a queue nobody drains — still holding their chunk payloads, still entitled to spend LLM permits and provider tokens on an operation that already failed.

The stage label was telling the truth. set_stage(f"llm.{provider}.{scope}{structured}") fires before the LLM semaphore acquire, and the provider only re-stamped .attempt=N/M when attempt > 0, with each attempt hard-capped by asyncio.wait_for(timeout=self.timeout). At LLM_TIMEOUT=500, a coroutine actually inside a Bedrock call cannot sit at the bare label for more than 500s — it flips to .attempt=2/N. stage_age=2405s at the bare label means op ce12eba3 never reached the provider: it was queued on an LLM concurrency semaphore. Not Bedrock, not the DB. That also explains why consolidation kept running — it isn't behind retain's per-op semaphore.

What holds permits that long: retain's retry budget multiplies. extract_facts runs an outer loop of llm_max_retries + 1, and each iteration calls llm_config.call(max_retries=llm_max_retries), which runs its own N+1 transport attempts with the permit held across the whole inner ladder — 4 × 4 × 500s ≈ 2.2 hours per chunk at the reporter's settings.

Changes

1. HINDSIGHT_API_RETAIN_WALL_TIMEOUT (default 3600s, 0 disables) — the issue's requested fix. Applied in the poller at _execute_task_inner, resolved per operation type, rather than around the gather: the poller is where the slot is held, so it bounds anything that wedges a retain, not just the one site we identified. On expiry the executor is cancelled and the operation is marked failed — which the API will retry, unlike processing, which it refuses to retry or cancel.

Uses asyncio.timeout() rather than wait_for() so cm.expired() distinguishes our ceiling firing from an inner TimeoutError bubbling out (an asyncpg command timeout). wait_for surfaces both as the same exception, and reporting a task's own timeout as a wedge sends operators after the wrong thing. The failure message carries the stage current when the ceiling hit.

Only retain is bounded: reflect self-bounds in the engine (reflect_wall_timeout), and nothing else has a reported wedge.

2. Explicit cancellation of both pipeline halves in _streaming_retain_batch, plus a finally in the producer covering tasks created before it reaches its own gather. Fixes the leak above, and the mirror case (producer fails, consumer parked on get()).

3. Stage breadcrumbs distinguish waiting from calling. .queued until the permits are held, then the in-flight label. Providers now stamp attempt 1 too, so a retry ladder is visible from the first attempt instead of only from the second.

4. bulk_insert_entities orders by LOWER(name). The issue's fix #1, downgraded to hardening: the caller already sorts (sorted_groups = sorted(groups.items()) on lowercased keys), so this isn't the headline fix. The residual hole is Python str.lower() vs PG LOWER() divergence (documented in entity_resolver for Turkish İ) — ordering in SQL makes the database's collation the single arbiter for all writers.

5. HINDSIGHT_API_DB_ACQUIRE_TIMEOUT now bounds the wait it names. It was only passed to asyncpg.create_pool(timeout=...), a connect kwarg; Pool.acquire() was called with no timeout and asyncpg's _acquire skips wait_for entirely when timeout is None. Pool exhaustion never surfaced as an error — it just hung. Included because the issue tabulates this knob as one that should have bounded the incident. Behaviour change: an exhausted pool now raises after the configured wait (retried up to 3× by acquire_with_retry) instead of blocking indefinitely; 0 restores the old behaviour.

Deliberately not fixed

The retry-ladder multiplication. It's real and it's my best candidate for what saturated the permits, but capping it would turn currently-succeeding slow retains into failures, and the wall timeout already defuses the systemic damage — cancelling the task releases the permits, so a ladder can no longer hold them past the ceiling. Worth a separate issue.

Tests

4 new files, 15 tests:

• test_worker_wall_timeout.py — the ceiling cancels the executor and marks failed; the message carries the stage; fast tasks untouched; an inner TimeoutError is not reported as a wedge; non-retain types stay unbounded; 0 disables.
• test_retain_pipeline_cancellation.py — drives the real pipeline with a DB error injected into the consumer while extractions are in flight. Confirmed it fails without the fix, reproducing the actual leak (producer + 3 _extract_one tasks surviving a failed operation).
• test_llm_stage_breadcrumbs.py — stage says .queued while blocked on a saturated semaphore, and names the in-flight call once the permit is granted.
• test_pool_acquire_timeout.py — acquire() and transaction() pass the configured timeout; 0 maps to asyncpg's unbounded None.

Full sweep of the worker, retain, entity-resolver, provider and pool suites green; lint.sh and ty check clean.

Open questions for the reporter

Neither blocks this PR, but both would confirm the incident is closed:

• Is HINDSIGHT_API_RETAIN_LLM_MAX_CONCURRENT set, and to what? That's the choke point in this reading, and its value predicts how fast a worker re-wedges.
• In /llm-requests, are there retain rows in that 50-minute window with errors/timeouts (not just successes)? That distinguishes "permits held by the retry ladder" from "permits leaked".