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 b4d012936c5 Add integrity metadata to TypeScript bundles (#72046)
b4d012936c5 is described below
commit b4d012936c5abe5f6b9b68eff085f98f30018370
Author: Shivam Rastogi <[email protected]>
AuthorDate: Mon Sep 14 02:00:00 2026 -0700
Add integrity metadata to TypeScript bundles (#72046)
* Detect corrupt TypeScript bundles before execution
Embed checksummed metadata and executable byte ranges so the coordinator
can validate packed bundles without executing user code. Select the first
verified bundle containing the requested Dag, with useful diagnostics for
rejected candidates.
Keep parsing and integrity checks in a private reader, document the
single-version JSON contract, and cover malformed bundles, cache invalidation,
routing, and corruption rejection.
* Clarify TypeScript bundle reader flow
Make private names and purpose comments explain declared sections, computed
digests, file identity, and cache bounds. Inline two shallow validators so the
read path is easier to follow, without changing validation behavior or error
messages.
---
.../language-sdks/typescript.rst | 13 +-
task-sdk/docs/airflow-metadata.schema.json | 2 +-
task-sdk/docs/index.rst | 1 +
task-sdk/docs/ts-bundle-spec.rst | 178 +++++++++
.../sdk/coordinators/node/_bundle_reader.py | 376 +++++++++++++++++++
.../airflow/sdk/coordinators/node/coordinator.py | 138 +++----
.../coordinators/node/_bundle_test_utils.py | 124 +++++++
.../coordinators/node/test_bundle_reader.py | 412 +++++++++++++++++++++
.../task_sdk/coordinators/node/test_coordinator.py | 250 +++++++------
ts-sdk/README.md | 12 +-
ts-sdk/src/cli/bundle-encoder.ts | 174 +++++++++
ts-sdk/src/cli/pack.ts | 74 +---
ts-sdk/tests/cli/fixtures/bundle-v1.mjs | 24 ++
ts-sdk/tests/cli/pack.test.ts | 210 ++++++++---
14 files changed, 1658 insertions(+), 330 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 279092c5dab..38616832436 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -261,7 +261,8 @@ 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) as a leading ``//#
airflowMetadata=<base64>`` comment — one file to
+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 — one file to
deploy, with no separate manifest or ``node_modules``.
``esbuild`` is an optional peer dependency: packing is build-time only, so the
runtime install of
@@ -280,7 +281,8 @@ 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 usable bundle with ``node``.
+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.
.. _typescript-sdk/coordinator-config:
@@ -299,8 +301,8 @@ All ``kwargs`` in the ``coordinators`` config entry are
passed to the
- Description
* - ``bundles_root``
- *(required)*
- - One or more directories searched, in order, for a ``bundle.mjs`` with
embedded metadata. Accepts a
- string, a path, or a list of strings/paths.
+ - 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.
* - ``node_executable``
- ``"node"``
- Path to the ``node`` binary. Defaults to ``node`` on ``$PATH``.
@@ -316,8 +318,5 @@ Limitations
languages, so task names and dependencies are declared in Python with
:func:`@task.stub <airflow.sdk.task.stub>`.
* **Beta status.** The SDK API may change in incompatible ways between
releases.
-* **One bundle per coordinator.**
:class:`~airflow.sdk.coordinators.node.NodeCoordinator` launches the first
- usable bundle found in ``bundles_root``; it does not yet route different
Dags or tasks to different
- bundles. To serve multiple bundles, register multiple coordinators on
separate queues.
* **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.
diff --git a/task-sdk/docs/airflow-metadata.schema.json
b/task-sdk/docs/airflow-metadata.schema.json
index 56e11109e26..d4a5cd240e7 100644
--- a/task-sdk/docs/airflow-metadata.schema.json
+++ b/task-sdk/docs/airflow-metadata.schema.json
@@ -37,7 +37,7 @@
},
"source": {
"type": "string",
- "description": "Original filename of the primary DAG source file (e.g.
'example.go'). The file's bytes are embedded in the bundle's source region;
this field is a display name used by the Airflow UI.",
+ "description": "Original filename of the primary Dag source file (e.g.
'example.go'). Bundle formats with an embedded source region use this as its
display name; other formats treat it as the logical authoring name.",
"minLength": 1
},
"dags": {
diff --git a/task-sdk/docs/index.rst b/task-sdk/docs/index.rst
index 1bb23c0a962..9a125a316b5 100644
--- a/task-sdk/docs/index.rst
+++ b/task-sdk/docs/index.rst
@@ -179,3 +179,4 @@ For the full public API reference, see the :doc:`api` page.
api
concepts
executable-bundle-spec
+ ts-bundle-spec
diff --git a/task-sdk/docs/ts-bundle-spec.rst b/task-sdk/docs/ts-bundle-spec.rst
new file mode 100644
index 00000000000..9aa46545132
--- /dev/null
+++ b/task-sdk/docs/ts-bundle-spec.rst
@@ -0,0 +1,178 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+TypeScript Bundle Format
+========================
+
+This document specifies the ``bundle.mjs`` format produced by
``airflow-ts-pack`` and consumed by
:class:`~airflow.sdk.coordinators.node.NodeCoordinator`.
+
+Container
+---------
+
+The bundle remains an ECMAScript module that runs directly with ``node
bundle.mjs``. It has three regions:
+
+.. code-block:: text
+
+ //# airflowBundle=<compact JSON layout>\n
+ //# airflowMetadata=<compact JSON>\n
+ <bundled ECMAScript code>
+
+The layout comes first so readers can locate and verify the other regions. The
+current format has no embedded source region.
+
+Layout Header
+-------------
+
+The ``airflowBundle`` payload is a compact UTF-8 JSON object:
+
+.. code-block:: json
+
+ {
+ "code": {
+ "start": "0000000000000401",
+ "end": "0000000000001200",
+ "sha256":
"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
+ },
+ "metadata": {
+ "start": "0000000000000300",
+ "end": "0000000000000400",
+ "sha256":
"123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0"
+ }
+ }
+
+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.
+
+Metadata
+--------
+
+The ``airflowMetadata`` payload is compact UTF-8 JSON with this logical shape:
+
+.. code-block:: json
+
+ {
+ "airflow_bundle_metadata_version": "1.0",
+ "sdk": {
+ "language": "typescript",
+ "version": "0.1.0-beta1",
+ "supervisor_schema_version": "2026-06-16"
+ },
+ "source": "main.ts",
+ "dags": {
+ "example": {
+ "tasks": ["extract", "load"]
+ }
+ }
+ }
+
+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.
+
+Reader and Selection Algorithm
+------------------------------
+
+For each directory in ``bundles_root``, in configured order, the coordinator:
+
+1. Looks for ``bundle.mjs`` and opens it once.
+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.
+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``.
+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.
+
+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
+---------------------------------------
+
+The digests detect truncation, corruption, or modification when the stored
+digests remain unchanged. They do not authenticate the producer: someone able
+to replace the bundle can replace its header and recompute both digests.
+
+The format also makes no provenance claim about which TypeScript sources or
+build process produced the JavaScript. Authenticity requires a signature or a
+digest delivered through a separately trusted channel. Provenance requires a
+build attestation or reproducible-build verification.
+
+The coordinator launches Node using the verified path. Replacing that path
+between verification and process launch remains a time-of-check/time-of-use
+window. Deployments should use controlled write permissions and atomic artifact
+replacement. The digest cache is a performance optimization, not a trust
anchor.
+
+Versioning and Compatibility
+----------------------------
+
+The Node coordinator accepts TypeScript bundle contract versions with major
+version 1 and ignores unknown optional header or metadata fields added by later
+minor versions. It rejects a missing, malformed, or different major version.
+Any incompatible change to either metadata or the meaning, encoding, or order
+of physical regions requires a new major version and an explicit coordinator
+change.
+
+The current strict marker, range, adjacency, file-size, and digest checks make
+older readers fail closed when they encounter incompatible physical framing,
+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.
diff --git a/task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py
b/task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py
new file mode 100644
index 00000000000..a03576aeac7
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py
@@ -0,0 +1,376 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Read and verify TypeScript Dag bundles.
+
+File order: layout comment, metadata comment, then executable JavaScript.
+Read the headers, verify the section digests, and return coordinator metadata.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import pathlib
+import re
+from collections import OrderedDict
+from typing import TYPE_CHECKING, Any
+
+import attrs
+
+from airflow.sdk.coordinators._bundle_metadata import
extract_supervisor_schema_version
+
+if TYPE_CHECKING:
+ from typing import BinaryIO
+
+# Format prefixes and whole-line limits must agree with the TypeScript encoder.
+_LAYOUT_COMMENT_PREFIX = b"//# airflowBundle="
+_MAX_LAYOUT_LINE_BYTES = 4096
+_METADATA_COMMENT_PREFIX = b"//# airflowMetadata="
+_MAX_METADATA_LINE_BYTES = 1024 * 1024
+_SUPPORTED_BUNDLE_MAJOR_VERSION = 1
+
+# Bound hashing memory and process-local cache growth independently of the
format.
+_HASH_CHUNK_BYTES = 1024 * 1024
+_MAX_DIGEST_CACHE_ENTRIES = 256
+_LOWER_HEX_DIGITS = frozenset("0123456789abcdef")
+
+
[email protected](frozen=True)
+class _DeclaredSection:
+ """A declared byte range (end exclusive) and its expected SHA-256
digest."""
+
+ start: int
+ end: int
+ sha256: bytes
+
+
[email protected](frozen=True)
+class _BundleLayout:
+ """The metadata and executable sections described by the layout comment."""
+
+ metadata: _DeclaredSection
+ code: _DeclaredSection
+
+
[email protected](frozen=True)
+class _DigestCacheKey:
+ """File identity, timestamps, size, and declared layout for one cached
calculation."""
+
+ path: str
+ metadata: _DeclaredSection
+ code: _DeclaredSection
+ device: int
+ inode: int
+ mtime_ns: int
+ ctime_ns: int
+ size: int
+
+
[email protected](frozen=True)
+class _ComputedDigests:
+ """SHA-256 digests calculated from the actual section bytes."""
+
+ metadata: bytes
+ code: bytes
+
+
[email protected](frozen=True)
+class BundleMetadata:
+ """
+ Metadata fields needed by the coordinator, extracted after integrity
verification.
+
+ This is not the full metadata document. Supervisor schema compatibility is
+ checked when the coordinator selects the bundle.
+ """
+
+ dag_ids: frozenset[str]
+ supervisor_schema_version: str
+
+
+def read_bundle(bundle_path: pathlib.Path) -> BundleMetadata:
+ """Read and verify one exact TypeScript bundle file."""
+ try:
+ bundle_file = bundle_path.open("rb")
+ except OSError as exc:
+ raise OSError(f"cannot read {bundle_path.name}: {exc}") from exc
+
+ with bundle_file:
+ try:
+ # Save file identity, size, and timestamps to detect changes
during reading.
+ initial_file_info = os.fstat(bundle_file.fileno())
+ except OSError as exc:
+ raise OSError(f"cannot read {bundle_path.name}: {exc}") from exc
+
+ layout, metadata_payload = _read_bundle_headers(
+ bundle_file, path=bundle_path, file_size=initial_file_info.st_size
+ )
+ _verify_integrity(bundle_file, path=bundle_path, layout=layout,
initial_file_info=initial_file_info)
+
+ # Interpret metadata only after checking its serialized bytes against the
declared digest.
+ return _parse_bundle_metadata(metadata_payload)
+
+
+class _BundleDigestCache:
+ """Process-local LRU of computed digests, not file contents or
verification verdicts."""
+
+ def __init__(self, maxsize: int) -> None:
+ self._maxsize = maxsize
+ self._entries: OrderedDict[_DigestCacheKey, _ComputedDigests] =
OrderedDict()
+
+ def get(self, key: _DigestCacheKey) -> _ComputedDigests | None:
+ digests = self._entries.get(key)
+ if digests is not None:
+ self._entries.move_to_end(key)
+ return digests
+
+ def put(self, key: _DigestCacheKey, digests: _ComputedDigests) -> None:
+ self._entries[key] = digests
+ self._entries.move_to_end(key)
+ while len(self._entries) > self._maxsize:
+ self._entries.popitem(last=False)
+
+ def clear(self) -> None:
+ self._entries.clear()
+
+
+_digest_cache = _BundleDigestCache(maxsize=_MAX_DIGEST_CACHE_ENTRIES)
+
+
+def _parse_offset(section: dict[str, Any], field: str) -> int:
+ value = section.get(field)
+ if (
+ not isinstance(value, str)
+ or len(value) != 16
+ or any(character not in _LOWER_HEX_DIGITS for character in value)
+ ):
+ raise ValueError(f"bundle layout {field} offset must be a 16-digit
lowercase hexadecimal string")
+ return int(value, 16)
+
+
+def _parse_section(layout: dict[str, Any], name: str) -> _DeclaredSection:
+ section = layout.get(name)
+ if not isinstance(section, dict):
+ raise ValueError(f"bundle layout is missing the {name} section")
+ start = _parse_offset(section, "start")
+ end = _parse_offset(section, "end")
+ if start >= end:
+ raise ValueError(f"bundle layout {name} section must contain at least
one byte")
+ sha256 = section.get("sha256")
+ if (
+ not isinstance(sha256, str)
+ or len(sha256) != 64
+ or any(character not in _LOWER_HEX_DIGITS for character in sha256)
+ ):
+ raise ValueError(f"bundle layout {name}.sha256 must be 64 lowercase
hexadecimal digits")
+ return _DeclaredSection(start=start, end=end, sha256=bytes.fromhex(sha256))
+
+
+def _parse_layout(payload: bytes) -> _BundleLayout:
+ try:
+ layout = json.loads(payload.decode("utf-8"))
+ except (UnicodeDecodeError, json.JSONDecodeError, RecursionError) as exc:
+ raise ValueError(f"cannot parse embedded airflow bundle layout:
{exc}") from exc
+ if not isinstance(layout, dict):
+ raise ValueError("embedded airflow bundle layout must contain a
mapping")
+ return _BundleLayout(
+ metadata=_parse_section(layout, "metadata"),
+ code=_parse_section(layout, "code"),
+ )
+
+
+def _is_supported_bundle_version(value: Any) -> bool:
+ if not isinstance(value, str) or
re.fullmatch(r"[0-9]+\.[0-9]+(?:\.[0-9]+)?", value) is None:
+ return False
+ # Compare decimal text without converting an arbitrarily long number to
int.
+ major = value.partition(".")[0].lstrip("0") or "0"
+ return major == str(_SUPPORTED_BUNDLE_MAJOR_VERSION)
+
+
+def _hash_region(bundle_file: BinaryIO, *, start: int, end: int, path:
pathlib.Path, section: str) -> bytes:
+ # Hash incrementally so large executable sections do not need to fit in
memory.
+ bundle_file.seek(start)
+ hasher = hashlib.sha256()
+ remaining_bytes = end - start
+ while remaining_bytes:
+ chunk = bundle_file.read(min(_HASH_CHUNK_BYTES, remaining_bytes))
+ if not chunk:
+ raise ValueError(f"{path.name} was truncated while hashing its
{section} region")
+ hasher.update(chunk)
+ remaining_bytes -= len(chunk)
+ return hasher.digest()
+
+
+def _get_file_identity_fields(file_info: os.stat_result) -> tuple[int, int,
int, int, int]:
+ """Select file identity, timestamps, and size for before/after
comparisons."""
+ return (
+ file_info.st_dev,
+ file_info.st_ino,
+ file_info.st_mtime_ns,
+ file_info.st_ctime_ns,
+ file_info.st_size,
+ )
+
+
+def _read_prefixed_line(
+ bundle_file: BinaryIO,
+ *,
+ path: pathlib.Path,
+ marker: bytes,
+ max_bytes: int,
+ section: str,
+ missing_error: str,
+) -> bytes:
+ """Read one bounded, newline-terminated bundle line and return its
payload."""
+ try:
+ line = bundle_file.readline(max_bytes + 1)
+ except OSError as exc:
+ raise OSError(f"cannot read {path.name}: {exc}") from exc
+ if not line.startswith(marker):
+ raise ValueError(missing_error)
+ if len(line) > max_bytes:
+ raise ValueError(f"embedded airflow {section} exceeds {max_bytes}
bytes")
+ if not line.endswith(b"\n"):
+ raise ValueError(f"embedded airflow {section} is not
newline-terminated")
+ payload = line[len(marker) : -1]
+ # These characters would end the JavaScript comment even inside JSON
strings.
+ if b"\r" in payload or b"\xe2\x80\xa8" in payload or b"\xe2\x80\xa9" in
payload:
+ raise ValueError(f"embedded airflow {section} contains a JavaScript
line terminator")
+ return payload
+
+
+def _compute_stable_digests(
+ bundle_file: BinaryIO,
+ *,
+ path: pathlib.Path,
+ layout: _BundleLayout,
+ initial_file_info: os.stat_result,
+) -> _ComputedDigests:
+ digests = _ComputedDigests(
+ metadata=_hash_region(
+ bundle_file, start=layout.metadata.start, end=layout.metadata.end,
path=path, section="metadata"
+ ),
+ code=_hash_region(
+ bundle_file, start=layout.code.start, end=layout.code.end,
path=path, section="code"
+ ),
+ )
+ try:
+ post_hash_file_info = os.fstat(bundle_file.fileno())
+ except OSError as exc:
+ raise OSError(f"cannot stat {path.name} after verification: {exc}")
from exc
+ # Reject a detected file change before caching the calculated digests.
+ if _get_file_identity_fields(post_hash_file_info) !=
_get_file_identity_fields(initial_file_info):
+ raise ValueError(f"{path.name} changed while its integrity was being
verified")
+ return digests
+
+
+def _verify_integrity(
+ bundle_file: BinaryIO,
+ *,
+ path: pathlib.Path,
+ layout: _BundleLayout,
+ initial_file_info: os.stat_result,
+) -> None:
+ cache_key = _DigestCacheKey(
+ path=os.fspath(path),
+ metadata=layout.metadata,
+ code=layout.code,
+ device=initial_file_info.st_dev,
+ inode=initial_file_info.st_ino,
+ mtime_ns=initial_file_info.st_mtime_ns,
+ ctime_ns=initial_file_info.st_ctime_ns,
+ size=initial_file_info.st_size,
+ )
+ # Reuse calculated hashes only when the file information and declared
layout match.
+ computed_digests = _digest_cache.get(cache_key)
+ if computed_digests is None:
+ computed_digests = _compute_stable_digests(
+ bundle_file, path=path, layout=layout,
initial_file_info=initial_file_info
+ )
+ _digest_cache.put(cache_key, computed_digests)
+
+ for section, computed_digest, declared_digest in (
+ ("metadata", computed_digests.metadata, layout.metadata.sha256),
+ ("code", computed_digests.code, layout.code.sha256),
+ ):
+ if computed_digest != declared_digest:
+ raise ValueError(f"{path.name} {section} SHA-256 mismatch")
+
+ # Check again on both cache-hit and cache-miss paths.
+ try:
+ final_file_info = os.fstat(bundle_file.fileno())
+ except OSError as exc:
+ raise OSError(f"cannot stat {path.name} after reading it: {exc}") from
exc
+ if _get_file_identity_fields(final_file_info) !=
_get_file_identity_fields(initial_file_info):
+ raise ValueError(f"{path.name} changed while it was being read")
+
+
+def _parse_bundle_metadata(payload: bytes) -> BundleMetadata:
+ """Validate the metadata document and extract the fields needed by the
coordinator."""
+ try:
+ metadata = json.loads(payload.decode("utf-8"))
+ except (UnicodeDecodeError, json.JSONDecodeError, RecursionError) as exc:
+ raise ValueError(f"cannot parse embedded airflow metadata: {exc}")
from exc
+ if not isinstance(metadata, dict):
+ raise ValueError("embedded airflow metadata must contain a mapping")
+ value = metadata.get("airflow_bundle_metadata_version")
+ if not _is_supported_bundle_version(value):
+ raise ValueError(
+ f"unsupported airflow bundle metadata version {value!r}; "
+ f"this runtime supports major version
{_SUPPORTED_BUNDLE_MAJOR_VERSION}"
+ )
+ dags = metadata.get("dags")
+ if not isinstance(dags, dict):
+ raise ValueError("embedded airflow metadata must contain a dags
mapping")
+ return BundleMetadata(
+ dag_ids=frozenset(dags),
+ supervisor_schema_version=extract_supervisor_schema_version(metadata),
+ )
+
+
+def _read_bundle_headers(
+ bundle_file: BinaryIO, *, path: pathlib.Path, file_size: int
+) -> tuple[_BundleLayout, bytes]:
+ """Read both comment payloads and check their declared ranges against the
file."""
+ layout_payload = _read_prefixed_line(
+ bundle_file,
+ path=path,
+ marker=_LAYOUT_COMMENT_PREFIX,
+ max_bytes=_MAX_LAYOUT_LINE_BYTES,
+ section="bundle layout",
+ missing_error=f"{path.name} has no airflow bundle layout; rebuild with
airflow-ts-pack",
+ )
+ layout = _parse_layout(layout_payload)
+ metadata_payload = _read_prefixed_line(
+ bundle_file,
+ path=path,
+ marker=_METADATA_COMMENT_PREFIX,
+ max_bytes=_MAX_METADATA_LINE_BYTES,
+ section="metadata",
+ missing_error=f"{path.name} has no embedded airflow metadata after its
layout",
+ )
+ layout_line_size = len(_LAYOUT_COMMENT_PREFIX) + len(layout_payload) + 1
+ metadata_start = layout_line_size + len(_METADATA_COMMENT_PREFIX)
+ metadata_end = metadata_start + len(metadata_payload)
+ code_start = metadata_end + 1
+ if (layout.metadata.start, layout.metadata.end) != (metadata_start,
metadata_end):
+ raise ValueError("bundle layout metadata offsets do not match the
metadata section")
+ if (layout.code.start, layout.code.end) != (code_start, file_size):
+ raise ValueError("bundle layout code offsets do not match the
executable section")
+ return layout, metadata_payload
diff --git a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
index 38e5224336c..98de32d5761 100644
--- a/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
+++ b/task-sdk/src/airflow/sdk/coordinators/node/coordinator.py
@@ -19,106 +19,75 @@
from __future__ import annotations
-import base64
import os
import pathlib
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
import attrs
import structlog
-from airflow.sdk.coordinators._bundle_metadata import (
- ResolvedBundle,
- convert_roots,
- extract_supervisor_schema_version,
- parse_metadata_mapping,
-)
+from airflow.sdk.coordinators._bundle_metadata import ResolvedBundle,
convert_roots
from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.coordinators.node._bundle_reader import read_bundle
if TYPE_CHECKING:
from collections.abc import Sequence
from structlog.typing import FilteringBoundLogger
+ from typing_extensions import Self
from airflow.sdk.api.datamodels._generated import TaskInstance
log: FilteringBoundLogger =
structlog.get_logger(logger_name="coordinators.node")
BUNDLE_FILENAME = "bundle.mjs"
-EMBEDDED_METADATA_MARKER = b"//# airflowMetadata="
-EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024
-def _read_embedded_metadata(bundle_path: pathlib.Path) -> dict[str, Any]:
- """
- Read the manifest ``airflow-ts-pack`` embeds in the bundle itself.
-
- The packer prepends the metadata as a leading
- ``//# airflowMetadata=<base64>`` line comment, keeping bundle and metadata
- a single artifact. Raises ``ValueError`` when the bundle has no such
marker.
- """
- try:
- with bundle_path.open("rb") as bundle_file:
- line = bundle_file.readline(EMBEDDED_METADATA_MAX_BYTES + 1)
- except OSError as exc:
- raise ValueError(f"cannot read {bundle_path.name}: {exc}") from exc
-
- if not line.startswith(EMBEDDED_METADATA_MARKER):
- raise ValueError(f"{bundle_path.name} has no embedded airflow
metadata; rebuild with airflow-ts-pack")
- if len(line) > EMBEDDED_METADATA_MAX_BYTES:
- raise ValueError(
- f"embedded airflow metadata exceeds {EMBEDDED_METADATA_MAX_BYTES}
bytes; "
- f"rebuild {bundle_path.name} with airflow-ts-pack"
- )
-
- payload = line[len(EMBEDDED_METADATA_MARKER) :].strip()
- try:
- decoded = base64.b64decode(payload, validate=True)
- except ValueError as exc:
- raise ValueError(f"cannot parse embedded airflow metadata: {exc}")
from exc
- return parse_metadata_mapping(decoded, source="embedded airflow metadata")
-
-
-def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> ResolvedBundle:
- """
- Locate the ``.mjs`` entry point in *bundles_root*.
-
- Scans each configured directory for ``bundle.mjs`` and reads the bundle's
- supervisor schema version from the metadata embedded in the bundle.
-
- This is an ordered fallback search, not Dag/task-aware multi-bundle
- routing. The first bundle found wins. A future version can use the
- metadata's ``dags`` section together with ``TaskInstance.dag_id`` and
- ``TaskInstance.task_id`` to select the bundle that owns a specific task.
- """
- rejected: list[tuple[pathlib.Path, str]] = []
- for root in bundles_root:
- candidate = root / BUNDLE_FILENAME
- if not candidate.is_file():
- continue
- try:
- metadata = _read_embedded_metadata(candidate)
- log.debug("Selected TypeScript bundle", path=candidate, root=root)
- return ResolvedBundle(
- path=candidate,
- schema_version=extract_supervisor_schema_version(metadata),
[email protected]
+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*."""
+ rejected: list[tuple[pathlib.Path, str]] = []
+ for root in bundles_root:
+ candidate = root / BUNDLE_FILENAME
+ 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(
+ (candidate, f"verified bundle declares
dag_ids={sorted(metadata.dag_ids)!r}")
+ )
+ continue
+ bundle = cls(path=candidate,
schema_version=metadata.supervisor_schema_version)
+ except (OSError, TypeError, ValueError) as exc:
+ 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)
+ return bundle
+
+ searched = os.pathsep.join(os.fspath(root) for root in bundles_root)
+ if rejected:
+ details = "; ".join(f"{path}: {reason}" for path, reason in
rejected)
+ raise FileNotFoundError(
+ f"Cannot find usable TypeScript bundle containing
dag_id={dag_id!r} in {searched}: "
+ f"rejected candidates ({details})"
)
- except (TypeError, ValueError) as exc:
- log.debug(
- "TypeScript bundle metadata rejected; skipping",
- path=candidate,
- root=root,
- exc_info=True,
- )
- rejected.append((candidate.resolve(), str(exc)))
-
- searched = os.pathsep.join(os.fspath(p.resolve()) for p in bundles_root)
- if rejected:
- details = "; ".join(f"{path}: {reason}" for path, reason in rejected)
- raise FileNotFoundError(
- f"Cannot find usable TypeScript bundle in {searched}: matching
bundles were rejected ({details})"
- )
- raise FileNotFoundError(f"Cannot find {BUNDLE_FILENAME} in {searched}")
+ raise FileNotFoundError(f"Cannot find TypeScript bundle containing
dag_id={dag_id!r} in {searched}")
@attrs.define(kw_only=True)
@@ -141,11 +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 a usable
- TypeScript bundle. Each bundle directory must contain ``bundle.mjs``
- with embedded metadata (as produced by ``airflow-ts-pack``). This is a
- fallback search path; it does not yet route different Dag/task pairs
- to different bundles.
+ :param bundles_root: Ordered list of directories scanned for the first
+ verified ``bundle.mjs`` that declares 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.
"""
@@ -157,7 +123,5 @@ class NodeCoordinator(SubprocessCoordinator):
)
def _build_execute_task_command(self, *, what: TaskInstance) ->
tuple[list[str], str | None]:
- # Multi-bundle routing should be added here by passing `what.dag_id`
and
- # `what.task_id` into bundle selection and matching against
metadata["dags"].
- bundle = _find_bundle(self.bundles_root)
+ bundle = _Bundle.find(self.bundles_root, what.dag_id)
return [self.node_executable, os.fspath(bundle.path)],
bundle.schema_version
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
new file mode 100644
index 00000000000..303c95b77b7
--- /dev/null
+++ b/task-sdk/tests/task_sdk/coordinators/node/_bundle_test_utils.py
@@ -0,0 +1,124 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import hashlib
+import json
+import pathlib
+
+SCHEMA_VERSION = "2026-06-16"
+LAYOUT_PREFIX = b"//# airflowBundle="
+METADATA_PREFIX = b"//# airflowMetadata="
+OFFSET_WIDTH = 16
+
+
+def metadata_json(
+ *dag_ids: str,
+ schema_version: str = SCHEMA_VERSION,
+ metadata_version: str | None = "1.0",
+) -> bytes:
+ metadata = {
+ "sdk": {
+ "language": "typescript",
+ "version": "0.1.0",
+ "supervisor_schema_version": schema_version,
+ },
+ "source": "main.ts",
+ "dags": {dag_id: {"tasks": ["test_task"]} for dag_id in dag_ids},
+ }
+ if metadata_version is not None:
+ metadata = {"airflow_bundle_metadata_version": metadata_version,
**metadata}
+ return json.dumps(metadata, separators=(",", ":"),
ensure_ascii=False).encode()
+
+
+def _section(start: int, end: int, payload: bytes) -> dict[str, str]:
+ return {
+ "start": f"{start:0{OFFSET_WIDTH}x}",
+ "end": f"{end:0{OFFSET_WIDTH}x}",
+ "sha256": hashlib.sha256(payload).hexdigest(),
+ }
+
+
+def _layout_line(layout: dict[str, object]) -> bytes:
+ payload = json.dumps(layout, separators=(",", ":")).encode("ascii")
+ return LAYOUT_PREFIX + payload + b"\n"
+
+
+def write_bundle(
+ root: pathlib.Path,
+ *dag_ids: str,
+ code: bytes = b"export {};\n",
+ schema_version: str = SCHEMA_VERSION,
+ metadata_version: str | None = "1.0",
+ metadata_payload: bytes | None = None,
+) -> pathlib.Path:
+ if metadata_payload is None:
+ metadata_payload = metadata_json(
+ *dag_ids,
+ schema_version=schema_version,
+ metadata_version=metadata_version,
+ )
+ metadata_line = METADATA_PREFIX + metadata_payload + b"\n"
+ placeholder = _layout_line(
+ {
+ "code": _section(0, 0, code),
+ "metadata": _section(0, 0, metadata_payload),
+ }
+ )
+ metadata_start = len(placeholder) + len(METADATA_PREFIX)
+ metadata_end = metadata_start + len(metadata_payload)
+ code_start = len(placeholder) + len(metadata_line)
+ layout_line = _layout_line(
+ {
+ "code": _section(code_start, code_start + len(code), code),
+ "metadata": _section(metadata_start, metadata_end,
metadata_payload),
+ }
+ )
+ assert len(layout_line) == len(placeholder)
+
+ bundle = root / "bundle.mjs"
+ bundle.write_bytes(layout_line + metadata_line + code)
+ return bundle
+
+
+def read_layout(bundle: pathlib.Path) -> dict[str, object]:
+ line = bundle.read_bytes().splitlines(keepends=True)[0]
+ return json.loads(line[len(LAYOUT_PREFIX) :].strip())
+
+
+def rewrite_layout(bundle: pathlib.Path, layout: dict[str, object]) -> None:
+ contents = bundle.read_bytes()
+ _, separator, remainder = contents.partition(b"\n")
+ assert separator
+ replacement = _layout_line(layout)
+ assert len(replacement) == len(contents) - len(remainder)
+ bundle.write_bytes(replacement + remainder)
+
+
+def replace_layout_payload(bundle: pathlib.Path, payload: bytes) -> None:
+ contents = bundle.read_bytes()
+ _, separator, remainder = contents.partition(b"\n")
+ assert separator
+ bundle.write_bytes(LAYOUT_PREFIX + payload + b"\n" + remainder)
+
+
+def mutate_byte(bundle: pathlib.Path, offset: int) -> None:
+ contents = bytearray(bundle.read_bytes())
+ contents[offset] = ord("A") if contents[offset] != ord("A") else ord("B")
+ bundle.write_bytes(contents)
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
new file mode 100644
index 00000000000..1ff5808cf6f
--- /dev/null
+++ b/task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py
@@ -0,0 +1,412 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import pathlib
+from unittest import mock
+
+import pytest
+from task_sdk.coordinators.node._bundle_test_utils import (
+ LAYOUT_PREFIX,
+ METADATA_PREFIX,
+ OFFSET_WIDTH,
+ SCHEMA_VERSION,
+ metadata_json as _metadata_json,
+ mutate_byte as _mutate_byte,
+ read_layout as _read_layout,
+ replace_layout_payload as _replace_layout_payload,
+ rewrite_layout as _rewrite_layout,
+ write_bundle,
+)
+
+from airflow.sdk.coordinators.node import _bundle_reader as _reader
+from airflow.sdk.coordinators.node._bundle_reader import _digest_cache,
_hash_region, read_bundle
+
+from tests_common.test_utils.paths import AIRFLOW_ROOT_PATH
+
+TYPESCRIPT_V1_FIXTURE = AIRFLOW_ROOT_PATH / "ts-sdk" / "tests" / "cli" /
"fixtures" / "bundle-v1.mjs"
+
+
[email protected](autouse=True)
+def clear_digest_cache():
+ _digest_cache.clear()
+
+
+class TestBundleReader:
+ def test_reads_bundle_produced_by_typescript_encoder(self):
+ assert "version" not in _read_layout(TYPESCRIPT_V1_FIXTURE)
+ metadata = read_bundle(TYPESCRIPT_V1_FIXTURE)
+
+ assert metadata.dag_ids == frozenset({"test_dag"})
+ assert metadata.supervisor_schema_version == SCHEMA_VERSION
+
+ 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")
+
+ with pytest.raises(ValueError, match="no airflow bundle layout"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ @pytest.mark.parametrize(
+ "metadata_version",
+ [None, "banana", "١.٠", "2.0", pytest.param(f"{'9' * 5_000}.0",
id="unbounded-major")],
+ )
+ def test_rejects_unsupported_metadata_version(self, tmp_path,
metadata_version):
+ 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")
+
+ def test_accepts_newer_minor_version_with_unknown_optional_fields(self,
tmp_path):
+ payload = json.loads(_metadata_json("sales", metadata_version="1.7.3"))
+ payload["future_metadata_field"] = {"enabled": True}
+ bundle = write_bundle(tmp_path,
metadata_payload=json.dumps(payload).encode())
+ layout = _read_layout(bundle)
+ original_header_size = len(bundle.read_bytes().partition(b"\n")[0])
+ layout["future_header_field"] = True
+ layout["code"]["future_section_field"] = True # type: ignore[index]
+ # Fixed-width offsets let us relocate both sections after extending
the header.
+ header_growth = len(LAYOUT_PREFIX) + len(json.dumps(layout).encode())
- original_header_size
+ for name in ("metadata", "code"):
+ for field in ("start", "end"):
+ value = int(layout[name][field], 16) + header_growth # type:
ignore[index, call-overload]
+ layout[name][field] = f"{value:0{OFFSET_WIDTH}x}" # type:
ignore[index]
+ _replace_layout_payload(bundle, json.dumps(layout).encode())
+
+ metadata = read_bundle(bundle)
+
+ assert metadata.dag_ids == frozenset({"sales"})
+ assert metadata.supervisor_schema_version == SCHEMA_VERSION
+
+ @pytest.mark.parametrize(
+ ("payload", "message"),
+ [
+ (b"not-json", "cannot parse embedded airflow bundle layout"),
+ (b"[]", "bundle layout must contain a mapping"),
+ ],
+ )
+ def test_rejects_malformed_layout(self, tmp_path, payload, message):
+ bundle = write_bundle(tmp_path, "sales")
+ _replace_layout_payload(bundle, payload)
+
+ with pytest.raises(ValueError, match=message):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ @pytest.mark.parametrize("encoding", ["utf-16", "utf-32"])
+ def test_rejects_layout_that_is_not_utf8(self, tmp_path, encoding):
+ bundle = write_bundle(tmp_path, "sales")
+ layout = _read_layout(bundle)
+ payload = json.dumps(layout).encode(encoding)
+ _replace_layout_payload(bundle, payload)
+
+ with pytest.raises(ValueError, match="cannot parse embedded airflow
bundle layout"):
+ read_bundle(bundle)
+
+ def test_rejects_oversized_layout(self, tmp_path):
+ bundle = write_bundle(tmp_path, "sales")
+ _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")
+
+ def test_rejects_unterminated_layout(self, tmp_path):
+ bundle = write_bundle(tmp_path, "sales")
+ layout_line = bundle.read_bytes().partition(b"\n")[0]
+ bundle.write_bytes(layout_line)
+
+ with pytest.raises(ValueError, match="bundle layout is not
newline-terminated"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ def test_requires_metadata_immediately_after_layout(self, tmp_path):
+ bundle = write_bundle(tmp_path, "sales")
+ contents = bundle.read_bytes().replace(METADATA_PREFIX, b"//#
notMetadata=", 1)
+ bundle.write_bytes(contents)
+
+ with pytest.raises(ValueError, match="no embedded airflow metadata
after its layout"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ def test_rejects_unterminated_metadata(self, tmp_path):
+ bundle = write_bundle(tmp_path, "sales")
+ layout_line, metadata_line, _ = bundle.read_bytes().split(b"\n",
maxsplit=2)
+ 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")
+
+ @pytest.mark.parametrize(
+ "metadata_payload",
+ [
+ _metadata_json("sales").replace(b"{", b"{\r", 1),
+ _metadata_json("sales").replace(b"sales",
"sales\u2028dag".encode(), 1),
+ _metadata_json("sales").replace(b"sales",
"sales\u2029dag".encode(), 1),
+ ],
+ ids=["carriage-return", "line-separator", "paragraph-separator"],
+ )
+ def test_rejects_metadata_with_javascript_line_terminator(self, tmp_path,
metadata_payload):
+ bundle = write_bundle(tmp_path, metadata_payload=metadata_payload)
+
+ with pytest.raises(ValueError, match="metadata contains a JavaScript
line terminator"):
+ read_bundle(bundle)
+
+ @pytest.mark.parametrize("section", ["code", "metadata"])
+ def test_requires_every_layout_section(self, tmp_path, section):
+ bundle = write_bundle(tmp_path, "sales")
+ layout = _read_layout(bundle)
+ del layout[section]
+ _replace_layout_payload(bundle, json.dumps(layout).encode())
+
+ with pytest.raises(ValueError, match=f"missing the {section} section"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ def test_rejects_invalid_section_digest(self, tmp_path):
+ bundle = write_bundle(tmp_path, "sales")
+ layout = _read_layout(bundle)
+ layout["code"]["sha256"] = "z" * 64 # type: ignore[index]
+ _rewrite_layout(bundle, layout)
+
+ with pytest.raises(ValueError, match="64 lowercase hexadecimal
digits"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ def test_rejects_malformed_offset(self, tmp_path):
+ bundle = write_bundle(tmp_path, "sales")
+ layout = _read_layout(bundle)
+ layout["code"]["start"] = "Z" * OFFSET_WIDTH # type: ignore[index]
+ _rewrite_layout(bundle, layout)
+
+ with pytest.raises(ValueError, match="16-digit lowercase hexadecimal"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ @pytest.mark.parametrize(("start", "end"), [(1, 1), (2, 1)])
+ def test_rejects_empty_or_reversed_section(self, tmp_path, start, end):
+ bundle = write_bundle(tmp_path, "sales")
+ layout = _read_layout(bundle)
+ layout["code"]["start"] = f"{start:0{OFFSET_WIDTH}x}" # type:
ignore[index]
+ layout["code"]["end"] = f"{end:0{OFFSET_WIDTH}x}" # type:
ignore[index]
+ _rewrite_layout(bundle, layout)
+
+ with pytest.raises(ValueError, match="code section must contain at
least one byte"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ def test_rejects_metadata_offset_mismatch(self, tmp_path):
+ bundle = write_bundle(tmp_path, "sales")
+ layout = _read_layout(bundle)
+ metadata_start = int(layout["metadata"]["start"], 16) # type:
ignore[index, call-overload]
+ layout["metadata"]["start"] = f"{metadata_start + 1:0{OFFSET_WIDTH}x}"
# type: ignore[index]
+ _rewrite_layout(bundle, layout)
+
+ with pytest.raises(ValueError, match="metadata offsets do not match"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ @pytest.mark.parametrize("section", [pytest.param("metadata",
id="metadata-before-decode"), "code"])
+ def test_rejects_section_digest_mismatch(self, tmp_path, section):
+ bundle = write_bundle(tmp_path, "sales")
+ layout = _read_layout(bundle)
+ # For metadata, corrupt the opening brace so decoding would also fail
if attempted first.
+ _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")
+
+ 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")
+
+ 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")
+
+ 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")
+
+ @pytest.mark.parametrize(
+ "metadata_payload",
+ [
+ b'{"value":"\xff"}',
+ _metadata_json("sales").decode().encode("utf-16"),
+ ],
+ ids=["invalid-utf8", "utf16"],
+ )
+ def test_rejects_metadata_that_is_not_utf8(self, tmp_path,
metadata_payload):
+ 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")
+
+ 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")
+
+ @pytest.mark.parametrize("dags", [None, []], ids=["missing",
"not-a-mapping"])
+ def test_rejects_missing_or_malformed_dags(self, tmp_path, dags):
+ metadata = json.loads(_metadata_json("sales"))
+ if dags is None:
+ del metadata["dags"]
+ else:
+ metadata["dags"] = dags
+ 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")
+
+ def test_rejects_oversized_metadata(self, tmp_path):
+ write_bundle(
+ tmp_path,
+ "sales",
+ metadata_payload=b"A" * _reader._MAX_METADATA_LINE_BYTES,
+ )
+
+ with pytest.raises(ValueError, match="embedded airflow metadata
exceeds"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ def test_rejects_change_during_verification(self, tmp_path, monkeypatch):
+ bundle = write_bundle(tmp_path, "sales")
+ original_hash_region = _hash_region
+
+ def hash_then_touch(*args, **kwargs):
+ digest = original_hash_region(*args, **kwargs)
+ if kwargs["section"] == "metadata":
+ stat_result = bundle.stat()
+ os.utime(bundle, ns=(stat_result.st_atime_ns,
stat_result.st_mtime_ns + 1_000_000_000))
+ return digest
+
+ 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")
+
+ def test_rejects_change_after_verification_before_metadata_decode(self,
tmp_path, monkeypatch):
+ bundle = write_bundle(tmp_path, "sales")
+ original_put = _digest_cache.put
+
+ def put_then_touch(*args, **kwargs):
+ original_put(*args, **kwargs)
+ stat_result = bundle.stat()
+ os.utime(bundle, ns=(stat_result.st_atime_ns,
stat_result.st_mtime_ns + 1_000_000_000))
+
+ 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")
+
+ @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"),
+ ],
+ )
+ def test_translates_fstat_errors(self, tmp_path, monkeypatch,
failure_call, message):
+ bundle = write_bundle(tmp_path, "sales")
+ stat_result = bundle.stat()
+ call_count = 0
+
+ def fail_selected_fstat(_):
+ nonlocal call_count
+ call_count += 1
+ if call_count == failure_call:
+ raise OSError("test failure")
+ return stat_result
+
+ monkeypatch.setattr(_reader.os, "fstat", fail_selected_fstat)
+
+ with pytest.raises(OSError, match=message):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ def test_translates_metadata_read_error(self, tmp_path, monkeypatch):
+ bundle = write_bundle(tmp_path, "sales")
+ layout_line = bundle.read_bytes().splitlines(keepends=True)[0]
+ bundle_file = mock.MagicMock(spec=io.BufferedReader)
+ bundle_file.fileno.return_value = 1
+ bundle_file.readline.side_effect = [layout_line, OSError("test
failure")]
+ path_open = mock.create_autospec(pathlib.Path.open)
+ path_open.return_value = bundle_file
+ 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")
+
+ @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")
+
+ assert hash_region.call_count == 2
+
+ @mock.patch.object(_reader.os, "fstat", autospec=True)
+ def test_cache_uses_ctime_to_detect_corruption_with_restored_mtime(self,
fstat, tmp_path):
+ bundle = write_bundle(tmp_path, "sales")
+ original_stat = bundle.stat()
+ # Control the reported timestamps rather than relying on filesystem
clock resolution.
+ fstat.return_value = mock.Mock(
+ spec=os.stat_result,
+ st_dev=original_stat.st_dev,
+ st_ino=original_stat.st_ino,
+ st_mtime_ns=original_stat.st_mtime_ns,
+ st_ctime_ns=original_stat.st_ctime_ns,
+ st_size=original_stat.st_size,
+ )
+ read_bundle(bundle)
+ layout = _read_layout(bundle)
+ _mutate_byte(bundle, int(layout["code"]["start"], 16)) # type:
ignore[index, call-overload]
+ fstat.return_value.st_ctime_ns += 1
+
+ with pytest.raises(ValueError, match="code SHA-256 mismatch"):
+ read_bundle(tmp_path / "bundle.mjs")
+
+ def test_digest_cache_evicts_least_recently_used_entry(self):
+ cache = _reader._BundleDigestCache(maxsize=2)
+ section = _reader._DeclaredSection(start=0, end=1, sha256=b"0" * 32)
+ digests = _reader._ComputedDigests(metadata=b"1" * 32, code=b"2" * 32)
+
+ def build_key(inode):
+ return _reader._DigestCacheKey(
+ path="bundle.mjs",
+ metadata=section,
+ code=section,
+ device=1,
+ inode=inode,
+ mtime_ns=1,
+ ctime_ns=1,
+ size=1,
+ )
+
+ cache.put(build_key(1), digests)
+ cache.put(build_key(2), digests)
+ assert cache.get(build_key(1)) == digests
+ cache.put(build_key(3), digests)
+
+ assert cache.get(build_key(2)) is None
+ assert cache.get(build_key(1)) == digests
+ assert cache.get(build_key(3)) == digests
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 6efe45566a6..304de1832b4 100644
--- a/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py
+++ b/task-sdk/tests/task_sdk/coordinators/node/test_coordinator.py
@@ -18,22 +18,31 @@
from __future__ import annotations
-import base64
+import json
import pathlib
+from unittest import mock
import pytest
+from task_sdk.coordinators.node._bundle_test_utils import (
+ mutate_byte,
+ read_layout,
+ write_bundle,
+)
from uuid6 import uuid7
from airflow.sdk.api.datamodels._generated import TaskInstance
-from airflow.sdk.coordinators.node.coordinator import (
- EMBEDDED_METADATA_MAX_BYTES,
- NodeCoordinator,
- _find_bundle,
-)
+from airflow.sdk.coordinators.node import _bundle_reader as _reader
+from airflow.sdk.coordinators.node._bundle_reader import _digest_cache
+from airflow.sdk.coordinators.node.coordinator import NodeCoordinator, _Bundle
SCHEMA_VERSION = "2026-06-16"
[email protected](autouse=True)
+def clear_digest_cache():
+ _digest_cache.clear()
+
+
def _make_ti(dag_id: str = "test_dag", queue: str = "ts") -> TaskInstance:
return TaskInstance(
id=uuid7(),
@@ -47,31 +56,6 @@ def _make_ti(dag_id: str = "test_dag", queue: str = "ts") ->
TaskInstance:
)
-def _metadata_yaml(schema_version: str) -> str:
- return f"""\
-airflow_bundle_metadata_version: "1.0"
-sdk:
- language: typescript
- version: "0.1.0"
- supervisor_schema_version: "{schema_version}"
-source: src/airflow.ts
-dags:
- test_dag:
- tasks:
- - test_task
-"""
-
-
-def write_bundle(
- root: pathlib.Path, schema_version: str = SCHEMA_VERSION, payload: str |
None = None
-) -> pathlib.Path:
- if payload is None:
- payload =
base64.b64encode(_metadata_yaml(schema_version).encode("utf-8")).decode("ascii")
- bundle = root / "bundle.mjs"
- bundle.write_text(f"//# airflowMetadata={payload}\nexport {{}};\n",
encoding="utf-8")
- return bundle
-
-
class TestNodeCoordinatorAttributes:
def test_default_kwargs(self):
coordinator = NodeCoordinator(bundles_root="/airflow/ts-bundles")
@@ -99,125 +83,169 @@ class TestNodeCoordinatorAttributes:
NodeCoordinator(bundles_root=None)
-class TestNodeCoordinatorBundleSelection:
- def test_find_bundle_returns_bundle_mjs(self, tmp_path):
- bundle = write_bundle(tmp_path)
+class TestNodeCoordinatorExecuteTaskCommand:
+ def test_selects_bundle_by_dag_id(self, tmp_path):
+ selected = write_bundle(tmp_path, "sales")
+ coordinator = NodeCoordinator(
+ node_executable="/opt/node/bin/node",
+ bundles_root=tmp_path,
+ )
+
+ command, schema_version =
coordinator._build_execute_task_command(what=_make_ti(dag_id="sales"))
- found = _find_bundle([tmp_path])
+ assert command == ["/opt/node/bin/node", str(selected)]
+ assert schema_version == SCHEMA_VERSION
- assert found.path == bundle
- assert found.schema_version == SCHEMA_VERSION
- @pytest.mark.parametrize(
- ("payload", "message"),
- [
- ("not-base64!", "cannot parse embedded airflow metadata"),
- (base64.b64encode(b"[not-a-mapping]").decode("ascii"), "must
contain a mapping"),
- (base64.b64encode(b"sdk: [not-a-mapping]").decode("ascii"),
"missing sdk metadata mapping"),
- (
- base64.b64encode(b"sdk:\n language:
typescript").decode("ascii"),
- "missing or invalid sdk.supervisor_schema_version",
- ),
- ],
- )
- def test_find_bundle_rejects_invalid_embedded_metadata(self, tmp_path,
payload, message):
- write_bundle(tmp_path, payload=payload)
+class TestBundleFind:
+ def test_ignores_roots_without_bundle_mjs(self, tmp_path):
+ (tmp_path / "tasks.mjs").write_bytes(b"export {};\n")
- with pytest.raises(FileNotFoundError, match=message):
- _find_bundle([tmp_path])
+ with pytest.raises(FileNotFoundError, match="dag_id='sales'"):
+ _Bundle.find([tmp_path], "sales")
- def test_find_bundle_rejects_oversized_embedded_metadata(self, tmp_path):
- write_bundle(tmp_path, payload="A" * EMBEDDED_METADATA_MAX_BYTES)
+ def test_reports_unreadable_bundle(self, tmp_path, monkeypatch):
+ write_bundle(tmp_path, "sales")
+ original_open = pathlib.Path.open
- with pytest.raises(FileNotFoundError, match="embedded airflow metadata
exceeds"):
- _find_bundle([tmp_path])
+ def raise_os_error(self, *args, **kwargs):
+ if self.name == "bundle.mjs":
+ raise PermissionError("denied")
+ return original_open(self, *args, **kwargs)
- def test_find_bundle_rejects_empty_marker(self, tmp_path):
- (tmp_path / "bundle.mjs").write_text("//# airflowMetadata=\nexport
{};\n", encoding="utf-8")
+ monkeypatch.setattr(pathlib.Path, "open", raise_os_error)
- with pytest.raises(FileNotFoundError, match="must contain a mapping"):
- _find_bundle([tmp_path])
+ with pytest.raises(FileNotFoundError, match="cannot read bundle.mjs"):
+ _Bundle.find([tmp_path], "sales")
- def test_find_bundle_checks_multiple_roots(self, tmp_path):
+ def test_skips_root_when_bundle_probe_fails(self, tmp_path, monkeypatch):
first = tmp_path / "first"
second = tmp_path / "second"
first.mkdir()
second.mkdir()
- bundle = write_bundle(second)
-
- found = _find_bundle([first, second])
+ write_bundle(first, "sales")
+ expected = write_bundle(second, "sales")
+ original_is_file = pathlib.Path.is_file
- assert found.path == bundle
- assert found.schema_version == SCHEMA_VERSION
+ def fail_first_probe(self):
+ if self.parent == first:
+ raise PermissionError("denied")
+ return original_is_file(self)
- def test_find_bundle_ignores_other_mjs_names(self, tmp_path):
- (tmp_path / "tasks.mjs").write_text("export {};\n")
+ monkeypatch.setattr(pathlib.Path, "is_file", fail_first_probe)
- with pytest.raises(FileNotFoundError, match="Cannot find bundle.mjs"):
- _find_bundle([tmp_path])
+ found = _Bundle.find([first, second], "sales")
- def test_find_bundle_rejects_bundle_without_metadata(self, tmp_path):
- (tmp_path / "bundle.mjs").write_text("export {};\n", encoding="utf-8")
+ assert found.path == expected
- with pytest.raises(FileNotFoundError, match="no embedded airflow
metadata"):
- _find_bundle([tmp_path])
+ def test_selects_later_bundle_containing_requested_dag(self, tmp_path):
+ first = tmp_path / "first"
+ second = tmp_path / "second"
+ first.mkdir()
+ second.mkdir()
+ write_bundle(first, "inventory")
+ expected = write_bundle(second, "sales")
- def test_find_bundle_reports_unreadable_bundle(self, tmp_path,
monkeypatch):
- write_bundle(tmp_path)
+ found = _Bundle.find([first, second], "sales")
- def raise_os_error(self, *args, **kwargs):
- if self.name == "bundle.mjs":
- raise PermissionError("denied")
- return original_open(self, *args, **kwargs)
+ assert found.path == expected
- original_open = pathlib.Path.open
- monkeypatch.setattr(pathlib.Path, "open", raise_os_error)
+ def test_first_configured_match_wins_for_duplicate_dag(self, tmp_path):
+ first = tmp_path / "first"
+ second = tmp_path / "second"
+ first.mkdir()
+ second.mkdir()
+ expected = write_bundle(first, "sales",
code=b'console.log("first");\n')
+ write_bundle(second, "sales", code=b'console.log("second");\n')
- with pytest.raises(FileNotFoundError, match="cannot read bundle.mjs"):
- _find_bundle([tmp_path])
+ found = _Bundle.find([first, second], "sales")
- def test_find_bundle_rejects_invalid_schema_version(self, tmp_path):
- write_bundle(tmp_path, schema_version="banana")
+ assert found.path == expected
- with pytest.raises(FileNotFoundError, match="Version 'banana' not
found"):
- _find_bundle([tmp_path])
+ @mock.patch("airflow.sdk.coordinators.node.coordinator.log.debug",
autospec=True)
+ def test_skips_corrupt_candidate_and_selects_later_match(self, log_debug,
tmp_path):
+ first = tmp_path / "first"
+ second = tmp_path / "second"
+ first.mkdir()
+ second.mkdir()
+ corrupt = write_bundle(first, "sales")
+ layout = read_layout(corrupt)
+ mutate_byte(corrupt, int(layout["code"]["start"], 16)) # type:
ignore[index, call-overload]
+ expected = write_bundle(second, "sales")
+
+ found = _Bundle.find([first, second], "sales")
+
+ assert found.path == expected
+ rejected_log = next(
+ call
+ for call in log_debug.call_args_list
+ if call.args == ("TypeScript bundle rejected; skipping",)
+ )
+ assert rejected_log.kwargs["exc_info"] is True
- def test_find_bundle_skips_rejected_bundle_metadata(self, tmp_path):
+ def test_skips_deeply_nested_metadata_and_selects_later_match(self,
tmp_path):
first = tmp_path / "first"
second = tmp_path / "second"
first.mkdir()
second.mkdir()
- (first / "bundle.mjs").write_text("export {};\n", encoding="utf-8")
- bundle = write_bundle(second)
+ deeply_nested_json = b'{"nested":' + (b"[" * 10_000) + b"0" + (b"]" *
10_000) + b"}"
+ write_bundle(first, "sales", metadata_payload=deeply_nested_json)
+ expected = write_bundle(second, "sales")
- found = _find_bundle([first, second])
+ found = _Bundle.find([first, second], "sales")
- assert found.path == bundle
- assert found.schema_version == SCHEMA_VERSION
+ assert found.path == expected
- def test_find_bundle_raises_with_searched_roots(self, tmp_path):
+ def test_skips_layout_decoder_recursion_and_selects_later_match(self,
tmp_path, monkeypatch):
first = tmp_path / "first"
second = tmp_path / "second"
first.mkdir()
second.mkdir()
+ write_bundle(first, "sales")
+ expected = write_bundle(second, "sales")
+ original_loads = json.loads
+ call_count = 0
- with pytest.raises(FileNotFoundError) as exc_info:
- _find_bundle([first, second])
+ def recurse_once(payload):
+ nonlocal call_count
+ call_count += 1
+ if call_count == 1:
+ raise RecursionError("test recursion")
+ return original_loads(payload)
- msg = str(exc_info.value)
- assert str(first.resolve()) in msg
- assert str(second.resolve()) in msg
+ monkeypatch.setattr(_reader.json, "loads", recurse_once)
+ found = _Bundle.find([first, second], "sales")
-class TestNodeCoordinatorExecuteTaskCommand:
- def
test_build_execute_task_command_returns_node_bundle_and_schema_version(self,
tmp_path):
- bundle = write_bundle(tmp_path)
- coordinator = NodeCoordinator(
- node_executable="/opt/node/bin/node",
- bundles_root=tmp_path,
- )
+ assert found.path == expected
- command, schema_version =
coordinator._build_execute_task_command(what=_make_ti())
+ def test_skips_matching_bundle_with_invalid_schema_version(self, tmp_path):
+ first = tmp_path / "first"
+ second = tmp_path / "second"
+ first.mkdir()
+ second.mkdir()
+ write_bundle(first, "sales", schema_version="banana")
+ expected = write_bundle(second, "sales")
- assert command == ["/opt/node/bin/node", str(bundle)]
- assert schema_version == SCHEMA_VERSION
+ found = _Bundle.find([first, second], "sales")
+
+ assert found.path == expected
+
+ def test_error_names_dag_roots_and_rejected_candidates(self, tmp_path):
+ first = tmp_path / "first"
+ second = tmp_path / "second"
+ first.mkdir()
+ second.mkdir()
+ (first / "bundle.mjs").write_bytes(b"export {};\n")
+ write_bundle(second, "inventory")
+
+ with pytest.raises(FileNotFoundError) as exc_info:
+ _Bundle.find([first, second], "sales")
+
+ message = str(exc_info.value)
+ assert "dag_id='sales'" in message
+ assert str(first) in message
+ assert str(second) in message
+ assert "rejected candidates" in message
+ assert "verified bundle declares dag_ids=['inventory']" in message
+ assert "matching bundles were rejected" not in message
diff --git a/ts-sdk/README.md b/ts-sdk/README.md
index b89d846969b..33b25fea65d 100644
--- a/ts-sdk/README.md
+++ b/ts-sdk/README.md
@@ -173,7 +173,7 @@ uses a Python stub Dag.
## Packing bundles
-`airflow-ts-pack` produces everything `NodeCoordinator` needs in one command.
+`airflow-ts-pack` produces a single self-contained bundle in one command.
Packing is build-time only, so `esbuild` is an optional peer dependency the
runtime install skips:
@@ -185,9 +185,13 @@ 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 leading `//# airflowMetadata=<base64>` comment. The result is a
-single deployable file whose metadata cannot drift from its code; no
-hand-written sidecar is needed.
+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.
Options:
diff --git a/ts-sdk/src/cli/bundle-encoder.ts b/ts-sdk/src/cli/bundle-encoder.ts
new file mode 100644
index 00000000000..5870b0faf84
--- /dev/null
+++ b/ts-sdk/src/cli/bundle-encoder.ts
@@ -0,0 +1,174 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Encodes a self-contained TypeScript Dag bundle that remains directly
+ * executable by Node.
+ *
+ * Final byte order:
+ *
+ * airflowBundle header
+ * -> airflowMetadata
+ * -> executable JavaScript
+ *
+ * The header tells Airflow where each region begins and ends and carries the
+ * digest used to verify each one. Metadata describes what the bundle can
serve,
+ * and executable JavaScript runs its task handlers.
+ *
+ * This module owns the on-disk encoding. Readers must use the header's named
+ * byte ranges rather than relying on incidental line positions.
+ */
+
+import { createHash } from "node:crypto";
+
+import type { BundleManifest } from "../coordinator/manifest.js";
+
+const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
+const EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024;
+const OFFSET_HEX_WIDTH = 16;
+
+export const EMBEDDED_METADATA_PREFIX = "//# airflowMetadata=";
+export const EMBEDDED_LAYOUT_PREFIX = "//# airflowBundle=";
+
+export interface BundleEncoderInput {
+ bundleManifest: BundleManifest;
+ sdkVersion: string;
+ entrypointName: string;
+ executable: Uint8Array;
+}
+
+interface BundleMetadata {
+ airflow_bundle_metadata_version: string;
+ sdk: { language: string; version: string; supervisor_schema_version: string
};
+ source: string;
+ dags: BundleManifest["dags"];
+}
+
+interface VerifiedByteRange {
+ end: string;
+ sha256: string;
+ start: string;
+}
+
+interface BundleHeader {
+ code: VerifiedByteRange;
+ metadata: VerifiedByteRange;
+}
+
+export function encodeBundle(input: BundleEncoderInput): Buffer {
+ const metadata = encodeMetadata(input);
+ const executable = encodeExecutable(input.executable);
+ const header = encodeHeader({ metadata, executable });
+
+ return Buffer.concat([header, metadata, executable]);
+}
+
+function encodeHeader(regions: { metadata: Buffer; executable: Buffer }):
Buffer {
+ const metadataPayload = regions.metadata.subarray(
+ Buffer.byteLength(EMBEDDED_METADATA_PREFIX),
+ -1,
+ );
+ const digests = {
+ code: computeSha256(regions.executable),
+ metadata: computeSha256(metadataPayload),
+ };
+ const zeroOffset = "0".repeat(OFFSET_HEX_WIDTH);
+ const placeholderHeader = renderHeader({
+ code: { start: zeroOffset, end: zeroOffset, sha256: digests.code },
+ metadata: { start: zeroOffset, end: zeroOffset, sha256: digests.metadata },
+ });
+ const metadataStart = placeholderHeader.length +
Buffer.byteLength(EMBEDDED_METADATA_PREFIX);
+ const metadataEnd = metadataStart + metadataPayload.length;
+ const codeStart = placeholderHeader.length + regions.metadata.length;
+ const codeEnd = codeStart + regions.executable.length;
+ const header = renderHeader({
+ code: {
+ start: formatOffset(codeStart),
+ end: formatOffset(codeEnd),
+ sha256: digests.code,
+ },
+ metadata: {
+ start: formatOffset(metadataStart),
+ end: formatOffset(metadataEnd),
+ sha256: digests.metadata,
+ },
+ });
+ if (header.length !== placeholderHeader.length) {
+ throw new Error("Bundle header changed length while resolving section
offsets");
+ }
+ return header;
+}
+
+function encodeMetadata(input: BundleEncoderInput): Buffer {
+ const payload = Buffer.from(
+ JSON.stringify(buildBundleMetadata(input))
+ .replaceAll("\u2028", "\\u2028")
+ .replaceAll("\u2029", "\\u2029"),
+ "utf-8",
+ );
+ const metadata = Buffer.concat([
+ Buffer.from(EMBEDDED_METADATA_PREFIX, "ascii"),
+ payload,
+ Buffer.from("\n", "ascii"),
+ ]);
+ if (metadata.length > EMBEDDED_METADATA_MAX_BYTES) {
+ throw new Error(
+ `Embedded airflow metadata is ${metadata.length} bytes, ` +
+ `over the ${EMBEDDED_METADATA_MAX_BYTES} byte limit; reduce the number
of registered tasks`,
+ );
+ }
+ return metadata;
+}
+
+function encodeExecutable(executable: Uint8Array): Buffer {
+ const bytes = Buffer.from(executable);
+ if (bytes[0] !== 0x23 || bytes[1] !== 0x21) return bytes;
+ const newline = bytes.indexOf(0x0a);
+ return newline === -1 ? Buffer.alloc(0) : bytes.subarray(newline + 1);
+}
+
+function buildBundleMetadata(input: BundleEncoderInput): BundleMetadata {
+ return {
+ airflow_bundle_metadata_version: AIRFLOW_BUNDLE_METADATA_VERSION,
+ sdk: {
+ language: "typescript",
+ version: input.sdkVersion,
+ supervisor_schema_version:
input.bundleManifest.supervisor_schema_version,
+ },
+ source: input.entrypointName,
+ dags: input.bundleManifest.dags,
+ };
+}
+
+function renderHeader(header: BundleHeader): Buffer {
+ const payload = JSON.stringify(header);
+ return Buffer.from(`${EMBEDDED_LAYOUT_PREFIX}${payload}\n`, "ascii");
+}
+
+function formatOffset(offset: number): string {
+ const value = offset.toString(16);
+ if (value.length > OFFSET_HEX_WIDTH) {
+ throw new Error(`Bundle offset ${offset} exceeds the 16-digit hexadecimal
layout limit`);
+ }
+ return value.padStart(OFFSET_HEX_WIDTH, "0");
+}
+
+function computeSha256(contents: Uint8Array): string {
+ return createHash("sha256").update(contents).digest("hex");
+}
diff --git a/ts-sdk/src/cli/pack.ts b/ts-sdk/src/cli/pack.ts
index 0642f0e13d3..040c80e4414 100644
--- a/ts-sdk/src/cli/pack.ts
+++ b/ts-sdk/src/cli/pack.ts
@@ -18,11 +18,11 @@
*/
// airflow-ts-pack: bundle a TypeScript entrypoint into the single-file
-// artifact NodeCoordinator consumes — `bundle.mjs` with the airflow
-// metadata embedded as a leading `//# airflowMetadata=<base64>` comment.
+// artifact NodeCoordinator consumes — `bundle.mjs` with metadata and an
+// integrity layout descriptor embedded in JavaScript comments.
//
// Build first, then run the built bundle with --airflow-metadata so the
-// manifest comes from the bundle's own task registry and schema version,
+// manifest comes from the bundle's own Dag registry and schema version,
// never from a hand-written sidecar.
import { execFileSync } from "node:child_process";
@@ -34,21 +34,20 @@ import {
AIRFLOW_METADATA_SENTINEL,
type BundleManifest,
} from "../coordinator/manifest.js";
+import { encodeBundle } from "./bundle-encoder.js";
import { warnOnSuspiciousIds } from "./validate.js";
-const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
const BUNDLE_FILENAME = "bundle.mjs";
-// bundle.mjs is written only after validation, so failures leave no partial
artifact.
+// Write bundle.mjs only after the build and manifest checks succeed, so a
+// failed pack cannot leave a partial final artifact.
const STAGING_FILENAME = "bundle.pack-staging.mjs";
const MANIFEST_TIMEOUT_MS = 60_000;
const MANIFEST_MAX_BUFFER_BYTES = 64 * 1024 * 1024;
-const EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024;
-export const EMBEDDED_METADATA_PREFIX = "//# airflowMetadata=";
const USAGE = `Usage: airflow-ts-pack <entry> [--outdir <dir>] [--source
<name>]
Bundles <entry> into <outdir>/${BUNDLE_FILENAME} with esbuild and embeds the
-airflow metadata generated from the bundle's registered tasks.
+airflow metadata generated from the bundle's served Dags.
Options:
--outdir <dir> Output directory (default: dist)
@@ -89,32 +88,6 @@ export function parsePackArgs(argv: readonly string[]):
PackArgs {
return { entry, outdir, source: source ?? path.basename(entry) };
}
-export interface PackMetadata {
- airflow_bundle_metadata_version: string;
- sdk: { language: string; version: string; supervisor_schema_version: string
};
- source: string;
- dags: BundleManifest["dags"];
-}
-
-// JSON string literals are valid YAML double-quoted scalars, so every
-// scalar below is emitted through JSON.stringify for correct escaping.
-export function renderMetadataYaml(metadata: PackMetadata): string {
- const lines = [
- `airflow_bundle_metadata_version:
${JSON.stringify(metadata.airflow_bundle_metadata_version)}`,
- "sdk:",
- ` language: ${JSON.stringify(metadata.sdk.language)}`,
- ` version: ${JSON.stringify(metadata.sdk.version)}`,
- ` supervisor_schema_version:
${JSON.stringify(metadata.sdk.supervisor_schema_version)}`,
- `source: ${JSON.stringify(metadata.source)}`,
- "dags:",
- ];
- for (const [dagId, dag] of Object.entries(metadata.dags)) {
- lines.push(` ${JSON.stringify(dagId)}:`);
- lines.push(` tasks: [${dag.tasks.map((task) =>
JSON.stringify(task)).join(", ")}]`);
- }
- return `${lines.join("\n")}\n`;
-}
-
function readSdkVersion(): string {
const packageJsonUrl = new URL("../../package.json", import.meta.url);
const { version } = JSON.parse(readFileSync(packageJsonUrl, "utf-8")) as {
version: string };
@@ -197,14 +170,6 @@ function isTaskIdList(value: unknown): value is string[] {
return Array.isArray(value) && value.every((item) => typeof item ===
"string" && item.length > 0);
}
-// esbuild keeps an entry hashbang as line 1, where the metadata comment must
go;
-// NodeCoordinator always runs the bundle through `node`, so drop it.
-function stripShebang(bundle: string): string {
- if (!bundle.startsWith("#!")) return bundle;
- const newline = bundle.indexOf("\n");
- return newline === -1 ? "" : bundle.slice(newline + 1);
-}
-
async function loadEsbuild(): Promise<typeof import("esbuild")> {
try {
return await import("esbuild");
@@ -246,27 +211,16 @@ export async function runPack(argv: readonly string[]):
Promise<void> {
}
warnOnSuspiciousIds(manifest.dags);
- const metadataYaml = renderMetadataYaml({
- airflow_bundle_metadata_version: AIRFLOW_BUNDLE_METADATA_VERSION,
- sdk: {
- language: "typescript",
- version: readSdkVersion(),
- supervisor_schema_version: manifest.supervisor_schema_version,
- },
- source: args.source,
- dags: manifest.dags,
+ const bundle = encodeBundle({
+ bundleManifest: manifest,
+ sdkVersion: readSdkVersion(),
+ entrypointName: args.source,
+ executable: readFileSync(stagingPath),
});
- const metadataLine =
`${EMBEDDED_METADATA_PREFIX}${Buffer.from(metadataYaml,
"utf-8").toString("base64")}\n`;
- if (metadataLine.length > EMBEDDED_METADATA_MAX_BYTES) {
- throw new Error(
- `Embedded airflow metadata is ${metadataLine.length} bytes, ` +
- `over the ${EMBEDDED_METADATA_MAX_BYTES} byte limit; reduce the
number of registered tasks`,
- );
- }
- writeFileSync(bundlePath, metadataLine +
stripShebang(readFileSync(stagingPath, "utf-8")));
+ writeFileSync(bundlePath, bundle);
} finally {
rmSync(stagingPath, { force: true });
}
- console.log(`Wrote ${bundlePath} (airflow metadata embedded)`);
+ console.log(`Wrote ${bundlePath} (airflow metadata and integrity embedded)`);
}
diff --git a/ts-sdk/tests/cli/fixtures/bundle-v1.mjs
b/ts-sdk/tests/cli/fixtures/bundle-v1.mjs
new file mode 100644
index 00000000000..af75e60c244
--- /dev/null
+++ b/ts-sdk/tests/cli/fixtures/bundle-v1.mjs
@@ -0,0 +1,24 @@
+//#
airflowBundle={"code":{"start":"0000000000000203","end":"0000000000000592","sha256":"f814358e0d4aa5d38c10c365515179049171bb9d4bba5b80c06ea549d6f16337"},"metadata":{"start":"000000000000013e","end":"0000000000000202","sha256":"a51dfd6f0c9e8ea867900e55c0387b556d3cb0e98321b62d4625f522ed465041"}}
+//#
airflowMetadata={"airflow_bundle_metadata_version":"1.0","sdk":{"language":"typescript","version":"0.1.0","supervisor_schema_version":"2026-06-16"},"source":"entry.ts","dags":{"test_dag":{"tasks":["test_task"]}}}
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { DagRegistry, serveDags } from "../../../src/index.js";
+
+await serveDags(new DagRegistry());
diff --git a/ts-sdk/tests/cli/pack.test.ts b/ts-sdk/tests/cli/pack.test.ts
index 75b2ccb5a45..51928359c03 100644
--- a/ts-sdk/tests/cli/pack.test.ts
+++ b/ts-sdk/tests/cli/pack.test.ts
@@ -18,6 +18,7 @@
*/
import { execFileSync } from "node:child_process";
+import { createHash } from "node:crypto";
import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from
"node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
@@ -25,15 +26,16 @@ import { fileURLToPath } from "node:url";
import { afterEach, describe, expect, it, vi } from "vitest";
import {
+ EMBEDDED_LAYOUT_PREFIX,
EMBEDDED_METADATA_PREFIX,
- parsePackArgs,
- renderMetadataYaml,
- runPack,
-} from "../../src/cli/pack.js";
+ encodeBundle,
+} from "../../src/cli/bundle-encoder.js";
+import { parsePackArgs, runPack } from "../../src/cli/pack.js";
import { SUPERVISOR_API_VERSION } from "../../src/coordinator/protocol.js";
import { AIRFLOW_METADATA_SENTINEL } from "../../src/coordinator/manifest.js";
const FIXTURE_ENTRY = fileURLToPath(new URL("fixtures/entry.ts",
import.meta.url));
+const GOLDEN_BUNDLE = fileURLToPath(new URL("fixtures/bundle-v1.mjs",
import.meta.url));
const NOISY_ENTRY = fileURLToPath(new URL("fixtures/noisy-entry.ts",
import.meta.url));
const EMPTY_ENTRY = fileURLToPath(new URL("fixtures/empty-entry.ts",
import.meta.url));
const SDK_INDEX = fileURLToPath(new URL("../../src/index.ts",
import.meta.url));
@@ -43,6 +45,15 @@ const SDK_VERSION = (
}
).version;
+interface TestBundleHeader {
+ code: { end: string; sha256: string; start: string };
+ metadata: { end: string; sha256: string; start: string };
+}
+
+function parseHeader(line: string): TestBundleHeader {
+ return JSON.parse(line.slice(EMBEDDED_LAYOUT_PREFIX.length)) as
TestBundleHeader;
+}
+
describe("parsePackArgs", () => {
it("parses entry with defaults", () => {
expect(parsePackArgs(["src/main.ts"])).toEqual({
@@ -70,34 +81,93 @@ describe("parsePackArgs", () => {
});
});
-describe("renderMetadataYaml", () => {
- it("emits schema-conformant YAML with escaped scalars", () => {
- const yaml = renderMetadataYaml({
- airflow_bundle_metadata_version: "1.0",
- sdk: { language: "typescript", version: "0.1.0",
supervisor_schema_version: "2026-06-16" },
- source: 'we"ird.ts',
- dags: { my_dag: { tasks: ["a", 'b"c'] } },
+describe("encodeBundle", () => {
+ it("assembles header, metadata, and executable in physical order", () => {
+ const executable = Buffer.from('console.log("hello");\n');
+ const bundle = encodeBundle({
+ bundleManifest: {
+ supervisor_schema_version: "2026-06-16",
+ dags: { my_dag: { tasks: ["a", 'b"c'] } },
+ },
+ sdkVersion: "0.1.0",
+ entrypointName: 'we"ird.ts',
+ executable,
});
- expect(yaml).toBe(
- [
- 'airflow_bundle_metadata_version: "1.0"',
- "sdk:",
- ' language: "typescript"',
- ' version: "0.1.0"',
- ' supervisor_schema_version: "2026-06-16"',
- 'source: "we\\"ird.ts"',
- "dags:",
- ' "my_dag":',
- ' tasks: ["a", "b\\"c"]',
- "",
- ].join("\n"),
+
+ const firstNewline = bundle.indexOf("\n");
+ const header = parseHeader(bundle.subarray(0,
firstNewline).toString("ascii"));
+ const offset = (value: string): number => Number.parseInt(value, 16);
+ const metadataStart = offset(header.metadata.start);
+ const metadataEnd = offset(header.metadata.end);
+ const executableStart = offset(header.code.start);
+ const executableEnd = offset(header.code.end);
+
+ expect(metadataStart).toBe(firstNewline + 1 +
Buffer.byteLength(EMBEDDED_METADATA_PREFIX));
+ expect(executableStart).toBe(metadataEnd + 1);
+ expect(executableEnd).toBe(bundle.length);
+ expect(bundle.subarray(executableStart,
executableEnd)).toEqual(executable);
+
+ const metadata = bundle.subarray(metadataStart,
metadataEnd).toString("utf-8");
+ expect(metadata).toBe(
+
'{"airflow_bundle_metadata_version":"1.0","sdk":{"language":"typescript","version":"0.1.0","supervisor_schema_version":"2026-06-16"},"source":"we\\"ird.ts","dags":{"my_dag":{"tasks":["a","b\\"c"]}}}',
+ );
+
+ expect(header).not.toHaveProperty("source");
+ expect(header).not.toHaveProperty("version");
+ expect(bundle.toString("utf-8")).not.toContain("airflowSource");
+ });
+
+ it("matches the golden bundle", () => {
+ const executable = readFileSync(EMPTY_ENTRY);
+ const bundle = encodeBundle({
+ bundleManifest: {
+ supervisor_schema_version: "2026-06-16",
+ dags: { test_dag: { tasks: ["test_task"] } },
+ },
+ sdkVersion: "0.1.0",
+ entrypointName: "entry.ts",
+ executable,
+ });
+
+ expect(bundle).toEqual(readFileSync(GOLDEN_BUNDLE));
+ const firstNewline = bundle.indexOf("\n");
+ const header = parseHeader(bundle.subarray(0,
firstNewline).toString("ascii"));
+ for (const section of [header.metadata, header.code]) {
+ expect(section.start).toMatch(/^[0-9a-f]{16}$/);
+ expect(section.end).toMatch(/^[0-9a-f]{16}$/);
+ }
+ });
+
+ it("escapes JavaScript line separators inside the metadata comment", () => {
+ const bundle = encodeBundle({
+ bundleManifest: {
+ supervisor_schema_version: "2026-06-16",
+ dags: { "line\u2028separator": { tasks: ["paragraph\u2029separator"] }
},
+ },
+ sdkVersion: "0.1.0",
+ entrypointName: "entry.ts",
+ executable: Buffer.from("export {};\n"),
+ });
+ const metadataLine = bundle.toString("utf-8").split("\n")[1]!;
+
+ expect(metadataLine).not.toContain("\u2028");
+ expect(metadataLine).not.toContain("\u2029");
+ expect(metadataLine).toContain("\\u2028");
+ expect(metadataLine).toContain("\\u2029");
+
expect(JSON.parse(metadataLine.slice(EMBEDDED_METADATA_PREFIX.length))).toHaveProperty(
+ "dags.line\u2028separator.tasks",
+ ["paragraph\u2029separator"],
);
});
});
function readEmbeddedMetadata(bundlePath: string): string {
- const firstLine = readFileSync(bundlePath, "utf-8").split("\n")[0]!;
- return Buffer.from(firstLine.slice(EMBEDDED_METADATA_PREFIX.length),
"base64").toString("utf-8");
+ const bundle = readFileSync(bundlePath);
+ const firstNewline = bundle.indexOf("\n");
+ const header = parseHeader(bundle.subarray(0,
firstNewline).toString("utf-8"));
+ const start = Number.parseInt(header.metadata.start, 16);
+ const end = Number.parseInt(header.metadata.end, 16);
+ return bundle.subarray(start, end).toString("utf-8");
}
/** Collect what runPack writes to stderr; returns a reader for the text so
far. */
@@ -126,28 +196,23 @@ describe("runPack", () => {
const bundlePath = path.join(nested, "bundle.mjs");
expect(existsSync(path.join(nested, "airflow-metadata.yaml"))).toBe(false);
- const firstLine = readFileSync(bundlePath, "utf-8").split("\n")[0]!;
- expect(firstLine.startsWith(EMBEDDED_METADATA_PREFIX)).toBe(true);
- const metadata = Buffer.from(
- firstLine.slice(EMBEDDED_METADATA_PREFIX.length),
- "base64",
- ).toString("utf-8");
- expect(metadata).toBe(
- [
- 'airflow_bundle_metadata_version: "1.0"',
- "sdk:",
- ' language: "typescript"',
- ` version: ${JSON.stringify(SDK_VERSION)}`,
- ` supervisor_schema_version:
${JSON.stringify(SUPERVISOR_API_VERSION)}`,
- 'source: "entry.ts"',
- "dags:",
- ' "fixture_dag":',
- ' tasks: ["extract", "transform"]',
- ' "other_dag":',
- ' tasks: ["solo"]',
- "",
- ].join("\n"),
- );
+ const [layoutLine, metadataLine] = readFileSync(bundlePath,
"utf-8").split("\n");
+ expect(layoutLine!.startsWith(EMBEDDED_LAYOUT_PREFIX)).toBe(true);
+ expect(metadataLine!.startsWith(EMBEDDED_METADATA_PREFIX)).toBe(true);
+ const metadata =
JSON.parse(metadataLine!.slice(EMBEDDED_METADATA_PREFIX.length));
+ expect(metadata).toEqual({
+ airflow_bundle_metadata_version: "1.0",
+ sdk: {
+ language: "typescript",
+ version: SDK_VERSION,
+ supervisor_schema_version: SUPERVISOR_API_VERSION,
+ },
+ source: "entry.ts",
+ dags: {
+ fixture_dag: { tasks: ["extract", "transform"] },
+ other_dag: { tasks: ["solo"] },
+ },
+ });
const dumped = execFileSync(process.execPath, [bundlePath,
"--airflow-metadata"], {
encoding: "utf-8",
@@ -158,21 +223,45 @@ describe("runPack", () => {
).toBe(SUPERVISOR_API_VERSION);
});
+ it("embeds verifiable metadata and code regions", async () => {
+ outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
+ await runPack([FIXTURE_ENTRY, "--outdir", outdir]);
+
+ const bundle = readFileSync(path.join(outdir, "bundle.mjs"));
+ const firstNewline = bundle.indexOf("\n");
+ const layoutLine = bundle.subarray(0, firstNewline).toString("utf-8");
+ expect(layoutLine.startsWith(EMBEDDED_LAYOUT_PREFIX)).toBe(true);
+
+ const layout = parseHeader(layoutLine);
+ const offset = (value: string): number => Number.parseInt(value, 16);
+
+ expect(layout).not.toHaveProperty("version");
+ const code = bundle.subarray(offset(layout.code.start),
offset(layout.code.end));
+
expect(createHash("sha256").update(code).digest("hex")).toBe(layout.code.sha256);
+
+ const metadataPayload = bundle.subarray(
+ offset(layout.metadata.start),
+ offset(layout.metadata.end),
+ );
+
expect(createHash("sha256").update(metadataPayload).digest("hex")).toBe(layout.metadata.sha256);
+
expect(JSON.parse(metadataPayload.toString("utf-8"))).toHaveProperty("dags.fixture_dag");
+ expect(layout).not.toHaveProperty("source");
+ expect(bundle.toString("utf-8")).not.toContain("airflowSource");
+ });
+
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 bundle = readFileSync(bundlePath, "utf-8");
- expect(bundle.startsWith(EMBEDDED_METADATA_PREFIX)).toBe(true);
+ expect(bundle.startsWith(EMBEDDED_LAYOUT_PREFIX)).toBe(true);
expect(bundle).not.toContain("#!/usr/bin/env node");
expect(existsSync(path.join(outdir,
"bundle.pack-staging.mjs"))).toBe(false);
- const metadata = Buffer.from(
- bundle.split("\n")[0]!.slice(EMBEDDED_METADATA_PREFIX.length),
- "base64",
- ).toString("utf-8");
- expect(metadata).toContain(' "noisy_dag":');
+ const metadataLine = bundle.split("\n")[1]!;
+ const metadata =
JSON.parse(metadataLine.slice(EMBEDDED_METADATA_PREFIX.length));
+ expect(metadata).toHaveProperty("dags.noisy_dag");
execFileSync(process.execPath, [bundlePath, "--airflow-metadata"], {
encoding: "utf-8" });
});
@@ -185,7 +274,7 @@ describe("runPack", () => {
[
`import { Dag, DagRegistry, serveDags } from
${JSON.stringify(SDK_INDEX)};`,
'const bigDag = new Dag("big_dag");',
- 'for (let i = 0; i < 4000; i += 1) bigDag.task(String(i).padStart(240,
"t"), async () => undefined);',
+ 'for (let i = 0; i < 5000; i += 1) bigDag.task(String(i).padStart(240,
"t"), async () => undefined);',
"await serveDags(new DagRegistry(bigDag));",
].join("\n"),
);
@@ -304,8 +393,9 @@ describe("runPack", () => {
await runPack([entry, "--outdir", outdir]);
expect(stderr()).toContain('warning: dag "empty_dag" has no tasks\n');
- expect(readEmbeddedMetadata(path.join(outdir, "bundle.mjs"))).toContain(
- ' "empty_dag":\n tasks: []',
+ expect(JSON.parse(readEmbeddedMetadata(path.join(outdir,
"bundle.mjs")))).toHaveProperty(
+ "dags.empty_dag.tasks",
+ [],
);
});
@@ -326,8 +416,8 @@ describe("runPack", () => {
await runPack([entry, "--outdir", outdir]);
- const metadata = readEmbeddedMetadata(path.join(outdir, "bundle.mjs"));
- expect(metadata).toContain(' "sales_dag":');
- expect(metadata).not.toContain("billing_dag");
+ const metadata = JSON.parse(readEmbeddedMetadata(path.join(outdir,
"bundle.mjs")));
+ expect(metadata).toHaveProperty("dags.sales_dag");
+ expect(metadata).not.toHaveProperty("dags.billing_dag");
});
});
