Donniedarko45 opened a new pull request, #76987: URL: https://github.com/apache/airflow/pull/76987
closes: #70614 Since 3.2 (#60006) the scheduler writes `dag.exceeds_max_non_backfill` when a run starts or finishes. The Dag processor holds `FOR NO KEY UPDATE` on every `dag` row of a file for its whole parse transaction (`DagModelOperation.find_orm_dags`), which can last minutes for a file that generates many Dags. A scheduler write to one of those rows waits for the parse to commit, and the scheduling loop stalls with it. This is the `Lock` / `transactionid` wait in the issue. Reproduced on Postgres 16 against `main`, using the real `find_orm_dags` and `_set_exceeds_max_active_runs` in two sessions, with the parse transaction held for 8s: | | flag unchanged | flag flips | |---|---|---| | before | no `UPDATE` sent, 0.00s | blocked 8.00s (`Lock` / `transactionid`) | | after | no `UPDATE` sent, 0.00s | 0.00s, write deferred; next loop writes the right value | The "flag unchanged" column is also why skipping redundant writes (#70712) can't fix this: SQLAlchemy already sends no `UPDATE` when the value doesn't change, so the blocking write only happens when the flag really flips. **Change** `_set_exceeds_max_active_runs` no longer waits for the row lock. When the flag needs to change, it first selects the row with `SKIP LOCKED`. - If it gets the row, it writes the flag as before. In the run-creation path the scheduler already holds this lock through `dags_needing_dagruns`, so that path behaves the same. - If the row is held elsewhere, it skips the write and records the Dag. At the start of the next loop, before Dag runs are created, the scheduler recomputes the flag for those Dags from fresh active-run counts. **Why a deferred write is safe** - While the row stays locked, `dags_needing_dagruns` skips it (it selects with `SKIP LOCKED` too), so a flag that should have become `True` can't lead to an extra Dag run. - A flag that should have become `False` delays run creation for that Dag by at most one scheduler loop. - The Dag processor computes its own value of the flag from a run count read during the parse. If that count is already stale when it commits, the retry on the next loop corrects it. Today the blocked scheduler write is what corrects it. - The retry set is in memory. If a scheduler dies with a pending Dag, the flag is corrected at the next run state change or the next parse of the Dag's file, the same as for any other missed write today. - Databases without row locking (`USE_ROW_LEVEL_LOCKING` off) skip the lock check, so behaviour there is unchanged. MySQL 8 gets `FOR UPDATE SKIP LOCKED`. I considered narrowing the Dag processor's lock scope instead. It is the bigger lever, but it changes the parse transaction for every writer of `dag`. This change only touches the scheduler side and doesn't stop that being done later. **Tests** - `test_set_exceeds_max_active_runs_does_not_wait_on_dag_row_locked_elsewhere` (Postgres/MySQL) holds the row the way the Dag processor does and asserts the scheduler returns without waiting. Without the change it fails with `TimeoutError`. - Tests for the retry: recomputing from fresh counts, dropping deleted Dags, clearing the retry when the flag is already correct, and `_do_scheduling` running the retry before Dag run creation. - Full `test_scheduler_job.py` on SQLite and on Postgres 16, compared with the same command on unchanged `main`: no new failures on either. The run without breeze has the same pre-existing environment failures on both. - Pre-commit hooks pass, including `mypy-airflow-core`. `check-template-fields-valid` needs breeze and was not run locally. ) -- 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]
