Honor execution timeout across task deferrals

[GautamSharma99]

PR body

Deferred tasks could exceed their configured execution_timeout because every resumed execution received the full timeout again. Their deferred trigger
deadline also ignored the task execution deadline.

This change:

• Sets the deferred deadline to the earlier of trigger_timeout and the task's execution deadline.
• Applies only the remaining execution time when a deferred task resumes.
• Immediately raises AirflowTaskTimeout when the timeout elapsed while the task was deferred.
• Adds regression coverage for execution-only, trigger-first, execution-first, and already-expired cases.

Tests:

• Execution API task instance tests: 190 passed, 4 skipped
• Task SDK execute callable tests: 8 passed
• Execution API suite: 521 passed, 4 skipped
• Ruff, formatting, mypy, and prek checks passed

---

Important

🛠️ Maintainer triage note for @GautamSharma99 · by @potiuk · 2026-07-28 16:32 UTC

Some review feedback from kaxil is waiting on you:

• There is 1 unresolved review thread on this PR from kaxil.

The ball is in your court — you've been assigned to this PR. Push a fix or reply in each thread explaining why the feedback does not apply, then mark them resolved and ping the reviewer (kaxil) for a final look.

See the Pull Request quality criteria for how to fix each item. There is no rush.

Note: your branch is 386 commits behind main — please rebase and push again to get up-to-date CI results.

_{Automated triage — may be imperfect; a maintainer takes the next look. We use this two-stage triage process so maintainers' limited time goes to the conversation with you.}

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[kaxil]

This deserializes the full SerializedDAG on every deferral just to read one task attribute (execution_timeout). ti_update_state with state=deferred is a hot path -- every deferrable-sensor poke and every deferrable-operator await hits it.

The sibling helper _handle_fail_fast_for_dag in this same file deliberately avoids exactly this:

# Check fail_fast from DagModel (simple column lookup) - early exit if False
# This avoids loading 5-50 MB SerializedDAG in 99% of cases
fail_fast = session.scalar(select(DagModel.fail_fast).where(DagModel.dag_id == dag_id))
if not fail_fast:
    return
# Only load SerializedDAG when fail_fast=True (rare case ~1%)
ser_dag = dag_bag.get_dag_for_run(dag_run=dr, session=session)

That guard only works because fail_fast is a DagModel column. execution_timeout is task-level and lives only in the serialized blob, so there's no cheap column to gate on -- every deferral here pays the full deserialization unconditionally, even for the tasks (the majority) that set no execution_timeout.

The worker already has the value with no DB hit: _run_execute_callable reads task.execution_timeout, and the worker already ships trigger_timeout=defer.timeout in the defer payload (task_runner.py:1494 -> TIDeferredStatePayload.trigger_timeout). Compute the execution deadline (or the remaining budget) on the worker at defer time and send it in the payload the way trigger_timeout already flows; the server then does min(...) with zero deserialization.

Tradeoff: a new payload field needs an Execution API version bump (Cadwyn), while the server-side load doesn't touch the wire. But a 5-50 MB deserialization on every defer isn't worth saving a versioned field, and it regresses the fail-fast optimization this file just added. The extra session.get(TI, ...) full-ORM reload (the route only selected the columns it needed) folds into the same fix.