diff --git a/task-sdk/src/airflow/sdk/execution_time/timeout.py b/task-sdk/src/airflow/sdk/execution_time/timeout.py index 925916ed4f947..6754ddd7754be 100644 --- a/task-sdk/src/airflow/sdk/execution_time/timeout.py +++ b/task-sdk/src/airflow/sdk/execution_time/timeout.py @@ -16,6 +16,7 @@ # under the License. from __future__ import annotations +import contextlib import os import structlog @@ -35,7 +36,11 @@ def __init__(self, seconds=1, error_message="Timeout"): def handle_timeout(self, signum, frame): """Log information and raises AirflowTaskTimeout.""" - self.log.error("Process timed out", pid=os.getpid()) + # Logging must never mask the timeout: task-runner logs go through + # a pipe owned by the supervisor, and a broken pipe here would + # replace AirflowTaskTimeout with BrokenPipeError. See #64212. + with contextlib.suppress(OSError): + self.log.error("Process timed out", pid=os.getpid()) raise AirflowTaskTimeout(self.error_message) def __enter__(self): diff --git a/task-sdk/tests/task_sdk/execution_time/test_timeout.py b/task-sdk/tests/task_sdk/execution_time/test_timeout.py index 94f7babd77d07..8545a2b0219d0 100644 --- a/task-sdk/tests/task_sdk/execution_time/test_timeout.py +++ b/task-sdk/tests/task_sdk/execution_time/test_timeout.py @@ -75,3 +75,35 @@ def test_timeout_happens(self): tp = TimeoutPosix(seconds=1, error_message="boom") with pytest.raises(AirflowTaskTimeout, match="boom"): tp.handle_timeout(signal.SIGALRM, None) + + def test_handle_timeout_raises_even_when_log_write_fails(self): + # Task-runner logs go through a pipe owned by the supervisor. If that + # pipe is broken when SIGALRM fires (e.g. supervisor already died), + # self.log.error raises BrokenPipeError inside the signal handler. + # Without a guard, that OSError replaces the AirflowTaskTimeout, so + # the task exits with the wrong exception class and skips the + # timeout-handling path in task_runner. See #64212. + if not hasattr(signal, "SIGALRM"): + pytest.skip("SIGALRM not supported on this platform") + + tp = TimeoutPosix(seconds=1, error_message="boom") + tp.log = mock.MagicMock() + tp.log.error.side_effect = BrokenPipeError("supervisor pipe closed") + + with pytest.raises(AirflowTaskTimeout, match="boom"): + tp.handle_timeout(signal.SIGALRM, None) + + tp.log.error.assert_called_once_with("Process timed out", pid=mock.ANY) + + def test_handle_timeout_normal_path_logs_and_raises(self): + """Regression guard: on the happy path the log is emitted AND AirflowTaskTimeout is raised.""" + if not hasattr(signal, "SIGALRM"): + pytest.skip("SIGALRM not supported on this platform") + + tp = TimeoutPosix(seconds=1, error_message="boom") + tp.log = mock.MagicMock() + + with pytest.raises(AirflowTaskTimeout, match="boom"): + tp.handle_timeout(signal.SIGALRM, None) + + tp.log.error.assert_called_once_with("Process timed out", pid=mock.ANY)