diff --git a/providers/ssh/src/airflow/providers/ssh/utils/remote_job.py b/providers/ssh/src/airflow/providers/ssh/utils/remote_job.py index 496b0179d8913..d8d7d8ff3fd59 100644 --- a/providers/ssh/src/airflow/providers/ssh/utils/remote_job.py +++ b/providers/ssh/src/airflow/providers/ssh/utils/remote_job.py @@ -223,10 +223,14 @@ def build_posix_wrapper_command( exit 0 ' +# Redirect stdin from /dev/null too, not just stdout/stderr: a fresh setsid session +# leader that keeps the launching terminal on any fd re-acquires it as its controlling +# terminal, so a hangup (SSH session with a PTY dropping) would SIGHUP the detached job +# and defeat the whole point of running it in its own session. if command -v setsid >/dev/null 2>&1; then - setsid bash -c "$job_script" >/dev/null 2>&1 & + setsid bash -c "$job_script" /dev/null 2>&1 & else - nohup bash -c "$job_script" >/dev/null 2>&1 & + nohup bash -c "$job_script" /dev/null 2>&1 & fi echo "{paths.job_id}" """ diff --git a/providers/ssh/tests/unit/ssh/utils/test_remote_job.py b/providers/ssh/tests/unit/ssh/utils/test_remote_job.py index be93502057ee8..53c1f6f80b455 100644 --- a/providers/ssh/tests/unit/ssh/utils/test_remote_job.py +++ b/providers/ssh/tests/unit/ssh/utils/test_remote_job.py @@ -397,11 +397,6 @@ def test_kill_terminates_whole_job_tree(self, tmp_path): pgid = self._await_recorded_pid(paths) self._assert_kill_tears_down(paths, pgid, marker) - # Same environment-dependent process-group race that #69384 added reruns for on the - # sibling test; that marker was dropped in #69490 when this pty variant was written. - # The launch still depends on how the runner schedules the setsid fork, so keep main - # green on a fresh draw - the first attempt's assertion text stays in the CI log. - @pytest.mark.flaky(reruns=5) def test_kill_terminates_whole_job_tree_under_job_control(self, tmp_path): """With job control on, setsid(1) forks and the launcher's ``$!`` would name the short-lived setsid parent, not the job -- the condition the old wrapper orphaned @@ -414,8 +409,9 @@ def test_kill_terminates_whole_job_tree_under_job_control(self, tmp_path): self._run_bash_mc_under_pty( wrapper + "\necho SUBMIT_DONE\n", b"SUBMIT_DONE", - # The job records its pid only after setsid(2) has put it in its own session, so - # a non-empty pid file is proof the pty hangup below can no longer reach it. + # Hang up only once the job is up (pid file written), so the pgrep below sees a + # started job. The job survives the hangup regardless: the wrapper detaches its + # stdin from the terminal, so the setsid session never adopts the pty. detached=lambda: pid_path.exists() and bool(pid_path.read_text().strip()), ) pgid = self._await_recorded_pid(paths)