From 4664effee6338e5d5ecfd15dae34fef7af1a3b24 Mon Sep 17 00:00:00 2001 From: Justin Date: Fri, 16 Oct 2026 20:05:54 -0400 Subject: [PATCH 1/5] fix(cli): delete dev endpoints when rp flash dev is stopped by a signal A dev session started in the background from a non-interactive shell inherits SIGINT as ignored, so Python never raises KeyboardInterrupt, and SIGTERM has no handler. Either way the process stopped without its teardown and left the session's endpoints running and billing. A stop during a long entrypoint call also waited for that call indefinitely. Install SIGINT and SIGTERM handlers on the session loop. The first signal cancels the session, which runs teardown and deletes the endpoints. A second signal stops waiting for a running entrypoint. The entrypoint thread is a daemon and tolerates the loop closing under it. --- runpod/rp_cli/main.py | 63 ++++++++++++++++++-- tests/test_cli/test_deploy_command.py | 85 +++++++++++++++++++++++++++ 2 files changed, 143 insertions(+), 5 deletions(-) diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index 5c4b4bfdc..01aa7fe6f 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -8,6 +8,7 @@ import asyncio import os +import signal import sys from pathlib import Path @@ -405,11 +406,16 @@ async def _wait_for_rerun(watcher: "FileWatcher") -> str: pass await asyncio.sleep(0.5) + # set once a second stop signal arrives: teardown stops waiting for + # a running entrypoint + force_stop: "list[asyncio.Event]" = [] + async def _run_entrypoint_cancellable(fn) -> None: """run user code without blocking the event loop. cancellation waits for the entrypoint to finish so its remote - calls cannot race with session teardown. + calls cannot race with session teardown, unless a second stop + signal forces teardown first. """ loop = asyncio.get_running_loop() done: asyncio.Future[None] = loop.create_future() @@ -422,6 +428,13 @@ def _finish(exc: "BaseException | None") -> None: else: done.set_exception(exc) + def _report(exc: "BaseException | None") -> None: + try: + loop.call_soon_threadsafe(_finish, exc) + except RuntimeError: + # a forced stop closed the loop while this entrypoint ran + pass + def _runner() -> None: try: run_entrypoint(fn) @@ -430,17 +443,56 @@ def _runner() -> None: except (Exception, KeyboardInterrupt, SystemExit) as exc: # bind the exception now: `as exc` is unbound once the # except block exits, before the loop callback runs - loop.call_soon_threadsafe(_finish, exc) + _report(exc) else: - loop.call_soon_threadsafe(_finish, None) + _report(None) - threading.Thread(target=_runner).start() + # daemon: a forced stop must not leave process exit waiting on + # a long remote call + threading.Thread(target=_runner, daemon=True).start() try: return await asyncio.shield(done) except asyncio.CancelledError: - await done + if force_stop and not done.done(): + forced = asyncio.ensure_future(force_stop[0].wait()) + await asyncio.wait({done, forced}, return_when=asyncio.FIRST_COMPLETED) + forced.cancel() + # an abandoned entrypoint's error is not retrieved + done.add_done_callback(lambda f: f.cancelled() or f.exception()) + else: + await done raise + def _install_stop_signals() -> None: + """stop cleanly on sigint and sigterm. + + python only turns sigint into KeyboardInterrupt when it was not + ignored at startup, and shells start background jobs with sigint + ignored; sigterm has no handler at all. without these a stopped + session would leave its endpoints running. the first signal + cancels the session (teardown deletes the endpoints), the second + stops waiting for a running entrypoint. + """ + loop = asyncio.get_running_loop() + task = asyncio.current_task() + force_stop.append(asyncio.Event()) + seen = [] + + def _on_signal() -> None: + seen.append(True) + if len(seen) == 1: + task.cancel() + else: + force_stop[0].set() + + for sig in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(sig, _on_signal) + except (NotImplementedError, RuntimeError, ValueError): + # windows event loops have no signal handlers; ctrl-c + # still raises KeyboardInterrupt there + pass + async def _session() -> int: nonlocal apps, entrypoint ui.set_name_width(_all_resource_names(apps)) @@ -448,6 +500,7 @@ async def _session() -> int: watcher = FileWatcher([module.parent]) ui.dev_banner([a.name for a in apps], str(module.name)) + _install_stop_signals() try: await session.start() ui.resources_table(_table_rows(session)) diff --git a/tests/test_cli/test_deploy_command.py b/tests/test_cli/test_deploy_command.py index 2fe5942e7..29b4ca6ff 100644 --- a/tests/test_cli/test_deploy_command.py +++ b/tests/test_cli/test_deploy_command.py @@ -1,5 +1,6 @@ """tests for rp flash deploy and rp flash dev command wiring.""" +import sys from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -290,6 +291,90 @@ async def delete(endpoint_id): assert result.exit_code == 0, result.output assert events == ["cancelled", "finished", "deleted"] + def _signal_session(self, tmp_path, monkeypatch, entrypoint): + import asyncio + + module = tmp_path / "main.py" + module.write_text(ENTRYPOINT_SOURCE) + monkeypatch.chdir(tmp_path) + events = [] + api = AsyncMock() + api.list_my_endpoints.return_value = [] + api.save_endpoint.return_value = {"id": "ep-active"} + + async def delete(endpoint_id): + events.append("deleted") + return True + + api.delete_endpoint.side_effect = delete + with ( + patch("runpod.apps.dev.AppsApiClient", return_value=api), + patch( + "runpod.apps.entrypoint.run_entrypoint", + side_effect=lambda fn: entrypoint(events), + ), + ): + result = _runner().invoke(cli, ["flash", "dev", str(module)]) + return result, events + + @pytest.mark.skipif(sys.platform == "win32", reason="posix signals") + def test_sigterm_deletes_endpoints(self, tmp_path, monkeypatch): + import os + import signal + import time + + def entrypoint(events): + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(0.1) + events.append("finished") + + result, events = self._signal_session(tmp_path, monkeypatch, entrypoint) + assert result.exit_code == 0, result.output + assert events == ["finished", "deleted"] + + @pytest.mark.skipif(sys.platform == "win32", reason="posix signals") + def test_sigint_cleans_up_even_when_ignored_at_start(self, tmp_path, monkeypatch): + import os + import signal + import time + + # shells start background jobs with sigint ignored + previous = signal.signal(signal.SIGINT, signal.SIG_IGN) + + def entrypoint(events): + os.kill(os.getpid(), signal.SIGINT) + time.sleep(0.1) + events.append("finished") + + try: + result, events = self._signal_session(tmp_path, monkeypatch, entrypoint) + finally: + signal.signal(signal.SIGINT, previous) + assert result.exit_code == 0, result.output + assert events == ["finished", "deleted"] + + @pytest.mark.skipif(sys.platform == "win32", reason="posix signals") + def test_second_signal_forces_teardown(self, tmp_path, monkeypatch): + import os + import signal + import threading + + release = threading.Event() + + def entrypoint(events): + os.kill(os.getpid(), signal.SIGTERM) + threading.Timer(0.2, os.kill, (os.getpid(), signal.SIGTERM)).start() + # a long remote call that outlives the stop request + release.wait(5) + events.append("finished") + + try: + result, events = self._signal_session(tmp_path, monkeypatch, entrypoint) + finally: + release.set() + assert result.exit_code == 0, result.output + assert events[0] == "deleted" + def test_module_without_entrypoint(self, tmp_path, monkeypatch): module = tmp_path / "main.py" module.write_text(APP_SOURCE) From 0e88f577454272ab0cdac1f059f29163d8c9ebb7 Mon Sep 17 00:00:00 2001 From: Justin Date: Mon, 19 Oct 2026 23:09:08 -0400 Subject: [PATCH 2/5] fix(cli): finish dev teardown and track in-flight creates on a stop signal a signal during teardown cancelled the deletes still pending, and a signal during endpoint creation dropped the id the platform had already assigned. teardown now ignores signals, and a cancelled create waits for its reply and records the endpoint so teardown deletes it. --- runpod/apps/dev.py | 23 ++++++++-- runpod/rp_cli/main.py | 9 +++- tests/test_cli/test_deploy_command.py | 62 +++++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 4 deletions(-) diff --git a/runpod/apps/dev.py b/runpod/apps/dev.py index c4e95ad1c..fc9bc6cc0 100644 --- a/runpod/apps/dev.py +++ b/runpod/apps/dev.py @@ -6,6 +6,7 @@ api is the only source of truth; nothing is persisted locally. """ +import asyncio import logging import os as _os from typing import Dict, List, Optional, Union @@ -288,7 +289,7 @@ async def start(self) -> None: log.info("adopted dev endpoint %s (%s)", name, endpoint_id) else: self._emit("provisioning", spec.name, spec.kind.value, hardware) - result = await self.api.save_endpoint(payload) + result = await self._create_tracked(name, spec.name, payload) endpoint_id = result["id"] log.info("provisioned dev endpoint %s (%s)", name, endpoint_id) @@ -360,8 +361,9 @@ async def refresh(self, apps: List[App]) -> None: if existing_id: payload["id"] = existing_id - - result = await self.api.save_endpoint(payload) + result = await self.api.save_endpoint(payload) + else: + result = await self._create_tracked(name, handle.spec.name, payload) endpoint_id = result["id"] self._endpoints[name] = endpoint_id @@ -393,6 +395,21 @@ async def refresh(self, apps: List[App]) -> None: self.generation, ) + async def _create_tracked(self, name: str, resource: str, payload: Dict) -> Dict: + """create an endpoint, recording it even if the session is + cancelled mid-request, so teardown deletes it.""" + create = asyncio.ensure_future(self.api.save_endpoint(payload)) + try: + return await asyncio.shield(create) + except asyncio.CancelledError: + try: + result = await create + except Exception: # noqa: BLE001 - the create failed; nothing to track + raise asyncio.CancelledError from None + self._endpoints[name] = result["id"] + self._resource_names[name] = resource + raise + async def stop(self, events: Optional[object] = None) -> None: """delete every endpoint this session owns. diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index 01aa7fe6f..0a669b365 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -409,6 +409,8 @@ async def _wait_for_rerun(watcher: "FileWatcher") -> str: # set once a second stop signal arrives: teardown stops waiting for # a running entrypoint force_stop: "list[asyncio.Event]" = [] + # set when teardown starts: signals then leave it to finish + tearing_down: "list[bool]" = [] async def _run_entrypoint_cancellable(fn) -> None: """run user code without blocking the event loop. @@ -471,7 +473,8 @@ def _install_stop_signals() -> None: ignored; sigterm has no handler at all. without these a stopped session would leave its endpoints running. the first signal cancels the session (teardown deletes the endpoints), the second - stops waiting for a running entrypoint. + stops waiting for a running entrypoint. signals during teardown + are ignored so every endpoint still gets deleted. """ loop = asyncio.get_running_loop() task = asyncio.current_task() @@ -480,6 +483,9 @@ def _install_stop_signals() -> None: def _on_signal() -> None: seen.append(True) + if tearing_down: + # cancelling teardown would skip the remaining deletes + return if len(seen) == 1: task.cancel() else: @@ -525,6 +531,7 @@ async def _session() -> int: ui.set_name_width(_all_resource_names(apps)) await session.refresh(apps) finally: + tearing_down.append(True) events = ui.CleanupEvents() try: await session.stop(events=events) diff --git a/tests/test_cli/test_deploy_command.py b/tests/test_cli/test_deploy_command.py index 29b4ca6ff..48fd9c567 100644 --- a/tests/test_cli/test_deploy_command.py +++ b/tests/test_cli/test_deploy_command.py @@ -375,6 +375,68 @@ def entrypoint(events): assert result.exit_code == 0, result.output assert events[0] == "deleted" + @pytest.mark.skipif(sys.platform == "win32", reason="posix signals") + def test_signal_during_teardown_still_deletes(self, tmp_path, monkeypatch): + import asyncio + import os + import signal + + module = tmp_path / "main.py" + module.write_text(ENTRYPOINT_SOURCE) + monkeypatch.chdir(tmp_path) + events = [] + api = AsyncMock() + api.list_my_endpoints.return_value = [] + api.save_endpoint.return_value = {"id": "ep-active"} + + async def delete(endpoint_id): + os.kill(os.getpid(), signal.SIGTERM) + await asyncio.sleep(0.1) + events.append(("deleted", endpoint_id)) + return True + + api.delete_endpoint.side_effect = delete + with ( + patch("runpod.apps.dev.AppsApiClient", return_value=api), + patch("runpod.apps.entrypoint.run_entrypoint"), + ): + result = _runner().invoke(cli, ["flash", "dev", str(module), "--once"]) + assert result.exit_code == 0, result.output + assert events == [("deleted", "ep-active")] + + @pytest.mark.skipif(sys.platform == "win32", reason="posix signals") + def test_signal_during_create_deletes_the_new_endpoint(self, tmp_path, monkeypatch): + import asyncio + import os + import signal + + module = tmp_path / "main.py" + module.write_text(ENTRYPOINT_SOURCE) + monkeypatch.chdir(tmp_path) + deleted = [] + api = AsyncMock() + api.list_my_endpoints.return_value = [] + + async def create(payload): + # the platform creates the endpoint before the reply arrives + os.kill(os.getpid(), signal.SIGTERM) + await asyncio.sleep(0.1) + return {"id": "ep-new"} + + async def delete(endpoint_id): + deleted.append(endpoint_id) + return True + + api.save_endpoint.side_effect = create + api.delete_endpoint.side_effect = delete + with ( + patch("runpod.apps.dev.AppsApiClient", return_value=api), + patch("runpod.apps.entrypoint.run_entrypoint"), + ): + result = _runner().invoke(cli, ["flash", "dev", str(module)]) + assert result.exit_code == 0, result.output + assert deleted == ["ep-new"] + def test_module_without_entrypoint(self, tmp_path, monkeypatch): module = tmp_path / "main.py" module.write_text(APP_SOURCE) From ecac29a23c09525d8059f6356e9896536afa4a40 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 20 Oct 2026 12:28:31 -0400 Subject: [PATCH 3/5] refactor(cli): plain flags for dev stop signals --- runpod/rp_cli/main.py | 43 +++++++++++++++------------ tests/test_cli/test_deploy_command.py | 2 -- 2 files changed, 24 insertions(+), 21 deletions(-) diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index 0a669b365..da5ddd580 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -406,11 +406,11 @@ async def _wait_for_rerun(watcher: "FileWatcher") -> str: pass await asyncio.sleep(0.5) - # set once a second stop signal arrives: teardown stops waiting for - # a running entrypoint - force_stop: "list[asyncio.Event]" = [] - # set when teardown starts: signals then leave it to finish - tearing_down: "list[bool]" = [] + # set by a second stop signal: teardown stops waiting for a running + # entrypoint + force_stop = asyncio.Event() + # once teardown starts, signals leave it to finish + tearing_down = False async def _run_entrypoint_cancellable(fn) -> None: """run user code without blocking the event loop. @@ -430,6 +430,12 @@ def _finish(exc: "BaseException | None") -> None: else: done.set_exception(exc) + def _discard_result(future: "asyncio.Future[None]") -> None: + # an abandoned entrypoint's error would otherwise be logged + # as never retrieved + if not future.cancelled(): + future.exception() + def _report(exc: "BaseException | None") -> None: try: loop.call_soon_threadsafe(_finish, exc) @@ -455,14 +461,13 @@ def _runner() -> None: try: return await asyncio.shield(done) except asyncio.CancelledError: - if force_stop and not done.done(): - forced = asyncio.ensure_future(force_stop[0].wait()) + if done.done(): + await done + else: + forced = asyncio.ensure_future(force_stop.wait()) await asyncio.wait({done, forced}, return_when=asyncio.FIRST_COMPLETED) forced.cancel() - # an abandoned entrypoint's error is not retrieved - done.add_done_callback(lambda f: f.cancelled() or f.exception()) - else: - await done + done.add_done_callback(_discard_result) raise def _install_stop_signals() -> None: @@ -478,18 +483,18 @@ def _install_stop_signals() -> None: """ loop = asyncio.get_running_loop() task = asyncio.current_task() - force_stop.append(asyncio.Event()) - seen = [] + stopping = False def _on_signal() -> None: - seen.append(True) + nonlocal stopping if tearing_down: # cancelling teardown would skip the remaining deletes return - if len(seen) == 1: - task.cancel() + if stopping: + force_stop.set() else: - force_stop[0].set() + stopping = True + task.cancel() for sig in (signal.SIGINT, signal.SIGTERM): try: @@ -500,7 +505,7 @@ def _on_signal() -> None: pass async def _session() -> int: - nonlocal apps, entrypoint + nonlocal apps, entrypoint, tearing_down ui.set_name_width(_all_resource_names(apps)) session = DevSession(apps, events=ui.DevEvents()) watcher = FileWatcher([module.parent]) @@ -531,7 +536,7 @@ async def _session() -> int: ui.set_name_width(_all_resource_names(apps)) await session.refresh(apps) finally: - tearing_down.append(True) + tearing_down = True events = ui.CleanupEvents() try: await session.stop(events=events) diff --git a/tests/test_cli/test_deploy_command.py b/tests/test_cli/test_deploy_command.py index 48fd9c567..09183584e 100644 --- a/tests/test_cli/test_deploy_command.py +++ b/tests/test_cli/test_deploy_command.py @@ -292,8 +292,6 @@ async def delete(endpoint_id): assert events == ["cancelled", "finished", "deleted"] def _signal_session(self, tmp_path, monkeypatch, entrypoint): - import asyncio - module = tmp_path / "main.py" module.write_text(ENTRYPOINT_SOURCE) monkeypatch.chdir(tmp_path) From fdf0757b0da9afb6abbf7463a9adc895ee58c199 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 20 Oct 2026 13:00:22 -0400 Subject: [PATCH 4/5] refactor(cli): name the post-stop wait and the late-create record --- runpod/apps/dev.py | 19 +++++++++++++------ runpod/rp_cli/main.py | 32 +++++++++++++++++++------------- 2 files changed, 32 insertions(+), 19 deletions(-) diff --git a/runpod/apps/dev.py b/runpod/apps/dev.py index fc9bc6cc0..1c955f604 100644 --- a/runpod/apps/dev.py +++ b/runpod/apps/dev.py @@ -402,14 +402,21 @@ async def _create_tracked(self, name: str, resource: str, payload: Dict) -> Dict try: return await asyncio.shield(create) except asyncio.CancelledError: - try: - result = await create - except Exception: # noqa: BLE001 - the create failed; nothing to track - raise asyncio.CancelledError from None - self._endpoints[name] = result["id"] - self._resource_names[name] = resource + await self._record_late_create(create, name, resource) raise + async def _record_late_create( + self, create: "asyncio.Future[Dict]", name: str, resource: str + ) -> None: + """wait for a create the session was cancelled during, and record + the endpoint it made.""" + try: + result = await create + except Exception: # noqa: BLE001 - the create failed; nothing to track + return + self._endpoints[name] = result["id"] + self._resource_names[name] = resource + async def stop(self, events: Optional[object] = None) -> None: """delete every endpoint this session owns. diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index da5ddd580..77a23d173 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -430,12 +430,6 @@ def _finish(exc: "BaseException | None") -> None: else: done.set_exception(exc) - def _discard_result(future: "asyncio.Future[None]") -> None: - # an abandoned entrypoint's error would otherwise be logged - # as never retrieved - if not future.cancelled(): - future.exception() - def _report(exc: "BaseException | None") -> None: try: loop.call_soon_threadsafe(_finish, exc) @@ -461,15 +455,20 @@ def _runner() -> None: try: return await asyncio.shield(done) except asyncio.CancelledError: - if done.done(): - await done - else: - forced = asyncio.ensure_future(force_stop.wait()) - await asyncio.wait({done, forced}, return_when=asyncio.FIRST_COMPLETED) - forced.cancel() - done.add_done_callback(_discard_result) + await _settle_after_stop(done) raise + async def _settle_after_stop(done: "asyncio.Future[None]") -> None: + """after a stop, wait for the still-running entrypoint so its remote + calls cannot race teardown, unless a second signal forces the stop.""" + if done.done(): + await done # surfaces the entrypoint's own error + return + forced = asyncio.ensure_future(force_stop.wait()) + await asyncio.wait({done, forced}, return_when=asyncio.FIRST_COMPLETED) + forced.cancel() + done.add_done_callback(_discard_result) + def _install_stop_signals() -> None: """stop cleanly on sigint and sigterm. @@ -551,6 +550,13 @@ async def _session() -> int: raise click.ClickException(str(exc)) from exc +def _discard_result(future: "asyncio.Future[None]") -> None: + """mark an abandoned entrypoint's error as retrieved, so asyncio does + not log it as never retrieved.""" + if not future.cancelled(): + future.exception() + + # ------------------------------------------------------------------ logs From 26edffa9bf8bacd3c6d9c64b827dac6a84c1e1f9 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:03:28 -0700 Subject: [PATCH 5/5] fix(cli): preserve dev cancellation when entrypoint fails --- runpod/rp_cli/main.py | 2 +- tests/test_cli/test_deploy_command.py | 42 +++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index 77a23d173..5be6aaef0 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -462,7 +462,7 @@ async def _settle_after_stop(done: "asyncio.Future[None]") -> None: """after a stop, wait for the still-running entrypoint so its remote calls cannot race teardown, unless a second signal forces the stop.""" if done.done(): - await done # surfaces the entrypoint's own error + _discard_result(done) return forced = asyncio.ensure_future(force_stop.wait()) await asyncio.wait({done, forced}, return_when=asyncio.FIRST_COMPLETED) diff --git a/tests/test_cli/test_deploy_command.py b/tests/test_cli/test_deploy_command.py index 09183584e..59ea111ee 100644 --- a/tests/test_cli/test_deploy_command.py +++ b/tests/test_cli/test_deploy_command.py @@ -330,6 +330,48 @@ def entrypoint(events): assert result.exit_code == 0, result.output assert events == ["finished", "deleted"] + @pytest.mark.skipif(sys.platform == "win32", reason="posix signals") + @pytest.mark.parametrize("stop_signal", ["SIGINT", "SIGTERM"]) + def test_stop_signal_preserves_cancellation_on_entrypoint_failure( + self, tmp_path, monkeypatch, stop_signal + ): + import asyncio + import signal + + handlers = {} + shield = asyncio.shield + + def install_handler(loop, sig, callback): + handlers[sig] = callback + + def stop_on_completion(future): + waiter = shield(future) + if not isinstance(future, asyncio.Task): + future.add_done_callback( + lambda _: handlers[getattr(signal, stop_signal)]() + ) + return waiter + + def entrypoint(events): + events.append("failed") + raise RuntimeError("entrypoint failed while stopping") + + with ( + patch.object( + asyncio.SelectorEventLoop, "add_signal_handler", install_handler + ), + patch("asyncio.shield", side_effect=stop_on_completion), + patch( + "runpod.apps.watch.FileWatcher.changed", + side_effect=AssertionError("a stopped session must not resume watching"), + ) as changed, + ): + result, events = self._signal_session(tmp_path, monkeypatch, entrypoint) + + assert result.exit_code == 0, result.output + assert events == ["failed", "deleted"] + changed.assert_not_called() + @pytest.mark.skipif(sys.platform == "win32", reason="posix signals") def test_sigint_cleans_up_even_when_ignored_at_start(self, tmp_path, monkeypatch): import os