Do not fail scheduled tasks when serialized Dag is briefly missing

[Vamsi-klu]

What is the change?

_task_concurrency_allows_execution no longer bulk-UPDATEs every SCHEDULED task instance to FAILED when get_dag_for_run returns None. It logs the same error as _create_dag_runs and returns False so this tick skips, and the next tick retries.

Why did I do it?

closes: #62050

The miss sits inside a concurrency check. A missing serialized row is not a concurrency answer. On HA schedulers, a parse or version hole then failed the whole warehouse load; retry worked because the next parse had a row. The create path already continues (test_scheduler_create_dag_runs_does_not_raise_error_when_no_serdag). Queue never got the same treatment. #58259 / #56422 made misses rarer; they did not remove the UPDATE.

How did I do it?

Deleted the session.execute(update(TI)...FAILED) block. return False was already there. Session stays uncommitted. I did not add a miss counter, did not fail only the one TI, and did not change _create_dag_runs. Permanently missing Dags stay SCHEDULED for the existing stale/import-error cleanup.

This path only runs when dag_model.has_task_concurrency_limits is True (max_active_tis_per_dag / max_active_tis_per_dagrun). Tests set max_active_tis_per_dag so the helper is actually entered.

What's the impact?

A transient serialized_dag hole no longer fails every SCHEDULED TI for that Dag (including every mapped index, backfill slice, and asset-triggered run that hits this helper). HA schedulers all skip instead of racing to stamp FAILED. Deleted Dags can sit in SCHEDULED until other cleanup; that is intentional.

What's the test plan?

New tests next to the create-path skip test:

• test_executable_task_instances_skip_when_serialized_dag_missing: two tasks with concurrency limits, mock get_dag_for_run to None, queued list empty, both TIs still SCHEDULED
• test_executable_task_instances_queue_when_serialized_dag_present: control, t1/t2 still queue

I restored the UPDATE and re-ran the skip test: TIs became FAILED (the first TI's UPDATE failed all SCHEDULED TIs including t2).

uv run --project airflow-core pytest \
  airflow-core/tests/unit/jobs/test_scheduler_job.py \
  -k 'skip_when_serialized_dag_missing or queue_when_serialized_dag_present or no_serdag'

3 passed. Ruff and airflow-core mypy passed via prek.

---

Was generative AI tooling used to co-author this PR?

• Yes — Grok 4.6

Generated-by: Grok 4.6 following the guidelines

---

Drafted-by: Grok 4.6 (no human review before posting)

[potiuk]

Have you thought about side-effect of it? What are they? Do ypu (not your LLM) understand what your are doing here?have How it can happen that serialized dag is missing? Is it maybye a sign that something else is wrong - and you are just masking a problem?

Note - I have years of experience in Airflow but if I were to touch this code - I would think100 times and talk to someone over slack explaining how I reproduced .

Did you actually experience and reproduce it on running Airlfow instance? Do you have some proof of that?

Looking at the patterns of your contribution - you contribute like a shotgun - wherever your LLM thinks there is an issue, but you do not have deeper undersrtanding.

i am provisionally closing that - the "critical" parts of the code shoudl not be contributed to significantly without either experience or significant proof that you have run it locally, reproduced the issue and solved it and that you thought and reasoned about the consequences.

Thare are other areas where you can contribute smaller things in Airflow - if all you do is put your LLLm on it.