Fix TypeScript coordinator hanging forever without a supervisor greeting

[ColtenOuO]

Summary

CommChannel.connect() (ts-sdk/src/coordinator/comm-channel.ts) opens the comm TCP socket and then awaits channel.greeting.promise — the supervisor's first frame (StartupDetails or DagFileParseRequest) — with no timeout of its own. Once the TCP connection succeeds, nothing bounds how long the Node coordinator process will wait for that first frame: if the socket stays open but the supervisor never writes the greeting (a wedged or misbehaving supervisor, a stuck process on the Python side, a protocol bug), connect() never resolves and never rejects, and the coordinator subprocess hangs indefinitely with no way to recover on its own.

This is inconsistent with the rest of the same class: every subsequent request() call on the channel already times out after COORDINATOR_REQUEST_TIMEOUT_MS (30 seconds, ts-sdk/src/coordinator/comm-channel.ts:54) via Deferred.rejectAfter(), and sendResponse() supports an equivalent optional timeout that destroys the socket when a terminal write wedges. The greeting wait — which happens once, at startup, before any request/response traffic — was the one gap left uncovered.

Change

• Added ConnectOptions ({ timeoutMs?: number }) and a third parameter on CommChannel.connect().
• connect() now arms channel.greeting.rejectAfter(timeoutMs, ...) right after opening the socket, defaulting timeoutMs to the existing COORDINATOR_REQUEST_TIMEOUT_MS (30s) for consistency with the rest of the channel. This reuses the same self-clearing-timer Deferred mechanism already used by request() and sendResponse() — no new timer bookkeeping.
• On timeout, the socket is destroyed (sock.destroy(err)) with a descriptive error, mirroring the existing sendResponse timeout pattern, so a wedged supervisor connection is actually torn down instead of left dangling.
• A normal greeting arrival still resolves connect() immediately and clears the timer, so there's no behavior change on the working path.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Sonnet 5)

[jason810496]

Thanks for the PR.

IIUC, the purpose of the timeout here is to teardown the subprocess itself. In another word, self-destroy to avoid the subprocess being hanging forever.

The supervisor side (the coordinator interface) will ensure the Lang SDKs subprocess will be cleaned up:

airflow/task-sdk/src/airflow/sdk/coordinators/_subprocess.py

Lines 288 to 321 in be4a1e6

	@classmethod
	def start( # type: ignore[override]
	cls,
	*,
	what: TaskInstance,
	dag_rel_path: str | os.PathLike[str],
	bundle_info,
	logger: FilteringBoundLogger | None = None,
	sentry_integration: str = "",
	command: Sequence[str],
	subprocess_schema_version: str | None = None,
	startup_timeout: float = 10.0,
	**kwargs,
	) -> Self:
	with _ResourceTracker(timeout=startup_timeout) as tracker:
	comm_server, logs_server = tracker.track(_start_server(), _start_server())
	stdout_r, stdout_w = tracker.track(*socket.socketpair())
	stderr_r, stderr_w = tracker.track(*socket.socketpair())
	# A language SDK runtime cannot read Airflow's config, so propagate the
	# resolved log levels via the environment at launch. StartupDetails
	# arrives too late, the logs might already be produced by then.
	env = {
	**os.environ,
	"AIRFLOW__LOGGING__LOGGING_LEVEL": conf.get("logging", "logging_level", fallback="INFO"),
	"AIRFLOW__LOGGING__NAMESPACE_LEVELS": conf.get("logging", "namespace_levels", fallback=""),
	}
	proc = subprocess.Popen(
	[
	*command,
	"--comm={0[0]}:{0[1]}".format(comm_server.getsockname()),
	"--logs={0[0]}:{0[1]}".format(logs_server.getsockname()),
	],

airflow/task-sdk/src/airflow/sdk/coordinators/_subprocess.py

Lines 232 to 260 in be4a1e6

	@attrs.define(kw_only=True)
	class _ResourceTracker:
	"""
	Context manager that auto-closes tracked sockets and terminates tracked Popen objects.
	A subprocess startup is built up incrementally: bind sockets, spawn the
	child, accept its connections. If any step fails, the half-set-up state
	must be released. Calling :meth:`track` after each successful step records
	what to release; :meth:`untrack` removes ownership once another component
	(e.g. the activity subprocess instance) has taken over.
	"""
	timeout: float
	tracked: dict[int, socket.socket | subprocess.Popen] = attrs.field(init=False, factory=dict)
	def __enter__(self):
	return self
	def __exit__(self, *exc_info):
	for o in self.tracked.values():
	match o:
	case socket.socket():
	o.close()
	case subprocess.Popen():
	o.terminate()
	try:
	o.wait(self.timeout)
	except subprocess.TimeoutExpired:
	o.kill()

Additionally, the Execution API client on the supervisor side will retry the Execution API (which might takes over 30s) and the ti.run (or ti.patch) endpoint sometime take more than 30s. Forcing kill the subprocess itself terminates the chance of a success task execution.

Therefore I don't think this PR is necessary.