Make the supervisor the only clock for execution_timeout #75394

Under which category would you file this issue?

Task SDK

Apache Airflow version

main

What happened and how to reproduce it?

Since #74869, the supervisor enforces execution_timeout for every task, using the value from TIRunContext. But it only acts killed_task_cleanup_time after the timeout, and the Python runner still keeps its own timers. With execution_timeout=10s and the default killed_task_cleanup_time=60:

Problems:

1. Language SDK tasks overrun by killed_task_cleanup_time (Language SDK runtimes do not handle execution_timeout in the task process #74717).
2. Python has several timers: the runner's SIGALRM, the supervisor backstop, and asyncio.wait_for in BaseAsyncOperator.execute, which reads task.execution_timeout and raises TimeoutError without calling on_kill().
3. The timeout is not recorded as the reason. A language SDK task stopped by SIGTERM reports a generic failure. If it exits without reporting and has retries left, the supervisor does not report UP_FOR_RETRY (it is in STATES_SENT_DIRECTLY), so the task instance stays running until the scheduler marks it "killed externally".
4. A task that swallows the cancellation succeeds, e.g. a Go task that returns nil after ctx.Done(). A Python task cannot do this by accident, since AirflowTaskTimeout is a BaseException.

What you think should happen instead?

The supervisor keeps the only clock, and the task process only reacts to a signal:

• Supervisor: at execution_timeout, signal the task process, then SIGKILL after killed_task_cleanup_time. Record that the task timed out, and report the outcome with the timeout as the reason, including RetryTask or a failed state when the runtime exits without reporting.
• Language SDK runtimes: no change. Go, TypeScript and Java (Make Java SDK context signal-aware #74690) already cancel the task's context on SIGTERM; they just get it at the deadline.
• Python task runner: drop the SIGALRM timer and the asyncio.wait_for timer, and raise AirflowTaskTimeout when the supervisor signals a timeout, so on_kill(), callbacks and retries behave as today.

A signal rather than a comms message, because the supervisor only sends frames in response to a request (comms.py), and all four runtimes would need a protocol change to handle a pushed message.

1. Which signal tells a Python task it timed out? SIGTERM already means a server-requested stop (on_kill() without AirflowTaskTimeout). SIGUSR1 is unused in the task process (it would need resetting in _reset_signals, like SIGUSR2). SIGALRM is taken by airflow.sdk.execution_time.timeout, which Dag parsing and providers use.
2. Should a task that reports success after its timeout be failed by the supervisor?
3. Should the TypeScript runtime's 30s ABORT_GRACE_PERIOD_MS follow killed_task_cleanup_time?

Out of scope: deferral (the runner's TODO: handle timeout in case of deferral), and letting a language SDK task opt out of [core] default_task_execution_timeout.

Operating System

No response

Deployment

None

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

No response

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct