[bnspopi]

Trigger.clean_unused() retries its UPDATE task_instance SET trigger_id=NULL through run_with_db_retries(), but reuses the same session across attempts without rolling it back. When the metadata database connection drops mid-statement, the transaction is left invalid, and the retried session.execute() raises PendingRollbackError ("Can't reconnect until invalid transaction is rolled back"). That is not one of the retried exception types, so it propagates through TriggerRunnerSupervisor.run and the triggerer exits — a pod restart for what was a momentary disconnect. Traceback and reporter's logs are in #74614.

The fix is the one retry_db_transaction already applies for the same reason: roll the session back when the statement raises DBAPIError/StaleDataError, before the retry runs. That decorator cannot be reused here directly — its docstring says it must not be combined with @provide_session, which clean_unused needs — so the rollback is inline, catching the same two exception types the retry predicate matches.

Nothing changes on the happy path: no new statement, no signature change, no new config.

Tests: test_clean_unused_rolls_back_before_retrying makes the first session.execute raise OperationalError, then asserts the session was rolled back once and the cleanup still completed on the retry.

Scope I left alone deliberately:

• The DELETE after the loop is not inside the retry at all, so a disconnect there still surfaces to the caller. That is pre-existing and a separate decision about where the retry boundary should sit.
• SchedulerJobRunner.check_trigger_timeouts and check_awaiting_input_timeouts use the same un-rolled-back inline retry pattern. Same latent bug, different component; happy to follow up if you want them in one sweep.

Relationship to #70961: that PR rewrites clean_unused to open a fresh session per attempt and adds batching plus a new [triggerer] unreferenced_triggers_cleanup_batch_size option, which would also resolve this crash as a side effect. This is the minimal fix for the crash on its own — two files, no config or signature change — so it can land (and backport) without waiting on that discussion. If #70961 goes in first, this becomes unnecessary and I will close it.

I could not run the test suite on this machine, so the first run will be CI here; happy to follow up on anything it flags. Let me know if you want a newsfragment — I did not have the PR number when committing.

closes: #74614
related: #70961

---

Was generative AI tooling used to co-author this PR?

• Yes (Claude Code)

Generated-by: Claude Code following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[RishabhBajaj97]

this looks like a duplicate of the PR I had raised earlier for the issue, can you please close this?

[namanjain24-sudo]

Current main no longer has the code this patches. #75096 (merged 19 Oct) moved the retry out of Trigger.clean_unused into TriggerRunnerSupervisor.clean_unused:

def clean_unused(self) -> None:
    """Remove triggers that are no longer needed, retrying database errors in a new transaction."""
    for attempt in run_with_db_retries():
        with attempt, create_session() as session:
            Trigger.clean_unused(session=session)

Trigger.clean_unused itself no longer retries at all — its docstring now reads "Database errors are not retried; to retry, run the call again in a new transaction."

That removes the mechanism this PR describes rather than patching it: create_session() runs inside the loop, so every attempt gets its own session and there is no invalidated transaction left for the next attempt to reuse. test_clean_unused_retries_failed_batches_in_new_transactions asserts exactly that — three delete attempts across three separate transactions. There is also test_clean_unused_recovers_after_the_database_connection_drops, which terminates the backend mid-DELETE with pg_terminate_backend; it is marked @pytest.mark.backend("postgres"), so it runs on the Postgres CI jobs rather than on a default local run.

This branch was last updated on 13 Oct, before that landed, so it needs a rebase to show the current state.

Since #74614 is still open, the more useful next step is probably to check whether it still reproduces on current main and where the escape is now, rather than rebasing this onto a function that no longer retries.

---

Drafted-by: Claude Code (Opus 5); reviewed by @namanjain24-sudo before posting