Execution API: Task deferrals do not enforce execution_timeout

[KushagraB424]

Under which category would you file this issue?

Airflow Core

Apache Airflow version

main

What happened and how to reproduce it?

When a task instance updates its state to DEFERRED via the Execution API, the system correctly records the trigger_timeout, but it fails to enforce the task's overall execution_timeout.

In airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py, the ti_update_state route handles TIDeferredStatePayload but leaves behind this TODO:
# TODO: HANDLE execution timeout later as it requires a call to the DB either get it from the serialised DAG or get it from the API

Without fetching and enforcing the execution_timeout here, tasks that enter a deferred state might completely ignore their configured maximum execution duration limits.

What you think should happen instead?

When processing a deferred state update, the Execution API should fetch the task's configured execution_timeout and validate it against the elapsed execution time or persist it correctly so the scheduler/supervisor can enforce it. This ensures that deferred tasks strictly abide by their execution_timeout constraints as they did in Airflow 2, preventing resource leakage or indefinitely stalled workflows.

Operating System

Not Applicable

Deployment

None

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

Not Applicable

Official Helm Chart version

Not Applicable

Kubernetes Version

Not Applicable

Helm Chart configuration

Not Applicable

Docker Image customizations

Not Applicable

Anything else?

Not Applicable

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