seanmuth commented on issue #65708:
URL: https://github.com/apache/airflow/issues/65708#issuecomment-5668081781
## 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:**
1. 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.
2. 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`.
3. 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*:
```python
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:
```python
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:
```python
@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 (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
