Putting back the task.queued_duration metric Airflow 3 lost

Why this issue

This wasn’t my first pick. I’d started on #66726 (a SQLite backfill returning 500) and already had a branch. When I checked again before writing any code, someone had opened a PR for it that same morning. The issue had been quiet for six months, so I assumed it was safe. Lesson learned: an empty assignee field isn’t enough. Read the issue comments and search recent PRs too.

From the shortlist I made next, I picked #63503: after upgrading to Airflow 3, task.queued_duration was gone. It measures how long a task sits in the queue before a worker picks it up. The legacy name is dag.<dag_id>.<task_id>.queued_duration.

It’s close to my day job. dbt source freshness checks and Cloud Monitoring alerts are both about catching things that break without anyone noticing. The scope was right too: enough to learn the task instance lifecycle, not enough to eat a whole weekend.

Why it disappeared

There was exactly one place in the repo that sent this metric: TaskInstance.emit_state_change_metric. The only caller was the old LocalTaskJob flow, where the worker set the task to RUNNING itself.

In Airflow 3 the worker runs on the Task SDK. It no longer touches the database. It calls the ti_run Execution API endpoint and the API server sets RUNNING. The state still changes; nothing on the new path sends the metric. ti.start and ti.finish went missing the same way (#62019).

So the fix is obvious: send it from ti_run. The hard part is deciding when.

The guards in my first version, and how both went away

My first version copied the old logic: only send when end_date is empty. In other words, count the first run and skip retries.

On the day I opened the PR, henry3260 pointed out that a task resuming from deferral also comes back through ti_run, with end_date still empty, so it would be counted as a first run. He suggested adding next_method is None. I went through the code, agreed, wrote out a table of the three cases to be sure, and pushed the change the next day.

Both conditions were removed later.

• June: ashb said retries shouldn’t be skipped. Each retry is a new task instance row; the old try moves to history. The scheduler rewrites queued_dttm every time it queues a task, so a retry really did wait in the queue again. Skipping it undercounts.
• August: kaxil pointed out that henry’s condition made it disagree with task.scheduled_duration. Thinking it through again, deferral resumes should count too: after resuming, the task waits for a worker like anything else. Some operators (start_from_trigger=True) defer before they’re ever queued, so the resume is their only queue wait. Skip it and that task never gets a number.

The final rule is simple: one sample per queue wait. First run, retry, deferral resume and reschedule-mode sensor pokes all count. Only two cases skip it: runs that never went through the scheduler’s queue (like dag.test(), where queued_dttm is empty), and a duplicate start request from the client.

The guard I’d checked and agreed with was a patch on top of the old logic. I asked whether the condition was complete. I never asked whether it should exist at all.

Other things review caught

kaxil caught two more things I’d missed.

First, the tags should match the sibling metric, task.scheduled_duration. Mine didn’t, so you couldn’t slice the two the same way on a dashboard. Adding Dag tags also meant two extra lazy loads on every task start, so they’re now preloaded only when the metric will actually be sent and that setting is on.

Second, the timing. I sent the metric right after the state change, but a few later steps can still return a 5xx. The SDK retries a 5xx, the rollback puts the task back in QUEUED, and one wait gets counted twice. So it moved to the very end, just before return:

The tests came to 360 lines. After writing them I’d stash the production change and run them once to see them fail, then restore it and see them pass. That check is worth more than writing a few extra tests.

If you’re upgrading from Airflow 2

This doesn’t just bring the metric back; the numbers won’t match Airflow 2. Airflow 2 skipped retries and reschedule pokes (because end_date was still set). Now both count.

Why it took four and a half months

The review rounds above took a while on their own, and there was a lot of waiting in between. The milestone moved to 3.3.2 (“not critical”), then to 3.4.0. As a first-time contributor, every rebase meant asking a maintainer to approve CI again. In September someone commented asking whether I was still working on it. They needed the fix and offered to take it over.

In the last round kaxil said the code was fine, but the PR description still described the old guard. Airflow squash-merges, so the description becomes the commit message and has to match the final behaviour. I updated it, and it was merged the next day.

What I’ll remember next time

I wrote this PR with Claude Code (Opus 4.7) and disclosed it with Generated-by: in the PR, as Airflow’s guidelines ask.