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 37df9a5a4ac Wait for the Lang-SDK runtime's descriptors to settle
before checking them (#74288)
37df9a5a4ac is described below
commit 37df9a5a4acffdbbe81094d60b2c0d7592fd0dbf
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Oct 5 22:11:38 2026 +0200
Wait for the Lang-SDK runtime's descriptors to settle before checking them
(#74288)
The test read the runtime's open descriptors as soon as the process was
called "sleep". The kernel renames the process during exec, before the
dynamic loader opens and closes the libraries it loads on the lowest free
descriptor, so the test intermittently saw a short-lived fd 3 and failed
although nothing was inherited. An inherited descriptor stays open, so
waiting for the set to settle keeps the check meaningful, and the
failure now shows what each descriptor points to.
Generated-by: Claude Opus 5
---
.../unit/dag_processing/test_lang_sdk_processor.py | 20 ++++++++++++++++++--
1 file changed, 18 insertions(+), 2 deletions(-)
diff --git a/airflow-core/tests/unit/dag_processing/test_lang_sdk_processor.py
b/airflow-core/tests/unit/dag_processing/test_lang_sdk_processor.py
index 879e43710ed..373f8c63ae9 100644
--- a/airflow-core/tests/unit/dag_processing/test_lang_sdk_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_lang_sdk_processor.py
@@ -88,6 +88,15 @@ def _get_open_fds() -> set[int]:
return {int(fd) for fd in os.listdir("/proc/self/fd")} if
os.path.isdir("/proc/self/fd") else set()
+def _read_fds(fd_dir: Path) -> dict[str, str]:
+ fds = {}
+ for fd in fd_dir.iterdir():
+ # A descriptor can close between listing the directory and reading its
link.
+ with contextlib.suppress(FileNotFoundError):
+ fds[fd.name] = os.readlink(fd)
+ return fds
+
+
def _is_running(pid: int) -> bool:
try:
# Init may not have reaped the killed process yet.
@@ -497,11 +506,18 @@ class TestLangSDKDagFileProcessorProcess:
assert time.monotonic() < deadline, "the runtime did not start"
proc._service_subprocess(max_wait_time=0.1)
fd_dir = Path(f"/proc/{proc.pid}/fd")
- fds = {fd.name: os.readlink(fd) for fd in fd_dir.iterdir()}
+ # The process takes its new name during exec, before the dynamic
loader has opened and
+ # closed the libraries it loads, so wait for such a short-lived
descriptor to go away.
+ # An inherited descriptor stays open for good.
+ settle_deadline = time.monotonic() + 5
+ while (fds := _read_fds(fd_dir)).keys() != {"0", "1", "2"} and (
+ time.monotonic() < settle_deadline
+ ):
+ time.sleep(0.05)
proc.kill(signal.SIGKILL)
proc.close()
- assert sorted(fds) == ["0", "1", "2"]
+ assert sorted(fds) == ["0", "1", "2"], fds
assert fds["0"] == "/dev/null"
