Pass the scheduler's session to set_state when processing executor events

process_executor_events calls ti.set_state() without a session when the
Dag of a finished task instance cannot be loaded. set_state is
@provide_session and settings.Session is scoped, so create_session()
returned the scheduler's own session and committed and closed it on
exit, in the middle of the batch.

A second call site with the same bug (a cleared task instance reported
terminated) was superseded by #73554, which replaced that whole code
path with TaskInstance.complete_restart(), already passing the
scheduler's session correctly.

Author: namanjain24-sudo

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -1671,7 +1671,7 @@ def process_executor_events(
                     task = dag.get_task(ti.task_id)
                 except Exception:
                     cls.logger().exception("Marking task instance %s as %s", ti, state)
-                    ti.set_state(state)
+                    ti.set_state(state, session=session)
                     continue
                 ti.task = task
                 if task.has_on_retry_callback or task.has_on_failure_callback:

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -759,6 +759,39 @@ def test_retired_attempt_events_do_not_modify_replacement(
         assert ti.try_number == 2
         assert ti.external_executor_id == ("current_worker" if include_current else "replacement")
 
+    def test_process_executor_events_sets_state_in_callers_transaction(self, dag_maker):
+        """
+        Setting a task instance's state must not commit the caller's transaction.
+
+        ``settings.Session`` is scoped, so ``ti.set_state()`` without ``session`` resolved to the
+        scheduler's own session and ``create_session()`` committed and closed it on exit. That released
+        the scheduler's row locks mid-batch and detached the task instances still to be processed, so
+        changes made to them afterwards were never written.
+
+        This covers the Dag-not-found path: the Dag can't be loaded, so the task instance is marked
+        with the executor's reported state directly, without a session it would otherwise resolve to
+        the scheduler's own scoped session and commit early.
+        """
+        session = settings.Session()
+        with dag_maker(dag_id="test_executor_events_callers_transaction", fileloc="/test_path1/"):
+            task1 = EmptyOperator(task_id="test_task", retries=2)
+        ti1 = dag_maker.create_dagrun().get_task_instance(task1.task_id)
+        ti1.state = TaskInstanceState.QUEUED
+        session.merge(ti1)
+        session.commit()
+
+        executor = MockExecutor(do_update=False)
+        job_runner = SchedulerJobRunner(Job(), executors=[executor])
+        job_runner.scheduler_dag_bag = mock.MagicMock()
+        job_runner.scheduler_dag_bag.get_dag_for_run.side_effect = Exception("failed")
+        executor.event_buffer[ti1.key] = State.FAILED, None
+
+        job_runner._process_executor_events(executor=executor, session=session)
+        session.rollback()
+
+        ti1.refresh_from_db(session=session)
+        assert ti1.state == TaskInstanceState.QUEUED
+
     @mock.patch("airflow.jobs.scheduler_job_runner.TaskCallbackRequest")
     @mock.patch("airflow._shared.observability.metrics.stats._get_backend")
     def test_process_executor_events_with_no_callback(self, mock_get_backend, mock_task_callback, dag_maker):