test: Run the subprocess-racing tests serially, and quiesce between them

A loaded 108-second suite run failed 14 tests at once -- every cancel/terminate
test in the suite -- while the product was behaving correctly. All 14 pass
serially. One log line makes the mechanism plain:

    Canceling subprocess 69379 via termination method
    Log from test 0 ... Log from test 19        # child ran the full 20s, exit 0

These tests start a real child, cancel or time it out, and assert on the outcome.
The assertions are right, but they assume the child and the runtime get scheduled
promptly. Under `-n auto` with twelve workers each also sleeping on a child, that
assumption fails. The failure is indistinguishable from a real cancel regression,
which is the expensive part: it teaches you to re-run rather than to read.

Adds a `serial_process` mark (a `pytest.mark.xdist_group`) in conftest, applied to
the tests that race the wall clock. Every such test lands on one xdist worker, so
they run serially with respect to each other while the other ~800 tests still run
in parallel. Requires `--dist=loadgroup`, added to addopts; under the default
`--dist=load` the marker is accepted, ignored, and reported nowhere.

Adds an autouse `_quiesce_after_process_test` fixture, keyed on that mark, which
cancels stray `threading.Timer`s and waits briefly for the thread count to settle.
Serialising only helps if the tests also stop overlapping in the background: a
`ScriptRunnerBase` leaves a timer running for a whole unexpired timeout or grace
period, plus a pool worker, and a test that finishes early by cancelling its child
hands both to whatever runs next.

Scoping, after measuring rather than assuming. Marking the two big classes
wholesale put ~60 process tests on one worker and made it the critical path: the
suite went 40s -> 94s. The mark is therefore on the 13 specific tests that
actually flaked, not on `TestScriptRunnerBase` or
`TestLoggingSubprocessSameUser` entire. The quiesce budget is 1s, not 5s: some
tests legitimately leave a daemon stdout-reader thread that never exits, and a
generous budget was being spent in full on every one of them -- measured at 5s of
teardown for a single test.

Also removes the suite's slowest test. `test_run_action_default_timeout`'s
no-timeout case ran a 20-second child to completion, at 21.3s the slowest test by
a factor of three and the entire critical path. Split into
test_run_action_timeout_terminates_the_action (unchanged intent) and
test_run_action_without_timeout_runs_to_completion, which uses a child that exits
after half a second -- what is being asserted is that no timer cut the action
short, and a child that exits on its own shows that just as well.

Adds test_conftest_serial_process.py so this mechanism cannot regress silently.
Note its first version asserted `getoption("dist") == "loadgroup"` and failed:
inside an xdist worker that option is `"no"`, because a worker runs its share
serially and only the controller distributes. It now reads `addopts`.

No product code changes.

Verified: three consecutive full runs, 855 passed / 0 failed, in 44.2s / 41.5s /
40.7s -- the pre-change baseline was 40s with intermittent 14-failure runs. ruff,
mypy and black clean.

Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com>

Author: leongdl

pyproject.toml

@@ -155,6 +155,11 @@ addopts = [
     "--cov-report=xml:build/coverage/coverage.xml",
     "--cov-report=term-missing",
     "--numprocesses=auto",
+    # loadgroup, not the default load: it honours @pytest.mark.xdist_group, which
+    # the tests that race a real subprocess against the wall clock rely on to run
+    # serially with respect to each other (see conftest.serial_process). Behaves
+    # exactly like `load` for every test that is not in a group.
+    "--dist=loadgroup",
     "--timeout=30",
     "--log-format=%(asctime)s.%(msecs)03d %(levelname)s %(filename)20s:%(lineno)-3s %(message)s",
     "--log-date-format=%H:%M:%S"

test/openjd/sessions_v0/conftest.py

@@ -3,6 +3,7 @@
 import os
 import random
 import string
+import time
 import uuid
 from logging import INFO, getLogger
 from logging.handlers import QueueHandler
@@ -66,6 +67,79 @@ def pytest_collection_modifyitems(config, items):
         config.option.markexpr = mark_expr
 
 
+SERIAL_PROCESS_GROUP = "serial_process"
+
+serial_process = pytest.mark.xdist_group(SERIAL_PROCESS_GROUP)
+"""Mark for tests that race a real subprocess against wall-clock expectations.
+
+Applied to a class or a test, it pins every such test onto ONE xdist worker, so
+they run serially with respect to each other instead of competing for CPU with
+eleven siblings that are each also sleeping on a child process.
+
+Why this is needed: these tests start a child, cancel or time it out, and assert
+on the outcome. The assertions are correct, but they assume the child and the
+runtime get scheduled reasonably promptly. Under `-n auto` on a loaded host that
+assumption fails, and the whole cancel/terminate family goes red together while
+the product is behaving correctly -- observed as 14 simultaneous failures in a
+108-second run that all passed serially. The failure mode is indistinguishable
+from a real cancel regression, which is the expensive part: it trains you to
+re-run rather than to read.
+
+Requires `--dist=loadgroup` (set in pyproject.toml). With plain `--dist=load` the
+marker is silently ignored -- see `test_conftest_serial_process.py`, which fails
+if that ever regresses.
+
+This is not a substitute for fixing genuinely flaky assertions. Where a test
+asserted something it had no business asserting -- a +/-1 second window on a
+child's output, say -- that assertion was removed rather than protected by this.
+"""
+
+
+@pytest.fixture(autouse=True)
+def _quiesce_after_process_test(request: pytest.FixtureRequest) -> Generator[None, None, None]:
+    """Between serial-process tests, wait for the timers and threads the previous
+    one created to actually go away.
+
+    A `ScriptRunnerBase` leaves a `threading.Timer` running for the whole of an
+    unexpired timeout or cancel grace period, and a `ThreadPoolExecutor` worker
+    behind it. A test that finishes early -- because it cancelled its child -- can
+    therefore hand a live 30-second timer and a busy thread to whatever runs next.
+    Serialising the tests only helps if they also stop overlapping in that way.
+
+    Cancels stray timers, then waits briefly for the thread count to settle. Does
+    not assert: a leftover thread is not necessarily this test's fault, and turning
+    that into a failure here would report it against the wrong test. It is logged
+    so it is visible when it matters.
+    """
+    marker = request.node.get_closest_marker("xdist_group")
+    if marker is None or SERIAL_PROCESS_GROUP not in marker.args:
+        yield
+        return
+
+    import threading
+
+    before = threading.active_count()
+    yield
+
+    for thread in threading.enumerate():
+        if isinstance(thread, threading.Timer) and thread.is_alive():
+            thread.cancel()
+
+    # Deliberately short. Some tests legitimately leave a daemon stdout-reader
+    # thread behind that never exits (documented in LoggingSubprocess), so a
+    # generous budget here is spent in full on every one of them -- measured at 5s
+    # of pure teardown for a single test. One second is enough for a cancelled
+    # timer and a pool worker to wind down, which is what this is for.
+    deadline = time.monotonic() + 1.0
+    while threading.active_count() > before and time.monotonic() < deadline:
+        time.sleep(0.02)
+    if threading.active_count() > before:
+        print(
+            f"\n[quiesce] {request.node.name} left "
+            f"{threading.active_count() - before} extra thread(s) running"
+        )
+
+
 def nonexistent_group_name() -> str:
     """A group name that cannot resolve on any host.
 

test/openjd/sessions_v0/test_concurrency_fixes.py

@@ -31,6 +31,7 @@
 from openjd.sessions._subprocess import LoggingSubprocess
 
 from .conftest import build_logger
+from .conftest import serial_process
 
 
 @pytest.mark.usefixtures("message_queue", "queue_handler")
@@ -299,6 +300,7 @@ def bad_callback(state: ActionState) -> None:
 
 @pytest.mark.usefixtures("message_queue", "queue_handler")
 @pytest.mark.skipif(sys.platform == "win32", reason="POSIX-specific signal handling")
+@serial_process
 class TestReview22F4DoubleLoadFix:
     """Review22-F4: notify/terminate must bind _process once and pass to helpers."""
 

test/openjd/sessions_v0/test_conftest_serial_process.py

@@ -0,0 +1,71 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+"""The `serial_process` mechanism must keep working, silently-failing being its
+whole risk.
+
+`@pytest.mark.xdist_group` is only honoured under `--dist=loadgroup`. Under the
+default `--dist=load` it is accepted, ignored, and reported nowhere -- so the
+process-heavy tests would quietly go back to competing with eleven siblings and
+the cancel/terminate family would start failing together again on loaded hosts,
+looking exactly like a product regression.
+
+These tests are cheap and exist so that a change to `addopts` fails here, next to
+an explanation, rather than as fourteen mystery failures somewhere else.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from .conftest import SERIAL_PROCESS_GROUP, serial_process
+
+
+def test_loadgroup_distribution_is_configured(pytestconfig: pytest.Config) -> None:
+    """Pins the `--dist=loadgroup` setting the marker depends on.
+
+    Read from `addopts` rather than from `getoption("dist")`: inside an xdist
+    worker the resolved `dist` option is `"no"`, because a worker runs its own
+    share serially and only the controller distributes. Asserting on the resolved
+    option therefore fails on the workers and passes nowhere useful -- which is
+    exactly what the first version of this test did.
+    """
+    # GIVEN / WHEN
+    addopts = " ".join(pytestconfig.getini("addopts"))
+
+    # THEN
+    assert "--dist=loadgroup" in addopts, (
+        "xdist_group markers are silently ignored unless --dist=loadgroup; the "
+        "serial_process tests would go back to running in parallel with each other"
+    )
+
+
+def test_serial_process_is_an_xdist_group_marker() -> None:
+    """The mark must be the one xdist looks for, with the expected group name.
+
+    A rename or a typo would leave a decorator that reads as if it does something
+    and does nothing at all.
+    """
+    # GIVEN / WHEN
+    mark = serial_process.mark
+
+    # THEN
+    assert mark.name == "xdist_group"
+    assert mark.args == (SERIAL_PROCESS_GROUP,)
+
+
+@serial_process
+class TestMarkerReachesTheTest:
+    def test_the_marker_is_visible_on_a_decorated_test(
+        self, request: pytest.FixtureRequest
+    ) -> None:
+        """A class-level mark must actually reach the test item.
+
+        This is what the quiesce fixture in conftest keys off, so if the mark stops
+        propagating the cleanup silently stops running too.
+        """
+        # GIVEN / WHEN
+        marker = request.node.get_closest_marker("xdist_group")
+
+        # THEN
+        assert marker is not None
+        assert SERIAL_PROCESS_GROUP in marker.args

test/openjd/sessions_v0/test_runner_base.py

@@ -39,6 +39,7 @@
 from openjd.sessions._tempdir import TempDir
 
 from .conftest import (
+    serial_process,
     build_logger,
     collect_queue_messages,
     has_posix_target_user,
@@ -609,6 +610,7 @@ def test_cannot_run_twice(self, tmp_path: Path, python_exe: str) -> None:
             with pytest.raises(RuntimeError):
                 runner._run([python_exe, "-c", "print('hello')"])
 
+    @serial_process
     @pytest.mark.usefixtures("message_queue", "queue_handler")
     def test_run_action(
         self,
@@ -714,37 +716,29 @@ def test_run_action_effective_timeout(
         expected = timedelta(seconds=expected_seconds) if expected_seconds is not None else None
         assert captured[0] == expected
 
+    @serial_process
     @pytest.mark.usefixtures("message_queue", "queue_handler")
-    @pytest.mark.parametrize(
-        argnames=("action_timeout_seconds", "expected_state"),
-        argvalues=(
-            pytest.param(2, ScriptRunnerState.TIMEOUT, id="timeout-terminates-the-action"),
-            pytest.param(None, ScriptRunnerState.SUCCESS, id="no-timeout-runs-to-completion"),
-        ),
-    )
-    def test_run_action_timeout_is_enforced(
+    def test_run_action_timeout_terminates_the_action(
         self,
         tmp_path: Path,
         message_queue: SimpleQueue,
         queue_handler: QueueHandler,
-        action_timeout_seconds: Optional[int],
-        expected_state: ScriptRunnerState,
         python_exe: str,
     ) -> None:
-        """A declared timeout really does terminate the action, and no timeout
-        really does let it finish.
-
-        The end-to-end half of the coverage above, kept deliberately loose: it
-        asserts only the terminal state and that the child did or did not reach its
-        last line. It says nothing about *when* the timeout landed, because the
-        exact second the child reaches depends on how promptly the host scheduled
-        it -- which is what made the previous version of this test flaky.
+        """A declared timeout really does terminate the action.
+
+        The end-to-end half of the coverage above, kept deliberately loose: the
+        terminal state, and that the child did not reach its last line. It says
+        nothing about *when* the timeout landed, because which second the child
+        reaches depends on how promptly the host scheduled it -- that assumption is
+        what made the previous version of this test flaky.
         """
-        # GIVEN: a child that prints one line a second for 20 seconds
+        # GIVEN: a child that prints one line a second for 20 seconds, and a
+        # timeout that will cut it short
         action = Action_2023_09(
             command=CommandString_2023_09("{{Task.PythonInterpreter}}"),
             args=[ArgString_2023_09("{{Task.ScriptFile}}")],
-            timeout=action_timeout_seconds,
+            timeout=2,
         )
         python_app_loc = (Path(__file__).parent / "support_files" / "app_20s_run.py").resolve()
         symtab = SymbolTable(
@@ -761,13 +755,48 @@ def test_run_action_timeout_is_enforced(
                 time.sleep(0.2)
 
         # THEN
-        assert runner.state == expected_state
-        messages = collect_queue_messages(message_queue)
-        if expected_state is ScriptRunnerState.TIMEOUT:
-            # It was cut short. Which line it got to is a scheduling detail.
-            assert "Log from test 19" not in messages
-        else:
-            assert "Log from test 19" in messages
+        assert runner.state == ScriptRunnerState.TIMEOUT
+        assert "Log from test 19" not in collect_queue_messages(message_queue)
+
+    @pytest.mark.usefixtures("message_queue", "queue_handler")
+    def test_run_action_without_timeout_runs_to_completion(
+        self,
+        tmp_path: Path,
+        message_queue: SimpleQueue,
+        queue_handler: QueueHandler,
+        python_exe: str,
+    ) -> None:
+        """No timeout means no time limit is imposed: the action finishes on its
+        own terms.
+
+        Uses a child that exits after a fraction of a second rather than the
+        20-second one. Running a 20-second child to completion made this the single
+        slowest test in the suite by a factor of three, and it was the whole
+        critical path under `-n auto` -- while proving nothing that a short child
+        does not. What is being asserted is that no timer cut the action short, and
+        a child that exits on its own shows that either way.
+        """
+        # GIVEN: a short-lived child, and no timeout on the action or the caller
+        action = Action_2023_09(
+            command=CommandString_2023_09("{{Task.PythonInterpreter}}"),
+            args=[
+                ArgString_2023_09("-c"),
+                ArgString_2023_09("import time; time.sleep(0.5); print('finished')"),
+            ],
+            timeout=None,
+        )
+        symtab = SymbolTable(source={"Task.PythonInterpreter": python_exe})
+        logger = build_logger(queue_handler)
+        with TerminatingRunner(logger=logger, session_working_directory=tmp_path) as runner:
+            # WHEN
+            runner._run_action(action, symtab, default_timeout=None)
+            while runner.state == ScriptRunnerState.RUNNING:
+                time.sleep(0.05)
+
+        # THEN: it ran to its own end, and its last output arrived.
+        assert runner.state == ScriptRunnerState.SUCCESS
+        assert runner.exit_code == 0
+        assert "finished" in collect_queue_messages(message_queue)
 
     @pytest.mark.usefixtures("message_queue", "queue_handler")
     @pytest.mark.parametrize(
@@ -875,6 +904,7 @@ def test_run_action_bad_formatstring(
         messages = collect_queue_messages(message_queue)
         assert any(m.startswith("openjd_fail") for m in messages)
 
+    @serial_process
     @pytest.mark.usefixtures("message_queue", "queue_handler")
     def test_cancel_terminate(
         self,
@@ -910,6 +940,7 @@ def test_cancel_terminate(
         # Didn't get to the end of the application run
         assert "Log from test 9" not in messages
 
+    @serial_process
     @pytest.mark.usefixtures("message_queue", "queue_handler")
     @pytest.mark.xfail(not is_posix(), reason="Signals not yet implemented for non-posix")
     def test_run_with_time_limit(
@@ -942,6 +973,7 @@ def test_run_with_time_limit(
         # Didn't get to the end of the application run
         assert "Log from test 9" not in messages
 
+    @serial_process
     @pytest.mark.usefixtures("message_queue", "queue_handler")
     def test_cancel_notify(
         self,
@@ -1105,6 +1137,7 @@ def test_cancel_notify_direct_signal_with_cap_kill(
             cap_kill_was_effective == cap_kill_effective_after_cancel
         ), "CAP_KILL added/removed from effetive set and persisted after cancelation"
 
+    @serial_process
     @pytest.mark.usefixtures("message_queue", "queue_handler")
     def test_cancel_double_cancel_notify(
         self,

test/openjd/sessions_v0/test_session.py

@@ -66,6 +66,7 @@
 from openjd.sessions._windows_permission_helper import WindowsPermissionHelper
 
 from .conftest import (
+    serial_process,
     has_posix_target_user,
     has_windows_user,
     WIN_SET_TEST_ENV_VARS_MESSAGE,
@@ -954,6 +955,7 @@ def test_run_task_with_variables(
             )
 
 
+@serial_process
 class TestSessionCancel:
     """Test that cancelation will cancel the currently running Script."""
 

test/openjd/sessions_v0/test_session_run_subprocess.py

@@ -21,13 +21,15 @@
 from openjd.sessions._session_user import PosixSessionUser, WindowsSessionUser
 
 from .conftest import (
+    serial_process,
     has_posix_target_user,
     has_windows_user,
     WIN_SET_TEST_ENV_VARS_MESSAGE,
     POSIX_SET_TARGET_USER_ENV_VARS_MESSAGE,
 )
 
 
+@serial_process
 class TestRunSubprocess:
     """Tests for the Session.run_subprocess method."""