diff --git a/airflow-core/newsfragments/71704.bugfix.rst b/airflow-core/newsfragments/71704.bugfix.rst new file mode 100644 index 0000000000000..5cd2aed7ec0dd --- /dev/null +++ b/airflow-core/newsfragments/71704.bugfix.rst @@ -0,0 +1 @@ +The scheduler's Dag cache is now a bounded LRU of 512 versions, so scheduler memory no longer grows with every Dag version the process has ever seen. diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index a6343c4ae6c71..d03b1d456f466 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -154,6 +154,18 @@ TASK_STUCK_IN_QUEUED_RESCHEDULE_EVENT = "stuck in queued reschedule" """:meta private:""" +SCHEDULER_DAG_CACHE_SIZE = 512 +""" +Max deserialized Dag versions the scheduler keeps in memory. + +The scheduler reaches its DagBag through the Dag version of each active Dag run, so an +unbounded cache retains every version the process has ever seen and grows for the life of +the process. Sized to sit above the versions-with-runs-in-flight working set of a typical +deployment, so eviction costs a re-fetch only where that working set is genuinely larger. + +:meta private: +""" + # Per-tick cap on pending AssetPartitionDagRun rows the scheduler evaluates. # Bounds the per-tick transaction so executor heartbeats and regular scheduling # aren't starved; remaining APDRs drain across subsequent ticks. @@ -348,7 +360,7 @@ def __init__( if log: self._log = log - self.scheduler_dag_bag = DBDagBag(load_op_links=False) + self.scheduler_dag_bag = DBDagBag(load_op_links=False, cache_size=SCHEDULER_DAG_CACHE_SIZE) # Set of (dag_id, asset_name, asset_uri) tuples for trigger policies that # are permanently unreachable for the rollup window's cardinality — the diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index 72ed2065bdbff..419b459bc3619 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -57,7 +57,7 @@ from airflow.executors.executor_utils import ExecutorName from airflow.executors.local_executor import LocalExecutor from airflow.jobs.job import Job, run_job -from airflow.jobs.scheduler_job_runner import SchedulerJobRunner +from airflow.jobs.scheduler_job_runner import SCHEDULER_DAG_CACHE_SIZE, SchedulerJobRunner from airflow.models.asset import ( AssetActive, AssetAliasModel, @@ -410,6 +410,15 @@ def test_executor_loaded_in_scheduler_job(self, mock_init_executors, mock_defaul assert scheduler_job.executor == mock_local_executor assert scheduler_job.executors == [mock_local_executor] + def test_scheduler_dag_bag_is_bounded(self): + """The scheduler's Dag cache must evict, or it retains every version it has ever seen.""" + from cachetools import LRUCache + + job_runner = SchedulerJobRunner(Job()) + + assert isinstance(job_runner.scheduler_dag_bag._dags, LRUCache) + assert job_runner.scheduler_dag_bag._dags.maxsize == SCHEDULER_DAG_CACHE_SIZE + @pytest.mark.parametrize( "heartrate", [10, 5],