diff --git a/ts-sdk/adr/0001-mixed-lang-dag-interface.md b/ts-sdk/adr/0001-mixed-lang-dag-interface.md new file mode 100644 index 0000000000000..f12226c9f2781 --- /dev/null +++ b/ts-sdk/adr/0001-mixed-lang-dag-interface.md @@ -0,0 +1,182 @@ + + +# ADR-0001: Mixed-Lang Dag — TypeScript Task Handler Interface + +## Status + +Proposed. Revised after the review on #72047. + +## Decision + +1. TypeScript registers **task handlers, not Dags**. `new TaskHandler(dagId, taskId, handler)` binds a + handler to the Python-owned task it implements; `Dag` is exclusively the native case + ([ADR-0002](0002-native-dag-interface.md)). +2. **A bundle has one registration verb and serves itself.** `bundle.register(...)` takes Dags and + task handlers alike, in any mixture, and `await bundle.serve()` starts the runtime over them. + `Bundle` replaces `DagRegistry`, and the free `serveDags(registry)` function goes with it. +3. **task_id is always written out**; nothing is derived from the handler's function name. +4. **A handler is a plain function of its own data**, destructured by name. `getContext()` and + `getClient()` supply the rest, so nothing the SDK injects shares a namespace with an author's + arguments. +5. **Names bind by folding on both sides** — lowercased, separators removed — so Python's + `region_code` reaches a handler's `regionCode` with nothing declared. `withArgNames` is for a + genuine rename, never for a spelling difference. +6. **An upstream's return value is not a bound argument.** Read it with + `client.getXCom({ key: "return_value", taskId })`. + +## Context + +The Python `@task.stub` call site already defines task data flow. TypeScript tasks should consume +those bindings directly as named arguments instead of re-fetching each value with +`client.getXCom(...)`. A mixed-language Dag declares its structure in Python, and TypeScript supplies +task bodies and nothing else: it owns no dag_id, no schedule, no task order. + +This ADR covers only the TypeScript call-site interface. The argument-binding spec itself — its +shape, how it is materialized, how it travels over the wire — is a protocol-level decision recorded +in [`airflow-core/adr/lang-sdk/0007`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Example + +```ts +import { Bundle, TaskHandler, getClient, getContext } from "apache-airflow-ts-sdk"; + +interface TransformArgs { + regionCode: string; + threshold: number; +} + +async function transform({ regionCode, threshold }: TransformArgs) { + const client = getClient(); + const rows = await client.getXCom({ key: "return_value", taskId: "extract" }); + if (rows === null) { + throw new Error(`task ${getContext().taskId} has no upstream row count to transform`); + } + return { regionCode, passed: rows >= threshold }; +} + +const bundle = new Bundle(); +bundle.register(new TaskHandler("etl", "transform", transform)); +await bundle.serve(); +``` + +A bundle usually provides both kinds, and one call lists everything it exposes: + +```ts +bundle.register( + nativeEtl, // a Dag, from ADR-0002 + new TaskHandler("py_etl", "transform", transform), +); +``` + +### Wire names + +Arguments bind by folding both sides, so `region_code` reaches `regionCode`, `Name` reaches `name`, +and `s3_uri` reaches `s3Uri` with nothing declared on either side. The Go SDK folds the same way +(`strings.ToLower(strings.ReplaceAll(name, "_", ""))`), so one Python signature binds identically in +either SDK. + +```ts +// Python: def transform(region_code: str, s3_uri: str, threshold: float) +async function transform({ regionCode, s3Uri, threshold }: TransformArgs) { + // ... +} +``` + +`withArgNames` states a binding explicitly — its first argument is the mapping, its second the +handler. Folding absorbs spelling differences, so this is for a name the Python side never used: a +clearer word, or a TypeScript reserved word like `enum`. + +```ts +const report = withArgNames( + { label: "run_label" }, + async ({ label, transformed }: ReportArgs) => { + if (label !== "nightly") { + throw new Error(`expected run label "nightly" but got "${label}"`); + } + }, +); + +bundle.register(new TaskHandler("etl", "report", report)); +``` + +The map's keys are checked against the handler's own parameter type, so `{ labl: "run_label" }` is a +compile error naming the right key. Its values are Python names, which `tsc` cannot see and does not +check. + +## Consequences + +- One binding mechanism serves every mixed-language handler, and the Python call site stays the + single source of data-flow wiring. +- Folding matches the Go SDK's rule, so the same Python signature binds the same way in either SDK, + and neither needs a rename declared for ordinary snake_case parameters. +- A handler is directly unit-testable as a plain function of its data: no fixture to construct and no + intersection type to remember. +- This breaks 0.1.0-beta1 authors. `DagRegistry` becomes `Bundle`, and `TaskHandlerArgs` and the + `TaskHandler` type that takes it (`src/sdk/task.ts`) no longer describe a handler and are removed. + The shipped call sites change with the implementation — `src/index.ts`, and the `new Dag(...)` + + `dag.task(...)` pattern in `README.md`, `docs/index.md`, and `example/src/main.ts`. The package's + status line already reads "API may change". +- `serveDags(registry)` is removed in favour of `bundle.serve()`, since a bundle no longer holds + only Dags. Its name is also quoted in a user-facing error string + (`ts-sdk/src/cli/pack.ts:237`), which changes with it. + +## Alternatives + +- **Annotating the Python name on the field with a phantom type** + (`type Arg = T & { readonly __arg?: N }`), the way Java's `@ArgName` annotates + a `TaskInput` field. Rejected: an `interface` is erased, so the annotation cannot reach dispatch and + needs a runtime companion regardless. It also has two silent failure modes — + `Arg` collapses to a required `string`, because `undefined & object` is + `never`, and the phantom key appears in `keyof` for an object-valued argument. +- **Reading the expected names from `handler.toString()`** and parsing the destructuring pattern. + Property names do survive bundling, but a handler whose parameter is not destructured + (`async (a: ReportArgs) => a.runLabel`) exposes no names at all, so the check would disappear + silently for ordinary code. +- **Two registration verbs**, `registerDag(...dags)` beside `registerTaskHandler(dagId, taskId, fn)`. + Rejected once a task handler became a value carrying its own ids: the asymmetry that justified the + split — a Dag knows its id, a bare handler does not — disappears, and a bundle that provides both + kinds had to say so in two calls. + +## Appendix: Implementation Notes + +- **`register` widens rather than splits.** The shipped `DagRegistry.register(...dags: Dag[])` + (`ts-sdk/src/sdk/registry.ts`) already narrows each argument with `instanceof Dag` and rejects a + foreign copy by brand. `Bundle.register(...items: Registerable[])`, over + `type Registerable = Dag | TaskHandler`, follows the same path with one more arm — a discriminated + union being TypeScript's equivalent of the sealed interface the Go SDK uses for the same purpose. +- **`serve` is a method so the coordinator stays unnamed.** `startCoordinator` is deliberately not + exported — "Dag authors reach the runtime through `serveDags()`, and never name the coordinator + itself" (`ts-sdk/src/coordinator/index.ts`) — and a method on the object that already holds the + Dags and handlers keeps that intent while dropping the free function. +- **A task handler has no factory to call**, so wiring one the way a native task is wired + (`transform()`) is a compile error rather than a runtime throw. That is the guarantee an earlier + draft's separate `MixedLangDag` class existed to provide. +- **`getClient()` and `getContext()` read from an `AsyncLocalStorage` store** the runtime wraps around + the handler call, and throw outside a handler. TypeScript keeps type and value namespaces separate, + so `getContext()` coexists with the `TaskContext` type without either being renamed. +- **The bound argument object is a `Proxy`.** Destructuring a name triggers a lookup that folds that + name on demand and matches it against the folded wire names, so binding needs nothing declared + anywhere, and an entry in `withArgNames` takes precedence over folding. +- **An unmatched name is logged**, with both the requested name and the names actually delivered. It + cannot throw: a destructuring default (`{ runId = "manual" }`) is a legitimate miss, and the runtime + cannot tell one from a typo. +- `in` folds like a read. `Object.keys` and rest destructuring (`{ ...rest }`) yield Python's names, + since the SDK has no TypeScript-side names to enumerate. Two Python names that fold to the same + token fail the task at dispatch, naming both. diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md new file mode 100644 index 0000000000000..03c9a9e119745 --- /dev/null +++ b/ts-sdk/adr/0002-native-dag-interface.md @@ -0,0 +1,161 @@ + + +# ADR-0002: Native TypeScript Dag — Interface Design + +## Status + +Proposed. Revised after the review on #72047. + +## Decision + +1. **`dag.task(taskId, handler)` returns a factory.** Calling the factory both places the task in the + Dag and supplies its arguments by name — the shape Python TaskFlow uses for + `load(transformed=transform(...))`. +2. **The call graph is the task graph.** `tsc` checks every wired key against the handler's own + parameter type, and a `TaskRef` exists only once its producing call has returned, so a cycle + through arguments is unrepresentable rather than rejected by a validator. +3. **Every task is called exactly once.** An uncalled task fails when the Dag is read, so none can be + silently left out of the graph. +4. **`before` and `after` draw order-only edges** — the TypeScript pair for `>>` and `<<`, both + variadic so one call fans out. +5. **The Dag file owns Dag-level and task-level configuration.** `new Dag(dagId, spec)` carries the + schedule and the rest of `DagSpec`; `dag.task(taskId, handler, spec)` carries per-task options + such as retries. Python owns both in the mixed-language case + ([ADR-0001](0001-mixed-lang-dag-interface.md)), which is the difference between the two modes. +6. **A handler is a plain function of its own data**; `getContext()` and `getClient()` supply the rest. +7. **One registration verb**: `bundle.register(dag)`, the same call that takes task handlers, with + `await bundle.serve()` starting the runtime ([ADR-0001](0001-mixed-lang-dag-interface.md)). + +## Context + +A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so +TypeScript itself must express everything Python would otherwise own: the schedule and the rest of +the Dag-level configuration, each task's own options, the graph, and the task bodies. This ADR covers only what that +call site looks like for a user. `Dag` here is exclusively the native case; the mixed-language case +registers task handlers instead ([ADR-0001](0001-mixed-lang-dag-interface.md)). Both share the +protocol substrate recorded in +[`airflow-core/adr/lang-sdk/0007`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md). + +## Example + +```ts +import { Bundle, Dag, getClient } from "apache-airflow-ts-sdk"; + +const dag = new Dag("ts_etl", { schedule: "@daily", catchup: false, tags: ["etl"] }); + +const extract = dag.task("extract", async (): Promise => 42); + +const transform = dag.task("transform", async ({ extracted }: { extracted: number }) => extracted * 2); + +const load = dag.task( + "load", + async ({ transformed }: { transformed: number }) => { + await getClient().setXCom({ key: "loaded", value: transformed }); + }, + { retries: 2 }, +); + +const extracted = extract(); +const transformed = transform({ extracted }); +const loaded = load({ transformed }); + +const bundle = new Bundle(); +bundle.register(dag); +await bundle.serve(); +``` + +The schedule on the `Dag` and the retries on `load` are the point of a native Dag: nothing outside +this file declares them. A mixed-language handler cannot carry either, because the Python Dag it +belongs to already does. + +One statement per task, with each ref named, is the form to write. Nesting the calls +(`load({ transformed: transform({ extracted: extract() }) })`) is legal and equivalent, but it is +shorthand for a two-task chain, not the general shape: a Dag of twenty tasks reads as twenty flat +statements, never as a twenty-deep expression. + +### Order-only edges: `>>` and `<<` + +An edge that carries no data has no key to put in the wiring object, so it is drawn directly between +refs: + +```ts +const cleaned = cleanup(); + +cleaned.after(loaded, transformed); // [loaded, transformed] >> cleaned +loaded.before(cleaned); // loaded >> cleaned +``` + +Both return their own receiver, since a fan-out has no single "next" ref to hand back. Fan-*in* with +data is the wiring object itself — `summarize({ north: extractNorth(), south: extractSouth() })` — so +`[a, b] >> c` has an answer in each direction: named keys when values flow, `after` when only order +does. This matches `Before`/`After` in the Go SDK's native Dag interface, spelled to TypeScript +convention. + +## Consequences + +- One authoring surface (`dag.task()` plus its factory) covers the graph and each task's arguments, + and `before`/`after` cover edges that carry nothing. +- Handlers are unit-testable as plain functions of their data, with no SDK fixture to construct. +- `DagSpec` and `TaskSpec` are empty placeholders today — `Record` + (`ts-sdk/src/sdk/dag.ts`), so `new Dag("d", { schedule: "@daily" })` is currently a compile error + by design. Native declaration is what fills them, generated from the serialized-Dag JSON schema the + way `src/generated/supervisor.ts` is. This ADR does not choose those fields; it fixes where an + author writes them. +- `TaskOptions` collapses into `TaskSpec`. The shipped third argument to `dag.task` is + `{ inputs, spec }`; with wiring moved to the factory call, `inputs` is no longer an option and the + third argument is the spec itself. +- `TaskHandlerArgs` is removed from the public API, `DagRegistry` becomes `Bundle`, and + `serveDags(registry)` becomes `bundle.serve()`, which breaks + 0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for the shipped call sites + that change. + +## Alternatives + +- **Positional wiring** (`load(transform(extract()))`), which becomes expressible once data no longer + shares an object with `ctx`/`client`, since a handler can then take its arguments positionally and + `Parameters` is a real tuple. Rejected: it removes the key names from every call + site, and those names are what keeps flat, one-statement-per-task wiring readable at twenty tasks. + A handler may still take several positional arguments; only the *wiring* stays named. +- **Injected `ctx`/`client` arguments**, mimicking the Python signature. Rejected, per the above and + because feeling native to TypeScript matters more than matching Python's parameter list. + +## Appendix: Implementation Notes + +- **`getClient()` and `getContext()` read from an `AsyncLocalStorage`** (`node:async_hooks`) store + that the runtime wraps around the handler call, at the single dispatch site in + `ts-sdk/src/coordinator/runtime.ts`. The store propagates across every `await` and every promise + created inside that scope by construction. What it does not cover is work that outlives the + handler: a floating promise still calling `getClient()` after the handler resolved runs after the + task's success has been reported, which is equally true of a closed-over client today. +- **A handler's parameter type is exactly its own data.** An earlier draft merged `ctx`/`client` into + the same object, which forced every typed handler to declare `TArgs & TaskHandlerArgs` and left the + top-level argument namespace open to collisions with an author's own parameter names. Getters close + both. +- **The spec argument already has its slot.** `dag.task(taskId, handler, options)` reads + `{ inputs = {}, spec = {} }` and runs `validateEmptySpec` on the spec today + (`ts-sdk/src/sdk/dag.ts`), so task fields land on a path that exists rather than a new one. +- **A `TaskRef` is inert** — a handle for wiring, not a promise. Nothing in a Dag file executes a task + body. +- **`withArgNames` and the name folding behind it** ([ADR-0001](0001-mixed-lang-dag-interface.md)) + exist for the mixed-language case and are never needed here: both ends of every name are + TypeScript, so `tsc` checks the wiring end to end and there is no foreign name to reconcile. +- **`bundle.register(...)` takes any number of Dags and task handlers**, since each carries its own + ids; the earlier `registerDag`/`registerTaskHandler` split is recorded as a rejected alternative in + [ADR-0001](0001-mixed-lang-dag-interface.md).