Fix deactivation of stale zip-packaged DAGs

[SamWheating]

The Issue:

When deactivating DAGs, airflow compares the DagFileStat generated by the manager to a DAG's fileloc in order to identify DAGs which were not seen on the most recent file parse.

However, when processing zip-packaged DAGs, the DAG's relative_fileloc will be the value of __file__ reported by zipfile, which includes the inner path within the zip archive (ex: path/to/some/dag_package.zip/my_dag.py).

Meanwhile the parsing stats used to index the last_parsed dictionary use the path of the archive file itself (ex: path/to/some/dag_package.zip) since there is a single entry per file-parsed.

As a result, when looking at zip-packaged DAGs the last_parsed.get() lookup here will never return a value:

airflow/airflow-core/src/airflow/dag_processing/manager.py

Lines 481 to 489 in 1438ea3

	file_info = DagFileInfo(rel_path=Path(dag.relative_fileloc), bundle_name=dag.bundle_name)
	if last_finish_time := last_parsed.get(file_info, None):
	if dag.last_parsed_time + timedelta(seconds=self.stale_dag_threshold) < last_finish_time:
	self.log.info(
	"Deactivating stale DAG %s. Not parsed for %s seconds (last parsed: %s).",
	dag.dag_id,
	int((last_finish_time - dag.last_parsed_time).total_seconds()),
	dag.last_parsed_time,
	)

And thus zip-packaged DAGs are never properly deactivated.

This issue doesn't surface too often, since most DAGs will also be caught by the deactivate_deleted_dags function which properly evaluates them based on inner paths. However for files which produce multiple DAGs ("dag factories") or DAGs which are renamed within a file, this second mechanism won't work either, since DAGs can be removed without the source file being removed.

As a result, certain DAGs within a zip file will never be removed and airflow will continue to attempt to schedule and run them. I have reproduced this issue in breeze on the latest version of main.

The Fix:

Within the deactivate_stale_dags function, we can just remove the inner part of the fileloc for zip-packaged DAGs to obtain the actual filepath of the archive itself. I've added this as a private method on the class, but it could also be added as a property on the DAG model itself.

---

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

No

[SamWheating]

Test failure seems unrelated (timeout from dockerhub) could someone re-run when you have a chance? Thanks!

[amoghrajesh]

Conflicts on this need to be handled.

[github-actions]

Backport successfully created: v3-3-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-3-test