Dag processor orphan cleanup drops versioned callback files before DAG callbacks execute

[hkc-8010]

Apache Airflow version

3.1.8 and current main

What happened?

We found a case where DAG-level callbacks for versioned DAG bundles are fetched and queued by the dag processor, but then get treated as orphaned before callback execution.

The observed sequence is:

1. Scheduler creates a DagCallbackRequest
2. Dag processor fetches it from the DB
3. Dag processor queues it for file processing
4. Dag processor starts or prepares callback processing for the file
5. The same callback file is later treated as "not present" / orphaned and is purged or the processor is stopped before _execute_dag_callbacks runs

This is distinct from the stale serialized DAG metadata problem in #66301 and the fix in #66474.

That earlier issue was about SerializedDagModel not refreshing when only bundle version changed. Here, the failure is later in the dag processor manager's callback/orphan cleanup path.

What you think should happen instead?

If a callback request is queued for bundle_name=X, rel_path=Y, and the same file is still present in the current bundle scan, it should not be treated as removed solely because the callback queue entry includes bundle_version while the scanned "present files" entry does not.

In other words, present/orphan checks in the dag processor should treat file presence as:

• bundle_name
• rel_path

and should not require exact DagFileInfo equality including bundle_version.

How to reproduce

A minimal pattern is:

1. Use a DAG bundle implementation with supports_versioning=True
2. Have a DAG with a DAG-level on_failure_callback
3. Trigger a failing DAG run so a DagCallbackRequest is created
4. In the dag processor:
   • _add_callback_to_queue() creates a DagFileInfo with bundle_version=<version>
   • _refresh_dag_bundles() builds known_files using DagFileInfo(rel_path=..., bundle_name=..., bundle_path=...) with no bundle_version
5. Let purge_removed_files_from_queue() / terminate_orphan_processes() run
6. Observe that the versioned callback file can be treated as absent even though the same bundle_name + rel_path is still present in the scanned files

Operating System

Linux

Deployment

Other

Anything else?

The relevant code path in airflow-core/src/airflow/dag_processing/manager.py is:

• _add_callback_to_queue() creates:

  file_info = DagFileInfo(
   rel_path=Path(request.filepath),
   bundle_path=bundle.path,
   bundle_name=request.bundle_name,
   bundle_version=request.bundle_version,
  )

• _refresh_dag_bundles() builds known_files with:

  DagFileInfo(rel_path=p, bundle_name=bundle.name, bundle_path=bundle.path)

• purge_removed_files_from_queue() and terminate_orphan_processes() then use full DagFileInfo equality against the present set

Because DagFileInfo includes bundle_version in equality/hashing, a versioned callback entry and an unversioned scanned entry for the same file do not compare equal.

That means the callback queue entry or active processor can be treated as orphaned even though the file is still present.

A small local proof of the mismatch is:

from pathlib import Path
from airflow.dag_processing.manager import DagFileInfo

versioned = DagFileInfo(
 rel_path=Path("dags/example.py"),
 bundle_name="main",
 bundle_path=Path("/tmp/bundle"),
 bundle_version="v1",
)
present = DagFileInfo(
 rel_path=Path("dags/example.py"),
 bundle_name="main",
 bundle_path=Path("/tmp/bundle"),
)

assert versioned != present

The fix I tested locally is to keep callback identity version-aware, but make present/orphan checks compare only a stable file-presence key of:

• bundle_name
• rel_path

That preserves distinct callback entries per version while preventing false orphan cleanup.

Are you willing to submit PR?

Yes