TS SDK: replace registerTask with Dag and registerDags - #71144
TS SDK: replace registerTask with Dag and registerDags#71144jason810496 wants to merge 2 commits into
Conversation
dab6b7b to
1676164
Compare
|
What’s the plan to do the external vs pure TS dags after this? |
Along with #71213, the user should set the Additionally, we can introduce a |
| /** | ||
| * Opaque handle to a task registered on a {@link Dag}. | ||
| * | ||
| * Pass it as a downstream task's input to declare that the downstream task |
There was a problem hiding this comment.
inputs does not declare anything today, but this headline says it does, and the gap can produce wrong data rather than an error.
Grepping ts-sdk/src/, nothing outside this file reads TaskRecord.inputs: registry.ts reads .handler, .task and .keys(), manifest.ts reads only dagId/tasks. There is also no channel to deliver an input to a handler, since TaskHandlerArgs is still {ctx, client} (src/sdk/task.ts).
The reachable failure: a user who reads this headline concludes the value is wired, so writes await client.getXCom({ key: "return_value" }) with no taskId. Per src/sdk/client-types.ts:32-33, taskId defaults to the running task's context, so that reads the task's own XCom, returns null, and the task succeeds with the wrong value. No error, no warning, no failed task instance.
Neither ts-sdk/README.md nor typescript.rst mentions inputs, and the rst note next to the paragraph this PR edits still says dependencies are declared in the Python stub Dag, so autocomplete plus this JSDoc is the only discovery channel.
Either resolution works:
- Drop
inputs/TaskInputs/#validateInputsfor now and landDag+dag.task(taskId, handler)+registerDags, which is the whole refactor the title promises. Adding an optional trailingoptionsparameter later is not source-breaking in TypeScript, so waiting costs nothing. - Keep it, but lead the headline with what it does today (nothing), and add a
dag.taskoptions table to the rst markinginputsreserved and inert.
There was a problem hiding this comment.
I went with your second option. inputs, spec, DagSpec and TaskSpec now lead with "Reserved: inert today" on every declaration, the inputs docstring names the exact trap (client.getXCom without taskId reads the running task's own XCom), and explicitly mentioned intypescript.rst.
| export function listRegisteredTasks(): TaskRegistration[] { | ||
| return defaultRegistry.list(); | ||
| /** Record Dags in the default registry so the coordinator can run their tasks. */ | ||
| export function registerDags(...dags: Dag[]): void { |
There was a problem hiding this comment.
Splitting declaration from registration makes this third step forgettable, and the chain fails quietly when it is missed.
buildBundleManifest() reads defaultRegistry.listDags(), so a Dag that was constructed and populated but never passed here is absent from the manifest. runPack catches only the all-or-nothing case (registered no Dags), so registering 3 of 4 Dags packs successfully. NodeCoordinator._find_bundle ignores the manifest's dags map when selecting a bundle (task-sdk/src/airflow/sdk/coordinators/node/coordinator.py), so the bundle is still launched for the missing Dag's tasks, handleTask finds no handler, and it returns {type: "TaskState", state: "removed"}. The user sees task instances quietly marked removed, which normally means the task is no longer in the Dag, with only a warning line in the task log.
Fair caveat: forgetting import "./sales/tasks" under the old side-effect pattern registered nothing either, so this is a new instance of an existing mistake rather than a regression in kind. What is new is that it is now cheaply detectable, because the Dag is an object the SDK can see.
Cheapest fix that keeps this shape: have the Dag constructor push this onto a module-level list, emit it in --airflow-metadata mode, and have pack fail with Dag "billing" was declared but never passed to registerDags(...).
The structural version is worth a thought too. Java uses this same constructor shape (java.rst: var dag = new Dag("my_dag"); dag.addTask("fetch", FetchTask.class); return List.of(dag);) but returns the Dags from getDags(), so omission is unrepresentable; Go inverts ownership instead, with Registry.AddDag(dagId) Dag. A TypeScript analogue of either, startCoordinator({ dags: [salesDag, billingDag] }) or an addDag(dagId, spec?) that registers eagerly and returns the handle, removes the forgettable step rather than detecting it.
On naming, while the package is still unpublished: Task here is an opaque {dagId, taskId} identity, whereas Go's Task is the executable thing the runtime calls and Java's is the interface a task class implements. TaskRef or TaskHandle would keep the vocabulary aligned across the three SDKs and leave Task free for the real task object native Dag declaration will need.
There was a problem hiding this comment.
registerDags(...) is now the single entrypoint, since a Dag author never needs to know Airflow's coordinator at all. A second registerDags call is rejected, which would otherwise start a second runtime.
For the Dag instance omission, I prefer to let airflow-ts-pack warns instead of having restrict error (same as Python side behavior, it's fine to "define the Dag but don't construct it".
I agreed to rename Task to TaskRef to match the convention of Go and Java and avoid having the Task too ambigious.
| // eslint-disable-next-line @typescript-eslint/no-empty-object-type -- extension point for future native-Dag fields | ||
| export interface DagSpec {} |
There was a problem hiding this comment.
interface DagSpec {} is the empty object type, so spec accepts every non-nullish value, and the rule suppressed here is the one that reports exactly that. Today new Dag("d", 42) and dag.task("t", fn, { spec: { retries: 3 } }) both typecheck and store the value unvalidated. TaskSpec's own docstring advertises retries as a future field, so that second call is a natural thing to write and have silently ignored.
The repo has already taken a position on the spelling: airflow-core/src/airflow/ui/rules/typescript.js:668 sets @typescript-eslint/no-empty-object-type to ERROR, and its doc block lists type FooType = {} as incorrect while prescribing type FooType = object. That is the UI package's config rather than ts-sdk/eslint.config.js, so precedent rather than a rule this PR breaks, but the question has an in-repo answer.
The stated rationale also inverts. Once DagSpec gains its first field, necessarily optional, it becomes a weak type, and any call site passing an object with no overlapping property starts failing with TS2559. Every other extension the comment anticipates is already non-breaking: an optional trailing parameter, an optional property on TaskOptions (a type users pass and never implement), and Task becoming Task<TReturn = unknown> are all source-compatible. The only source-breaking change in that set is the one the placeholder introduces.
export type DagSpec = object rejects primitives and matches the spelling the repo prescribes; Record<string, never> also rejects {retries: 3}. Either one removes both eslint-disable lines.
One caveat on my side: I verified the lint policy and these declarations, but not the assignability behaviour with a tsc run, since ts-sdk/ has no installed node_modules in my checkout. Both are one-line checks.
There was a problem hiding this comment.
Both are now Record<string, never>, and both eslint-disable lines are gone. I ran the tsc check you flagged as unverified:
| written today | interface DagSpec {} |
type DagSpec = object |
Record<string, never> |
|---|---|---|---|
new Dag("d", 42) |
compiles | TS2345 | TS2345 |
new Dag("d", { schedule: "@daily" }) |
compiles, dropped | compiles, dropped | TS2322 |
object fixes only the primitive case, it still silently accepts the field a user would actually write. Your TS2559 point checks out too (but it can only bite call sites that pass a non-{} spec).
Record<string, never> makes unwritable, and {} stays assignable to an all-optional generated type.
| handler: TaskHandler<TReturn>, | ||
| options: TaskOptions = {}, | ||
| ): Task { | ||
| const { inputs = {}, spec = {} } = options; |
There was a problem hiding this comment.
This destructure accepts any other shape silently, which is out of step with the three checks around it. taskId, handler, and every inputs value are runtime-validated precisely because TypeScript can be bypassed, but dag.task("transform", fn, { input: { extracted } }) (singular typo) or { inpts: {...} } registers a task with no declared upstream and says nothing. Reachable from plain JavaScript, since airflow-ts-pack bundles whatever esbuild accepts, and from an as TaskOptions cast. This package already fails closed on the same class of mistake: parsePackArgs throws Unknown option ${arg}.
Worth fixing in the same place: the new comment at tests/public-api.test.ts:178 says "these constructor/method misuses also throw at runtime", and that is false for one of the five lines it covers. dag.task("transform2", async () => undefined, { upstream }) does not throw; it registers with zero inputs and discards the declared upstream. The other four do throw. Because the arrow function is never invoked, nothing catches it.
for (const key of Object.keys(options)) {
if (key !== "inputs" && key !== "spec") {
throw new Error(`Unknown option "${key}" for Dag "${this.dagId}" task "${taskId}"`);
}
}Also worth handling while here: options === null currently fails with an opaque "Cannot destructure property", because the default only applies to undefined.
There was a problem hiding this comment.
Added #validateOptions: unknown keys throw Unknown option "input" for Dag "d" task "t", and a non-object (including null and arrays) throws before the destructure, so no more opaque "Cannot destructure property".
Good catch on tests/public-api.test.ts:178 — that comment was false for the { upstream } line. It is now true rather than reworded, and a runtime test covers all three misuse shapes.
| // The line is whatever the bundle printed, so the per-Dag shape is not | ||
| // guaranteed by the type assertion above. | ||
| for (const [dagId, dag] of Object.entries(manifest.dags)) { | ||
| if (dag == null || !Array.isArray(dag.tasks)) { |
There was a problem hiding this comment.
This stops one level short of the schema it protects. Element types are unchecked, so {"dags":{"d":{"tasks":[null,{},"",123]}}} passes here and flows through renderMetadataYaml into the base64 trailer, while task-sdk/docs/airflow-metadata.schema.json defines dagEntry.tasks.items as {"type": "string", "minLength": 1}. The Go reference packer cannot emit that, because it decodes into Tasks []string, and the Python reader does not close the gap either: _bundle_metadata.py validates only that the document is a mapping and that sdk.supervisor_schema_version is a non-empty string, so bad task ids propagate past pack.
if (dag == null || !Array.isArray(dag.tasks) ||
dag.tasks.some((t) => typeof t !== "string" || t.length === 0)) {Cheap while you are here: the guard on the line above passes for an array, which Object.entries then turns into Dags named "0", "1". Pre-existing, but this loop is the natural place to close it.
There was a problem hiding this comment.
Both fixed. Task IDs are now checked as non-empty strings per airflow-metadata.schema.json, and the array case is rejected before Object.entries can turn it into Dags named "0", "1". Parametrized tests cover each malformed shape.
| const emptyDags = dagEntries | ||
| .filter(([, dag]) => dag.tasks.length === 0) | ||
| .map(([dagId]) => dagId); | ||
| if (emptyDags.length > 0) { |
There was a problem hiding this comment.
This is a policy divergence from the packer this file says it mirrors. For the same condition airflow-go-pack prints warning: dag %q has no tasks and continues (go-sdk/cmd/airflow-go-pack/pack.go), while this throws. The two agree on rejecting zero Dags, and the shared schema permits an empty list, since dagEntry.tasks has no minItems.
Failing closed is arguably the better call, and there is a forward-looking argument for it, since executable/coordinator.py claims bundle ownership from set(dags.keys()) alone and Node is slated to grow the same routing. But it means one placeholder Dag, or conditional attachment like if (process.env.FEATURE_X) dag.task(...), blocks the whole bundle build in TypeScript and not in Go. Worth picking one deliberately and applying it across both: match Go's warning, or hard-error in both and add "minItems": 1 to the schema.
There was a problem hiding this comment.
Matched all the handling behavior with Go side: warning: dag %q has no tasks and zero registered Dags is still a hard error.
registerTask bound one handler at a time to a Dag/task pair declared in a Python stub file, and left no object that could later carry a natively declared TypeScript Dag. Reworking only the authoring interface now means native Dag support and TaskFlow-style data passing can arrive without a second breaking change for users: a Dag instance keeps its spec and every task's handler, spec and declared inputs, which is what a future serialize() needs to emit the serialized Dag JSON. Producing that JSON stays out of scope, so Dag parsing still answers with no serialized Dags. The coordinator wire protocol and the bundle manifest shape are unchanged.
1676164 to
6bfacba
Compare
A Dag author has no reason to know Airflow's coordinator exists, and the
three-step shape left two steps that failed quietly when one was missed: a Dag
that was built but never registered was dropped from the packed bundle, and its
task instances showed up as *removed* at runtime with only a warning in the task
log. Folding the runtime handoff into registerDags removes one of those steps
outright; the packer now names the Dags that fell through the other.
Reserved options were the second quiet failure. `inputs` read as though it
declared a dependency, so a user could conclude the value was wired and call
getXCom without a taskId, which reads the running task's own XCom and returns
null without failing the task. Saying plainly that these fields are inert, and
accepting only `{}` until they are real, turns a wrong value into a compile
error. Generated specs will be all-optional types that `{}` still satisfies, so
filling them in later cannot break a call site.
Empty-Dag handling followed airflow-go-pack rather than diverging from it: one
placeholder Dag should not block a bundle build in TypeScript and not in Go.
6bfacba to
cedbd25
Compare
jason810496
left a comment
jason810496
left a comment
There was a problem hiding this comment.
Thanks kaxil for the review, I just resolved all the comments.
| /** | ||
| * Opaque handle to a task registered on a {@link Dag}. | ||
| * | ||
| * Pass it as a downstream task's input to declare that the downstream task |
There was a problem hiding this comment.
I went with your second option. inputs, spec, DagSpec and TaskSpec now lead with "Reserved: inert today" on every declaration, the inputs docstring names the exact trap (client.getXCom without taskId reads the running task's own XCom), and explicitly mentioned intypescript.rst.
| export function listRegisteredTasks(): TaskRegistration[] { | ||
| return defaultRegistry.list(); | ||
| /** Record Dags in the default registry so the coordinator can run their tasks. */ | ||
| export function registerDags(...dags: Dag[]): void { |
There was a problem hiding this comment.
registerDags(...) is now the single entrypoint, since a Dag author never needs to know Airflow's coordinator at all. A second registerDags call is rejected, which would otherwise start a second runtime.
For the Dag instance omission, I prefer to let airflow-ts-pack warns instead of having restrict error (same as Python side behavior, it's fine to "define the Dag but don't construct it".
I agreed to rename Task to TaskRef to match the convention of Go and Java and avoid having the Task too ambigious.
| // eslint-disable-next-line @typescript-eslint/no-empty-object-type -- extension point for future native-Dag fields | ||
| export interface DagSpec {} |
There was a problem hiding this comment.
Both are now Record<string, never>, and both eslint-disable lines are gone. I ran the tsc check you flagged as unverified:
| written today | interface DagSpec {} |
type DagSpec = object |
Record<string, never> |
|---|---|---|---|
new Dag("d", 42) |
compiles | TS2345 | TS2345 |
new Dag("d", { schedule: "@daily" }) |
compiles, dropped | compiles, dropped | TS2322 |
object fixes only the primitive case, it still silently accepts the field a user would actually write. Your TS2559 point checks out too (but it can only bite call sites that pass a non-{} spec).
Record<string, never> makes unwritable, and {} stays assignable to an all-optional generated type.
| handler: TaskHandler<TReturn>, | ||
| options: TaskOptions = {}, | ||
| ): Task { | ||
| const { inputs = {}, spec = {} } = options; |
There was a problem hiding this comment.
Added #validateOptions: unknown keys throw Unknown option "input" for Dag "d" task "t", and a non-object (including null and arrays) throws before the destructure, so no more opaque "Cannot destructure property".
Good catch on tests/public-api.test.ts:178 — that comment was false for the { upstream } line. It is now true rather than reworded, and a runtime test covers all three misuse shapes.
| // The line is whatever the bundle printed, so the per-Dag shape is not | ||
| // guaranteed by the type assertion above. | ||
| for (const [dagId, dag] of Object.entries(manifest.dags)) { | ||
| if (dag == null || !Array.isArray(dag.tasks)) { |
There was a problem hiding this comment.
Both fixed. Task IDs are now checked as non-empty strings per airflow-metadata.schema.json, and the array case is rejected before Object.entries can turn it into Dags named "0", "1". Parametrized tests cover each malformed shape.
| const emptyDags = dagEntries | ||
| .filter(([, dag]) => dag.tasks.length === 0) | ||
| .map(([dagId]) => dagId); | ||
| if (emptyDags.length > 0) { |
There was a problem hiding this comment.
Matched all the handling behavior with Go side: warning: dag %q has no tasks and zero registered Dags is still a hard error.
| dag.task("read_connection", readConnection); | ||
|
|
||
| await startCoordinator(); | ||
| await registerDags(dag); |
There was a problem hiding this comment.
I’m not sure how I feel with a function named registerDags handling sockets and a waiting server. Maybe use something more similar to the Java SDK such as serveDags()?
| // so a later mutation of their object would silently change what is packed. | ||
| // Shallow — a nested value in a future generated spec stays mutable. | ||
| this.spec = Object.freeze({ ...spec }); | ||
| declaredDagIds.add(dagId); |
There was a problem hiding this comment.
Also not a big fan with using a global variable like this, especially in the constructor. Why not make the registry a public interface instead, similar to how the Java SDK has a Bundle class?
|
Maybe we can do something like this instead const dag = new Dag(...);
// Add tasks...
const registry = new DagRegistry(dag);
registry.serve();Or, even more similar to Java SDK, |
3 participants
Why
registerTask({ dagId, taskId }, handler)binds one handler to a Python stub Dag/task pair at a time and leaves no object that can carry native TypeScript Dag declaration later (#69288). This PR changes only the user-facing authoring interface so it can grow into TaskFlow and native Dag support without another breaking change.How
Please noted that the spec at both Dag and Task level arguments (
DagSpec,TaskSpec) and the taskinputsare no-op, I intentionally placeholder them in this PR to prevent further breaking change.new Dag(dagId, spec?)takes a positional id plus an optional trailing spec object.dag.task(taskId, handler, options?)returns aTaskRefhandle, with placeholderinputsandspecas named options (the further native Dag:{ inputs: { extracted }, spec: { retries: 2 } }) so the future options need no new parameter breaking change. Unknown option keys are rejected, so a typo fails at import time instead of being silently ignored.DagSpecandTaskSpecareRecord<string, never>while they are placeholders, so a field that would be silently dropped does not compile. They will be all-optional types once real, which{}still satisfies, so filling them in later is not a breaking change either.dag.taskIdslists the attached task IDs, so a user can assert their handlers match the Python@task.stubnames.Daginstance retains its spec and each task's(taskId, handler, spec, inputs), so a futureserialize()can produce the serialized Dag JSON forDagFileParsingResult.serialized_dags.registerTask,listRegisteredTasks, andTaskRegistrationare replaced withregisterDags(...dags)as entrypoint. It both registers and runs the Dags, sostartCoordinatoris no longer part of the public API — a Dag author never needs to name the coordinator.airflow-ts-packwarns instead of failing when a registered Dag has no tasks, matchingairflow-go-pack, and warns by name when a Dag was built but never passed toregisterDags(...).Was generative AI tooling used to co-author this PR?