diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/local/process.py b/packages/nemo_platform_ext/src/nemo_platform_ext/local/process.py index 7ee0cd4a9b..e211ea340a 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/local/process.py +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/local/process.py @@ -60,6 +60,10 @@ _SIGTERM_POLL_INTERVAL = 0.25 _SIGKILL_WAIT_TIMEOUT = 5.0 _DEFAULT_STOP_TIMEOUT = 30.0 +# How long ``stop_instance`` waits for the flock to be released after the last process it +# signaled has exited. The kernel drops the lock as part of closing the fd during process +# teardown, so this only has to cover the tail of that teardown, not a whole shutdown. +_LOCK_RELEASE_TIMEOUT = 5.0 def _pause(seconds: float) -> None: @@ -569,9 +573,14 @@ def _snapshot_children(pid: int) -> list[psutil.Process]: def _sweep_orphans(children: list[psutil.Process], timeout: float = 5.0) -> list[int]: """Terminate any still-alive processes from a prior snapshot. - Sends SIGTERM, waits up to *timeout*, then SIGKILL survivors. + Sends SIGTERM, waits up to *timeout*, then SIGKILL survivors and waits again. Returns PIDs that were signaled. Handles ``NoSuchProcess`` gracefully since children may have already exited during graceful shutdown. + + The second wait matters: these children inherited the instance flock fd from the + parent, and the kernel only releases the lock once the last holder's fd is closed. + Returning while a SIGKILLed child is still being torn down leaves the lock held, so + an ``is_instance_alive`` probe immediately after a "successful" stop reports True. """ alive_children = [c for c in children if c.is_running()] if not alive_children: @@ -586,12 +595,19 @@ def _sweep_orphans(children: list[psutil.Process], timeout: float = 5.0) -> list pass # Already exited or not owned by us — skip. _, still_alive = psutil.wait_procs(alive_children, timeout=timeout) + kill_sent: list[psutil.Process] = [] for child in still_alive: try: child.kill() + kill_sent.append(child) except (psutil.NoSuchProcess, psutil.AccessDenied): pass # Raced with exit or not owned — nothing to do. + if kill_sent: + _, unreaped = psutil.wait_procs(kill_sent, timeout=timeout) + if unreaped: + logger.warning("Orphaned child %s still alive after SIGKILL", [c.pid for c in unreaped]) + if killed: logger.info( "Swept %d orphaned child %s: %s", len(killed), "process" if len(killed) == 1 else "processes", killed @@ -673,10 +689,32 @@ def stop_instance( swept = _sweep_orphans(children) if children else [] + # A process exiting and its flock being released are not the same instant: the kernel + # drops the lock while closing fds during teardown. Callers treat a successful stop as + # "the scope is free now" (and immediately probe with ``is_instance_alive``), so hold the + # post-condition here rather than making every caller poll for it. + if not _wait_for_lock_release(scope, base_dir=base_dir, timeout=_LOCK_RELEASE_TIMEOUT): + logger.warning( + "Instance '%s' flock still held %.0fs after pid %d exited; a child may have outlived the sweep", + scope, + _LOCK_RELEASE_TIMEOUT, + pid, + ) + remove_descriptor(scope, base_dir=base_dir) return StopResult(stopped_pids=[pid], swept_children=swept) +def _wait_for_lock_release(scope: str, *, base_dir: Path | None, timeout: float) -> bool: + """Poll until the instance flock is free, or *timeout* elapses.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if not is_instance_alive(scope, base_dir=base_dir): + return True + _pause(_SIGTERM_POLL_INTERVAL) + return not is_instance_alive(scope, base_dir=base_dir) + + def _wait_for_pid_exit(pid: int, *, timeout: float) -> bool: deadline = time.monotonic() + timeout while time.monotonic() < deadline: diff --git a/packages/nemo_platform_ext/tests/cli/commands/test_services_process.py b/packages/nemo_platform_ext/tests/cli/commands/test_services_process.py index 26478c20d5..f5be14ec80 100644 --- a/packages/nemo_platform_ext/tests/cli/commands/test_services_process.py +++ b/packages/nemo_platform_ext/tests/cli/commands/test_services_process.py @@ -5,6 +5,7 @@ from __future__ import annotations +import logging import os import signal import subprocess @@ -845,6 +846,49 @@ def test_escalates_to_sigkill(self) -> None: proc.kill() proc.wait(timeout=5) + def test_warns_when_child_survives_sigkill(self, caplog: pytest.LogCaptureFixture) -> None: + """A child that outlives SIGKILL is reported rather than silently dropped. + + Nothing survives SIGKILL on demand — the real trigger is a process wedged + in uninterruptible sleep, which a test cannot arrange — so `wait_procs` is + stubbed to keep reporting the child as alive. The terminate/kill calls are + still real; only the observation is simulated. + """ + proc = subprocess.Popen( + [ + sys.executable, + "-c", + "import signal, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(60)", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + time.sleep(0.3) + ps_child = psutil.Process(proc.pid) + # The first wait drives the SIGTERM -> SIGKILL escalation; the second + # is the one whose non-empty result trips the warning. + with ( + patch.object( + process_module.psutil, + "wait_procs", + side_effect=[([], [ps_child]), ([], [ps_child])], + ) as wait_procs, + caplog.at_level(logging.WARNING, logger=process_module.logger.name), + ): + killed = _sweep_orphans([ps_child], timeout=0.1) + + # Two waits, not one. The second is the post-SIGKILL wait this PR adds, + # so a single-wait implementation must not be able to satisfy this test. + assert wait_procs.call_count == 2 + assert proc.pid in killed + assert f"Orphaned child [{proc.pid}] still alive after SIGKILL" in caplog.text + proc.wait(timeout=5) + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=5) + # --------------------------------------------------------------------------- # stop_instance — child sweep integration diff --git a/packages/nemo_platform_ext/tests/local/test_daemon_lifecycle.py b/packages/nemo_platform_ext/tests/local/test_daemon_lifecycle.py index e1a3c5c7e1..fae6aed834 100644 --- a/packages/nemo_platform_ext/tests/local/test_daemon_lifecycle.py +++ b/packages/nemo_platform_ext/tests/local/test_daemon_lifecycle.py @@ -376,6 +376,149 @@ def test_stop_instance_sweeps_orphaned_children(tmp_path: Path) -> None: pass +def test_wait_for_lock_release_blocks_until_holder_exits(tmp_path: Path) -> None: + """``_wait_for_lock_release`` should return only once the flock is actually free.""" + scope = "unit-lock-release" + state_dir = tmp_path / "state" + lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME + lock_path.parent.mkdir(parents=True) + lock_path.touch() + + # A holder that keeps the flock for ~1s, then exits and releases it. + holder = subprocess.Popen( + [ + sys.executable, + "-c", + "import fcntl, sys, time; " + "fd = open(sys.argv[1], 'r+'); " + "fcntl.flock(fd, fcntl.LOCK_EX); " + "print('held', flush=True); " + "time.sleep(1.0)", + str(lock_path), + ], + stdout=subprocess.PIPE, + text=True, + ) + try: + assert holder.stdout is not None + assert holder.stdout.readline().strip() == "held" + assert process.is_instance_alive(scope, base_dir=state_dir) + + started = time.monotonic() + assert process._wait_for_lock_release(scope, base_dir=state_dir, timeout=10.0) + assert not process.is_instance_alive(scope, base_dir=state_dir) + # It waited rather than returning eagerly on a still-held lock. + assert time.monotonic() - started >= 0.5 + finally: + holder.kill() + holder.wait(timeout=5) + + +def test_wait_for_lock_release_times_out_while_held(tmp_path: Path) -> None: + """A lock held for the whole window should report failure, not success.""" + scope = "unit-lock-held" + state_dir = tmp_path / "state" + lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME + lock_path.parent.mkdir(parents=True) + lock_path.touch() + + holder = subprocess.Popen( + [ + sys.executable, + "-c", + "import fcntl, sys, time; " + "fd = open(sys.argv[1], 'r+'); " + "fcntl.flock(fd, fcntl.LOCK_EX); " + "print('held', flush=True); " + "time.sleep(300)", + str(lock_path), + ], + stdout=subprocess.PIPE, + text=True, + ) + try: + assert holder.stdout is not None + assert holder.stdout.readline().strip() == "held" + assert not process._wait_for_lock_release(scope, base_dir=state_dir, timeout=0.5) + finally: + holder.kill() + holder.wait(timeout=5) + + +@pytest.mark.integration +def test_stop_instance_releases_lock_held_by_surviving_child(tmp_path: Path) -> None: + """Regression: ``stop_instance`` must not report success while the scope's + flock is still held. + + A process exiting and its flock being released are not the same instant — the + kernel drops the lock while closing fds during teardown, and any process that + inherited the fd keeps it held until *it* is gone too. ``stop_instance`` used + to return as soon as the descriptor PID was dead, so an immediate + ``is_instance_alive`` probe raced that teardown and intermittently saw True. + + The holder here is deliberately outside the parent's process tree: that makes + the window wide and fixed instead of scheduler-dependent, and it models the + case the sweep can't reach (a lock-inheriting process spawned after the child + snapshot was taken). + """ + scope = "integ-lock-release" + state_dir = tmp_path / "state" + lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME + lock_path.parent.mkdir(parents=True) + lock_path.touch() + + hold_seconds = 2.0 + holder = subprocess.Popen( + [ + sys.executable, + "-c", + "import fcntl, sys, time; " + "fd = open(sys.argv[1], 'r+'); " + "fcntl.flock(fd, fcntl.LOCK_EX); " + "print('held', flush=True); " + f"time.sleep({hold_seconds})", + str(lock_path), + ], + stdout=subprocess.PIPE, + text=True, + ) + # The descriptor PID: exits promptly on SIGTERM, and does not hold the lock. + parent = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(300)"], start_new_session=True) + try: + assert holder.stdout is not None + assert holder.stdout.readline().strip() == "held" + assert process.is_instance_alive(scope, base_dir=state_dir) + + process.write_descriptor( + process.InstanceDescriptor( + pid=parent.pid, + config=PlatformAppConfig(scope=scope, host="127.0.0.1", port=0, state_root=state_dir), + transport="tcp", + mode="daemon", + create_time=psutil.Process(parent.pid).create_time(), + ), + base_dir=state_dir, + ) + + started = time.monotonic() + result = process.stop_instance(scope, base_dir=state_dir, timeout=10, force=True) + elapsed = time.monotonic() - started + assert parent.pid in result.stopped_pids + + # The post-condition callers rely on, checked with no grace period. + assert not process.is_instance_alive(scope, base_dir=state_dir) + assert process.read_descriptor(scope, base_dir=state_dir) is None + # It blocked on the lock rather than returning the moment the PID died. + assert elapsed >= hold_seconds / 2 + finally: + for proc in (holder, parent): + try: + proc.kill() + proc.wait(timeout=3) + except (OSError, subprocess.TimeoutExpired): + pass + + @pytest.mark.integration def test_stop_instance_foreground_mode_requires_force(tmp_path: Path) -> None: """Stopping a foreground-mode instance without ``force=True`` should raise diff --git a/plugins/nemo-deployments/tests/integration/backends/docker/test_docker_backend.py b/plugins/nemo-deployments/tests/integration/backends/docker/test_docker_backend.py index a2938160fa..7fc8d23781 100644 --- a/plugins/nemo-deployments/tests/integration/backends/docker/test_docker_backend.py +++ b/plugins/nemo-deployments/tests/integration/backends/docker/test_docker_backend.py @@ -93,7 +93,14 @@ def _docker_backend_with_observe_timeout( *, oneshot_observe_timeout_seconds: int, ) -> DockerDeploymentBackend: - return _build_docker_backend(oneshot_observe_timeout_seconds=oneshot_observe_timeout_seconds) + # `pull_images` is off so the caller can pre-pull and keep the registry + # round-trip out of any window it times. `create_deployment` pulls + # unconditionally, not just when the image is missing locally, so leaving + # this on would put ~2s of Docker Hub latency inside the measurement. + return _build_docker_backend( + oneshot_observe_timeout_seconds=oneshot_observe_timeout_seconds, + pull_images=False, + ) def _always_http_config() -> DeploymentConfig: @@ -158,7 +165,13 @@ async def test_never_deployment_succeeds(docker_backend: DockerDeploymentBackend @pytest.mark.asyncio async def test_never_deployment_outlives_observe_wait_then_succeeds() -> None: """Long Never jobs return STARTING on create and finish via read_status polling.""" - job_sleep_seconds = 5 + # The job has to outlive the observe wait by a wide enough margin that a + # create_deployment which blocked until exit is unmistakable. The margin is + # what makes the timing assertion below meaningful, and it has to clear the + # cost of container create/start on a contended CI docker daemon (~3s + # observed) -- this test shares an xdist loadgroup with the heavier + # test_reconcile_docker cases, all hitting the same daemon. + job_sleep_seconds = 20 observe_timeout_seconds = 1 docker_backend = _docker_backend_with_observe_timeout( oneshot_observe_timeout_seconds=observe_timeout_seconds, @@ -169,8 +182,9 @@ async def test_never_deployment_outlives_observe_wait_then_succeeds() -> None: client = docker.from_env() try: - # Warm the image cache so the timed window below measures the observe wait - # rather than an uncached image pull. + # The backend is built with `pull_images=False`, so this pull is what puts + # the image on the host. Doing it here keeps it out of the timed window + # below, which is measuring the observe wait. await asyncio.to_thread(client.images.pull, ALPINE_IMAGE) started = time.monotonic() @@ -185,9 +199,13 @@ async def test_never_deployment_outlives_observe_wait_then_succeeds() -> None: assert created.status == "STARTING" assert "after observe wait" in (created.status_message or "") - assert create_elapsed < observe_timeout_seconds + 2.0 + # Deliberately loose. The STARTING assertions above are what pin "returned + # during the observe wait"; this one only has to catch the gross regression + # of blocking for the whole job, which would land at ~job_sleep_seconds. + # Keeping it well clear of CI jitter is worth more than a tight bound. + assert create_elapsed < observe_timeout_seconds + 8.0 - deadline = time.monotonic() + 15.0 + deadline = time.monotonic() + 45.0 status = created while time.monotonic() < deadline: status = await docker_backend.read_status(workspace="itest", name="sleep-job") diff --git a/sdk/python/nemo-platform/src/nemo_platform/local/process.py b/sdk/python/nemo-platform/src/nemo_platform/local/process.py index 7ee0cd4a9b..e211ea340a 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/local/process.py +++ b/sdk/python/nemo-platform/src/nemo_platform/local/process.py @@ -60,6 +60,10 @@ _SIGTERM_POLL_INTERVAL = 0.25 _SIGKILL_WAIT_TIMEOUT = 5.0 _DEFAULT_STOP_TIMEOUT = 30.0 +# How long ``stop_instance`` waits for the flock to be released after the last process it +# signaled has exited. The kernel drops the lock as part of closing the fd during process +# teardown, so this only has to cover the tail of that teardown, not a whole shutdown. +_LOCK_RELEASE_TIMEOUT = 5.0 def _pause(seconds: float) -> None: @@ -569,9 +573,14 @@ def _snapshot_children(pid: int) -> list[psutil.Process]: def _sweep_orphans(children: list[psutil.Process], timeout: float = 5.0) -> list[int]: """Terminate any still-alive processes from a prior snapshot. - Sends SIGTERM, waits up to *timeout*, then SIGKILL survivors. + Sends SIGTERM, waits up to *timeout*, then SIGKILL survivors and waits again. Returns PIDs that were signaled. Handles ``NoSuchProcess`` gracefully since children may have already exited during graceful shutdown. + + The second wait matters: these children inherited the instance flock fd from the + parent, and the kernel only releases the lock once the last holder's fd is closed. + Returning while a SIGKILLed child is still being torn down leaves the lock held, so + an ``is_instance_alive`` probe immediately after a "successful" stop reports True. """ alive_children = [c for c in children if c.is_running()] if not alive_children: @@ -586,12 +595,19 @@ def _sweep_orphans(children: list[psutil.Process], timeout: float = 5.0) -> list pass # Already exited or not owned by us — skip. _, still_alive = psutil.wait_procs(alive_children, timeout=timeout) + kill_sent: list[psutil.Process] = [] for child in still_alive: try: child.kill() + kill_sent.append(child) except (psutil.NoSuchProcess, psutil.AccessDenied): pass # Raced with exit or not owned — nothing to do. + if kill_sent: + _, unreaped = psutil.wait_procs(kill_sent, timeout=timeout) + if unreaped: + logger.warning("Orphaned child %s still alive after SIGKILL", [c.pid for c in unreaped]) + if killed: logger.info( "Swept %d orphaned child %s: %s", len(killed), "process" if len(killed) == 1 else "processes", killed @@ -673,10 +689,32 @@ def stop_instance( swept = _sweep_orphans(children) if children else [] + # A process exiting and its flock being released are not the same instant: the kernel + # drops the lock while closing fds during teardown. Callers treat a successful stop as + # "the scope is free now" (and immediately probe with ``is_instance_alive``), so hold the + # post-condition here rather than making every caller poll for it. + if not _wait_for_lock_release(scope, base_dir=base_dir, timeout=_LOCK_RELEASE_TIMEOUT): + logger.warning( + "Instance '%s' flock still held %.0fs after pid %d exited; a child may have outlived the sweep", + scope, + _LOCK_RELEASE_TIMEOUT, + pid, + ) + remove_descriptor(scope, base_dir=base_dir) return StopResult(stopped_pids=[pid], swept_children=swept) +def _wait_for_lock_release(scope: str, *, base_dir: Path | None, timeout: float) -> bool: + """Poll until the instance flock is free, or *timeout* elapses.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if not is_instance_alive(scope, base_dir=base_dir): + return True + _pause(_SIGTERM_POLL_INTERVAL) + return not is_instance_alive(scope, base_dir=base_dir) + + def _wait_for_pid_exit(pid: int, *, timeout: float) -> bool: deadline = time.monotonic() + timeout while time.monotonic() < deadline: diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_services_process.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_services_process.py index 6008aed5c9..a6f827326e 100644 --- a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_services_process.py +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_services_process.py @@ -5,6 +5,7 @@ from __future__ import annotations +import logging import os import signal import subprocess @@ -845,6 +846,49 @@ def test_escalates_to_sigkill(self) -> None: proc.kill() proc.wait(timeout=5) + def test_warns_when_child_survives_sigkill(self, caplog: pytest.LogCaptureFixture) -> None: + """A child that outlives SIGKILL is reported rather than silently dropped. + + Nothing survives SIGKILL on demand — the real trigger is a process wedged + in uninterruptible sleep, which a test cannot arrange — so `wait_procs` is + stubbed to keep reporting the child as alive. The terminate/kill calls are + still real; only the observation is simulated. + """ + proc = subprocess.Popen( + [ + sys.executable, + "-c", + "import signal, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(60)", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + time.sleep(0.3) + ps_child = psutil.Process(proc.pid) + # The first wait drives the SIGTERM -> SIGKILL escalation; the second + # is the one whose non-empty result trips the warning. + with ( + patch.object( + process_module.psutil, + "wait_procs", + side_effect=[([], [ps_child]), ([], [ps_child])], + ) as wait_procs, + caplog.at_level(logging.WARNING, logger=process_module.logger.name), + ): + killed = _sweep_orphans([ps_child], timeout=0.1) + + # Two waits, not one. The second is the post-SIGKILL wait this PR adds, + # so a single-wait implementation must not be able to satisfy this test. + assert wait_procs.call_count == 2 + assert proc.pid in killed + assert f"Orphaned child [{proc.pid}] still alive after SIGKILL" in caplog.text + proc.wait(timeout=5) + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=5) + # --------------------------------------------------------------------------- # stop_instance — child sweep integration diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/local/test_daemon_lifecycle.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/local/test_daemon_lifecycle.py index d38ac7c8eb..b7c3b6bf31 100644 --- a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/local/test_daemon_lifecycle.py +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/local/test_daemon_lifecycle.py @@ -376,6 +376,149 @@ def test_stop_instance_sweeps_orphaned_children(tmp_path: Path) -> None: pass +def test_wait_for_lock_release_blocks_until_holder_exits(tmp_path: Path) -> None: + """``_wait_for_lock_release`` should return only once the flock is actually free.""" + scope = "unit-lock-release" + state_dir = tmp_path / "state" + lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME + lock_path.parent.mkdir(parents=True) + lock_path.touch() + + # A holder that keeps the flock for ~1s, then exits and releases it. + holder = subprocess.Popen( + [ + sys.executable, + "-c", + "import fcntl, sys, time; " + "fd = open(sys.argv[1], 'r+'); " + "fcntl.flock(fd, fcntl.LOCK_EX); " + "print('held', flush=True); " + "time.sleep(1.0)", + str(lock_path), + ], + stdout=subprocess.PIPE, + text=True, + ) + try: + assert holder.stdout is not None + assert holder.stdout.readline().strip() == "held" + assert process.is_instance_alive(scope, base_dir=state_dir) + + started = time.monotonic() + assert process._wait_for_lock_release(scope, base_dir=state_dir, timeout=10.0) + assert not process.is_instance_alive(scope, base_dir=state_dir) + # It waited rather than returning eagerly on a still-held lock. + assert time.monotonic() - started >= 0.5 + finally: + holder.kill() + holder.wait(timeout=5) + + +def test_wait_for_lock_release_times_out_while_held(tmp_path: Path) -> None: + """A lock held for the whole window should report failure, not success.""" + scope = "unit-lock-held" + state_dir = tmp_path / "state" + lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME + lock_path.parent.mkdir(parents=True) + lock_path.touch() + + holder = subprocess.Popen( + [ + sys.executable, + "-c", + "import fcntl, sys, time; " + "fd = open(sys.argv[1], 'r+'); " + "fcntl.flock(fd, fcntl.LOCK_EX); " + "print('held', flush=True); " + "time.sleep(300)", + str(lock_path), + ], + stdout=subprocess.PIPE, + text=True, + ) + try: + assert holder.stdout is not None + assert holder.stdout.readline().strip() == "held" + assert not process._wait_for_lock_release(scope, base_dir=state_dir, timeout=0.5) + finally: + holder.kill() + holder.wait(timeout=5) + + +@pytest.mark.integration +def test_stop_instance_releases_lock_held_by_surviving_child(tmp_path: Path) -> None: + """Regression: ``stop_instance`` must not report success while the scope's + flock is still held. + + A process exiting and its flock being released are not the same instant — the + kernel drops the lock while closing fds during teardown, and any process that + inherited the fd keeps it held until *it* is gone too. ``stop_instance`` used + to return as soon as the descriptor PID was dead, so an immediate + ``is_instance_alive`` probe raced that teardown and intermittently saw True. + + The holder here is deliberately outside the parent's process tree: that makes + the window wide and fixed instead of scheduler-dependent, and it models the + case the sweep can't reach (a lock-inheriting process spawned after the child + snapshot was taken). + """ + scope = "integ-lock-release" + state_dir = tmp_path / "state" + lock_path = state_dir / "instances" / scope / process.LOCK_FILENAME + lock_path.parent.mkdir(parents=True) + lock_path.touch() + + hold_seconds = 2.0 + holder = subprocess.Popen( + [ + sys.executable, + "-c", + "import fcntl, sys, time; " + "fd = open(sys.argv[1], 'r+'); " + "fcntl.flock(fd, fcntl.LOCK_EX); " + "print('held', flush=True); " + f"time.sleep({hold_seconds})", + str(lock_path), + ], + stdout=subprocess.PIPE, + text=True, + ) + # The descriptor PID: exits promptly on SIGTERM, and does not hold the lock. + parent = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(300)"], start_new_session=True) + try: + assert holder.stdout is not None + assert holder.stdout.readline().strip() == "held" + assert process.is_instance_alive(scope, base_dir=state_dir) + + process.write_descriptor( + process.InstanceDescriptor( + pid=parent.pid, + config=PlatformAppConfig(scope=scope, host="127.0.0.1", port=0, state_root=state_dir), + transport="tcp", + mode="daemon", + create_time=psutil.Process(parent.pid).create_time(), + ), + base_dir=state_dir, + ) + + started = time.monotonic() + result = process.stop_instance(scope, base_dir=state_dir, timeout=10, force=True) + elapsed = time.monotonic() - started + assert parent.pid in result.stopped_pids + + # The post-condition callers rely on, checked with no grace period. + assert not process.is_instance_alive(scope, base_dir=state_dir) + assert process.read_descriptor(scope, base_dir=state_dir) is None + # It blocked on the lock rather than returning the moment the PID died. + assert elapsed >= hold_seconds / 2 + finally: + for proc in (holder, parent): + try: + proc.kill() + proc.wait(timeout=3) + except (OSError, subprocess.TimeoutExpired): + pass + + @pytest.mark.integration def test_stop_instance_foreground_mode_requires_force(tmp_path: Path) -> None: """Stopping a foreground-mode instance without ``force=True`` should raise