Don't deactivate DAG bundles owned by other dag-processors

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

@@ -271,7 +271,7 @@ def parse_config(self) -> None:
         self.log.info("DAG bundles loaded: %s", ", ".join(self._bundle_config.keys()))
 
     @provide_session
-    def sync_bundles_to_db(self, *, session: Session = NEW_SESSION) -> None:
+    def sync_bundles_to_db(self, *, deactivate_missing: bool = True, session: Session = NEW_SESSION) -> None:
         """
         Persist the configured DAG bundles into ``DagBundleModel`` rows.
 
@@ -280,6 +280,14 @@ def sync_bundles_to_db(self, *, session: Session = NEW_SESSION) -> None:
         ``DagModel`` / ``SerializedDagModel`` rows is the responsibility of
         ``DagBag`` plus ``sync_bag_to_db`` (or, in production, the DAG
         processor); calling this method does not trigger that work.
+
+        :param deactivate_missing: When ``True`` (the default), any bundle stored in
+            the database that is not present in this manager's config is marked
+            inactive. This is only correct when the calling process sees the
+            *complete* bundle configuration. When multiple dag-processors each run
+            with a partial config (e.g. one bundle per processor), pass ``False`` so a
+            processor does not disable bundles owned by other processors -- otherwise
+            the processors repeatedly deactivate each other and never converge.
         """
         self.log.debug("Syncing DAG bundles to the database")
 
@@ -356,6 +364,11 @@ def _extract_and_sign_template(bundle_name: str) -> tuple[str | None, dict]:
                 )
                 bundle.teams = []
 
+        if not deactivate_missing:
+            # This process only owns a subset of the bundles, so the remaining stored
+            # bundles may well be owned by another dag-processor. Leave them alone.
+            return
+
         # Import here to avoid circular import
         from airflow.models.errors import ParseImportError
 

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

@@ -333,7 +333,9 @@ def _exit_gracefully(self, signum, frame):
 
     def sync_bundles(self) -> None:
         """Sync configured DAG bundles to the metadata database."""
-        DagBundlesManager().sync_bundles_to_db()
+        # When this processor only parses a subset of bundles, it does not see the full
+        # bundle configuration and must not deactivate bundles owned by other processors.
+        DagBundlesManager().sync_bundles_to_db(deactivate_missing=not self.bundle_names_to_parse)
 
     def get_all_bundles(self) -> list[BaseDagBundle]:
         """Return configured DAG bundles filtered by ``bundle_names_to_parse`` if provided."""

airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py

@@ -119,6 +119,14 @@ def path(self):
     }
 ]
 
+OTHER_BUNDLE_CONFIG = [
+    {
+        "name": "other-test-bundle",
+        "classpath": "unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle",
+        "kwargs": {"refresh_interval": 1},
+    }
+]
+
 
 def test_get_bundle():
     """Test that get_bundle builds and returns a bundle."""
@@ -217,6 +225,54 @@ def test_sync_bundles_to_db_does_not_log_removing_none_team(clear_db, caplog):
     assert "Removing ownership of team 'None'" not in caplog.text
 
 
+@pytest.mark.db_test
+@conf_vars({("core", "LOAD_EXAMPLES"): "False"})
+def test_sync_bundles_to_db_partial_config_does_not_disable_other_bundles(clear_db, session):
+    """
+    Multiple dag-processors, each configured with only its own bundle, must not
+    disable each other's bundles.
+
+    Each processor is started with ``--bundle-name`` (``bundle_names_to_parse``)
+    and a ``dag_bundle_config_list`` containing only the bundle it owns. When
+    ``sync_bundles_to_db`` is told it only owns a subset (``deactivate_missing=False``),
+    it must leave bundles it does not know about untouched instead of marking them
+    inactive. Otherwise processor A disables B's bundle, B disables A's, and neither
+    ever makes progress.
+    """
+
+    def _get_bundle_names_and_active():
+        return session.execute(
+            select(DagBundleModel.name, DagBundleModel.active).order_by(DagBundleModel.name)
+        ).all()
+
+    # Processor A: only knows "my-test-bundle".
+    with patch.dict(
+        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
+    ):
+        DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
+    assert _get_bundle_names_and_active() == [("my-test-bundle", True)]
+
+    # Processor B: only knows "other-test-bundle". It must not touch A's bundle.
+    with patch.dict(
+        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(OTHER_BUNDLE_CONFIG)}
+    ):
+        DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
+    assert _get_bundle_names_and_active() == [
+        ("my-test-bundle", True),
+        ("other-test-bundle", True),
+    ]
+
+    # Processor A runs again: still must not disable B's bundle.
+    with patch.dict(
+        os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
+    ):
+        DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
+    assert _get_bundle_names_and_active() == [
+        ("my-test-bundle", True),
+        ("other-test-bundle", True),
+    ]
+
+
 @conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(BASIC_BUNDLE_CONFIG)})
 @pytest.mark.parametrize("version", [None, "hello"])
 def test_view_url(version):

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

@@ -295,6 +295,20 @@ def test_get_observed_filelocs_expands_zip_inner_paths(self, tmp_path):
             "test_zip.zip/broken_dag.py",
         }
 
+    def test_sync_bundles_deactivates_missing_when_owning_all_bundles(self):
+        """A processor with no bundle filter owns the full config and may deactivate missing bundles."""
+        manager = DagFileProcessorManager(max_runs=1)
+        with mock.patch("airflow.dag_processing.manager.DagBundlesManager") as mock_bundles_manager:
+            manager.sync_bundles()
+        mock_bundles_manager.return_value.sync_bundles_to_db.assert_called_once_with(deactivate_missing=True)
+
+    def test_sync_bundles_does_not_deactivate_missing_when_filtered(self):
+        """A processor started with ``--bundle-name`` owns a subset and must not deactivate others."""
+        manager = DagFileProcessorManager(max_runs=1, bundle_names_to_parse=["only-mine"])
+        with mock.patch("airflow.dag_processing.manager.DagBundlesManager") as mock_bundles_manager:
+            manager.sync_bundles()
+        mock_bundles_manager.return_value.sync_bundles_to_db.assert_called_once_with(deactivate_missing=False)
+
     @pytest.mark.usefixtures("clear_parse_import_errors")
     def test_refresh_dag_bundles_keeps_zip_inner_file_errors(self, session, tmp_path, configure_dag_bundles):
         bundle_path = tmp_path / "bundleone"