hussein-awala opened a new pull request, #76213: URL: https://github.com/apache/airflow/pull/76213
The triggerer's claim queries join `task_instance` and `callback` to order triggers by priority, and lock their result without naming a table. On PostgreSQL and MySQL 8 that locks the joined task instance and callback rows along with the `trigger` rows until the claim commits, so scheduler writes to those task instances wait for the triggerer, and `SKIP LOCKED` makes the triggerer pass over a trigger whenever the scheduler holds its task instance row. Lock only the `trigger` rows, the only rows the claim updates. related: #76209 On PostgreSQL each joined claim query locks up to `[triggerer] max_trigger_to_select_per_loop` (default 50) task instance or callback rows. On MySQL 8 InnoDB locks rows as it reads them, before the `ORDER BY` sort and the `LIMIT` are applied, so a claim can lock every joined row it reads. `assign_unassigned` claims on every triggerer loop. The claim queries are in `Trigger.get_sorted_triggers`: https://github.com/apache/airflow/blob/b95498cd56/airflow-core/src/airflow/models/trigger.py#L485-L506 The deferred-task timeout sweep's `UPDATE task_instance` is one scheduler write that waits on those locks: https://github.com/apache/airflow/blob/b95498cd56/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L3891-L3913 The claim decides ownership only through `trigger.triggerer_id`, so passing `of=Trigger` is enough. The lock clause becomes `FOR NO KEY UPDATE OF trigger SKIP LOCKED` on PostgreSQL and ``FOR UPDATE OF `trigger` SKIP LOCKED`` on MySQL 8, the same `of=` form `_delete_unused_batch` already uses on this table. Lock strength, the claim predicate, ordering and `LIMIT` are unchanged, and there is no new configuration or schema change. Nothing changes where no row lock is rendered today: SQLite, MariaDB, MySQL before 8, and `[scheduler] use_row_level_locking = False`. The behaviour change is that a trigger whose task instance row another transaction holds can now be claimed in that loop instead of a later one. If the task instance leaves the deferred state in the meantime, the triggerer already skips a trigger that has no task instance and cancels or releases triggers it no longer owns. #75962 makes the timeout sweep claim task instances with `with_row_locks(of=TaskInstance, skip_locked=True)`, rendered as `FOR NO KEY UPDATE OF task_instance SKIP LOCKED` on PostgreSQL. The two are complementary: with this change the sweep no longer skips timed-out task instances because a triggerer claim holds them, and the sweep's locks no longer make the triggerer skip claims. #72849 changes task instance locking in `submit_event` and `submit_failure` in the same file; it does not touch the claim. Recording the claim statements that `Trigger.assign_unassigned` sends to PostgreSQL 16 shows the same statements with a narrower lock target: | Claim SELECTs in `assign_unassigned` (PostgreSQL 16) | Before | After | |---|---|---| | Locked claim statements | 3 | 3 | | Lock clause on each statement | `FOR NO KEY UPDATE SKIP LOCKED` | `FOR NO KEY UPDATE OF trigger SKIP LOCKED` | The check attached an engine-wide statement listener around `Trigger.assign_unassigned` with three claimable deferred task instances and one live triggerer job, on PostgreSQL 16 in Docker on a laptop, once on `main` and once with this change. It shows lock scope only; no timings were taken, and the contention case itself was not benchmarked. `test_get_sorted_triggers_locks_only_trigger_rows` wraps `with_row_locks`, compiles the three locked claim queries for PostgreSQL and for MySQL 8, and asserts that the `task_instance` and `callback` joins are still present while every lock clause names only `trigger`: `FOR NO KEY UPDATE OF trigger SKIP LOCKED` on PostgreSQL and ``FOR UPDATE OF `trigger` SKIP LOCKED`` on MySQL 8, where `trigger` is a reserved word and has to be quoted. SQLAlchemy's MySQL dialect renders `OF` only after it has detected MySQL 8 on connect, so the test puts the dialect in that state (server version 8.0.36 with `supports_for_update_of`), the same way SQLAlchemy's own `FOR UPDATE OF` compile tests do. Both cases fail without the change. Validation: - Triggerer job and trigger model tests: 207 passed, 1 skipped on SQLite. - Same tests through Breeze with the CI image: 208 passed on PostgreSQL 14; 207 passed, 1 skipped on MySQL 8.0 (the skip is a PostgreSQL-only test). - Regular and manual static checks passed on the full diff. - Selected core suites and the documentation build were not run locally; CI validation is pending. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Opus 5.5) Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
