This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new a52b0a23321 Fix edge3 worker deadlock regression test on Python 3.14
(#66247)
a52b0a23321 is described below
commit a52b0a2332113c9f9033e63153051797bb1ee25b
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sat May 2 07:14:24 2026 +0200
Fix edge3 worker deadlock regression test on Python 3.14 (#66247)
The test relied on
`@patch("airflow.sdk.execution_time.supervisor.supervise")`
propagating into the worker subprocess via fork-inheritance. Python 3.14
flipped the POSIX default multiprocessing start method from `fork` to
`forkserver`, where children spawn from a fresh interpreter and parent-side
patches no longer apply — so `supervise()` ran for real and the test timed
out instead of triggering the queue-deadlock condition.
Patch `EdgeWorker._launch_job` and use an explicit `fork` context to spawn
a small top-level subprocess that pushes a >64 KB exception to the queue.
That reproduces the actual production deadlock (Queue feeder thread blocked
on the 64 KB OS pipe buffer with `is_alive()` still True) without depending
on the start method or on test-process patches surviving.
Verified locally on Python 3.10 (passes) and 3.14 (was timing out, now
passes); confirmed regression detection by temporarily reverting the
`worker.py` deadlock fix and observing the test fails.
---
.../edge3/tests/unit/edge3/cli/test_worker.py | 65 +++++++++++++++-------
1 file changed, 45 insertions(+), 20 deletions(-)
diff --git a/providers/edge3/tests/unit/edge3/cli/test_worker.py
b/providers/edge3/tests/unit/edge3/cli/test_worker.py
index d056d3af1e1..933793a4f06 100644
--- a/providers/edge3/tests/unit/edge3/cli/test_worker.py
+++ b/providers/edge3/tests/unit/edge3/cli/test_worker.py
@@ -20,6 +20,7 @@ import asyncio
import contextlib
import importlib
import json
+import multiprocessing
import signal
from datetime import datetime
from io import StringIO
@@ -79,6 +80,18 @@ MOCK_COMMAND = {
}
+def _emit_large_exception_target(results_queue):
+ """Worker-process target used by
``test_fetch_and_run_job_possible_deadlock``.
+
+ Pushes a >64 KB pickled exception to ``results_queue``. On Linux the OS
pipe
+ backing ``multiprocessing.Queue`` only has ~64 KB of buffer, so the queue's
+ feeder thread blocks on ``send_bytes`` and the subprocess can't terminate
+ until the parent reads from the queue — exactly the production deadlock
+ condition that #66144 fixed.
+ """
+ results_queue.put(Exception(f"Task execution failed with large error
message {'-' * 66000}"))
+
+
class _MockProcess(Process):
def __init__(self, returncode=None):
self.generated_returncode = None
@@ -310,7 +323,6 @@ class TestEdgeWorker:
assert len(worker_with_job.jobs) == 1 # no new job added (was removed
at the end...)
mock_logs_push.assert_not_called()
- @patch("airflow.sdk.execution_time.supervisor.supervise")
@patch("airflow.providers.edge3.cli.worker.jobs_fetch")
@patch("airflow.providers.edge3.cli.worker.jobs_set_state")
@patch("airflow.providers.edge3.cli.worker.EdgeWorker._push_logs_in_chunks")
@@ -322,14 +334,17 @@ class TestEdgeWorker:
mock_push_log_chunks,
mock_jobs_set_state,
mock_jobs_fetch,
- mock_supervise,
worker_with_job: EdgeWorker,
):
- """Verify that a large exception from the subprocess does not deadlock
fetch_and_run_job."""
-
- large_exception = Exception(f"Task execution failed with large error
message {'-' * 66000}")
- mock_supervise.side_effect = large_exception
-
+ """Verify that a large exception from the subprocess does not deadlock
fetch_and_run_job.
+
+ Uses an explicit ``fork`` context to spawn the simulated worker
subprocess. Python 3.14
+ flipped the POSIX default ``multiprocessing`` start method to
``forkserver``, which
+ spawns a fresh interpreter for the child — patches applied in the test
process do not
+ propagate, so older variants of this test that mocked ``supervise`` no
longer triggered
+ the deadlock condition. Forking a small top-level target sidesteps
that and reproduces
+ the actual queue-feeder/pipe-buffer deadlock the fix targets.
+ """
mock_jobs_fetch.side_effect = [
EdgeJobFetched(
dag_id="test",
@@ -342,22 +357,32 @@ class TestEdgeWorker:
),
None,
]
- worker_with_job.__dict__["_execution_api_server_url"] =
"https://mock-server/execution";
worker_with_job.concurrency = 1 # only one job at a time
assert worker_with_job.free_concurrency == 0
- try:
- await asyncio.wait_for(worker_with_job.fetch_and_run_job(),
timeout=10.0)
- except asyncio.TimeoutError:
- # Clean up any hanging subprocess to prevent blocking pytest
- for job in list(worker_with_job.jobs):
- if job.process.is_alive():
- job.process.terminate()
- job.process.join(timeout=1.0)
- if job.process.is_alive():
- job.process.kill()
- job.process.join()
- pytest.fail("fetch_and_run_job timed out after 10s - DEADLOCK
DETECTED. ")
+ ctx = multiprocessing.get_context("fork")
+ results_queue = ctx.Queue()
+ process = ctx.Process(target=_emit_large_exception_target,
args=(results_queue,))
+
+ with patch.object(EdgeWorker, "_launch_job", return_value=(process,
results_queue)):
+ process.start()
+ try:
+ await asyncio.wait_for(worker_with_job.fetch_and_run_job(),
timeout=10.0)
+ except asyncio.TimeoutError:
+ # Clean up any hanging subprocess to prevent blocking pytest
+ if process.is_alive():
+ process.terminate()
+ process.join(timeout=1.0)
+ if process.is_alive():
+ process.kill()
+ process.join()
+ pytest.fail("fetch_and_run_job timed out after 10s - DEADLOCK
DETECTED. ")
+ finally:
+ if process.is_alive():
+ process.join(timeout=1.0)
+ if process.is_alive():
+ process.terminate()
+ process.join(timeout=1.0)
# If we reach here without timeout, the deadlock was not triggered
assert mock_jobs_set_state.call_count >= 1
