From 02cfd96be98fd06c9debbaaee5ced2b18b374524 Mon Sep 17 00:00:00 2001 From: Sam Wheating Date: Tue, 28 Jul 2026 11:02:18 +0100 Subject: [PATCH 1/3] use archive file path when checking for stale DAGs in zip packages --- .../src/airflow/dag_processing/manager.py | 18 ++++- .../tests/unit/dag_processing/test_manager.py | 66 +++++++++++++++++++ 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index df0f8f8748135..298bc7f6106e0 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -452,6 +452,20 @@ def cleanup_stale_bundle_versions(self) -> None: """Clean up stale DAG bundle version usage records.""" BundleUsageTrackingManager().remove_stale_bundle_versions() + @staticmethod + def _file_name_from_fileloc(fileloc: str) -> str: + """ + If a python file comes from a zip module, return just the path to the zip file. + + file parsing stats are keyed by the top-level filename (path/to/archive.zip). + in order to correctly link DAGs to their relevant parsing stats, we need to normalize + fileloc to the archive file location, if applicable. + """ + if ".zip" in fileloc: + return fileloc[: fileloc.index(".zip") + 4] + + return fileloc + @provide_session def deactivate_stale_dags( self, @@ -506,7 +520,9 @@ def deactivate_stale_dags( # When the Dag's last_parsed_time is more than the stale_dag_threshold older than the # Dag file's last_finish_time, the Dag is considered stale as has apparently been removed from the file, # This is especially relevant for Dag files that generate Dags in a dynamic manner. - file_info = DagFileInfo(rel_path=Path(dag.relative_fileloc), bundle_name=dag.bundle_name) + file_info = DagFileInfo( + rel_path=Path(self._file_name_from_fileloc(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( diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index f1fb4b66c81fd..b561924be7064 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -1046,6 +1046,72 @@ def test_scan_stale_dags(self, session): # SerializedDagModel gives history about Dags assert serialized_dag_count == 1 + @pytest.mark.usefixtures("testing_dag_bundle") + def test_scan_stale_dags_deactivates_zip_packaged_dags(self, session, test_zip_path): + """ + Ensure that zip-packaged DAGs are marked inactive when the file is parsed but the + DagModel.last_parsed_time is not updated. + """ + manager = DagFileProcessorManager( + max_runs=1, + processor_timeout=10 * 60, + ) + bundle = MagicMock() + bundle.name = "testing" + manager._dag_bundles = [bundle] + + test_dag_path = DagFileInfo( + rel_path=Path(test_zip_path), + bundle_name="testing", + ) + dagbag = DagBag( + test_zip_path, + bundle_path=test_dag_path.bundle_path, + ) + + # Add stale DAG to the DB + dag = dagbag.get_dag("test_zip_dag") + sync_dag_to_db(dag, session=session) + + # Add DAG to the file_parsing_stats + stat = DagFileStat( + num_dags=1, + import_errors=0, + last_finish_time=timezone.utcnow() + timedelta(hours=1), + last_duration=1, + run_count=1, + last_num_of_db_queries=1, + ) + manager._files = [test_dag_path] + manager._file_stats[test_dag_path] = stat + + active_dag_count = session.scalar( + select(func.count(DagModel.dag_id)).where( + DagModel.dag_id == "test_zip_dag", + ~DagModel.is_stale, + DagModel.relative_fileloc == str(test_dag_path.rel_path / "test_zip.py"), + ) + ) + assert active_dag_count == 1 + + manager._scan_stale_dags() + + active_dag_count = session.scalar( + select(func.count(DagModel.dag_id)).where( + DagModel.dag_id == "test_zip_dag", + ~DagModel.is_stale, + DagModel.relative_fileloc == str(test_dag_path.rel_path / "test_zip.py"), + ) + ) + assert active_dag_count == 0 + + serialized_dag_count = session.scalar( + select(func.count(SerializedDagModel.dag_id)).where(SerializedDagModel.dag_id == dag.dag_id) + ) + # Deactivating the DagModel should not delete the SerializedDagModel + # SerializedDagModel gives history about Dags + assert serialized_dag_count == 1 + @pytest.mark.usefixtures("testing_dag_bundle") def test_deactivate_stale_dags_marks_dags_in_inactive_bundles(self, session): """Dags whose bundle is no longer active should be marked stale even without a parse signal.""" From 04bc2db46e09b994d1f3fa6fa763188b3ba73a18 Mon Sep 17 00:00:00 2001 From: Sam Wheating Date: Fri, 31 Jul 2026 13:15:42 +0100 Subject: [PATCH 2/3] suggestions from code review --- .../src/airflow/dag_processing/manager.py | 22 +++++-------------- .../tests/unit/dag_processing/test_manager.py | 8 +++---- 2 files changed, 8 insertions(+), 22 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index 298bc7f6106e0..a4d4917958c11 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -452,20 +452,6 @@ def cleanup_stale_bundle_versions(self) -> None: """Clean up stale DAG bundle version usage records.""" BundleUsageTrackingManager().remove_stale_bundle_versions() - @staticmethod - def _file_name_from_fileloc(fileloc: str) -> str: - """ - If a python file comes from a zip module, return just the path to the zip file. - - file parsing stats are keyed by the top-level filename (path/to/archive.zip). - in order to correctly link DAGs to their relevant parsing stats, we need to normalize - fileloc to the archive file location, if applicable. - """ - if ".zip" in fileloc: - return fileloc[: fileloc.index(".zip") + 4] - - return fileloc - @provide_session def deactivate_stale_dags( self, @@ -520,9 +506,11 @@ def deactivate_stale_dags( # When the Dag's last_parsed_time is more than the stale_dag_threshold older than the # Dag file's last_finish_time, the Dag is considered stale as has apparently been removed from the file, # This is especially relevant for Dag files that generate Dags in a dynamic manner. - file_info = DagFileInfo( - rel_path=Path(self._file_name_from_fileloc(dag.relative_fileloc)), bundle_name=dag.bundle_name - ) + rel_path = Path(dag.relative_fileloc) + file_info = DagFileInfo(rel_path=rel_path, bundle_name=dag.bundle_name) + if file_info not in last_parsed: + # Zip-packaged dags are keyed by the archive path, not the inner file, so try the parent as well + file_info = DagFileInfo(rel_path=rel_path.parent, 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( diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index b561924be7064..c666b4d62ac1e 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -576,7 +576,6 @@ def test_files_sorted_random_seeded_by_host(self): assert manager._file_queue == expected # Verify running it again produces same order - manager._files = [] manager.prepare_file_queue(known_files=known_files) assert manager._file_queue == expected @@ -1016,7 +1015,6 @@ def test_scan_stale_dags(self, session): run_count=1, last_num_of_db_queries=1, ) - manager._files = [test_dag_path] manager._file_stats[test_dag_path] = stat active_dag_count = session.scalar( @@ -1061,11 +1059,12 @@ def test_scan_stale_dags_deactivates_zip_packaged_dags(self, session, test_zip_p manager._dag_bundles = [bundle] test_dag_path = DagFileInfo( - rel_path=Path(test_zip_path), + rel_path=Path("test_zip.zip"), + bundle_path=Path(test_zip_path).parent, bundle_name="testing", ) dagbag = DagBag( - test_zip_path, + test_dag_path.absolute_path, bundle_path=test_dag_path.bundle_path, ) @@ -1082,7 +1081,6 @@ def test_scan_stale_dags_deactivates_zip_packaged_dags(self, session, test_zip_p run_count=1, last_num_of_db_queries=1, ) - manager._files = [test_dag_path] manager._file_stats[test_dag_path] = stat active_dag_count = session.scalar( From 46afe881fcba88d3e0530773c6f90e107e3a2c11 Mon Sep 17 00:00:00 2001 From: Sam Wheating Date: Tue, 4 Aug 2026 12:19:07 +0100 Subject: [PATCH 3/3] suggestions from code review --- airflow-core/tests/unit/dag_processing/test_manager.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index c666b4d62ac1e..b136a4d9e8fe3 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -1048,7 +1048,8 @@ def test_scan_stale_dags(self, session): def test_scan_stale_dags_deactivates_zip_packaged_dags(self, session, test_zip_path): """ Ensure that zip-packaged DAGs are marked inactive when the file is parsed but the - DagModel.last_parsed_time is not updated. + DagModel.last_parsed_time is not updated, testing fallback to the parent path when + comparing DAG.relative_fileloc to last_parsed entries. """ manager = DagFileProcessorManager( max_runs=1,