[vuonghoangbntt]

Once a Dag-level callback has run for a file, the Dag processor reparses that file on
every loop and stops honouring [dag_processor] min_file_process_interval for it. It
doesn't clear on its own. Restarting the processor fixes it, until the next Dag run
failure.

Why

_add_callback_to_queue is the only place that sets bundle_version on a
DagFileInfo; the bundle scan leaves it unset. That's deliberate. bundle_path is
compare=False, so bundle_version is the only field keeping a version-pinned callback
distinct from the scan entry. Without it, a pinned callback would parse against the
tracking checkout instead of the version its Dag run used.

The side effect is that one file can hold two _file_stats entries.
prepare_file_queue collapses them with

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

which keeps whichever went in last. That's the callback entry, since the file must be
parsed before a run can fail.

That entry never gets a timestamp. process_parse_results builds a callback-only stat
without last_finish_time (deliberately, to avoid disturbing 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.

It stays invisible in the stats table, because _log_file_processing_stats reads
_file_stats by exact key. It reports the scan entry's healthy timestamp while the
queue logic reads the callback entry's None.

Only versioning-capable bundles hit this. For unversioned requests
request.bundle_version is None (see #72930), the two keys coincide, and the merge is
correct.

Fix

Resolve the presence_key collapse from the bundle scan's entry and ignore versioned
(callback) entries. A callback run never records a parse, so a versioned stat is only
ever None or a timeout stamp that _kill_timed_out_processors writes although nothing
was parsed. prepare_file_queue and _sort_by_mtime share the new
_resolve_stats_by_presence_key, and emit_metrics uses it too, so a timed-out
callback's import_errors=1 no longer keeps dag_processing.import_errors at 1
(reported by @fahadabdul). #66484's two versioned-stat tests are flipped to expect the
file to be queued. processed_recently is left as is and will be removed separately.

This leaves identity and lifecycle alone, so a versioned callback entry still survives
cleanup while the unversioned file is present.

Reproduction

Local Airflow built from main, sqlite, GitDagBundle (versioning on),
min_file_process_interval = 30, two Dags: one with a Dag-level on_failure_callback
and a task that raises, one control. Measured by polling dag.last_parsed_time:

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
after this change, fresh callback fired	1 per 30.0s	1 per 30.0s

The third row triggers a fresh failure after the restart. A restart on its own clears
the state, so measuring that alone would prove nothing.

I found this on 3.3.1, where it held a Dag processor at its 1 CPU limit indefinitely.
Two Dag files were reparsing about 43x more often than configured while the other five
in the same bundle sat exactly on interval.

Tests

The tests produce the callback stat through the code that writes it in production: a
callback-only handle_parsing_result, in both insertion orders, and
_kill_timed_out_processors, for the interval check, modified_time change detection
and the import-error gauge. Each asserts its setup before checking the result.

Verification run locally

• The new tests and the two flipped ones pass with the change. Against unpatched
  manager.py they all fail except the callback-recorded-first case, where insertion
  order happens to pick the scan entry.
• airflow-core/tests/unit/dag_processing/test_manager.py: 192 passed, 5 skipped
  (postgres-only). 12 tests error at fixture setup in my local env, and the same 12 error
  on unpatched main, so they are unrelated.
• ruff check and ruff format --check are clean, and so is mypy on the changed
  module.

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

• Yes (please specify the tool below)

Generated-by: Claude Opus 5 following the guidelines

closes: #73432

[khmelevskiy]

We're hitting this on 3.3.2 in production: standalone dag-processors (one per team), GitDagBundle, min_file_process_interval = 120.

Within two and a half days of a redeploy it started on four files across three dag-processors. Each looping file is parsed 2-3 times a second instead of every 120 s, costs 1-1.5 CPU cores, and its child-process log grows by 80-280 MB a day. A restart clears it; the file's next callback brings it back, usually within a day.

One addition to the issue: here the trigger was task-level callbacks, not Dag-level ones. These are the TaskCallbackRequests the scheduler sends for tasks it fails itself. The loop started in the same minute as the file's first callback. The request carries bundle_version either way, so this PR covers both.

main still resolves the stats by insertion order in prepare_file_queue and _sort_by_mtime.

@jedcunningham @ephraimbuddy could you take a look? A backport to v3-3-test would help too.

[fahadabdul]

A related symptom from the same versioned callback entry, which this PR doesn't seem to cover: dag_processing.import_errors stays at 1 when there are no import errors.

What seems to happen

1. A callback runs under a DagFileInfo with bundle_version set, so it gets its own _file_stats entry.
2. If that callback run exceeds dag_file_processor_timeout, _kill_timed_out_processors writes DagFileStat(import_errors=1, ...) to the versioned entry. No ParseImportError row is written, so the UI and API show 0.
3. remove_orphaned_file_stats compares presence_key, so the versioned entry survives while the file exists (as intended since Fix dag processor callback cleanup for versioned bundle files #66484).
4. emit_metrics sums import_errors across all _file_stats values, so the gauge keeps reporting 1 until the DAG processor restarts.

In the production case, the stats table showed 0 errors for the same file at the same time, since _log_file_processing_stats reads the scan entry by exact key.

Repro (Astronomer Runtime build of 3.3.2, 3.3.2+astro.2, versioned bundle, dag_file_processor_timeout = 30):

import time
from datetime import datetime
from airflow.sdk import dag, task

def slow_callback(context):
    time.sleep(120)

@dag(start_date=datetime(2026, 1, 1), schedule=None, catchup=False, on_failure_callback=slow_callback)
def ghost_import_error_repro():
    @task
    def fail():
        raise ValueError("fail on purpose")
    fail()

ghost_import_error_repro()

After one triggered run, the callback was killed after about 30s, and dag_processing.import_errors went from 0 to 1. The import_error table was empty while the gauge was at 1. The gauge stayed at 1 after a new bundle version was picked up (about 27 minutes on the same DAG processor), and went back to 0 once the DAG processor was replaced.

We first saw this in production on a 3.2.2 build with #66484 backported. The relevant code (presence_key cleanup, the timeout path, and emit_metrics) is the same in OSS 3.3.2 and on main.

Possible directions (not sure which fits best):

• Don't record import_errors=1 when the timed-out processor was a versioned callback run, since a slow callback isn't an import error.
• Or have emit_metrics resolve per presence_key the same way this PR now does for the queue checks (preferring the scan entry).

Happy to open a separate issue if you'd prefer to keep this PR focused.

[kaxil]

• The PR description still describes the newest-parse approach (>=, processed_recently sharing the helper, one test per lookup site). Could you update it to match the diff?
• The timed-out tests (L1246, L1269) and the callback-only test (L1223) pass whether or not the setup wrote the versioned entry, because the helper ignores it once a scan entry exists. Asserting it before prepare_file_queue (mock_kill called with SIGKILL and last_finish_time set, or None in the callback-only case) keeps them meaningful if the setup drifts.
• @fahadabdul's gauge report checks out: prepare_file_queue passes every _file_stats value to emit_metrics (L1648), so a timed-out versioned callback keeps dag_processing.import_errors at 1 while the UI shows 0. Passing self._resolve_stats_by_presence_key().values() there fixes it with the helper this PR adds, so it seems worth folding in.

[vuonghoangbntt]

Thanks @kaxil and @fahadabdul, all three are in the latest commit:

• The description now matches the diff.
• The tests assert their setup before prepare_file_queue
• emit_metrics now gets self._resolve_stats_by_presence_key().values(), with a test that a timed-out versioned callback no longer adds to dag_processing.import_errors.