[v3-4-test] Stop a stale callback stat from forcing constant Dag file reparsing (#73433)

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

@@ -1573,8 +1573,21 @@ def _resort_file_queue(self):
             # Put callback files at the front, then sorted regular files
             self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files)
 
+    def _resolve_stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]:
+        """
+        Resolve one stat per physical file from the bundle scan's entries.
+
+        A callback pins ``bundle_version``, so its entry is separate from the scan's and does
+        not track parses of the current file: a callback-only result has no ``last_finish_time``,
+        and a timed-out callback processor is stamped although nothing parsed the current version.
+        A file with no scan entry has no stat, so it is queued.
+        """
+        return {
+            file.presence_key: stat for file, stat in self._file_stats.items() if file.bundle_version is None
+        }
+
     def _sort_by_mtime(self, files: Iterable[DagFileInfo]):
-        file_stats_by_presence_key = {file.presence_key: stat for file, stat in self._file_stats.items()}
+        file_stats_by_presence_key = self._resolve_stats_by_presence_key()
         files_with_mtime: dict[DagFileInfo, float] = {}
         changed_recently = set()
         for file in files:
@@ -1624,19 +1637,19 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
 
         Note this method is only called when the file path queue is empty
         """
+        file_stats_by_presence_key = self._resolve_stats_by_presence_key()
         # We only emit metrics after processing all files in the queue. If `self._parsing_start_time` is None
         # when this method is called, no files have yet been added to the queue so we shouldn't emit metrics.
         if self._parsing_start_time is not None:
             emit_metrics(
                 parse_time=time.perf_counter() - self._parsing_start_time,
-                dag_file_stats=list(self._file_stats.values()),
+                dag_file_stats=list(file_stats_by_presence_key.values()),
             )
             self._parsing_start_time = None
 
         # If the file path is already being processed, or if a file was
         # processed recently, wait until the next batch
         in_progress_keys = {file.presence_key for file in self._processors}
-        file_stats_by_presence_key = {file.presence_key: stat for file, stat in self._file_stats.items()}
         now = timezone.utcnow()
 
         # Sort the file paths by the parsing order mode

airflow-core/tests/unit/dag_processing/test_manager.py

@@ -1147,7 +1147,8 @@ def test_prepare_file_queue_skips_file_when_versioned_processor_is_in_progress(s
         assert manager._file_queue == OrderedDict()
 
     @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
-    def test_prepare_file_queue_skips_file_when_versioned_stat_is_at_run_limit(self):
+    def test_prepare_file_queue_ignores_versioned_stat_at_run_limit(self):
+        """A callback's versioned stat doesn't stand for a parse, so it doesn't count towards the run limit."""
         manager = DagFileProcessorManager(max_runs=1)
         versioned_file = _get_versioned_file_info("file_1.py")
         known_file = _get_file_infos(["file_1.py"])[0]
@@ -1156,10 +1157,11 @@ def test_prepare_file_queue_skips_file_when_versioned_stat_is_at_run_limit(self)
 
         manager.prepare_file_queue(known_files={"testing": {known_file}})
 
-        assert manager._file_queue == OrderedDict()
+        assert manager._file_queue == OrderedDict.fromkeys([known_file])
 
     @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
-    def test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats(self):
+    def test_prepare_file_queue_ignores_recent_versioned_stat(self):
+        """A callback's versioned stat doesn't stand for a parse, so a recent one doesn't hold the file back."""
         manager = DagFileProcessorManager(max_runs=3)
         versioned_file = _get_versioned_file_info("file_1.py")
         known_file = _get_file_infos(["file_1.py"])[0]
@@ -1171,11 +1173,11 @@ def test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats(s
 
         manager.prepare_file_queue(known_files={"testing": {known_file}})
 
-        assert manager._file_queue == OrderedDict()
+        assert manager._file_queue == OrderedDict.fromkeys([known_file])
 
     @conf_vars({("dag_processor", "file_parsing_sort_mode"): "modified_time"})
     @mock.patch("airflow.utils.file.os.path.getmtime")
-    def test_recently_modified_file_uses_versioned_stats_without_creating_duplicate_entries(
+    def test_recently_modified_file_with_versioned_stat_does_not_create_duplicate_entries(
         self, mock_getmtime
     ):
         freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0)
@@ -1197,6 +1199,106 @@ def test_recently_modified_file_uses_versioned_stats_without_creating_duplicate_
         assert known_file not in manager._file_stats
         assert versioned_file in manager._file_stats
 
+    @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
+    @pytest.mark.parametrize(
+        "callback_first",
+        [
+            pytest.param(False, id="scan-recorded-first"),
+            pytest.param(True, id="callback-recorded-first"),
+        ],
+    )
+    def test_callback_only_parse_does_not_requeue_recently_parsed_file(self, callback_first):
+        """A callback-only parse records no ``last_finish_time``, which must not requeue the file."""
+        freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0)
+        known_file = _get_file_infos(["file_1.py"])[0]
+        versioned_file = _get_versioned_file_info("file_1.py")
+        scan_stat = DagFileStat(1, 0, freezed_base_time - timedelta(seconds=5), 1.0, 1, 1)
+
+        manager = DagFileProcessorManager(max_runs=3)
+        processor, _ = self.mock_processor(start_time=time.monotonic() - 1)
+        processor.had_callbacks = True
+        processor.parsing_result = None
+
+        with time_machine.travel(freezed_base_time):
+            if not callback_first:
+                manager._file_stats[known_file] = scan_stat
+            manager.handle_parsing_result(versioned_file, processor)
+            assert manager._file_stats[versioned_file].last_finish_time is None
+            if callback_first:
+                manager._file_stats[known_file] = scan_stat
+            manager.prepare_file_queue(known_files={"testing": {known_file}})
+
+        assert manager._file_queue == OrderedDict()
+
+    @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
+    @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True)
+    def test_timed_out_callback_does_not_delay_due_parse(self, mock_kill):
+        """A callback processor that times out must not make a due file look recently parsed."""
+        freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0)
+        known_file = _get_file_infos(["file_1.py"])[0]
+        versioned_file = _get_versioned_file_info("file_1.py")
+
+        manager = DagFileProcessorManager(max_runs=3, processor_timeout=5)
+        manager._file_stats[known_file] = DagFileStat(
+            1, 0, freezed_base_time - timedelta(seconds=60), 1.0, 1, 1
+        )
+        processor, _ = self.mock_processor(start_time=time.monotonic() - manager.processor_timeout - 1)
+        manager._processors = {versioned_file: processor}
+
+        with time_machine.travel(freezed_base_time):
+            manager._kill_timed_out_processors()
+            mock_kill.assert_called_once_with(processor, signal.SIGKILL)
+            assert manager._file_stats[versioned_file].last_finish_time is not None
+            manager.prepare_file_queue(known_files={"testing": {known_file}})
+
+        assert known_file in manager._file_queue
+
+    @conf_vars({("dag_processor", "file_parsing_sort_mode"): "modified_time"})
+    @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True)
+    @mock.patch("airflow.utils.file.os.path.getmtime")
+    def test_timed_out_callback_does_not_hide_an_edit(self, mock_getmtime, mock_kill):
+        """An edit after the last parse is picked up even if a callback processor times out later."""
+        freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0)
+        known_file = _get_file_infos(["file_1.py"])[0]
+        versioned_file = _get_versioned_file_info("file_1.py")
+
+        manager = DagFileProcessorManager(max_runs=3, processor_timeout=5)
+        manager._file_stats[known_file] = DagFileStat(
+            1, 0, freezed_base_time - timedelta(seconds=10), 1.0, 1, 1
+        )
+        processor, _ = self.mock_processor(start_time=time.monotonic() - manager.processor_timeout - 1)
+        manager._processors = {versioned_file: processor}
+        mock_getmtime.return_value = (freezed_base_time - timedelta(seconds=7)).timestamp()
+
+        with time_machine.travel(freezed_base_time - timedelta(seconds=5)):
+            manager._kill_timed_out_processors()
+        mock_kill.assert_called_once_with(processor, signal.SIGKILL)
+        assert manager._file_stats[versioned_file].last_finish_time is not None
+        with time_machine.travel(freezed_base_time):
+            manager.prepare_file_queue(known_files={"testing": {known_file}})
+
+        assert known_file in manager._file_queue
+
+    @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"})
+    @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True)
+    @mock.patch("airflow.dag_processing.manager.stats.gauge")
+    def test_timed_out_callback_is_not_reported_as_an_import_error(self, mock_gauge, mock_kill):
+        """The ``import_errors=1`` a timeout leaves on a callback's entry must not reach the gauge."""
+        known_file = _get_file_infos(["file_1.py"])[0]
+        versioned_file = _get_versioned_file_info("file_1.py")
+
+        manager = DagFileProcessorManager(max_runs=3, processor_timeout=5)
+        manager._file_stats[known_file] = DagFileStat(1, 2, timezone.datetime(2020, 1, 5), 1.0, 1, 1)
+        processor, _ = self.mock_processor(start_time=time.monotonic() - manager.processor_timeout - 1)
+        manager._processors = {versioned_file: processor}
+
+        manager._kill_timed_out_processors()
+        assert manager._file_stats[versioned_file].import_errors == 1
+        manager._parsing_start_time = time.perf_counter()
+        manager.prepare_file_queue(known_files={"testing": {known_file}})
+
+        mock_gauge.assert_any_call("dag_processing.import_errors", 2)
+
     def test_file_paths_in_queue_sorted_by_priority(self):
         from airflow.models.dagbag import DagPriorityParsingRequest