[HIGH — silent failure / observability] When the retry succeeds, the original OperationalError/InterfaceError is gone forever — no exc_info, and the exception type/message isn't interpolated. You can't tell a benign PgBouncer idle reap from a recurring statement timeout or deadlock that the retry is silently papering over. The message also hard-asserts "lost its DB connection", which (see line 163) will be a lie for many of the conditions this except actually catches.

Suggested:

except (psycopg2.OperationalError, psycopg2.InterfaceError) as exc:
    if attempt >= _BARRIER_WRITE_ATTEMPTS:
        raise
    logger.warning(
        "PgBarrier: %s — DB write failed (%s: %s); cached connection "
        "was likely stale, reconnecting and retrying (attempt %d/%d)",
        what, type(exc).__name__, exc, attempt, _BARRIER_WRITE_ATTEMPTS,
    )

[MEDIUM — broad catch + no backoff] psycopg2.OperationalError is the base class for many server-side, non-connection conditions: QueryCanceled (statement_timeout), DeadlockDetected/SerializationFailure (via TransactionRollbackError), TooManyConnections, AdminShutdown, etc. All are now caught, mislabeled (line 170), and immediately re-run with zero backoff — which just adds load against an already-struggling DB. Re-running the idempotent UPSERT once is correctness-safe, so this isn't a bug, but consider either (a) narrowing the retry to genuine connection-drop signals (e.g. only when _cursor actually discarded the cached conn), or at minimum (b) excluding TransactionRollbackError so deadlocks aren't blindly hammered. A small fixed sleep before the retry would also widen the self-heal window during a DB failover (the idle-reap case succeeds instantly either way).

[MEDIUM — comment inaccuracy] "fire the callback twice" isn't the actual failure mode. Decrements each run in their own committed txn, serialized by the row lock, and the callback fires only when a decrement's RETURNING observes remaining == 0. A double-applied decrement skips over 0 (1→0 committed-but-unacked, retry 0→-1 → returns -1 → abandoned branch), so two observations of exactly 0 can't occur. The real hazards — still justifying the exclusion — are firing the callback prematurely with incomplete results, or skipping past 0 and stranding the barrier to expiry. Suggest: "...double-counts: it can drive remaining to 0 early and fire the callback prematurely with incomplete results, or skip past 0 and strand the barrier."

[LOW — comment precision] "same row, same state" slightly overstates idempotency: the UPSERT's DO UPDATE re-runs created_at = now() and expires_at = ... (lines 375-376), so timestamps advance on a retry. Harmless here (retries are milliseconds apart, execution_id is the PK so no row duplication), but a future reader relying on created_at for staleness/reaper logic could be surprised. Suggest softening to "→ same row, remaining/results reset identically (timestamps refresh, harmlessly)."

[MEDIUM — type design] The defining invariant — that operation must be idempotent — lives entirely in the docstring; Callable[[Any], None] would happily accept the non-idempotent decrement. The -> None return is a nice structural guard (it blocks passing a RETURNING-reading op like claim_batch), but idempotency itself is unenforceable in Python's type system. Pragmatic mitigations: keep it private (done) and consider a name that makes the contract impossible to miss at call sites, e.g. _run_idempotent_pre_dispatch_write.

Also minor: the cursor is typed Any here, but the module already imports connection as PgConnection under TYPE_CHECKING — adding cursor as PgCursor and typing Callable[[PgCursor], None] would tighten intent at zero runtime cost (worth doing file-wide for consistency).

[LOW — polish] _BARRIER_WRITE_ATTEMPTS = 2 means "total attempts", so 2 = "one retry" — a name/value relationship only the docstring disambiguates. Suggest _BARRIER_WRITE_ATTEMPTS: Final = 2 # total attempts: 1 initial + 1 retry. Keeping it a literal (not env-driven) is the right call — it keeps the idempotency bound from being weakened operationally.

[LOW — redundancy] This 3-line comment restates the _run_idempotent_write docstring it points to (directly above the call). It's the kind of duplicated rationale that rots independently of the source of truth. Suggest trimming to a one-liner pointer, e.g. # Idempotent + pre-dispatch → safe to retry; see _run_idempotent_write.

[HIGH — test gap] All three tests exercise _run_idempotent_write in isolation with a no-op operation — none go through the actual enqueue() path the fix protects. Concrete regression this misses: reverting enqueue to the old inline with _cursor() as cur: (the exact pre-fix code) leaves _run_idempotent_write as dead code and silently removes the self-heal, yet every test here still passes. Suggest an integration test using the real barrier_db fixture: set pg_barrier._local.conn to a stub whose first .execute raises OperationalError, monkeypatch create_pg_connection to return the real test connection, call PgBarrier().enqueue(...), then assert (a) the pg_barrier_state row exists exactly once with remaining == N, and (b) header dispatch was called exactly N times. This pins both the self-heal and the "headers still dispatched, no double-dispatch" idempotency claim end-to-end.

[MEDIUM — test gap] Both except clauses (this fix at pg_barrier.py:163 and _cursor at :119) catch (OperationalError, InterfaceError), but all three tests raise only OperationalError. InterfaceError ("connection already closed") is arguably the more common psycopg2 symptom of a stale cached socket — exactly this PR's target scenario. Narrowing either clause to drop InterfaceError would break the self-heal with no failing test. Suggest parametrizing this test over [psycopg2.OperationalError("server closed..."), psycopg2.InterfaceError("connection already closed")]. While here, also asserting dead.commits == 0 (no partial commit on the failed attempt) and a caplog warning would lock the "log on retry, not on success" contract.

[LOW — dead test scaffolding] _FakeConn increments self.rollbacks (line 159) and self.executes (line 149) but neither counter is ever asserted anywhere — test 1 counts invocations via its own attempts list. The rollback() method must stay (called by _cursor's recovery path), but the two counters are dead state. Either drop them, or better, assert on them (e.g. dead.executes == 1, healthy.executes == 1) to tightly bound the retry to exactly one extra attempt. Also closed is bool here but real psycopg2 connection.closed is an int (0/1/2) — fine for the truthiness check, minor fidelity drift.