(airflow) branch main updated: Ensure that DB migrations run in a single connection. (#65231)

[as...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ash pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new e3fea3a5556 Ensure that DB migrations run in a single connection. (#65231)
e3fea3a5556 is described below

commit e3fea3a55564fb4b0daeafe43c521f0ee2001f7a
Author: Ash Berlin-Taylor <as...@apache.org>
AuthorDate: Tue Apr 14 22:40:34 2026 +0100

    Ensure that DB migrations run in a single connection. (#65231)
    
    This was discovered by running with a custom External DB manager that had some
    gnarly queries that ended up being locked behind this transaction.
    
    `_single_connection_pool` replaces `settings.engine` with a
    SingletonThreadPool engine. But `work_session` was created before that — it
    still holds an internal reference to the old engine object.
    _single_connection_pool has no way to rebind work_session.
    
    So when _get_current_revision(session=work_session) runs on line 1203 — inside
    the _single_connection_pool() block — it calls session.connection() which goes
    through the old engine, not the SingletonThreadPool. The old engine's pool was
    disposed and recreated empty by engine.dispose(), so it creates a brand new
    connection. That connection is completely outside _single_connection_pool's
    control.
    
    _single_connection_pool guarantees one connection on the new engine. It can't
    prevent work_session from creating connections on the old one. The name is a
    bit of a lie — it's really "single connection pool for new code that uses
    settings.engine", not "single connection total."
---
 airflow-core/src/airflow/utils/db.py | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py
index c16a797b1a2..9293f1c49ac 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -1204,6 +1204,13 @@ def _run_upgradedb(
         with _configured_alembic_environment() as env:
             source_heads = env.script.get_heads()
 
+        # End the read-only transaction from _get_current_revision before
+        # external DB manager migrations, which may run DDL that is blocked by
+        # open transactions (e.g. CREATE INDEX CONCURRENTLY). The advisory lock
+        # from create_global_lock() is unaffected: it is session-level and held
+        # on a separate connection.
+        work_session.rollback()
+
         if current_revision == source_heads[0] and not _SKIP_EXTERNAL_DB_MANAGERS_UPGRADE.get():
             external_db_manager = RunDBManager()
             external_db_manager.upgradedb(work_session, use_migration_files=use_migration_files)