diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index b87f27c509364..7d30836c1292f 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml @@ -2780,6 +2780,18 @@ scheduler: type: integer example: ~ default: "60" + unreferenced_triggers_cleanup_batch_size: + description: | + Maximum number of unreferenced ``trigger`` rows the scheduler removes per + transaction during its periodic trigger cleanup (see + ``[scheduler] parsing_cleanup_interval``). Batching avoids holding row locks + on the ``trigger`` table for the duration of a single large delete, which + would otherwise block the triggerer's own writers and stall the scheduler + main loop. Must be at least ``1``. + version_added: 3.3.0 + type: integer + example: ~ + default: "500" dag_cache_size: description: | Max number of deserialized SerializedDAG objects the scheduler keeps in memory, keyed by Dag version diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index 198ef4bdc0ad9..62f833549851e 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -85,7 +85,6 @@ AssetEvent, AssetModel, AssetPartitionDagRun, - AssetWatcherModel, DagScheduleAssetAliasReference, DagScheduleAssetReference, PartitionedAssetKeyLog, @@ -4193,18 +4192,22 @@ def _generate_task_instance_heartbeat_timeout_message_details( @provide_session def _remove_unreferenced_triggers(self, *, session: Session = NEW_SESSION) -> None: - """Remove triggers that are no longer used by anything.""" - session.execute( - delete(Trigger) - .where( - ~exists( - select(AssetWatcherModel.trigger_id).where(AssetWatcherModel.trigger_id == Trigger.id) - ), - ~exists(select(Callback.trigger_id).where(Callback.trigger_id == Trigger.id)), - ~exists(select(TaskInstance.trigger_id).where(TaskInstance.trigger_id == Trigger.id)), - ) - .execution_options(synchronize_session="fetch") - ) + """ + Remove triggers that are no longer used by anything. + + Delegates to ``Trigger._delete_unused_batch`` so the same LIMIT/row-lock/commit + pattern the triggerer uses (#70961) applies here too — holding row locks on + ``trigger`` for the duration of an unbounded DELETE stalls the triggerer's + writers and the scheduler main loop on installs with large trigger tables. + + Batch size is controlled by ``[scheduler] unreferenced_triggers_cleanup_batch_size`` + (default 500). One batch is removed per call; later scheduler ticks drain the + remainder, matching the triggerer's per-tick cadence. + """ + batch_size = conf.getint("scheduler", "unreferenced_triggers_cleanup_batch_size", fallback=500) + if batch_size <= 0: + raise ValueError("[scheduler] unreferenced_triggers_cleanup_batch_size must be at least 1") + Trigger._delete_unused_batch(batch_size, session=session) @provide_session def _update_asset_orphanage(self, *, session: Session = NEW_SESSION) -> None: diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index 9f7d83b9787d2..a11724dd2c9e5 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -10279,6 +10279,100 @@ def test_delete_unreferenced_triggers_with_null_trigger_id_ti(self, dag_maker, s self.job_runner._remove_unreferenced_triggers(session=session) assert session.scalar(select(func.count()).select_from(Trigger)) == 0 + @pytest.mark.need_serialized_dag(False) + @conf_vars({("scheduler", "unreferenced_triggers_cleanup_batch_size"): "2"}) + def test_remove_unreferenced_triggers_drains_in_batches_per_call(self, session): + """One scheduler tick removes at most one batch (5 triggers, batch=2 → 3 stay). + + Mirrors the triggerer's cadence (``Trigger.clean_unused`` does one batch per call; + later scheduler ticks drain the remainder). Holding locks across unbounded DELETEs + would block the triggerer's writers — see #75141. + """ + self.job_runner = SchedulerJobRunner(job=Job()) + + for _ in range(5): + session.add( + Trigger( + classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger", + kwargs={}, + ) + ) + session.commit() + + self.job_runner._remove_unreferenced_triggers(session=session) + + # batch_size=2 ⇒ exactly 2 rows deleted; 3 remain to be drained next tick. + assert session.scalar(select(func.count()).select_from(Trigger)) == 3 + + # Second tick drains the remaining batch. + self.job_runner._remove_unreferenced_triggers(session=session) + assert session.scalar(select(func.count()).select_from(Trigger)) == 1 + + # Third tick removes the final row. + self.job_runner._remove_unreferenced_triggers(session=session) + assert session.scalar(select(func.count()).select_from(Trigger)) == 0 + + @pytest.mark.need_serialized_dag(False) + @pytest.mark.parametrize("batch_size", ["0", "-1"]) + def test_remove_unreferenced_triggers_rejects_non_positive_batch_size(self, session, batch_size): + """A non-positive batch size must raise before any DELETE is issued.""" + self.job_runner = SchedulerJobRunner(job=Job()) + + session.add( + Trigger( + classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger", + kwargs={}, + ) + ) + session.commit() + + with conf_vars({("scheduler", "unreferenced_triggers_cleanup_batch_size"): batch_size}): + with pytest.raises( + ValueError, + match=r"\[scheduler\] unreferenced_triggers_cleanup_batch_size must be at least 1", + ): + self.job_runner._remove_unreferenced_triggers(session=session) + + assert session.scalar(select(func.count()).select_from(Trigger)) == 1 + + @pytest.mark.need_serialized_dag(False) + def test_remove_unreferenced_triggers_preserves_referenced_triggers(self, session, dag_maker): + """Cleanup must skip triggers still referenced by a deferred TaskInstance. + + Asset-watcher and Callback preservation are covered by the model's own + tests in ``tests/unit/models/test_trigger.py``; here we just verify the + scheduler delegates the same preservation contract. + """ + self.job_runner = SchedulerJobRunner(job=Job()) + + with dag_maker(dag_id="dag", schedule=None, session=session): + EmptyOperator(task_id="fake0") + + dr = dag_maker.create_dagrun() + + referenced = Trigger( + classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger", kwargs={} + ) + unreferenced = Trigger( + classpath="airflow.providers.standard.triggers.file.FileDeleteTrigger", kwargs={} + ) + session.add_all([referenced, unreferenced]) + session.flush() + + ti = next(iter(dr.task_instances)) + ti.state = TaskInstanceState.DEFERRED + ti.trigger_id = referenced.id + session.commit() + + unreferenced_id = unreferenced.id + referenced_id = referenced.id + self.job_runner._remove_unreferenced_triggers(session=session) + session.expire_all() + + remaining_ids = set(session.scalars(select(Trigger.id)).all()) + assert unreferenced_id not in remaining_ids + assert referenced_id in remaining_ids + @patch("airflow.serialization.serialized_objects.SerializedDAG.create_dagrun") def test_misconfigured_dags_doesnt_crash_scheduler(self, mock_create, session, dag_maker, caplog): """Test that if dagrun creation throws an exception, the scheduler doesn't crash"""