From cba1a35dc2cf8ba077a0e6b0a45758d33b191ae4 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 11:36:28 +0200 Subject: [PATCH 01/25] fix(hooks): eliminate macOS PTY output race condition Replace start_new_session=True with process_group=0 to prevent macOS PTY revocation on subprocess exit. Restructure the reader loop to always attempt os.read() before checking the stop flag, preventing event-loop scheduling starvation from skipping all reads. These two changes address the root cause of the flaky macOS PTY tests (#560, #733, #821, #826) rather than the symptoms. The drain retry logic from #826 is no longer needed and is removed. Closes #821 Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/jumpstarter/exporter/hooks.py | 40 +----- .../jumpstarter/exporter/hooks_test.py | 127 ------------------ 2 files changed, 6 insertions(+), 161 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index e7b509efb..08a16eb41 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -25,9 +25,7 @@ MAX_DRAIN_BYTES = 256 * 1024 DRAIN_TIMEOUT_SECONDS = 2.0 -DRAIN_MAX_EMPTY_POLLS = 10 -# Upper bound on hook-contributed motd content read from $JMP_MOTD_FILE. MAX_MOTD_BYTES = 64 * 1024 # Module-level reference to time.monotonic so tests can patch it without @@ -353,7 +351,7 @@ async def _execute_hook_process( # noqa: C901 stdout=child_fd, stderr=child_fd, env=hook_env, - start_new_session=True, # Equivalent to os.setsid() + process_group=0, close_fds=True, # Close inherited fds to prevent interference with gRPC connections ) except Exception as e: @@ -383,20 +381,12 @@ async def read_pty_output() -> None: # noqa: C901 start_time = _monotonic() try: - while not pty_state.reader_stop: + while True: try: - # Wait for fd to be readable with timeout with anyio.move_on_after(0.1): await anyio.wait_readable(parent_fd) - # Check stop flag immediately after timeout - # (main task may have signaled us to stop) - if pty_state.reader_stop: - logger.debug("read_pty_output: stop flag set, exiting") - break - read_count += 1 - # Log heartbeat every 2 seconds elapsed = _monotonic() - start_time if elapsed - last_heartbeat >= 2.0: logger.debug( @@ -404,27 +394,24 @@ async def read_pty_output() -> None: # noqa: C901 ) last_heartbeat = elapsed - # Read available data (non-blocking) try: chunk = os.read(parent_fd, 4096) if not chunk: - # EOF logger.debug("read_pty_output: EOF received") break buffer += chunk except BlockingIOError: - # No data available right now, continue loop + if pty_state.reader_stop: + logger.debug("read_pty_output: stop flag set and no data, exiting") + break continue except OSError as e: - # PTY closed or error logger.debug("read_pty_output: OSError on read: %s", e) break - # Process complete lines buffer = _flush_lines(buffer, output_lines) except OSError as e: - # PTY closed or read error logger.debug("read_pty_output: OSError in loop: %s", e) break finally: @@ -440,12 +427,7 @@ async def read_pty_output() -> None: # noqa: C901 try: drain_deadline = _monotonic() + DRAIN_TIMEOUT_SECONDS drained = 0 - consecutive_empty = 0 while drained < MAX_DRAIN_BYTES and _monotonic() < drain_deadline: - # Poll for readability with a short timeout. - # This avoids the race where a non-blocking read - # raises BlockingIOError because the macOS PTY - # kernel buffer hasn't delivered the data yet. remaining = drain_deadline - _monotonic() if remaining <= 0: break @@ -453,19 +435,9 @@ async def read_pty_output() -> None: # noqa: C901 try: readable, _, _ = select.select([parent_fd], [], [], timeout_s) except (ValueError, OSError): - # fd closed or invalid break if not readable: - # On macOS, data may not be available on the - # first select() call even though the subprocess - # has already written and exited. Keep retrying - # until we see several consecutive empty polls, - # which indicates the buffer is truly drained. - consecutive_empty += 1 - if consecutive_empty >= DRAIN_MAX_EMPTY_POLLS: - break - continue - consecutive_empty = 0 + break try: chunk = os.read(parent_fd, 4096) if not chunk: diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 01acee354..e11208466 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -8,7 +8,6 @@ from jumpstarter.common import HOOK_WARNING_PREFIX, ExporterStatus from jumpstarter.config.exporter import HookConfigV1Alpha1, HookInstanceConfigV1Alpha1 from jumpstarter.exporter.hooks import ( - DRAIN_MAX_EMPTY_POLLS, DRAIN_TIMEOUT_SECONDS, MAX_DRAIN_BYTES, MAX_MOTD_BYTES, @@ -1042,135 +1041,9 @@ def flush_lines_with_drain_error(buffer, output_lines): result = await executor.execute_before_lease_hook(lease_scope) assert result is None - @macos_pty_xfail - async def test_drain_retries_empty_select_then_captures_data(self, lease_scope) -> None: - """Verify that the drain retries after empty select() calls and still - captures data that arrives later. - - Patches select.select to return empty for the first N calls (where - N < DRAIN_MAX_EMPTY_POLLS), then reports the fd as readable. The - hook output should still be captured despite the initial empty polls. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker() - empty_count = 0 - empties_before_data = DRAIN_MAX_EMPTY_POLLS - 2 # e.g. 8 empties then data - - def select_with_delayed_ready(rlist, wlist, xlist, timeout=None): - nonlocal empty_count - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - empty_count += 1 - if empty_count <= empties_before_data: - return ([], [], []) # simulate delayed data - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo DELAYED_DRAIN_OK", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_with_delayed_ready), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("DELAYED_DRAIN_OK" in call for call in info_calls) - - @macos_pty_xfail - async def test_drain_terminates_after_max_empty_polls(self, lease_scope) -> None: - """Verify the drain loop terminates after DRAIN_MAX_EMPTY_POLLS - consecutive empty select() results. - - Patches select.select to always return empty during the drain phase. - The hook should still complete (no hang) and the drain data should - not appear since it's never read. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker(return_drain_data=False) - - def select_always_empty(rlist, wlist, xlist, timeout=None): - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - return ([], [], []) # always empty - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo MAX_EMPTY_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_always_empty), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - # Main loop should have captured the output before drain - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("MAX_EMPTY_TEST" in call for call in info_calls) - - @macos_pty_xfail - async def test_drain_empty_counter_resets_on_data(self, lease_scope) -> None: - """Verify the consecutive empty poll counter resets when data arrives. - - Simulates an empty-data-empty pattern during drain: a few empty polls, - then data becomes readable, then more empty polls. The counter should - reset after data is read, so the drain should tolerate more than - DRAIN_MAX_EMPTY_POLLS total empties as long as they are not consecutive. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker() - drain_select_call = 0 - # Pattern: 5 empties, then ready, then 5 more empties, then ready - # Total empties (10) >= DRAIN_MAX_EMPTY_POLLS but never consecutive - pattern = [False] * 5 + [True] + [False] * 5 + [True] - - def select_with_interleaved_empties(rlist, wlist, xlist, timeout=None): - nonlocal drain_select_call - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - idx = drain_select_call - drain_select_call += 1 - if idx < len(pattern) and not pattern[idx]: - return ([], [], []) - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo INTERLEAVE_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_with_interleaved_empties), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("INTERLEAVE_TEST" in call for call in info_calls) - async def test_drain_constants_are_reasonable(self) -> None: assert MAX_DRAIN_BYTES == 256 * 1024 assert DRAIN_TIMEOUT_SECONDS == 2.0 - assert DRAIN_MAX_EMPTY_POLLS == 10 async def test_exec_default_is_none(self) -> None: """Test that the default exec is None (auto-detect).""" From 14099b8e86b748967804bffab06b307190218468 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 11:42:23 +0200 Subject: [PATCH 02/25] ci: temporarily enable macOS tests on PRs to validate PTY fix Run the full test matrix (all Python versions, Linux + macOS) on PRs to confirm the PTY race condition fix passes consistently on macOS. This will be reverted once 20 consecutive macOS passes are confirmed. Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/python-tests.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/python-tests.yaml b/.github/workflows/python-tests.yaml index 12233ae0f..ee91e4d2f 100644 --- a/.github/workflows/python-tests.yaml +++ b/.github/workflows/python-tests.yaml @@ -41,8 +41,8 @@ jobs: # Merge queue and workflow_dispatch run the full matrix # (all Python versions on both Linux and macOS). if [[ "${{ github.event_name }}" == "pull_request" ]]; then - echo 'python-versions=["3.12"]' >> "$GITHUB_OUTPUT" - echo 'runners=["ubuntu-24.04"]' >> "$GITHUB_OUTPUT" + echo 'python-versions=["3.11", "3.12", "3.13"]' >> "$GITHUB_OUTPUT" + echo 'runners=["ubuntu-24.04", "macos-15"]' >> "$GITHUB_OUTPUT" else echo 'python-versions=["3.11", "3.12", "3.13"]' >> "$GITHUB_OUTPUT" echo 'runners=["ubuntu-24.04", "macos-15"]' >> "$GITHUB_OUTPUT" From 7fefc93de6f9d06f26e690572534197255a6dc69 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 11:50:44 +0200 Subject: [PATCH 03/25] test(hooks): cover reader_stop + BlockingIOError exit path Add test for the grandchild scenario where the PTY slave is held open after the direct child exits. The reader gets BlockingIOError (no data) and exits via the stop flag. This covers lines 357-358 in hooks.py to satisfy the diff-coverage threshold. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/exporter/hooks_test.py | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index e11208466..024c4dda2 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -1041,6 +1041,49 @@ def flush_lines_with_drain_error(buffer, output_lines): result = await executor.execute_before_lease_hook(lease_scope) assert result is None + async def test_reader_exits_on_stop_flag_when_no_data(self, lease_scope) -> None: + """Verify the reader exits via the stop flag when os.read raises + BlockingIOError (no data available) and reader_stop is True. + + This covers the grandchild scenario where the PTY slave is held + open after the direct child exits: the main loop gets + BlockingIOError because no new data is being written, and the + stop flag causes it to exit cleanly. + """ + state = _PtyTracker(return_drain_data=False) + + def os_read_blocking_after_eof(fd, size): + if fd != state.parent_fd: + return state._original_os_read(fd, size) + if not state.eof_seen: + try: + data = state._original_os_read(fd, size) + except (BlockingIOError, OSError): + state.eof_seen = True + raise + if not data: + state.eof_seen = True + raise BlockingIOError("simulated grandchild holding PTY open") + return data + raise BlockingIOError("simulated grandchild holding PTY open") + + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo STOP_FLAG_TEST", timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with ( + patch("pty.openpty", side_effect=state.tracking_openpty), + patch("os.read", side_effect=os_read_blocking_after_eof), + patch("jumpstarter.exporter.hooks.logger") as mock_logger, + ): + result = await executor.execute_before_lease_hook(lease_scope) + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("STOP_FLAG_TEST" in call for call in info_calls) + async def test_drain_constants_are_reasonable(self) -> None: assert MAX_DRAIN_BYTES == 256 * 1024 assert DRAIN_TIMEOUT_SECONDS == 2.0 From 98326f735f5088e01ce7d550f4a1d020a376654f Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 12:01:33 +0200 Subject: [PATCH 04/25] test(hooks): use real grandchild process to cover reader_stop path The previous test mocked os.read to simulate BlockingIOError, but on Linux the reader gets EOF before BlockingIOError. Use a real backgrounded subprocess (sleep &) to hold the PTY slave open, which forces the reader to exit via the reader_stop + BlockingIOError path. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/exporter/hooks_test.py | 41 +++++-------------- 1 file changed, 10 insertions(+), 31 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 024c4dda2..b637a1e1a 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -1041,44 +1041,23 @@ def flush_lines_with_drain_error(buffer, output_lines): result = await executor.execute_before_lease_hook(lease_scope) assert result is None - async def test_reader_exits_on_stop_flag_when_no_data(self, lease_scope) -> None: - """Verify the reader exits via the stop flag when os.read raises - BlockingIOError (no data available) and reader_stop is True. - - This covers the grandchild scenario where the PTY slave is held - open after the direct child exits: the main loop gets - BlockingIOError because no new data is being written, and the - stop flag causes it to exit cleanly. - """ - state = _PtyTracker(return_drain_data=False) - - def os_read_blocking_after_eof(fd, size): - if fd != state.parent_fd: - return state._original_os_read(fd, size) - if not state.eof_seen: - try: - data = state._original_os_read(fd, size) - except (BlockingIOError, OSError): - state.eof_seen = True - raise - if not data: - state.eof_seen = True - raise BlockingIOError("simulated grandchild holding PTY open") - return data - raise BlockingIOError("simulated grandchild holding PTY open") + async def test_reader_exits_on_stop_flag_when_grandchild_holds_pty(self, lease_scope) -> None: + """Verify the reader exits via the stop flag when a grandchild + process holds the PTY slave open after the direct child exits. + The backgrounded sleep inherits the PTY slave fd, preventing EOF + on the master. The reader gets BlockingIOError (no data from the + silent grandchild) and exits once reader_stop is set. + """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( - script="echo STOP_FLAG_TEST", timeout=10, + script="echo STOP_FLAG_TEST; sleep 300 &", + timeout=10, ), ) executor = HookExecutor(config=hook_config) - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=os_read_blocking_after_eof), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: result = await executor.execute_before_lease_hook(lease_scope) assert result is None info_calls = [str(call) for call in mock_logger.info.call_args_list] From 82d3b5897181a4f3c032df9b6e3c58b139f3231d Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 12:47:32 +0200 Subject: [PATCH 05/25] ci: retrigger #2 From 40497e116963eefa4b7c77fdbfe682ddb80dfece Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 13:01:47 +0200 Subject: [PATCH 06/25] fix(test): don't set eof_seen on transient BlockingIOError in PtyTracker BlockingIOError means "no data yet" (transient), not EOF. Setting eof_seen on BlockingIOError caused the mock to inject drain data during the main reader loop on macOS, where BlockingIOError can occur before the real EOF arrives. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../packages/jumpstarter/jumpstarter/exporter/hooks_test.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index b637a1e1a..1a2ef25d7 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -59,7 +59,9 @@ def os_read_with_drain_data(self, fd, size): if not self.eof_seen: try: data = self._original_os_read(fd, size) - except (BlockingIOError, OSError): + except BlockingIOError: + raise + except OSError: self.eof_seen = True raise if not data: From bffad383260eddcbd81326897a4ce835073f30f0 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 13:11:07 +0200 Subject: [PATCH 07/25] ci: run macOS pytest 20x to validate PTY fix Temporarily run only macOS runners with 20 repetitions via matrix index to confirm the PTY race condition fix is stable. Will be reverted after validation. Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/python-tests.yaml | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/.github/workflows/python-tests.yaml b/.github/workflows/python-tests.yaml index ee91e4d2f..7aa2f5434 100644 --- a/.github/workflows/python-tests.yaml +++ b/.github/workflows/python-tests.yaml @@ -42,7 +42,7 @@ jobs: # (all Python versions on both Linux and macOS). if [[ "${{ github.event_name }}" == "pull_request" ]]; then echo 'python-versions=["3.11", "3.12", "3.13"]' >> "$GITHUB_OUTPUT" - echo 'runners=["ubuntu-24.04", "macos-15"]' >> "$GITHUB_OUTPUT" + echo 'runners=["macos-15"]' >> "$GITHUB_OUTPUT" else echo 'python-versions=["3.11", "3.12", "3.13"]' >> "$GITHUB_OUTPUT" echo 'runners=["ubuntu-24.04", "macos-15"]' >> "$GITHUB_OUTPUT" @@ -53,14 +53,11 @@ jobs: if: needs.changes.outputs.should_run == 'true' || github.event_name == 'workflow_dispatch' runs-on: ${{ matrix.runs-on }} strategy: + fail-fast: false matrix: runs-on: ${{ fromJson(needs.changes.outputs.runners) }} - # Floor: oldest Python in supported platforms (RHEL 9 appstream) - # Ceiling: newest Python in latest Fedora - # Review on each RHEL/Fedora release - # PRs run only 3.12 on Linux; merge queue runs all versions - # on both Linux and macOS. python-version: ${{ fromJson(needs.changes.outputs.python-versions) }} + run-index: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20] steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 with: From 32c5aa8b9614992a46fffa98a3f56ab239d23ae9 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 13:38:21 +0200 Subject: [PATCH 08/25] fix(hooks): drain polls for full timeout instead of breaking on first empty On macOS, PTY internal buffer delivery can lag behind slave closure for very fast commands. The drain must keep polling (bounded by DRAIN_TIMEOUT_SECONDS) rather than breaking on the first empty select(), giving the kernel time to deliver remaining data. Co-Authored-By: Claude Opus 4.6 (1M context) --- python/packages/jumpstarter/jumpstarter/exporter/hooks.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index 08a16eb41..cc28371a7 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -437,7 +437,7 @@ async def read_pty_output() -> None: # noqa: C901 except (ValueError, OSError): break if not readable: - break + continue try: chunk = os.read(parent_fd, 4096) if not chunk: From 4b5896ac333f96418ef3e0e430a2d82325632b4a Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 24 Jun 2026 14:00:45 +0200 Subject: [PATCH 09/25] fix(hooks): extend reader grace period to DRAIN_TIMEOUT_SECONDS Replace the 0.2s grace period with DRAIN_TIMEOUT_SECONDS (2s) after subprocess exit. This gives the macOS PTY kernel buffer sufficient time to deliver data to the master fd before the reader_stop flag forces the reader to exit. The 0.2s window was the fundamental cause of the remaining failures: on loaded macOS CI runners, the PTY buffer delivery can lag behind slave closure by more than 200ms. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/jumpstarter/exporter/hooks.py | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index cc28371a7..173439903 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -498,22 +498,19 @@ async def wait_for_process() -> int: await anyio.sleep(0) with anyio.move_on_after(timeout) as cancel_scope: - # Run output reading and process waiting concurrently async with anyio.create_task_group() as tg: logger.debug("Task group created, starting tasks...") tg.start_soon(read_pty_output) logger.debug("Waiting for subprocess to complete...") returncode = await wait_for_process() logger.debug("Subprocess completed with code: %s", returncode) - # Give a brief moment for any final output to be read - await anyio.sleep(0.2) - # Signal the read task to stop via the dedicated stop flag. - # The read task checks this flag after each 0.1s timeout - # and also receives EOF when the subprocess exits. - # Note: pty_state.parent_fd_open stays True so the finally block - # properly closes parent_fd. + # After the subprocess exits, the PTY slave has no + # more writers. The reader will get EOF and exit. + # Set reader_stop after a grace period to handle + # grandchild processes that hold the PTY slave open. + await anyio.sleep(DRAIN_TIMEOUT_SECONDS) pty_state.reader_stop = True - logger.debug("Stop flag set, waiting for read task to exit") + logger.debug("Reader grace period expired, stop flag set") # Don't cancel - let the task exit naturally via EOF or flag check # Cancellation can cause unexpected side effects on gRPC connections From 0f0d4f69dd803c42ca890bae54bcdbfe1e0611a2 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Thu, 25 Jun 2026 16:13:50 +0200 Subject: [PATCH 10/25] ci: bump e2e timeout from 30m to 35m The PTY fix adds a 2s grace period after each hook subprocess exits, giving the macOS PTY kernel buffer time to deliver data. With ~24 hook e2e tests running both beforeLease and afterLease hooks, this adds ~2-3 minutes total. Bump the timeout to accommodate. Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/e2e.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml index 7fcbdec27..d8103271f 100644 --- a/.github/workflows/e2e.yaml +++ b/.github/workflows/e2e.yaml @@ -57,7 +57,7 @@ jobs: matrix: include: ${{ fromJson(needs.changes.outputs.e2e-matrix) }} runs-on: ${{ matrix.os }} - timeout-minutes: 30 + timeout-minutes: 35 steps: - name: Checkout repository uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 @@ -96,7 +96,7 @@ jobs: matrix: include: ${{ fromJson(needs.changes.outputs.e2e-matrix) }} runs-on: ${{ matrix.os }} - timeout-minutes: 30 + timeout-minutes: 35 steps: - name: Checkout repository uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 From 131f51eb6e2bf9871a756864b0f8acc892df08c7 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Thu, 25 Jun 2026 16:18:43 +0200 Subject: [PATCH 11/25] test(hooks): remove macos_pty_xfail markers Remove the xfail markers added by upstream for the macOS PTY race condition. The fix in this PR (process_group=0 + read-before-stop + drain with continue + extended grace period) addresses the root cause, making these markers unnecessary. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/exporter/hooks_test.py | 43 +++++++------------ 1 file changed, 16 insertions(+), 27 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 1a2ef25d7..d5d67bba4 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -1,5 +1,4 @@ import os -import sys from contextlib import nullcontext from unittest.mock import AsyncMock, MagicMock, patch @@ -19,16 +18,6 @@ pytestmark = pytest.mark.anyio -# Tests that spawn real subprocesses via PTY and assert on captured logger -# output are flaky on macOS due to a PTY kernel buffer timing race condition. -# See https://github.com/jumpstarter-dev/jumpstarter/issues/821 -# Targeted for proper fix in 0.10.0. -macos_pty_xfail = pytest.mark.xfail( - condition=sys.platform == "darwin", - reason="PTY output race condition on macOS (#821)", - strict=False, -) - class _PtyTracker: """Tracks PTY fd and EOF state for drain tests that need to intercept @@ -211,7 +200,7 @@ async def test_hook_timeout(self, lease_scope) -> None: assert "timed out after 1 seconds" in str(exc_info.value) assert exc_info.value.on_failure == "exit" - @macos_pty_xfail + async def test_hook_environment_variables(self, lease_scope) -> None: hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( @@ -311,7 +300,7 @@ def test_append_hook_motd_caps_size(self, tmp_path) -> None: HookExecutor._append_hook_motd(session, str(big)) assert len(session.motd) <= MAX_MOTD_BYTES - @macos_pty_xfail + async def test_real_time_output_logging(self, lease_scope) -> None: """Test that hook output is logged in real-time at INFO level.""" hook_config = HookConfigV1Alpha1( @@ -329,7 +318,7 @@ async def test_real_time_output_logging(self, lease_scope) -> None: assert any("Line 2" in call for call in info_calls) assert any("Line 3" in call for call in info_calls) - @macos_pty_xfail + async def test_post_lease_hook_execution_on_completion(self, lease_scope) -> None: """Test that post-lease hook executes when called directly.""" hook_config = HookConfigV1Alpha1( @@ -440,7 +429,7 @@ async def test_successful_hook_returns_none(self, lease_scope) -> None: result = await executor.execute_before_lease_hook(lease_scope) assert result is None - @macos_pty_xfail + async def test_exec_bash(self, lease_scope) -> None: """Test that exec=/bin/bash allows bash-specific syntax. @@ -462,7 +451,7 @@ async def test_exec_bash(self, lease_scope) -> None: info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("BASH_OK: world" in call for call in info_calls) - @macos_pty_xfail + async def test_exec_python3(self, lease_scope) -> None: """Test that exec=python3 runs inline Python. @@ -485,7 +474,7 @@ async def test_exec_python3(self, lease_scope) -> None: # Expected total: 0 + 1 + 4 + 9 == 14 assert any("PYTHON_OK: 14" in call for call in info_calls) - @macos_pty_xfail + async def test_script_file_sh(self, lease_scope, tmp_path) -> None: """Test that a .sh file auto-detects /bin/sh as interpreter.""" script_file = tmp_path / "hook_script.sh" @@ -508,7 +497,7 @@ async def test_script_file_sh(self, lease_scope, tmp_path) -> None: debug_calls = [str(call) for call in mock_logger.debug.call_args_list] assert any("Executing script file" in call for call in debug_calls) - @macos_pty_xfail + async def test_script_file_py_autodetects_python(self, lease_scope, tmp_path) -> None: """Test that a .py file auto-detects the exporter's Python as interpreter.""" import sys @@ -535,7 +524,7 @@ async def test_script_file_py_autodetects_python(self, lease_scope, tmp_path) -> # Verify it used the exporter's own Python interpreter assert any(sys.executable in call for call in debug_calls) - @macos_pty_xfail + async def test_script_file_py_exec_override(self, lease_scope, tmp_path) -> None: """Test that explicit exec overrides .py auto-detection.""" script_file = tmp_path / "hook_script.py" @@ -559,7 +548,7 @@ async def test_script_file_py_exec_override(self, lease_scope, tmp_path) -> None debug_calls = [str(call) for call in mock_logger.debug.call_args_list] assert not any("Auto-detected" in call for call in debug_calls) - @macos_pty_xfail + async def test_noninteractive_environment(self, lease_scope) -> None: """Test that hooks receive noninteractive environment variables. @@ -819,7 +808,7 @@ async def test_drain_handles_oserror_gracefully(self) -> None: assert output_lines == [] assert drained == 0 - @macos_pty_xfail + async def test_drain_captures_output_without_trailing_newline(self, lease_scope) -> None: """Verify output without a trailing newline is still captured.""" hook_config = HookConfigV1Alpha1( @@ -836,7 +825,7 @@ async def test_drain_captures_output_without_trailing_newline(self, lease_scope) info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("NO_NEWLINE_OUTPUT" in call for call in info_calls) - @macos_pty_xfail + async def test_drain_reads_data_remaining_in_pty_buffer(self, lease_scope) -> None: """Verify the drain loop inside read_pty_output reads data left in the PTY kernel buffer after the main read loop exits. @@ -901,7 +890,7 @@ def os_read_with_drain_data(fd, size): info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("DRAIN_CAPTURED" in call for call in info_calls) - @macos_pty_xfail + async def test_drain_select_oserror_exits_gracefully(self, lease_scope) -> None: """Verify the drain loop exits gracefully when select.select() raises OSError (e.g. fd closed during drain). @@ -938,7 +927,7 @@ def select_with_oserror(rlist, wlist, xlist, timeout=None): info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("SELECT_ERROR_TEST" in call for call in info_calls) - @macos_pty_xfail + async def test_drain_select_valueerror_exits_gracefully(self, lease_scope) -> None: """Verify the drain loop exits gracefully when select.select() raises ValueError (e.g. negative fd). @@ -973,7 +962,7 @@ def select_with_valueerror(rlist, wlist, xlist, timeout=None): info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("VALUEERROR_TEST" in call for call in info_calls) - @macos_pty_xfail + async def test_drain_exits_when_deadline_exceeded_before_select(self, lease_scope) -> None: """Verify the drain loop exits when the deadline is exceeded between the while condition and the remaining-time check (line: if remaining <= 0). @@ -1008,7 +997,7 @@ async def test_drain_exits_when_deadline_exceeded_before_select(self, lease_scop # exited early due to remaining <= 0 before select could run assert not any("SHOULD_NOT_APPEAR" in call for call in info_calls) - @macos_pty_xfail + async def test_drain_exception_is_suppressed(self, lease_scope) -> None: """Verify that an unexpected exception raised during the drain is caught by the except-Exception handler and does not propagate to the caller. @@ -1078,7 +1067,7 @@ async def test_exec_default_is_none(self) -> None: class TestHookExecutorPRRegressions: """Regression tests for issues reported during PR review of hooks feature.""" - @macos_pty_xfail + async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> None: """Issue A1: Hook infrastructure messages should be at DEBUG, not INFO. From 58dfcf70458ec02d58f26d7031ed05bdaca0f8a2 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Fri, 26 Jun 2026 11:10:20 +0200 Subject: [PATCH 12/25] fix(hooks): replace PTY with subprocess.PIPE for output capture The XNU kernel discards buffered PTY data when the slave fd closes without S_CTTYREF set (documented in Ruby #20682, pexpect #662, Apple Developer Forums #663632). This is POSIX-compliant behavior that no userspace I/O strategy can work around. Replace the PTY-based output capture with subprocess.PIPE, which uses kernel pipes instead of the PTY subsystem. Pipes guarantee data delivery before EOF on all platforms, eliminating the macOS race condition entirely. This removes ~400 lines of PTY management code (non-blocking polling, drain loops, retry logic, PtyState tracking) and replaces it with a simple blocking pipe read in a thread. Closes #821 Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/jumpstarter/exporter/hooks.py | 196 +------- .../jumpstarter/exporter/hooks_test.py | 472 +----------------- 2 files changed, 22 insertions(+), 646 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index 173439903..3c7159a5c 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -2,10 +2,8 @@ import logging import os -import select import stat import tempfile -import time from collections.abc import Awaitable from dataclasses import dataclass from typing import TYPE_CHECKING, Callable, Literal @@ -23,15 +21,8 @@ logger = logging.getLogger(__name__) -MAX_DRAIN_BYTES = 256 * 1024 -DRAIN_TIMEOUT_SECONDS = 2.0 - MAX_MOTD_BYTES = 64 * 1024 -# Module-level reference to time.monotonic so tests can patch it without -# affecting the asyncio event loop (which also uses time.monotonic). -_monotonic = time.monotonic - def _flush_lines(buffer: bytes, output_lines: list[str]) -> bytes: """Extract and log complete lines from a byte buffer. @@ -76,19 +67,6 @@ def should_end_lease(self) -> bool: return self.on_failure in ("endLease", "exit") -@dataclass -class PtyState: - """Mutable state for PTY file descriptors and reader coordination. - - Tracks which fds are still open (for cleanup) and provides a separate - stop flag to signal the reader task without affecting fd lifecycle. - """ - - parent_fd_open: bool = True - child_fd_open: bool = True - reader_stop: bool = False - - @dataclass(kw_only=True) class HookExecutor: """Executes lifecycle hooks with access to the j CLI.""" @@ -278,52 +256,32 @@ async def _execute_hook_process( # noqa: C901 logging_session: Session, hook_type: Literal["before_lease", "after_lease"], ) -> str | None: - """Execute the hook process with the given environment and logging session. - - Uses subprocess with a PTY to force line buffering in the subprocess, - ensuring logs stream in real-time rather than being block-buffered. + """Execute the hook process and capture its output via pipes. Returns: Warning message string if hook failed with on_failure='warn', None otherwise """ - import pty import subprocess command = hook_config.script timeout = hook_config.timeout on_failure = hook_config.on_failure - # Exception handling error_msg: str | None = None cause: Exception | None = None timed_out = False - # Route hook output logs to the client via the session's log stream logger.debug("Entering log source context for %s", log_source) with logging_session.context_log_source(__name__, log_source): - # Create a PTY pair - this forces line buffering in the subprocess logger.debug("Starting hook subprocess...") - logger.debug("Creating PTY pair...") - try: - parent_fd, child_fd = pty.openpty() - except Exception as e: - logger.error("Failed to create PTY: %s", e, exc_info=True) - raise - logger.debug("PTY created: parent_fd=%d, child_fd=%d", parent_fd, child_fd) - - pty_state = PtyState() process: subprocess.Popen | None = None try: - # Use subprocess.Popen with the PTY child as stdin/stdout/stderr - # This avoids the issues with os.fork() in async contexts - # Determine interpreter and invocation mode script_stripped = command.strip() is_file = "\n" not in script_stripped and os.path.isfile(script_stripped) interpreter = hook_config.exec_ if is_file and interpreter is None: - # Auto-detect interpreter from file extension import sys ext = os.path.splitext(script_stripped)[1].lower() @@ -347,111 +305,36 @@ async def _execute_hook_process( # noqa: C901 try: process = subprocess.Popen( cmd, - stdin=child_fd, - stdout=child_fd, - stderr=child_fd, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, env=hook_env, process_group=0, - close_fds=True, # Close inherited fds to prevent interference with gRPC connections + close_fds=True, ) except Exception as e: logger.error("Failed to spawn subprocess: %s", e, exc_info=True) raise logger.debug("Subprocess spawned with PID %d", process.pid) - # Close child fd in parent process - subprocess has it now - os.close(child_fd) - pty_state.child_fd_open = False - logger.debug("Closed child_fd in parent process") output_lines: list[str] = [] - # Set parent fd to non-blocking mode - import fcntl - - flags = fcntl.fcntl(parent_fd, fcntl.F_GETFL) - fcntl.fcntl(parent_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - logger.debug("Parent fd set to non-blocking") - - async def read_pty_output() -> None: # noqa: C901 - """Read from PTY parent fd line by line using non-blocking I/O.""" - logger.debug("read_pty_output task started") + async def read_output() -> None: + """Read subprocess output line by line via pipe.""" buffer = b"" - read_count = 0 - last_heartbeat = 0 - - start_time = _monotonic() try: while True: - try: - with anyio.move_on_after(0.1): - await anyio.wait_readable(parent_fd) - - read_count += 1 - elapsed = _monotonic() - start_time - if elapsed - last_heartbeat >= 2.0: - logger.debug( - "read_pty_output: heartbeat at %.1fs, iterations=%d", elapsed, read_count - ) - last_heartbeat = elapsed - - try: - chunk = os.read(parent_fd, 4096) - if not chunk: - logger.debug("read_pty_output: EOF received") - break - buffer += chunk - except BlockingIOError: - if pty_state.reader_stop: - logger.debug("read_pty_output: stop flag set and no data, exiting") - break - continue - except OSError as e: - logger.debug("read_pty_output: OSError on read: %s", e) - break - - buffer = _flush_lines(buffer, output_lines) - - except OSError as e: - logger.debug("read_pty_output: OSError in loop: %s", e) + chunk = await anyio.to_thread.run_sync( + lambda: process.stdout.read(4096), + abandon_on_cancel=True, + ) + if not chunk: break - finally: - # Drain any remaining data from the PTY buffer. - # On macOS, PTY output may still be in the kernel buffer - # after the subprocess exits and the stop flag is set. - # Use select() with a timeout to poll for readability - # instead of immediately breaking on BlockingIOError, - # giving the macOS PTY kernel buffer time to deliver - # remaining data. - # Bound the drain to prevent spinning indefinitely if a - # grandchild process holds the PTY slave fd open. - try: - drain_deadline = _monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and _monotonic() < drain_deadline: - remaining = drain_deadline - _monotonic() - if remaining <= 0: - break - timeout_s = min(remaining, 0.1) - try: - readable, _, _ = select.select([parent_fd], [], [], timeout_s) - except (ValueError, OSError): - break - if not readable: - continue - try: - chunk = os.read(parent_fd, 4096) - if not chunk: - break - buffer += chunk - drained += len(chunk) - except (BlockingIOError, OSError): - break - + buffer += chunk buffer = _flush_lines(buffer, output_lines) - except Exception: - logger.debug("read_pty_output: error during drain", exc_info=True) - - logger.debug("read_pty_output: exiting, processed %d iterations", read_count) + except OSError as e: + logger.debug("read_output: OSError: %s", e) + finally: if buffer: line_decoded = buffer.decode(errors="replace").rstrip() if line_decoded: @@ -459,75 +342,48 @@ async def read_pty_output() -> None: # noqa: C901 logger.info("%s", line_decoded) async def wait_for_process() -> int: - """Wait for the subprocess to complete. - - Ensures the subprocess is properly reaped even if cancelled, - preventing zombie processes. - """ + """Wait for the subprocess to complete.""" logger.debug("wait_for_process: waiting for PID %d", process.pid) try: result = await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) logger.debug("wait_for_process: PID %d exited with code %d", process.pid, result) return result finally: - # Ensure subprocess is reaped on cancellation to prevent zombies if process.poll() is None: logger.debug("wait_for_process: cleaning up still-running PID %d", process.pid) try: process.terminate() - # Give it a moment to terminate gracefully for _ in range(10): if process.poll() is not None: break await anyio.sleep(0.1) - # Force kill if still running if process.poll() is None: logger.debug("wait_for_process: force killing PID %d", process.pid) process.kill() - # Final reap with non-abandoning wait await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=False) except Exception as e: logger.debug("wait_for_process: error during cleanup: %s", e) - # Use move_on_after for timeout returncode: int | None = None - logger.debug("Starting PTY output reader and process waiter (timeout=%d)", timeout) - - # Yield to event loop to ensure other tasks can progress - # This helps prevent race conditions in task scheduling - await anyio.sleep(0) + logger.debug("Starting output reader and process waiter (timeout=%d)", timeout) with anyio.move_on_after(timeout) as cancel_scope: async with anyio.create_task_group() as tg: - logger.debug("Task group created, starting tasks...") - tg.start_soon(read_pty_output) - logger.debug("Waiting for subprocess to complete...") + tg.start_soon(read_output) returncode = await wait_for_process() logger.debug("Subprocess completed with code: %s", returncode) - # After the subprocess exits, the PTY slave has no - # more writers. The reader will get EOF and exit. - # Set reader_stop after a grace period to handle - # grandchild processes that hold the PTY slave open. - await anyio.sleep(DRAIN_TIMEOUT_SECONDS) - pty_state.reader_stop = True - logger.debug("Reader grace period expired, stop flag set") - # Don't cancel - let the task exit naturally via EOF or flag check - # Cancellation can cause unexpected side effects on gRPC connections if cancel_scope.cancelled_caught: timed_out = True error_msg = f"Hook timed out after {timeout} seconds" logger.error(error_msg) - # Terminate the process if process and process.poll() is None: process.terminate() - # Give it a moment to terminate gracefully try: with anyio.move_on_after(5): await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) except Exception: pass - # Force kill if still running if process.poll() is None: process.kill() try: @@ -546,23 +402,13 @@ async def wait_for_process() -> int: cause = e logger.error(error_msg, exc_info=True) finally: - # Clean up file descriptors - only close those still open to avoid - # closing an unrelated fd that reused the same number. - if pty_state.parent_fd_open: - try: - os.close(parent_fd) - except OSError: - pass - if pty_state.child_fd_open: + if process and process.stdout: try: - os.close(child_fd) + process.stdout.close() except OSError: pass - # Handle failure inside context_log_source so the WARNING log is - # routed to the client as a hook log (visible without --exporter-logs). if error_msg is not None: - # For timeout, create a TimeoutError as the cause if timed_out and cause is None: cause = TimeoutError(error_msg) return self._handle_hook_failure(error_msg, on_failure, hook_type, cause) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index d5d67bba4..7b0930103 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -1,4 +1,3 @@ -import os from contextlib import nullcontext from unittest.mock import AsyncMock, MagicMock, patch @@ -7,90 +6,15 @@ from jumpstarter.common import HOOK_WARNING_PREFIX, ExporterStatus from jumpstarter.config.exporter import HookConfigV1Alpha1, HookInstanceConfigV1Alpha1 from jumpstarter.exporter.hooks import ( - DRAIN_TIMEOUT_SECONDS, - MAX_DRAIN_BYTES, MAX_MOTD_BYTES, HookExecutionError, HookExecutor, _flush_lines, - _monotonic, ) pytestmark = pytest.mark.anyio -class _PtyTracker: - """Tracks PTY fd and EOF state for drain tests that need to intercept - os.read and pty.openpty calls. - - When ``return_drain_data`` is True (default), the first os.read after EOF - returns ``b"SHOULD_NOT_APPEAR\\n"``; otherwise it returns ``b""``. - """ - - def __init__(self, *, return_drain_data: bool = True) -> None: - import pty - - self.parent_fd: int | None = None - self.eof_seen: bool = False - self._drain_data_returned: bool = False - self._return_drain_data = return_drain_data - self._original_openpty = pty.openpty - self._original_os_read = os.read - - def tracking_openpty(self): - parent, child = self._original_openpty() - self.parent_fd = parent - return parent, child - - def os_read_with_drain_data(self, fd, size): - if fd != self.parent_fd: - return self._original_os_read(fd, size) - if not self.eof_seen: - try: - data = self._original_os_read(fd, size) - except BlockingIOError: - raise - except OSError: - self.eof_seen = True - raise - if not data: - self.eof_seen = True - return b"" - return data - if self._return_drain_data and not self._drain_data_returned: - self._drain_data_returned = True - return b"SHOULD_NOT_APPEAR\n" - return b"" - - -class _DrainDeadlineClock: - """A callable that replaces ``_monotonic`` to simulate the drain - deadline being exceeded between the ``while`` condition check and the - ``remaining`` calculation. - - Only patches the hooks module's ``_monotonic`` reference, leaving - ``time.monotonic`` (used by the asyncio event loop) unaffected. - """ - - def __init__(self, real_monotonic, state: _PtyTracker) -> None: - self._real = real_monotonic - self._state = state - self._call_count = 0 - self._deadline: float | None = None - - def __call__(self) -> float: - real_time = self._real() - if not self._state.eof_seen: - return real_time - self._call_count += 1 - if self._call_count == 1: - self._deadline = real_time + DRAIN_TIMEOUT_SECONDS - return real_time - if self._call_count == 2: - return self._deadline - 0.001 # type: ignore[operator] - return self._deadline + 1.0 # type: ignore[operator] - - class TestFlushLines: def test_extracts_complete_lines(self) -> None: output: list[str] = [] @@ -648,168 +572,8 @@ async def test_before_lease_hook_endlease_handles_release_error(self, lease_scop assert lease_scope.skip_after_lease_hook is True mock_request_lease_release.assert_called_once() - async def test_pty_output_drained_after_stop_flag_set(self) -> None: - """Test that PTY drain captures data remaining after the stop flag is set. - - Simulates the macOS scenario where PTY output is still in the kernel - buffer after the subprocess exits and reader_stop is set. Uses a pipe - to inject data, sets reader_stop=True to skip the main loop, and - verifies the finally-block drain captures all lines. - """ - import fcntl - import time - - read_fd, write_fd = os.pipe() - try: - flags = fcntl.fcntl(read_fd, fcntl.F_GETFL) - fcntl.fcntl(read_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - - os.write(write_fd, b"DRAIN_LINE_1\nDRAIN_LINE_2\nDRAIN_LINE_3\n") - os.close(write_fd) - write_fd = -1 - - output_lines: list[str] = [] - buffer = b"" - - drain_deadline = time.monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and time.monotonic() < drain_deadline: - try: - chunk = os.read(read_fd, 4096) - if not chunk: - break - buffer += chunk - drained += len(chunk) - except (BlockingIOError, OSError): - break - - buffer = _flush_lines(buffer, output_lines) - - assert "DRAIN_LINE_1" in output_lines - assert "DRAIN_LINE_2" in output_lines - assert "DRAIN_LINE_3" in output_lines - finally: - os.close(read_fd) - if write_fd != -1: - os.close(write_fd) - - async def test_drain_respects_byte_limit(self) -> None: - """Verify the drain loop stops after MAX_DRAIN_BYTES to prevent - indefinite blocking when a grandchild process holds the PTY open. - - Directly tests the drain logic using a pipe with data exceeding the - byte limit. Uses non-blocking writes to fill the pipe without blocking. - """ - import fcntl - import time - - read_fd, write_fd = os.pipe() - try: - flags = fcntl.fcntl(read_fd, fcntl.F_GETFL) - fcntl.fcntl(read_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - wflags = fcntl.fcntl(write_fd, fcntl.F_GETFL) - fcntl.fcntl(write_fd, fcntl.F_SETFL, wflags | os.O_NONBLOCK) - - total_written = 0 - chunk = b"X" * 4000 + b"\n" - try: - while True: - os.write(write_fd, chunk) - total_written += len(chunk) - except BlockingIOError: - pass - - assert total_written > 0 - - output_lines: list[str] = [] - buffer = b"" - drain_deadline = time.monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and time.monotonic() < drain_deadline: - try: - data = os.read(read_fd, 4096) - if not data: - break - buffer += data - drained += len(data) - except (BlockingIOError, OSError): - break - - buffer = _flush_lines(buffer, output_lines) - - assert drained <= MAX_DRAIN_BYTES - assert len(output_lines) > 0 - finally: - os.close(read_fd) - os.close(write_fd) - - async def test_drain_completes_immediately_on_empty_buffer(self) -> None: - """Verify drain exits quickly when the PTY buffer is empty (EOF).""" - import time - - read_fd, write_fd = os.pipe() - os.close(write_fd) - try: - import fcntl - - flags = fcntl.fcntl(read_fd, fcntl.F_GETFL) - fcntl.fcntl(read_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - - output_lines: list[str] = [] - buffer = b"" - start = time.monotonic() - - drain_deadline = time.monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and time.monotonic() < drain_deadline: - try: - chunk = os.read(read_fd, 4096) - if not chunk: - break - buffer += chunk - drained += len(chunk) - except (BlockingIOError, OSError): - break - - buffer = _flush_lines(buffer, output_lines) - elapsed = time.monotonic() - start - - assert output_lines == [] - assert drained == 0 - assert elapsed < 0.5 - finally: - os.close(read_fd) - - async def test_drain_handles_oserror_gracefully(self) -> None: - """Verify drain exits gracefully when os.read raises OSError (e.g. EIO).""" - import time - - read_fd, write_fd = os.pipe() - os.close(write_fd) - os.close(read_fd) - - output_lines: list[str] = [] - buffer = b"" - - drain_deadline = time.monotonic() + DRAIN_TIMEOUT_SECONDS - drained = 0 - while drained < MAX_DRAIN_BYTES and time.monotonic() < drain_deadline: - try: - chunk = os.read(read_fd, 4096) - if not chunk: - break - buffer += chunk - drained += len(chunk) - except (BlockingIOError, OSError): - break - - buffer = _flush_lines(buffer, output_lines) - assert output_lines == [] - assert drained == 0 - - - async def test_drain_captures_output_without_trailing_newline(self, lease_scope) -> None: + async def test_output_captured_without_trailing_newline(self, lease_scope) -> None: """Verify output without a trailing newline is still captured.""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( @@ -825,239 +589,6 @@ async def test_drain_captures_output_without_trailing_newline(self, lease_scope) info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("NO_NEWLINE_OUTPUT" in call for call in info_calls) - - async def test_drain_reads_data_remaining_in_pty_buffer(self, lease_scope) -> None: - """Verify the drain loop inside read_pty_output reads data left in the - PTY kernel buffer after the main read loop exits. - - Patches os.read so that, once the main loop has consumed the initial - subprocess output via EOF from the specific PTY fd, a subsequent read - returns additional data - simulating the macOS scenario where the - kernel buffers output that arrives after the reader stop flag is set. - """ - import pty - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo MAIN_OUTPUT", - timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - original_os_read = os.read - original_openpty = pty.openpty - pty_parent_fd = None - eof_seen_on_pty = False - - def tracking_openpty(): - nonlocal pty_parent_fd - parent, child = original_openpty() - pty_parent_fd = parent - return parent, child - - drain_data_returned = False - - def os_read_with_drain_data(fd, size): - nonlocal eof_seen_on_pty, drain_data_returned - if fd != pty_parent_fd: - return original_os_read(fd, size) - if not eof_seen_on_pty: - try: - data = original_os_read(fd, size) - except (BlockingIOError, OSError): - if not eof_seen_on_pty: - eof_seen_on_pty = True - raise - if not data: - eof_seen_on_pty = True - return b"" - return data - if not drain_data_returned: - drain_data_returned = True - return b"DRAIN_CAPTURED\n" - return b"" - - with ( - patch("pty.openpty", side_effect=tracking_openpty), - patch("os.read", side_effect=os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - assert pty_parent_fd is not None - assert eof_seen_on_pty - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("DRAIN_CAPTURED" in call for call in info_calls) - - - async def test_drain_select_oserror_exits_gracefully(self, lease_scope) -> None: - """Verify the drain loop exits gracefully when select.select() raises - OSError (e.g. fd closed during drain). - - Patches select.select inside the drain to raise OSError, simulating a - closed or invalid fd. The hook should still complete successfully. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker() - - def select_with_oserror(rlist, wlist, xlist, timeout=None): - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - raise OSError("simulated fd closed during drain") - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo SELECT_ERROR_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_with_oserror), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - assert state.eof_seen - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("SELECT_ERROR_TEST" in call for call in info_calls) - - - async def test_drain_select_valueerror_exits_gracefully(self, lease_scope) -> None: - """Verify the drain loop exits gracefully when select.select() raises - ValueError (e.g. negative fd). - - This covers the except (ValueError, OSError) handler in the drain loop. - """ - import select as select_mod - - original_select = select_mod.select - state = _PtyTracker(return_drain_data=False) - - def select_with_valueerror(rlist, wlist, xlist, timeout=None): - if state.eof_seen and rlist and rlist[0] == state.parent_fd: - raise ValueError("file descriptor cannot be a negative integer (-1)") - return original_select(rlist, wlist, xlist, timeout) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo VALUEERROR_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks.select.select", side_effect=select_with_valueerror), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("VALUEERROR_TEST" in call for call in info_calls) - - - async def test_drain_exits_when_deadline_exceeded_before_select(self, lease_scope) -> None: - """Verify the drain loop exits when the deadline is exceeded between the - while condition and the remaining-time check (line: if remaining <= 0). - - Patches ``jumpstarter.exporter.hooks._monotonic`` (not ``time.monotonic`` - globally) to simulate a jump past the deadline after the while condition - passes but before the remaining check. Using the module-level - ``_monotonic`` reference avoids breaking the asyncio event loop, which - also relies on ``time.monotonic``. - """ - state = _PtyTracker() - clock = _DrainDeadlineClock(_monotonic, state) - - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo DEADLINE_TEST", timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with ( - patch("pty.openpty", side_effect=state.tracking_openpty), - patch("os.read", side_effect=state.os_read_with_drain_data), - patch("jumpstarter.exporter.hooks._monotonic", side_effect=clock), - patch("jumpstarter.exporter.hooks.logger") as mock_logger, - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("DEADLINE_TEST" in call for call in info_calls) - # SHOULD_NOT_APPEAR should not be in output because the drain - # exited early due to remaining <= 0 before select could run - assert not any("SHOULD_NOT_APPEAR" in call for call in info_calls) - - - async def test_drain_exception_is_suppressed(self, lease_scope) -> None: - """Verify that an unexpected exception raised during the drain is caught - by the except-Exception handler and does not propagate to the caller. - - Patches _flush_lines so that the second call (inside the drain) raises - a RuntimeError. The hook should still complete successfully because the - drain's except-Exception block suppresses it. - """ - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo BEFORE_DRAIN_ERROR", - timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - original_flush = _flush_lines - call_count = 0 - - def flush_lines_with_drain_error(buffer, output_lines): - nonlocal call_count - call_count += 1 - result = original_flush(buffer, output_lines) - if call_count > 1: - raise RuntimeError("simulated drain error") - return result - - with ( - patch("jumpstarter.exporter.hooks._flush_lines", side_effect=flush_lines_with_drain_error), - patch("jumpstarter.exporter.hooks.logger"), - ): - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - - async def test_reader_exits_on_stop_flag_when_grandchild_holds_pty(self, lease_scope) -> None: - """Verify the reader exits via the stop flag when a grandchild - process holds the PTY slave open after the direct child exits. - - The backgrounded sleep inherits the PTY slave fd, preventing EOF - on the master. The reader gets BlockingIOError (no data from the - silent grandchild) and exits once reader_stop is set. - """ - hook_config = HookConfigV1Alpha1( - before_lease=HookInstanceConfigV1Alpha1( - script="echo STOP_FLAG_TEST; sleep 300 &", - timeout=10, - ), - ) - executor = HookExecutor(config=hook_config) - - with patch("jumpstarter.exporter.hooks.logger") as mock_logger: - result = await executor.execute_before_lease_hook(lease_scope) - assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("STOP_FLAG_TEST" in call for call in info_calls) - - async def test_drain_constants_are_reasonable(self) -> None: - assert MAX_DRAIN_BYTES == 256 * 1024 - assert DRAIN_TIMEOUT_SECONDS == 2.0 - async def test_exec_default_is_none(self) -> None: """Test that the default exec is None (auto-detect).""" hook = HookInstanceConfigV1Alpha1(script="echo hello") @@ -1091,7 +622,6 @@ async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> N # Infrastructure messages should be at DEBUG level infra_messages = [ "Starting hook subprocess", - "Creating PTY", "Spawning subprocess", "Subprocess spawned", "Hook executed successfully", From a6ae7f668254711339376fb1b93b403592e14f7e Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Fri, 26 Jun 2026 14:15:14 +0200 Subject: [PATCH 13/25] fix(hooks): use async non-blocking read on pipe fd for real-time output Replace thread-based process.stdout.read() with non-blocking os.read() + anyio.wait_readable() on the pipe fd. This delivers hook output line-by-line in real-time (like PTY did) instead of in a batch after subprocess exit. The thread-based reader had a scheduling race: the thread pool could delay the reader's first read until after the subprocess exited, causing all logger.info calls to happen in a burst right before the hook reported completion. The LogStream couldn't deliver the messages to the client in time. The async reader starts immediately (no thread pool dependency), reads each line as it's produced, and logs it via logger.info which the LogStream delivers to the client in real-time. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/jumpstarter/exporter/hooks.py | 27 ++++++++++++------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index 3c7159a5c..a7517f0d0 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -319,21 +319,30 @@ async def _execute_hook_process( # noqa: C901 output_lines: list[str] = [] + import fcntl + + pipe_fd = process.stdout.fileno() + flags = fcntl.fcntl(pipe_fd, fcntl.F_GETFL) + fcntl.fcntl(pipe_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) + async def read_output() -> None: - """Read subprocess output line by line via pipe.""" + """Read subprocess output via pipe using async non-blocking I/O.""" buffer = b"" try: while True: - chunk = await anyio.to_thread.run_sync( - lambda: process.stdout.read(4096), - abandon_on_cancel=True, - ) - if not chunk: + try: + with anyio.move_on_after(0.1): + await anyio.wait_readable(pipe_fd) + chunk = os.read(pipe_fd, 4096) + if not chunk: + break + buffer += chunk + except BlockingIOError: + continue + except OSError as e: + logger.debug("read_output: OSError: %s", e) break - buffer += chunk buffer = _flush_lines(buffer, output_lines) - except OSError as e: - logger.debug("read_output: OSError: %s", e) finally: if buffer: line_decoded = buffer.decode(errors="replace").rstrip() From 171f340c08c26aa930dc698d82c489ffe254308d Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Fri, 26 Jun 2026 15:44:57 +0200 Subject: [PATCH 14/25] fix(hooks): yield to event loop after reader to flush LogStream After the reader finishes and the task group exits, yield to the event loop before reporting the hook result. This lets the LogStream deliver pending messages to the client before the hook status changes to AVAILABLE, preventing the client from disconnecting before receiving the last output lines. Co-Authored-By: Claude Opus 4.6 (1M context) --- python/packages/jumpstarter/jumpstarter/exporter/hooks.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index a7517f0d0..ae4ef2456 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -381,6 +381,9 @@ async def wait_for_process() -> int: tg.start_soon(read_output) returncode = await wait_for_process() logger.debug("Subprocess completed with code: %s", returncode) + # Yield to let the LogStream deliver any pending + # messages before reporting the hook result. + await anyio.sleep(0) if cancel_scope.cancelled_caught: timed_out = True From 4d7f11d6003204345aa5228f0097b2079ada79a5 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Tue, 30 Jun 2026 10:46:37 +0200 Subject: [PATCH 15/25] revert(ci): remove temporary validation CI changes The PTY approach was replaced with subprocess.PIPE, so the 35m e2e timeout bump and the 20x macOS validation matrix are no longer needed. Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/e2e.yaml | 4 ++-- .github/workflows/python-tests.yaml | 11 +++++++---- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/.github/workflows/e2e.yaml b/.github/workflows/e2e.yaml index d8103271f..7fcbdec27 100644 --- a/.github/workflows/e2e.yaml +++ b/.github/workflows/e2e.yaml @@ -57,7 +57,7 @@ jobs: matrix: include: ${{ fromJson(needs.changes.outputs.e2e-matrix) }} runs-on: ${{ matrix.os }} - timeout-minutes: 35 + timeout-minutes: 30 steps: - name: Checkout repository uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 @@ -96,7 +96,7 @@ jobs: matrix: include: ${{ fromJson(needs.changes.outputs.e2e-matrix) }} runs-on: ${{ matrix.os }} - timeout-minutes: 35 + timeout-minutes: 30 steps: - name: Checkout repository uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 diff --git a/.github/workflows/python-tests.yaml b/.github/workflows/python-tests.yaml index 7aa2f5434..12233ae0f 100644 --- a/.github/workflows/python-tests.yaml +++ b/.github/workflows/python-tests.yaml @@ -41,8 +41,8 @@ jobs: # Merge queue and workflow_dispatch run the full matrix # (all Python versions on both Linux and macOS). if [[ "${{ github.event_name }}" == "pull_request" ]]; then - echo 'python-versions=["3.11", "3.12", "3.13"]' >> "$GITHUB_OUTPUT" - echo 'runners=["macos-15"]' >> "$GITHUB_OUTPUT" + echo 'python-versions=["3.12"]' >> "$GITHUB_OUTPUT" + echo 'runners=["ubuntu-24.04"]' >> "$GITHUB_OUTPUT" else echo 'python-versions=["3.11", "3.12", "3.13"]' >> "$GITHUB_OUTPUT" echo 'runners=["ubuntu-24.04", "macos-15"]' >> "$GITHUB_OUTPUT" @@ -53,11 +53,14 @@ jobs: if: needs.changes.outputs.should_run == 'true' || github.event_name == 'workflow_dispatch' runs-on: ${{ matrix.runs-on }} strategy: - fail-fast: false matrix: runs-on: ${{ fromJson(needs.changes.outputs.runners) }} + # Floor: oldest Python in supported platforms (RHEL 9 appstream) + # Ceiling: newest Python in latest Fedora + # Review on each RHEL/Fedora release + # PRs run only 3.12 on Linux; merge queue runs all versions + # on both Linux and macOS. python-version: ${{ fromJson(needs.changes.outputs.python-versions) }} - run-index: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20] steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 with: From fa80961e783a5394cf7659e2beaa941625069bc7 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Tue, 30 Jun 2026 17:10:46 +0200 Subject: [PATCH 16/25] test(hooks): add edge case tests for pipe-based output capture Cover pipe-specific edge cases that were untested after the PTY-to-pipe migration: stderr merge, large output spanning multiple reads, non-UTF8 decoding, rapid exit buffering, spawn failure cleanup, interleaved stdout/stderr, and grandchild process holding the pipe open past timeout. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/exporter/hooks_test.py | 142 ++++++++++++++++++ 1 file changed, 142 insertions(+) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 7b0930103..4fa569401 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -1211,3 +1211,145 @@ async def mock_report_status(status, msg): assert lease_scope.before_lease_hook.is_set(), ( "before_lease_hook event must be set to unblock downstream waiters" ) + + +class TestPipeOutputEdgeCases: + """Edge cases for pipe-based output capture (PR #837).""" + + async def test_stderr_captured_via_pipe_merge(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo STDOUT_LINE; echo STDERR_LINE >&2", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("STDOUT_LINE" in call for call in info_calls) + assert any("STDERR_LINE" in call for call in info_calls) + + async def test_large_output_spanning_multiple_reads(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script=( + "seq 1 200 | while read n; do " + "echo \"LINE_${n}_PADDING_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX\"; " + "done" + ), + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("LINE_1_" in call for call in info_calls) + assert any("LINE_100_" in call for call in info_calls) + assert any("LINE_200_" in call for call in info_calls) + + async def test_non_utf8_output_decoded_with_replacement(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="python3", + script=( + "import sys; " + "sys.stdout.buffer.write(b'VALID_PREFIX\\x80VALID_SUFFIX\\n')" + ), + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + matching = [ + call for call in info_calls + if "VALID_PREFIX" in call and "VALID_SUFFIX" in call + ] + assert len(matching) > 0 + + async def test_rapid_exit_with_buffered_output(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo FAST_1; echo FAST_2; echo FAST_3; echo FAST_4; echo FAST_5", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + for i in range(1, 6): + assert any(f"FAST_{i}" in call for call in info_calls), ( + f"FAST_{i} was not captured" + ) + + async def test_spawn_failure_cleans_up_without_crash(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="/nonexistent/interpreter", + script="echo should not run", + timeout=10, + on_failure="warn", + ), + ) + executor = HookExecutor(config=hook_config) + + result = await executor.execute_before_lease_hook(lease_scope) + assert result is not None + assert "error" in result.lower() + + async def test_interleaved_stdout_and_stderr_captured(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script=( + "echo OUT_1; echo ERR_1 >&2; " + "echo OUT_2; echo ERR_2 >&2; " + "echo OUT_3" + ), + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + for label in ("OUT_1", "OUT_2", "OUT_3", "ERR_1", "ERR_2"): + assert any(label in call for call in info_calls), ( + f"{label} was not captured" + ) + + async def test_timeout_with_grandchild_holding_pipe(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo GRANDCHILD_TEST; sleep 10 &", + timeout=2, + on_failure="warn", + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is not None + assert "timed out" in result.lower() + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("GRANDCHILD_TEST" in call for call in info_calls) From 83a62af27254efd461ee37ac146abb0dd17d609c Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 1 Jul 2026 10:44:29 +0200 Subject: [PATCH 17/25] fix(hooks): address review findings for pipe-based output capture - Close pipe read-end after child exits so grandchild processes holding the write end do not block the reader until timeout - Use os.killpg to terminate the entire process group on timeout, preventing orphaned child processes - Replace fcntl non-blocking setup with os.set_blocking - Remove stale PTY references from comments and docstrings - Trim verbose docstrings to single-line summaries per project rules - Add assertion for process.stdout before accessing fileno() - Improve finally-block cleanup with process group termination - Add TestReadOutputErrorPaths covering BlockingIOError, OSError, and partial buffer flush paths Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/jumpstarter/exporter/hooks.py | 233 +++++------------- .../jumpstarter/exporter/hooks_test.py | 215 ++++++++-------- 2 files changed, 182 insertions(+), 266 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index ae4ef2456..d4e70c4cc 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -2,6 +2,7 @@ import logging import os +import signal import stat import tempfile from collections.abc import Awaitable @@ -25,13 +26,7 @@ def _flush_lines(buffer: bytes, output_lines: list[str]) -> bytes: - """Extract and log complete lines from a byte buffer. - - Splits the buffer on newline boundaries, decodes each complete line, - and appends non-empty lines to output_lines while logging them. - - Returns the remaining bytes after the last newline (incomplete line). - """ + """Extract complete lines from buffer, log them, return the remainder.""" while b"\n" in buffer: line, buffer = buffer.split(b"\n", 1) line_decoded = line.decode(errors="replace").rstrip() @@ -43,13 +38,7 @@ def _flush_lines(buffer: bytes, output_lines: list[str]) -> bytes: @dataclass class HookExecutionError(Exception): - """Raised when a hook fails and on_failure is set to 'endLease' or 'exit'. - - Attributes: - message: Error message describing the failure - on_failure: The on_failure mode that triggered this error ('endLease' or 'exit') - hook_type: The type of hook that failed ('before_lease' or 'after_lease') - """ + """Raised when a hook fails and on_failure is set to 'endLease' or 'exit'.""" message: str on_failure: Literal["endLease", "exit"] @@ -74,22 +63,13 @@ class HookExecutor: config: HookConfigV1Alpha1 def _create_hook_env(self, lease_scope: "LeaseContext") -> dict[str, str]: - """Create standardized hook environment variables. - - Args: - lease_scope: LeaseScope containing lease metadata and socket paths + """Create environment variables for hook execution. - Returns: - Dictionary of environment variables for hook execution - - Note: - Uses the hook_socket_path (if available) instead of the main socket_path - to prevent SSL frame corruption when hook j commands access the session - concurrently with client LogStream connections. + Uses hook_socket_path (if available) instead of the main socket_path + to prevent SSL frame corruption when hook j commands access the session + concurrently with client LogStream connections. """ hook_env = os.environ.copy() - # Use dedicated hook socket to prevent SSL corruption - # Falls back to main socket if hook socket not available (backward compatibility) socket_path = lease_scope.hook_socket_path or lease_scope.socket_path if lease_scope.hook_socket_path: logger.info( @@ -110,9 +90,8 @@ def _create_hook_env(self, lease_scope: "LeaseContext") -> dict[str, str]: "LEASE_NAME": lease_scope.lease_name, "CLIENT_NAME": lease_scope.client_name, # Signal noninteractive mode to the child process. - # Even though hooks run in a PTY (for line-buffered output), they - # are not interactive sessions. These variables prevent programs - # from displaying prompts or interactive UI. + # Hooks are not interactive sessions. These variables prevent + # programs from displaying prompts or interactive UI. "TERM": "dumb", "DEBIAN_FRONTEND": "noninteractive", "GIT_TERMINAL_PROMPT": "0", @@ -128,16 +107,7 @@ async def _execute_hook( lease_scope: "LeaseContext", log_source: LogSource, ) -> str | None: - """Execute a single hook command. - - Args: - hook_config: Hook configuration including script, timeout, and on_failure - lease_scope: LeaseScope containing lease metadata and session - log_source: Log source for hook output - - Returns: - Warning message string if hook failed with on_failure='warn', None otherwise - """ + """Execute a single hook command.""" command = hook_config.script if not command or not command.strip(): logger.debug("Hook command is empty, skipping") @@ -145,14 +115,11 @@ async def _execute_hook( logger.debug("Executing hook: %s", command.strip().split("\n")[0][:100]) - # Determine hook type from log source hook_type = "before_lease" if log_source == LogSource.BEFORE_LEASE_HOOK else "after_lease" - # Validate session is available for logging if lease_scope.session is None: raise RuntimeError("Cannot execute hook: lease_scope.session is None") - # Use existing session from lease_scope hook_env = self._create_hook_env(lease_scope) logger.debug( "Hook environment: JUMPSTARTER_HOST=%s, LEASE_NAME=%s, CLIENT_NAME=%s", @@ -215,20 +182,7 @@ def _handle_hook_failure( hook_type: Literal["before_lease", "after_lease"], cause: Exception | None = None, ) -> str | None: - """Handle hook failure according to on_failure setting. - - Args: - error_msg: Error message describing the failure - on_failure: The on_failure mode ('warn', 'endLease', or 'exit') - hook_type: The type of hook that failed - cause: Optional exception that caused the failure - - Returns: - Warning message string if on_failure is 'warn', None otherwise - - Raises: - HookExecutionError: If on_failure is 'endLease' or 'exit' - """ + """Handle hook failure according to on_failure setting.""" if on_failure == "warn": logger.warning("%s (on_failure=warn, continuing)", error_msg) return error_msg @@ -241,7 +195,6 @@ def _handle_hook_failure( hook_type=hook_type, ) - # Properly handle exception chaining if cause is not None: raise error from cause else: @@ -256,11 +209,7 @@ async def _execute_hook_process( # noqa: C901 logging_session: Session, hook_type: Literal["before_lease", "after_lease"], ) -> str | None: - """Execute the hook process and capture its output via pipes. - - Returns: - Warning message string if hook failed with on_failure='warn', None otherwise - """ + """Execute the hook process and capture its output via pipes.""" import subprocess command = hook_config.script @@ -319,11 +268,9 @@ async def _execute_hook_process( # noqa: C901 output_lines: list[str] = [] - import fcntl - + assert process.stdout is not None pipe_fd = process.stdout.fileno() - flags = fcntl.fcntl(pipe_fd, fcntl.F_GETFL) - fcntl.fcntl(pipe_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) + os.set_blocking(pipe_fd, False) async def read_output() -> None: """Read subprocess output via pipe using async non-blocking I/O.""" @@ -351,27 +298,19 @@ async def read_output() -> None: logger.info("%s", line_decoded) async def wait_for_process() -> int: - """Wait for the subprocess to complete.""" + """Wait for the subprocess to complete. + + After the child exits, closes pipe_fd so that read_output + sees EOF even if a grandchild still holds the write end. + """ logger.debug("wait_for_process: waiting for PID %d", process.pid) + result = await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) + logger.debug("wait_for_process: PID %d exited with code %d", process.pid, result) try: - result = await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) - logger.debug("wait_for_process: PID %d exited with code %d", process.pid, result) - return result - finally: - if process.poll() is None: - logger.debug("wait_for_process: cleaning up still-running PID %d", process.pid) - try: - process.terminate() - for _ in range(10): - if process.poll() is not None: - break - await anyio.sleep(0.1) - if process.poll() is None: - logger.debug("wait_for_process: force killing PID %d", process.pid) - process.kill() - await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=False) - except Exception as e: - logger.debug("wait_for_process: error during cleanup: %s", e) + os.close(pipe_fd) + except OSError: + pass + return result returncode: int | None = None logger.debug("Starting output reader and process waiter (timeout=%d)", timeout) @@ -390,14 +329,20 @@ async def wait_for_process() -> int: error_msg = f"Hook timed out after {timeout} seconds" logger.error(error_msg) if process and process.poll() is None: - process.terminate() + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass try: with anyio.move_on_after(5): await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) except Exception: pass if process.poll() is None: - process.kill() + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass try: await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) except Exception: @@ -414,11 +359,28 @@ async def wait_for_process() -> int: cause = e logger.error(error_msg, exc_info=True) finally: - if process and process.stdout: - try: - process.stdout.close() - except OSError: - pass + if process: + if process.poll() is None: + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + process.wait(timeout=5) + except Exception: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + try: + process.wait(timeout=5) + except Exception: + pass + if process.stdout: + try: + process.stdout.close() + except OSError: + pass if error_msg is not None: if timed_out and cause is None: @@ -427,17 +389,7 @@ async def wait_for_process() -> int: return None async def execute_before_lease_hook(self, lease_scope: "LeaseContext") -> str | None: - """Execute the before-lease hook. - - Args: - lease_scope: LeaseScope with lease metadata and session - - Returns: - Warning message string if hook failed with on_failure='warn', None otherwise - - Raises: - HookExecutionError: If hook fails and on_failure is set to 'endLease' or 'exit' - """ + """Execute the before-lease hook.""" if not self.config.before_lease: logger.debug("No before-lease hook configured") return None @@ -450,17 +402,7 @@ async def execute_before_lease_hook(self, lease_scope: "LeaseContext") -> str | ) async def execute_after_lease_hook(self, lease_scope: "LeaseContext") -> str | None: - """Execute the after-lease hook. - - Args: - lease_scope: LeaseScope with lease metadata and session - - Returns: - Warning message string if hook failed with on_failure='warn', None otherwise - - Raises: - HookExecutionError: If hook fails and on_failure is set to 'endLease' or 'exit' - """ + """Execute the after-lease hook.""" if not self.config.after_lease: logger.debug("No after-lease hook configured") return None @@ -523,25 +465,14 @@ async def run_before_lease_hook( ) -> None: """Execute before-lease hook with full orchestration. - This method handles the complete lifecycle of running a before-lease hook: - - Waits for the lease scope to be ready (session/socket populated) - - Reports status changes via the provided callback - - Sets up the hook executor with the session for logging - - Executes the hook and handles errors - - Always signals the before_lease_hook event to unblock connections - - Args: - lease_scope: LeaseScope containing session, socket_path, and sync event - report_status: Async callback to report status changes to controller - shutdown: Callback to trigger exporter shutdown (accepts optional exit_code kwarg) - request_lease_release: Async callback to request lease release from controller + Always signals the before_lease_hook event to unblock connections, + even on failure. """ should_release = False try: if not await self._wait_for_lease_ready(lease_scope, report_status): return - # Check if hook is configured if not self.config.before_lease: logger.debug("No before-lease hook configured") await report_status(ExporterStatus.LEASE_READY, "Ready for commands") @@ -559,7 +490,6 @@ async def run_before_lease_hook( await report_status(ExporterStatus.BEFORE_LEASE_HOOK, "Running beforeLease hook") - # Execute hook with lease scope logger.info("Executing before-lease hook for lease %s", lease_scope.lease_name) warning = await self._execute_hook( self.config.before_lease, @@ -583,7 +513,6 @@ async def run_before_lease_hook( except HookExecutionError as e: if e.should_shutdown_exporter(): - # on_failure='exit' - defer shutdown until client handles the failure logger.error("beforeLease hook failed with on_failure='exit': %s", e) lease_scope.skip_after_lease_hook = True await report_status( @@ -594,10 +523,8 @@ async def run_before_lease_hook( ExporterStatus.OFFLINE, "Exporter shutting down due to beforeLease hook failure", ) - # Defer shutdown: sets _stop_requested=True, actual stop after lease cleanup shutdown(exit_code=1, wait_for_lease_exit=True, should_unregister=True) else: - # on_failure='endLease' - report failure, release in finally block logger.error("beforeLease hook failed with on_failure='endLease': %s", e) lease_scope.skip_after_lease_hook = True should_release = True @@ -612,13 +539,9 @@ async def run_before_lease_hook( ExporterStatus.BEFORE_LEASE_HOOK_FAILED, f"beforeLease hook failed: {e}", ) - # Unexpected errors don't trigger shutdown - just block the lease - finally: - # Always set the event to unblock connections lease_scope.before_lease_hook.set() - # Release lease for endLease failure mode. # Shielded from cancellation to ensure the release completes # even if the task group is being torn down. if should_release: @@ -633,32 +556,14 @@ async def run_after_lease_hook( shutdown: Callable[..., None], request_lease_release: Callable[[], Awaitable[None]] | None = None, ) -> None: - """Execute after-lease hook with full orchestration. - - This method handles the complete lifecycle of running an after-lease hook: - - Validates that the lease scope is ready - - Reports status changes via the provided callback - - Sets up the hook executor with the session for logging - - Executes the hook and handles errors - - Triggers shutdown on critical failures (HookExecutionError) - - Requests lease release from controller after hook completes - - Args: - lease_scope: LeaseScope containing session, socket_path, and client info - report_status: Async callback to report status changes to controller - shutdown: Callback to trigger exporter shutdown (accepts optional exit_code kwarg) - request_lease_release: Async callback to request lease release from controller - """ + """Execute after-lease hook with full orchestration.""" shutdown_called = False try: - # Verify lease scope is ready - for after-lease this should always be true - # since we've already processed the lease, but check defensively if not lease_scope.is_ready(): - logger.warning("LeaseScope not ready for after-lease hook, skipping") + logger.warning("LeaseContext not ready for after-lease hook, skipping") await report_status(ExporterStatus.AVAILABLE, "Available for new lease") return - # Check if hook is configured if not self.config.after_lease: logger.debug("No after-lease hook configured") await report_status(ExporterStatus.AVAILABLE, "Available for new lease") @@ -666,7 +571,6 @@ async def run_after_lease_hook( await report_status(ExporterStatus.AFTER_LEASE_HOOK, "Running afterLease hooks") - # Execute hook with lease scope logger.info("Executing after-lease hook for lease %s", lease_scope.lease_name) warning = await self._execute_hook( self.config.after_lease, @@ -683,7 +587,6 @@ async def run_after_lease_hook( except HookExecutionError as e: if e.should_shutdown_exporter(): - # on_failure='exit' - shut down the entire exporter logger.error("afterLease hook failed with on_failure='exit': %s", e) await report_status( ExporterStatus.AFTER_LEASE_HOOK_FAILED, @@ -693,16 +596,10 @@ async def run_after_lease_hook( ExporterStatus.OFFLINE, "Exporter shutting down due to afterLease hook failure", ) - # No delay needed - client is already polling and will see the failure logger.error("Shutting down exporter due to afterLease hook failure with on_failure='exit'") - # Exit code 1 tells the CLI not to restart the exporter shutdown(exit_code=1, should_unregister=True, wait_for_lease_exit=True) shutdown_called = True else: - # on_failure='endLease' - report failure to the client, then release the lease. - # AFTER_LEASE_HOOK_FAILED is a transient status: the client sees the failure, - # the lease is released in the finally block, and the exporter's main loop - # clears the lease context and accepts new leases. logger.error("afterLease hook failed with on_failure='endLease': %s", e) await report_status( ExporterStatus.AFTER_LEASE_HOOK_FAILED, @@ -710,9 +607,6 @@ async def run_after_lease_hook( ) except Exception as e: - # Unexpected errors: report failure but do not shut down. - # Same transient status - the lease is released and the exporter - # accepts new leases after the finally block completes. logger.error("afterLease hook failed with unexpected error: %s", e, exc_info=True) await report_status( ExporterStatus.AFTER_LEASE_HOOK_FAILED, @@ -720,11 +614,10 @@ async def run_after_lease_hook( ) finally: - # Always delay to give client time to poll the final status await anyio.sleep(1.0) - # Don't release lease when exporter is shutting down - unregistration handles cleanup. - # Releasing here would report AVAILABLE to the controller right before shutdown. + # Don't release lease when exporter is shutting down -- + # releasing here would report AVAILABLE right before shutdown. if request_lease_release and not shutdown_called: try: await request_lease_release() diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 4fa569401..32656aeb7 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -66,9 +66,7 @@ def lease_scope(): before_lease_hook=Event(), client_name="test-client", ) - # Add mock session to lease_scope mock_session = MagicMock() - # Return a no-op context manager for context_log_source mock_session.context_log_source.return_value = nullcontext() # Session.motd is str | None; model it so the beforeLease motd append works mock_session.motd = None @@ -87,7 +85,6 @@ async def test_empty_hook_execution(self, lease_scope) -> None: empty_config = HookConfigV1Alpha1() executor = HookExecutor(config=empty_config) - # Both hooks should return None for empty/None commands assert await executor.execute_before_lease_hook(lease_scope) is None assert await executor.execute_after_lease_hook(lease_scope) is None @@ -269,7 +266,6 @@ async def test_hook_timeout_with_warn(self, lease_scope) -> None: result = await executor.execute_before_lease_hook(lease_scope) assert result is not None assert "timed out" in result.lower() - # Verify WARNING log was created warning_calls = [str(call) for call in mock_logger.warning.call_args_list] assert any("on_failure=warn, continuing" in call for call in warning_calls) @@ -285,12 +281,8 @@ async def test_failed_hook_with_warn_returns_warning(self, lease_scope) -> None: assert "exit code 1" in result.lower() async def test_failed_hook_with_warn_logs_warning_inside_log_source_context(self) -> None: - """Test that the WARNING log for on_failure='warn' is emitted inside context_log_source. - - Issue #246: The WARNING log from _handle_hook_failure must be emitted while - the context_log_source context manager is active. This ensures the warning - is tagged with the hook source (BEFORE_LEASE_HOOK / AFTER_LEASE_HOOK) and - is visible to the client even without --exporter-logs. + """The WARNING log for on_failure='warn' must be emitted inside context_log_source + so the warning is tagged with the hook source and visible to the client. """ from contextlib import contextmanager @@ -303,7 +295,6 @@ async def test_failed_hook_with_warn_logs_warning_inside_log_source_context(self ) executor = HookExecutor(config=hook_config) - # Track whether context_log_source is active when warning is logged context_active = False warning_logged_in_context = False @@ -442,10 +433,8 @@ async def test_script_file_py_autodetects_python(self, lease_scope, tmp_path) -> assert result is None info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("PYFILE_OK" in call for call in info_calls) - # Verify it auto-detected Python (now logged at DEBUG level) debug_calls = [str(call) for call in mock_logger.debug.call_args_list] assert any("Auto-detected Python script" in call for call in debug_calls) - # Verify it used the exporter's own Python interpreter assert any(sys.executable in call for call in debug_calls) @@ -468,7 +457,6 @@ async def test_script_file_py_exec_override(self, lease_scope, tmp_path) -> None assert result is None info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("OVERRIDE_OK" in call for call in info_calls) - # Should NOT say "Auto-detected" since exec was explicitly set debug_calls = [str(call) for call in mock_logger.debug.call_args_list] assert not any("Auto-detected" in call for call in debug_calls) @@ -480,8 +468,8 @@ async def test_noninteractive_environment(self, lease_scope) -> None: and that PS1 is not set in the env dict passed to the subprocess. Note: PS1 is verified via _create_hook_env directly because shells - started in a PTY may re-set PS1 from init files despite it being - removed from the environment. + may re-set PS1 from init files despite it being removed from the + environment. """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( @@ -496,7 +484,7 @@ async def test_noninteractive_environment(self, lease_scope) -> None: executor = HookExecutor(config=hook_config) # Verify PS1 is removed from the env dict (not via subprocess, since - # shells in a PTY may re-set PS1 from profile/init files) + # shells may re-set PS1 from profile/init files) hook_env = executor._create_hook_env(lease_scope) assert "PS1" not in hook_env @@ -600,13 +588,8 @@ class TestHookExecutorPRRegressions: async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> None: - """Issue A1: Hook infrastructure messages should be at DEBUG, not INFO. - - Infrastructure messages like 'Starting hook subprocess', 'Creating PTY', - 'Spawning subprocess', 'Subprocess spawned', 'Subprocess completed', and - 'Hook executed successfully' must be logged at DEBUG level so they don't - appear in the client LogStream at the default INFO level. Only user output - from the hook script should be at INFO. + """Infrastructure messages must be at DEBUG, not INFO, so they + don't appear in the client LogStream. """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="echo 'user output'", timeout=10), @@ -619,7 +602,6 @@ async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> N debug_calls = [str(call) for call in mock_logger.debug.call_args_list] info_calls = [str(call) for call in mock_logger.info.call_args_list] - # Infrastructure messages should be at DEBUG level infra_messages = [ "Starting hook subprocess", "Spawning subprocess", @@ -634,15 +616,11 @@ async def test_infrastructure_messages_at_debug_not_info(self, lease_scope) -> N f"Infrastructure message '{msg}' should NOT be at INFO level" ) - # User output should be at INFO level assert any("user output" in call for call in info_calls) async def test_before_lease_hook_always_sets_event_on_failure(self, lease_scope) -> None: - """Issue C3: before_lease_hook event must be set even when hook fails. - - When the beforeLease hook fails with on_failure=endLease, the event must - still be set to unblock process_connections in handle_lease. Otherwise - the lease hangs indefinitely. + """before_lease_hook event must be set even when hook fails, to + unblock process_connections in handle_lease. """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="endLease"), @@ -660,15 +638,10 @@ async def test_before_lease_hook_always_sets_event_on_failure(self, lease_scope) mock_shutdown, ) - # Event must always be set to unblock connections assert lease_scope.before_lease_hook.is_set() async def test_before_lease_hook_always_sets_event_on_exit(self, lease_scope) -> None: - """Issue C3b: before_lease_hook event must be set when hook fails with exit. - - Same as C3 but for on_failure=exit. The event must be set, shutdown called, - and skip_after_lease_hook set to True. - """ + """before_lease_hook event must be set when hook fails with on_failure=exit.""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), ) @@ -688,12 +661,7 @@ async def test_before_lease_hook_always_sets_event_on_exit(self, lease_scope) -> mock_shutdown.assert_called_once() async def test_no_hooks_transitions_to_lease_ready(self, lease_scope) -> None: - """Issue D1: No hooks configured should transition directly to LEASE_READY. - - When no hooks are configured, run_before_lease_hook should report - LEASE_READY immediately, preventing the 'create lease, never use → stuck' - scenario. - """ + """No hooks configured should transition directly to LEASE_READY.""" empty_config = HookConfigV1Alpha1() executor = HookExecutor(config=empty_config) @@ -710,19 +678,13 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # Should have reported LEASE_READY assert any( status == ExporterStatus.LEASE_READY and msg == "Ready for commands" for status, msg in status_calls ), f"Expected LEASE_READY status, got: {status_calls}" async def test_skip_after_lease_prevents_after_hook_execution(self, lease_scope) -> None: - """Issue E1: beforeLease fail+exit should prevent afterLease hook execution. - - When beforeLease fails with on_failure=exit, skip_after_lease_hook is set - to True. The handle_lease finally block checks this flag and skips the - afterLease hook. This test verifies the orchestration sequence. - """ + """beforeLease fail+exit should prevent afterLease hook execution.""" # Config with both hooks hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), @@ -737,7 +699,6 @@ async def mock_report_status(status, msg): mock_shutdown = MagicMock() - # Run before hook (which fails and sets skip flag) await executor.run_before_lease_hook( lease_scope, mock_report_status, @@ -746,8 +707,6 @@ async def mock_report_status(status, msg): assert lease_scope.skip_after_lease_hook is True - # Now simulate what handle_lease does: check the flag before running after hook - # This mirrors the actual code: `if not lease_scope.skip_after_lease_hook:` if not lease_scope.skip_after_lease_hook: await executor.run_after_lease_hook( lease_scope, @@ -755,19 +714,13 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # AFTER_LEASE_HOOK status should never have been reported after_hook_statuses = [s for s, _ in status_calls if s == ExporterStatus.AFTER_LEASE_HOOK] assert len(after_hook_statuses) == 0, ( f"afterLease hook should have been skipped, but AFTER_LEASE_HOOK was reported: {status_calls}" ) async def test_before_hook_exit_reports_failed_not_available(self, lease_scope) -> None: - """Issue E2: beforeLease fail+exit should report FAILED, not AVAILABLE. - - When beforeLease hook fails with on_failure=exit, the last status must be - BEFORE_LEASE_HOOK_FAILED. It should NOT report AVAILABLE, which would - incorrectly tell the controller the exporter is ready for new leases. - """ + """beforeLease fail+exit should report FAILED, not AVAILABLE.""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), ) @@ -786,35 +739,25 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # Last status should be OFFLINE (reported before shutdown to prevent new leases) last_status, _ = status_calls[-1] assert last_status == ExporterStatus.OFFLINE, ( f"Expected last status to be OFFLINE, got {last_status}" ) - # BEFORE_LEASE_HOOK_FAILED should also be present (reported before OFFLINE) failed_statuses = [s for s, _ in status_calls if s == ExporterStatus.BEFORE_LEASE_HOOK_FAILED] assert len(failed_statuses) > 0, ( f"Expected BEFORE_LEASE_HOOK_FAILED status, got: {status_calls}" ) - # AVAILABLE should never have been reported available_statuses = [s for s, _ in status_calls if s == ExporterStatus.AVAILABLE] assert len(available_statuses) == 0, ( f"AVAILABLE should NOT be reported when beforeLease exits, got: {status_calls}" ) - # Shutdown should have been called with correct args mock_shutdown.assert_called_once_with(exit_code=1, wait_for_lease_exit=True, should_unregister=True) async def test_after_hook_exit_reports_failed_calls_shutdown(self, lease_scope) -> None: - """Issue E3: afterLease fail+exit should report FAILED and call shutdown. - - When afterLease hook fails with on_failure=exit: - - AFTER_LEASE_HOOK_FAILED status must be reported - - AVAILABLE must NOT be reported - - shutdown must be called (not request_lease_release) - """ + """afterLease fail+exit should report FAILED and call shutdown.""" hook_config = HookConfigV1Alpha1( after_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), ) @@ -835,27 +778,22 @@ async def mock_report_status(status, msg): mock_request_release, ) - # AFTER_LEASE_HOOK_FAILED should be in statuses failed_statuses = [s for s, _ in status_calls if s == ExporterStatus.AFTER_LEASE_HOOK_FAILED] assert len(failed_statuses) > 0, ( f"Expected AFTER_LEASE_HOOK_FAILED status, got: {status_calls}" ) - # AVAILABLE should NOT be in statuses available_statuses = [s for s, _ in status_calls if s == ExporterStatus.AVAILABLE] assert len(available_statuses) == 0, ( f"AVAILABLE should NOT be reported when afterLease exits, got: {status_calls}" ) - # Shutdown called (not request_lease_release) mock_shutdown.assert_called_once_with(exit_code=1, should_unregister=True, wait_for_lease_exit=True) mock_request_release.assert_not_called() async def test_before_hook_warn_includes_warning_prefix(self, lease_scope) -> None: - """Issue E5: beforeLease hook fail with warn should include HOOK_WARNING_PREFIX. - - The status message for LEASE_READY must start with '[HOOK_WARNING] ' so that - shell.py can detect it and display a user-visible warning. + """beforeLease hook fail with warn should include HOOK_WARNING_PREFIX + so shell.py can detect and display a user-visible warning. """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="warn"), @@ -875,7 +813,6 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # Find the LEASE_READY status call ready_calls = [(s, m) for s, m in status_calls if s == ExporterStatus.LEASE_READY] assert len(ready_calls) == 1, f"Expected exactly one LEASE_READY, got: {status_calls}" _, msg = ready_calls[0] @@ -884,11 +821,7 @@ async def mock_report_status(status, msg): ) async def test_before_hook_exit_reports_offline_before_shutdown(self, lease_scope) -> None: - """When beforeLease hook fails with on_failure=exit, the exporter must - report OFFLINE status to the controller before initiating shutdown. - This prevents the controller from assigning new leases to a dying - exporter during the shutdown window. - """ + """OFFLINE must be reported before shutdown to prevent new lease assignment.""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="exit"), ) @@ -988,11 +921,9 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # beforeLease with warn should still transition to LEASE_READY ready_calls = [s for s, _ in status_calls if s == ExporterStatus.LEASE_READY] assert len(ready_calls) == 1 - # Now run afterLease (simulating premature lease-end cleanup) await executor.run_after_lease_hook( lease_scope, mock_report_status, @@ -1000,18 +931,13 @@ async def mock_report_status(status, msg): mock_request_release, ) - # afterLease hook should run and transition to AVAILABLE available_calls = [s for s, _ in status_calls if s == ExporterStatus.AVAILABLE] assert len(available_calls) > 0, ( f"Expected AVAILABLE status after warn+afterLease, got: {status_calls}" ) async def test_after_hook_warn_includes_warning_prefix(self, lease_scope) -> None: - """Issue E5b: afterLease hook fail with warn should include HOOK_WARNING_PREFIX. - - The status message for AVAILABLE must start with '[HOOK_WARNING] ' so that - shell.py can detect it and display a user-visible warning after session ends. - """ + """afterLease hook fail with warn should include HOOK_WARNING_PREFIX.""" hook_config = HookConfigV1Alpha1( after_lease=HookInstanceConfigV1Alpha1(script="exit 1", timeout=10, on_failure="warn"), ) @@ -1030,7 +956,6 @@ async def mock_report_status(status, msg): mock_shutdown, ) - # Find the AVAILABLE status call available_calls = [(s, m) for s, m in status_calls if s == ExporterStatus.AVAILABLE] assert len(available_calls) == 1, f"Expected exactly one AVAILABLE, got: {status_calls}" _, msg = available_calls[0] @@ -1336,7 +1261,10 @@ async def test_interleaved_stdout_and_stderr_captured(self, lease_scope) -> None f"{label} was not captured" ) - async def test_timeout_with_grandchild_holding_pipe(self, lease_scope) -> None: + async def test_grandchild_holding_pipe_does_not_block(self, lease_scope) -> None: + """After child exits, pipe_fd is closed so a grandchild holding + the write end does not block the reader until the full timeout. + """ hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( script="echo GRANDCHILD_TEST; sleep 10 &", @@ -1349,7 +1277,102 @@ async def test_timeout_with_grandchild_holding_pipe(self, lease_scope) -> None: with patch("jumpstarter.exporter.hooks.logger") as mock_logger: result = await executor.execute_before_lease_hook(lease_scope) - assert result is not None - assert "timed out" in result.lower() + assert result is None info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("GRANDCHILD_TEST" in call for call in info_calls) + + +class TestReadOutputErrorPaths: + """Tests for BlockingIOError and OSError handling in read_output. + + These exercise the read_output error paths via real subprocesses + that produce controlled output patterns, and via _flush_lines + for the partial buffer flush path. + """ + + async def test_blocking_io_error_path_via_nonblocking_pipe(self, lease_scope) -> None: + """On a non-blocking pipe, reading before data arrives raises + BlockingIOError. read_output handles this by continuing the loop. + Verified via a script that delays output. + """ + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="sleep 0.1; echo DELAYED_OUTPUT", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("DELAYED_OUTPUT" in call for call in info_calls) + + async def test_os_error_on_closed_pipe_breaks_reader(self, lease_scope) -> None: + """When the pipe fd is closed (by wait_for_process after child exit), + os.read raises OSError (Bad file descriptor) and read_output breaks + out of the loop. + """ + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="echo BEFORE_CLOSE", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("BEFORE_CLOSE" in call for call in info_calls) + + def test_flush_lines_partial_buffer_preserved(self) -> None: + """Partial buffer (no trailing newline) is returned for later flush.""" + output: list[str] = [] + remainder = _flush_lines(b"complete\npartial_data", output) + assert output == ["complete"] + assert remainder == b"partial_data" + + async def test_partial_buffer_flushed_on_exit(self, lease_scope) -> None: + """When the subprocess exits with output lacking a trailing newline, + the finally block in read_output flushes the partial buffer. + """ + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="printf 'NO_TRAILING_NEWLINE'", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("NO_TRAILING_NEWLINE" in call for call in info_calls) + + async def test_mixed_complete_and_partial_lines(self, lease_scope) -> None: + """Complete lines are flushed immediately; the trailing partial + is flushed when the subprocess exits. + """ + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + script="printf 'LINE_A\\nLINE_B\\nPARTIAL_C'", + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("LINE_A" in call for call in info_calls) + assert any("LINE_B" in call for call in info_calls) + assert any("PARTIAL_C" in call for call in info_calls) From 7eae6f745e04ac693754c39be336df57b9440786 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 1 Jul 2026 14:11:59 +0200 Subject: [PATCH 18/25] fix(hooks): do not close pipe_fd eagerly after child exit Closing pipe_fd in wait_for_process races with read_output: if the kernel pipe buffer still holds data when the fd is closed, read_output gets EBADF and loses unread output. This caused test_large_output_spanning_multiple_reads to fail in CI. Revert to letting read_output see natural EOF when all processes (including grandchildren) close their end of the pipe. The timeout handles the grandchild case. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/jumpstarter/exporter/hooks.py | 10 +--------- .../jumpstarter/exporter/hooks_test.py | 15 +++++---------- 2 files changed, 6 insertions(+), 19 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py index d4e70c4cc..ee96e1f1c 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks.py @@ -298,18 +298,10 @@ async def read_output() -> None: logger.info("%s", line_decoded) async def wait_for_process() -> int: - """Wait for the subprocess to complete. - - After the child exits, closes pipe_fd so that read_output - sees EOF even if a grandchild still holds the write end. - """ + """Wait for the subprocess to complete.""" logger.debug("wait_for_process: waiting for PID %d", process.pid) result = await anyio.to_thread.run_sync(process.wait, abandon_on_cancel=True) logger.debug("wait_for_process: PID %d exited with code %d", process.pid, result) - try: - os.close(pipe_fd) - except OSError: - pass return result returncode: int | None = None diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 32656aeb7..5f9ce0de6 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -1261,10 +1261,7 @@ async def test_interleaved_stdout_and_stderr_captured(self, lease_scope) -> None f"{label} was not captured" ) - async def test_grandchild_holding_pipe_does_not_block(self, lease_scope) -> None: - """After child exits, pipe_fd is closed so a grandchild holding - the write end does not block the reader until the full timeout. - """ + async def test_timeout_with_grandchild_holding_pipe(self, lease_scope) -> None: hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( script="echo GRANDCHILD_TEST; sleep 10 &", @@ -1277,7 +1274,8 @@ async def test_grandchild_holding_pipe_does_not_block(self, lease_scope) -> None with patch("jumpstarter.exporter.hooks.logger") as mock_logger: result = await executor.execute_before_lease_hook(lease_scope) - assert result is None + assert result is not None + assert "timed out" in result.lower() info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("GRANDCHILD_TEST" in call for call in info_calls) @@ -1310,11 +1308,8 @@ async def test_blocking_io_error_path_via_nonblocking_pipe(self, lease_scope) -> info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("DELAYED_OUTPUT" in call for call in info_calls) - async def test_os_error_on_closed_pipe_breaks_reader(self, lease_scope) -> None: - """When the pipe fd is closed (by wait_for_process after child exit), - os.read raises OSError (Bad file descriptor) and read_output breaks - out of the loop. - """ + async def test_reader_exits_on_eof(self, lease_scope) -> None: + """read_output exits cleanly when os.read returns empty bytes (EOF).""" hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( script="echo BEFORE_CLOSE", From e495ad059e96734e0a1dc899383a74ef3a112735 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 1 Jul 2026 14:18:55 +0200 Subject: [PATCH 19/25] ci: add temporary stress test for hooks (60 parallel runs) Temporary workflow to validate hooks test stability across 60 parallel runs. Will be removed after validation. Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/stress-test-hooks.yaml | 30 ++++++++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 .github/workflows/stress-test-hooks.yaml diff --git a/.github/workflows/stress-test-hooks.yaml b/.github/workflows/stress-test-hooks.yaml new file mode 100644 index 000000000..07abd45ab --- /dev/null +++ b/.github/workflows/stress-test-hooks.yaml @@ -0,0 +1,30 @@ +name: Stress Test Hooks + +on: + workflow_dispatch: + +permissions: + contents: read + +jobs: + stress-test: + runs-on: ubuntu-24.04 + strategy: + fail-fast: false + matrix: + run: [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60] + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 + + - id: uv + run: echo "version=$(cat .uv-version)" >> "$GITHUB_OUTPUT" + - name: Install uv + uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + with: + version: ${{ steps.uv.outputs.version }} + python-version: "3.12" + + - name: Run hooks tests (iteration ${{ matrix.run }}) + working-directory: python + run: | + uv run --package jumpstarter pytest packages/jumpstarter/jumpstarter/exporter/hooks_test.py -v 2>&1 | tail -60 From 1be7dc4b2871ff2c9cf795b98082c86a9bee875b Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 1 Jul 2026 14:19:26 +0200 Subject: [PATCH 20/25] ci: trigger stress test via push event Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/stress-test-hooks.yaml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/stress-test-hooks.yaml b/.github/workflows/stress-test-hooks.yaml index 07abd45ab..878c05e17 100644 --- a/.github/workflows/stress-test-hooks.yaml +++ b/.github/workflows/stress-test-hooks.yaml @@ -1,7 +1,9 @@ name: Stress Test Hooks on: - workflow_dispatch: + push: + branches: + - fix/pty-read-before-stop permissions: contents: read From 611a5dff26b842e52bae9f2ec84a691a42a056ac Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 1 Jul 2026 14:22:16 +0200 Subject: [PATCH 21/25] ci: remove temporary stress test workflow All 60 parallel runs passed with zero failures. Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/stress-test-hooks.yaml | 32 ------------------------ 1 file changed, 32 deletions(-) delete mode 100644 .github/workflows/stress-test-hooks.yaml diff --git a/.github/workflows/stress-test-hooks.yaml b/.github/workflows/stress-test-hooks.yaml deleted file mode 100644 index 878c05e17..000000000 --- a/.github/workflows/stress-test-hooks.yaml +++ /dev/null @@ -1,32 +0,0 @@ -name: Stress Test Hooks - -on: - push: - branches: - - fix/pty-read-before-stop - -permissions: - contents: read - -jobs: - stress-test: - runs-on: ubuntu-24.04 - strategy: - fail-fast: false - matrix: - run: [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60] - steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 - - - id: uv - run: echo "version=$(cat .uv-version)" >> "$GITHUB_OUTPUT" - - name: Install uv - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 - with: - version: ${{ steps.uv.outputs.version }} - python-version: "3.12" - - - name: Run hooks tests (iteration ${{ matrix.run }}) - working-directory: python - run: | - uv run --package jumpstarter pytest packages/jumpstarter/jumpstarter/exporter/hooks_test.py -v 2>&1 | tail -60 From 422e4a1b72441576300f4c23ccc790344d992bc4 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Wed, 1 Jul 2026 14:29:26 +0200 Subject: [PATCH 22/25] ci: stress test hooks 60x on ubuntu, macos-13, macos-14, macos-15 240 total runs (60 per OS) to validate pipe-based output capture stability across Linux and all macOS variants (Intel + Apple Silicon). Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/stress-test-hooks.yaml | 33 ++++++++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 .github/workflows/stress-test-hooks.yaml diff --git a/.github/workflows/stress-test-hooks.yaml b/.github/workflows/stress-test-hooks.yaml new file mode 100644 index 000000000..54abb8c06 --- /dev/null +++ b/.github/workflows/stress-test-hooks.yaml @@ -0,0 +1,33 @@ +name: Stress Test Hooks + +on: + push: + branches: + - fix/pty-read-before-stop + +permissions: + contents: read + +jobs: + stress-test: + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-24.04, macos-13, macos-14, macos-15] + run: [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60] + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 + + - id: uv + run: echo "version=$(cat .uv-version)" >> "$GITHUB_OUTPUT" + - name: Install uv + uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 + with: + version: ${{ steps.uv.outputs.version }} + python-version: "3.12" + + - name: Run hooks tests (iteration ${{ matrix.run }} on ${{ matrix.os }}) + working-directory: python + run: | + uv run --package jumpstarter pytest packages/jumpstarter/jumpstarter/exporter/hooks_test.py -v 2>&1 | tail -60 From dd97f3ab1912a09aa29f1d0568f45acf8f3aea9d Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Thu, 2 Jul 2026 11:09:38 +0200 Subject: [PATCH 23/25] ci: remove temporary stress test workflow Results: 180/180 passed, 0 failures. - ubuntu-24.04: 60/60 passed - macos-14 (M1): 60/60 passed - macos-15 (M3): 60/60 passed - macos-13 (Intel): cancelled (no runners available after 14h) Co-Authored-By: Claude Opus 4.6 (1M context) --- .github/workflows/stress-test-hooks.yaml | 33 ------------------------ 1 file changed, 33 deletions(-) delete mode 100644 .github/workflows/stress-test-hooks.yaml diff --git a/.github/workflows/stress-test-hooks.yaml b/.github/workflows/stress-test-hooks.yaml deleted file mode 100644 index 54abb8c06..000000000 --- a/.github/workflows/stress-test-hooks.yaml +++ /dev/null @@ -1,33 +0,0 @@ -name: Stress Test Hooks - -on: - push: - branches: - - fix/pty-read-before-stop - -permissions: - contents: read - -jobs: - stress-test: - runs-on: ${{ matrix.os }} - strategy: - fail-fast: false - matrix: - os: [ubuntu-24.04, macos-13, macos-14, macos-15] - run: [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60] - steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 - - - id: uv - run: echo "version=$(cat .uv-version)" >> "$GITHUB_OUTPUT" - - name: Install uv - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0 - with: - version: ${{ steps.uv.outputs.version }} - python-version: "3.12" - - - name: Run hooks tests (iteration ${{ matrix.run }} on ${{ matrix.os }}) - working-directory: python - run: | - uv run --package jumpstarter pytest packages/jumpstarter/jumpstarter/exporter/hooks_test.py -v 2>&1 | tail -60 From 93af24723f5ac630943fd8fa0a386235afe8b903 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Mon, 6 Jul 2026 15:01:03 +0200 Subject: [PATCH 24/25] test(hooks): add coverage for defensive cleanup and error paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tests cover timeout cleanup with ProcessLookupError, finally-block process reaping, OSError handling in pipe reader, and the LeaseContext-not-ready guard — raising diff coverage from 52% to 92%. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/exporter/hooks_test.py | 142 ++++++++++++++++++ 1 file changed, 142 insertions(+) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 5f9ce0de6..3b3af3fbb 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -1,6 +1,9 @@ +import os +import subprocess from contextlib import nullcontext from unittest.mock import AsyncMock, MagicMock, patch +import anyio import pytest from jumpstarter.common import HOOK_WARNING_PREFIX, ExporterStatus @@ -936,6 +939,38 @@ async def mock_report_status(status, msg): f"Expected AVAILABLE status after warn+afterLease, got: {status_calls}" ) + async def test_after_lease_hook_skips_when_lease_context_not_ready(self) -> None: + from anyio import Event + + from jumpstarter.exporter.lease_context import LeaseContext + + hook_config = HookConfigV1Alpha1( + after_lease=HookInstanceConfigV1Alpha1(script="echo should-not-run", timeout=10), + ) + executor = HookExecutor(config=hook_config) + + lease_scope = LeaseContext( + lease_name="test-lease", + before_lease_hook=Event(), + client_name="test-client", + ) + + status_calls: list[tuple] = [] + + async def mock_report_status(status, msg): + status_calls.append((status, msg)) + + mock_shutdown = MagicMock() + + await executor.run_after_lease_hook( + lease_scope, + mock_report_status, + mock_shutdown, + ) + + assert any(s == ExporterStatus.AVAILABLE for s, _ in status_calls) + assert not any(s == ExporterStatus.AFTER_LEASE_HOOK for s, _ in status_calls) + async def test_after_hook_warn_includes_warning_prefix(self, lease_scope) -> None: """afterLease hook fail with warn should include HOOK_WARNING_PREFIX.""" hook_config = HookConfigV1Alpha1( @@ -1280,6 +1315,81 @@ async def test_timeout_with_grandchild_holding_pipe(self, lease_scope) -> None: assert any("GRANDCHILD_TEST" in call for call in info_calls) + async def test_timeout_cleanup_handles_process_lookup_errors(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="python3", + script=( + "import signal, time\n" + "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" + "time.sleep(300)\n" + ), + timeout=1, + on_failure="warn", + ), + ) + executor = HookExecutor(config=hook_config) + + original_killpg = os.killpg + + def killpg_raises_after_real_signal(pgid, sig): + try: + original_killpg(pgid, sig) + except ProcessLookupError: + pass + raise ProcessLookupError + + with patch("os.killpg", side_effect=killpg_raises_after_real_signal): + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is not None + assert "timed out" in result.lower() + + async def test_exception_during_hook_triggers_finally_cleanup(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="python3", + script=( + "import signal, time\n" + "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" + "time.sleep(300)\n" + ), + timeout=30, + on_failure="warn", + ), + ) + executor = HookExecutor(config=hook_config) + + original_wait = subprocess.Popen.wait + original_killpg = os.killpg + wait_calls = [0] + + def failing_then_real_wait(self_popen, timeout=None): + wait_calls[0] += 1 + if wait_calls[0] == 1: + import time as _time + + _time.sleep(0.3) + raise RuntimeError("simulated wait failure") + return original_wait(self_popen, timeout=timeout) + + def killpg_raises_after_real_signal(pgid, sig): + try: + original_killpg(pgid, sig) + except ProcessLookupError: + pass + raise ProcessLookupError + + with ( + patch.object(subprocess.Popen, "wait", failing_then_real_wait), + patch("os.killpg", side_effect=killpg_raises_after_real_signal), + ): + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is not None + assert "error" in result.lower() + + class TestReadOutputErrorPaths: """Tests for BlockingIOError and OSError handling in read_output. @@ -1351,6 +1461,38 @@ async def test_partial_buffer_flushed_on_exit(self, lease_scope) -> None: info_calls = [str(call) for call in mock_logger.info.call_args_list] assert any("NO_TRAILING_NEWLINE" in call for call in info_calls) + async def test_oserror_during_pipe_read_exits_gracefully(self, lease_scope) -> None: + hook_config = HookConfigV1Alpha1( + before_lease=HookInstanceConfigV1Alpha1( + exec_="python3", + script=( + "import sys, time\n" + "sys.stdout.write('VISIBLE\\n')\n" + "sys.stdout.flush()\n" + "time.sleep(0.5)\n" + ), + timeout=10, + ), + ) + executor = HookExecutor(config=hook_config) + + original_wait_readable = anyio.wait_readable + call_count = [0] + + async def wait_readable_then_oserror(fd): + call_count[0] += 1 + if call_count[0] >= 3: + raise OSError("simulated fd error") + return await original_wait_readable(fd) + + with patch("anyio.wait_readable", side_effect=wait_readable_then_oserror): + with patch("jumpstarter.exporter.hooks.logger") as mock_logger: + result = await executor.execute_before_lease_hook(lease_scope) + + assert result is None + info_calls = [str(call) for call in mock_logger.info.call_args_list] + assert any("VISIBLE" in call for call in info_calls) + async def test_mixed_complete_and_partial_lines(self, lease_scope) -> None: """Complete lines are flushed immediately; the trailing partial is flushed when the subprocess exits. From 1e351bf7356c2f24fd1c27822132ce75a0640a22 Mon Sep 17 00:00:00 2001 From: Paul Wallrabe Date: Mon, 6 Jul 2026 15:40:45 +0200 Subject: [PATCH 25/25] fix(test): make OSError read test deterministic across CI runners The previous test depended on subprocess output timing -- on slow CI runners the mock OSError fired before output was produced. Now raises OSError immediately and asserts on the debug log instead. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../jumpstarter/exporter/hooks_test.py | 25 +++++-------------- 1 file changed, 6 insertions(+), 19 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py index 3b3af3fbb..35662f7d5 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/hooks_test.py @@ -3,7 +3,6 @@ from contextlib import nullcontext from unittest.mock import AsyncMock, MagicMock, patch -import anyio import pytest from jumpstarter.common import HOOK_WARNING_PREFIX, ExporterStatus @@ -1464,34 +1463,22 @@ async def test_partial_buffer_flushed_on_exit(self, lease_scope) -> None: async def test_oserror_during_pipe_read_exits_gracefully(self, lease_scope) -> None: hook_config = HookConfigV1Alpha1( before_lease=HookInstanceConfigV1Alpha1( - exec_="python3", - script=( - "import sys, time\n" - "sys.stdout.write('VISIBLE\\n')\n" - "sys.stdout.flush()\n" - "time.sleep(0.5)\n" - ), + script="echo done", timeout=10, ), ) executor = HookExecutor(config=hook_config) - original_wait_readable = anyio.wait_readable - call_count = [0] + async def wait_readable_oserror(fd): + raise OSError("simulated fd error") - async def wait_readable_then_oserror(fd): - call_count[0] += 1 - if call_count[0] >= 3: - raise OSError("simulated fd error") - return await original_wait_readable(fd) - - with patch("anyio.wait_readable", side_effect=wait_readable_then_oserror): + with patch("anyio.wait_readable", side_effect=wait_readable_oserror): with patch("jumpstarter.exporter.hooks.logger") as mock_logger: result = await executor.execute_before_lease_hook(lease_scope) assert result is None - info_calls = [str(call) for call in mock_logger.info.call_args_list] - assert any("VISIBLE" in call for call in info_calls) + debug_calls = [str(call) for call in mock_logger.debug.call_args_list] + assert any("OSError" in call for call in debug_calls) async def test_mixed_complete_and_partial_lines(self, lease_scope) -> None: """Complete lines are flushed immediately; the trailing partial