fix(worker): serialise graph_maintenance per bank at claim time

[nicoloboschi]

Summary

graph_maintenance has no per-bank serialisation at claim time. Two concurrent runs against one bank do no extra work — the payload carries only bank_id, run_graph_maintenance_job discards the request context, and the relink pass drains the whole queue — so both runs are the same bank-wide sweep.

It is worse than redundant. claim_graph_maintenance_batch locks queue rows FOR UPDATE without SKIP LOCKED, deliberately, because it is written assuming one runner per bank (its docstring said submit-time dedup guaranteed that; it does not — dedup only inspects 'pending' rows). So the runs convoy on each other's row locks while each holds a worker slot. Reported in #3230: nine concurrent runs on one bank, a six-deep blocking chain, retain starved to zero for ~4h.

The fix

The same guarantee consolidation already has, expressed as a predicate on the existing claim queries and shared between both dialects (graph_maintenance_bank_serialization_sql in db/ops.py).

Two things differ from the consolidation form, both forced by the shape of the problem:

It is a predicate, not a separate claim phase. A trailing phase works for consolidation only because it has a reserved-slot floor of 2 (WORKER_SLOT_TYPE_DEFAULTS); graph_maintenance has 0, and the poller's fairness pass calls claim_tasks with shared_limit=1 (poller.py:459). A single pending retain would then consume the shared slot in the generic query and graph_maintenance would never be claimed at all — the queue grows, links are never topped up. As a predicate on that same query it keeps its place in the created_at ordering. test_not_starved_by_newer_pending_work covers this.

It also takes at most one same-bank row per batch. Excluding busy banks alone does not cover that: with several pending rows and nothing yet processing, one batch claims them all — the convoy, unchanged. Several pending rows per bank are reachable through the recovery paths: recover_own_tasks resets every processing row for a worker back to pending in one statement, and _schedule_retry / _defer_operation / hindsight-admin recover each restore rows independently. That, not the submit path, is where #3230's 22 simultaneously-pending rows come from — submit-time dedupe_by_bank is atomic (it takes FOR NO KEY UPDATE on the bank row before the check), so it yields at most one pending row per bank.

Inherited caveat, unchanged from consolidation: a row wedged in processing holds its bank until something releases it — hindsight-admin recover, or a restart with a stable HINDSIGHT_API_WORKER_ID so recover_own_tasks matches it. That is a general gap in claim recovery (there is no time-based stale-claim reaper for any operation type), not something specific to graph_maintenance, and it is not addressed here.

No advisory locks (#2817), no new table, no new config flag.

Relationship to #3231

Same issue, different mechanism. #3231 adds a dedicated claim phase after the generic shared-pool query, which introduces the starvation described above. This supersedes it.

Tests

tests/test_graph_maintenance_claim_serialization.py, real Postgres, 9 tests. These call ops.claim_tasks directly rather than WorkerPoller.claim_batch, so the slot limits under test are exact and not a function of ambient in-flight work in the shared test DB.

Fail on main, pass with the fix:

• test_not_claimed_while_bank_has_run_in_flight
• test_single_batch_claims_at_most_one_per_bank (also asserts it is the oldest row)
• test_idle_bank_still_claimed_while_another_is_busy
• test_reserved_pool_is_serialised_too
• test_reserved_and_shared_phases_do_not_double_claim

Guard against over-blocking (pass both ways — these are what a naive guard breaks):

• test_not_starved_by_newer_pending_work
• test_retry_blocked_older_row_does_not_block_a_claimable_one
• test_other_operation_types_unaffected

Dialect parity: test_guard_survives_the_oracle_sql_rewrite asserts on _rewrite_pg_to_oracle's output directly (NOW() → SYSTIMESTAMP in both the outer query and the correlated subquery, != ALL expanded, LIMIT → ROWNUM injected on the outer WHERE rather than the subquery's). Oracle integration tests need an ORACLE_TEST_DSN I don't have, so this is the check that the shared fragment stays valid there.

$ uv run pytest -n0 tests/test_graph_maintenance_claim_serialization.py tests/test_worker.py
                                                                            -> 113 passed
$ uv run pytest -n0 tests/test_graph_maintenance.py tests/test_graph_maintenance_deadlock.py \
      tests/test_graph_maintenance_queue_race.py tests/test_enqueue_graph_maintenance_ordered.py
                                                                            -> 32 passed
$ ./scripts/hooks/lint.sh                                                   -> All lints passed
$ uv run ty check hindsight_api/                                            -> All checks passed

Fixes #3230