[I] Scheduler does an unbounded DELETE on unreferenced triggers every cleanup tick [airflow]

["muditagrawal2007 (via GitHub)" <gi...@apache.org>]

muditagrawal2007 opened a new issue, #75141:
URL: https://github.com/apache/airflow/issues/75141

   ### Summary
   
   `SchedulerJobRunner._remove_unreferenced_triggers` issues an **unbounded single-statement `DELETE` against the `trigger` table every `parsing_cleanup_interval` (default 60s)**. On any deployment where the trigger table grows large — which is expected over time, since deferrable operators and asset/asset-alias watchers each leave a trigger row behind — this holds row locks for the entire transaction and stalls concurrent writers on the same table, including the triggerer's own `Trigger.clean_unused()` / `submit_event` paths. It also forces a full table materialisation in Python via `synchronize_session="fetch"` before the DELETE runs.
   
   ### Why this matters
   
   - **AGENTS.md (Coding Standards) is explicit:** *"Bulk `DELETE`/`UPDATE` in the scheduler loop or any synchronous interval task (e.g. `call_regular_interval` callbacks) must be batched with `LIMIT` and committed between batches — never issue a single unbounded bulk write against a user-driven table. … Follow the batching pattern in `airflow-core/src/airflow/utils/db_cleanup.py`."* The current implementation predates and contradicts that rule. Affected column joins (`task_instance.trigger_id`, `asset_watcher.trigger_id`, `callback.trigger_id`) are already indexed.
   - **Two cleanup paths, one batched and one not.** #70961 (merge commit `c3cc5573b8`) moved the **triggerer's** version (`Trigger.clean_unused()`) onto the existing pattern at `models/trigger.py:_delete_unused_batch`: `select id LIMIT, with_row_locks(skip_locked=True), delete by id, commit`, driven by `[triggerer] unreferenced_triggers_cleanup_batch_size`. The scheduler-side counterpart was missed by that change — only the triggerer's `clean_unused` callback in `jobs/triggerer_job_runner.py:clean_unused` was rewired. The scheduler still calls its own `_remove_unreferenced_triggers` from `jobs/scheduler_job_runner.py:2034`.
   
   ### Evidence
   
   `airflow-core/src/airflow/jobs/scheduler_job_runner.py:4206-4219`:
   
   ```python
   @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")
       )
   ```
   
   Existing tests are green at `airflow-core/tests/unit/jobs/test_scheduler_job.py:10051` and `:10087` (covering the unreferenced/referenced-path decision tree) but assume the unbounded form — they won't fail today because the table is small.
   
   ### Why I'm filing this
   
   A previous PR — **#68241** "Batch the scheduler's unreferenced-trigger cleanup deletes" — implemented the same fix and was closed as a stale draft on 2026-07-02 (author inactive; `potiuk` moved it to draft because static / docs checks were failing). It would also re-roll the batching logic that already exists in `Trigger._delete_unused_batch`. **The triggerer side of this work landed (#70961); the scheduler side did not.**
   
   ### Proposed fix
   
   Reuse the helper `Trigger._delete_unused_batch` from the scheduler rather than re-rolling the LIMIT/with_row_locks/commit-between-batches loop. Concretely:
   
   1. Add a new config knob `[scheduler] unreferenced_triggers_cleanup_batch_size` (default `500`, `version_added: 3.3.0` per the existing `[triggerer]` option pattern).
   2. Rewrite `_remove_unreferenced_triggers` to read that knob, validate it is `≥ 1`, and call `Trigger._delete_unused_batch(batch_size, session=session)`. The triggerer's `clean_unused` will keep the same cadence it has today (one batch per tick → drain over multiple ticks), so the net behaviour is the same modulo per-tick lock duration.
   3. Extend tests in `tests/unit/jobs/test_scheduler_job.py` with: a batching-commit-count test (e.g. 5 triggers, `batch_size=2` → 3 commits), a non-positive-batch-size rejection test mirroring `tests/unit/models/test_trigger.py:325`, and a sanity test that referenced triggers survive.
   
   ### Areas
   
   - area:scheduler
   - area:core
   
   ### Reproduction
   
   Hard to reproduce locally on a small fixture, but reproducible on any long-running install by tailing the scheduler logs for one cleanup tick after seeding N >> 500 unreferenced trigger rows; you will see the tick stretch to multiple seconds and any concurrent writer hitting `SELECT … FOR UPDATE` on `trigger` block.
   
   ### Airflow version
   
   main at `7fbcfeea27` (post-#75096).
   
   ### Follow-ups
   
   None expected — once the scheduler delegates to `Trigger._delete_unused_batch`, the cleanup pattern is fully owned by `Trigger`, and any future tuning (e.g. lock-skip policy, column-level WHERE tightening) edits one class.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org