Skip to content

Add TypeScript coordinator runtime#69079

Open
shivaam wants to merge 5 commits into
apache:mainfrom
shivaam:feat/add-ts-sdk-coordinator-runtime
Open

Add TypeScript coordinator runtime#69079
shivaam wants to merge 5 commits into
apache:mainfrom
shivaam:feat/add-ts-sdk-coordinator-runtime

Conversation

@shivaam

@shivaam shivaam commented Jun 27, 2026

Copy link
Copy Markdown
Contributor

Adds the TypeScript coordinator runtime to @apache-airflow/ts-sdk, the
follow-up to #67908 (TypeScript task SDK public interface). This lets Airflow
execute registered TypeScript task handlers through a bundled Node.js entrypoint.

  • Adds the coordinator runtime that speaks the supervisor wire protocol
    (length-prefixed msgpack frames) over the --comm and --logs sockets:
    connects, receives the task startup message, dispatches to the registered
    handler for the Dag/task pair, and reports the terminal task state.
  • Adds TaskClient-backed task-time data access: Variables (get / getOrThrow),
    XCom (get / set, including the automatic return_value push for non-undefined
    handler returns), and Connections.
  • Adds cooperative cancellation via ctx.signal (AbortSignal); SIGTERM/SIGINT
    abort the signal with a grace period before force-exit.
  • Adds generated supervisor wire types (src/generated/supervisor.ts) plus a
    generate:supervisor script, and a ./coordinator package export subpath.
  • Adds an example/ showing a Python stub Dag + TypeScript handlers, bundled
    with esbuild, and documents the bundle layout and Airflow coordinator config.

This is intentionally still Python-stub-Dag mode: TypeScript Dag declaration is
not introduced here. TypeScript registers handlers for Dag/task IDs declared by
Python stub tasks. The example bundles with esbuild and writes
airflow-metadata.yaml manually; folding metadata generation into the build
script is left to a separate follow-up.

Testing performed:

  • Unit tests and full e2e test with the airflow coordinator. Readme to run tests is available under examples.

Next steps

  • Combine airflow-metadata.yml with the .mjs file and introduce a ts pack tool.

@shivaam shivaam force-pushed the feat/add-ts-sdk-coordinator-runtime branch 3 times, most recently from 503a0e1 to cb139b3 Compare June 28, 2026 23:43
@shivaam shivaam marked this pull request as ready for review June 29, 2026 00:06
@shivaam

shivaam commented Jun 29, 2026

Copy link
Copy Markdown
Contributor Author

@uranusjr @jason810496 PR for TS sdk runtime.

shivaam added 5 commits June 28, 2026 21:16
The runtime should keep generated supervisor payloads as the source of truth while making the discriminators required where the coordinator constructs or narrows wire messages.
@shivaam shivaam force-pushed the feat/add-ts-sdk-coordinator-runtime branch from cb139b3 to 5de8276 Compare June 29, 2026 04:17
Comment on lines +97 to +105
this.pendingReplies.set(id, (frame) => {
this.logs?.debug("Response received", {
id,
request_type: type,
response_type: describeFrameType(frame.body),
error: frame.error ?? null,
});
resolve(frame);
});

@uranusjr uranusjr Jun 29, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if I missed anything. This seems to have the potential to be stuck forever (until the process is killed) if the frame it expects got dropped or corrupt for whatever reason. This should probably add a timeout or something.

logs.debug("Dispatching to handler", { task_id: ctx.taskId });

try {
const result = await handler(args);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If a cancel signal arrives after the result is actually sent, the handler’s result would have already been collected, but discard into the ether and never received by anyone. I would scope the abort signal to only apply if cancellable happened during the communication. Something like

try {
  const result = await handler(args);
  // Don't check isCancellationRequested here, just process.
} catch (err) {
  // Check here instead.
  if (isCancellationRequested()) {
    return buildCancellationResponse(details, logs, ctx);
  }
  // Same...
}

@uranusjr uranusjr Jun 29, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should have a pre-commit hook to ensure this file is up-to-date.

event: `[${this.name}] ${record.event}`,
timestamp: record.timestamp ?? new Date().toISOString(),
});
this.sock.write(Buffer.from(line + "\n", "utf8"));

@uranusjr uranusjr Jun 29, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Both Java and Go implement a buffer here to temporarily hold logs if sock is not connected. This is important because we might need to log before all sockets are fully connected, and those logs should still be sent after connection. (This can be hardened later.)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants