(airflow) branch main updated: Sync the session's copy with the DB for removed task instances when scheduling (#75974)

[ash]

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

ashb 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 56b44decc8f Sync the session's copy with the DB for removed task 
instances when scheduling (#75974)
56b44decc8f is described below

commit 56b44decc8f9b31b04cc55f47d32a77210c59a91
Author: Christos Bisias <christos...@gmail.com>
AuthorDate: Fri Oct 30 11:59:04 2026 +0200

    Sync the session's copy with the DB for removed task instances when 
scheduling (#75974)
---
 airflow-core/src/airflow/models/dagrun.py     | 1 -
 airflow-core/tests/unit/models/test_dagrun.py | 5 +++++
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/airflow-core/src/airflow/models/dagrun.py 
b/airflow-core/src/airflow/models/dagrun.py
index 7be6c87be56..1880e2424ad 100644
--- a/airflow-core/src/airflow/models/dagrun.py
+++ b/airflow-core/src/airflow/models/dagrun.py
@@ -1625,7 +1625,6 @@ class DagRun(Base, LoggingMixin):
                         TI.state != TaskInstanceState.REMOVED,
                     )
                     .values(state=TaskInstanceState.REMOVED)
-                    .execution_options(synchronize_session=False)
                 )
             return finished
 
diff --git a/airflow-core/tests/unit/models/test_dagrun.py 
b/airflow-core/tests/unit/models/test_dagrun.py
index 8d583b4c0da..62cae9bce0d 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -5209,7 +5209,12 @@ def 
test_scheduling_decisions_mark_orphaned_finished_tis_removed(dag_maker, sess
     )
     assert (finished_ti.region_id, finished_ti.region_index) == 
(stays.region_id, stays.region_index)
     assert finished_ti.dag_version_id == stays.dag_version_id
+    # This read returns the ORM copy loaded in memory, and the assertion checks
+    # that the bulk UPDATE also synchronised the session.
+    assert dag_run_1.get_task_instance("vanishes", session=session).state == 
TaskInstanceState.REMOVED
     session.expire_all()
+    # After expiring the session, the following reads hit the database,
+    # and the assertions check what the UPDATE actually wrote.
     assert dag_run_1.get_task_instance("vanishes", session=session).state == 
TaskInstanceState.REMOVED
     assert dag_run_1_later.get_task_instance("vanishes", 
session=session).state == TaskInstanceState.SUCCESS
     assert dag_run_2.get_task_instance("vanishes", session=session).state == 
TaskInstanceState.SUCCESS