This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f3faf90ff69 TS SDK: minify packed bundles and find them by metadata,
not filename (#73126)
f3faf90ff69 is described below
commit f3faf90ff69a92054474de2bc70a4875a7ee36cd
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Fri Sep 18 20:55:51 2026 +0800
TS SDK: minify packed bundles and find them by metadata, not filename
(#73126)
NodeCoordinator looked for exactly `bundle.mjs` in each configured root, so
a
`bundles_root` could hold only one bundle and its name was fixed. That
defeats
the Dag-to-bundle routing the embedded metadata already supports, and it is
out
of step with JavaCoordinator and ExecutableCoordinator, which walk their
roots
and select on metadata. Each root is now searched recursively for
`*.min.mjs`
bundles, so one directory can hold several and each Dag is routed to
whichever
bundle declares it. Roots are searched in configured order and each
directory
in sorted path order, so which bundle wins no longer depends on the order a
filesystem returns entries in. `airflow-ts-pack` gains `--outfile` for
naming
the artifact.
esbuild output is also minified now. An integrity digest over the code
region
is only worth taking when the artifact is not something anyone is expected
to
read or edit in place, which is what `.min` records. Dependencies' license
banners are preserved. Identifier names are not: nothing identifies a Dag, a
task, or a handler by a function name, so there is nothing for minification
to
break, and ADR-0002 already settled that a bundle-wide `keepNames` is not
how
a name reaches the registration that needs it.
The bundle container format is unchanged; minification only changes which
bytes
land inside the already-digested code region.
---
.../language-sdks/typescript.rst | 42 +++++---
.../tests/airflow_e2e_tests/conftest.py | 3 +-
task-sdk/docs/ts-bundle-spec.rst | 116 +++++++++++----------
.../airflow/sdk/coordinators/_bundle_metadata.py | 58 ++++++++++-
.../airflow/sdk/coordinators/node/coordinator.py | 22 ++--
.../coordinators/node/_bundle_test_utils.py | 5 +-
.../coordinators/node/test_bundle_reader.py | 67 ++++++------
.../task_sdk/coordinators/node/test_coordinator.py | 82 ++++++++++++---
ts-sdk/README.md | 25 +++--
ts-sdk/example/README.md | 4 +-
ts-sdk/src/cli/pack.ts | 59 +++++++----
ts-sdk/tests/cli/pack.test.ts | 81 +++++++++++---
12 files changed, 386 insertions(+), 178 deletions(-)
diff --git
a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
index 7aaf6408ac8..bf7c78bc6a7 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -49,8 +49,8 @@ Prerequisites
-------------
* Node.js 22 or later must be available on the Airflow worker nodes.
-* The packed bundle (a single ``bundle.mjs`` file, see
:ref:`typescript-sdk/build`) must be accessible from
- the worker, under a directory the coordinator scans.
+* The packed bundle (a single ``bundle.min.mjs`` file, see
:ref:`typescript-sdk/build`) must be accessible
+ from the worker, under a directory the coordinator scans.
* The ``apache-airflow-task-sdk`` package (installed with Airflow) provides
the coordinator; no additional
Python packages are needed.
* In the TypeScript project, install the ``apache-airflow-ts-sdk`` npm package
to author task handlers:
@@ -163,8 +163,8 @@ task instance.
.. note::
- The coordinator runs inside the Airflow worker, so the ``[sdk]`` config (and
the packed ``bundle.mjs``
- files in ``bundles_root``) only need to be present wherever tasks actually
execute. With
+ The coordinator runs inside the Airflow worker, so the ``[sdk]`` config (and
the packed ``*.min.mjs``
+ bundles in ``bundles_root``) only need to be present wherever tasks actually
execute. With
``CeleryExecutor``, setting them on the Celery workers is sufficient. With
``LocalExecutor``, tasks run
inside the scheduler process, so they must be present where the scheduler
can read them. The API server
and Dag processor do not need them.
@@ -271,10 +271,15 @@ Building and packaging
----------------------
``airflow-ts-pack`` (shipped with the SDK) bundles the entry module and all of
its imports with esbuild into
-a single self-contained ESM file, ``bundle.mjs``, and embeds the manifest (the
``dag_id`` and ``task_id``
-map plus the supervisor schema version) after a leading compact JSON ``//#
airflowBundle=...`` layout header.
-The layout records the byte ranges and SHA-256 digests of the manifest and
executable code, so one file to
-deploy, with no separate manifest or ``node_modules``.
+a single self-contained, minified ESM file, ``bundle.min.mjs``, and embeds the
manifest (the ``dag_id`` and
+``task_id`` map plus the supervisor schema version) after a leading compact
JSON ``//# airflowBundle=...``
+layout header. The layout records the byte ranges and SHA-256 digests of the
manifest and executable code,
+so there is one file to deploy, with no separate manifest or ``node_modules``.
+
+The code is minified because an integrity digest is only worth taking over an
artifact nobody is expected to
+read or edit in place. The ``/*! */`` license banners of bundled dependencies
are kept. Nothing is identified by
+a function name, so minified names are safe: a Dag and a task are named by the
string ids their registration
+states, and a handler is dispatched by reference.
``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``.
@@ -284,16 +289,21 @@ deploy, with no separate manifest or ``node_modules``.
npm install --save-dev esbuild
npx airflow-ts-pack src/main.ts --outdir dist
-Use ``--outdir <dir>`` to choose the output directory (default ``dist``) and
``--source <name>`` to set the
-source name displayed in the Airflow UI (default: the entry file's basename).
+Use ``--outdir <dir>`` to choose the output directory (default ``dist``),
``--outfile <path>`` to name the
+artifact exactly, which helps when one ``bundles_root`` holds several bundles,
and ``--source <name>`` to set
+the source name displayed in the Airflow UI (default: the entry file's
basename). ``--outdir`` and
+``--outfile`` are mutually exclusive, and an ``--outfile`` name must end in
``.min.mjs`` so the coordinator
+can find it.
Deploying
~~~~~~~~~
-Copy or mount ``bundle.mjs`` into a directory listed in the coordinator's
``bundles_root``.
-:class:`~airflow.sdk.coordinators.node.NodeCoordinator` searches the
configured directories in order and
-launches the first integrity-verified bundle whose metadata declares the task
instance's Dag. If multiple
-bundles declare the same Dag, the first configured match wins.
+Copy or mount the bundle into a directory listed in the coordinator's
``bundles_root``.
+:class:`~airflow.sdk.coordinators.node.NodeCoordinator` searches the
configured directories in order,
+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
root can hold several bundles
+and a Dag is routed to whichever declares it. If multiple bundles declare the
same Dag, the first configured
+root wins, and within a root the first in sorted path order.
.. _typescript-sdk/coordinator-config:
@@ -312,8 +322,8 @@ All ``kwargs`` in the ``coordinators`` config entry are
passed to the
- Description
* - ``bundles_root``
- *(required)*
- - One or more directories searched, in order, for an integrity-verified
``bundle.mjs`` that declares
- the requested Dag. Accepts a string, a path, or a list of strings/paths.
+ - One or more directories searched recursively, in order, for an
integrity-verified ``*.min.mjs``
+ bundle that declares the requested Dag. Accepts a string, a path, or a
list of strings/paths.
* - ``node_executable``
- ``"node"``
- Path to the ``node`` binary. Defaults to ``node`` on ``$PATH``.
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
index 961f367b907..39ff3fd0a64 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
@@ -722,7 +722,8 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir):
# version from the metadata airflow-ts-pack embedded in the bundle.
ts_bundles_dir = tmp_dir / "ts-bundles"
ts_bundles_dir.mkdir()
- copyfile(TS_SDK_EXAMPLE_PATH / "dist" / "bundle.mjs", ts_bundles_dir /
"bundle.mjs")
+ # Deliberately renamed: the coordinator routes on embedded metadata, not
on a fixed name.
+ copyfile(TS_SDK_EXAMPLE_PATH / "dist" / "bundle.min.mjs", ts_bundles_dir /
"example.min.mjs")
# Both of the example bundle's Dags: one bundle.mjs provides for two
dag_ids,
# and the tests check that dispatch tells their same-named tasks apart.
diff --git a/task-sdk/docs/ts-bundle-spec.rst b/task-sdk/docs/ts-bundle-spec.rst
index 9aa46545132..82147f95e5c 100644
--- a/task-sdk/docs/ts-bundle-spec.rst
+++ b/task-sdk/docs/ts-bundle-spec.rst
@@ -18,18 +18,27 @@
TypeScript Bundle Format
========================
-This document specifies the ``bundle.mjs`` format produced by
``airflow-ts-pack`` and consumed by
:class:`~airflow.sdk.coordinators.node.NodeCoordinator`.
+This document specifies the bundle format produced by ``airflow-ts-pack`` and
consumed by
+:class:`~airflow.sdk.coordinators.node.NodeCoordinator`.
+
+Artifact Name
+-------------
+
+A bundle's name must end in ``.min.mjs``. Nothing else about it is
significant: the coordinator searches each
+configured root recursively and routes on embedded metadata, so one root may
hold several differently named bundles.
+``airflow-ts-pack`` writes ``bundle.min.mjs`` by default and accepts
``--outfile`` for any other name ending in
+that suffix.
Container
---------
-The bundle remains an ECMAScript module that runs directly with ``node
bundle.mjs``. It has three regions:
+The bundle remains an ECMAScript module that runs directly with ``node
bundle.min.mjs``. It has three regions:
.. code-block:: text
//# airflowBundle=<compact JSON layout>\n
//# airflowMetadata=<compact JSON>\n
- <bundled ECMAScript code>
+ <minified, bundled ECMAScript code>
The layout comes first so readers can locate and verify the other regions. The
current format has no embedded source region.
@@ -54,32 +63,25 @@ The ``airflowBundle`` payload is a compact UTF-8 JSON
object:
}
}
-Offsets are bytes from the beginning of ``bundle.mjs``. They use exactly 16
-lowercase hexadecimal digits and describe half-open ranges: ``start`` is
-included and ``end`` is excluded.
-
-The layout stays on one line and contains only controlled ASCII field names,
-fixed-width hexadecimal offsets, and SHA-256 digests. It therefore needs no
-additional encoding layer.
-
-The first-line layout is a stable bootstrap descriptor, not a separately
-versioned payload. The ``airflow_bundle_metadata_version`` stored in the
-metadata region versions the entire TypeScript bundle contract, including this
-physical framing and the decoded metadata schema. A reader parses the bounded,
-named ranges before it can locate and verify that version.
-
-The metadata range points to the UTF-8 JSON payload only, excluding the
-JavaScript comment marker and newline. Its digest therefore covers the exact
-JSON bytes stored in that range. The code range covers every byte after the
-metadata line through the end of the file. Its digest covers those raw
-JavaScript bytes.
-
-The file begins with the layout line. The metadata marker immediately follows
-that line, and exactly one newline separates the metadata payload from the code
-range. These prescribed framing bytes are outside the hashed metadata and code
-ranges; no additional bytes are permitted before, between, or after them.
-Post-pack formatters, compressors, source-map injectors, and other tools that
-rewrite ``bundle.mjs`` invalidate the offsets or digests.
+Offsets are bytes from the beginning of the bundle file. They use exactly 16
lowercase hexadecimal digits and
+describe half-open ranges: ``start`` is included and ``end`` is excluded.
+
+The layout stays on one line and contains only controlled ASCII field names,
fixed-width hexadecimal offsets, and
+SHA-256 digests. It therefore needs no additional encoding layer.
+
+The first-line layout is a stable bootstrap descriptor, not a separately
versioned payload. The
+``airflow_bundle_metadata_version`` stored in the metadata region versions the
entire TypeScript bundle contract,
+including this physical framing and the decoded metadata schema. A reader
parses the bounded, named ranges before it
+can locate and verify that version.
+
+The metadata range points to the UTF-8 JSON payload only, excluding the
JavaScript comment marker and newline. Its
+digest therefore covers the exact JSON bytes stored in that range. The code
range covers every byte after the
+metadata line through the end of the file, and its digest covers those raw
JavaScript bytes.
+
+The file begins with the layout line. The metadata marker immediately follows
that line, and exactly one newline
+separates the metadata payload from the code range. These prescribed framing
bytes are outside the hashed metadata
+and code ranges, and no additional bytes are permitted before, between, or
after them. Post-pack formatters,
+compressors, source-map injectors, and other tools that rewrite the bundle
invalidate the offsets or digests.
Metadata
--------
@@ -103,43 +105,43 @@ The ``airflowMetadata`` payload is compact UTF-8 JSON
with this logical shape:
}
}
-The packer serializes this object without insignificant whitespace and escapes
-the ECMAScript line and paragraph separators (U+2028 and U+2029), keeping it in
-one newline-terminated JavaScript comment without a second encoding layer. The
-SHA-256 digest detects changes to the exact serialized bytes.
+The packer serializes this object without insignificant whitespace and escapes
the ECMAScript line and paragraph
+separators (U+2028 and U+2029), keeping it in one newline-terminated
JavaScript comment without a second encoding
+layer. The SHA-256 digest detects changes to the exact serialized bytes.
-The coordinator uses the ``dags`` keys to choose a bundle for a task instance.
-The ``source`` value is a logical authoring name only; it is not embedded
source
-content and is not used to execute the bundle.
+The coordinator uses the ``dags`` keys to choose a bundle for a task instance.
The ``source`` value is a logical
+authoring name only, not embedded source content, and it is not used to
execute the bundle.
Reader and Selection Algorithm
------------------------------
-For each directory in ``bundles_root``, in configured order, the coordinator:
+For each candidate in ``bundles_root``, the coordinator:
-1. Looks for ``bundle.mjs`` and opens it once.
+1. Opens it once. Candidates are the files whose name ends in ``.min.mjs``,
found by walking each root recursively,
+ roots in configured order and each directory's entries in sorted order, so
selection does not depend on the order
+ a filesystem returns entries in. Directories are deduplicated by ``(st_dev,
st_ino)``, so a symlink loop
+ terminates the walk instead of exhausting the interpreter stack.
2. Reads a bounded first line and decodes the named metadata and code ranges.
-3. Reads the bounded metadata line and checks that the declared metadata and
- code ranges exactly match their physical locations and the file size.
+3. Reads the bounded metadata line and checks that the declared metadata and
code ranges exactly match their
+ physical locations and the file size.
4. Computes SHA-256 for both ranges before parsing or using metadata.
-5. Confirms with ``fstat`` that the open file did not change during
- verification.
-6. Parses metadata and requires a supported TypeScript bundle contract major
- version from ``airflow_bundle_metadata_version``.
-7. Skips the verified bundle if its ``dags`` mapping does not contain the
- requested ``dag_id``.
+5. Confirms with ``fstat`` that the open file did not change during
verification.
+6. Parses metadata and requires a supported bundle contract major version from
+ ``airflow_bundle_metadata_version``.
+7. Skips the verified bundle if its ``dags`` mapping does not contain the
requested ``dag_id``.
8. Resolves the supervisor schema version and selects the first usable match.
-A missing, unrelated, unreadable, malformed, corrupt, or incompatible earlier
-candidate does not prevent selection of a later usable match. When more than
one
-usable bundle declares the same Dag, the first configured match wins. If none
-matches, the error identifies the requested Dag, searched roots, and rejected
-candidates.
+A missing, unrelated, unreadable, malformed, corrupt, or incompatible earlier
candidate does not prevent selection
+of a later usable match. When more than one usable bundle declares the same
Dag, the first configured match wins. If
+none matches, the error identifies the requested Dag, searched roots, and
rejected candidates.
+
+Every ``.min.mjs`` file under a root is therefore opened, and one that is not
a usable bundle is named among those
+rejected candidates. ``bundles_root`` names directories of deployed Airflow
bundles, so unrelated minified modules do
+not belong there.
-The coordinator does not cache Dag-to-path routing. It checks root ordering and
-the current deployed files for each task selection. It may reuse section
digests
-from a bounded process-local cache when the open file identity, timestamps,
-size, layout ranges, and declared digests have not changed.
+The coordinator does not cache Dag-to-path routing. It checks root ordering
and the current deployed files for each
+task selection. It may reuse section digests from a bounded process-local
cache when the open file identity,
+timestamps, size, layout ranges, and declared digests have not changed.
Integrity, Authenticity, and Provenance
---------------------------------------
@@ -174,5 +176,5 @@ even when they cannot reach the metadata version. A future
container that
cannot preserve the readable first-line descriptor must use a new marker rather
than reinterpret the current one.
-The TypeScript packing workflow was unreleased when this format was added. The
-coordinator therefore does not accept the earlier metadata-first prototype.
+The TypeScript packing workflow was unreleased when this format was added. The
coordinator therefore does not accept
+the earlier metadata-first prototype.
diff --git a/task-sdk/src/airflow/sdk/coordinators/_bundle_metadata.py
b/task-sdk/src/airflow/sdk/coordinators/_bundle_metadata.py
index e95ffe52396..9a0ee5e5b32 100644
--- a/task-sdk/src/airflow/sdk/coordinators/_bundle_metadata.py
+++ b/task-sdk/src/airflow/sdk/coordinators/_bundle_metadata.py
@@ -21,13 +21,22 @@ from __future__ import annotations
import os
import pathlib
-from typing import Any
+import stat
+from typing import TYPE_CHECKING, Any
import attrs
+import structlog
import yaml
from airflow.sdk.execution_time.schema import get_schema_version_migrator
+if TYPE_CHECKING:
+ from collections.abc import Callable, Iterable, Iterator
+
+ from structlog.typing import FilteringBoundLogger
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators")
+
def convert_roots(
value: None | os.PathLike[str] | pathlib.Path | list[os.PathLike[str] |
pathlib.Path],
@@ -40,6 +49,53 @@ def convert_roots(
return [pathlib.Path(v).expanduser() for v in value]
+def walk_files(
+ roots: Iterable[pathlib.Path], *, match: Callable[[pathlib.Path], bool]
+) -> Iterator[pathlib.Path]:
+ """
+ Yield the regular files under *roots* that satisfy *match*, descending
into directories.
+
+ Roots are visited in order and each directory's entries sorted, so
coordinator selection does
+ not depend on filesystem ordering.
+
+ ``JavaCoordinator`` and ``ExecutableCoordinator`` still carry equivalent
walks and should move
+ onto this one.
+ """
+ yield from _walk_files(roots, match, set())
+
+
+def _walk_files(
+ items: Iterable[pathlib.Path],
+ match: Callable[[pathlib.Path], bool],
+ seen_dirs: set[tuple[int, int]],
+) -> Iterator[pathlib.Path]:
+ for item in items:
+ try:
+ file_info = item.stat()
+ except OSError:
+ # A broken symlink or unreadable parent must not abort the scan.
+ # The caller reports a genuinely missing artifact once every root
is searched.
+ continue
+ if stat.S_ISDIR(file_info.st_mode):
+ # Dedupe by identity so a symlink loop cannot recurse until the
stack is exhausted.
+ key = (file_info.st_dev, file_info.st_ino)
+ if key in seen_dirs:
+ log.debug("Skipping already-visited directory", path=item)
+ continue
+ seen_dirs.add(key)
+ yield from _walk_files(_sorted_children(item), match, seen_dirs)
+ elif stat.S_ISREG(file_info.st_mode) and match(item):
+ yield item
+
+
+def _sorted_children(directory: pathlib.Path) -> list[pathlib.Path]:
+ # iterdir() is lazy, so an unreadable directory raises only once iteration
starts.
+ try:
+ return sorted(directory.iterdir())
+ except OSError:
+ return []
+
+
def validate_schema_version(instance, _, value) -> str:
"""Attrs validator resolving a bundle's supervisor schema version to a
known one."""
return get_schema_version_migrator().resolve_version(str(value))
diff --git a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
index 98de32d5761..cd352f3871b 100644
--- a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
+++ b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
@@ -26,7 +26,7 @@ from typing import TYPE_CHECKING
import attrs
import structlog
-from airflow.sdk.coordinators._bundle_metadata import ResolvedBundle,
convert_roots
+from airflow.sdk.coordinators._bundle_metadata import ResolvedBundle,
convert_roots, walk_files
from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
from airflow.sdk.coordinators.node._bundle_reader import read_bundle
@@ -40,7 +40,11 @@ if TYPE_CHECKING:
log: FilteringBoundLogger =
structlog.get_logger(logger_name="coordinators.node")
-BUNDLE_FILENAME = "bundle.mjs"
+BUNDLE_SUFFIX = ".min.mjs"
+
+
+def _is_bundle(path: pathlib.Path) -> bool:
+ return path.name.endswith(BUNDLE_SUFFIX)
@attrs.define
@@ -48,18 +52,15 @@ class _Bundle(ResolvedBundle):
@classmethod
def find(cls, bundles_root: Sequence[pathlib.Path], dag_id: str) -> Self:
"""Return the first verified configured bundle that declares
*dag_id*."""
+ log.debug("Finding TypeScript bundles recursively",
roots=bundles_root, dag_id=dag_id)
rejected: list[tuple[pathlib.Path, str]] = []
- for root in bundles_root:
- candidate = root / BUNDLE_FILENAME
+ for candidate in walk_files(bundles_root, match=_is_bundle):
try:
- if not candidate.is_file():
- continue
metadata = read_bundle(candidate)
if dag_id not in metadata.dag_ids:
log.debug(
"TypeScript bundle does not contain requested Dag;
skipping",
path=candidate,
- root=root,
dag_id=dag_id,
)
rejected.append(
@@ -71,13 +72,12 @@ class _Bundle(ResolvedBundle):
log.debug(
"TypeScript bundle rejected; skipping",
path=candidate,
- root=root,
reason=str(exc),
exc_info=True,
)
rejected.append((candidate, str(exc)))
continue
- log.debug("Selected TypeScript bundle", path=candidate, root=root,
dag_id=dag_id)
+ log.debug("Selected TypeScript bundle", path=candidate,
dag_id=dag_id)
return bundle
searched = os.pathsep.join(os.fspath(root) for root in bundles_root)
@@ -110,8 +110,8 @@ class NodeCoordinator(SubprocessCoordinator):
:param node_executable: Path to the ``node`` binary (defaults to
``"node"``, which relies on ``$PATH``).
- :param bundles_root: Ordered list of directories scanned for the first
- verified ``bundle.mjs`` that declares the task instance's Dag.
+ :param bundles_root: Directories searched recursively, in order, for the
first verified
+ ``*.min.mjs`` bundle declaring the task instance's Dag.
:param task_startup_timeout: Maximum time the coordinator waits for a task
process to start, in seconds. The default is 10 seconds.
"""
diff --git a/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py
b/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py
index 303c95b77b7..0ec0a13cf7b 100644
--- a/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py
+++ b/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py
@@ -23,6 +23,7 @@ import json
import pathlib
SCHEMA_VERSION = "2026-06-16"
+BUNDLE_NAME = "bundle.min.mjs"
LAYOUT_PREFIX = b"//# airflowBundle="
METADATA_PREFIX = b"//# airflowMetadata="
OFFSET_WIDTH = 16
@@ -67,6 +68,7 @@ def write_bundle(
schema_version: str = SCHEMA_VERSION,
metadata_version: str | None = "1.0",
metadata_payload: bytes | None = None,
+ name: str = BUNDLE_NAME,
) -> pathlib.Path:
if metadata_payload is None:
metadata_payload = metadata_json(
@@ -92,7 +94,8 @@ def write_bundle(
)
assert len(layout_line) == len(placeholder)
- bundle = root / "bundle.mjs"
+ bundle = root / name
+ bundle.parent.mkdir(parents=True, exist_ok=True)
bundle.write_bytes(layout_line + metadata_line + code)
return bundle
diff --git a/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py
b/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py
index 1ff5808cf6f..3866028b658 100644
--- a/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py
+++ b/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py
@@ -26,6 +26,7 @@ from unittest import mock
import pytest
from task_sdk.coordinators.node._bundle_test_utils import (
+ BUNDLE_NAME,
LAYOUT_PREFIX,
METADATA_PREFIX,
OFFSET_WIDTH,
@@ -61,10 +62,10 @@ class TestBundleReader:
def test_rejects_metadata_first_legacy_bundle(self, tmp_path):
payload = _metadata_json("sales")
- (tmp_path / "bundle.mjs").write_bytes(METADATA_PREFIX + payload +
b"\nexport {};\n")
+ (tmp_path / BUNDLE_NAME).write_bytes(METADATA_PREFIX + payload +
b"\nexport {};\n")
with pytest.raises(ValueError, match="no airflow bundle layout"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
@pytest.mark.parametrize(
"metadata_version",
@@ -74,7 +75,7 @@ class TestBundleReader:
write_bundle(tmp_path, "sales", metadata_version=metadata_version)
with pytest.raises(ValueError, match="unsupported airflow bundle
metadata version"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_accepts_newer_minor_version_with_unknown_optional_fields(self,
tmp_path):
payload = json.loads(_metadata_json("sales", metadata_version="1.7.3"))
@@ -109,7 +110,7 @@ class TestBundleReader:
_replace_layout_payload(bundle, payload)
with pytest.raises(ValueError, match=message):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
@pytest.mark.parametrize("encoding", ["utf-16", "utf-32"])
def test_rejects_layout_that_is_not_utf8(self, tmp_path, encoding):
@@ -126,7 +127,7 @@ class TestBundleReader:
_replace_layout_payload(bundle, b"A" * _reader._MAX_LAYOUT_LINE_BYTES)
with pytest.raises(ValueError, match="bundle layout exceeds"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_unterminated_layout(self, tmp_path):
bundle = write_bundle(tmp_path, "sales")
@@ -134,7 +135,7 @@ class TestBundleReader:
bundle.write_bytes(layout_line)
with pytest.raises(ValueError, match="bundle layout is not
newline-terminated"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_requires_metadata_immediately_after_layout(self, tmp_path):
bundle = write_bundle(tmp_path, "sales")
@@ -142,7 +143,7 @@ class TestBundleReader:
bundle.write_bytes(contents)
with pytest.raises(ValueError, match="no embedded airflow metadata
after its layout"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_unterminated_metadata(self, tmp_path):
bundle = write_bundle(tmp_path, "sales")
@@ -150,7 +151,7 @@ class TestBundleReader:
bundle.write_bytes(layout_line + b"\n" + metadata_line)
with pytest.raises(ValueError, match="embedded airflow metadata is not
newline-terminated"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
@pytest.mark.parametrize(
"metadata_payload",
@@ -175,7 +176,7 @@ class TestBundleReader:
_replace_layout_payload(bundle, json.dumps(layout).encode())
with pytest.raises(ValueError, match=f"missing the {section} section"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_invalid_section_digest(self, tmp_path):
bundle = write_bundle(tmp_path, "sales")
@@ -184,7 +185,7 @@ class TestBundleReader:
_rewrite_layout(bundle, layout)
with pytest.raises(ValueError, match="64 lowercase hexadecimal
digits"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_malformed_offset(self, tmp_path):
bundle = write_bundle(tmp_path, "sales")
@@ -193,7 +194,7 @@ class TestBundleReader:
_rewrite_layout(bundle, layout)
with pytest.raises(ValueError, match="16-digit lowercase hexadecimal"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
@pytest.mark.parametrize(("start", "end"), [(1, 1), (2, 1)])
def test_rejects_empty_or_reversed_section(self, tmp_path, start, end):
@@ -204,7 +205,7 @@ class TestBundleReader:
_rewrite_layout(bundle, layout)
with pytest.raises(ValueError, match="code section must contain at
least one byte"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_metadata_offset_mismatch(self, tmp_path):
bundle = write_bundle(tmp_path, "sales")
@@ -214,7 +215,7 @@ class TestBundleReader:
_rewrite_layout(bundle, layout)
with pytest.raises(ValueError, match="metadata offsets do not match"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
@pytest.mark.parametrize("section", [pytest.param("metadata",
id="metadata-before-decode"), "code"])
def test_rejects_section_digest_mismatch(self, tmp_path, section):
@@ -224,24 +225,24 @@ class TestBundleReader:
_mutate_byte(bundle, int(layout[section]["start"], 16)) # type:
ignore[index, call-overload]
with pytest.raises(ValueError, match=f"{section} SHA-256 mismatch"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_truncated_code(self, tmp_path):
bundle = write_bundle(tmp_path, "sales")
bundle.write_bytes(bundle.read_bytes()[:-1])
with pytest.raises(ValueError, match="code offsets do not match"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_reports_truncation_while_hashing(self, tmp_path):
with pytest.raises(ValueError, match="was truncated while hashing its
code region"):
- _hash_region(io.BytesIO(), start=0, end=1, path=tmp_path /
"bundle.mjs", section="code")
+ _hash_region(io.BytesIO(), start=0, end=1, path=tmp_path /
BUNDLE_NAME, section="code")
def test_rejects_invalid_metadata_json_after_verification(self, tmp_path):
write_bundle(tmp_path, "sales", metadata_payload=b"not-json")
with pytest.raises(ValueError, match="cannot parse embedded airflow
metadata"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
@pytest.mark.parametrize(
"metadata_payload",
@@ -255,13 +256,13 @@ class TestBundleReader:
write_bundle(tmp_path, "sales", metadata_payload=metadata_payload)
with pytest.raises(ValueError, match="cannot parse embedded airflow
metadata"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_metadata_that_is_not_a_mapping(self, tmp_path):
write_bundle(tmp_path, "sales", metadata_payload=b"[]")
with pytest.raises(ValueError, match="embedded airflow metadata must
contain a mapping"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
@pytest.mark.parametrize("dags", [None, []], ids=["missing",
"not-a-mapping"])
def test_rejects_missing_or_malformed_dags(self, tmp_path, dags):
@@ -273,7 +274,7 @@ class TestBundleReader:
write_bundle(tmp_path, metadata_payload=json.dumps(metadata).encode())
with pytest.raises(ValueError, match="metadata must contain a dags
mapping"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_oversized_metadata(self, tmp_path):
write_bundle(
@@ -283,7 +284,7 @@ class TestBundleReader:
)
with pytest.raises(ValueError, match="embedded airflow metadata
exceeds"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_change_during_verification(self, tmp_path, monkeypatch):
bundle = write_bundle(tmp_path, "sales")
@@ -299,7 +300,7 @@ class TestBundleReader:
monkeypatch.setattr(_reader, "_hash_region", hash_then_touch)
with pytest.raises(ValueError, match="changed while its integrity was
being verified"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_rejects_change_after_verification_before_metadata_decode(self,
tmp_path, monkeypatch):
bundle = write_bundle(tmp_path, "sales")
@@ -313,14 +314,14 @@ class TestBundleReader:
monkeypatch.setattr(_digest_cache, "put", put_then_touch)
with pytest.raises(ValueError, match="changed while it was being
read"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
@pytest.mark.parametrize(
("failure_call", "message"),
[
- (1, "cannot read bundle.mjs"),
- (2, "cannot stat bundle.mjs after verification"),
- (3, "cannot stat bundle.mjs after reading"),
+ (1, "cannot read bundle.min.mjs"),
+ (2, "cannot stat bundle.min.mjs after verification"),
+ (3, "cannot stat bundle.min.mjs after reading"),
],
)
def test_translates_fstat_errors(self, tmp_path, monkeypatch,
failure_call, message):
@@ -338,7 +339,7 @@ class TestBundleReader:
monkeypatch.setattr(_reader.os, "fstat", fail_selected_fstat)
with pytest.raises(OSError, match=message):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_translates_metadata_read_error(self, tmp_path, monkeypatch):
bundle = write_bundle(tmp_path, "sales")
@@ -351,16 +352,16 @@ class TestBundleReader:
monkeypatch.setattr(pathlib.Path, "open", path_open)
monkeypatch.setattr(_reader.os, "fstat", lambda _: bundle.stat())
- with pytest.raises(OSError, match="cannot read bundle.mjs"):
- read_bundle(tmp_path / "bundle.mjs")
+ with pytest.raises(OSError, match="cannot read bundle.min.mjs"):
+ read_bundle(tmp_path / BUNDLE_NAME)
@mock.patch.object(_reader, "_hash_region", autospec=True)
def test_reuses_cached_digests_for_unchanged_bundle(self, hash_region,
tmp_path):
write_bundle(tmp_path, "sales")
hash_region.side_effect = _hash_region
- read_bundle(tmp_path / "bundle.mjs")
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
+ read_bundle(tmp_path / BUNDLE_NAME)
assert hash_region.call_count == 2
@@ -383,7 +384,7 @@ class TestBundleReader:
fstat.return_value.st_ctime_ns += 1
with pytest.raises(ValueError, match="code SHA-256 mismatch"):
- read_bundle(tmp_path / "bundle.mjs")
+ read_bundle(tmp_path / BUNDLE_NAME)
def test_digest_cache_evicts_least_recently_used_entry(self):
cache = _reader._BundleDigestCache(maxsize=2)
@@ -392,7 +393,7 @@ class TestBundleReader:
def build_key(inode):
return _reader._DigestCacheKey(
- path="bundle.mjs",
+ path="bundle.min.mjs",
metadata=section,
code=section,
device=1,
diff --git a/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py
b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py
index 304de1832b4..492c410c8aa 100644
--- a/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py
+++ b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py
@@ -24,6 +24,7 @@ from unittest import mock
import pytest
from task_sdk.coordinators.node._bundle_test_utils import (
+ BUNDLE_NAME,
mutate_byte,
read_layout,
write_bundle,
@@ -98,24 +99,33 @@ class TestNodeCoordinatorExecuteTaskCommand:
class TestBundleFind:
- def test_ignores_roots_without_bundle_mjs(self, tmp_path):
- (tmp_path / "tasks.mjs").write_bytes(b"export {};\n")
+ @pytest.mark.parametrize(
+ "name",
+ ["tasks.mjs", "tasks.js", "tasks.min.js", "bundle.min.mjs.bak",
"min.mjs.txt"],
+ ids=["mjs", "js", "min-js", "suffixed", "embedded"],
+ )
+ def test_ignores_files_without_the_bundle_suffix(self, tmp_path, name):
+ # Written as a real bundle, so only the name can exclude it.
+ write_bundle(tmp_path, "sales", name=name)
- with pytest.raises(FileNotFoundError, match="dag_id='sales'"):
+ with pytest.raises(FileNotFoundError, match="dag_id='sales'") as
exc_info:
_Bundle.find([tmp_path], "sales")
+ # Never opened, so it cannot appear among the rejected candidates.
+ assert "rejected candidates" not in str(exc_info.value)
+
def test_reports_unreadable_bundle(self, tmp_path, monkeypatch):
write_bundle(tmp_path, "sales")
original_open = pathlib.Path.open
def raise_os_error(self, *args, **kwargs):
- if self.name == "bundle.mjs":
+ if self.name == BUNDLE_NAME:
raise PermissionError("denied")
return original_open(self, *args, **kwargs)
monkeypatch.setattr(pathlib.Path, "open", raise_os_error)
- with pytest.raises(FileNotFoundError, match="cannot read bundle.mjs"):
+ with pytest.raises(FileNotFoundError, match="cannot read
bundle.min.mjs"):
_Bundle.find([tmp_path], "sales")
def test_skips_root_when_bundle_probe_fails(self, tmp_path, monkeypatch):
@@ -123,21 +133,69 @@ class TestBundleFind:
second = tmp_path / "second"
first.mkdir()
second.mkdir()
- write_bundle(first, "sales")
+ unstattable = write_bundle(first, "sales")
expected = write_bundle(second, "sales")
- original_is_file = pathlib.Path.is_file
+ original_stat = pathlib.Path.stat
- def fail_first_probe(self):
- if self.parent == first:
+ def fail_first_probe(self, *args, **kwargs):
+ if self == unstattable:
raise PermissionError("denied")
- return original_is_file(self)
+ return original_stat(self, *args, **kwargs)
- monkeypatch.setattr(pathlib.Path, "is_file", fail_first_probe)
+ monkeypatch.setattr(pathlib.Path, "stat", fail_first_probe)
found = _Bundle.find([first, second], "sales")
assert found.path == expected
+ def test_finds_bundle_nested_below_a_root(self, tmp_path):
+ expected = write_bundle(tmp_path / "team" / "sales", "sales")
+
+ found = _Bundle.find([tmp_path], "sales")
+
+ assert found.path == expected
+
+ def test_selects_bundle_by_dag_id_within_one_root(self, tmp_path):
+ write_bundle(tmp_path, "inventory", name="inventory.min.mjs")
+ expected = write_bundle(tmp_path, "sales", name="sales.min.mjs")
+
+ found = _Bundle.find([tmp_path], "sales")
+
+ assert found.path == expected
+
+ def test_orders_candidates_in_one_root_by_path(self, tmp_path):
+ # Directory iteration order is filesystem-dependent, so sorted name
decides the winner.
+ expected = write_bundle(tmp_path, "sales", name="a.min.mjs")
+ write_bundle(tmp_path, "sales", name="b.min.mjs")
+ write_bundle(tmp_path / "nested", "sales")
+
+ found = _Bundle.find([tmp_path], "sales")
+
+ assert found.path == expected
+
+ def test_survives_a_directory_symlink_loop(self, tmp_path):
+ expected = write_bundle(tmp_path, "sales")
+ loop = tmp_path / "loop"
+ try:
+ loop.symlink_to(tmp_path, target_is_directory=True)
+ except (OSError, NotImplementedError):
+ pytest.skip("filesystem does not support directory symlinks")
+
+ found = _Bundle.find([tmp_path], "sales")
+
+ assert found.path == expected
+
+ def test_names_unrelated_min_mjs_file_among_rejected_candidates(self,
tmp_path):
+ stray = tmp_path / "vendor.min.mjs"
+ stray.write_bytes(b"export {};\n")
+
+ with pytest.raises(FileNotFoundError) as exc_info:
+ _Bundle.find([tmp_path], "sales")
+
+ message = str(exc_info.value)
+ assert str(stray) in message
+ assert "no airflow bundle layout" in message
+
def test_selects_later_bundle_containing_requested_dag(self, tmp_path):
first = tmp_path / "first"
second = tmp_path / "second"
@@ -236,7 +294,7 @@ class TestBundleFind:
second = tmp_path / "second"
first.mkdir()
second.mkdir()
- (first / "bundle.mjs").write_bytes(b"export {};\n")
+ (first / BUNDLE_NAME).write_bytes(b"export {};\n")
write_bundle(second, "inventory")
with pytest.raises(FileNotFoundError) as exc_info:
diff --git a/ts-sdk/README.md b/ts-sdk/README.md
index 6e4c44079da..a4af960e622 100644
--- a/ts-sdk/README.md
+++ b/ts-sdk/README.md
@@ -98,9 +98,8 @@ coordinators = {
queue_to_coordinator = {"typescript": "ts"}
```
-Each configured bundle directory must contain a `bundle.mjs` built with
-`airflow-ts-pack` (see [Packing bundles](#packing-bundles)), which embeds the
-Airflow metadata in the bundle itself.
+Each configured bundle directory is searched recursively for `*.min.mjs`
bundles built with `airflow-ts-pack`
+(see [Packing bundles](#packing-bundles)), which embeds the Airflow metadata
in the bundle itself.
TypeScript entrypoint:
@@ -185,20 +184,20 @@ npm install --save-dev esbuild
airflow-ts-pack src/main.ts --outdir dist
```
-It bundles the entrypoint into `dist/bundle.mjs` with esbuild, runs the
-bundle with `--airflow-metadata` so the bundle reports its own registered
-Dag/task pairs and supervisor schema version, and embeds that manifest in the
-bundle as a compact JSON `//# airflowMetadata=...` comment after a leading
-compact JSON `//# airflowBundle=...` layout descriptor. The descriptor records
-fixed-width byte ranges and SHA-256 digests for the metadata and bundled code,
-allowing a coordinator reader to detect corruption before using either region.
-These in-bundle digests do not authenticate who produced the bundle because
-someone who can replace the content can also replace its digests. The result is
-one deployable file with no hand-written metadata sidecar.
+It bundles the entrypoint into a minified `dist/bundle.min.mjs` with esbuild,
then runs that bundle with
+`--airflow-metadata` so it reports its own registered Dag/task pairs and
supervisor schema version. The manifest is
+embedded as a compact JSON `//# airflowMetadata=...` comment after a leading
compact JSON `//# airflowBundle=...`
+layout descriptor. The CLI records the integrity metadata for both regions in
that descriptor, so a coordinator that
+is handed a bundle whose content was replaced fails loudly instead of running
it. The result is one deployable file
+with no hand-written metadata sidecar.
+
+Pass `--outfile <path>` instead of `--outdir` to name the artifact yourself,
so one bundle directory can hold several
+bundles. The name must still end in `.min.mjs`, which is how `NodeCoordinator`
finds bundles.
Options:
- `--outdir <dir>`: output directory (default `dist`)
+- `--outfile <path>`: exact output path, whose name must end in `.min.mjs`
- `--source <name>`: display name of the primary source file shown in the
Airflow UI (default: entry basename)
## TaskClient
diff --git a/ts-sdk/example/README.md b/ts-sdk/example/README.md
index e9a1e43967b..4af4af57471 100644
--- a/ts-sdk/example/README.md
+++ b/ts-sdk/example/README.md
@@ -25,7 +25,7 @@ This example shows the coordinator-mode shape for TypeScript
task handlers:
- `src/main.ts` and `src/taskflow.ts` register a `TaskHandler` per stub task
and start the coordinator runtime.
One bundle provides for both Dags, and both declare a task called
`build_message`.
A handler binds the `(dag_id, task_id)` pair, so the two are different tasks
with different bodies.
-- `dist/bundle.mjs` is the generated Node.js bundle that Airflow launches.
+- `dist/bundle.min.mjs` is the generated Node.js bundle that Airflow launches.
The build uses the SDK's `airflow-ts-pack` tool, which bundles the entrypoint
with esbuild and embeds the Airflow metadata generated from the bundle's
@@ -53,7 +53,7 @@ The coordinator expects this layout:
```text
ts-sdk/example/dist/
- bundle.mjs
+ bundle.min.mjs
```
## Airflow Configuration
diff --git a/ts-sdk/src/cli/pack.ts b/ts-sdk/src/cli/pack.ts
index 08a3aff008b..0403d8047bd 100644
--- a/ts-sdk/src/cli/pack.ts
+++ b/ts-sdk/src/cli/pack.ts
@@ -17,13 +17,11 @@
* under the License.
*/
-// airflow-ts-pack: bundle a TypeScript entrypoint into the single-file
-// artifact NodeCoordinator consumes: `bundle.mjs` with metadata and an
-// integrity layout descriptor embedded in JavaScript comments.
+// airflow-ts-pack: bundle a TypeScript entrypoint into the single artifact
NodeCoordinator consumes.
+// `bundle.min.mjs` carries the metadata and an integrity layout descriptor in
JavaScript comments.
//
-// Build first, then run the built bundle with --airflow-metadata so the
-// manifest comes from the bundle's own Dag registry and schema version,
-// never from a hand-written sidecar.
+// Build first, then run the built bundle with --airflow-metadata so the
manifest comes from the
+// bundle's own Dag registry and schema version, never from a hand-written
sidecar.
import { execFileSync } from "node:child_process";
import { readFileSync, rmSync, writeFileSync } from "node:fs";
@@ -37,26 +35,31 @@ import {
import { encodeBundle } from "./bundle-encoder.js";
import { warnOnSuspiciousIds } from "./validate.js";
-const BUNDLE_FILENAME = "bundle.mjs";
-// Write bundle.mjs only after the build and manifest checks succeed, so a
-// failed pack cannot leave a partial final artifact.
+// NodeCoordinator discovers bundles by this suffix, so keep the two in step.
+const BUNDLE_FILENAME = "bundle.min.mjs";
+// Write the bundle only after the build and manifest checks succeed, so a
failed pack
+// cannot leave a partial artifact.
const STAGING_FILENAME = "bundle.pack-staging.mjs";
const MANIFEST_TIMEOUT_MS = 60_000;
const MANIFEST_MAX_BUFFER_BYTES = 64 * 1024 * 1024;
-const USAGE = `Usage: airflow-ts-pack <entry> [--outdir <dir>] [--source
<name>]
+const USAGE = `Usage: airflow-ts-pack <entry> [--outdir <dir> | --outfile
<path>] [--source <name>]
-Bundles <entry> into <outdir>/${BUNDLE_FILENAME} with esbuild and embeds the
+Bundles <entry> into a minified ${BUNDLE_FILENAME} with esbuild and embeds the
airflow metadata generated from the bundle's served Dags.
Options:
- --outdir <dir> Output directory (default: dist)
- --source <name> Display name of the primary source file (default: <entry>
basename)
+ --outdir <dir> Output directory, holding ${BUNDLE_FILENAME} (default:
dist)
+ --outfile <path> Exact output path; its name must end in .min.mjs
+ --source <name> Display name of the primary source file (default: <entry>
basename)
`;
+/** A bundle written without this suffix is invisible to NodeCoordinator. */
+const REQUIRED_OUTFILE_SUFFIX = ".min.mjs";
+
export interface PackArgs {
entry: string;
- outdir: string;
+ outfile: string;
source: string;
}
@@ -66,14 +69,16 @@ function usageError(message: string): Error {
export function parsePackArgs(argv: readonly string[]): PackArgs {
let entry: string | null = null;
- let outdir = "dist";
+ let outdir: string | null = null;
+ let outfile: string | null = null;
let source: string | null = null;
for (let i = 0; i < argv.length; i += 1) {
const arg = argv[i]!;
- if (arg === "--outdir" || arg === "--source") {
+ if (arg === "--outdir" || arg === "--outfile" || arg === "--source") {
const value = argv[i + 1];
if (!value) throw usageError(`${arg} requires a value`);
if (arg === "--outdir") outdir = value;
+ else if (arg === "--outfile") outfile = value;
else source = value;
i += 1;
} else if (arg.startsWith("-")) {
@@ -85,7 +90,20 @@ export function parsePackArgs(argv: readonly string[]):
PackArgs {
}
}
if (!entry) throw usageError("Missing entry file");
- return { entry, outdir, source: source ?? path.basename(entry) };
+ // Silently preferring one would write the bundle somewhere the caller did
not ask for.
+ if (outdir !== null && outfile !== null) {
+ throw usageError("--outdir and --outfile are mutually exclusive");
+ }
+ if (outfile !== null &&
!path.basename(outfile).endsWith(REQUIRED_OUTFILE_SUFFIX)) {
+ throw usageError(
+ `--outfile name must end in ${REQUIRED_OUTFILE_SUFFIX}; NodeCoordinator
finds bundles by that suffix`,
+ );
+ }
+ return {
+ entry,
+ outfile: outfile ?? path.join(outdir ?? "dist", BUNDLE_FILENAME),
+ source: source ?? path.basename(entry),
+ };
}
function readSdkVersion(): string {
@@ -183,8 +201,8 @@ async function loadEsbuild(): Promise<typeof
import("esbuild")> {
export async function runPack(argv: readonly string[]): Promise<void> {
const args = parsePackArgs(argv);
- const bundlePath = path.join(args.outdir, BUNDLE_FILENAME);
- const stagingPath = path.join(args.outdir, STAGING_FILENAME);
+ const bundlePath = args.outfile;
+ const stagingPath = path.join(path.dirname(bundlePath), STAGING_FILENAME);
const { build } = await loadEsbuild();
try {
@@ -194,6 +212,9 @@ export async function runPack(argv: readonly string[]):
Promise<void> {
platform: "node",
format: "esm",
target: "node22",
+ // A digest is only worth taking over an artifact nobody reads or edits
in place.
+ minify: true,
+ // The manifest is read by running the staged bundle, so the metadata
describes what ships.
outfile: stagingPath,
});
diff --git a/ts-sdk/tests/cli/pack.test.ts b/ts-sdk/tests/cli/pack.test.ts
index 34d4b3a3d86..fb45586a254 100644
--- a/ts-sdk/tests/cli/pack.test.ts
+++ b/ts-sdk/tests/cli/pack.test.ts
@@ -58,7 +58,7 @@ describe("parsePackArgs", () => {
it("parses entry with defaults", () => {
expect(parsePackArgs(["src/main.ts"])).toEqual({
entry: "src/main.ts",
- outdir: "dist",
+ outfile: path.join("dist", "bundle.min.mjs"),
source: "main.ts",
});
});
@@ -66,16 +66,33 @@ describe("parsePackArgs", () => {
it("parses --outdir and --source overrides", () => {
expect(parsePackArgs(["src/main.ts", "--outdir", "build", "--source",
"pipeline.ts"])).toEqual({
entry: "src/main.ts",
- outdir: "build",
+ outfile: path.join("build", "bundle.min.mjs"),
source: "pipeline.ts",
});
});
+ it("parses --outfile as the exact output path", () => {
+ expect(parsePackArgs(["src/main.ts", "--outfile",
"out/sales.min.mjs"])).toEqual({
+ entry: "src/main.ts",
+ outfile: "out/sales.min.mjs",
+ source: "main.ts",
+ });
+ });
+
it.each([
[[], "Missing entry file"],
[["--outdir"], "--outdir requires a value"],
+ [["--outfile"], "--outfile requires a value"],
[["a.ts", "b.ts"], "Unexpected argument b.ts"],
[["a.ts", "--bogus"], "Unknown option --bogus"],
+ [
+ ["a.ts", "--outdir", "dist", "--outfile", "dist/sales.min.mjs"],
+ "--outdir and --outfile are mutually exclusive",
+ ],
+ // A bundle written without the suffix would never be found.
+ [["a.ts", "--outfile", "dist/sales.mjs"], "--outfile name must end in
.min.mjs"],
+ [["a.ts", "--outfile", "dist/sales.min.js"], "--outfile name must end in
.min.mjs"],
+ [["a.ts", "--outfile", "dist/min.mjs.txt"], "--outfile name must end in
.min.mjs"],
])("rejects %j", (argv, message) => {
expect(() => parsePackArgs(argv)).toThrow(message);
});
@@ -193,7 +210,7 @@ describe("runPack", () => {
const nested = path.join(outdir, "dist");
await runPack([FIXTURE_ENTRY, "--outdir", nested]);
- const bundlePath = path.join(nested, "bundle.mjs");
+ const bundlePath = path.join(nested, "bundle.min.mjs");
expect(existsSync(path.join(nested, "airflow-metadata.yaml"))).toBe(false);
const [layoutLine, metadataLine] = readFileSync(bundlePath,
"utf-8").split("\n");
@@ -227,7 +244,7 @@ describe("runPack", () => {
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
await runPack([FIXTURE_ENTRY, "--outdir", outdir]);
- const bundle = readFileSync(path.join(outdir, "bundle.mjs"));
+ const bundle = readFileSync(path.join(outdir, "bundle.min.mjs"));
const firstNewline = bundle.indexOf("\n");
const layoutLine = bundle.subarray(0, firstNewline).toString("utf-8");
expect(layoutLine.startsWith(EMBEDDED_LAYOUT_PREFIX)).toBe(true);
@@ -249,11 +266,51 @@ describe("runPack", () => {
expect(bundle.toString("utf-8")).not.toContain("airflowSource");
});
+ it("minifies the code region and keeps it runnable", async () => {
+ outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
+ await runPack([FIXTURE_ENTRY, "--outdir", outdir]);
+
+ const bundlePath = path.join(outdir, "bundle.min.mjs");
+ const bundle = readFileSync(bundlePath);
+ const layout = parseHeader(bundle.subarray(0,
bundle.indexOf("\n")).toString("utf-8"));
+ const code = bundle
+ .subarray(Number.parseInt(layout.code.start, 16),
Number.parseInt(layout.code.end, 16))
+ .toString("utf-8");
+
+ // esbuild indents unminified output; minified output has no indented line.
+ expect(code).not.toContain("\n ");
+ // Minification must not strip the dependencies' license banners.
+ expect(code).toContain("/*!");
+ // A digest over minified bytes only means something if those bytes still
execute.
+ const dumped = execFileSync(process.execPath, [bundlePath,
"--airflow-metadata"], {
+ encoding: "utf-8",
+ });
+ expect(
+
JSON.parse(dumped.slice(AIRFLOW_METADATA_SENTINEL.length)).supervisor_schema_version,
+ ).toBe(SUPERVISOR_API_VERSION);
+ });
+
+ it("writes the bundle to an explicit --outfile", async () => {
+ outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
+ const target = path.join(outdir, "nested", "sales.min.mjs");
+
+ await runPack([FIXTURE_ENTRY, "--outfile", target]);
+
+ expect(existsSync(target)).toBe(true);
+ expect(existsSync(path.join(outdir, "nested",
"bundle.min.mjs"))).toBe(false);
+ // Staging is written beside the target and cleaned up there.
+ expect(existsSync(path.join(outdir, "nested",
"bundle.pack-staging.mjs"))).toBe(false);
+ expect(readFileSync(target).subarray(0,
EMBEDDED_LAYOUT_PREFIX.length).toString()).toBe(
+ EMBEDDED_LAYOUT_PREFIX,
+ );
+
expect(JSON.parse(readEmbeddedMetadata(target))).toHaveProperty("dags.fixture_dag");
+ });
+
it("keeps a shebang entry runnable and reads the manifest past import-time
logging", async () => {
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
await runPack([NOISY_ENTRY, "--outdir", outdir]);
- const bundlePath = path.join(outdir, "bundle.mjs");
+ const bundlePath = path.join(outdir, "bundle.min.mjs");
const bundle = readFileSync(bundlePath, "utf-8");
expect(bundle.startsWith(EMBEDDED_LAYOUT_PREFIX)).toBe(true);
expect(bundle).not.toContain("#!/usr/bin/env node");
@@ -282,7 +339,7 @@ describe("runPack", () => {
await expect(runPack([entry, "--outdir", outdir])).rejects.toThrow(
"over the 1048576 byte limit",
);
- expect(existsSync(path.join(outdir, "bundle.mjs"))).toBe(false);
+ expect(existsSync(path.join(outdir, "bundle.min.mjs"))).toBe(false);
expect(existsSync(path.join(outdir,
"bundle.pack-staging.mjs"))).toBe(false);
});
@@ -292,7 +349,7 @@ describe("runPack", () => {
await expect(runPack([EMPTY_ENTRY, "--outdir", outdir])).rejects.toThrow(
"served nothing; register Dags or task handlers",
);
- expect(existsSync(path.join(outdir, "bundle.mjs"))).toBe(false);
+ expect(existsSync(path.join(outdir, "bundle.min.mjs"))).toBe(false);
expect(existsSync(path.join(outdir,
"bundle.pack-staging.mjs"))).toBe(false);
});
@@ -328,7 +385,7 @@ describe("runPack", () => {
await runPack([entry, "--outdir", outdir]);
expect(stderr()).toContain(expected);
- expect(existsSync(path.join(outdir, "bundle.mjs"))).toBe(true);
+ expect(existsSync(path.join(outdir, "bundle.min.mjs"))).toBe(true);
});
it("reports the last error from a failed bundle", async () => {
@@ -343,7 +400,7 @@ describe("runPack", () => {
"message",
"Error: final failure",
);
- expect(existsSync(path.join(outdir, "bundle.mjs"))).toBe(false);
+ expect(existsSync(path.join(outdir, "bundle.min.mjs"))).toBe(false);
expect(existsSync(path.join(outdir,
"bundle.pack-staging.mjs"))).toBe(false);
});
@@ -374,7 +431,7 @@ describe("runPack", () => {
);
await expect(runPack([entry, "--outdir",
outdir])).rejects.toThrow(message);
- expect(existsSync(path.join(outdir, "bundle.mjs"))).toBe(false);
+ expect(existsSync(path.join(outdir, "bundle.min.mjs"))).toBe(false);
expect(existsSync(path.join(outdir,
"bundle.pack-staging.mjs"))).toBe(false);
});
@@ -395,7 +452,7 @@ describe("runPack", () => {
await runPack([entry, "--outdir", outdir]);
expect(stderr()).toContain('warning: dag "empty_dag" has no tasks\n');
- expect(JSON.parse(readEmbeddedMetadata(path.join(outdir,
"bundle.mjs")))).toHaveProperty(
+ expect(JSON.parse(readEmbeddedMetadata(path.join(outdir,
"bundle.min.mjs")))).toHaveProperty(
"dags.empty_dag.tasks",
[],
);
@@ -418,7 +475,7 @@ describe("runPack", () => {
await runPack([entry, "--outdir", outdir]);
- const metadata = JSON.parse(readEmbeddedMetadata(path.join(outdir,
"bundle.mjs")));
+ const metadata = JSON.parse(readEmbeddedMetadata(path.join(outdir,
"bundle.min.mjs")));
expect(metadata).toHaveProperty("dags.sales_dag");
expect(metadata).not.toHaveProperty("dags.billing_dag");
});
