shivaam opened a new pull request, #68961: URL: https://github.com/apache/airflow/pull/68961
### Summary - Handle pending DagRun deadlines immediately when `DagRun.update_state()` transitions a run to `FAILED`. - Keep `Deadline.prune_deadlines()` limited to successful DagRun cleanup. - Persist deadline callback context/routing data by reassigning `callback.data`, so both triggerer and executor callback paths survive commit/reload. - Serialize `deadline.id` as a string in callback context so async trigger kwargs can be stored and run by the triggerer. Closes: #60927 ### Details Failed DagRuns currently leave pending deadlines untouched until the scheduler's expired-deadline loop reaches `deadline_time`. This can delay the configured deadline alert by hours after the run has already failed. This change treats scheduler-driven DagRun failure as an immediate missed deadline for pending deadlines on that DagRun. It uses the same row-locking shape as the scheduler deadline loop and filters already-missed rows to avoid duplicate handling. While validating the callback path, the executor callback path also needed top-level routing fields persisted in `callback.data`; otherwise the scheduler could mark the callback pending but not enqueue it. The triggerer callback path needed the same JSON reassignment discipline and a string deadline id because trigger kwargs are serialized before the triggerer runs them. Manual/API/UI "mark failed" remains out of scope for this patch. ### Tests ```bash uv run --project airflow-core pytest -q airflow-core/tests/unit/models/test_deadline.py::TestDeadline::test_handle_miss airflow-core/tests/unit/models/test_deadline.py::TestDeadline::test_handle_miss_persists_triggerer_callback_context airflow-core/tests/unit/models/test_deadline.py::TestDeadline::test_handle_miss_persists_executor_callback_routing_data airflow-core/tests/unit/models/test_dagrun.py::TestDagRun::test_dagrun_success_prunes_dagrun_deadlines airflow-core/tests/unit/models/test_dagrun.py::TestDagRun::test_dagrun_success_skips_pruning_non_dagrun_deadlines airflow-core/tests/unit/models/test_dagrun.py::TestDagRun::test_dagrun_success_handles_empty_deadline_list airflow-core/tests/unit/models/test_dagrun.py::TestDagRun::test_dagrun_failure_handles_pending_deadline airflow-core/tests/unit/models/test_dagrun.py::TestDagRun::test_dagrun_deadline_variable_interval_stable --with-db-init ``` Result: `8 passed, 1 warning`. ```bash uv run ruff check airflow-core/src/airflow/models/dagrun.py airflow-core/src/airflow/models/deadline.py airflow-core/tests/unit/models/test_dagrun.py airflow-core/tests/unit/models/test_deadline.py ``` Result: `All checks passed!` Also tested manually in a real local Airflow daemon environment: - API server, dag processor, triggerer, and two schedulers. - Local Postgres metadata DB and LocalExecutor. - Sync/executor deadline callback reached `success` and wrote a marker row. - Async/triggerer deadline callback reached `success` and wrote a marker row. Final E2E rows: ```text async: dagrun_state=failed, deadline_missed=True, callback_type=triggerer, callback_state=success sync: dagrun_state=failed, deadline_missed=True, callback_type=executor, callback_state=success ``` -- 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]
