Task SDK: Supervisor IPC deadlocks child task runner on unvalidated request frames #76550

Under which category would you file this issue?

Task SDK

Apache Airflow version

3.4.0rc3 (and main)

What happened and how to reproduce it?

Description

When an IPC request frame sent from a task runner fails validation or deserialization on the supervisor (for example, due to a malformed payload, schema drift, or unrecognized discriminator), handle_requests logs the exception and runs continue without returning a response frame.

Because CommsDecoder.send() on the task runner blocks synchronously on socket.recv(4) waiting for a response, the child process deadlocks permanently. At the same time, the supervisor continues its event loop and sends regular heartbeats to the API server, so the scheduler's zombie detector never fires. The task stays stuck in RUNNING and holds worker capacity until manually killed.

Affected Code

In task-sdk/src/airflow/sdk/execution_time/supervisor.py:1176-1180:

try:
    msg = self.decoder.validate_python(self._deserialize_request(request.body))
except Exception:
    log.exception("Unable to decode message", body=request.body)
    continue

And in task-sdk/src/airflow/sdk/execution_time/comms.py:347-348:

frame = await asyncio.to_thread(self._read_frame)
return self._from_frame(frame)

Steps to Reproduce

1. Set up an ActivitySubprocess over a socketpair.
2. Send a request frame with an invalid body (e.g. {"type": "GetVariable"} without required keys).
3. The supervisor logs Unable to decode message and loops.
4. The child process thread hangs on recv(4) waiting for _ResponseFrame(id=request.id).
5. The supervisor continues heartbeating the API server.

Minimal reproduction script:

import socket
import uuid
from unittest.mock import Mock

import msgspec
from airflow.sdk.execution_time.comms import _RequestFrame
from airflow.sdk.execution_time.supervisor import (
    ActivitySubprocess,
    length_prefixed_frame_reader,
)

read_end, write_end = socket.socketpair()
ti_id = uuid.uuid4()
mock_client = Mock()

subprocess = ActivitySubprocess(
    id=ti_id,
    socket=read_end,
    client=mock_client,
    dag_rel_path="test_dag.py",
    bundle_info=Mock(),
)

gen = subprocess.handle_requests(log=Mock())
frame_reader = length_prefixed_frame_reader(
    gen, on_close=subprocess._on_socket_closed
)

req = _RequestFrame(id=101, body={"type": "InvalidMessage"})
encoder = msgspec.msgpack.Encoder()
payload = encoder.encode(req)
write_end.sendall(len(payload).to_bytes(4, "big") + payload)

frame_reader(read_end)

write_end.settimeout(2.0)
try:
    write_end.recv(4)
except TimeoutError:
    print(
        "DEADLOCK: Child blocked waiting for response frame that was never sent"
    )

Expected Behavior

If the supervisor cannot deserialize or validate request.body, it should send back an error response frame (_ResponseFrame(id=request.id, error=...)) so the task runner receives an error and fails or retries the task, rather than deadlocking the process.

Proposed Fix

In supervisor.py:handle_requests, send an error response before continuing:

try:
    msg = self.decoder.validate_python(self._deserialize_request(request.body))
except Exception as e:
    log.exception("Unable to decode message", body=request.body)
    err_resp = _ResponseFrame(
        id=request.id,
        error={"detail": f"Supervisor decoding failed: {e}"},
    )
    self._send_response(err_resp)
    continue

I have reproduced this with the test above and am working on a PR with unit tests to fix this.

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