Fix scheduler crash from dag_run listener Connection/Variable lookups

[udsy19]

MetastoreBackend.get_connection() and get_variable() are decorated with @provide_session. Called with no session — the shape every public caller uses, including BaseHook.get_connection(conn_id) and Variable.get(key) from a dag_run listener hook (on_dag_run_running/on_dag_run_success/on_dag_run_failed) — the decorator resolves the session to settings.Session(), a thread-local scoped session.

The scheduler's own _do_scheduling loop runs dag_run listener hooks in-process (DagRun.notify_dagrun_state_changed) while holding that exact same scoped session under prohibit_commit, a guard that raises on any unexpected commit to protect its HA locking. The @provide_session wrapper commits on exit, so a listener that reads a Connection or Variable trips the guard with RuntimeError("UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!"). That error is swallowed per-backend inside Connection.get_connection_from_secrets/Variable.get_variable_from_secrets, so it first surfaces as a spurious AirflowNotFoundException for a Connection/Variable that genuinely exists — and then corrupts the scheduler's session for the rest of that scheduling pass, observed cascading into DetachedInstanceError on unrelated DagRun objects processed in the same batch.

This is the root cause of #39646: a listener plugin calling BaseHook.get_connection()/Variable.get() from on_dag_run_running/on_dag_run_success/on_dag_run_failed gets AirflowNotFoundException for connections/variables that are defined, while the identical call from on_task_instance_* (which runs in a separate worker process, not the scheduler) works fine.

Impact: crash-on-valid-input

WHO reaches this / entry point: any dag_run listener plugin registered the documented way (a hookimpl implementing on_dag_run_running/on_dag_run_success/on_dag_run_failed, exactly as shown in the Listeners how-to doc and in #39646's own repro) that calls the standard BaseHook.get_connection(conn_id) or Variable.get(key) — no session argument, which is the only shape a plugin author has, since the hookspec passes no session. Triggered by: the scheduler reaching SchedulerJobRunner._do_scheduling -> _schedule_all_dag_runs/_start_queued_dagruns -> DagRun.notify_dagrun_state_changed for any DagRun transitioning to running/success/failed while such a listener is registered -- an entirely ordinary scheduling pass, valid input by construction (a real, already-committed Connection/Variable). What they observe: first a wrong AirflowNotFoundException for a Connection/Variable that exists, then (per the DetachedInstanceError repro below) the scheduler's own session breaks for the rest of that scheduling pass.

Fix

Give get_connection()/get_variable() a genuinely independent, non-scoped session (create_session(scoped=False)) when no session is passed, instead of the scoped one @provide_session's default would resolve to. A listener's read no longer aliases into — and commits — the scheduler's locked session. @provide_session itself can't be reused for this (its NEW_SESSION path is exactly the scoped session that's the problem), so the session handling is inlined; the check-new-session-in-provide-session prek hook's None-typed exemption (session: Session | None = None) covers the new shape.

This is a narrower, session-less-caller-focused companion to #71968/#72121, which added an optional session-reuse parameter for internal scheduler call sites that already have a session in scope (Variable.get_variable_from_secrets). Neither of those covers BaseHook.get_connection()/Variable.get() called with no session at all, which is the public, session-agnostic shape every dag_run listener (and every non-scheduler caller) actually uses — #71968's own body names Connection.get_connection_from_secrets as exactly this left-open follow-up.

Testing

Two new tests in TestMetastoreBackendSessionSafety (tests/unit/always/test_secrets_metastore.py), mirroring the prohibit_commit-under-guard pattern from #67980's task_instance_mutation_hook regression tests:

• test_get_connection_survives_prohibit_commit_without_explicit_session
• test_get_variable_survives_prohibit_commit_without_explicit_session

Both reproduce the exact scheduler shape (with create_session() as session: with prohibit_commit(session): MetastoreBackend().get_connection(...)/get_variable(...), no session= argument) and assert the lookup succeeds.

Negative control: reverting only metastore.py to main and running the class (pytest tests/unit/always/test_secrets_metastore.py::TestMetastoreBackendSessionSafety):

tests/unit/always/test_secrets_metastore.py::TestMetastoreBackendSessionSafety::test_get_connection_survives_prohibit_commit_without_explicit_session FAILED
tests/unit/always/test_secrets_metastore.py::TestMetastoreBackendSessionSafety::test_get_variable_survives_prohibit_commit_without_explicit_session FAILED
E   RuntimeError: UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!
2 failed, 5 passed in 5.04s

With the fix, all 7 tests in the class pass (the 5 pre-existing tests plus the 2 new ones), 7 passed. Also re-verified the original issue's exact scenario end to end against a real SchedulerJobRunner._do_scheduling() run with a registered on_dag_run_success listener calling BaseHook.get_connection()/Variable.get() on a real, committed Connection/Variable: fails with DetachedInstanceError on pristine main, passes with this fix.

closes: #39646

Signed-off-by: Udaya Tejas udayatejas2004@gmail.com

---

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

• Yes — Claude Code (Sonnet 5)

Generated-by: Claude Code (Sonnet 5) following the guidelines