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.