Parse native TypeScript Dags with the Node coordinator

airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst

@@ -49,6 +49,9 @@ Prerequisites
 -------------
 
 * Node.js 22 or later must be available on the Airflow worker nodes and the Dag processor.
+  Once a ``NodeCoordinator`` is configured, the Dag processor runs ``node`` on every packed ``*.min.mjs`` bundle
+  in every Dag bundle, including ones that only register ``TaskHandler`` objects,
+  and reports an import error for each when ``node`` is missing.
 * The packed bundle (a single ``bundle.min.mjs`` file, see :ref:`typescript-sdk/build`) must be accessible
   from the worker and the Dag processor, in the Dag bundle the coordinator scans.
 * The ``apache-airflow-task-sdk`` package (installed with Airflow) provides the coordinator; no additional
@@ -241,13 +244,24 @@ located.
 There is no separate Node.js worker to run: the Airflow worker launches the bundle with ``node`` once per
 task instance.
 
+A Dag processor with this ``[sdk]`` configuration also parses the ``*.min.mjs`` bundles of every Dag bundle,
+and needs Node.js to do so (see :ref:`typescript-sdk/native-parsing`).
+The ``ts-task-handlers`` Dag bundle only holds the bundles that the Python stub Dag's tasks run,
+so keep the Dag processor from parsing it by listing ``*`` in its ``.airflowignore``:
+
+.. code-block:: bash
+
+    echo '*' > /opt/airflow/ts-bundles/.airflowignore
+
 .. note::
 
   The ``[sdk]`` config, the packed ``*.min.mjs`` bundles and Node.js must be present wherever tasks execute
   and on the Dag processor. With ``CeleryExecutor``, tasks execute on the Celery workers; with
   ``LocalExecutor``, they run inside the scheduler process. The Dag processor checks the stub tasks of each
-  Python Dag against the task handlers the packed bundles register, so it runs them too. The API server
-  does not need any of it. Register the Dag bundle in ``[dag_processor] dag_bundle_config_list`` on every
+  Python Dag against the task handlers the packed bundles register, so it runs them too.
+  It also runs them to parse the bundles of every Dag bundle, see :ref:`typescript-sdk/native-parsing`.
+  The API server does not need any of it.
+  Register the Dag bundle in ``[dag_processor] dag_bundle_config_list`` on every
   component, like your other Dag bundles: the worker and the Dag processor resolve
   ``task_handler_bundle_name`` through it, and wherever the ``[sdk]`` config is read it is rejected if the
   name is missing there.
@@ -258,8 +272,8 @@ Declaring a Dag in TypeScript
 -----------------------------
 
 A ``Dag`` is declared on this side rather than in Python: its schedule, its tasks, their options and
-the edges between them are all written in TypeScript. The surface is still growing, so a Dag declared
-this way is not served to Airflow yet.
+the edges between them are all written in TypeScript. Airflow parses such a Dag from its packed bundle,
+see :ref:`typescript-sdk/native-parsing`.
 
 ``dag.task(taskId, handler)`` returns a *factory*. A handler takes one object of named arguments, and
 calling the factory names each input, so the call graph is the task graph:
@@ -497,12 +511,11 @@ The code is minified because an integrity digest is only worth taking over an ar
 read or edit in place. Function names are kept through minification, since a task id defaults to its
 handler's name. The ``/*! */`` license banners of bundled dependencies are kept.
 
-Because the shipped code is not the code anyone wrote, the packer also embeds each Dag-defining source
-file verbatim in its own ``/*# airflowSource:<path> ... #*/`` block comment, verified by its own digest,
-so Airflow has something readable to display for each Dag. Each native Dag's file is embedded — the file
-its ``new Dag(...)`` constructor ran in — so a bundle that declares its Dags across several files gets one
-source region per file, mapped to their ``dag_id`` by the ``dag_source_paths`` field in the manifest. Files
-that only supply utilities or types are not embedded.
+Because the shipped code is not the code anyone wrote, the packer also embeds source files verbatim, each in
+its own ``/*# airflowSource:<path> ... #*/`` block comment verified by its own digest: the entry module, and
+each native Dag's file (the file its ``new Dag(...)`` constructor ran in). The ``dag_source_paths`` field in
+the manifest maps each ``dag_id`` to its file. Files that only supply utilities or types are not embedded.
+The Code view currently shows the embedded entry module for each Dag in the bundle.
 
 ``esbuild`` is an optional peer dependency: packing is build-time only, so the runtime install of
 ``apache-airflow-ts-sdk`` skips it, and it must be installed separately before running ``airflow-ts-pack``.
@@ -522,10 +535,71 @@ Deploying
 ~~~~~~~~~
 
 Copy or mount the bundle into the Dag bundle named by the coordinator's ``task_handler_bundle_name``.
+For a task of a Python stub Dag,
 :class:`~airflow.sdk.coordinators.node.NodeCoordinator` searches that Dag bundle recursively and launches the
 first integrity-verified ``*.min.mjs`` bundle whose metadata declares the task instance's Dag. The artifact's
 name does not matter beyond that suffix, so one Dag bundle can hold several bundles and a Dag is routed to
 whichever declares it. If multiple bundles declare the same Dag, the first in sorted path order wins.
+A task of a native TypeScript Dag runs the bundle its Dag was parsed from, see
+:ref:`typescript-sdk/native-parsing`.
+
+.. _typescript-sdk/native-parsing:
+
+Parsing native Dags
+-------------------
+
+A Dag declared in TypeScript ships in its packed bundle, so the bundle goes into a Dag bundle,
+next to any Python Dag files.
+To have Airflow parse it, configure a :class:`~airflow.sdk.coordinators.node.NodeCoordinator`.
+The Dag processor runs ``node`` on the bundle to list its Dags, so it needs Node.js, as the workers do:
+
+.. code-block:: ini
+
+    [sdk]
+    coordinators = {
+      "ts": {
+        "classpath": "airflow.sdk.coordinators.node.NodeCoordinator",
+        "kwargs": {"node_executable": "/usr/local/bin/node"}
+      }
+    }
+    queue_to_coordinator = {"typescript": "ts"}
+
+Once a ``NodeCoordinator`` is configured, the Dag processor parses the packed bundles of every Dag bundle,
+so it needs this ``[sdk]`` configuration and Node.js. With one ``NodeCoordinator``, it parses them all. With several,
+map each Dag bundle that holds native TypeScript Dags to one of them in ``[sdk] dag_bundle_to_coordinator``.
+Each packed bundle of a Dag bundle that has no entry, or whose entry names no ``NodeCoordinator``,
+fails to parse with an import error:
+
+.. code-block:: ini
+
+    [sdk]
+    dag_bundle_to_coordinator = {"dags-folder": "ts"}
+
+A Dag bundle that holds only the bundles that Python stub Dags' tasks run should list ``*`` in its ``.airflowignore``.
+Otherwise, with several Node coordinators, its bundles fail to parse.
+
+A task of a native Dag runs the bundle file its Dag was parsed from, on the ``NodeCoordinator`` its queue routes to,
+which need not be the one that parsed it. For example, a queue can route to a coordinator that uses another Node.js.
+A task whose queue routes to another kind of coordinator fails without retries. A task whose bundle is missing,
+or that the coordinator cannot run (for example, because the bundle fails its integrity check),
+fails and retries while it has retries left.
+A task of a Python stub Dag runs the first bundle, in sorted path order, that declares its Dag.
+
+The Dag processor parses only files that end in ``.min.mjs`` and start with the header ``airflow-ts-pack`` writes.
+A bundle that fails its integrity check, or whose Dags cannot be serialized or fail validation,
+such as a cycle drawn with ``before`` and ``after``, is reported as an import error.
+
+Do not declare a Dag in TypeScript that a Python file in the same bundle also defines. When you move a Dag
+such as the Quick start's ``typescript_example`` to ``new Dag(...)``, remove its Python stub, otherwise the
+two files overwrite each other's Dag on every parse.
+
+The Code view currently shows the bundle's entry module for each of its Dags, as ``airflow-ts-pack``
+embeds it (see :ref:`typescript-sdk/build`). If the source cannot be read from the bundle, the view shows a
+short notice instead.
+
+A bundle that only registers ``TaskHandler`` objects is parsed too,
+since its metadata does not say whether it declares Dags. Each parse then launches ``node`` once and finds no Dags.
+To avoid that cost, list such bundles in ``.airflowignore``; the coordinator still finds them to run tasks.
 
 .. _typescript-sdk/coordinator-config:
 
@@ -568,12 +642,22 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
   * If ``task_handler_bundle_name`` is unset, the bundle is read from the **task's own** Dag bundle, pinned
     to the version the run was created with.
 
+  A task of a native TypeScript Dag ignores ``task_handler_bundle_name``:
+  it runs the bundle of its Dag from the Dag's own bundle, at the version the run was created with.
+  See :ref:`typescript-sdk/native-parsing`.
+
 Limitations
 -----------
 
 * **A Python stub Dag is still required.** The Execution API does not yet carry Dag structure for non-Python
   languages, so task names and dependencies are declared in Python with
   :func:`@task.stub <airflow.sdk.task.stub>`.
+* **Cluster policies do not apply to a native Dag.** ``dag_policy`` and ``task_policy`` are not run on
+  it, so they cannot change or reject it.
+* **Some CLI commands do not take a native Dag.** ``airflow dags test``, ``tasks test`` and ``tasks render`` refuse it,
+  and ``airflow dags reserialize`` does not store the Dags of ``*.min.mjs`` bundles,
+  which only the Dag processor stores.
+  ``airflow tasks list`` lists its tasks by running the bundle with ``node``, so it needs Node.js.
 * **Beta status.** The SDK API may change in incompatible ways between releases.
 * **One Node.js subprocess per task instance.** Tasks that need to share in-process state between instances
   should use XCom or an external store instead.

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

@@ -619,7 +619,7 @@ def update_dag_parsing_results_in_db(
     :param files_parsed: Set of (bundle_name, relative_fileloc) tuples for all files that were parsed.
         If None, will be inferred from dags and import_errors. Passing this explicitly ensures that
         import errors are cleared for files that were parsed but no longer contain DAGs.
-    :param dag_source_codes: Source code read by the Dag importers, keyed by Dag fileloc. Dags
+    :param dag_source_codes: Source code read by the Dag importers, keyed by dag_id. Dags
         without an entry have their source read from ``fileloc``.
     """
     accepted = _reject_other_teams_plugin_classes(bundle_name, dags, import_errors, session=session)
@@ -670,7 +670,7 @@ def update_dag_parsing_results_in_db(
                             version_data=version_data,
                             session=session,
                             _prefetched=prefetched_metadata.get(dag.dag_id),
-                            dag_source_code=dag_source_codes.get(dag.fileloc),
+                            dag_source_code=dag_source_codes.get(dag.dag_id),
                         )
                     )
             except OperationalError:

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

@@ -327,17 +327,21 @@ def _process_definition(
 
         self.captured_warnings.pop(fileloc, None)
         self._import_warnings.pop(fileloc, None)
-        self.dag_source_codes.pop(fileloc, None)
+        # Keyed by dag_id, not fileloc: a bundle can embed several Dags' own source under one
+        # fileloc, so a stale entry is found by which dag_ids this fileloc previously bagged.
+        for dag_id, dag in list(self.dags.items()):
+            if dag.fileloc == fileloc:
+                self.dag_source_codes.pop(dag_id, None)
         bagged_dags: list[BaggedDAG]
         if self.parse_lang_sdk_files and is_coordinator_importer(importer):
             bagged_dags = list(self._import_lang_sdk_definition(definition))
         else:
             bagged_dags = list(self._import_sdk_definition(importer, definition))
 
         self.file_last_changed[fileloc] = freshness_token
-        if bagged_dags:
+        for dag in bagged_dags:
             try:
-                self.dag_source_codes[fileloc] = importer.get_source_code(definition)
+                self.dag_source_codes[dag.dag_id] = importer.get_source_code(definition, dag.dag_id)
             except Exception:
                 self.log.exception("Failed to read source code of %s", fileloc)
         return bagged_dags

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

@@ -473,7 +473,7 @@ def _apply_team_rules(self, data: dict, dag: SerializedDAG) -> None:
 
     def _read_dag_source_codes(self, serialized_dags: list[LazyDeserializedDAG]) -> dict[str, DagSourceCode]:
         """
-        Read the file's source with its Dag importer, for the fileloc of each Dag.
+        Read each Dag's own source with its Dag importer, keyed by dag_id.
 
         A binary artifact cannot be read as text, so a source that cannot be read is a placeholder.
         """
@@ -483,11 +483,25 @@ def _read_dag_source_codes(self, serialized_dags: list[LazyDeserializedDAG]) ->
         try:
             if (importer := get_claiming_importer(file, self.bundle_name)) is None:
                 raise RuntimeError(f"No coordinator Dag importer claims {file}")
-            source = importer.get_source_code(FilesystemDagDefinition(Path(file)))
         except Exception as e:
             self.process_log.warning("Cannot read the Dag source", fileloc=file, error=str(e))
-            source = DagSourceCode(f"Cannot read the source of {self.dag_file_rel_path}: {e}", "text")
-        return {dag.data["dag"].get("fileloc", file): source for dag in serialized_dags}
+            placeholder = DagSourceCode(f"Cannot read the source of {self.dag_file_rel_path}: {e}", "text")
+            return dict.fromkeys((dag.dag_id for dag in serialized_dags), placeholder)
+
+        sources: dict[str, DagSourceCode] = {}
+        for dag in serialized_dags:
+            try:
+                sources[dag.dag_id] = importer.get_source_code(
+                    FilesystemDagDefinition(Path(file)), dag.dag_id
+                )
+            except Exception as e:
+                self.process_log.warning(
+                    "Cannot read the Dag source", fileloc=file, dag_id=dag.dag_id, error=str(e)
+                )
+                sources[dag.dag_id] = DagSourceCode(
+                    f"Cannot read the source of {self.dag_file_rel_path}: {e}", "text"
+                )
+        return sources
 
     _request_handlers: ClassVar[dict[type[BaseModel], RequestHandler[LangSDKDagFileProcessorProcess]]] = {
         **BaseDagFileProcessorProcess._common_request_handlers,

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

@@ -132,7 +132,7 @@ class DagFileParsingResult(BaseModel):
     parsed_definitions: list[str] = Field(default_factory=list)
     """Bundle-relative locations of the Dag definitions imported from ``fileloc``."""
     dag_source_codes: dict[str, DagSourceCode] = Field(default_factory=dict)
-    """Source code of the parsed Dags, keyed by Dag fileloc."""
+    """Source code of the parsed Dags, keyed by dag_id."""
     type: Literal["DagFileParsingResult"] = "DagFileParsingResult"
 
 

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

@@ -68,7 +68,7 @@ class FakeCoordinatorDagImporter(CoordinatorDagImporter):
     artifact_suffix = ".native"
     supported_extensions = [".native"]
 
-    def get_source_code(self, definition) -> DagSourceCode:
+    def get_source_code(self, definition, dag_id: str | None = None) -> DagSourceCode:
         return DagSourceCode(definition.read_text(), "fake")
 
 

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

@@ -956,7 +956,7 @@ def test_dag_source_codes_are_written_to_dag_code(self, testing_dag_bundle, sess
             parse_duration=None,
             warnings=set(),
             session=session,
-            dag_source_codes={dag.fileloc: DagSourceCode(source_code="dag_id: yaml_dag\n", language="yaml")},
+            dag_source_codes={dag.dag_id: DagSourceCode(source_code="dag_id: yaml_dag\n", language="yaml")},
         )
 
         dag_code = DagCode.get_latest_dagcode("yaml_dag", session=session)

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

@@ -396,7 +396,7 @@ def import_definition(self, definition, bundle):
             ],
         )
 
-    def get_source_code(self, definition):
+    def get_source_code(self, definition, dag_id: str | None = None):
         return DagSourceCode(source_code=definition.read_text(), language="text")
 
 
@@ -775,12 +775,8 @@ def test_dag_source_codes(self, tmp_path):
         dagbag = DagBag(dag_folder=tmp_path, bundle_path=tmp_path, bundle_name="test_bundle")
 
         assert dagbag.dag_source_codes == {
-            os.fspath(tmp_path / "plain.py"): DagSourceCode(
-                source_code=_dag_source("plain"), language="python"
-            ),
-            os.fspath(tmp_path / "dags.zip" / "member.py"): DagSourceCode(
-                source_code=_dag_source("member"), language="python"
-            ),
+            "plain": DagSourceCode(source_code=_dag_source("plain"), language="python"),
+            "member": DagSourceCode(source_code=_dag_source("member"), language="python"),
         }
 
     def test_dag_source_codes_drops_stale_source_when_read_fails(self, tmp_path):
@@ -808,9 +804,7 @@ def test_sync_bag_to_db_passes_parsed_files_and_source_codes(self, mock_update,
         kwargs = mock_update.call_args.kwargs
         assert kwargs["files_parsed"] == {("test_bundle", "dags.zip/member.py"), ("test_bundle", "dags.zip")}
         assert kwargs["dag_source_codes"] == {
-            os.fspath(tmp_path / "dags.zip" / "member.py"): DagSourceCode(
-                source_code=_dag_source("member"), language="python"
-            )
+            "member": DagSourceCode(source_code=_dag_source("member"), language="python")
         }
 
     def test_zip(self, tmp_path, test_zip_path):
@@ -1402,7 +1396,7 @@ def test_custom_importer_dags_are_bound_to_their_definition(self, tmp_path):
         assert dag.fileloc == os.fspath(dag_file)
         assert dag.relative_fileloc == "nested/text_dag.dagtxt"
         assert dagbag.dag_source_codes == {
-            os.fspath(dag_file): DagSourceCode(source_code="text_dag\n", language="text")
+            "text_dag": DagSourceCode(source_code="text_dag\n", language="text")
         }
         assert dagbag.dag_warnings == {DagWarning("text_dag", "test:deprecated_field", "Deprecated field")}
         assert dagbag.captured_warnings == {
@@ -1664,7 +1658,8 @@ def test_native_file_is_bagged_as_serialized_dags(self, mock_parse_dag, tmp_path
         assert isinstance(dag.task_dict["extract"], SerializedBaseOperator)
         assert (dag.fileloc, dag.relative_fileloc) == (os.fspath(native), "sub/dags.native")
         assert dag.bundle_name == "testing"
-        assert dagbag.dag_source_codes[os.fspath(native)].source_code == "{}"
+        assert dagbag.dag_source_codes["native_a"].source_code == "{}"
+        assert dagbag.dag_source_codes["native_b"].source_code == "{}"
 
     def test_native_file_is_an_import_error_by_default(self, tmp_path):
         write_native_file(tmp_path / "dags.native")