Stop emitting the legacy first_task_scheduling_delay metric twice

airflow-core/src/airflow/models/dagrun.py

@@ -1799,9 +1799,6 @@ def _emit_true_scheduling_delay_stats_for_finished_state(self, finished_tis: lis
             else:
                 true_delay = first_start_date - self.run_after
                 if true_delay.total_seconds() > 0:
-                    stats.timing(
-                        f"dagrun.{dag.dag_id}.first_task_scheduling_delay", true_delay, tags=self.stats_tags
-                    )
                     stats.timing("dagrun.first_task_scheduling_delay", true_delay, tags=self.stats_tags)
                 if self.queued_at is not None:
                     start_delay = first_start_date - self.queued_at

airflow-core/tests/unit/models/test_dagrun.py

@@ -1122,20 +1122,27 @@ def test_no_scheduling_delay_for_nonscheduled_runs(self, stats_mock, session, te
         initial_task_states = {dag_task.task_id: TaskInstanceState.SUCCESS}
         dag_run = self.create_dag_run(scheduler_dag, task_states=initial_task_states, session=session)
         dag_run.update_state(session=session)
-        assert call(f"dagrun.{dag.dag_id}.first_task_scheduling_delay") not in stats_mock.mock_calls
+        assert (
+            call("dagrun.first_task_scheduling_delay", mock.ANY, tags=mock.ANY) not in stats_mock.mock_calls
+        )
 
     @pytest.mark.parametrize(
-        ("schedule", "expected"),
+        ("schedule", "export_legacy_names", "expected"),
         [
-            ("*/5 * * * *", True),
-            (None, False),
-            ("@once", False),
+            ("*/5 * * * *", True, True),
+            ("*/5 * * * *", False, True),
+            (None, True, False),
+            ("@once", True, False),
         ],
     )
-    def test_emit_scheduling_delay(self, session, schedule, expected, testing_dag_bundle):
+    def test_emit_scheduling_delay(
+        self, session, schedule, export_legacy_names, expected, testing_dag_bundle
+    ):
         """
         Tests that dag scheduling delay stat is set properly once running scheduled dag.
         dag_run.update_state() invokes the _emit_true_scheduling_delay_stats_for_finished_state method.
+        The legacy ``dagrun.<dag_id>.first_task_scheduling_delay`` name must only come from the metrics
+        registry, so the backend receives it exactly once and only when legacy names are exported.
         """
         dag = DAG(dag_id="test_emit_dag_stats", start_date=DEFAULT_DATE, schedule=schedule)
         dag_task = EmptyOperator(task_id="dummy", dag=dag, owner="airflow")
@@ -1179,26 +1186,30 @@ def test_emit_scheduling_delay(self, session, schedule, expected, testing_dag_bu
             ti.set_state(TaskInstanceState.SUCCESS, session=session)
             session.flush()
 
-            with mock.patch("airflow._shared.observability.metrics.stats.timing") as stats_mock:
+            backend = mock.MagicMock(spec=StatsLogger)
+            with (
+                mock.patch("airflow._shared.observability.metrics.stats._get_backend", return_value=backend),
+                mock.patch(
+                    "airflow._shared.observability.metrics.stats._export_legacy_names", export_legacy_names
+                ),
+            ):
                 dag_run.update_state(session=session)
 
-            metric_name = f"dagrun.{dag.dag_id}.first_task_scheduling_delay"
-
+            scheduling_delay_calls = [
+                c for c in backend.timing.call_args_list if "first_task_scheduling_delay" in c.args[0]
+            ]
             if expected:
                 true_delay = ti.start_date - dag_run.run_after
-                sched_delay_stat_call = call(metric_name, true_delay, tags=expected_stat_tags)
-                sched_delay_stat_call_with_tags = call(
-                    "dagrun.first_task_scheduling_delay", true_delay, tags=expected_stat_tags
+                legacy_call = call(
+                    f"dagrun.{dag.dag_id}.first_task_scheduling_delay",
+                    true_delay,
+                    tags={"run_type": DagRunType.SCHEDULED},
                 )
-                assert sched_delay_stat_call in stats_mock.mock_calls
-                assert sched_delay_stat_call_with_tags in stats_mock.mock_calls
+                modern_call = call("dagrun.first_task_scheduling_delay", true_delay, tags=expected_stat_tags)
+                expected_calls = [legacy_call, modern_call] if export_legacy_names else [modern_call]
+                assert scheduling_delay_calls == expected_calls
             else:
-                # Assert that we never passed the metric
-                sched_delay_stat_call = call(
-                    metric_name,
-                    mock.ANY,
-                )
-                assert sched_delay_stat_call not in stats_mock.mock_calls
+                assert scheduling_delay_calls == []
         finally:
             # Don't write anything to the DB
             session.rollback()