You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "shivaam (via GitHub)" <gi...@apache.org> on 2026/08/25 04:15:00 UTC
[PR] Add integrity metadata to TypeScript bundles [airflow]
shivaam opened a new pull request, #72046:
URL: https://github.com/apache/airflow/pull/72046
This PR adds the TypeScript writer portion of the integrity format discussed in #69905.
`airflow-ts-pack` now emits a directly executable `bundle.mjs` with three physical regions:
1. a versioned `airflowBundle` header;
2. the existing Airflow metadata; and
3. the executable JavaScript.
The header records fixed-width byte ranges and independent SHA-256 digests for the metadata and executable payloads. Keeping the encoding in a dedicated `bundle-encoder` module leaves `pack.ts` responsible for CLI handling, the esbuild run, and runtime-manifest collection.
The digests let the coordinator detect accidental truncation or modification. They do not authenticate the bundle producer because someone able to replace the payload can also replace the header and its digests.
This PR intentionally does not embed author-written source. Jason's [issue clarification](https://github.com/apache/airflow/issues/69905#issuecomment-5174764657) narrows the required precedent to the integrity-check idea rather than the complete AFBNDL01 feature set. The TypeScript SDK is not released, so this PR also adopts the layout-first format without preserving the previous metadata-first physical layout.
A separate coordinator/spec PR will add reading, digest verification, corruption rejection, caching, and the formal Task SDK format documentation.
Tests cover the versioned golden bundle, exact section order and offsets, metadata and executable digests, direct Node execution, shebang handling, metadata collection, malformed manifests, and failed builds.
Related: #69905
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Codex
Generated-by: Codex following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3997433425
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -45,80 +40,46 @@
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.
- """
Review Comment:
Done. Thanks for the suggestion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3997446607
##########
task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py:
##########
@@ -0,0 +1,417 @@
+#
+# 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 base64
+import io
+import json
+import os
+import pathlib
+from unittest import mock
+
+import pytest
+from task_sdk.coordinators.node._bundle_test_utils import (
+ 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"
+
+
+@pytest.fixture(autouse=True)
+def clear_digest_cache():
+ _digest_cache.clear()
+
+
+class TestBundleReader:
+ def test_reader_returns_verified_bundle_metadata(self):
+ metadata = read_bundle(TYPESCRIPT_V1_FIXTURE)
+
+ assert metadata.dag_ids == frozenset({"test_dag"})
+ assert metadata.supervisor_schema_version == SCHEMA_VERSION
+
+ def test_reads_bundle_produced_by_typescript_encoder(self, tmp_path):
+ bundle = tmp_path / "bundle.mjs"
+ bundle.write_bytes(TYPESCRIPT_V1_FIXTURE.read_bytes())
+
+ metadata = read_bundle(bundle)
+
+ assert metadata.dag_ids == frozenset({"test_dag"})
+ assert metadata.supervisor_schema_version == SCHEMA_VERSION
Review Comment:
Combined the tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3859229525
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
Review Comment:
```suggestion
airflow_bundle_metadata_version: string;
```
Since we pass the `AIRFLOW_BUNDLE_METADATA_VERSION` when construct.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
+ 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;
+ version: 1;
Review Comment:
IIUC, the `BUNDLE_LAYOUT_VERSION` is not necessary, since the `AIRFLOW_BUNDLE_METADATA_VERSION` should be enough.
The coordinator side will know whether the artifact is parseable by checking against the `AIRFLOW_BUNDLE_METADATA_VERSION` flag.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
+ 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;
+ version: 1;
+}
+
+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({
+ version: BUNDLE_LAYOUT_VERSION,
+ 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({
+ version: BUNDLE_LAYOUT_VERSION,
+ 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(
+ Buffer.from(renderMetadata(buildBundleMetadata(input)), "utf-8").toString("base64"),
Review Comment:
I wonder if it's necessary to encode to base64 or not? Since the hash can ensure the integrity. Additionally, having base64 encoding introduce another overhead on the coordinator side and doesn't seem to provide additional value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3996857664
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -45,80 +40,46 @@
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.
- """
+def _select_bundle(bundles_root: Sequence[pathlib.Path], dag_id: str) -> ResolvedBundle:
+ """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
- 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),
- )
- except (TypeError, ValueError) as exc:
+ 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,
+ )
+ continue
+ bundle = ResolvedBundle(path=candidate, schema_version=metadata.supervisor_schema_version)
+ except (OSError, TypeError, ValueError) as exc:
log.debug(
- "TypeScript bundle metadata rejected; skipping",
+ "TypeScript bundle rejected; skipping",
path=candidate,
root=root,
- exc_info=True,
+ reason=str(exc),
Review Comment:
Kept exc_info
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3974978467
##########
task-sdk/docs/ts-bundle-spec.rst:
##########
@@ -0,0 +1,177 @@
+ .. 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:
Review Comment:
```suggestion
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:
```
##########
task-sdk/tests/task_sdk/coordinators/node/test_bundle_reader.py:
##########
@@ -0,0 +1,417 @@
+#
+# 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 base64
+import io
+import json
+import os
+import pathlib
+from unittest import mock
+
+import pytest
+from task_sdk.coordinators.node._bundle_test_utils import (
+ 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"
+
+
+@pytest.fixture(autouse=True)
+def clear_digest_cache():
+ _digest_cache.clear()
+
+
+class TestBundleReader:
+ def test_reader_returns_verified_bundle_metadata(self):
+ metadata = read_bundle(TYPESCRIPT_V1_FIXTURE)
+
+ assert metadata.dag_ids == frozenset({"test_dag"})
+ assert metadata.supervisor_schema_version == SCHEMA_VERSION
+
+ def test_reads_bundle_produced_by_typescript_encoder(self, tmp_path):
+ bundle = tmp_path / "bundle.mjs"
+ bundle.write_bytes(TYPESCRIPT_V1_FIXTURE.read_bytes())
+
+ metadata = read_bundle(bundle)
+
+ assert metadata.dag_ids == frozenset({"test_dag"})
+ assert metadata.supervisor_schema_version == SCHEMA_VERSION
Review Comment:
Let's combine the test.
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -45,80 +40,46 @@
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.
- """
+def _select_bundle(bundles_root: Sequence[pathlib.Path], dag_id: str) -> ResolvedBundle:
+ """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
- 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),
- )
- except (TypeError, ValueError) as exc:
+ 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,
+ )
+ continue
Review Comment:
Bundles that pass full integrity verification but declare a different `dag_id` are dropped without being recorded in `rejected`.
```suggestion
rejected.append((candidate, f"verified bundle declares dag_ids={sorted(metadata.dag_ids)!r}"))
continue
```
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -45,80 +40,46 @@
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.
- """
Review Comment:
No sure would introduce the `_Bundle` attrs be more readable?
https://github.com/apache/airflow/blob/1952520a9eedb63c530be89eade6863ea975bb2e/task-sdk/src/airflow/sdk/coordinators/executable/coordinator.py#L293-L323
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -45,80 +40,46 @@
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.
- """
+def _select_bundle(bundles_root: Sequence[pathlib.Path], dag_id: str) -> ResolvedBundle:
+ """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
- 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),
- )
- except (TypeError, ValueError) as exc:
+ 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,
+ )
+ continue
+ bundle = ResolvedBundle(path=candidate, schema_version=metadata.supervisor_schema_version)
+ except (OSError, TypeError, ValueError) as exc:
log.debug(
- "TypeScript bundle metadata rejected; skipping",
+ "TypeScript bundle rejected; skipping",
path=candidate,
root=root,
- exc_info=True,
+ reason=str(exc),
Review Comment:
Let's keep the `exc_info`
```suggestion
reason=str(exc),
exc_info=True,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3997444115
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -45,80 +40,46 @@
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.
- """
+def _select_bundle(bundles_root: Sequence[pathlib.Path], dag_id: str) -> ResolvedBundle:
+ """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
- 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),
- )
- except (TypeError, ValueError) as exc:
+ 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,
+ )
+ continue
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3868217108
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
+ 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;
+ version: 1;
Review Comment:
> This lets the coordinator detect metadata corruption as well as executable corruption.
This is why the top level shasum comes to play. Additionally, the only consumer of the bundle metadata is the coordinator.
> Keeping them separate also allows the layout header to carry an independent digest for the metadata section. This lets the coordinator detect metadata corruption as well as executable corruption. I think the separation will also help if we later add source, signatures, or other physical sections.
Even if we add the additional metadata fields, we just need to bump the `AIRFLOW_BUNDLE_METADATA_VERSION` with the corresponding coordinator change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r4003576658
##########
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"}}
Review Comment:
I just realized the `.mjs` here isn't the min js.
From my perspative, we should compile into `.min.js` so that user ourself only need to deploy one artifact instead of multiple artifacts in the DagBundle.
The shape I imagined of `.mjs` is
- the current metadata layout as comments
- have a multiple line comments that comment out the visible entry point file source
- then the actual compiled min JS content (should be human unreadable)
Otherwise the `code` in the `airflowBundle` means nothing.
##########
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),
+@attrs.define
+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
Review Comment:
Just found out that we hard coded the artifact file name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #72046:
URL: https://github.com/apache/airflow/pull/72046
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3868054523
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
Review Comment:
Good catch. Updated this locally to `string` so the constant remains the single source of truth.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
+ 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;
+ version: 1;
Review Comment:
The two versions protect different contracts. The layout version describes the physical bundle structure—how the coordinator locates and verifies the metadata and executable sections. `airflow_bundle_metadata_version` describes the logical schema inside the metadata section.
Keeping them separate also allows the layout header to carry an independent digest for the metadata section. This lets the coordinator detect metadata corruption as well as executable corruption. I think the separation will also help if we later add source, signatures, or other physical sections.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
+ 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;
+ version: 1;
+}
+
+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({
+ version: BUNDLE_LAYOUT_VERSION,
+ 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({
+ version: BUNDLE_LAYOUT_VERSION,
+ 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(
+ Buffer.from(renderMetadata(buildBundleMetadata(input)), "utf-8").toString("base64"),
Review Comment:
Good point. I kept the base64 encoding to support the YAML-based metadata format added by the original TypeScript packer. Since the runtime already produces structured JSON and the TypeScript SDK is unreleased, I think it makes sense to simplify this to compact JSON embedded directly in the single-line comment and remove the metadata base64 encoding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #72046:
URL: https://github.com/apache/airflow/pull/72046#issuecomment-5433257339
Regarding the coordinator/spec scope in [this review](https://github.com/apache/airflow/pull/72046#pullrequestreview-5026271011): agreed. I’ve combined the coordinator reader, digest verification, corruption rejection, caching, format documentation, and tests into this PR locally so the existing E2E path can consume the new bundles. I’ll update the PR description when I publish the revised branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3868054523
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
Review Comment:
Good catch. Updated this locally to `string` so the constant remains the single source of truth.
---
Drafted-by: Codex; reviewed by @shivaam before posting
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
+ 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;
+ version: 1;
Review Comment:
The two versions protect different contracts. The layout version describes the physical bundle structure—how the coordinator locates and verifies the metadata and executable sections. `airflow_bundle_metadata_version` describes the logical schema inside the metadata section.
Keeping them separate also allows the layout header to carry an independent digest for the metadata section. This lets the coordinator detect metadata corruption as well as executable corruption. I think the separation will also help if we later add source, signatures, or other physical sections.
---
Drafted-by: Codex; reviewed by @shivaam before posting
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
+ 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;
+ version: 1;
+}
+
+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({
+ version: BUNDLE_LAYOUT_VERSION,
+ 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({
+ version: BUNDLE_LAYOUT_VERSION,
+ 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(
+ Buffer.from(renderMetadata(buildBundleMetadata(input)), "utf-8").toString("base64"),
Review Comment:
Good point. I kept the base64 encoding to support the YAML-based metadata format added by the original TypeScript packer. Since the runtime already produces structured JSON and the TypeScript SDK is unreleased, I think it makes sense to simplify this to compact JSON embedded directly in the single-line comment and remove the metadata base64 encoding.
---
Drafted-by: Codex; reviewed by @shivaam before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add integrity metadata to TypeScript bundles [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3868622109
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -0,0 +1,195 @@
+/*!
+ * 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 version
+ * and 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 BUNDLE_LAYOUT_VERSION = 1;
+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: "1.0";
+ 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;
+ version: 1;
Review Comment:
> > This lets the coordinator detect metadata corruption as well as executable corruption.
>
> This is why the top level shasum comes to play. Additionally, the only consumer of the bundle metadata is the coordinator.
>
> > Keeping them separate also allows the layout header to carry an independent digest for the metadata section. This lets the coordinator detect metadata corruption as well as executable corruption. I think the separation will also help if we later add source, signatures, or other physical sections.
>
> Even if we add the additional metadata fields, we just need to bump the `AIRFLOW_BUNDLE_METADATA_VERSION` with the corresponding coordinator change.
I think that makes sense .As long as we keep the first-line descriptor stable enough to find and verify the metadata, one version is enough for the whole bundle. I will remove the bundle version
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org