This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new b5d9259f4f5 TS SDK: replace DagRegistry with Bundle, and serveDags
with bundle.serve (#73187)
b5d9259f4f5 is described below
commit b5d9259f4f5766bb41eaf37dab3a254e33051e1f
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Thu Sep 17 11:40:00 2026 +0800
TS SDK: replace DagRegistry with Bundle, and serveDags with bundle.serve
(#73187)
A bundle process was described entirely in terms of Dags: `DagRegistry` held
them, `serveDags(registry)` served them, and the pack error told an author
to
pass them to `serveDags(new DagRegistry(...))`. A mixed-language bundle
provides task handlers for a Dag that Python owns and registers no Dag at
all,
so all three are about to be wrong.
The collection becomes a `Bundle`, and it serves itself.
`register(...items: Registerable[])` is its one registration verb, over a
union
that gains an arm per kind rather than a second verb per kind. Today the
union
has one arm, `Dag`, and registering keeps the path it already had.
`bundle.serve()` replaces the free function. The existing intent was that
Dag
authors reach the runtime through one call and never name the coordinator; a
method on the object that already holds everything keeps that while dropping
the function, and the coordinator subpath now exports no entry point at all.
The latch that makes a serve one-shot moves to `coordinator/serve.ts`,
beside
the sockets it protects rather than on the bundle that holds none.
The guard `serveDags` performed on its argument survives as a guard on the
receiver, which is where it still has something to catch: `const { serve } =
bundle` detaches the method, and a bundle from a second resolved copy is
reported by its cause rather than as a missing private field.
`Bundle.serve()` reaches `serveBundle` through an inline `await
import(...)`.
The coordinator reads a bundle to dispatch and to enumerate it, so a static
import in the other direction would leave the authoring surface and the
coordinator mutually dependent. The module graph stays acyclic instead.
`tests/cli/fixtures/bundle-v1.mjs` is regenerated. That golden bundle embeds
the byte ranges and SHA-256 digest of `empty-entry.ts`, which this change
rewrites onto the new surface, so its header moves with it.
---
.../language-sdks/typescript.rst | 25 +--
ts-sdk/README.md | 40 ++---
ts-sdk/api-docs/dag-authoring-api.ts | 3 +-
ts-sdk/docs/index.md | 8 +-
ts-sdk/example/src/main.ts | 6 +-
ts-sdk/scripts/verify-package.mjs | 2 +-
ts-sdk/src/cli/pack.ts | 4 +-
ts-sdk/src/coordinator/index.ts | 6 +-
ts-sdk/src/coordinator/manifest.ts | 6 +-
ts-sdk/src/coordinator/runtime.ts | 87 ++--------
ts-sdk/src/coordinator/serve.ts | 61 +++++++
ts-sdk/src/index.ts | 5 +-
ts-sdk/src/sdk/brand.ts | 4 +-
ts-sdk/src/sdk/bundle.ts | 176 +++++++++++++++++++++
ts-sdk/src/sdk/dag.ts | 2 +-
ts-sdk/src/sdk/registry.ts | 124 ---------------
ts-sdk/tests/cli/fixtures/bundle-v1.mjs | 6 +-
ts-sdk/tests/cli/fixtures/empty-entry.ts | 4 +-
ts-sdk/tests/cli/fixtures/entry.ts | 4 +-
ts-sdk/tests/cli/fixtures/noisy-entry.ts | 4 +-
ts-sdk/tests/cli/pack.test.ts | 20 +--
ts-sdk/tests/coordinator/integration.test.ts | 14 +-
ts-sdk/tests/coordinator/protocol.test.ts | 6 +-
ts-sdk/tests/coordinator/public-api.test.ts | 13 +-
ts-sdk/tests/coordinator/runtime-manifest.test.ts | 26 +--
ts-sdk/tests/public-api.test.ts | 91 ++++++-----
.../tests/sdk/{registry.test.ts => bundle.test.ts} | 98 +++++++-----
ts-sdk/tests/sdk/dag.test.ts | 26 +--
28 files changed, 482 insertions(+), 389 deletions(-)
diff --git
a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
index a1d1137dd7f..9bfc50e0f67 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -37,7 +37,7 @@ The SDK is the ``apache-airflow-ts-sdk`` package (ESM-only).
It is currently in
.. seealso::
- For the full TypeScript API reference (``Dag``, ``DagRegistry``,
``serveDags``, the task handler getters,
+ For the full TypeScript API reference (``Dag``, ``Bundle``, the task handler
getters,
``TaskClient``, supporting types, and exceptions),
see the `TypeScript SDK API reference
<https://airflow.apache.org/docs/ts-sdk/stable/>`__.
@@ -97,12 +97,12 @@ A task is an ordinary (usually ``async``) function taking
no arguments:
``getContext()`` and ``getClient()`` reach the runtime from inside the call,
so nothing the SDK supplies is a parameter.
Create a ``Dag`` with the ``dag_id`` it implements, attach each handler with
``dag.task``,
-collect the Dags in a ``DagRegistry``, then serve them to Airflow with
``serveDags``.
+register it on a ``Bundle``, then serve it to Airflow with ``bundle.serve()``.
That top-level ``await`` makes the module a runnable bundle entry point.
.. code-block:: typescript
- import { Dag, DagRegistry, getClient, serveDags } from
"apache-airflow-ts-sdk";
+ import { Bundle, Dag, getClient } from "apache-airflow-ts-sdk";
export async function buildMessage() {
const client = getClient();
@@ -117,16 +117,19 @@ That top-level ``await`` makes the module a runnable
bundle entry point.
const dag = new Dag("typescript_example");
dag.task("build_message", buildMessage);
- await serveDags(new DagRegistry(dag));
+ const bundle = new Bundle();
+ bundle.register(dag);
+ await bundle.serve();
The ``dagId`` passed to ``new Dag(...)`` must match the ``dag_id`` of the
Python Dag, and each ``taskId``
-passed to ``dag.task`` must match a ``@task.stub`` function in that Dag. The
registry passed to
-``serveDags`` is the bundle's complete set of Dags; a second ``serveDags``
call is rejected. A Dag left out
-of the registry is not part of the packed bundle, and its tasks are marked
removed at runtime.
-
-``DagRegistry`` holds no sockets and starts nothing, so a unit test can build
one and dispatch a handler
-through ``registry.getTaskHandler(dagId, taskId)`` without a coordinator
runtime. A bundle that collects
-its Dags across several modules can add them incrementally with
``registry.register(...)``.
+passed to ``dag.task`` must match a ``@task.stub`` function in that Dag. What
the bundle holds is its
+complete set of Dags; a second ``bundle.serve()`` call is rejected. A Dag left
unregistered is not part of
+the packed bundle, and its tasks are marked removed at runtime.
+
+``register`` is the bundle's one registration verb, and takes any number of
items, so a bundle that
+collects what it provides across several modules can call it repeatedly
instead of passing everything to
+the constructor. Registering holds no sockets and starts nothing, so a unit
test can build a bundle and
+dispatch a handler through ``bundle.getTaskHandler(dagId, taskId)`` without a
coordinator runtime.
``new Dag`` and ``dag.task`` take a trailing options object: ``spec`` on both,
plus ``inputs`` on a task.
These are not used yet; do not set them. Any other key is rejected.
diff --git a/ts-sdk/README.md b/ts-sdk/README.md
index 2c5b9345748..4cc5c9b4e53 100644
--- a/ts-sdk/README.md
+++ b/ts-sdk/README.md
@@ -35,7 +35,7 @@ npm install [email protected]
## Task Handlers
```ts
-import { Dag, DagRegistry, getClient, getContext, serveDags } from
"apache-airflow-ts-sdk";
+import { Bundle, Dag, getClient, getContext } from "apache-airflow-ts-sdk";
export async function sayHello() {
const greeting = await getClient().getVariable("greeting");
@@ -45,7 +45,9 @@ export async function sayHello() {
const dag = new Dag("example_dag");
dag.task("say_hello", sayHello);
-await serveDags(new DagRegistry(dag));
+const bundle = new Bundle();
+bundle.register(dag);
+await bundle.serve();
```
A handler is a plain function. `getContext()` and `getClient()` reach the
runtime from inside the call,
@@ -103,7 +105,7 @@ Airflow metadata in the bundle itself.
TypeScript entrypoint:
```ts
-import { Dag, DagRegistry, getClient, serveDags } from "apache-airflow-ts-sdk";
+import { Bundle, Dag, getClient } from "apache-airflow-ts-sdk";
export async function extract() {
const client = getClient();
@@ -131,22 +133,22 @@ const salesPipeline = new Dag("sales_pipeline");
salesPipeline.task("extract", extract);
salesPipeline.task("transform", transform);
-await serveDags(new DagRegistry(salesPipeline));
+const bundle = new Bundle();
+bundle.register(salesPipeline);
+await bundle.serve();
```
The Python stub defines the Dag dependency graph. The TypeScript handler does
the work and uses `TaskClient` for task-time Airflow data access. Create a
`Dag` with the Python Dag's `dag_id` and attach each handler with the stub
task's `task_id`. The handler function is the reusable task implementation;
-`dag.task` binds that handler to a Python stub task identity, a `DagRegistry`
-collects the Dags this bundle can execute, and `serveDags` serves them to
-Airflow.
+`dag.task` binds that handler to a Python stub task identity, a `Bundle` holds
+what this bundle process provides, and `bundle.serve()` serves it to Airflow.
-`serveDags` is the entrypoint, and the registry it is given is the whole
bundle:
-a Dag left out of the registry is not part of the bundle, and its tasks are
-marked removed at runtime. The registry itself holds no sockets and starts
-nothing, so a unit test can build one and dispatch through
-`registry.getTaskHandler(dagId, taskId)` without any runtime involved.
+`bundle.serve()` is the entrypoint: a Dag left unregistered is not part of the
bundle,
+and its tasks are marked removed at runtime.
+Registering holds no sockets and starts nothing, so a unit test can build a
bundle
+and dispatch through `bundle.getTaskHandler(dagId, taskId)` without any
runtime involved.
`new Dag` and `dag.task` take a trailing options object: `spec` on both, plus
`inputs` on a task.
These are not used yet; do not set them.
@@ -157,18 +159,18 @@ entrypoint that serves them all:
```ts
import { salesDag } from "./sales/dag";
import { billingDag } from "./billing/dag";
-import { DagRegistry, serveDags } from "apache-airflow-ts-sdk";
+import { Bundle } from "apache-airflow-ts-sdk";
-await serveDags(new DagRegistry(salesDag, billingDag));
+await new Bundle(salesDag, billingDag).serve();
```
-A bundle that collects its Dags across several modules can add them
-incrementally with `registry.register(...)` instead of passing them all to the
-constructor.
+`register` is the bundle's one registration verb: a bundle that collects what
it
+provides across several modules can call it repeatedly instead of passing
+everything to the constructor.
Airflow launches the bundled entrypoint with `--comm=host:port` and
-`--logs=host:port`. `serveDags()` connects to those sockets, receives the task
-startup message, finds the registered handler for the Dag/task pair, and
+`--logs=host:port`. `bundle.serve()` connects to those sockets, receives the
+task startup message, finds the registered handler for the Dag/task pair, and
reports the terminal task state back to Airflow.
See [`example/`](https://github.com/apache/airflow/tree/main/ts-sdk/example)
for
diff --git a/ts-sdk/api-docs/dag-authoring-api.ts
b/ts-sdk/api-docs/dag-authoring-api.ts
index 065cd0ae689..b1e92900f56 100644
--- a/ts-sdk/api-docs/dag-authoring-api.ts
+++ b/ts-sdk/api-docs/dag-authoring-api.ts
@@ -19,9 +19,10 @@
/** @module Authoring */
-export { Dag, DagRegistry, getClient, getContext, serveDags } from
"../src/index.js";
+export { Bundle, Dag, getClient, getContext } from "../src/index.js";
export type {
DagSpec,
+ Registerable,
TaskClient,
TaskContext,
TaskFunction,
diff --git a/ts-sdk/docs/index.md b/ts-sdk/docs/index.md
index 5082acd4a5f..e53f5b8dc86 100644
--- a/ts-sdk/docs/index.md
+++ b/ts-sdk/docs/index.md
@@ -35,14 +35,14 @@ Install the beta package from npm:
npm install [email protected]
```
-Define a Dag and register its task handlers.
+Define a Dag, register it on a `Bundle`, and serve it.
A handler is a plain function: `getContext()` returns the `TaskContext` and
`getClient()` the `TaskClient`
for as long as it runs, so neither is a parameter.
Any non-`undefined` return value is pushed to XCom under the `"return_value"`
key by the active runtime,
matching Python `@task` behavior:
```ts
-import { Dag, DagRegistry, getClient, getContext, serveDags } from
"apache-airflow-ts-sdk";
+import { Bundle, Dag, getClient, getContext } from "apache-airflow-ts-sdk";
export async function sayHello() {
const greeting = await getClient().getVariable("greeting");
@@ -52,7 +52,9 @@ export async function sayHello() {
const dag = new Dag("example_dag");
dag.task("say_hello", sayHello);
-await serveDags(new DagRegistry(dag));
+const bundle = new Bundle();
+bundle.register(dag);
+await bundle.serve();
```
## Coordinators
diff --git a/ts-sdk/example/src/main.ts b/ts-sdk/example/src/main.ts
index e98be36b92e..3914148937a 100644
--- a/ts-sdk/example/src/main.ts
+++ b/ts-sdk/example/src/main.ts
@@ -17,7 +17,7 @@
* under the License.
*/
-import { Dag, DagRegistry, getClient, serveDags } from "apache-airflow-ts-sdk";
+import { Bundle, Dag, getClient } from "apache-airflow-ts-sdk";
const dag = new Dag("typescript_example");
@@ -53,4 +53,6 @@ export async function readConnection() {
dag.task("build_message", buildMessage);
dag.task("read_connection", readConnection);
-await serveDags(new DagRegistry(dag));
+const bundle = new Bundle();
+bundle.register(dag);
+await bundle.serve();
diff --git a/ts-sdk/scripts/verify-package.mjs
b/ts-sdk/scripts/verify-package.mjs
index 5655310a646..e8064682da4 100644
--- a/ts-sdk/scripts/verify-package.mjs
+++ b/ts-sdk/scripts/verify-package.mjs
@@ -34,7 +34,7 @@ export const REQUIRED_ROOT_FILES = ["LICENSE", "NOTICE",
"README.md", "package.j
// The Dag-authoring entrypoints a consumer must be able to reach from the
package root.
// The two getters are here because a handler cannot reach the runtime without
them: a
// published build that dropped them would still import, and fail at the first
task.
-export const REQUIRED_ROOT_EXPORTS = ["Dag", "DagRegistry", "getClient",
"getContext", "serveDags"];
+export const REQUIRED_ROOT_EXPORTS = ["Bundle", "Dag", "getClient",
"getContext"];
function run(command, args, options = {}) {
const result = spawnSync(command, args, {
diff --git a/ts-sdk/src/cli/pack.ts b/ts-sdk/src/cli/pack.ts
index 040c80e4414..5acea1b9626 100644
--- a/ts-sdk/src/cli/pack.ts
+++ b/ts-sdk/src/cli/pack.ts
@@ -18,7 +18,7 @@
*/
// airflow-ts-pack: bundle a TypeScript entrypoint into the single-file
-// artifact NodeCoordinator consumes — `bundle.mjs` with metadata and an
+// artifact NodeCoordinator consumes: `bundle.mjs` with metadata and an
// integrity layout descriptor embedded in JavaScript comments.
//
// Build first, then run the built bundle with --airflow-metadata so the
@@ -200,7 +200,7 @@ export async function runPack(argv: readonly string[]):
Promise<void> {
const manifest = readBundleManifest(stagingPath);
const dagEntries = Object.entries(manifest.dags);
if (dagEntries.length === 0) {
- throw new Error(`${args.entry} served no Dags; pass them to
serveDags(new DagRegistry(...))`);
+ throw new Error(`${args.entry} served no Dags; register them with
bundle.register(...)`);
}
// Warn rather than fail, as airflow-go-pack does: the shared schema
allows a
// Dag with no tasks.
diff --git a/ts-sdk/src/coordinator/index.ts b/ts-sdk/src/coordinator/index.ts
index f06214a4c2f..50996b9b275 100644
--- a/ts-sdk/src/coordinator/index.ts
+++ b/ts-sdk/src/coordinator/index.ts
@@ -22,11 +22,9 @@
// TaskClient and related types are exported from the package root. This
// barrel only exports coordinator-specific entry points.
//
-// `startCoordinator` is deliberately not exported: Dag authors reach the
-// runtime through `serveDags()`, and never name the coordinator itself.
+// Dag authors reach the runtime through `bundle.serve()`.
-export { serveDags } from "./runtime.js";
/** Cadwyn schema version this SDK was generated against. Not sent on
- * the wire — exposed so callers can read it for bundle metadata,
+ * the wire, but exposed so callers can read it for bundle metadata,
* health checks, or to confirm which schema their build is pinned to. */
export { SUPERVISOR_API_VERSION } from "./protocol.js";
diff --git a/ts-sdk/src/coordinator/manifest.ts
b/ts-sdk/src/coordinator/manifest.ts
index 145cd350167..3a35fc46e09 100644
--- a/ts-sdk/src/coordinator/manifest.ts
+++ b/ts-sdk/src/coordinator/manifest.ts
@@ -18,7 +18,7 @@
*/
import { SUPERVISOR_API_VERSION } from "./protocol.js";
-import { listRegistryDags, type DagRegistry } from "../sdk/registry.js";
+import { listBundleDags, type Bundle } from "../sdk/bundle.js";
export const AIRFLOW_METADATA_FLAG = "--airflow-metadata";
@@ -35,9 +35,9 @@ export interface BundleManifest {
dags: Record<string, { tasks: string[] }>;
}
-export function buildBundleManifest(registry: DagRegistry): BundleManifest {
+export function buildBundleManifest(bundle: Bundle): BundleManifest {
const dags: BundleManifest["dags"] = {};
- for (const { dagId, tasks } of listRegistryDags(registry)) {
+ for (const { dagId, tasks } of listBundleDags(bundle)) {
if (typeof dagId !== "string") {
throw new Error("Dag ID must be a string");
}
diff --git a/ts-sdk/src/coordinator/runtime.ts
b/ts-sdk/src/coordinator/runtime.ts
index 79a273e4278..0e05ec24b58 100644
--- a/ts-sdk/src/coordinator/runtime.ts
+++ b/ts-sdk/src/coordinator/runtime.ts
@@ -25,9 +25,9 @@
//
// where `my-bundle.mjs` is a user-bundled Node script that imports
// the SDK, creates `Dag` objects, attaches a handler per task with
-// `dag.task(...)`, collects them in a `DagRegistry`, then awaits
-// `serveDags(registry)`. Each handler runs inside a task scope, which is what
-// `getContext()` and `getClient()` read.
+// `dag.task(...)`, registers them on a `Bundle`, then awaits `bundle.serve()`.
+// Each handler runs inside a task scope, which is what `getContext()` and
+// `getClient()` read.
//
// Lifecycle:
// 1. Parse --comm / --logs from argv
@@ -53,68 +53,13 @@ import {
type RuntimeTaskState,
type StartupDetails,
} from "./protocol.js";
-import { DagRegistry, isDagRegistry, listRegistryTasks } from
"../sdk/registry.js";
-import { DUPLICATE_COPY_HINT } from "../sdk/brand.js";
+import { listBundleTasks, type Bundle } from "../sdk/bundle.js";
import { runInTaskScope, type TaskContext } from "../sdk/task.js";
import type { JsonValue } from "../sdk/client-types.js";
export const ABORT_GRACE_PERIOD_MS = 30_000;
export const COORDINATOR_RESPONSE_TIMEOUT_MS = 30_000;
-// What must not happen twice is one process connecting two pairs of sockets,
so
-// this is keyed globally rather than held in a module variable: two resolved
-// copies of the package would each get their own, and both would be first.
-const SERVED = Symbol.for("airflow.ts-sdk.served");
-
-function serveLatch(): Record<symbol, boolean | undefined> {
- return globalThis as unknown as Record<symbol, boolean | undefined>;
-}
-
-/**
- * Serve a bundle's Dags to Airflow. The entry point of a TypeScript Dag
bundle.
- *
- * Build the Dags, attach a handler per task with `dag.task(...)`, collect them
- * in a {@link DagRegistry}, then await this at module top level:
- *
- * ```ts
- * const dag = new Dag("my_dag");
- * dag.task("extract", extractFn);
- * await serveDags(new DagRegistry(dag));
- * ```
- *
- * The registry is the bundle's complete set of Dags: this process serves one
- * supervisor request, so a second call, which would connect a second pair of
- * sockets, is rejected. A call that fails is not a serve, and may be retried.
- * Resolves when Airflow's supervisor has been sent the terminal frame for the
- * work this process was started for; the same call also answers the build-time
- * `--airflow-metadata` query `airflow-ts-pack` makes.
- */
-export async function serveDags(registry: DagRegistry): Promise<void> {
- // Checked here rather than at first use: a bad argument otherwise surfaces
- // only after the sockets are up, as a missing method deep in the runtime.
- if (!(registry instanceof DagRegistry)) {
- throw new Error(
- isDagRegistry(registry)
- ? `The registry passed to serveDags(...) ${DUPLICATE_COPY_HINT}`
- : "serveDags(...) takes a DagRegistry; build one with new
DagRegistry(...dags)",
- );
- }
- const latch = serveLatch();
- if (latch[SERVED]) {
- throw new Error("serveDags(...) was already called; serve every Dag from a
single registry");
- }
- // Set before the first await, so two concurrent calls cannot both pass.
- latch[SERVED] = true;
- try {
- await startCoordinator(registry);
- } catch (err) {
- // startCoordinator closes both sockets on its way out, so a failed serve
- // holds nothing open and may be retried.
- latch[SERVED] = undefined;
- throw err;
- }
-}
-
/** Options for `startCoordinator()`. */
export interface StartCoordinatorOptions {
/** Comm socket address (host:port). Must be supplied together with
`logsAddr`; otherwise parsed from argv. */
@@ -165,20 +110,20 @@ export function parseArgs(argv: readonly string[]):
ParsedArgs {
return { commAddr, logsAddr };
}
-/** Start the coordinator runtime, dispatching to `registry`. Resolves when the
+/** Start the coordinator runtime, dispatching to `bundle`. Resolves when the
* subprocess has delivered its terminal frame and closed both sockets.
*
- * Internal: `serveDags()` is the entry point Dag authors call, so this is
+ * Internal: `bundle.serve()` is the entry point Dag authors call, so this is
* deliberately absent from the package `"exports"` map. Tests drive it
* directly to supply explicit socket addresses. */
export async function startCoordinator(
- registry: DagRegistry,
+ bundle: Bundle,
opts: StartCoordinatorOptions = {},
): Promise<void> {
const argv = opts.argv ?? process.argv;
if (argv.includes(AIRFLOW_METADATA_FLAG)) {
process.stdout.write(
-
`${AIRFLOW_METADATA_SENTINEL}${JSON.stringify(buildBundleManifest(registry))}\n`,
+
`${AIRFLOW_METADATA_SENTINEL}${JSON.stringify(buildBundleManifest(bundle))}\n`,
);
return;
}
@@ -201,7 +146,7 @@ export async function startCoordinator(
const runtimeLogs = logs.child("runtime");
runtimeLogs.debug("Connecting log socket", { logs_addr: parsed.logsAddr });
await logs.connect(parsed.logsAddr);
- const tasks = listRegistryTasks(registry);
+ const tasks = listBundleTasks(bundle);
runtimeLogs.info("Coordinator runtime started", {
registered_tasks: tasks,
count: tasks.length,
@@ -223,7 +168,7 @@ export async function startCoordinator(
file: body.file,
bundle_path: body.bundle_path,
});
- const response = handleParse(body, registry, runtimeLogs);
+ const response = handleParse(body, bundle, runtimeLogs);
await sendSupervisorResponse(firstFrame.id, response, comm, runtimeLogs);
} else if (body.type === "StartupDetails") {
runtimeLogs.info("Received task startup details", {
@@ -236,7 +181,7 @@ export async function startCoordinator(
});
const response = await handleTask(
body,
- registry,
+ bundle,
comm,
runtimeLogs,
logs.child("client"),
@@ -317,13 +262,13 @@ export function createRuntimeAbort(
function handleParse(
request: { file: string; bundle_path: string },
- registry: DagRegistry,
+ bundle: Bundle,
logs: LogChannel,
): RuntimeDagFileParsingResult {
// TypeScript-native Dag parsing is not yet supported.
// Respond with an empty result so the Python-stub-Dag workflow works.
logs.info("Parse-mode response (TS Dag parsing not yet supported)", {
- registered_tasks: listRegistryTasks(registry),
+ registered_tasks: listBundleTasks(bundle),
});
const response: RuntimeDagFileParsingResult = {
type: "DagFileParsingResult",
@@ -335,20 +280,20 @@ function handleParse(
async function handleTask(
details: StartupDetails,
- registry: DagRegistry,
+ bundle: Bundle,
comm: CommChannel,
logs: LogChannel,
clientLogs: LogChannel,
signal: AbortSignal,
): Promise<RuntimeSucceedTask | RuntimeRetryTask | RuntimeTaskState> {
const ti = details.ti;
- const handler = registry.getTaskHandler(ti.dag_id, ti.task_id);
+ const handler = bundle.getTaskHandler(ti.dag_id, ti.task_id);
if (!handler) {
logs.warning("No handler registered for task", {
dag_id: ti.dag_id,
task_id: ti.task_id,
- available: listRegistryTasks(registry),
+ available: listBundleTasks(bundle),
});
// A missing handler means this bundle cannot run the task, so retrying the
// same bundle/configuration mismatch would not help.
diff --git a/ts-sdk/src/coordinator/serve.ts b/ts-sdk/src/coordinator/serve.ts
new file mode 100644
index 00000000000..2429e571ef0
--- /dev/null
+++ b/ts-sdk/src/coordinator/serve.ts
@@ -0,0 +1,61 @@
+/*!
+ * 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.
+ */
+
+// What `bundle.serve()` does: the one-shot latch, and the call into the
+// coordinator. Kept out of `sdk/bundle.ts` so the authoring surface never
+// imports the coordinator, and out of `runtime.ts` so importing it back from
+// the SDK side cannot close a cycle.
+
+import { startCoordinator } from "./runtime.js";
+import type { Bundle } from "../sdk/bundle.js";
+
+// What must not happen twice is one process connecting two pairs of sockets,
so
+// this is keyed globally rather than held in a module variable: two resolved
+// copies of the package would each get their own, and both would be first.
+const SERVED = Symbol.for("airflow.ts-sdk.served");
+
+function serveLatch(): Record<symbol, boolean | undefined> {
+ return globalThis as unknown as Record<symbol, boolean | undefined>;
+}
+
+/**
+ * Internal: serve `bundle` to Airflow. The latch lives here, with the sockets
+ * it protects, rather than on the bundle that holds none.
+ *
+ * Not exported from the package root: a Dag author reaches the runtime through
+ * `bundle.serve()`.
+ */
+export async function serveBundle(bundle: Bundle): Promise<void> {
+ const latch = serveLatch();
+ if (latch[SERVED]) {
+ throw new Error(
+ "bundle.serve() was already called; serve everything a bundle provides
from a single bundle",
+ );
+ }
+ // Set before the first await, so two concurrent calls cannot both pass.
+ latch[SERVED] = true;
+ try {
+ await startCoordinator(bundle);
+ } catch (err) {
+ // startCoordinator closes both sockets on its way out, so a failed serve
+ // holds nothing open and may be retried.
+ latch[SERVED] = undefined;
+ throw err;
+ }
+}
diff --git a/ts-sdk/src/index.ts b/ts-sdk/src/index.ts
index b4a412f172e..edf2ceaafc9 100644
--- a/ts-sdk/src/index.ts
+++ b/ts-sdk/src/index.ts
@@ -18,10 +18,11 @@
*/
export { Dag } from "./sdk/dag.js";
-export { DagRegistry } from "./sdk/registry.js";
+export { Bundle } from "./sdk/bundle.js";
export { getClient, getContext } from "./sdk/task.js";
export { ConnectionNotFoundError, VariableNotFoundError } from
"./sdk/client.js";
-export { serveDags, SUPERVISOR_API_VERSION } from "./coordinator/index.js";
+export { SUPERVISOR_API_VERSION } from "./coordinator/index.js";
+export type { Registerable } from "./sdk/bundle.js";
export type { DagSpec, TaskInputs, TaskOptions, TaskRef, TaskSpec } from
"./sdk/dag.js";
export type { TaskClient } from "./sdk/client.js";
export type { ConnectionResult, GetXComOpts, JsonValue, SetXComOpts } from
"./sdk/client-types.js";
diff --git a/ts-sdk/src/sdk/brand.ts b/ts-sdk/src/sdk/brand.ts
index a785f736c02..5e09cb978a4 100644
--- a/ts-sdk/src/sdk/brand.ts
+++ b/ts-sdk/src/sdk/brand.ts
@@ -21,8 +21,8 @@
//
// Brands live in the cross-realm symbol registry, which two resolved copies of
// this package share even though each gets its own class object. That does not
-// make another copy's objects usable — `Dag` and `DagRegistry` read private
-// state keyed to the class that declared it — so callers still guard with
+// make another copy's objects usable, since `Dag` and `Bundle` read private
+// state keyed to the class that declared it, so callers still guard with
// `instanceof` and use these only to tell the two failures apart.
const PREFIX = "airflow.ts-sdk.";
diff --git a/ts-sdk/src/sdk/bundle.ts b/ts-sdk/src/sdk/bundle.ts
new file mode 100644
index 00000000000..615fe971ee9
--- /dev/null
+++ b/ts-sdk/src/sdk/bundle.ts
@@ -0,0 +1,176 @@
+/*!
+ * 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.
+ */
+
+// The bundle: what a TypeScript bundle process provides, and how it serves it.
+
+import { brand, DUPLICATE_COPY_HINT, hasBrand } from "./brand.js";
+import { Dag, getDagTaskRecords, isDag, type TaskRef } from "./dag.js";
+import type { TaskFunction } from "./task.js";
+
+// Assigned inside Bundle's static block, as Dag does for its tasks.
+let dagsOf: (bundle: Bundle) => ReadonlyMap<string, Dag>;
+
+/**
+ * Anything {@link Bundle.register} accepts.
+ *
+ * A union rather than a base class or an interface: TypeScript's equivalent of
+ * the sealed interface the Go SDK uses for the same purpose. Registering gains
+ * a kind by gaining an arm here, never a second verb.
+ */
+export type Registerable = Dag;
+
+/** Internal: whether `value` is a Bundle built by any copy of this package. */
+export function isBundle(value: unknown): value is Bundle {
+ return hasBrand(value, "Bundle");
+}
+
+/** Internal: a registered Dag with its task IDs, as {@link listBundleDags}
reports it.
+ * A task-less Dag is included, so the bundle manifest keeps it visible. */
+export interface RegisteredDag {
+ /** Identifier of the registered Dag. */
+ readonly dagId: string;
+ /** Airflow task IDs, including any TaskGroup prefix. */
+ readonly tasks: string[];
+}
+
+/**
+ * What a bundle process provides to Airflow, and the thing that serves it.
+ *
+ * A bundle entry point builds one, registers what it provides, and awaits
+ * `serve()` at module top level:
+ *
+ * ```ts
+ * const dag = new Dag("my_dag");
+ * dag.task("extract", extractFn);
+ *
+ * const bundle = new Bundle();
+ * bundle.register(dag);
+ * await bundle.serve();
+ * ```
+ *
+ * Registering holds no sockets and starts nothing, so a test can build a
bundle
+ * and invoke a handler through {@link getTaskHandler} without any runtime in
+ * scope. Only `serve()` connects to Airflow.
+ *
+ * Lookups delegate live to each Dag's task map, so tasks added to a Dag
+ * after registration are visible: the bundle records Dag identity, not
+ * a snapshot of its tasks.
+ */
+export class Bundle {
+ readonly #dags = new Map<string, Dag>();
+
+ static {
+ dagsOf = (bundle) => bundle.#dags;
+ }
+
+ /** Registers `items`, on the same terms as {@link register}. */
+ constructor(...items: Registerable[]) {
+ brand(this, "Bundle");
+ this.register(...items);
+ }
+
+ /** Register what this bundle provides. Registering an already-registered
+ * `dagId` throws, and a call that throws registers none of its items.
+ *
+ * The constructor covers the common case; this is for a bundle that
+ * collects what it provides across several modules. */
+ register(...items: Registerable[]): void {
+ const incoming = new Set<string>();
+ for (const item of items) {
+ // Typed as Registerable, so narrowing it would collapse to never; these
+ // guard callers reaching this from plain JavaScript.
+ const candidate: unknown = item;
+ // Another copy's Dag cannot be registered, since lookups read a private
+ // task map keyed to this copy's class, so it is rejected by its cause.
+ if (!(candidate instanceof Dag)) {
+ throw new Error(
+ isDag(candidate)
+ ? `Dag "${candidate.dagId}" ${DUPLICATE_COPY_HINT}`
+ : "only Dag instances can be registered",
+ );
+ }
+ if (this.#dags.has(item.dagId) || incoming.has(item.dagId)) {
+ throw new Error(`Dag "${item.dagId}" is already registered`);
+ }
+ incoming.add(item.dagId);
+ }
+ for (const item of items) {
+ this.#dags.set(item.dagId, item);
+ }
+ }
+
+ /**
+ * Serve this bundle to Airflow. The entry point of a TypeScript Dag bundle.
+ *
+ * A bundle process serves one supervisor request, so a second call, which
+ * would connect a second pair of sockets, is rejected. A call that fails is
+ * not a serve, and may be retried. Resolves once the supervisor has been
sent
+ * the terminal frame for the work this process was started for, and the same
+ * call answers the build-time `--airflow-metadata` query `airflow-ts-pack`
+ * makes.
+ *
+ * Everything this bundle provides must be registered before `serve()` is
+ * awaited: what is left out is not part of the bundle, and its tasks are
+ * marked removed at runtime.
+ */
+ async serve(): Promise<void> {
+ // `const { serve } = bundle` detaches the method, which would otherwise
+ // fail deep in the runtime on a missing private field rather than here.
+ validateOwnBundle(this, "bundle.serve()");
+ // Imported here rather than at module top: the coordinator reads a bundle,
+ // so a static import would make the authoring surface and the coordinator
+ // mutually dependent for the sake of one call.
+ const { serveBundle } = await import("../coordinator/serve.js");
+ await serveBundle(this);
+ }
+
+ /** Look up a registered handler, the way the runtime dispatches a task.
+ * Returns `undefined` when no handler exists. */
+ getTaskHandler(dagId: string, taskId: string): TaskFunction | undefined {
+ const dag = this.#dags.get(dagId);
+ return dag ? getDagTaskRecords(dag).get(taskId)?.fn : undefined;
+ }
+}
+
+/** Internal: reject a `this` that is not a Bundle built by this copy, naming
+ * the cause. */
+export function validateOwnBundle(value: unknown, accessor: string): asserts
value is Bundle {
+ if (value instanceof Bundle) return;
+ throw new Error(
+ isBundle(value)
+ ? `The bundle ${accessor} was called on ${DUPLICATE_COPY_HINT}`
+ : `${accessor} must be called on a Bundle; build one with new
Bundle(...)`,
+ );
+}
+
+/** Internal: the task handles across a bundle's Dags. Not re-exported from the
+ * package root: enumerating what the runtime dispatches is the runtime's
job. */
+export function listBundleTasks(bundle: Bundle): TaskRef[] {
+ return [...dagsOf(bundle).values()].flatMap((dag) =>
+ [...getDagTaskRecords(dag).values()].map((record) => record.task),
+ );
+}
+
+/** Internal: every registered Dag with its task IDs, empty Dags included. */
+export function listBundleDags(bundle: Bundle): RegisteredDag[] {
+ return [...dagsOf(bundle).values()].map((dag) => ({
+ dagId: dag.dagId,
+ tasks: [...dag.taskIds],
+ }));
+}
diff --git a/ts-sdk/src/sdk/dag.ts b/ts-sdk/src/sdk/dag.ts
index a329aa8c869..7374bf64883 100644
--- a/ts-sdk/src/sdk/dag.ts
+++ b/ts-sdk/src/sdk/dag.ts
@@ -130,7 +130,7 @@ export function isDag(value: unknown): value is Dag {
* TypeScript Dag declaration.
*
* Constructing a Dag has no effect beyond the instance itself. Collect the
ones
- * a bundle should serve in a `DagRegistry` and pass it to `serveDags(...)`.
+ * a bundle should serve on a `Bundle` and await `bundle.serve()`.
*/
export class Dag {
/** Identifier of this Dag. Must match the Python Dag's `dag_id`. */
diff --git a/ts-sdk/src/sdk/registry.ts b/ts-sdk/src/sdk/registry.ts
deleted file mode 100644
index 9d9259ad540..00000000000
--- a/ts-sdk/src/sdk/registry.ts
+++ /dev/null
@@ -1,124 +0,0 @@
-/*!
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import { brand, DUPLICATE_COPY_HINT, hasBrand } from "./brand.js";
-import { Dag, getDagTaskRecords, isDag, type TaskRef } from "./dag.js";
-import type { TaskFunction } from "./task.js";
-
-// Assigned inside DagRegistry's static block, as Dag does for its tasks.
-let dagsOf: (registry: DagRegistry) => ReadonlyMap<string, Dag>;
-
-/** Internal: whether `value` is a DagRegistry built by any copy of this
package. */
-export function isDagRegistry(value: unknown): value is DagRegistry {
- return hasBrand(value, "DagRegistry");
-}
-
-/** Internal: a registered Dag with its task IDs, as {@link listRegistryDags}
reports it.
- * A task-less Dag is included, so the bundle manifest keeps it visible. */
-export interface RegisteredDag {
- /** Identifier of the registered Dag. */
- readonly dagId: string;
- /** Airflow task IDs, including any TaskGroup prefix. */
- readonly tasks: string[];
-}
-
-/**
- * The Dags a bundle process can execute, keyed by Dag ID.
- *
- * This is what a bundle entry point builds and hands to `serveDags(registry)`:
- *
- * ```ts
- * const dag = new Dag("my_dag");
- * dag.task("extract", extractFn);
- * await serveDags(new DagRegistry(dag));
- * ```
- *
- * It holds no sockets and starts nothing, so a test can build one and invoke a
- * handler through {@link getTaskHandler} without any runtime in scope.
- *
- * Lookups delegate live to each Dag's task map, so tasks added to a Dag
- * after registration are visible: the registry records Dag identity, not
- * a snapshot of its tasks.
- */
-export class DagRegistry {
- readonly #dags = new Map<string, Dag>();
-
- static {
- dagsOf = (registry) => registry.#dags;
- }
-
- /** Registers `dags`, on the same terms as {@link register}. */
- constructor(...dags: Dag[]) {
- brand(this, "DagRegistry");
- this.register(...dags);
- }
-
- /** Register Dags. Registering an already-registered `dagId` throws,
- * and a call that throws registers none of its Dags.
- *
- * The constructor covers the common case; this is for a bundle that
- * collects its Dags across several modules. */
- register(...dags: Dag[]): void {
- const incoming = new Set<string>();
- for (const dag of dags) {
- // Typed as Dag, so narrowing it would collapse to never; these guard
- // callers reaching this from plain JavaScript.
- const candidate: unknown = dag;
- // Another copy's Dag cannot be registered, since lookups read a private
- // task map keyed to this copy's class, so it is rejected by its cause.
- if (!(candidate instanceof Dag)) {
- throw new Error(
- isDag(candidate)
- ? `Dag "${candidate.dagId}" ${DUPLICATE_COPY_HINT}`
- : "only Dag instances can be registered",
- );
- }
- if (this.#dags.has(dag.dagId) || incoming.has(dag.dagId)) {
- throw new Error(`Dag "${dag.dagId}" is already registered`);
- }
- incoming.add(dag.dagId);
- }
- for (const dag of dags) {
- this.#dags.set(dag.dagId, dag);
- }
- }
-
- /** Look up a registered handler, the way the runtime dispatches a task.
- * Returns `undefined` when no handler exists. */
- getTaskHandler(dagId: string, taskId: string): TaskFunction | undefined {
- const dag = this.#dags.get(dagId);
- return dag ? getDagTaskRecords(dag).get(taskId)?.fn : undefined;
- }
-}
-
-/** Internal: the task handles across a registry's Dags. Not re-exported from
the
- * package root: enumerating what the runtime dispatches is the runtime's
job. */
-export function listRegistryTasks(registry: DagRegistry): TaskRef[] {
- return [...dagsOf(registry).values()].flatMap((dag) =>
- [...getDagTaskRecords(dag).values()].map((record) => record.task),
- );
-}
-
-/** Internal: every registered Dag with its task IDs, empty Dags included. */
-export function listRegistryDags(registry: DagRegistry): RegisteredDag[] {
- return [...dagsOf(registry).values()].map((dag) => ({
- dagId: dag.dagId,
- tasks: [...dag.taskIds],
- }));
-}
diff --git a/ts-sdk/tests/cli/fixtures/bundle-v1.mjs
b/ts-sdk/tests/cli/fixtures/bundle-v1.mjs
index af75e60c244..9edb3b8d3c2 100644
--- a/ts-sdk/tests/cli/fixtures/bundle-v1.mjs
+++ b/ts-sdk/tests/cli/fixtures/bundle-v1.mjs
@@ -1,4 +1,4 @@
-//#
airflowBundle={"code":{"start":"0000000000000203","end":"0000000000000592","sha256":"f814358e0d4aa5d38c10c365515179049171bb9d4bba5b80c06ea549d6f16337"},"metadata":{"start":"000000000000013e","end":"0000000000000202","sha256":"a51dfd6f0c9e8ea867900e55c0387b556d3cb0e98321b62d4625f522ed465041"}}
+//#
airflowBundle={"code":{"start":"0000000000000203","end":"000000000000057a","sha256":"bd26f9a6295069aef9eff45213377a49a448dcc0973da182a29ffb927f259e8e"},"metadata":{"start":"000000000000013e","end":"0000000000000202","sha256":"a51dfd6f0c9e8ea867900e55c0387b556d3cb0e98321b62d4625f522ed465041"}}
//#
airflowMetadata={"airflow_bundle_metadata_version":"1.0","sdk":{"language":"typescript","version":"0.1.0","supervisor_schema_version":"2026-06-16"},"source":"entry.ts","dags":{"test_dag":{"tasks":["test_task"]}}}
/*!
* Licensed to the Apache Software Foundation (ASF) under one
@@ -19,6 +19,6 @@
* under the License.
*/
-import { DagRegistry, serveDags } from "../../../src/index.js";
+import { Bundle } from "../../../src/index.js";
-await serveDags(new DagRegistry());
+await new Bundle().serve();
diff --git a/ts-sdk/tests/cli/fixtures/empty-entry.ts
b/ts-sdk/tests/cli/fixtures/empty-entry.ts
index c48c6c24cf8..098b671188f 100644
--- a/ts-sdk/tests/cli/fixtures/empty-entry.ts
+++ b/ts-sdk/tests/cli/fixtures/empty-entry.ts
@@ -17,6 +17,6 @@
* under the License.
*/
-import { DagRegistry, serveDags } from "../../../src/index.js";
+import { Bundle } from "../../../src/index.js";
-await serveDags(new DagRegistry());
+await new Bundle().serve();
diff --git a/ts-sdk/tests/cli/fixtures/entry.ts
b/ts-sdk/tests/cli/fixtures/entry.ts
index 8549fb56a32..bd388974a2e 100644
--- a/ts-sdk/tests/cli/fixtures/entry.ts
+++ b/ts-sdk/tests/cli/fixtures/entry.ts
@@ -17,7 +17,7 @@
* under the License.
*/
-import { Dag, DagRegistry, serveDags } from "../../../src/index.js";
+import { Bundle, Dag } from "../../../src/index.js";
const fixtureDag = new Dag("fixture_dag");
fixtureDag.task("extract", async () => "extracted");
@@ -25,4 +25,4 @@ fixtureDag.task("transform", async () => "transformed");
const otherDag = new Dag("other_dag");
otherDag.task("solo", async () => undefined);
-await serveDags(new DagRegistry(fixtureDag, otherDag));
+await new Bundle(fixtureDag, otherDag).serve();
diff --git a/ts-sdk/tests/cli/fixtures/noisy-entry.ts
b/ts-sdk/tests/cli/fixtures/noisy-entry.ts
index dcd7b0d8c0b..370022287c7 100644
--- a/ts-sdk/tests/cli/fixtures/noisy-entry.ts
+++ b/ts-sdk/tests/cli/fixtures/noisy-entry.ts
@@ -18,11 +18,11 @@
* under the License.
*/
-import { Dag, DagRegistry, serveDags } from "../../../src/index.js";
+import { Bundle, Dag } from "../../../src/index.js";
console.log("noise from an import-time dependency");
const noisyDag = new Dag("noisy_dag");
noisyDag.task("only", async () => undefined);
-await serveDags(new DagRegistry(noisyDag));
+await new Bundle(noisyDag).serve();
diff --git a/ts-sdk/tests/cli/pack.test.ts b/ts-sdk/tests/cli/pack.test.ts
index 51928359c03..cd5dc1ca0b6 100644
--- a/ts-sdk/tests/cli/pack.test.ts
+++ b/ts-sdk/tests/cli/pack.test.ts
@@ -188,7 +188,7 @@ describe("runPack", () => {
if (outdir) rmSync(outdir, { recursive: true, force: true });
});
- it("bundles the entry and embeds metadata from the bundle's registry", async
() => {
+ it("bundles the entry and embeds metadata from the bundle's bundle", async
() => {
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
const nested = path.join(outdir, "dist");
await runPack([FIXTURE_ENTRY, "--outdir", nested]);
@@ -272,10 +272,10 @@ describe("runPack", () => {
writeFileSync(
entry,
[
- `import { Dag, DagRegistry, serveDags } from
${JSON.stringify(SDK_INDEX)};`,
+ `import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
'const bigDag = new Dag("big_dag");',
'for (let i = 0; i < 5000; i += 1) bigDag.task(String(i).padStart(240,
"t"), async () => undefined);',
- "await serveDags(new DagRegistry(bigDag));",
+ "await new Bundle(bigDag).serve();",
].join("\n"),
);
@@ -315,10 +315,10 @@ describe("runPack", () => {
writeFileSync(
entry,
[
- `import { Dag, DagRegistry, serveDags } from
${JSON.stringify(SDK_INDEX)};`,
+ `import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
`const suspiciousDag = new Dag(${JSON.stringify(dagId)});`,
`suspiciousDag.task(${JSON.stringify(taskId)}, async () =>
undefined);`,
- "await serveDags(new DagRegistry(suspiciousDag));",
+ "await new Bundle(suspiciousDag).serve();",
].join("\n"),
);
const stderr = captureStderr();
@@ -382,10 +382,10 @@ describe("runPack", () => {
writeFileSync(
entry,
[
- `import { Dag, DagRegistry, serveDags } from
${JSON.stringify(SDK_INDEX)};`,
+ `import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
'const salesDag = new Dag("sales_dag");',
'salesDag.task("extract", async () => undefined);',
- 'await serveDags(new DagRegistry(salesDag, new Dag("empty_dag")));',
+ 'await new Bundle(salesDag, new Dag("empty_dag")).serve();',
].join("\n"),
);
const stderr = captureStderr();
@@ -399,18 +399,18 @@ describe("runPack", () => {
);
});
- it("packs only the Dags the served registry holds", async () => {
+ it("packs only the Dags the served bundle holds", async () => {
outdir = mkdtempSync(path.join(tmpdir(), "ts-pack-"));
const entry = path.join(outdir, "forgotten-entry.ts");
writeFileSync(
entry,
[
- `import { Dag, DagRegistry, serveDags } from
${JSON.stringify(SDK_INDEX)};`,
+ `import { Bundle, Dag } from ${JSON.stringify(SDK_INDEX)};`,
'const salesDag = new Dag("sales_dag");',
'salesDag.task("extract", async () => undefined);',
'const billingDag = new Dag("billing_dag");',
'billingDag.task("charge", async () => undefined);',
- "await serveDags(new DagRegistry(salesDag));",
+ "await new Bundle(salesDag).serve();",
].join("\n"),
);
diff --git a/ts-sdk/tests/coordinator/integration.test.ts
b/ts-sdk/tests/coordinator/integration.test.ts
index a3293a57847..e6294154187 100644
--- a/ts-sdk/tests/coordinator/integration.test.ts
+++ b/ts-sdk/tests/coordinator/integration.test.ts
@@ -36,15 +36,15 @@ import {
startCoordinator,
} from "../../src/coordinator/runtime.js";
import { Dag } from "../../src/sdk/dag.js";
-import { DagRegistry } from "../../src/sdk/registry.js";
+import { Bundle } from "../../src/sdk/bundle.js";
import { getClient, getContext } from "../../src/sdk/task.js";
const testDag = new Dag("test_dag");
const otherDag = new Dag("other_dag");
-// The registry the runtime dispatches through. startCoordinator() is driven
-// directly rather than through serveDags(), so these tests can supply mock
+// The bundle the runtime dispatches through. startCoordinator() is driven
+// directly rather than through bundle.serve(), so these tests can supply mock
// socket addresses.
-const registry = new DagRegistry(testDag, otherDag);
+const bundle = new Bundle(testDag, otherDag);
interface MockResult {
firstResponse: { id: number; body: unknown; isResponse: boolean } | null;
@@ -150,7 +150,7 @@ async function driveSupervisor(initialFrame: unknown,
responder?: Responder): Pr
const commAccept = acceptOne(comm.server);
const logsAccept = acceptOne(logs.server);
- const runtimeDone = startCoordinator(registry, {
+ const runtimeDone = startCoordinator(bundle, {
commAddr: `127.0.0.1:${comm.port}`,
logsAddr: `127.0.0.1:${logs.port}`,
argv: [],
@@ -221,7 +221,7 @@ describe("coordinator runtime integration", () => {
const logsSockPromise = acceptOne(logs.server);
const commSockPromise = acceptOne(comm.server);
- const runtimeDone = startCoordinator(registry, {
+ const runtimeDone = startCoordinator(bundle, {
commAddr: `127.0.0.1:${comm.port}`,
logsAddr: `127.0.0.1:${logs.port}`,
argv: [],
@@ -297,7 +297,7 @@ describe("coordinator runtime integration", () => {
const logsAccept = acceptOne(logs.server);
testDag.task("terminal_timeout", async () => undefined);
- const runtimeDone = startCoordinator(registry, {
+ const runtimeDone = startCoordinator(bundle, {
commAddr: `127.0.0.1:${comm.port}`,
logsAddr: `127.0.0.1:${logs.port}`,
argv: [],
diff --git a/ts-sdk/tests/coordinator/protocol.test.ts
b/ts-sdk/tests/coordinator/protocol.test.ts
index 8089a595a5d..12ba947eeb3 100644
--- a/ts-sdk/tests/coordinator/protocol.test.ts
+++ b/ts-sdk/tests/coordinator/protocol.test.ts
@@ -20,7 +20,7 @@
import { describe, expect, it } from "vitest";
import { asMsgFromSupervisor } from "../../src/coordinator/protocol.js";
import { parseArgs, startCoordinator } from "../../src/coordinator/runtime.js";
-import { DagRegistry } from "../../src/sdk/registry.js";
+import { Bundle } from "../../src/sdk/bundle.js";
describe("protocol decode", () => {
it("accepts StartupDetails", () => {
@@ -86,14 +86,14 @@ describe("runtime arg parser", () => {
it("requires commAddr and logsAddr overrides to be supplied together", async
() => {
await expect(
- startCoordinator(new DagRegistry(), {
+ startCoordinator(new Bundle(), {
commAddr: "127.0.0.1:5001",
argv: ["node", "bundle.mjs", "--logs=127.0.0.1:5002"],
}),
).rejects.toThrow(/Missing --comm/);
await expect(
- startCoordinator(new DagRegistry(), {
+ startCoordinator(new Bundle(), {
logsAddr: "127.0.0.1:5002",
argv: ["node", "bundle.mjs", "--comm=127.0.0.1:5001"],
}),
diff --git a/ts-sdk/tests/coordinator/public-api.test.ts
b/ts-sdk/tests/coordinator/public-api.test.ts
index 16e91d4b31f..a4176952ced 100644
--- a/ts-sdk/tests/coordinator/public-api.test.ts
+++ b/ts-sdk/tests/coordinator/public-api.test.ts
@@ -18,14 +18,17 @@
*/
import { describe, expect, expectTypeOf, it } from "vitest";
-import type { DagRegistry } from "../../src/sdk/registry.js";
import * as coordinator from "../../src/coordinator/index.js";
-import { serveDags } from "../../src/coordinator/index.js";
describe("coordinator public API", () => {
- it("exposes serveDags, not the coordinator itself, from the coordinator
subpath", () => {
- expectTypeOf<typeof serveDags>().toEqualTypeOf<(registry: DagRegistry) =>
Promise<void>>();
- expect("startCoordinator" in coordinator).toBe(false);
+ it("names neither the coordinator nor the serve it performs", () => {
+ // A Dag author reaches the runtime through `bundle.serve()`. The subpath
+ // carries only the schema version, so nothing here is an entry point.
+ for (const name of ["startCoordinator", "serveBundle", "serveDags"]) {
+ expect(name in coordinator).toBe(false);
+ }
expectTypeOf<typeof coordinator>().not.toHaveProperty("startCoordinator");
+ expectTypeOf<typeof coordinator>().not.toHaveProperty("serveBundle");
+ expectTypeOf<typeof
coordinator>().toHaveProperty("SUPERVISOR_API_VERSION");
});
});
diff --git a/ts-sdk/tests/coordinator/runtime-manifest.test.ts
b/ts-sdk/tests/coordinator/runtime-manifest.test.ts
index f1e4ea76b17..09f9ee1a274 100644
--- a/ts-sdk/tests/coordinator/runtime-manifest.test.ts
+++ b/ts-sdk/tests/coordinator/runtime-manifest.test.ts
@@ -23,7 +23,7 @@ import { AIRFLOW_METADATA_SENTINEL, buildBundleManifest }
from "../../src/coordi
import { startCoordinator } from "../../src/coordinator/runtime.js";
import { SUPERVISOR_API_VERSION } from "../../src/coordinator/protocol.js";
import { Dag } from "../../src/sdk/dag.js";
-import { DagRegistry } from "../../src/sdk/registry.js";
+import { Bundle } from "../../src/sdk/bundle.js";
function buildDag(dagId: string, ...taskIds: string[]): Dag {
const dag = new Dag(dagId);
@@ -34,9 +34,9 @@ function buildDag(dagId: string, ...taskIds: string[]): Dag {
}
describe("buildBundleManifest", () => {
- it("maps a registry's Dags to their tasks under the SDK's schema version",
() => {
- const registry = new DagRegistry(buildDag("dag_a", "t1", "t3"),
buildDag("dag_b", "t2"));
- expect(buildBundleManifest(registry)).toEqual({
+ it("maps a bundle's Dags to their tasks under the SDK's schema version", ()
=> {
+ const bundle = new Bundle(buildDag("dag_a", "t1", "t3"), buildDag("dag_b",
"t2"));
+ expect(buildBundleManifest(bundle)).toEqual({
supervisor_schema_version: SUPERVISOR_API_VERSION,
dags: {
dag_a: { tasks: ["t1", "t3"] },
@@ -46,23 +46,23 @@ describe("buildBundleManifest", () => {
});
it("keeps a registered Dag without tasks visible in the manifest", () => {
- expect(buildBundleManifest(new
DagRegistry(buildDag("empty_dag"))).dags).toEqual({
+ expect(buildBundleManifest(new
Bundle(buildDag("empty_dag"))).dags).toEqual({
empty_dag: { tasks: [] },
});
});
it("keeps a Dag named __proto__ visible in serialized metadata", () => {
- const manifest = buildBundleManifest(new DagRegistry(buildDag("__proto__",
"task")));
+ const manifest = buildBundleManifest(new Bundle(buildDag("__proto__",
"task")));
const serializedDags = JSON.parse(JSON.stringify(manifest)).dags;
expect(Object.keys(serializedDags)).toEqual(["__proto__"]);
expect(serializedDags["__proto__"]).toEqual({ tasks: ["task"] });
});
- it("reports only the Dags the registry was given", () => {
- const registry = new DagRegistry(buildDag("dag_a", "t1"));
+ it("reports only the Dags the bundle was given", () => {
+ const bundle = new Bundle(buildDag("dag_a", "t1"));
buildDag("dag_b", "t2");
- expect(Object.keys(buildBundleManifest(registry).dags)).toEqual(["dag_a"]);
+ expect(Object.keys(buildBundleManifest(bundle).dags)).toEqual(["dag_a"]);
});
// The server would reject these ids. The manifest keeps them and
@@ -70,7 +70,7 @@ describe("buildBundleManifest", () => {
it.each(["", " ", "\t", "my dag", "a/b", "task@1", "d".repeat(251)])(
"keeps a dagId the server would reject visible in the manifest: %j",
(dagId) => {
- const manifest = buildBundleManifest(new DagRegistry(buildDag(dagId,
"t1")));
+ const manifest = buildBundleManifest(new Bundle(buildDag(dagId, "t1")));
expect(manifest.dags[dagId]).toEqual({ tasks: ["t1"] });
},
);
@@ -80,7 +80,7 @@ describe("buildBundleManifest", () => {
it.each([" ", "\t", "my task", "a/b", "task@1", "t".repeat(251)])(
"keeps a taskId the server would reject visible in the manifest: %j",
(taskId) => {
- const manifest = buildBundleManifest(new
DagRegistry(buildDag("example_dag", taskId)));
+ const manifest = buildBundleManifest(new Bundle(buildDag("example_dag",
taskId)));
expect(manifest.dags["example_dag"]).toEqual({ tasks: [taskId] });
},
);
@@ -88,7 +88,7 @@ describe("buildBundleManifest", () => {
it("rejects a non-string dagId before object-key coercion hides it", () => {
const dag = new Dag(123 as unknown as string);
dag.task("t1", async () => undefined);
- expect(() => buildBundleManifest(new DagRegistry(dag))).toThrowError(/Dag
ID must be a string/);
+ expect(() => buildBundleManifest(new Bundle(dag))).toThrowError(/Dag ID
must be a string/);
});
});
@@ -100,7 +100,7 @@ describe("startCoordinator --airflow-metadata", () => {
it("dumps the manifest to stdout and returns without connecting", async ()
=> {
const write = vi.spyOn(process.stdout, "write").mockReturnValue(true);
- await startCoordinator(new DagRegistry(buildDag("metadata_dag", "only")), {
+ await startCoordinator(new Bundle(buildDag("metadata_dag", "only")), {
argv: ["node", "bundle.mjs", "--airflow-metadata"],
});
diff --git a/ts-sdk/tests/public-api.test.ts b/ts-sdk/tests/public-api.test.ts
index 9f0f732f838..ad94cd389a0 100644
--- a/ts-sdk/tests/public-api.test.ts
+++ b/ts-sdk/tests/public-api.test.ts
@@ -25,6 +25,7 @@ import type {
GetXComOpts,
SetXComOpts,
TaskClient,
+ Registerable,
TaskContext,
TaskFunction,
TaskInputs,
@@ -34,12 +35,11 @@ import type {
} from "../src/index.js";
import * as sdk from "../src/index.js";
import {
+ Bundle,
ConnectionNotFoundError,
Dag,
- DagRegistry,
getClient,
getContext,
- serveDags,
SUPERVISOR_API_VERSION,
VariableNotFoundError,
} from "../src/index.js";
@@ -54,29 +54,33 @@ describe("public API", () => {
expect(upstream).toEqual({ dagId: "public_api_dag", taskId:
"public_api_task" });
expect(downstream).toEqual({ dagId: "public_api_dag", taskId:
"public_api_downstream" });
expect(dag.taskIds).toEqual(["public_api_task", "public_api_downstream"]);
- // serveDags hands the registry to the runtime, which needs the
supervisor's
+ // serve() hands the bundle to the runtime, which needs the supervisor's
// socket addresses that Airflow puts on argv.
- await expect(serveDags(new DagRegistry(dag))).rejects.toThrow("Missing
--comm");
+ await expect(new Bundle(dag).serve()).rejects.toThrow("Missing --comm");
});
- // The registry guard runs before the already-served latch, so this holds
- // regardless of whether another test in this file already served a registry.
+ // A detached `const { serve } = bundle` must say so here rather than fail
+ // deep in the runtime. The guard runs before the already-served latch, so
+ // this holds however many bundles earlier tests in this file served.
it.each([
- ["a bare Dag", new Dag("not_a_registry_dag")],
+ ["a bare Dag", new Dag("not_a_bundle_dag")],
["a plain object", { register: () => {} }],
["null", null],
- ])("rejects a %s in place of a registry", async (_label, value) => {
- await expect(serveDags(value as unknown as DagRegistry)).rejects.toThrow(
- /serveDags\(\.\.\.\) takes a DagRegistry/,
+ ["undefined, as a detached serve receives", undefined],
+ ])("rejects %s as the receiver of serve()", async (_label, value) => {
+ const detached = Bundle.prototype.serve;
+ await expect(detached.call(value as unknown as Bundle)).rejects.toThrow(
+ /bundle\.serve\(\) must be called on a Bundle/,
);
});
- it("names the duplicate-copy cause for a registry built by another copy",
async () => {
- // Stands in for a registry from a second resolved copy: same brand, other
+ it("names the duplicate-copy cause for a bundle built by another copy",
async () => {
+ // Stands in for a bundle from a second resolved copy: same brand, other
// class. It still cannot be served, so the point is only that it says why.
const foreign = {};
- Object.defineProperty(foreign, Symbol.for("airflow.ts-sdk.DagRegistry"), {
value: true });
- await expect(serveDags(foreign as unknown as DagRegistry)).rejects.toThrow(
+ Object.defineProperty(foreign, Symbol.for("airflow.ts-sdk.Bundle"), {
value: true });
+ const detached = Bundle.prototype.serve;
+ await expect(detached.call(foreign as unknown as Bundle)).rejects.toThrow(
/different copy of apache-airflow-ts-sdk/,
);
});
@@ -95,9 +99,9 @@ describe("public API", () => {
process.argv = [...argv, AIRFLOW_METADATA_FLAG];
vi.spyOn(process.stdout, "write").mockReturnValue(true);
try {
- await serveDags(new DagRegistry(new Dag("served_dag")));
- await expect(serveDags(new DagRegistry(new
Dag("second_call_dag")))).rejects.toThrow(
- /serveDags\(\.\.\.\) was already called/,
+ await new Bundle(new Dag("served_dag")).serve();
+ await expect(new Bundle(new
Dag("second_call_dag")).serve()).rejects.toThrow(
+ /bundle\.serve\(\) was already called/,
);
} finally {
process.argv = argv;
@@ -106,34 +110,30 @@ describe("public API", () => {
});
it("releases the latch when a serve fails, so the call can be retried",
async () => {
- await expect(serveDags(new DagRegistry(new
Dag("first_try")))).rejects.toThrow(
- "Missing --comm",
- );
+ await expect(new Bundle(new
Dag("first_try")).serve()).rejects.toThrow("Missing --comm");
// The retry reports why it actually failed, not "already called".
- await expect(serveDags(new DagRegistry(new
Dag("second_try")))).rejects.toThrow(
- "Missing --comm",
- );
+ await expect(new Bundle(new
Dag("second_try")).serve()).rejects.toThrow("Missing --comm");
});
});
- it("exports DagRegistry as the Dag collection a bundle serves", () => {
- const dag = new Dag("registry_api_dag");
+ it("exports Bundle as the thing that holds what a bundle provides and serves
it", () => {
+ const dag = new Dag("bundle_api_dag");
const handler = async () => "hello";
dag.task("extract", handler);
- // Building a registry starts nothing, so a test can dispatch through it
+ // Registering starts nothing, so a test can dispatch through a bundle
// exactly as the runtime does, with no sockets in scope.
- const registry = new DagRegistry(dag);
- expect(registry.getTaskHandler("registry_api_dag",
"extract")).toBe(handler);
- registry.register(new Dag("late_dag"));
- expect(registry.getTaskHandler("late_dag", "extract")).toBeUndefined();
+ const bundle = new Bundle(dag);
+ expect(bundle.getTaskHandler("bundle_api_dag", "extract")).toBe(handler);
+ bundle.register(new Dag("late_dag"));
+ expect(bundle.getTaskHandler("late_dag", "extract")).toBeUndefined();
});
- it("keeps registry enumeration out of the public surface", () => {
- const registry = new DagRegistry();
+ it("keeps bundle enumeration out of the public surface", () => {
+ const bundle = new Bundle();
for (const name of ["listTasks", "listDags"]) {
- expect(name in registry).toBe(false);
+ expect(name in bundle).toBe(false);
}
- expectTypeOf<keyof DagRegistry>().toEqualTypeOf<"register" |
"getTaskHandler">();
+ expectTypeOf<keyof Bundle>().toEqualTypeOf<"register" | "serve" |
"getTaskHandler">();
});
it("does not export the removed registerTask surface or the coordinator
itself", () => {
@@ -188,9 +188,18 @@ describe("public API", () => {
expect(connErr.connId).toBe("missing_conn");
});
- it("reaches the runtime only through serveDags, which takes a registry", ()
=> {
- expectTypeOf<typeof serveDags>().toEqualTypeOf<(registry: DagRegistry) =>
Promise<void>>();
- expectTypeOf<ConstructorParameters<typeof
DagRegistry>>().toEqualTypeOf<Dag[]>();
+ it("reaches the runtime only through bundle.serve(), which takes nothing",
() => {
+ // One verb in and one verb out: `serveDags` is gone, and the coordinator
+ // stays unnamed because the object that holds the Dags serves them itself.
+ expectTypeOf<Bundle["serve"]>().toEqualTypeOf<() => Promise<void>>();
+ expectTypeOf<Bundle["register"]>().toEqualTypeOf<(...items:
Registerable[]) => void>();
+ expectTypeOf<ConstructorParameters<typeof
Bundle>>().toEqualTypeOf<Registerable[]>();
+ expectTypeOf<Registerable>().toEqualTypeOf<Dag>();
+ for (const name of ["serveDags", "DagRegistry"]) {
+ expect(name in sdk).toBe(false);
+ }
+ expectTypeOf<typeof sdk>().not.toHaveProperty("serveDags");
+ expectTypeOf<typeof sdk>().not.toHaveProperty("DagRegistry");
expectTypeOf(SUPERVISOR_API_VERSION).toMatchTypeOf<string>();
});
@@ -317,10 +326,10 @@ describe("public API", () => {
dag.task("transform3", async () => undefined, { spec: { retries: 2 } });
// @ts-expect-error the TaskRef handle is data, not callable.
upstream();
- // @ts-expect-error serveDags takes the registry, not a bare Dag.
- serveDags(dag);
- // @ts-expect-error a registry is built from Dags, not from task handles.
- new DagRegistry(upstream);
+ // @ts-expect-error a bundle is built from Dags, not from task handles.
+ new Bundle(upstream);
+ // @ts-expect-error serve() takes nothing; the bundle already holds it
all.
+ new Bundle(dag).serve(dag);
};
void rejectsPositionalMisuse;
// @ts-expect-error the TaskRef handle is opaque and does not expose the
handler.
diff --git a/ts-sdk/tests/sdk/registry.test.ts b/ts-sdk/tests/sdk/bundle.test.ts
similarity index 57%
rename from ts-sdk/tests/sdk/registry.test.ts
rename to ts-sdk/tests/sdk/bundle.test.ts
index 201e559800c..e5725cb7810 100644
--- a/ts-sdk/tests/sdk/registry.test.ts
+++ b/ts-sdk/tests/sdk/bundle.test.ts
@@ -19,54 +19,54 @@
import { describe, it, expect } from "vitest";
import { Dag } from "../../src/sdk/dag.js";
-import { DagRegistry, listRegistryDags, listRegistryTasks } from
"../../src/sdk/registry.js";
+import { Bundle, listBundleDags, listBundleTasks } from
"../../src/sdk/bundle.js";
-describe("DagRegistry", () => {
+describe("Bundle", () => {
it("registers a Dag and retrieves its handlers", () => {
const handler = async () => "hello";
const dag = new Dag("example_dag");
dag.task("my_task", handler);
- const registry = new DagRegistry();
- registry.register(dag);
- expect(registry.getTaskHandler("example_dag", "my_task")).toBe(handler);
+ const bundle = new Bundle();
+ bundle.register(dag);
+ expect(bundle.getTaskHandler("example_dag", "my_task")).toBe(handler);
});
it("registers the Dags passed to its constructor", () => {
const handler = async () => "hello";
const dagA = new Dag("dag_a");
dagA.task("a", handler);
- const registry = new DagRegistry(dagA, new Dag("dag_b"));
- expect(registry.getTaskHandler("dag_a", "a")).toBe(handler);
- expect(listRegistryDags(registry)).toEqual([
+ const bundle = new Bundle(dagA, new Dag("dag_b"));
+ expect(bundle.getTaskHandler("dag_a", "a")).toBe(handler);
+ expect(listBundleDags(bundle)).toEqual([
{ dagId: "dag_a", tasks: ["a"] },
{ dagId: "dag_b", tasks: [] },
]);
});
it("rejects duplicate dagIds passed to the constructor", () => {
- expect(() => new DagRegistry(new Dag("example_dag"), new
Dag("example_dag"))).toThrowError(
+ expect(() => new Bundle(new Dag("example_dag"), new
Dag("example_dag"))).toThrowError(
/already registered/,
);
});
it("rejects constructor values that are not Dag instances", () => {
- expect(() => new DagRegistry({ dagId: "example_dag" } as unknown as
Dag)).toThrowError(
+ expect(() => new Bundle({ dagId: "example_dag" } as unknown as
Dag)).toThrowError(
/only Dag instances can be registered/,
);
});
it("returns undefined for unknown taskIds and dagIds", () => {
- const registry = new DagRegistry();
+ const bundle = new Bundle();
const dag = new Dag("example_dag");
dag.task("my_task", async () => undefined);
- registry.register(dag);
- expect(registry.getTaskHandler("example_dag", "nope")).toBeUndefined();
- expect(registry.getTaskHandler("unknown_dag", "my_task")).toBeUndefined();
+ bundle.register(dag);
+ expect(bundle.getTaskHandler("example_dag", "nope")).toBeUndefined();
+ expect(bundle.getTaskHandler("unknown_dag", "my_task")).toBeUndefined();
});
it("returns an empty list when no Dags are registered", () => {
- const registry = new DagRegistry();
- expect(listRegistryTasks(registry)).toEqual([]);
+ const bundle = new Bundle();
+ expect(listBundleTasks(bundle)).toEqual([]);
});
it("lists tasks across registered Dags", () => {
@@ -74,46 +74,46 @@ describe("DagRegistry", () => {
dagA.task("a", async () => undefined);
const dagB = new Dag("dag_b");
dagB.task("b", async () => undefined);
- const registry = new DagRegistry();
- registry.register(dagA, dagB);
- const registered = listRegistryTasks(registry);
+ const bundle = new Bundle();
+ bundle.register(dagA, dagB);
+ const registered = listBundleTasks(bundle);
expect(registered).toHaveLength(2);
expect(registered).toContainEqual({ dagId: "dag_a", taskId: "a" });
expect(registered).toContainEqual({ dagId: "dag_b", taskId: "b" });
});
it("rejects registering the same dagId in separate calls", () => {
- const registry = new DagRegistry();
- registry.register(new Dag("example_dag"));
- expect(() => registry.register(new
Dag("example_dag"))).toThrowError(/already registered/);
+ const bundle = new Bundle();
+ bundle.register(new Dag("example_dag"));
+ expect(() => bundle.register(new
Dag("example_dag"))).toThrowError(/already registered/);
});
it("rejects duplicate dagIds within a single call", () => {
- const registry = new DagRegistry();
- expect(() => registry.register(new Dag("example_dag"), new
Dag("example_dag"))).toThrowError(
+ const bundle = new Bundle();
+ expect(() => bundle.register(new Dag("example_dag"), new
Dag("example_dag"))).toThrowError(
/already registered/,
);
});
it("rejects registering the same Dag instance twice", () => {
- const registry = new DagRegistry();
+ const bundle = new Bundle();
const dag = new Dag("example_dag");
- registry.register(dag);
- expect(() => registry.register(dag)).toThrowError(/already registered/);
+ bundle.register(dag);
+ expect(() => bundle.register(dag)).toThrowError(/already registered/);
});
it("registers none of the Dags when a call throws", () => {
- const registry = new DagRegistry();
+ const bundle = new Bundle();
const dag = new Dag("dag_a");
dag.task("a", async () => undefined);
- expect(() => registry.register(dag, new
Dag("dag_a"))).toThrowError(/already registered/);
- expect(registry.getTaskHandler("dag_a", "a")).toBeUndefined();
- expect(listRegistryTasks(registry)).toEqual([]);
+ expect(() => bundle.register(dag, new Dag("dag_a"))).toThrowError(/already
registered/);
+ expect(bundle.getTaskHandler("dag_a", "a")).toBeUndefined();
+ expect(listBundleTasks(bundle)).toEqual([]);
});
it("rejects values that are not Dag instances", () => {
- const registry = new DagRegistry();
- expect(() => registry.register({ dagId: "example_dag" } as unknown as
Dag)).toThrowError(
+ const bundle = new Bundle();
+ expect(() => bundle.register({ dagId: "example_dag" } as unknown as
Dag)).toThrowError(
/only Dag instances can be registered/,
);
});
@@ -122,7 +122,7 @@ describe("DagRegistry", () => {
// Stands in for a Dag from a second resolved copy: same brand, other
class.
const foreign = { dagId: "foreign_dag" };
Object.defineProperty(foreign, Symbol.for("airflow.ts-sdk.Dag"), { value:
true });
- expect(() => new DagRegistry(foreign as unknown as Dag)).toThrowError(
+ expect(() => new Bundle(foreign as unknown as Dag)).toThrowError(
/different copy of apache-airflow-ts-sdk/,
);
});
@@ -131,24 +131,38 @@ describe("DagRegistry", () => {
const dagA = new Dag("dag_a");
dagA.task("a1", async () => undefined);
dagA.task("a2", async () => undefined);
- const registry = new DagRegistry();
- registry.register(dagA, new Dag("empty_dag"));
- expect(listRegistryDags(registry)).toEqual([
+ const bundle = new Bundle();
+ bundle.register(dagA, new Dag("empty_dag"));
+ expect(listBundleDags(bundle)).toEqual([
{ dagId: "dag_a", tasks: ["a1", "a2"] },
{ dagId: "empty_dag", tasks: [] },
]);
});
+ it("accepts a call that registers nothing", () => {
+ // `bundle.register(...maybeDags)` with an empty list should not need a
+ // guard at the call site.
+ const bundle = new Bundle();
+ expect(() => bundle.register()).not.toThrow();
+ expect(listBundleDags(bundle)).toEqual([]);
+ });
+
+ it("carries the brand its own serve guard reads", () => {
+ // The brand is how a bundle from a second resolved copy is told apart from
+ // a plain object; `serve()` reports the two differently.
+ expect(Symbol.for("airflow.ts-sdk.Bundle") in new Bundle()).toBe(true);
+ });
+
it("sees tasks added to a Dag after registration", () => {
- const registry = new DagRegistry();
+ const bundle = new Bundle();
const dag = new Dag("example_dag");
- registry.register(dag);
- expect(listRegistryTasks(registry)).toEqual([]);
+ bundle.register(dag);
+ expect(listBundleTasks(bundle)).toEqual([]);
const handler = async () => "late";
dag.task("late_task", handler);
- expect(registry.getTaskHandler("example_dag", "late_task")).toBe(handler);
- expect(listRegistryTasks(registry)).toContainEqual({
+ expect(bundle.getTaskHandler("example_dag", "late_task")).toBe(handler);
+ expect(listBundleTasks(bundle)).toContainEqual({
dagId: "example_dag",
taskId: "late_task",
});
diff --git a/ts-sdk/tests/sdk/dag.test.ts b/ts-sdk/tests/sdk/dag.test.ts
index 6177cf28b02..4dfe1652652 100644
--- a/ts-sdk/tests/sdk/dag.test.ts
+++ b/ts-sdk/tests/sdk/dag.test.ts
@@ -19,7 +19,7 @@
import { describe, it, expect } from "vitest";
import { Dag, getDagTaskRecords, type TaskRef } from "../../src/sdk/dag.js";
-import { DagRegistry } from "../../src/sdk/registry.js";
+import { Bundle } from "../../src/sdk/bundle.js";
describe("Dag", () => {
it("returns a frozen TaskRef handle with the Dag and task identity", () => {
@@ -202,19 +202,19 @@ describe("Dag", () => {
firstDag.task("extract", first);
secondDag.task("extract", second);
- const registry = new DagRegistry();
- registry.register(firstDag, secondDag);
- expect(registry.getTaskHandler("first_dag", "extract")).toBe(first);
- expect(registry.getTaskHandler("second_dag", "extract")).toBe(second);
+ const bundle = new Bundle();
+ bundle.register(firstDag, secondDag);
+ expect(bundle.getTaskHandler("first_dag", "extract")).toBe(first);
+ expect(bundle.getTaskHandler("second_dag", "extract")).toBe(second);
});
it("accepts a Unicode dagId that Python's word-character rule allows", () =>
{
const handler = async () => undefined;
const dag = new Dag("café_dag");
dag.task("任務", handler);
- const registry = new DagRegistry();
- registry.register(dag);
- expect(registry.getTaskHandler("café_dag", "任務")).toBe(handler);
+ const bundle = new Bundle();
+ bundle.register(dag);
+ expect(bundle.getTaskHandler("café_dag", "任務")).toBe(handler);
});
it("rejects non-function handlers", () => {
@@ -227,11 +227,11 @@ describe("Dag", () => {
it("treats a dotted TaskGroup taskId as a single taskId (group.task)", () =>
{
const dag = new Dag("example_dag");
dag.task("transforms.normalize", async () => "ok");
- const registry = new DagRegistry();
- registry.register(dag);
- expect(registry.getTaskHandler("example_dag",
"transforms.normalize")).toBeDefined();
+ const bundle = new Bundle();
+ bundle.register(dag);
+ expect(bundle.getTaskHandler("example_dag",
"transforms.normalize")).toBeDefined();
// Should NOT accidentally match the prefix alone
- expect(registry.getTaskHandler("example_dag",
"transforms")).toBeUndefined();
- expect(registry.getTaskHandler("example_dag",
"normalize")).toBeUndefined();
+ expect(bundle.getTaskHandler("example_dag", "transforms")).toBeUndefined();
+ expect(bundle.getTaskHandler("example_dag", "normalize")).toBeUndefined();
});
});
