Fix Calendar view hanging for Dags with high-frequency cron schedules (apache#71263)

The planned-runs computation for cron timetables iterates croniter until the
year boundary with no cap, while the generic-timetable path stops at
MAX_PLANNED_RUNS. For "*/5 * * * *" that is ~105k iterations (~2s) per calendar
request, ~520k (~7s) for a minutely cron, and ~31M (minutes of CPU) for a
seconds-resolution cron - enough for any user with Dag read access to pin an
API server worker just by opening the Calendar tab. Apply the same
MAX_PLANNED_RUNS cap the generic path has used since the endpoint was added.

Author: theJohnnyBrown

airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import collections
+import itertools
 from collections.abc import Iterator, Sequence
 from datetime import datetime
 from typing import Literal, cast
@@ -221,7 +222,10 @@ def _calculate_cron_planned_runs(
             ret_type=datetime,
         )
 
-        for dt in dates_iter:
+        # Cap the iteration like _calculate_timetable_planned_runs does; a high-frequency
+        # expression (e.g. "* * * * *", or a seconds-resolution cron) would otherwise take
+        # hundreds of thousands of steps before hitting the year boundary.
+        for dt in itertools.islice(dates_iter, self.MAX_PLANNED_RUNS):
             if dt is None or dt.year != year:
                 break
             if dag.end_date and dt > dag.end_date:

airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_calendar.py

@@ -24,6 +24,7 @@
 from sqlalchemy.orm import Session
 
 from airflow._shared.timezones import timezone
+from airflow.api_fastapi.core_api.services.ui.calendar import CalendarService
 from airflow.models.deadline import Deadline
 from airflow.providers.standard.operators.empty import EmptyOperator
 from airflow.sdk import CronPartitionTimetable
@@ -326,6 +327,44 @@ def test_hourly_calendar_partitioned(self, test_client, query_params, result):
         assert body == result
 
 
+class TestCalendarPlannedRunsCap:
+    """A high-frequency cron must stop at MAX_PLANNED_RUNS instead of iterating to the year boundary."""
+
+    DAG_NAME = "test_minutely_dag"
+
+    @pytest.fixture(autouse=True)
+    @provide_session
+    def setup_dag_runs(self, dag_maker, *, session: Session = NEW_SESSION) -> None:
+        clear_db_runs()
+        clear_db_dags()
+        with dag_maker(
+            self.DAG_NAME,
+            schedule="* * * * *",
+            start_date=datetime(2025, 6, 1),
+            catchup=False,
+            serialized=True,
+            session=session,
+        ):
+            EmptyOperator(task_id="test_task1")
+        dag_maker.create_dagrun(
+            run_id="run_1",
+            state=DagRunState.SUCCESS,
+            logical_date=datetime(2025, 6, 1),
+        )
+        dag_maker.sync_dagbag_to_db()
+        session.commit()
+
+    def teardown_method(self) -> None:
+        clear_db_runs()
+        clear_db_dags()
+
+    def test_planned_runs_capped_for_high_frequency_cron(self, test_client):
+        response = test_client.get(f"/calendar/{self.DAG_NAME}")
+        assert response.status_code == 200
+        planned = [r for r in response.json()["dag_runs"] if r["state"] == "planned"]
+        assert sum(r["count"] for r in planned) == CalendarService.MAX_PLANNED_RUNS
+
+
 _CALLBACK_PATH = "tests.unit.api_fastapi.core_api.routes.ui.test_calendar._noop_callback"