jason810496 commented on code in PR #69295: URL: https://github.com/apache/airflow/pull/69295#discussion_r3548939107
########## ts-sdk/src/cli/pack.ts: ########## @@ -0,0 +1,167 @@ +/*! + * 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. + */ + +// 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. +// +// Mirrors airflow-go-pack: build first, then run the built bundle with +// --airflow-metadata so the manifest comes from the bundle's own task +// registry and schema version, never from a hand-written sidecar. + +import { execFileSync } from "node:child_process"; +import { readFileSync, writeFileSync } from "node:fs"; +import path from "node:path"; + +import { build } from "esbuild"; + +import { AIRFLOW_METADATA_FLAG, type BundleManifest } from "../coordinator/runtime.js"; + +const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0"; +const BUNDLE_FILENAME = "bundle.mjs"; +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. + +Options: + --outdir <dir> Output directory (default: dist) + --source <name> Display name of the primary source file (default: <entry> basename) +`; + +export interface PackArgs { + entry: string; + outdir: string; + source: string; +} + +export function parsePackArgs(argv: readonly string[]): PackArgs { + let entry: string | null = null; + let outdir = "dist"; + let source: string | null = null; + for (let i = 0; i < argv.length; i += 1) { + const arg = argv[i]!; + if (arg === "--outdir" || arg === "--source") { + const value = argv[i + 1]; + if (!value) throw new Error(`${arg} requires a value\n\n${USAGE}`); + if (arg === "--outdir") outdir = value; + else source = value; + i += 1; + } else if (arg.startsWith("-")) { + throw new Error(`Unknown option ${arg}\n\n${USAGE}`); + } else if (entry) { + throw new Error(`Unexpected argument ${arg}\n\n${USAGE}`); + } else { + entry = arg; + } + } + if (!entry) throw new Error(`Missing entry file\n\n${USAGE}`); + 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 }; + return version; +} + +function readBundleManifest(bundlePath: string): BundleManifest { + const stdout = execFileSync(process.execPath, [bundlePath, AIRFLOW_METADATA_FLAG], { + encoding: "utf-8", + }); + let manifest: BundleManifest; + try { + manifest = JSON.parse(stdout) as BundleManifest; + } catch (error) { + throw new Error(`Bundle produced invalid --airflow-metadata output: ${String(error)}`, { + cause: error, + }); + } + if (!manifest.supervisor_schema_version || typeof manifest.dags !== "object") { + throw new Error("Bundle produced incomplete --airflow-metadata output"); + } + return manifest; +} + +export async function runPack(argv: readonly string[]): Promise<void> { + const args = parsePackArgs(argv); + const bundlePath = path.join(args.outdir, BUNDLE_FILENAME); + + await build({ + entryPoints: [args.entry], + bundle: true, + platform: "node", + format: "esm", + target: "node22", + outfile: bundlePath, + }); + + const manifest = readBundleManifest(bundlePath); + if (Object.keys(manifest.dags).length === 0) { + throw new Error( + `${args.entry} registered no tasks; call registerTask(...) before startCoordinator()`, + ); + } + + 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 metadataLine = `${EMBEDDED_METADATA_PREFIX}${Buffer.from(metadataYaml, "utf-8").toString("base64")}\n`; + writeFileSync( + bundlePath, + Buffer.concat([Buffer.from(metadataLine, "utf-8"), readFileSync(bundlePath)]), + ); Review Comment: **[major] Prepending metadata can break a bundle with a preserved shebang.** The metadata line is prepended unconditionally. If the entry starts with a hashbang, esbuild keeps `#!...` as the first line of `bundle.mjs`; after this prepend it becomes line 2 and `node bundle.mjs` fails with `SyntaxError: Invalid or unexpected token`. Consider detecting a leading `#!` line in the built output and inserting the metadata *after* it (or stripping/regenerating the shebang). Reproduced against esbuild 0.28.1 / Node: a `#!/usr/bin/env node` entry runs fine before packing and throws the SyntaxError after. _Nit, same spot:_ the artifact is read back into memory and rewritten via `Buffer.concat`, loading the whole bundle twice — fine for a build tool, just noting it. ########## ts-sdk/src/cli/pack.ts: ########## @@ -0,0 +1,167 @@ +/*! + * 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. + */ + +// 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. +// +// Mirrors airflow-go-pack: build first, then run the built bundle with +// --airflow-metadata so the manifest comes from the bundle's own task +// registry and schema version, never from a hand-written sidecar. + +import { execFileSync } from "node:child_process"; +import { readFileSync, writeFileSync } from "node:fs"; +import path from "node:path"; + +import { build } from "esbuild"; + +import { AIRFLOW_METADATA_FLAG, type BundleManifest } from "../coordinator/runtime.js"; + +const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0"; +const BUNDLE_FILENAME = "bundle.mjs"; +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. + +Options: + --outdir <dir> Output directory (default: dist) + --source <name> Display name of the primary source file (default: <entry> basename) +`; + +export interface PackArgs { + entry: string; + outdir: string; + source: string; +} + +export function parsePackArgs(argv: readonly string[]): PackArgs { + let entry: string | null = null; + let outdir = "dist"; + let source: string | null = null; + for (let i = 0; i < argv.length; i += 1) { + const arg = argv[i]!; + if (arg === "--outdir" || arg === "--source") { + const value = argv[i + 1]; + if (!value) throw new Error(`${arg} requires a value\n\n${USAGE}`); + if (arg === "--outdir") outdir = value; + else source = value; + i += 1; + } else if (arg.startsWith("-")) { + throw new Error(`Unknown option ${arg}\n\n${USAGE}`); + } else if (entry) { + throw new Error(`Unexpected argument ${arg}\n\n${USAGE}`); + } else { + entry = arg; + } + } + if (!entry) throw new Error(`Missing entry file\n\n${USAGE}`); + 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 }; + return version; +} + +function readBundleManifest(bundlePath: string): BundleManifest { + const stdout = execFileSync(process.execPath, [bundlePath, AIRFLOW_METADATA_FLAG], { + encoding: "utf-8", + }); + let manifest: BundleManifest; + try { + manifest = JSON.parse(stdout) as BundleManifest; + } catch (error) { + throw new Error(`Bundle produced invalid --airflow-metadata output: ${String(error)}`, { + cause: error, + }); + } + if (!manifest.supervisor_schema_version || typeof manifest.dags !== "object") { Review Comment: **[minor] `typeof manifest.dags !== "object"` lets `null` through.** `typeof null === "object"`, so a manifest with `{"dags": null}` passes this guard and then `Object.keys(manifest.dags)` (line 144) throws a raw `TypeError` rather than the intended "incomplete output" error. Add a `manifest.dags && …` check. ########## ts-sdk/package.json: ########## @@ -57,7 +60,8 @@ "node": ">=22" }, "dependencies": { - "@msgpack/msgpack": "^3.1.2" + "@msgpack/msgpack": "^3.1.2", + "esbuild": "^0.28.1" Review Comment: **[major] `esbuild` is now a hard runtime dependency.** Packing is build-time only (just `airflow-ts-pack` imports esbuild), but every consumer installing `apache-airflow-ts-sdk` purely for the coordinator runtime now pulls esbuild and its platform binaries. Consider `peerDependencies`/`optionalDependencies`, or splitting the packer into its own bin package, to keep the runtime install lean. ########## ts-sdk/src/cli/pack.ts: ########## @@ -0,0 +1,167 @@ +/*! + * 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. + */ + +// 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. +// +// Mirrors airflow-go-pack: build first, then run the built bundle with +// --airflow-metadata so the manifest comes from the bundle's own task +// registry and schema version, never from a hand-written sidecar. + +import { execFileSync } from "node:child_process"; +import { readFileSync, writeFileSync } from "node:fs"; +import path from "node:path"; + +import { build } from "esbuild"; + +import { AIRFLOW_METADATA_FLAG, type BundleManifest } from "../coordinator/runtime.js"; + +const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0"; +const BUNDLE_FILENAME = "bundle.mjs"; +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. + +Options: + --outdir <dir> Output directory (default: dist) + --source <name> Display name of the primary source file (default: <entry> basename) +`; + +export interface PackArgs { + entry: string; + outdir: string; + source: string; +} + +export function parsePackArgs(argv: readonly string[]): PackArgs { Review Comment: nit: `parsePackArgs` repeats `\n\n${USAGE}` at four throw sites; a small `usageError(msg)` helper would DRY these. ########## ts-sdk/src/cli/pack.ts: ########## @@ -0,0 +1,167 @@ +/*! + * 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. + */ + +// 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. +// +// Mirrors airflow-go-pack: build first, then run the built bundle with +// --airflow-metadata so the manifest comes from the bundle's own task +// registry and schema version, never from a hand-written sidecar. + +import { execFileSync } from "node:child_process"; +import { readFileSync, writeFileSync } from "node:fs"; +import path from "node:path"; + +import { build } from "esbuild"; + +import { AIRFLOW_METADATA_FLAG, type BundleManifest } from "../coordinator/runtime.js"; + +const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0"; +const BUNDLE_FILENAME = "bundle.mjs"; +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. + +Options: + --outdir <dir> Output directory (default: dist) + --source <name> Display name of the primary source file (default: <entry> basename) +`; + +export interface PackArgs { + entry: string; + outdir: string; + source: string; +} + +export function parsePackArgs(argv: readonly string[]): PackArgs { + let entry: string | null = null; + let outdir = "dist"; + let source: string | null = null; + for (let i = 0; i < argv.length; i += 1) { + const arg = argv[i]!; + if (arg === "--outdir" || arg === "--source") { + const value = argv[i + 1]; + if (!value) throw new Error(`${arg} requires a value\n\n${USAGE}`); + if (arg === "--outdir") outdir = value; + else source = value; + i += 1; + } else if (arg.startsWith("-")) { + throw new Error(`Unknown option ${arg}\n\n${USAGE}`); + } else if (entry) { + throw new Error(`Unexpected argument ${arg}\n\n${USAGE}`); + } else { + entry = arg; + } + } + if (!entry) throw new Error(`Missing entry file\n\n${USAGE}`); + 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 }; + return version; +} + +function readBundleManifest(bundlePath: string): BundleManifest { + const stdout = execFileSync(process.execPath, [bundlePath, AIRFLOW_METADATA_FLAG], { + encoding: "utf-8", + }); + let manifest: BundleManifest; + try { + manifest = JSON.parse(stdout) as BundleManifest; + } catch (error) { + throw new Error(`Bundle produced invalid --airflow-metadata output: ${String(error)}`, { + cause: error, + }); + } + if (!manifest.supervisor_schema_version || typeof manifest.dags !== "object") { + throw new Error("Bundle produced incomplete --airflow-metadata output"); + } + return manifest; +} + +export async function runPack(argv: readonly string[]): Promise<void> { + const args = parsePackArgs(argv); + const bundlePath = path.join(args.outdir, BUNDLE_FILENAME); + + await build({ + entryPoints: [args.entry], + bundle: true, + platform: "node", + format: "esm", + target: "node22", + outfile: bundlePath, + }); + + const manifest = readBundleManifest(bundlePath); + if (Object.keys(manifest.dags).length === 0) { Review Comment: **[minor] A metadata-less `bundle.mjs` is left behind on the empty-registry error path.** `build()` writes `bundle.mjs` before the registry is validated; when no tasks are registered, the throw below leaves a marker-less bundle in `outdir`. A later coordinator scan finds it, sees no embedded marker, and silently falls back to a sidecar. Consider writing the final artifact atomically (temp file + rename) only after validation succeeds. ########## task-sdk/src/airflow/sdk/coordinators/node/coordinator.py: ########## @@ -45,6 +46,34 @@ BUNDLE_FILENAME = "bundle.mjs" METADATA_FILENAME = "airflow-metadata.yaml" +EMBEDDED_METADATA_MARKER = b"//# airflowMetadata=" +# Metadata sits on the bundle's first line; a bounded read is enough. +EMBEDDED_METADATA_HEAD_BYTES = 1 << 20 + + +def _read_embedded_metadata(bundle_path: pathlib.Path) -> dict[str, Any] | None: + """ + Read the manifest ``airflow-ts-pack`` embeds in the bundle itself. + + The packer prepends the ``airflow-metadata.yaml`` content as a leading + ``//# airflowMetadata=<base64>`` line comment, keeping bundle and metadata + a single artifact. Returns ``None`` when the bundle has no such marker. + """ + try: + with bundle_path.open("rb") as bundle_file: + head = bundle_file.read(EMBEDDED_METADATA_HEAD_BYTES) + except OSError as exc: + raise ValueError(f"cannot read {bundle_path.name}: {exc}") from exc + + if not head.startswith(EMBEDDED_METADATA_MARKER): + return None Review Comment: Do you think it's a good idea to introduce the trailer like Go-SDK to support the integrity validation and support viewing the "code" (the entrypoint .ts / .js file) for the further pure TS Dag feature? https://github.com/apache/airflow/blob/dc5ddd26202f972f634a6ee9ce25f04f17669ad7/task-sdk/docs/executable-bundle-spec.rst#L113-L139 The trailer could still be at the top of the .mjs prefix with the comment and point the the start and end position for each section. ########## ts-sdk/src/cli/pack.ts: ########## @@ -0,0 +1,167 @@ +/*! + * 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. + */ + +// 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. +// +// Mirrors airflow-go-pack: build first, then run the built bundle with +// --airflow-metadata so the manifest comes from the bundle's own task +// registry and schema version, never from a hand-written sidecar. + +import { execFileSync } from "node:child_process"; +import { readFileSync, writeFileSync } from "node:fs"; +import path from "node:path"; + +import { build } from "esbuild"; + +import { AIRFLOW_METADATA_FLAG, type BundleManifest } from "../coordinator/runtime.js"; + +const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0"; +const BUNDLE_FILENAME = "bundle.mjs"; +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. + +Options: + --outdir <dir> Output directory (default: dist) + --source <name> Display name of the primary source file (default: <entry> basename) +`; + +export interface PackArgs { + entry: string; + outdir: string; + source: string; +} + +export function parsePackArgs(argv: readonly string[]): PackArgs { + let entry: string | null = null; + let outdir = "dist"; + let source: string | null = null; + for (let i = 0; i < argv.length; i += 1) { + const arg = argv[i]!; + if (arg === "--outdir" || arg === "--source") { + const value = argv[i + 1]; + if (!value) throw new Error(`${arg} requires a value\n\n${USAGE}`); + if (arg === "--outdir") outdir = value; + else source = value; + i += 1; + } else if (arg.startsWith("-")) { + throw new Error(`Unknown option ${arg}\n\n${USAGE}`); + } else if (entry) { + throw new Error(`Unexpected argument ${arg}\n\n${USAGE}`); + } else { + entry = arg; + } + } + if (!entry) throw new Error(`Missing entry file\n\n${USAGE}`); + 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 }; + return version; +} + +function readBundleManifest(bundlePath: string): BundleManifest { + const stdout = execFileSync(process.execPath, [bundlePath, AIRFLOW_METADATA_FLAG], { + encoding: "utf-8", + }); Review Comment: **[major] `readBundleManifest` is brittle for real task modules.** `execFileSync(node, [bundle, --airflow-metadata])` passes no `timeout` and `JSON.parse`s raw stdout. Any import-time `console.log` in user code or a dependency runs before `startCoordinator()` handles the flag, so it lands on stdout ahead of the manifest JSON and `JSON.parse` throws; a top-level `await` that hangs blocks `airflow-ts-pack` indefinitely. Consider isolating the manifest (sentinel-delimited line or a dedicated fd) and passing a `timeout`. **[minor]** Relatedly, this `execFileSync` sits outside the `try/catch` that wraps only `JSON.parse`, so a bundle that throws at module load surfaces a generic child-process error with no hint it came from running the user bundle — wrapping it with a clear prefix would help. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
