Fix missing task.queued_duration metric in Airflow 3

[myps6415]

Summary

task.queued_duration (and its registry-derived legacy name dag.<dag_id>.<task_id>.queued_duration) stopped firing entirely after the Airflow 3 worker switched to the Task SDK / supervisor / Execution API.

The metric was only emitted by TaskInstance.emit_state_change_metric, which is only reachable from _check_and_change_state_before_execution — the legacy LocalTaskJob path. Airflow 3 workers flip TI state to RUNNING through the ti_run Execution API endpoint instead, which bypasses the emit site.

This is the same regression pattern as #62019 (missing ti.start / ti.finish).

Fix

Emit task.queued_duration from ti_run at the moment it transitions the TI from QUEUED to RUNNING. Skip the emit on transitions that are not the genuine first run of a try:

• retry of a previous attempt, identified by an existing end_date;
• resume from deferral, identified by next_method (the trigger sets it and end_date stays None across a deferral, so end_date alone cannot catch it);
• missing queued_dttm (rare race / test setups).

The legacy dotted name is emitted automatically by stats.timing via the metrics_template.yaml registry — no manual second call needed.

Test plan

• New test_ti_run_emits_queued_duration_metric confirmed to fail before the fix and pass after (verified by stashing the production change and re-running the test).
• New parametrized test_ti_run_skips_queued_duration_metric covers all three skip conditions (end_date set / deferral_resume / queued_dttm missing).
• All 33 existing TestTIRunState tests still pass.
• ruff format / ruff check / mypy-airflow-core / prek run --from-ref upstream/main --stage pre-commit all green.

closes: #63503
closes: #66067

---

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

• Yes — Claude Code (Opus 4.7)

Generated-by: Claude Code (Opus 4.7) following the guidelines

---

Important

🛠️ Maintainer triage note for @myps6415 · by @potiuk · 2026-07-02 17:46 UTC

Some review feedback from @ashb, @henry3260 is waiting on you:

• 3 unresolved review thread(s) from @ashb, @henry3260 need a reply or a fix.

The ball is in your court — you've been assigned to this PR. Reply or push a fix in each thread, then mark them resolved.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[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

[henry3260]

Thanks for the fix!

[henry3260]

@eladkal gentle ping! Could you help trigger CI?

[henry3260]

looks good from my side

[myps6415]

@ashb Gentle ping! This has approval from @henry3260, would appreciate a look when you have a moment. The 4 failing checks are unrelated flaky test_wasb failures.

[myps6415]

@ashb thanks for the review!

On the end_date guard (task_instances.py):
You're right. Since each retry is a separate TI row and the scheduler refreshes queued_dttm on every queueing, a retry genuinely waits in the queue and its utcnow() - queued_dttm is a valid sample (this also matches the legacy emit_state_change_metric, which fired on every transition to RUNNING). The end_date check was both fragile and semantically wrong, so I've dropped it — the guard is now just queued_dttm is not None and next_method is None, keeping only the deferral-resume skip. Also removed the now-unused TI.end_date from the select.

On the test:
Reworked it to assert the real invariant rather than the test setup: the emit case now has a retry parametrization where end_date is set on the row (the retry signature) and asserts the metric still fires — so a future regression that re-suppresses on end_date would fail here. The skip case now only covers deferral_resume and queued_dttm_missing.

[myps6415]

@ashb Gentle ping — I've addressed both your comments: dropped the end_date guard and reworked the test to assert the real invariant. Threads are resolved and CI is green. Would appreciate another look when you have a moment. Thanks!

[myps6415]

@kaxil all threads are resolved. The rebase reset the workflow approvals, so CI is sitting at action_required on 948cbcd — could you trigger it when you have a moment?

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this as it not critical

[kaxil]

This guard came in at @henry3260's request and it does what it says, but it leaves task.queued_duration and its sibling task.scheduled_duration disagreeing about which transitions count, in opposite directions. Taking the three ways a TI reaches RUNNING:

• First run: both metrics emit.
• Retry: only queued_duration. schedule_tis sets SCHEDULED, scheduled_dttm and try_number without clearing end_date, so the previous attempt's end_date is still on the row when the scheduler queues the TI, and emit_state_change_metric returns early.
• Deferral resume: only scheduled_duration. The DEFERRED update never touches end_date and ti_run already set it to None, so that guard passes, and the trigger refreshes scheduled_dttm, so the sample is a real measurement.

The resume's queue wait is equally real: queued_dttm is refreshed on every queueing and the critical section selects SCHEDULED with no next_method filter, so a deferrable task sits in QUEUED again waiting for a worker slot before execute_complete runs. Skipping it means that wait is never measured, and for sensor-heavy deployments the resume leg is where most of the queue time lives.

I would drop and ti.next_method is None and emit per queue wait, which is also where @ashb's end_date change pointed: one sample per real wait rather than one per try. The per-try reading is defensible too, but then the retry case should not emit either, and the comment should name the axis so the asymmetry reads as deliberate.

Either answer works for me, and this is the only thing I would like settled before merge. Adding these samples after release shifts percentiles for anyone alerting on the timer, which is cheap to decide now and awkward to change later.

On why this did not come up in my last pass: I was looking at the tag set then, and only walked the resume path through the scheduler this time.

[myps6415]

start_from_trigger=True operators are deferred straight from SCHEDULED, never queued, so
the resume this guard skips is their only queue wait — they emit nothing at all today.
Test added.

Also fixed my comment: legacy skipped retries and emitted on resume (it runs before
end_date is cleared), the inverse of what I claimed.

Per queue wait is the only axis that measures those, so that's my vote — your call.

[kaxil]

When metrics.dag_tags_in_metrics is on, dr.stats_tags costs two lazy loads per task start here: dag_tags_for_stats touches self.dag_model and then dag_model.tags, and the dr select above only eager-loads consumed_asset_events. The earlier query does join DagModel, but it selects DagModel.owners as a column rather than the entity, so nothing lands in the identity map and the lazy load still fires.

get_running_dag_runs_to_examine eager-loads exactly this to keep it out of the scheduler loop, and the comment on dag_tags_for_stats calls the remaining paths low frequency. ti_run is once per task start on the execution API, so it belongs with the first group. Adding the same eager load to the dr select, conditional on the config so the join is not paid when the feature is off, would cover it.

Non-blocking, and the config is off by default.

[myps6415]

Agreed — the earlier select takes DagModel.owners as a column, not the entity. Left out
to keep the diff on the blocking question; happy to add it here or in a follow-up.

[kaxil]

On the _team_name point from your last reply: it is the convention the scheduler already uses for exactly this, ti.dag_run._team_name = team plus two more sites, and stats_tags reads it through getattr. Setting dr._team_name = team_name alongside the dr.team_name = team_name above would make these three lines and the comment redundant, and keeps the tag set derived in one place if stats_tags grows again. Optional, since what you have produces the same tags today.

On the test: yes please, pin the multi-team leg with conf_vars. Your read of the gap is right, and get_team_name_for_ti returns None unless core.multi_team is on, so today that key agrees for the wrong reason on both sides.

[myps6415]

Test added with conf_vars, and it fails if the team tag is dropped.

_team_name: agreed — left out since it touches the same lines as the axis decision.