You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "pierrejeambrun (via GitHub)" <gi...@apache.org> on 2026/09/25 13:29:49 UTC
[PR] TS SDK: embed one source region per native Dag file [airflow]
pierrejeambrun opened a new pull request, #73723:
URL: https://github.com/apache/airflow/pull/73723
A bundle whose native Dags live in more than one file only carried its
entrypoint's source in the `airflowSource` region. Airflow's Code tab
would show the entrypoint for every Dag: the file for the ones declared
there, and something unrelated for the ones declared in imported
modules. Multi-Dag bundles were effectively single-source in the UI.
## What changes
- **Each Dag's source file is embedded**, one per unique path. The
bundle header's single `source` region becomes `sources: [{path,
start, end, sha256}]`. Each region has its own `/*# airflowSource:<path> …
#*/` block comment so the path is visible in the raw bundle too.
- **Metadata gains `dag_source_paths: Record<dag_id, path>`** so a
reader can pick the right source per Dag without guessing.
- **`Dag`'s constructor records which file it ran in**. It reads a
cross-realm global slot (`Symbol.for("airflow.ts-sdk.current-module-source")`)
that `airflow-ts-pack`'s new onLoad plugin writes at the top of each
author-owned source file. ES module hoisting is what makes the slot
the current file for every top-level `new Dag(...)` call, so the
constructor just reads it. `Function.prototype.toString` and stack
parsing stay out. Same `Symbol.for` pattern the SDK already uses for
the task-scope AsyncLocalStorage.
- **Files that only supply utilities or types are not embedded.** Only
the files whose `new Dag(...)` constructor ran end up in the source
regions — matches Python's `dag.fileloc` behavior.
- **Mixed-language bundles** (Python owns the Dag, TS supplies handlers)
carry no source regions; `dag_source_paths` is empty. Their Python
side is what displays in the UI.
## Reproducing before/after
```
src/main.ts # declares `orders_dag`, imports and registers reports_dag
src/dags/reports.ts # declares `reports_dag`
```
Before: bundle carries only `main.ts` in the source region; `reports_dag`
has no source for the Code tab.
After: bundle carries both files as separate regions; metadata has
`dag_source_paths: {orders_dag: "src/main.ts", reports_dag: "src/dags/reports.ts"}`.
Verified end-to-end on a pack of that layout — both files land in the
header's `sources` list with their own byte ranges and digests.
## Test coverage
- `buildBundleManifest` records paths from the module-source slot.
- Encoder: one region per file, path-tagged marker, empty case (no
sources) for pure mixed-lang bundles.
- Pack: multi-Dag bundle where two files declare Dags across an import
boundary — asserts both sources are embedded and `dag_source_paths`
maps each `dag_id` to its file.
- Existing shebang test still passes: the plugin's slot-write goes
after the shebang line so `#!/usr/bin/env node` stays valid.
## Not in scope
- Server-side importer for `.min.mjs` bundles is future work; nothing
reads `dag_source_paths` yet on the Airflow side. This PR only fixes
the pack-time embedding shape so the data is present when the
importer lands.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 4.7)
Generated-by: Claude Code (Opus 4.7) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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] TS SDK: embed one source region per native Dag file [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #73723:
URL: https://github.com/apache/airflow/pull/73723#discussion_r4105491744
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -42,27 +44,45 @@ import type { BundleManifest } from "../coordinator/manifest.js";
const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
const EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024;
-const EMBEDDED_SOURCE_MAX_BYTES = 1024 * 1024;
+const EMBEDDED_SOURCES_MAX_BYTES = 4 * 1024 * 1024;
const OFFSET_HEX_WIDTH = 16;
export const EMBEDDED_METADATA_PREFIX = "//# airflowMetadata=";
export const EMBEDDED_LAYOUT_PREFIX = "//# airflowBundle=";
-/** A block comment, because the entrypoint spans more than one line. */
-export const EMBEDDED_SOURCE_OPEN = "/*# airflowSource\n";
+/**
+ * Each source is wrapped in a block comment. The path follows the marker so
+ * a reader can tell one region from another without cross-referencing the
+ * header — the header's byte ranges are the source of truth, but a human
+ * scanning the bundle can still find their file by name.
+ */
+export const EMBEDDED_SOURCE_OPEN_PREFIX = "/*# airflowSource:";
+export const EMBEDDED_SOURCE_OPEN_SUFFIX = "\n";
export const EMBEDDED_SOURCE_CLOSE = "\n#*/\n";
export interface BundleEncoderInput {
bundleManifest: BundleManifest;
sdkVersion: string;
+ /** The entrypoint the packer was invoked on; recorded in metadata so a reader can
+ * point at the primary source file when there is no per-Dag source path to use. */
entrypointName: string;
- entrypointSource: string;
+ /**
+ * Every author-owned source file that declares at least one Dag, keyed by
+ * the path used in `BundleManifest.dag_source_paths`. Files that only
+ * import from these (utilities, types) are not embedded — the Code tab
+ * reads what defines each Dag, not what it depends on.
+ *
+ * A bundle whose Dags are all task handlers (Python owns them) may have
+ * an empty map here.
+ */
+ entrypointSources: Record<string, string>;
executable: Uint8Array;
}
interface BundleMetadata {
airflow_bundle_metadata_version: string;
sdk: { language: string; version: string; supervisor_schema_version: string };
- source: string;
+ entrypoint: string;
+ dag_source_paths: BundleManifest["dag_source_paths"];
Review Comment:
Deliberate — the SDK is at 0.1.0-beta1 and the bundle format has no shipped consumers yet. The reader-side changes for these fields land alongside the importer PR. Bumping the metadata version before there's a reader that has to distinguish old from new would be back-compat theater.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -72,65 +92,86 @@ interface VerifiedByteRange {
start: string;
}
+interface VerifiedSourceRegion extends VerifiedByteRange {
+ path: string;
+}
+
interface BundleHeader {
code: VerifiedByteRange;
metadata: VerifiedByteRange;
- source: VerifiedByteRange;
+ sources: VerifiedSourceRegion[];
+}
+
+/** Per-source-region metadata within the concatenated sources buffer. */
+interface SourceRegion {
+ path: string;
+ payloadStart: number;
+ payloadEnd: number;
+ sha256: string;
+}
+
+interface EncodedSources {
+ buffer: Buffer;
+ regions: SourceRegion[];
}
export function encodeBundle(input: BundleEncoderInput): Buffer {
const metadata = encodeMetadata(input);
- const source = encodeSource(input.entrypointSource);
+ const sources = encodeSources(input.entrypointSources);
const executable = encodeExecutable(input.executable);
- const header = encodeHeader({ metadata, source, executable });
+ const header = encodeHeader({ metadata, sources, executable });
- return Buffer.concat([header, metadata, source, executable]);
+ return Buffer.concat([header, metadata, sources.buffer, executable]);
}
-function encodeHeader(regions: { metadata: Buffer; source: Buffer; executable: Buffer }): Buffer {
+function encodeHeader(regions: {
+ metadata: Buffer;
+ sources: EncodedSources;
+ executable: Buffer;
+}): Buffer {
// Each digest covers the payload only. The framing markers and newlines are re-derived.
const metadataPayload = regions.metadata.subarray(
Buffer.byteLength(EMBEDDED_METADATA_PREFIX),
-1,
);
- const sourcePayload = regions.source.subarray(
- Buffer.byteLength(EMBEDDED_SOURCE_OPEN),
- -Buffer.byteLength(EMBEDDED_SOURCE_CLOSE),
- );
- const digests = {
- code: computeSha256(regions.executable),
- metadata: computeSha256(metadataPayload),
- source: computeSha256(sourcePayload),
- };
+ const metadataDigest = computeSha256(metadataPayload);
+ const codeDigest = computeSha256(regions.executable);
const zeroOffset = "0".repeat(OFFSET_HEX_WIDTH);
+ // Placeholder header with zeroed offsets and real digests, to measure its
+ // length without recursing. Every source path is present, so the array
+ // length is what it will be in the final header.
const placeholderHeader = renderHeader({
- code: { start: zeroOffset, end: zeroOffset, sha256: digests.code },
- metadata: { start: zeroOffset, end: zeroOffset, sha256: digests.metadata },
- source: { start: zeroOffset, end: zeroOffset, sha256: digests.source },
+ code: { start: zeroOffset, end: zeroOffset, sha256: codeDigest },
+ metadata: { start: zeroOffset, end: zeroOffset, sha256: metadataDigest },
+ sources: regions.sources.regions.map((region) => ({
+ path: region.path,
+ start: zeroOffset,
+ end: zeroOffset,
+ sha256: region.sha256,
+ })),
});
const metadataStart = placeholderHeader.length + Buffer.byteLength(EMBEDDED_METADATA_PREFIX);
const metadataEnd = metadataStart + metadataPayload.length;
- const sourceStart =
- placeholderHeader.length + regions.metadata.length + Buffer.byteLength(EMBEDDED_SOURCE_OPEN);
- const sourceEnd = sourceStart + sourcePayload.length;
- const codeStart = placeholderHeader.length + regions.metadata.length + regions.source.length;
+ const sourcesBaseOffset = placeholderHeader.length + regions.metadata.length;
+ const codeStart = sourcesBaseOffset + regions.sources.buffer.length;
const codeEnd = codeStart + regions.executable.length;
const header = renderHeader({
code: {
start: formatOffset(codeStart),
end: formatOffset(codeEnd),
- sha256: digests.code,
+ sha256: codeDigest,
},
metadata: {
start: formatOffset(metadataStart),
end: formatOffset(metadataEnd),
- sha256: digests.metadata,
- },
- source: {
- start: formatOffset(sourceStart),
- end: formatOffset(sourceEnd),
- sha256: digests.source,
+ sha256: metadataDigest,
},
+ sources: regions.sources.regions.map((region) => ({
+ path: region.path,
+ start: formatOffset(sourcesBaseOffset + region.payloadStart),
+ end: formatOffset(sourcesBaseOffset + region.payloadEnd),
+ sha256: region.sha256,
+ })),
Review Comment:
Fixed in 709a42f1c30 — the header is now encoded as utf-8. A non-ASCII path in `sources[i].path` round-trips correctly.
--
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] TS SDK: embed one source region per native Dag file [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #73723:
URL: https://github.com/apache/airflow/pull/73723#discussion_r4105441642
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -42,27 +44,45 @@ import type { BundleManifest } from "../coordinator/manifest.js";
const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
const EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024;
-const EMBEDDED_SOURCE_MAX_BYTES = 1024 * 1024;
+const EMBEDDED_SOURCES_MAX_BYTES = 4 * 1024 * 1024;
const OFFSET_HEX_WIDTH = 16;
export const EMBEDDED_METADATA_PREFIX = "//# airflowMetadata=";
export const EMBEDDED_LAYOUT_PREFIX = "//# airflowBundle=";
-/** A block comment, because the entrypoint spans more than one line. */
-export const EMBEDDED_SOURCE_OPEN = "/*# airflowSource\n";
+/**
+ * Each source is wrapped in a block comment. The path follows the marker so
+ * a reader can tell one region from another without cross-referencing the
+ * header — the header's byte ranges are the source of truth, but a human
+ * scanning the bundle can still find their file by name.
+ */
+export const EMBEDDED_SOURCE_OPEN_PREFIX = "/*# airflowSource:";
+export const EMBEDDED_SOURCE_OPEN_SUFFIX = "\n";
export const EMBEDDED_SOURCE_CLOSE = "\n#*/\n";
export interface BundleEncoderInput {
bundleManifest: BundleManifest;
sdkVersion: string;
+ /** The entrypoint the packer was invoked on; recorded in metadata so a reader can
+ * point at the primary source file when there is no per-Dag source path to use. */
entrypointName: string;
- entrypointSource: string;
+ /**
+ * Every author-owned source file that declares at least one Dag, keyed by
+ * the path used in `BundleManifest.dag_source_paths`. Files that only
+ * import from these (utilities, types) are not embedded — the Code tab
+ * reads what defines each Dag, not what it depends on.
+ *
+ * A bundle whose Dags are all task handlers (Python owns them) may have
+ * an empty map here.
+ */
+ entrypointSources: Record<string, string>;
executable: Uint8Array;
}
interface BundleMetadata {
airflow_bundle_metadata_version: string;
sdk: { language: string; version: string; supervisor_schema_version: string };
- source: string;
+ entrypoint: string;
+ dag_source_paths: BundleManifest["dag_source_paths"];
Review Comment:
NOt released yet
--
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] TS SDK: embed one source region per native Dag file [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #73723:
URL: https://github.com/apache/airflow/pull/73723#discussion_r4105489050
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -139,26 +180,47 @@ function encodeHeader(regions: { metadata: Buffer; source: Buffer; executable: B
}
/**
- * Wrap the entrypoint as written in a block comment.
+ * Wrap each source in its own block comment, one per author-owned Dag file.
*
* A comment terminator would splice the rest of the payload into executable position, so it is
* escaped. `*\\` is escaped too, which keeps the transformation reversible.
+ *
+ * The returned regions are keyed by path and hold byte offsets within the
+ * concatenated sources buffer (not the final bundle); the header adds the
+ * bundle-relative base offset when it renders.
*/
-function encodeSource(entrypointSource: string): Buffer {
- const payload = Buffer.from(escapeBlockComment(entrypointSource), "utf-8");
- const source = Buffer.concat([
- Buffer.from(EMBEDDED_SOURCE_OPEN, "ascii"),
- payload,
- Buffer.from(EMBEDDED_SOURCE_CLOSE, "ascii"),
- ]);
- if (source.length > EMBEDDED_SOURCE_MAX_BYTES) {
+function encodeSources(sources: Record<string, string>): EncodedSources {
+ const chunks: Buffer[] = [];
+ const regions: SourceRegion[] = [];
+ let offset = 0;
+
+ for (const [path, content] of Object.entries(sources)) {
+ const openMarker = `${EMBEDDED_SOURCE_OPEN_PREFIX}${path}${EMBEDDED_SOURCE_OPEN_SUFFIX}`;
Review Comment:
Fixed in 709a42f1c30 — a source path containing `*/` (or a newline) now rejects at pack time. Real filesystem paths don't hold either, and rejecting keeps the marker line trivially readable and offsets valid.
##########
ts-sdk/src/cli/pack.ts:
##########
@@ -202,11 +216,82 @@ async function loadEsbuild(): Promise<typeof import("esbuild")> {
}
}
+/** Author source files esbuild's onLoad tags with their path. Same filter set the previous
+ * task-id plugin used, which matched every language a Dag can be declared in. */
+const AUTHOR_SOURCE_FILTER = /\.[cm]?[jt]sx?$/;
+
+/** Skip anything under node_modules: an installed dependency is not a Dag file, and tagging it
+ * would only cost bundle bytes while overwriting the slot with the wrong path. */
+const DEPENDENCY_PATH = /[\\/]node_modules[\\/]/;
+
+/**
+ * esbuild plugin: prepend a single line to each author-owned source file that
+ * writes the file's path into the SDK's module-source slot. `Dag`'s
+ * constructor reads that slot, so a Dag declared in `src/dags/reports.ts`
+ * carries that path even though esbuild will soon inline every module into
+ * one bundle.
+ *
+ * The prepend is text-only — no parsing, no AST — so it can never mis-identify
+ * a construction site. ES modules hoist imports above non-import statements
+ * at execution, so the tag runs *after* imported modules have written their
+ * own slots and *before* this module's own top-level statements, which is
+ * exactly when its `new Dag(...)` calls fire.
+ */
+function moduleSourceTagPlugin(cwd: string): import("esbuild").Plugin {
+ const slotKey = JSON.stringify(MODULE_SOURCE_SLOT_KEY);
+ return {
+ name: "airflow-module-source-tag",
+ setup(build) {
+ build.onLoad({ filter: AUTHOR_SOURCE_FILTER }, async ({ path: file, namespace }) => {
+ // Only the `file` namespace has a path on disk to attribute a Dag to;
+ // a virtual module from another plugin has nothing to tag.
+ if (namespace !== "file" || DEPENDENCY_PATH.test(file)) return undefined;
+ const source = await readFileAsync(file, "utf-8");
+ // Project-relative so the bundle is portable and readable (no host
+ // filesystem prefix), matching how esbuild's own metafile keys sources.
+ const relative = path.relative(cwd, file);
+ const tag = `globalThis[Symbol.for(${slotKey})]=${JSON.stringify(relative)};\n`;
Review Comment:
Partly addressed in 9c0fb20fea7 — the tag now saves the previous slot value on entry and restores it after the module's top-level code returns. Nested imports (A imports B) and top-level `await` within a single module hand the slot back on the way out, so a `new Dag(...)` after the import is still attributed to A.
Two limits remain and are documented in module-source.ts: (a) a Dag constructed in a scheduled callback (setTimeout, microtask, .then) after the module's epilog has restored lands under whichever module touched the slot most recently; (b) two modules genuinely evaluated in parallel through `Promise.all([import, import])` can interleave. Both are non-standard patterns for Dag files; the documented invariant is 'declare Dags at module top level'.
--
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] TS SDK: embed one source region per native Dag file [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #73723:
URL: https://github.com/apache/airflow/pull/73723#discussion_r4105494100
##########
ts-sdk/src/cli/pack.ts:
##########
@@ -215,8 +300,15 @@ export async function runPack(argv: readonly string[]): Promise<void> {
platform: "node",
format: "esm",
target: "node22",
- // A digest is only worth taking over an artifact nobody reads or edits in place.
- minify: true,
+ // Whitespace and syntax only. Identifiers stay: `dag.task(handler)` takes
+ // the task id from the handler's name at run time, and `keepNames` carries
+ // that name through any bundler-collision renames.
Review Comment:
Fixed in 709a42f1c30 — the three-flag minify split reverted to plain `minify: true`, and the rationale comment is gone. Nothing in this PR depends on identifier preservation; the earlier split had been kept in anticipation of a follow-up PR that isn't part of this diff.
--
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] TS SDK: embed one source region per native Dag file [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #73723:
URL: https://github.com/apache/airflow/pull/73723#discussion_r4105291977
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -139,26 +180,47 @@ function encodeHeader(regions: { metadata: Buffer; source: Buffer; executable: B
}
/**
- * Wrap the entrypoint as written in a block comment.
+ * Wrap each source in its own block comment, one per author-owned Dag file.
*
* A comment terminator would splice the rest of the payload into executable position, so it is
* escaped. `*\\` is escaped too, which keeps the transformation reversible.
+ *
+ * The returned regions are keyed by path and hold byte offsets within the
+ * concatenated sources buffer (not the final bundle); the header adds the
+ * bundle-relative base offset when it renders.
*/
-function encodeSource(entrypointSource: string): Buffer {
- const payload = Buffer.from(escapeBlockComment(entrypointSource), "utf-8");
- const source = Buffer.concat([
- Buffer.from(EMBEDDED_SOURCE_OPEN, "ascii"),
- payload,
- Buffer.from(EMBEDDED_SOURCE_CLOSE, "ascii"),
- ]);
- if (source.length > EMBEDDED_SOURCE_MAX_BYTES) {
+function encodeSources(sources: Record<string, string>): EncodedSources {
+ const chunks: Buffer[] = [];
+ const regions: SourceRegion[] = [];
+ let offset = 0;
+
+ for (const [path, content] of Object.entries(sources)) {
+ const openMarker = `${EMBEDDED_SOURCE_OPEN_PREFIX}${path}${EMBEDDED_SOURCE_OPEN_SUFFIX}`;
Review Comment:
The source path is interpolated directly into a block-comment opener, unlike the source payload below. A valid path such as `dir*/reports.ts` therefore emits `/*# airflowSource:dir*/reports.ts`, closes the comment early, and makes the declared source/code offsets invalid. Escape the path's block-comment terminators (or reject such paths) before building the marker.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -42,27 +44,45 @@ import type { BundleManifest } from "../coordinator/manifest.js";
const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
const EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024;
-const EMBEDDED_SOURCE_MAX_BYTES = 1024 * 1024;
+const EMBEDDED_SOURCES_MAX_BYTES = 4 * 1024 * 1024;
Review Comment:
The encoder now permits up to 4 MiB of source regions, while the current task-sdk reader still rejects any source payload over 1 MiB (`task-sdk/src/airflow/sdk/coordinators/node/_bundle_reader.py:50,430-432`). After the framing change is made readable, a bundle between these limits will pack successfully and then be rejected by `NodeCoordinator`; keep the producer and consumer limits aligned or version the contract together.
This issue also appears on line 102 of the same file.
##########
ts-sdk/src/cli/pack.ts:
##########
@@ -215,8 +300,15 @@ export async function runPack(argv: readonly string[]): Promise<void> {
platform: "node",
format: "esm",
target: "node22",
- // A digest is only worth taking over an artifact nobody reads or edits in place.
- minify: true,
+ // Whitespace and syntax only. Identifiers stay: `dag.task(handler)` takes
+ // the task id from the handler's name at run time, and `keepNames` carries
+ // that name through any bundler-collision renames.
Review Comment:
This rationale is inconsistent with the SDK API: `Dag.task` receives an explicit task ID, and `TaskHandler` documentation states that function names are not used for task identity. Please correct the explanation (or restore full minification if `keepNames` was added only for this assumption), because the current comment misleadingly justifies a bundle-size change.
##########
ts-sdk/src/cli/pack.ts:
##########
@@ -202,11 +216,82 @@ async function loadEsbuild(): Promise<typeof import("esbuild")> {
}
}
+/** Author source files esbuild's onLoad tags with their path. Same filter set the previous
+ * task-id plugin used, which matched every language a Dag can be declared in. */
+const AUTHOR_SOURCE_FILTER = /\.[cm]?[jt]sx?$/;
+
+/** Skip anything under node_modules: an installed dependency is not a Dag file, and tagging it
+ * would only cost bundle bytes while overwriting the slot with the wrong path. */
+const DEPENDENCY_PATH = /[\\/]node_modules[\\/]/;
+
+/**
+ * esbuild plugin: prepend a single line to each author-owned source file that
+ * writes the file's path into the SDK's module-source slot. `Dag`'s
+ * constructor reads that slot, so a Dag declared in `src/dags/reports.ts`
+ * carries that path even though esbuild will soon inline every module into
+ * one bundle.
+ *
+ * The prepend is text-only — no parsing, no AST — so it can never mis-identify
+ * a construction site. ES modules hoist imports above non-import statements
+ * at execution, so the tag runs *after* imported modules have written their
+ * own slots and *before* this module's own top-level statements, which is
+ * exactly when its `new Dag(...)` calls fire.
+ */
+function moduleSourceTagPlugin(cwd: string): import("esbuild").Plugin {
+ const slotKey = JSON.stringify(MODULE_SOURCE_SLOT_KEY);
+ return {
+ name: "airflow-module-source-tag",
+ setup(build) {
+ build.onLoad({ filter: AUTHOR_SOURCE_FILTER }, async ({ path: file, namespace }) => {
+ // Only the `file` namespace has a path on disk to attribute a Dag to;
+ // a virtual module from another plugin has nothing to tag.
+ if (namespace !== "file" || DEPENDENCY_PATH.test(file)) return undefined;
+ const source = await readFileAsync(file, "utf-8");
+ // Project-relative so the bundle is portable and readable (no host
+ // filesystem prefix), matching how esbuild's own metafile keys sources.
+ const relative = path.relative(cwd, file);
+ const tag = `globalThis[Symbol.for(${slotKey})]=${JSON.stringify(relative)};\n`;
Review Comment:
The slot is a mutable process-global value, so it is not scoped to this module's evaluation. If this module reaches a top-level `await`, or another module evaluates while it is suspended, that module can overwrite the slot before a later `new Dag(...)` runs; the resulting `dag_source_paths` then points to an unrelated file. The current integration test covers only synchronous constructors, so this needs a module-scoped capture/transform or an explicit restriction on constructors after suspension.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -72,65 +92,86 @@ interface VerifiedByteRange {
start: string;
}
+interface VerifiedSourceRegion extends VerifiedByteRange {
+ path: string;
+}
+
interface BundleHeader {
code: VerifiedByteRange;
metadata: VerifiedByteRange;
- source: VerifiedByteRange;
+ sources: VerifiedSourceRegion[];
+}
+
+/** Per-source-region metadata within the concatenated sources buffer. */
+interface SourceRegion {
+ path: string;
+ payloadStart: number;
+ payloadEnd: number;
+ sha256: string;
+}
+
+interface EncodedSources {
+ buffer: Buffer;
+ regions: SourceRegion[];
}
export function encodeBundle(input: BundleEncoderInput): Buffer {
const metadata = encodeMetadata(input);
- const source = encodeSource(input.entrypointSource);
+ const sources = encodeSources(input.entrypointSources);
const executable = encodeExecutable(input.executable);
- const header = encodeHeader({ metadata, source, executable });
+ const header = encodeHeader({ metadata, sources, executable });
- return Buffer.concat([header, metadata, source, executable]);
+ return Buffer.concat([header, metadata, sources.buffer, executable]);
}
-function encodeHeader(regions: { metadata: Buffer; source: Buffer; executable: Buffer }): Buffer {
+function encodeHeader(regions: {
+ metadata: Buffer;
+ sources: EncodedSources;
+ executable: Buffer;
+}): Buffer {
// Each digest covers the payload only. The framing markers and newlines are re-derived.
const metadataPayload = regions.metadata.subarray(
Buffer.byteLength(EMBEDDED_METADATA_PREFIX),
-1,
);
- const sourcePayload = regions.source.subarray(
- Buffer.byteLength(EMBEDDED_SOURCE_OPEN),
- -Buffer.byteLength(EMBEDDED_SOURCE_CLOSE),
- );
- const digests = {
- code: computeSha256(regions.executable),
- metadata: computeSha256(metadataPayload),
- source: computeSha256(sourcePayload),
- };
+ const metadataDigest = computeSha256(metadataPayload);
+ const codeDigest = computeSha256(regions.executable);
const zeroOffset = "0".repeat(OFFSET_HEX_WIDTH);
+ // Placeholder header with zeroed offsets and real digests, to measure its
+ // length without recursing. Every source path is present, so the array
+ // length is what it will be in the final header.
const placeholderHeader = renderHeader({
- code: { start: zeroOffset, end: zeroOffset, sha256: digests.code },
- metadata: { start: zeroOffset, end: zeroOffset, sha256: digests.metadata },
- source: { start: zeroOffset, end: zeroOffset, sha256: digests.source },
+ code: { start: zeroOffset, end: zeroOffset, sha256: codeDigest },
+ metadata: { start: zeroOffset, end: zeroOffset, sha256: metadataDigest },
+ sources: regions.sources.regions.map((region) => ({
+ path: region.path,
+ start: zeroOffset,
+ end: zeroOffset,
+ sha256: region.sha256,
+ })),
});
const metadataStart = placeholderHeader.length + Buffer.byteLength(EMBEDDED_METADATA_PREFIX);
const metadataEnd = metadataStart + metadataPayload.length;
- const sourceStart =
- placeholderHeader.length + regions.metadata.length + Buffer.byteLength(EMBEDDED_SOURCE_OPEN);
- const sourceEnd = sourceStart + sourcePayload.length;
- const codeStart = placeholderHeader.length + regions.metadata.length + regions.source.length;
+ const sourcesBaseOffset = placeholderHeader.length + regions.metadata.length;
+ const codeStart = sourcesBaseOffset + regions.sources.buffer.length;
const codeEnd = codeStart + regions.executable.length;
const header = renderHeader({
code: {
start: formatOffset(codeStart),
end: formatOffset(codeEnd),
- sha256: digests.code,
+ sha256: codeDigest,
},
metadata: {
start: formatOffset(metadataStart),
end: formatOffset(metadataEnd),
- sha256: digests.metadata,
- },
- source: {
- start: formatOffset(sourceStart),
- end: formatOffset(sourceEnd),
- sha256: digests.source,
+ sha256: metadataDigest,
},
+ sources: regions.sources.regions.map((region) => ({
+ path: region.path,
+ start: formatOffset(sourcesBaseOffset + region.payloadStart),
+ end: formatOffset(sourcesBaseOffset + region.payloadEnd),
+ sha256: region.sha256,
+ })),
Review Comment:
Source paths are now serialized into the layout header, but `renderHeader` still creates that header with `Buffer.from(..., "ascii")`. A valid non-ASCII filename such as `données.ts` is therefore silently corrupted in the header, so the metadata-to-region path mapping no longer round-trips. Encode the header as UTF-8 or escape paths to ASCII before rendering it.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -42,27 +44,45 @@ import type { BundleManifest } from "../coordinator/manifest.js";
const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
const EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024;
-const EMBEDDED_SOURCE_MAX_BYTES = 1024 * 1024;
+const EMBEDDED_SOURCES_MAX_BYTES = 4 * 1024 * 1024;
const OFFSET_HEX_WIDTH = 16;
export const EMBEDDED_METADATA_PREFIX = "//# airflowMetadata=";
export const EMBEDDED_LAYOUT_PREFIX = "//# airflowBundle=";
-/** A block comment, because the entrypoint spans more than one line. */
-export const EMBEDDED_SOURCE_OPEN = "/*# airflowSource\n";
+/**
+ * Each source is wrapped in a block comment. The path follows the marker so
+ * a reader can tell one region from another without cross-referencing the
+ * header — the header's byte ranges are the source of truth, but a human
+ * scanning the bundle can still find their file by name.
+ */
+export const EMBEDDED_SOURCE_OPEN_PREFIX = "/*# airflowSource:";
+export const EMBEDDED_SOURCE_OPEN_SUFFIX = "\n";
export const EMBEDDED_SOURCE_CLOSE = "\n#*/\n";
export interface BundleEncoderInput {
bundleManifest: BundleManifest;
sdkVersion: string;
+ /** The entrypoint the packer was invoked on; recorded in metadata so a reader can
+ * point at the primary source file when there is no per-Dag source path to use. */
entrypointName: string;
- entrypointSource: string;
+ /**
+ * Every author-owned source file that declares at least one Dag, keyed by
+ * the path used in `BundleManifest.dag_source_paths`. Files that only
+ * import from these (utilities, types) are not embedded — the Code tab
+ * reads what defines each Dag, not what it depends on.
+ *
+ * A bundle whose Dags are all task handlers (Python owns them) may have
+ * an empty map here.
+ */
+ entrypointSources: Record<string, string>;
executable: Uint8Array;
}
interface BundleMetadata {
airflow_bundle_metadata_version: string;
sdk: { language: string; version: string; supervisor_schema_version: string };
- source: string;
+ entrypoint: string;
+ dag_source_paths: BundleManifest["dag_source_paths"];
Review Comment:
The bundle contract is changed incompatibly (`source` becomes `entrypoint`, and one source section becomes `sources`) while `AIRFLOW_BUNDLE_METADATA_VERSION` remains `"1.0"`. Once the reader is updated, it will either reject all existing v1 bundles or need to branch on a version that is currently indistinguishable; bump the bundle metadata version or implement explicit backward compatibility.
--
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] TS SDK: embed one source region per native Dag file [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #73723:
URL: https://github.com/apache/airflow/pull/73723#discussion_r4105461008
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -42,27 +44,45 @@ import type { BundleManifest } from "../coordinator/manifest.js";
const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
const EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024;
-const EMBEDDED_SOURCE_MAX_BYTES = 1024 * 1024;
+const EMBEDDED_SOURCES_MAX_BYTES = 4 * 1024 * 1024;
Review Comment:
Stale — the per-file cap is back to 1 MiB in b4e803dbb5b (before this review); the reader-side 1 MiB limit still holds.
--
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] TS SDK: embed one source region per native Dag file [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #73723:
URL: https://github.com/apache/airflow/pull/73723
--
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