Dag processor reparses a file every loop forever after a Dag-level callback runs

[vuonghoangbntt]

Under which category would you file this issue?

Airflow Core

Apache Airflow version

3.3.1

What happened and how to reproduce it?

Issue Description

A Dag with a Dag-level on_failure_callback parses on schedule until one of its runs
fails. After that the Dag processor reparses that file on every loop and stops applying
[dag_processor] min_file_process_interval to it. It never recovers on its own.
Restarting the Dag processor clears it. The next failed run brings it straight back.

We hit this in production on 3.3.1, where it held a Dag processor at its 1 CPU limit for
days. Two Dag files were reparsing roughly 43x more often than configured. The other
five files in the same bundle stayed exactly on interval.

The DAG File Processing Stats table gives no hint. It shows recent, healthy
Last Run At values for the affected file the whole time.

Root cause

Of the five DagFileInfo(...) construction sites, only _add_callback_to_queue sets
bundle_version. The bundle scan leaves it unset, and that's on purpose: bundle_path
is field(compare=False), so bundle_version is the only field that keeps a
version-pinned callback distinct from the scan entry. Drop it and a pinned callback
parses against the tracking checkout instead of the version its Dag run used.

The side effect is that one physical file can end up holding two _file_stats entries.
prepare_file_queue collapses them:

file_stats_by_presence_key = {file.presence_key: stat for file, stat in self._file_stats.items()}

A dict comprehension keeps whichever key went in last. That is the callback entry, since
the file has to be parsed before a run can fail, so it shadows the scan entry.

That entry never gets a timestamp. process_parse_results builds a callback-only stat
without last_finish_time (deliberately, so it does not disturb stale-Dag detection),
and handle_parsing_result writes it over the entry wholesale, so the value is None.
It stays None, because every later parse writes to the scan key instead. From then on
the interval check sees a file that has apparently never been parsed, and requeues it
every loop.

Nothing cleans it up, either. remove_orphaned_file_stats matches on presence_key,
which is still present.

Two nearby places resolve the same lookup and disagree with each other. The identical
collapse is duplicated in _sort_by_mtime, which feeds changed_recently, the other
route that bypasses the interval. And processed_recently resolves it by first match,
the opposite tiebreak.

What keeps it hidden is _log_file_processing_stats, which reads _file_stats by exact
key. It reports the scan entry's healthy timestamp while the queue logic reads the
frozen one.

Scope

This only bites bundles that support versioning. When a request carries no version,
request.bundle_version is None (since #72930), the two keys coincide, and merging
them is correct.

Steps to reproduce

1. Airflow with a standalone dag-processor and a versioning bundle (GitDagBundle),
   so the bundle reports a non-null version.
2. Set [dag_processor] min_file_process_interval = 30.
3. Add two Dags to the bundle. One has a Dag-level on_failure_callback and a task
   that raises; the other is a control with no callback:

with DAG(dag_id="callback_dag", schedule=None, on_failure_callback=_notify):
    PythonOperator(task_id="boom", python_callable=_boom)  # raises

4. Trigger callback_dag and let the run fail, so the Dag-level callback is dispatched.
5. Watch the reparse rate, e.g. by polling dag.last_parsed_time, or the statsd counter
   dag_processing.last_duration.<file>.

Measured on main (sqlite, GitDagBundle, min_file_process_interval = 30):

phase	callback_dag	control_dag
baseline, before any callback	1 per 30.0s	1 per 30.0s
after the Dag-level callback fired	1 per 0.4s	1 per 30.0s

Restarting the Dag processor clears it, because the bad state lives in memory. The next
failed Dag run re-arms it.

What you think should happen instead?

min_file_process_interval` should keep applying to a file whether or not a Dag-level
callback has ever run for it. One callback should not change that file's parse cadence
permanently.

Operating System

No response

Deployment

Official Apache Airflow Helm Chart

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

No response

Official Helm Chart version

1.22.0 (latest released)

Kubernetes Version

1.34.9

Helm Chart configuration

No response

Docker Image customizations

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct