fix: Serialize the pending-cancel handoff against action launch

Addresses the review comment on the previous head: the _pending_cancel handoff
was unsynchronized, so the lost-cancel window it was meant to close was still
open. Two defects, both real:

- The reader in _run and the writer in _cancel_with_resolved_method both
  touched _pending_cancel outside self._lock, so a cancel could be recorded
  after _run had already consumed (and found nothing).
- The writer keyed on whether the LoggingSubprocess object existed, not on
  whether it had started. In the window where the object exists but the pool
  thread is still inside _start_subprocess, the canceller handed off to
  _cancel, which returns early because the process is not running -- while _run
  had already passed its consume point. Dropped by both sides.

Now the decide-and-record step happens under the lock and keys on
has_started, and _run consumes under the lock; _cancel is called outside it,
since it takes the lock itself.

The regression test reproduces the exact interleaving by blocking inside
_start_subprocess so the window is held open deterministically. Verified
against a mutant keyed on object existence: it ends SUCCESS after the full 30
second sleep instead of CANCELED, i.e. the cancel is silently dropped.

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

Author: leongdl

src/openjd/sessions/_runner_base.py

@@ -722,13 +722,16 @@ def _run(self, args: Sequence[str], time_limit: Optional[timedelta] = None) -> N
         # before we check self.state
         self._process.wait_until_started()
 
-        # A cancel that landed during setup, before there was a subprocess to
-        # signal, is applied now rather than dropped. See _pending_cancel.
-        pending = self._pending_cancel
-        if pending is not None:
+        # A cancel that landed during setup, before there was a running
+        # subprocess to signal, is applied now rather than dropped. Read under
+        # the lock so the handoff is serialized against the writer in
+        # _cancel_with_resolved_method; _cancel takes the lock itself, so it is
+        # called outside.
+        with self._lock:
+            pending = self._pending_cancel
             self._pending_cancel = None
-            if self._resolved_cancel_method is not None:
-                self._cancel(self._resolved_cancel_method, *pending)
+        if pending is not None and self._resolved_cancel_method is not None:
+            self._cancel(self._resolved_cancel_method, *pending)
 
         if self.state == ScriptRunnerState.RUNNING and self._callback is not None:
             # Let the caller know that the process is running.
@@ -913,22 +916,34 @@ def _cancel_with_resolved_method(
         the effective cancel method that :meth:`_run_action` resolved at
         launch time.
 
-        A cancel that arrives before the subprocess exists is remembered and
-        applied by :meth:`_run` as soon as it does (see :attr:`_pending_cancel`)
-        — `cancel()` is called from another thread, so "no subprocess yet" is a
+        A cancel that arrives before the subprocess is running is remembered and
+        applied by :meth:`_run` as soon as it starts (see :attr:`_pending_cancel`)
+        — `cancel()` is called from another thread, so "not running yet" is a
         race, not a no-op. Only when no action will ever be launched (setup
-        failed before resolution) is there genuinely nothing to cancel.
+        failed, or there was no action to run) is there genuinely nothing to
+        cancel.
         """
-        if self._process is None:
-            if self.state == ScriptRunnerState.FAILED:
-                # Setup already failed; no subprocess is coming.
+        with self._lock:
+            # Decide-and-record must be atomic with respect to _run creating and
+            # starting the subprocess, otherwise a cancel can land between the
+            # two and be dropped by both sides: this method would see a process
+            # and hand off to _cancel, which has nothing to signal yet, while
+            # _run has already passed the point where it consumes a pending
+            # cancel. Keyed on has_started rather than on the object existing,
+            # for the same reason.
+            if self._state_override is not None:
+                # Terminal before launch: setup failed, or there was no action.
                 return
-            self._pending_cancel = (time_limit, mark_action_failed)
+            process = self._process
+            if process is None or not process.has_started:
+                self._pending_cancel = (time_limit, mark_action_failed)
+                return
+            method = self._resolved_cancel_method
+        if method is None:  # pragma: no cover - defensive
             return
         # Note: If the given time_limit is less than that in the method, then the time_limit will be what's used.
-        if self._resolved_cancel_method is None:  # pragma: no cover - defensive
-            return
-        self._cancel(self._resolved_cancel_method, time_limit, mark_action_failed)
+        # Called outside the lock: _cancel takes it itself.
+        self._cancel(method, time_limit, mark_action_failed)
 
     def _cancel(
         self,

test/openjd/sessions_v0/test_runner_step_script.py

@@ -1,5 +1,6 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 
+import threading
 import time
 from datetime import timedelta
 from logging.handlers import QueueHandler
@@ -42,6 +43,7 @@
     TerminateCancelMethod,
 )
 from openjd.sessions._runner_step_script import StepScriptRunner
+from openjd.sessions._subprocess import LoggingSubprocess
 from openjd.sessions._tempdir import TempDir
 from openjd.sessions._os_checker import is_posix, is_windows
 
@@ -424,3 +426,76 @@ def test_cancel_with_no_subprocess_does_not_raise(
 
         # WHEN / THEN: the low-level cancel is a quiet no-op, not an AssertionError
         runner._cancel(TerminateCancelMethod())
+
+
+class TestCancelRacingLaunchIsSerialized:
+    """The pending-cancel handoff must be atomic with respect to launch.
+
+    `cancel()` runs on another thread. Without serialization there is an
+    interleaving where the canceller sees a subprocess object and hands off to
+    `_cancel`, which has nothing to signal because the process has not started
+    yet, while `_run` has already passed the point where it consumes a pending
+    cancel -- so the cancel is dropped by both sides and the action runs to
+    completion.
+    """
+
+    @pytest.mark.skipif(not is_posix(), reason="signal delivery is posix-only here")
+    @pytest.mark.timeout(120)
+    def test_cancel_during_launch_is_not_dropped(self, tmp_path: Path, python_exe: str) -> None:
+        # GIVEN: a long-running action, and a canceller that fires while the
+        # subprocess object exists but has NOT started -- held open by blocking
+        # inside _start_subprocess, which runs on the pool thread after _run has
+        # already assigned self._process.
+        script = StepScript_2023_09(
+            actions=StepActions_2023_09(
+                onRun=Action_2023_09(
+                    command=CommandString_2023_09(python_exe),
+                    args=[
+                        ArgString_2023_09("-c"),
+                        ArgString_2023_09("import time; time.sleep(30)"),
+                    ],
+                )
+            )
+        )
+        runner = StepScriptRunner(
+            logger=MagicMock(),
+            session_working_directory=tmp_path,
+            script=script,
+            symtab=SymbolTable(),
+            session_files_directory=tmp_path,
+        )
+        in_window = threading.Event()
+        canceller_done = threading.Event()
+        real_start = LoggingSubprocess._start_subprocess
+
+        def _start_after_cancel(subproc):  # type: ignore[no-untyped-def]
+            in_window.set()
+            # Hold the window: _has_started is not set until this returns.
+            canceller_done.wait(timeout=30)
+            return real_start(subproc)
+
+        def _cancel_in_window() -> None:
+            in_window.wait(timeout=30)
+            assert runner._process is not None
+            assert runner._process.has_started is False
+            runner.cancel()
+            canceller_done.set()
+
+        canceller = threading.Thread(target=_cancel_in_window, daemon=True)
+
+        # WHEN
+        with patch.object(LoggingSubprocess, "_start_subprocess", _start_after_cancel):
+            canceller.start()
+            runner.run()
+        canceller.join(timeout=30)
+
+        # THEN: the cancel was applied, so the 30 second sleep did not run to
+        # completion. Unsynchronized, it is dropped by both sides and the runner
+        # ends in SUCCESS after 30 seconds.
+        deadline = time.time() + 60
+        while runner.state in (ScriptRunnerState.RUNNING, ScriptRunnerState.CANCELING):
+            if time.time() > deadline:  # pragma: no cover - timing guard
+                pytest.fail(f"runner never settled; state={runner.state}")
+            time.sleep(0.05)
+        assert runner.state == ScriptRunnerState.CANCELED
+        assert runner._pending_cancel is None