[nicoloboschi]

Summary

Fixes the duplicate-pending-consolidation pile-up in #1842 (reporter saw
PENDING_BREAKDOWN: total=3 claimable=0 retry_blocked=3 — several pending ops for
one bank, all parked on backoff, leaving the bank with no claimable work).

Root cause — a non-atomic endpoint. _submit_async_operation's dedup is a
check-then-INSERT split across two separate connection acquisitions (pre-check
SELECT then a later INSERT in a different block) — not even one transaction. Under
READ COMMITTED, two concurrent submits both see no pending row and both insert. A
manual POST /consolidate loop racing retain-driven submits and the round-limit
re-queue therefore leaks duplicate pending consolidation ops for a bank; those extras
each enter retry-backoff and pile up as retry_blocked. Everything downstream — the
dedup-guard failing ops, the idle banks — is a consequence of the endpoint creating
duplicates it promised not to.

Fix — make the submit atomic. Run the dedup check-and-insert in a single
transaction that first locks the bank row, so concurrent submits for the same bank
serialize and the second observes the first's pending row. The lock releases on
commit, before submit_task runs.

This enforces the intended invariant — at most one pending full-bank consolidation
per bank — at the point of creation, rather than cleaning up duplicates downstream.
Combined with the already-merged #1857 (rounds auto-continue) and #1854 (5s backoff),
the pile-up can't form and the external re-POST workaround is no longer needed.

Lock choice — FOR NO KEY UPDATE, not FOR UPDATE

async_operations has an FK to banks, so every async-op insert for the bank (a
scoped consolidation, a batch-retain op, a webhook delivery, …) takes a FOR KEY SHARE
lock on the bank row. FOR UPDATE conflicts with FOR KEY SHARE and would briefly
block all of those during the submit. FOR NO KEY UPDATE conflicts only with itself —
so two submits serialize (what we want) while those inserts proceed unblocked. The
Oracle SQL rewriter maps FOR NO KEY UPDATE → FOR UPDATE (Oracle has only the
latter, and there it does not block indexed-FK child inserts).

Scopes

Consolidation scopes are handled in both directions so a full-bank sweep and a scoped
run never suppress each other:

• Scoped submits (observation_scopes set) pass dedupe_by_bank=False — they skip
  the lock and dedup entirely and always run (existing intent: "a targeted run
  should not be merged into a pending full-bank sweep").
• Unscoped submits dedup only against an existing unscoped pending op. A pending
  scoped consolidation covers only its tag subset, so it must not swallow a full-bank
  sweep. The scope check is done in Python because the JSON predicate isn't portable
  (Oracle's JSON_VALUE returns NULL for the array-valued observation_scopes).

SELECT … FOR NO KEY UPDATE keeps this app-level, cross-dialect, and consistent with
existing locking in the engine. No schema change.

Testing

• tests/test_consolidation_submit_atomic_dedup.py:
  • 5 concurrent unscoped submits on an empty bank → exactly one pending row, all calls share the op id, four deduplicated
  • sequential submit while one is pending → returns the existing op
  • unscoped submit is NOT deduped against a pending scoped op (full-bank still runs); a second unscoped then dedups against the unscoped pending
  • two scoped submits → two pending rows (exemption preserved)
• tests/test_db_abstraction.py::TestOracleQueryRewriter::test_for_no_key_update_rewrite — FOR NO KEY UPDATE → FOR UPDATE on Oracle

All pass; ruff + ty clean. (The submit_task inline executor is neutralized in-test so the pending rows survive inspection under the sync backend.)