Fix inc task duration on skipped tasks

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

@@ -2137,8 +2137,7 @@ def _schedule_dag_run(
                     .where(TI.state.in_(State.unfinished))
                 )
                 for task_instance in unfinished_task_instances:
-                    task_instance.state = TaskInstanceState.SKIPPED
-                    session.merge(task_instance)
+                    task_instance.set_state(TaskInstanceState.SKIPPED, session)
                 session.flush()
                 self.log.info("Run %s of %s has timed-out", dag_run.run_id, dag_run.dag_id)
 

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

@@ -3264,6 +3264,28 @@ def test_dagrun_timeout_callbacks_are_stored_in_database(self, dag_maker, sessio
 
         assert callback == expected_callback
 
+    def test_dagrun_timeout_sets_task_end_date_and_duration(self, dag_maker, session):
+        """Test that TaskInstance.set_state(SKIPPED) sets end_date and duration."""
+        with dag_maker(dag_id="test_set_state_skipped", session=session):
+            EmptyOperator(task_id="test_task")
+
+        dr = dag_maker.create_dagrun()
+        ti = dr.get_task_instance("test_task", session)
+
+        ti.state = TaskInstanceState.RUNNING
+        ti.start_date = timezone.utcnow()
+        session.flush()
+
+        assert ti.end_date is None
+        assert ti.duration is None
+
+        ti.set_state(TaskInstanceState.SKIPPED, session)
+
+        assert ti.end_date is not NotImplemented
+        assert ti.duration is not None
+        assert ti.duration >= 0
+        assert ti.start_date < ti.end_date
+
     @conf_vars({("scheduler", "use_job_schedule"): "False"})
     def test_dagrun_callbacks_commited_before_sent(self, dag_maker):
         """