Airflow 3.2 Worker Failures on TaskInstance Finish (HTTP 409 invalid_state)

Under which category would you file this issue?

Task SDK

Apache Airflow version

3.2.0. & 3.2.1

What happened and how to reproduce it?

Issue Description

The observed behavior starts after upgrading to Airflow 3.2.0 (and persists in 3.2.1) and disappears when downgrading to 3.1.8.

The failure manifests on Celery workers during the post-execution "finish" step (Task SDK / execution API), where the worker attempts to PATCH the TaskInstance state and receives:

HTTP 409
reason=invalid_state
message=TI was not in the running state so it cannot be updated
previous_state=success

This results in a worker-side exception (ServerResponseError) and causes the Celery workload to be treated as failed even when the underlying task logic has completed.

Airflow version(s) affected: 3.2.0, 3.2.1
Airflow version known-good: 3.1.8
Executor: CeleryExecutor
Deployment: Kubernetes
Python: 3.12
Database: Postgres

Typical worker log sequence:

Worker starts and loads secrets backend
Worker begins Task SDK API calls (often with retries)
API returns 409 invalid_state with previous_state=success
Supervisor raises ServerResponseError and the Celery task fails

Example:

Secrets backends loaded for worker ...
Starting call to 'airflow.sdk.api.client.Client.request', this is the 1st time calling it.
Starting call to ... this is the 2nd time calling it.
API server error ... status_code=409 ... previous_state='success'
airflow.sdk.api.client.ServerResponseError: Server returned error

Steps to reproduce

The issue appears intermittently; higher concurrency tends to increase the frequency.

Prerequisites

Airflow 3.2.0+
CeleryExecutor
Execution API enabled (Task SDK in use)

Deploy Airflow 3.2.0 or 3.2.1.
Trigger the DAG multiple times (or run multiple manual runs concurrently).
Observe worker logs for ServerResponseError with HTTP 409 invalid_state and previous_state=success.

What you think should happen instead?

Airflow workers should be able to finalise a task idempotently; if the API server has already moved the TaskInstance to a terminal state (e.g. success), the worker’s PATCH /execution/task-instances//state should not return a fatal 409 or fail the task.

Operating System

No response

Deployment

Official Apache Airflow Helm Chart

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

We noticed the problem where heartbeat DB queries queue up. This makes workers timeout on the heartbeat request and retry. I can imagine that the failed heartbeat can put task instance into a terminal state (but not sure though).

usename    state                xact_duration    query_duration    query
db_user  idle in transaction  00:00:22.17721   00:00:00.06658    SELECT task_instance.state AS task_instance_state
                                                                   FROM task_instance
                                                                   WHERE task_instance.id = '0<zzzyyy>54'::uuid::UUID
db_user  active               00:00:22.123495  00:00:22.120761   UPDATE task_instance SET last_heartbeat_at='2026-04-23T09:03:18.258390+00:00'::timestamptz, updated_at='2026-04-23T09:03:18.261693+00:00'::timestamptz WHERE task_instance.id
db_user  active               00:00:22.058237  00:00:22.054038   UPDATE task_instance SET last_heartbeat_at='2026-04-23T09:03:18.320462+00:00'::timestamptz, updated_at='2026-04-23T09:03:18.324080+00:00'::timestamptz WHERE task_instance.id
db_user  active               00:00:21.901896  00:00:21.89822    UPDATE task_instance SET last_heartbeat_at='2026-04-23T09:03:18.479643+00:00'::timestamptz, updated_at='2026-04-23T09:03:18.483318+00:00'::timestamptz WHERE task_instance.id
db_user  active               00:00:21.537818  00:00:21.534485   UPDATE task_instance SET last_heartbeat_at='2026-04-23T09:03:18.844164+00:00'::timestamptz, updated_at='2026-04-23T09:03:18.847447+00:00'::timestamptz WHERE task_instance.id
...

We noticed the problem where heartbeat DB queries queue up. This makes workers timeout on the heartbeat request and retry. I can imagine that the failed heartbeat can put task instance into a terminal state (but not sure though).

usename    state                xact_duration    query_duration    query
db_user  idle in transaction  00:00:22.17721   00:00:00.06658    SELECT task_instance.state AS task_instance_state
                                                                   FROM task_instance
                                                                   WHERE task_instance.id = '0<zzzyyy>54'::uuid::UUID
db_user  active               00:00:22.123495  00:00:22.120761   UPDATE task_instance SET last_heartbeat_at='2026-04-23T09:03:18.258390+00:00'::timestamptz, updated_at='2026-04-23T09:03:18.261693+00:00'::timestamptz WHERE task_instance.id
db_user  active               00:00:22.058237  00:00:22.054038   UPDATE task_instance SET last_heartbeat_at='2026-04-23T09:03:18.320462+00:00'::timestamptz, updated_at='2026-04-23T09:03:18.324080+00:00'::timestamptz WHERE task_instance.id
db_user  active               00:00:21.901896  00:00:21.89822    UPDATE task_instance SET last_heartbeat_at='2026-04-23T09:03:18.479643+00:00'::timestamptz, updated_at='2026-04-23T09:03:18.483318+00:00'::timestamptz WHERE task_instance.id
db_user  active               00:00:21.537818  00:00:21.534485   UPDATE task_instance SET last_heartbeat_at='2026-04-23T09:03:18.844164+00:00'::timestamptz, updated_at='2026-04-23T09:03:18.847447+00:00'::timestamptz WHERE task_instance.id
...

after further investigation it seems that the issue we are experiencing is the combination of the idle transactions (i do see idle transactions up to 50 secs durararion as well) and issue described here #63355

I patched this issue my MWAA Local runner in a kinda ugly way - I'm sure that someone with more experience can take this further.

inside airflow/sdk/api/client.py I replaced the floowing

 except ServerResponseError as e:
            if e.response.status_code == HTTPStatus.CONFLICT:
                detail = e.detail
                if (
                    isinstance(detail, dict)
                    and detail.get("reason") == "invalid_state"
                    and detail.get("previous_state") == "running"
                ):
                    raise TaskAlreadyRunningError(f"Task instance {id} is already running") from e
            raise

with

except ServerResponseError as e:
            if e.response.status_code in (HTTPStatus.CONFLICT, HTTPStatus.NOT_FOUND):
                # LOCAL PATCH: treat any 409 or 404 as TaskAlreadyRunningError.
                # SQS at-least-once delivery can cause duplicate messages. The original
                # code only handles previous_state=running 409s, but duplicate workers
                # can also see 409 with previous_state=success/failed, or 404 if the
                # task instance was already cleaned up. All of these should be ignored.
                raise TaskAlreadyRunningError(f"Task instance {id} is not in a startable state") from e
            raise

well I think things are the same under v3.1.8 with python3.10, worker logs as follow:

worker log

[2026-05-14T12:21:08.507917Z] {celery_executor_utils.py:193} INFO - [ed8da700-f2f2-48aa-b3b3-8faf4ffa3f49] Executing workload in Celery: token='eyJ***' ti=TaskInstance(id=UUID('019e266e-b5e5-7e40-b3ec-a6eba4da280e'), dag_version_id=UUID('019ce048-3d97-724d-bcc8-5d02fda96c3b'), task_id='upload_result', dag_id='prod_dump_input_set', run_id='prod_job_7866110', try_number=1, map_index=-1, pool_slots=1, queue='default', priority_weight=2, executor_config=None, parent_context_carrier={}, context_carrier={}) dag_rel_path=PurePosixPath('prod/dump_input_set.py') bundle_info=BundleInfo(name='dags-folder', version=None) log_path='dag_id=prod_dump_input_set/run_id=prod_job_7866110/task_id=upload_result/attempt=1.log' type='ExecuteTask'

we can see the TI UUID is: 019e266e-b5e5-7e40-b3ec-a6eba4da280e

worker traceback log

	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/sdk/api/client.py", line 226, in finish
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 1018, in update_task_state_if_needed
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/tenacity/__init__.py", line 331, in wrapped_f
	May 14th 2026, 20:21:20.365	    return copy(f, *args, **kw)
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/tenacity/__init__.py", line 371, in iter
	May 14th 2026, 20:21:20.365	    response = self._send_handling_auth(
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/httpx/_client.py", line 1218, in patch
	May 14th 2026, 20:21:20.365	    raise self._exception
	May 14th 2026, 20:21:20.365	    result = fn(*args, **kwargs)
	May 14th 2026, 20:21:20.365	    self.client.task_instances.finish(
	May 14th 2026, 20:21:20.365	    raise exc
	May 14th 2026, 20:21:20.365	    do = self.iter(retry_state=retry_state)
	May 14th 2026, 20:21:20.365	    return self.send(request, auth=auth, follow_redirects=follow_redirects)
	May 14th 2026, 20:21:20.365	  File "/usr/python/lib/python3.10/concurrent/futures/_base.py", line 451, in result
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/httpx/_client.py", line 942, in _send_handling_auth
	May 14th 2026, 20:21:20.365	    self.client.patch(f"task-instances/{id}/state", content=body.model_dump_json())
	May 14th 2026, 20:21:20.365	    result = action(retry_state)
	May 14th 2026, 20:21:20.365	    response = self._send_handling_redirects(
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/tenacity/__init__.py", line 473, in __call__
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/httpx/_client.py", line 999, in _send_handling_redirects
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/sdk/api/client.py", line 887, in request
	May 14th 2026, 20:21:20.365	    return get_json_error(response) or response.raise_for_status()
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/httpx/_client.py", line 982, in _send_handling_redirects
	May 14th 2026, 20:21:20.365	    exit_code = process.wait()
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/tenacity/__init__.py", line 470, in __call__
	May 14th 2026, 20:21:20.365	    return super().request(*args, **kwargs)
	May 14th 2026, 20:21:20.365	    raise err
	May 14th 2026, 20:21:20.365	    self.update_task_state_if_needed()
	May 14th 2026, 20:21:20.365	    hook(response)
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/httpx/_client.py", line 825, in request
	May 14th 2026, 20:21:20.365	    return self.__get_result()
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 1004, in wait
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/httpx/_client.py", line 914, in send
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/sdk/api/client.py", line 176, in get_json_error
	May 14th 2026, 20:21:20.365	    self._add_action_func(lambda rs: rs.outcome.result())
	May 14th 2026, 20:21:20.365	airflow.sdk.api.client.ServerResponseError: Server returned error
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/sdk/api/client.py", line 180, in raise_on_4xx_5xx
	May 14th 2026, 20:21:20.365	  File "/usr/python/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
	May 14th 2026, 20:21:20.365	  File "/home/airflow/.local/lib/python3.10/site-packages/tenacity/__init__.py", line 393, in <lambda>
	May 14th 2026, 20:21:20.365	    return self.request(
	May 14th 2026, 20:21:20.364	    supervise(
	May 14th 2026, 20:21:20.364	  File "/home/airflow/.local/lib/python3.10/site-packages/celery/app/trace.py", line 479, in trace_task
	May 14th 2026, 20:21:20.364	[2026-05-14T12:21:20.364122Z] {trace.py:285} ERROR - Task execute_workload[ed8da700-f2f2-48aa-b3b3-8faf4ffa3f49] raised unexpected: ServerResponseError('Server returned error')
	May 14th 2026, 20:21:20.364	  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 1993, in supervise
	May 14th 2026, 20:21:20.364	Traceback (most recent call last):
	May 14th 2026, 20:21:20.364	  File "/home/airflow/.local/lib/python3.10/site-packages/celery/app/trace.py", line 779, in __protected_call__
	May 14th 2026, 20:21:20.364	    R = retval = fun(*args, **kwargs)
	May 14th 2026, 20:21:20.364	    return self.run(*args, **kwargs)
	May 14th 2026, 20:21:20.364	  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/providers/celery/executors/celery_executor_utils.py", line 202, in execute_workload

check apiserver ingress log, status 409 found:

May 14th 2026, 20:21:20.363 | /execution/task-instances/019e266e-b5e5-7e40-b3ec-a6eba4da280e/state | 409
-- | -- | --
  | May 14th 2026, 20:21:08.707 | /execution/task-instances/019e266e-b5e5-7e40-b3ec-a6eba4da280e/rtif | 201
  | May 14th 2026, 20:21:08.647 | /execution/task-instances/019e266e-b5e5-7e40-b3ec-a6eba4da280e/heartbeat | 204
  | May 14th 2026, 20:21:08.622 | /execution/task-instances/019e266e-b5e5-7e40-b3ec-a6eba4da280e/run | 200

but here is something interesting, from ingress log, state update request was only sent once, however, we found update logs in 2 different apiserver pod:

May 14th 2026, 20:21:20.355 | INFO:     10.44.36.4:46524 - "PATCH /execution/task-instances/019e266e-b5e5-7e40-b3ec-a6eba4da280e/state HTTP/1.1" 409 Conflict | inferflow-airflow-api-server-585cc5559c-bmvfn
-- | -- | --
  | May 14th 2026, 20:21:20.351 | [2026-05-14T12:21:20.351643Z] {task_instances.py:385} WARNING - Cannot update Task Instance in invalid state previous_state=failed ti_id=019e266e-b5e5-7e40-b3ec-a6eba4da280e | inferflow-airflow-api-server-585cc5559c-bmvfn
  | May 14th 2026, 20:21:20.340 | [2026-05-14T12:21:20.340120Z] {task_instances.py:424} INFO - Task instance state updated new_state=failed rows_affected=1 ti_id=019e266e-b5e5-7e40-b3ec-a6eba4da280e | inferflow-airflow-api-server-585cc5559c-zdnvx

Adding a data point from an Astro Hosted deployment (Astronomer's managed Airflow), Runtime 3.1-15 (Airflow 3.1.8) — confirms this is not limited to 3.2.x, and confirms the concurrency-correlation observation with actual numbers.

Same exact signature, both variants seen in this thread:

Fatal: Starting call to 'airflow.sdk.api.client.Client.request'... → API server error, status_code=409, previous_state=success (supervisor.py:617, uncaught) → task exits with exit_code=1, final_state=up_for_retry, despite the task's own work having already completed successfully.
Non-fatal sibling, same root cause: heartbeat hits the same 409/previous_state=success conflict, caught by the existing ServerResponseError handler (supervisor.py:1115, "Server indicated the task shouldn't be running anymore") and just SIGTERMs the process — no crash, no misleading up_for_retry.

Confirms PR #63355 is the right shape of fix ("if API says TI is already in requested state... task runner will not fail the process") — verified it's not present in 3.1.8/3.2.x (git compare against the merge commit: diverged from 3.1.8, 3.2.0, 3.2.1, 3.2.2; first contained in 3.3.0).

New evidence on the concurrency correlation: cross-environment comparison over the same 24h window, filtering on this exact error signature — one high-task-density environment showed ~487 occurrences across its cluster, while a much lower-density environment (older Runtime, same platform, no shared code) showed only 5. On the DB side: RDS Performance Insights (Database Load, Top SQL) looked unremarkable around one occurrence, but that view is built from active/CPU-consuming sessions and wouldn't surface an idle-in-transaction pileup even if present. Broader CloudWatch RDS metrics for that same window show a real, if modest, correlation: write IOPS/throughput/transaction-log generation all spike right at the time of the failure, with DB load, connection count, and CPU climbing over the following several minutes (DB load ~14→20, connections ~252→316, CPU ~46%→53%). Not dramatic, but a genuine load increase lands right on the incident — consistent with the heartbeat-queuing theory raised earlier in this thread, though we can't confirm the specific mechanism (e.g., can't rule in/out idle-in-transaction sessions specifically). We're now watching pg_stat_activity live (state != 'idle', runtime > 5s) to try to catch it during a genuine load spike.

One detail from an earlier comment worth a second look: the report showing ingress logs with only one state-update request sent, yet two different apiserver pods logging update activity for the same TI. We saw something adjacent to this — two separate "...this is the 1st time calling it" tenacity-before log lines within ~300ms of each other in the same worker pod's log, for what should be non-competing calls. Neither of us has a full explanation for that piece yet; flagging in case it points at something below the request-retry layer (e.g. a stale/pooled connection landing on a different backend pod) rather than a pure application-level duplicate call.

Filed by Claude (Sonnet 5) on behalf of Sean Muth, from a support investigation on an Astro Hosted deployment.

@seanmuth

Root cause of the intermittent `409 invalid_state` on TI completion

Follow-up to the earlier comment with cross-environment concurrency correlation. We built a live instrumented repro (fanout burst of task completions under AstroExecutor, wrapping Client.request() and TaskInstanceOperations.succeed/finish/heartbeat to log full call-site stacks + monotonic sequence numbers) and reproduced the exact signature under load. Full mechanism, confirmed via direct stack-trace evidence on both sides of the call:

Sequence:

Child sends SucceedTask → Supervisor._handle_request() sets self._terminal_state = msg.state ("success") and calls client.task_instances.succeed() synchronously → PATCH .../state → apiserver returns 204, row updated, both sides confirm the round-trip.
Roughly ~600ms later, the same supervisor process calls update_task_state_if_needed() from wait(), which calls .finish() again → second PATCH .../state → apiserver correctly rejects it (previous_state=success) → 409.
This second call is unguarded (no try/except, unlike the graceful ServerResponseError handling inside handle_requests()), so the 409 propagates uncaught → "Task supervision failed" → task ends up exit_code=1, and gets marked up_for_retry even though the DB already correctly has success.

Why the guard that should prevent this (STATES_SENT_DIRECTLY, which includes SUCCESS) doesn't fire:

update_task_state_if_needed() checks self.final_state, not self._terminal_state directly. final_state's property body gates on self._exit_code first:

if self._exit_code == 0:
    return self._terminal_state or TaskInstanceState.SUCCESS
...
return TaskInstanceState.FAILED  # or UP_FOR_RETRY

_monitor_subprocess()'s loop (while self._exit_code is None or self._open_sockets:) only exits once the OS exit code is observed and every child socket (stdout/stderr/logs/requests) has reported EOF. Under scheduling delay, the non-blocking exit-code poll (_check_subprocess_exit, via psutil) can lose the race against socket closure: all four sockets can legitimately drain and close — including the requests socket having already delivered and processed SucceedTask — in a cycle where the exit-code poll simply hasn't yet observed the process's real exit_code=0.

wait() then does:

self._exit_code = self._exit_code if self._exit_code is not None else 1

silently defaulting a never-actually-observed exit code to 1. Because final_state checks self._exit_code == 0 before ever consulting self._terminal_state, this default discards the already-known-good, message-reported _terminal_state = "success" and recomputes a fallback state from the wrong exit code — landing outside STATES_SENT_DIRECTLY — which is what makes update_task_state_if_needed() decide to call .finish() a second time.

This is a precedence bug, not a missing guard: the supervisor already has authoritative information (the child told it directly, and the network round-trip for .succeed() confirmed it), but that information gets thrown away in favor of a defaulted, never-confirmed exit code.

Proposed fix — local ordering fix, no additional network/DB round-trip: trust self._terminal_state whenever a terminal message was actually received, independent of whether the exit code was genuinely observed or defaulted; only fall back to exit-code-derived state when no terminal message arrived at all:

@property
def final_state(self):
    if self._terminal_state is not None:
        return self._terminal_state
    if self._exit_code == 0:
        return TaskInstanceState.SUCCESS
    if self._should_retry:
        return TaskInstanceState.UP_FOR_RETRY
    return TaskInstanceState.FAILED

This differs from #63355 (merged, released 3.3.0), which swallows the resulting 409 at the API layer via an idempotency check — a correct and worthwhile hardening on its own, but a band-aid for the symptom. This fix addresses why the redundant call happens at all: it's a single-line reordering, purely local to the supervisor, costs nothing extra over the wire or in the DB.

Repro details available on request (anonymized instrumentation + burst-fanout DAG); happy to share as a reference if useful.

We'll open a PR with this fix.

Drafted-by: Claude Sonnet 5 (reviewed by @seanmuth)

Fix task incorrectly retried after a terminal state was already reported

@seanmuth

Correction and new evidence: the mechanism I described earlier was wrong

My comment above (and the framing in PR #73142's current docstring) described this as "an unobserved/defaulted exit code overriding an already-confirmed terminal state." That explanation was inference from source reading, not something I'd actually observed, and it doesn't hold up. I built live instrumentation (wrapping the relevant Client/ActivitySubprocess call sites in a plugin, no application behavior changed) and ran it under a deliberately CPU-saturated burst workload (10k mapped tasks, worker concurrency pushed past comfortable capacity) to catch a real occurrence. Retracting the earlier explanation and posting what actually happens, with full tracebacks.

What's actually happening

Two independent, separately-caused duplicate writes hit the same task instance, 20 seconds apart. Full apiserver-side timeline for one occurrence (times UTC, anonymized deployment):

14:16:41.554  PATCH .../run            -> 200
14:16:44.129  PUT   .../heartbeat       -> 204
14:16:44.404  PUT   .../rtif            -> 201
14:16:55.466  "Task instance state updated, new_state=success, rows_affected=1"
14:16:55.538  PATCH .../state           -> 204   (the real, successful .succeed() write)
14:16:58.254  "Cannot update Task Instance in invalid state, previous_state=success"
14:16:58.256  PATCH .../state           -> 409   (duplicate retry of the SAME .succeed() call, 2.7s later)
...
14:17:18.386  "Cannot update Task Instance in invalid state, previous_state=success"
14:17:18.406  PATCH .../state           -> 409   (a SEPARATE, later .finish() call)

First duplicate (the 204→409 pair, 2.7s apart): this is client-side tenacity retrying .succeed()'s PATCH after a timeout/connection hiccup on the client side, even though the server had already fully processed and committed the first physical request. Client.request's retry predicate only retries on httpx.RequestError or a 5xx — a 409 is neither, so once the retry lands on an already-success row, tenacity gives up immediately (reraise=True) and raises. This happens entirely inside Client.request(), upstream of any application-level state tracking.

Update: this is already fixed. PR #63355 added an idempotency short-circuit to this exact endpoint (airflow-core/.../execution_api/routes/task_instances.py): if the requested state already matches previous_state, it returns a plain 200 instead of raising the 409. That covers this first duplicate exactly (retried .succeed() requesting state=success against an already-success row). #63355 shipped in 3.3.0 alongside #66574, and — like #66574 and #67115 — is not present on the 3.1.x/3.2.x branch this was reproduced against.

Second duplicate (the later 409, and the one that actually crashes the worker): full traceback, identical across all 7 live occurrences I captured in one run (only ti_id and timestamp differ):

ServerResponseError: Server returned error

  File ".../astro_agent_client/execution/isolated/_astro/procs/worker.py", line 351, in _child_supervise_task
  File ".../airflow/sdk/execution_time/supervisor.py", line 1993, in supervise
  File ".../airflow/sdk/execution_time/supervisor.py", line 1004, in wait
  File ".../airflow/sdk/execution_time/supervisor.py", line 1018, in update_task_state_if_needed
  File ".../airflow/sdk/api/client.py", line 226, in finish
  File ".../httpx/_client.py", line 1218, in patch
  File ".../tenacity/__init__.py", line 338, in wrapped_f
  File ".../tenacity/__init__.py", line 477, in __call__
  File ".../tenacity/__init__.py", line 378, in iter
  File ".../tenacity/__init__.py", line 400, in <lambda>
  File ".../concurrent/futures/_base.py", line 449, in result
  File ".../concurrent/futures/_base.py", line 401, in __get_result
  File ".../tenacity/__init__.py", line 480, in __call__
  File ".../airflow/sdk/api/client.py", line 887, in request
  File ".../httpx/_client.py", line 825, in request
  File ".../httpx/_client.py", line 914, in send
  File ".../httpx/_client.py", line 942, in _send_handling_auth
  File ".../httpx/_client.py", line 999, in _send_handling_redirects
  File ".../httpx/_client.py", line 982, in _send_handling_redirects
  File ".../airflow/sdk/api/client.py", line 186, in raise_on_4xx_5xx_with_note
  File ".../airflow/sdk/api/client.py", line 176, in get_json_error

update_task_state_if_needed() computes final_state from exit_code in a way that doesn't correctly defer to the already-known _terminal_state='success' here, decides it needs to call .finish(), and that call hits the exact same "already success" guard — except this exception is genuinely uncaught in wait()/supervise(), and it's this one that kills task supervision. This is exactly the scenario PR #73142 targets. I traced it end to end against #73142's fix: with correct precedence, final_state resolves to SUCCESS, lands in STATES_SENT_DIRECTLY, and .finish() is never called — zero calls, not a caught failure.

I also want to flag PR #66574 ("Recover stuck TIs when direct terminal-state API call fails," in 3.3.0+) as related but distinct: it defers when _terminal_state gets set and gives update_task_state_if_needed() a safe replay path instead of the buggy fallback. In my traced scenario it would prevent the crash (the replay's failure is caught and logged rather than propagated) but it does not prevent the duplicate send — the replay itself re-issues .succeed() again, hitting the same 409 a third time. #73142 is the one that actually eliminates that second call outright.

Net: two independent bugs, three total duplicate/redundant PATCH .../state calls for one task completion, on one confirmed live occurrence. #73142 fixes the second (by not calling .finish() at all). #63355 already fixes the first (by making the endpoint a no-op when the retried request matches the already-persisted state). Both, along with #66574, ship together in 3.3.0; none are backported to the 3.1.x/3.2.x line this was reproduced against.

Drafted-by: Claude Code (Sonnet 5); reviewed by @seanmuth before posting

Task SDK: don't let exit code override a confirmed terminal state

Navigation Menu

Repository navigation

Airflow 3.2 Worker Failures on TaskInstance Finish (HTTP 409 invalid_state) #65708

Under which category would you file this issue?

Apache Airflow version

What happened and how to reproduce it?

Prerequisites

What you think should happen instead?

Operating System

Deployment

Apache Airflow Provider(s)

Versions of Apache Airflow Providers

Official Helm Chart version

Kubernetes Version

Helm Chart configuration

Docker Image customizations

Anything else?

Are you willing to submit PR?

Code of Conduct

Root cause of the intermittent `409 invalid_state` on TI completion

Correction and new evidence: the mechanism I described earlier was wrong

What's actually happening

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Airflow 3.2 Worker Failures on TaskInstance Finish (HTTP 409 invalid_state) #65708

Description

Under which category would you file this issue?

Apache Airflow version

What happened and how to reproduce it?

Prerequisites

What you think should happen instead?

Operating System

Deployment

Apache Airflow Provider(s)

Versions of Apache Airflow Providers

Official Helm Chart version

Kubernetes Version

Helm Chart configuration

Docker Image customizations

Anything else?

Are you willing to submit PR?

Code of Conduct

Activity

serhii-karelov commented on Apr 23, 2026

DominikBalga commented on Apr 23, 2026

KeeganCarter11 commented on May 8, 2026

Shimotsuk1-Rei commented on May 15, 2026

seanmuth commented on Sep 9, 2026

seanmuth commented on Sep 14, 2026

Root cause of the intermittent 409 invalid_state on TI completion

seanmuth commented on Sep 15, 2026

Correction and new evidence: the mechanism I described earlier was wrong

What's actually happening

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Root cause of the intermittent `409 invalid_state` on TI completion