From c82827a8d04738c4adb48736fda6e2a7eb0c0ee5 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Fri, 13 Nov 2026 10:26:21 -0800 Subject: [PATCH 1/5] fix(apps): roll back new endpoints when deployment fails --- runpod/apps/deploy.py | 147 ++++++++++----- runpod/apps/errors.py | 4 + runpod/apps/spec.py | 17 +- runpod/rp_cli/console.py | 22 ++- runpod/rp_cli/main.py | 8 +- tests/test_apps/test_partial_deploy.py | 252 +++++++++++++++++++++++++ 6 files changed, 400 insertions(+), 50 deletions(-) create mode 100644 tests/test_apps/test_partial_deploy.py diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py index c166248b..66a2b65c 100644 --- a/runpod/apps/deploy.py +++ b/runpod/apps/deploy.py @@ -4,6 +4,7 @@ import logging import os import secrets +import shlex import tarfile import tempfile from dataclasses import dataclass, field @@ -19,7 +20,11 @@ BuildResult, build_environment, ) -from .errors import InvalidResourceError, ScheduleNotSupported +from .errors import ( + EndpointReconciliationError, + InvalidResourceError, + ScheduleNotSupported, +) from .discovery import resolve_project, validate_apps from .schedule import SCHEDULES_ENABLED from .shim import runtime_launcher @@ -403,6 +408,8 @@ async def reconcile_endpoints( resource name -> endpoint id for everything provisioned. existing endpoints (matched by name) are updated in place; endpoints whose resources disappeared are deleted. + + failures trigger best-effort deletion of endpoints created by this call. """ for handle in app.resources.values(): handle.spec.validate() @@ -426,47 +433,94 @@ async def reconcile_endpoints( resolver = resolver or VolumeResolver(client) endpoints: Dict[str, str] = {} - for name, handle in sorted(provisionable.items()): - members = None - if needs_gpu_catalog(handle.spec.gpu): - members = await client.gpu_pool_members() - if name in existing: - template_name = existing_templates.get(name) - else: - template_name = _new_template_name(app, handle.spec, template_token) - payload = _deployed_endpoint_input( - app, - handle.spec, - environment["id"], - build_id, - python_version, - env_name=environment["name"], - pool_members=members or {}, - template_name=template_name, - ) - await attach_endpoint_volumes(payload, handle.spec, resolver, app) - auth_id = await resolve_registry_auth(handle.spec.registry_auth, api=client) - if auth_id: - payload["template"]["containerRegistryAuthId"] = auth_id - if handle.spec.model: - from .model import model_reference - - payload["modelReferences"] = [model_reference(handle.spec.model)] - if name in existing: - payload["id"] = existing[name] - result = await client.save_endpoint(payload) - endpoints[name] = result["id"] - log.info( - "%s endpoint %s (%s)", - "updated" if name in existing else "provisioned", - name, - result["id"], + completed: Dict[str, Dict[str, str]] = { + "created": {}, "updated": {}, "deleted": {} + } + try: + for name, handle in sorted(provisionable.items()): + operation = "prepare" + members = None + if needs_gpu_catalog(handle.spec.gpu): + members = await client.gpu_pool_members() + if name in existing: + template_name = existing_templates.get(name) + else: + template_name = _new_template_name(app, handle.spec, template_token) + payload = _deployed_endpoint_input( + app, + handle.spec, + environment["id"], + build_id, + python_version, + env_name=environment["name"], + pool_members=members or {}, + template_name=template_name, + ) + await attach_endpoint_volumes(payload, handle.spec, resolver, app) + auth_id = await resolve_registry_auth(handle.spec.registry_auth, api=client) + if auth_id: + payload["template"]["containerRegistryAuthId"] = auth_id + if handle.spec.model: + from .model import model_reference + + payload["modelReferences"] = [model_reference(handle.spec.model)] + if name in existing: + payload["id"] = existing[name] + operation = "update" if name in existing else "create" + result = await client.save_endpoint(payload) + endpoints[name] = result["id"] + completed["updated" if name in existing else "created"][name] = result["id"] + log.info("%s endpoint %s (%s)", operation, name, result["id"]) + + for name, endpoint_id in existing.items(): + if name not in provisionable: + operation = "delete" + await client.delete_endpoint(endpoint_id) + completed["deleted"][name] = endpoint_id + log.info("deleted removed endpoint %s (%s)", name, endpoint_id) + except Exception as exc: + failed_id = f" ({existing[name]})" if name in existing else "" + lines = [ + f"endpoint reconciliation failed for {app.name}/{environment['name']} " + f"(environment {environment['id']}, build {build_id}).", + f"failed to {operation} {name}{failed_id}: {exc}", + ] + rolled_back = {} + cleanup_failures = {} + for resource, endpoint_id in reversed(list(completed["created"].items())): + try: + if not await client.delete_endpoint(endpoint_id): + raise RuntimeError("endpoint deletion was not confirmed") + rolled_back[resource] = endpoint_id + except Exception as cleanup_exc: + cleanup_failures[resource] = endpoint_id + lines.append( + f"rollback failed for {resource} ({endpoint_id}): {cleanup_exc}" + ) + for status, changes in ( + ("rolled back created endpoints", rolled_back), + ("created endpoints still requiring cleanup", cleanup_failures), + ("updated endpoints (not rolled back)", completed["updated"]), + ("deleted endpoints (not restored)", completed["deleted"]), + ): + summary = ", ".join( + f"{resource} ({endpoint_id})" for resource, endpoint_id in changes.items() + ) + if summary: + lines.append(f"{status}: {summary}") + lines.append("Remaining operations were not attempted.") + if operation != "prepare": + lines.append( + "The failed request may have taken effect if its response was lost." + ) + lines.append( + f"Inspect this environment with `rp flash env get " + f"{shlex.quote(environment['name'])} --app {shlex.quote(app.name)}` " + "before retrying. " + "Endpoints with active workers can keep billing. " + "Delete unwanted endpoints or scale their minimum workers to 0." ) - - for name, endpoint_id in existing.items(): - if name not in provisionable: - await client.delete_endpoint(endpoint_id) - log.info("deleted removed endpoint %s (%s)", name, endpoint_id) + raise EndpointReconciliationError("\n".join(lines)) from exc return endpoints @@ -602,9 +656,14 @@ async def deploy_app( log.info("activated build %s on %s/%s", build["id"], app.name, env_name) _phase(events, "endpoints") - endpoints = await reconcile_endpoints( - client, app, environment, build["id"], python_version, resolver=resolver - ) + try: + endpoints = await reconcile_endpoints( + client, app, environment, build["id"], python_version, resolver=resolver + ) + except EndpointReconciliationError as exc: + raise EndpointReconciliationError( + f"build {build['id']} was activated on {app.name}/{env_name}.\n{exc}" + ) from exc endpoint_ready = getattr(events, "endpoint_ready", None) if endpoint_ready is not None: for name, endpoint_id in sorted(endpoints.items()): diff --git a/runpod/apps/errors.py b/runpod/apps/errors.py index 2d34177f..78a12ddb 100644 --- a/runpod/apps/errors.py +++ b/runpod/apps/errors.py @@ -49,3 +49,7 @@ def __init__(self) -> None: class InvalidResourceError(AppError): """a decorator was applied to an unsupported target or with bad config.""" + + +class EndpointReconciliationError(AppError): + """endpoint reconciliation failed with rollback results and remaining changes.""" diff --git a/runpod/apps/spec.py b/runpod/apps/spec.py index 391471e7..ead24ee7 100644 --- a/runpod/apps/spec.py +++ b/runpod/apps/spec.py @@ -13,7 +13,7 @@ from .errors import InvalidResourceError from .datacenter import DataCenter, datacenter_id -from .gpu import GpuGroup, GpuLike, GpuType +from .gpu import CpuInstanceType, GpuGroup, GpuLike, GpuType from .volume import Volume, normalize_mounts, validate_mounts log = logging.getLogger(__name__) @@ -126,6 +126,12 @@ def normalize_cpu( ) if not cpu or any(not isinstance(c, str) or not c.strip() for c in cpu): raise InvalidResourceError("cpu must contain non-empty instance ids") + supported = {flavor.value for flavor in CpuInstanceType} + if unsupported := set(cpu) - supported: + raise InvalidResourceError( + f"unsupported cpu instance ids: {', '.join(sorted(unsupported))}. " + f"expected one of {', '.join(sorted(supported))}" + ) return list(cpu) @@ -246,6 +252,15 @@ def validate(self) -> None: raise InvalidResourceError( f"resource '{self.name}': gpu_count must be >= 1" ) + if ( + isinstance(self.idle_timeout, bool) + or not isinstance(self.idle_timeout, int) + or not 1 <= self.idle_timeout <= 3600 + ): + raise InvalidResourceError( + f"resource '{self.name}': idle_timeout must be an integer " + "between 1 and 3600 seconds" + ) if self.max_concurrency < 1: raise InvalidResourceError( f"resource '{self.name}': max_concurrency must be >= 1, " diff --git a/runpod/rp_cli/console.py b/runpod/rp_cli/console.py index 3bb10b7b..423e2748 100644 --- a/runpod/rp_cli/console.py +++ b/runpod/rp_cli/console.py @@ -14,11 +14,13 @@ from rich.progress import ( Progress, SpinnerColumn, + Task, TaskID, TextColumn, ) from rich.spinner import SPINNERS from rich.theme import Theme +from rich.text import Text from runpod._hosts import console_url @@ -275,6 +277,13 @@ def _bar(fraction: float) -> str: ) +class _DeploySpinnerColumn(SpinnerColumn): + def render(self, task: Task) -> Text: + if task.fields.get("failed"): + return Text("✗", style="err") + return super().render(task) + + class DeployEvents: """deploy_app event sink: animated phase list with live progress. @@ -285,7 +294,9 @@ class DeployEvents: def __init__(self) -> None: self._progress = Progress( - _spinner_column(), + _DeploySpinnerColumn( + RUNPOD_SAND_SPINNER, style=ACCENT, finished_text="[ok]✓[/ok]" + ), TextColumn("[white]{task.description}[/white]"), TextColumn("{task.fields[detail]}"), console=console, @@ -358,8 +369,13 @@ def endpoint_ready(self, name: str, endpoint_id: str) -> None: self.endpoints: Dict[str, str] = getattr(self, "endpoints", {}) self.endpoints[name] = endpoint_id - def close(self) -> None: - self._finish_current() + def close(self, *, failed: bool = False) -> None: + if failed and self._current is not None: + self._progress.update(self._current, failed=True, detail="[err]failed[/err]") + self._progress.stop_task(self._current) + self._current = None + else: + self._finish_current() if self._started: self._progress.stop() self._started = False diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index eae26a80..d8f1ad1a 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -225,6 +225,7 @@ def deploy(target, env, python_version, exclude, build_only): for found in apps: ui.set_name_width(list(found.resources)) events = ui.DeployEvents() + succeeded = False try: artifact = build_artifact( found, @@ -234,10 +235,11 @@ def deploy(target, env, python_version, exclude, build_only): events=events, output=Path.cwd() / f"{found.name}-artifact.tar.gz", ) + succeeded = True except Exception as exc: # noqa: BLE001 - surface engine errors cleanly raise click.ClickException(str(exc)) from exc finally: - events.close() + events.close(failed=not succeeded) size_mb = artifact.stat().st_size / (1024 * 1024) ui.success( f"built [white]{found.name}[/white] " @@ -268,6 +270,7 @@ def deploy(target, env, python_version, exclude, build_only): source=_app_source(found, project_root), ) events = ui.DeployEvents() + succeeded = False with ui.Timer() as t: try: result = asyncio.run( @@ -280,10 +283,11 @@ def deploy(target, env, python_version, exclude, build_only): events=events, ) ) + succeeded = True except Exception as exc: # noqa: BLE001 - surface engine errors cleanly raise click.ClickException(str(exc)) from exc finally: - events.close() + events.close(failed=not succeeded) ui.success( f"[bold white]{result.app_name}/{env or found.env}[/bold white] " f"is live [dim]{t.elapsed:.1f}s[/dim]" diff --git a/tests/test_apps/test_partial_deploy.py b/tests/test_apps/test_partial_deploy.py new file mode 100644 index 00000000..73cbbe5d --- /dev/null +++ b/tests/test_apps/test_partial_deploy.py @@ -0,0 +1,252 @@ +"""partial deployment diagnostics and preflight validation.""" + +import io +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from click.testing import CliRunner +from rich.console import Console + +from runpod.apps import App +from runpod.apps.app import _clear_registry +from runpod.apps.build import BuildResult +from runpod.apps.deploy import deploy_app, reconcile_endpoints +from runpod.apps.errors import EndpointReconciliationError, InvalidResourceError +from runpod.apps.gpu import CpuInstanceType +from runpod.rp_cli import console as ui +from runpod.rp_cli.main import cli + + +@pytest.fixture(autouse=True) +def clean_registry(): + _clear_registry() + yield + _clear_registry() + + +def make_app(): + app = App("partial") + for name in ("c_later", "b_bad", "a_ok"): + app.queue(name=name, cpu="cpu3c-1-2")(lambda x: x) + return app + + +def environment(endpoints=None): + return {"id": "env-1", "name": "prod", "endpoints": endpoints or []} + + +@pytest.mark.parametrize("updating", [False, True]) +@pytest.mark.parametrize( + "cause", [RuntimeError("quota exceeded"), TimeoutError("timed out")] +) +async def test_reports_confirmed_changes_and_failed_request(updating, cause): + api = AsyncMock() + api.save_endpoint.side_effect = [{"id": "ep-ok"}, cause] + existing = [{"name": "a_ok", "id": "ep-ok"}] if updating else [] + existing.append({"name": "obsolete", "id": "ep-old"}) + + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, make_app(), environment(existing), "build-1", "3.12" + ) + + message = str(caught.value) + assert caught.value.__cause__ is cause + assert "partial/prod (environment env-1, build build-1)" in message + status = ( + "updated endpoints (not rolled back)" + if updating + else "rolled back created endpoints" + ) + assert f"{status}: a_ok (ep-ok)" in message + assert f"failed to create b_bad: {cause}" in message + assert "deleted endpoints (not restored)" not in message + assert "response was lost" in message + assert "keep billing" in message + assert "rp flash env get prod --app partial" in message + assert [call.args[0]["name"] for call in api.save_endpoint.await_args_list] == [ + "a_ok", + "b_bad", + ] + if updating: + api.delete_endpoint.assert_not_awaited() + else: + api.delete_endpoint.assert_awaited_once_with("ep-ok") + + +async def test_first_update_failure_identifies_existing_endpoint(): + api = AsyncMock() + api.save_endpoint.side_effect = RuntimeError("rejected") + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, + make_app(), + environment([{"name": "a_ok", "id": "ep-old"}]), + "build-1", + "3.12", + ) + assert "failed to update a_ok (ep-old): rejected" in str(caught.value) + assert "rolled back created endpoints" not in str(caught.value) + api.delete_endpoint.assert_not_awaited() + assert api.save_endpoint.await_count == 1 + + +async def test_preparation_failure_reports_earlier_changes(): + api = AsyncMock() + api.save_endpoint.return_value = {"id": "ep-ok"} + with patch( + "runpod.apps.volume.attach_endpoint_volumes", + AsyncMock(side_effect=[None, RuntimeError("volume unavailable")]), + ): + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints(api, make_app(), environment(), "build-1", "3.12") + assert "rolled back created endpoints: a_ok (ep-ok)" in str(caught.value) + assert "failed to prepare b_bad: volume unavailable" in str(caught.value) + assert "response was lost" not in str(caught.value) + api.save_endpoint.assert_awaited_once() + api.delete_endpoint.assert_awaited_once_with("ep-ok") + + +async def test_deletion_failure_reports_creates_updates_and_deletes(): + api = AsyncMock() + api.save_endpoint.side_effect = lambda payload: {"id": f"ep-{payload['name']}"} + api.delete_endpoint.side_effect = [True, RuntimeError("delete denied"), True, True] + existing = [ + {"name": name, "id": f"ep-{name}"} for name in ("b_bad", "old1", "old2", "old3") + ] + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, make_app(), environment(existing), "build-1", "3.12" + ) + message = str(caught.value) + assert ( + "rolled back created endpoints: c_later (ep-c_later), a_ok (ep-a_ok)" in message + ) + assert "updated endpoints (not rolled back): b_bad (ep-b_bad)" in message + assert "deleted endpoints (not restored): old1 (ep-old1)" in message + assert "failed to delete old2 (ep-old2): delete denied" in message + assert [call.args for call in api.delete_endpoint.await_args_list] == [ + ("ep-old1",), + ("ep-old2",), + ("ep-c_later",), + ("ep-a_ok",), + ] + + +@pytest.mark.parametrize("cleanup_result", [RuntimeError("cleanup denied"), False]) +async def test_rollback_continues_after_cleanup_failure(cleanup_result): + api = AsyncMock() + cause = RuntimeError("quota exceeded") + api.save_endpoint.side_effect = [{"id": "ep-a"}, {"id": "ep-b"}, cause] + api.delete_endpoint.side_effect = [cleanup_result, True] + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints(api, make_app(), environment(), "build-1", "3.12") + message = str(caught.value) + assert caught.value.__cause__ is cause + assert "failed to create c_later: quota exceeded" in message + assert "rollback failed for b_bad (ep-b)" in message + assert "created endpoints still requiring cleanup: b_bad (ep-b)" in message + assert "rolled back created endpoints: a_ok (ep-a)" in message + assert "keep billing" in message + assert [call.args for call in api.delete_endpoint.await_args_list] == [ + ("ep-b",), + ("ep-a",), + ] + + +async def test_success_keeps_created_endpoints(): + api = AsyncMock() + api.save_endpoint.side_effect = lambda payload: {"id": f"ep-{payload['name']}"} + result = await reconcile_endpoints( + api, make_app(), environment(), "build-1", "3.12" + ) + assert result == {"a_ok": "ep-a_ok", "b_bad": "ep-b_bad", "c_later": "ep-c_later"} + api.delete_endpoint.assert_not_awaited() + + +def test_cli_reports_partial_deploy_and_failed_phase(tmp_path, monkeypatch): + (tmp_path / "main.py").write_text( + 'from runpod import App\napp = App("partial")\n' + '@app.queue(name="a_ok", cpu="cpu3c-1-2")\n' + "def a_ok(x): return x\n" + '@app.queue(name="b_bad", cpu="cpu3c-1-2")\n' + "def b_bad(x): return x\n" + ) + env_dir = tmp_path / "vendor" + env_dir.mkdir() + api = AsyncMock() + api.get_app_by_name.return_value = { + "id": "app-1", + "flashEnvironments": [environment()], + } + api.prepare_artifact_upload.return_value = { + "uploadUrl": "https://upload.invalid", + "objectKey": "key", + } + api.finalize_artifact_upload.return_value = {"id": "build-1"} + api.save_endpoint.side_effect = [{"id": "ep-ok"}, RuntimeError("quota exceeded")] + buffer = io.StringIO() + monkeypatch.setattr(ui, "console", Console(file=buffer, theme=ui.theme, width=120)) + with ( + patch("runpod.apps.deploy.AppsApiClient", return_value=api), + patch( + "runpod.apps.deploy.build_environment", + return_value=BuildResult(env_dir, []), + ), + ): + result = CliRunner().invoke( + cli, ["flash", "deploy", str(tmp_path), "-e", "prod"] + ) + assert result.exit_code == 1, result.output + assert "build build-1 was activated on partial/prod" in result.output + assert "rolled back created endpoints: a_ok (ep-ok)" in result.output + assert "failed to create b_bad: quota exceeded" in result.output + assert "✗ reconciling endpoints" in buffer.getvalue() + assert "✓ reconciling endpoints" not in buffer.getvalue() + assert "is live" not in buffer.getvalue() + api.deploy_build.assert_awaited_once_with("env-1", "build-1") + api.delete_endpoint.assert_awaited_once_with("ep-ok") + + +@pytest.mark.parametrize("kind", ["queue", "api"]) +@pytest.mark.parametrize("timeout", [0, -1, 3601, 5000, 99999, 1.5, True, "60", None]) +def test_idle_timeout_rejected_at_decoration(kind, timeout): + app = App("invalid") + with pytest.raises(InvalidResourceError, match="idle_timeout"): + getattr(app, kind)(name="worker", cpu="cpu3c-1-2", idle_timeout=timeout)( + lambda: None + ) + + +@pytest.mark.parametrize("timeout", [1, 3600]) +@pytest.mark.parametrize("cpu", list(CpuInstanceType)) +def test_accepts_timeout_boundaries_and_supported_cpu_flavors(timeout, cpu): + app = App("valid") + handle = app.queue(name="worker", cpu=cpu, idle_timeout=timeout)(lambda: None) + assert handle.spec.idle_timeout == timeout + assert handle.spec.cpu == [cpu.value] + + +@pytest.mark.parametrize("kind", ["queue", "api", "task"]) +@pytest.mark.parametrize("cpu", ["cpu9z", ["cpu3c-1-2", "cpu9z"]]) +def test_cpu_flavor_rejected_at_decoration(kind, cpu): + app = App("invalid") + with pytest.raises( + InvalidResourceError, match="unsupported cpu instance ids: cpu9z" + ): + getattr(app, kind)(name="worker", cpu=cpu)(lambda: None) + + +@pytest.mark.parametrize("field,value", [("idle_timeout", 5000), ("cpu", ["cpu9z"])]) +async def test_mutated_invalid_spec_fails_before_build_and_remote_calls( + tmp_path, field, value +): + app = make_app() + setattr(app.resources["b_bad"].spec, field, value) + api = AsyncMock() + build = Mock() + with patch("runpod.apps.deploy.build_environment", build): + with pytest.raises(InvalidResourceError): + await deploy_app(app, tmp_path, api=api) + build.assert_not_called() + assert not api.mock_calls From 655bb7c98f1e8277ba1981867d67b6612db2f31a Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:03:21 -0800 Subject: [PATCH 2/5] fix(apps): restore endpoint settings after failed deployments --- runpod/api/queries/apps.py | 21 ++ runpod/apps/api.py | 34 +++ runpod/apps/deploy.py | 190 ++++++++----- runpod/apps/discovery.py | 3 + runpod/apps/spec.py | 7 +- runpod/rp_cli/console.py | 7 +- tests/test_apps/test_deploy_rollback.py | 354 ++++++++++++++++++++++++ tests/test_apps/test_partial_deploy.py | 95 ++++--- 8 files changed, 593 insertions(+), 118 deletions(-) create mode 100644 tests/test_apps/test_deploy_rollback.py diff --git a/runpod/api/queries/apps.py b/runpod/api/queries/apps.py index c5b11ac3..0d8b63cb 100644 --- a/runpod/api/queries/apps.py +++ b/runpod/api/queries/apps.py @@ -9,6 +9,27 @@ } """ +QUERY_ENDPOINT_SNAPSHOT = """ +query EndpointSnapshot($id: String!) { + myself { + endpoint(id: $id) { + id name type instanceIds gpuIds gpuCount locations + workersMin workersMax workersPFBTarget idleTimeout + scalerType scalerValue executionTimeoutMs workerInitTimeoutMs + flashBootType minCudaVersion allowedCudaVersions + modelReferences env { key value } + networkVolumeIds { networkVolumeId } + volumes { volumeId volumeType } + template { + name imageName containerDiskInGb dockerArgs + env { key value } + ports containerRegistryAuthId advancedStart startScript config + } + } + } +} +""" + QUERY_FLASH_APP_BY_NAME = """ query getFlashAppByName($flashAppName: String!) { flashAppByName(flashAppName: $flashAppName) { diff --git a/runpod/apps/api.py b/runpod/apps/api.py index 49dd1d6c..f9d9882d 100644 --- a/runpod/apps/api.py +++ b/runpod/apps/api.py @@ -5,6 +5,7 @@ absent from rest. management verbs for the wider sdk stay in runpod.api.ctl_commands. """ +from copy import deepcopy from typing import Any, Dict, List, Optional, Set import aiohttp @@ -243,6 +244,39 @@ async def _execute( ) return response["data"] + async def snapshot_endpoint(self, endpoint_id: str) -> Dict[str, Any]: + """capture a saveable endpoint configuration in memory for rollback.""" + data = await self._execute( + app_queries.QUERY_ENDPOINT_SNAPSHOT, {"id": endpoint_id}, retry=True + ) + endpoint = (data.get("myself") or {}).get("endpoint") + template = (endpoint or {}).get("template") + if ( + not endpoint + or endpoint.get("id") != endpoint_id + or not template + or not template.get("imageName") + ): + raise RuntimeError( + f"endpoint {endpoint_id} has no restorable configuration" + ) + snapshot = deepcopy(endpoint) + template = snapshot["template"] + snapshot["template"]["containerRegistryAuthId"] = ( + template.get("containerRegistryAuthId") or "" + ) + snapshot["template"]["dockerArgs"] = template.get("dockerArgs") or "" + snapshot["template"]["env"] = template.get("env") or [] + snapshot["env"] = snapshot.get("env") or [] + snapshot["modelReferences"] = snapshot.get("modelReferences") or [] + snapshot["networkVolumeIds"] = snapshot.get("networkVolumeIds") or [] + snapshot["volumes"] = [ + volume + for volume in snapshot.get("volumes") or [] + if volume["volumeType"] == "OBJECT_STORE_VOLUME" + ] + return snapshot + async def save_endpoint(self, endpoint_input: Dict[str, Any]) -> Dict[str, Any]: """save via rest unless the input requires an unsupported capability.""" body = _rest_endpoint_input(endpoint_input) diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py index 66a2b65c..4b3006ea 100644 --- a/runpod/apps/deploy.py +++ b/runpod/apps/deploy.py @@ -11,6 +11,8 @@ from pathlib import Path, PurePosixPath from typing import Any, Dict, List, Optional, Set +import aiohttp + from .api import AppsApiClient from .app import App from .build import ( @@ -402,21 +404,18 @@ async def reconcile_endpoints( python_version: str, *, resolver=None, + activate_build: bool = False, + environment_created: bool = False, + events=None, ) -> Dict[str, str]: - """converge the environment's endpoints to the app's resources. - - resource name -> endpoint id for everything provisioned. existing - endpoints (matched by name) are updated in place; endpoints whose - resources disappeared are deleted. + """snapshot and reconcile endpoints, restoring configurations on failure. - failures trigger best-effort deletion of endpoints created by this call. + obsolete endpoints are removed after the deployment commits. """ for handle in app.resources.values(): handle.spec.validate() existing_endpoints = environment.get("endpoints") or [] existing = {e["name"]: e["id"] for e in existing_endpoints} - # updates send the bound template's current name, so redeploying an - # existing endpoint sends the same template name as before. existing_templates = { e["name"]: (e.get("template") or {}).get("name") for e in existing_endpoints } @@ -433,12 +432,26 @@ async def reconcile_endpoints( resolver = resolver or VolumeResolver(client) endpoints: Dict[str, str] = {} - completed: Dict[str, Dict[str, str]] = { - "created": {}, "updated": {}, "deleted": {} - } + snapshots = {} + undo = [] + activation_attempted = False + creating = False try: + if provisionable.keys() & existing.keys(): + _phase(events, "snapshot") + for name in sorted(provisionable.keys() & existing.keys()): + action = f"read '{name}'" + snapshots[name] = await client.snapshot_endpoint(existing[name]) + snapshots[name]["flashEnvironmentId"] = environment["id"] + if activate_build: + action = "activate the build" + _phase(events, "activate") + activation_attempted = True + await client.deploy_build(environment["id"], build_id) + _phase(events, "endpoints") for name, handle in sorted(provisionable.items()): - operation = "prepare" + action = f"prepare '{name}'" + creating = False members = None if needs_gpu_catalog(handle.spec.gpu): members = await client.gpu_pool_members() @@ -466,62 +479,109 @@ async def reconcile_endpoints( payload["modelReferences"] = [model_reference(handle.spec.model)] if name in existing: payload["id"] = existing[name] - operation = "update" if name in existing else "create" + undo.append((name, existing[name], snapshots[name])) + action = f"deploy '{name}'" + creating = name not in existing result = await client.save_endpoint(payload) endpoints[name] = result["id"] - completed["updated" if name in existing else "created"][name] = result["id"] - log.info("%s endpoint %s (%s)", operation, name, result["id"]) - - for name, endpoint_id in existing.items(): - if name not in provisionable: - operation = "delete" - await client.delete_endpoint(endpoint_id) - completed["deleted"][name] = endpoint_id - log.info("deleted removed endpoint %s (%s)", name, endpoint_id) + if creating: + undo.append((name, result["id"], None)) + creating = False + log.info("saved endpoint %s (%s)", name, result["id"]) + except Exception as exc: - failed_id = f" ({existing[name]})" if name in existing else "" - lines = [ - f"endpoint reconciliation failed for {app.name}/{environment['name']} " - f"(environment {environment['id']}, build {build_id}).", - f"failed to {operation} {name}{failed_id}: {exc}", - ] - rolled_back = {} - cleanup_failures = {} - for resource, endpoint_id in reversed(list(completed["created"].items())): + lines = [f"Couldn't {action}: {str(exc) or type(exc).__name__}"] + rollback_errors = [] + if activation_attempted and not environment_created: try: - if not await client.delete_endpoint(endpoint_id): + previous_build = environment.get("activeBuildId") + if previous_build: + await client.deploy_build(environment["id"], previous_build) + else: + current = await client.get_environment_by_name( + app.name, environment["name"] + ) + if not isinstance(current, dict) or "activeBuildId" not in current: + raise RuntimeError("activation status unknown") + if current["activeBuildId"]: + raise RuntimeError("no previous build to restore") + except Exception as restore_exc: + rollback_errors.append( + f"build: {str(restore_exc) or type(restore_exc).__name__}" + ) + if creating and isinstance(exc, (aiohttp.ClientError, OSError)): + try: + current = await client.get_environment_by_name( + app.name, environment["name"] + ) + matches = [ + endpoint + for endpoint in (current or {}).get("endpoints", []) + if (endpoint.get("template") or {}).get("name") == template_name + and endpoint["id"] not in existing.values() + ] + if len(matches) != 1: + raise RuntimeError("creation status unknown") + undo.append((name, matches[0]["id"], None)) + except Exception: + rollback_errors.append(f"{name}: creation status unknown") + for resource, endpoint_id, snapshot in reversed(undo): + try: + if snapshot is not None: + await client.save_endpoint(snapshot) + elif not await client.delete_endpoint(endpoint_id): raise RuntimeError("endpoint deletion was not confirmed") - rolled_back[resource] = endpoint_id except Exception as cleanup_exc: - cleanup_failures[resource] = endpoint_id - lines.append( - f"rollback failed for {resource} ({endpoint_id}): {cleanup_exc}" + rollback_errors.append( + f"{resource} ({endpoint_id}): {str(cleanup_exc) or type(cleanup_exc).__name__}" ) - for status, changes in ( - ("rolled back created endpoints", rolled_back), - ("created endpoints still requiring cleanup", cleanup_failures), - ("updated endpoints (not rolled back)", completed["updated"]), - ("deleted endpoints (not restored)", completed["deleted"]), - ): - summary = ", ".join( - f"{resource} ({endpoint_id})" for resource, endpoint_id in changes.items() + if environment_created and not rollback_errors: + try: + current = await client.get_environment_by_name( + app.name, environment["name"] + ) + if ( + not isinstance(current, dict) + or "endpoints" not in current + or current["endpoints"] + ): + raise RuntimeError("couldn't confirm the environment is empty") + if current.get("activeBuildId") not in (None, build_id): + raise RuntimeError("environment changed during deployment") + if not await client.delete_environment(environment["id"]): + raise RuntimeError("environment deletion was not confirmed") + except Exception as cleanup_exc: + rollback_errors.append( + f"environment {environment['id']}: {str(cleanup_exc) or type(cleanup_exc).__name__}" + ) + if rollback_errors: + lines.extend( + ["Rollback incomplete:", *(f" {error}" for error in rollback_errors)] ) - if summary: - lines.append(f"{status}: {summary}") - lines.append("Remaining operations were not attempted.") - if operation != "prepare": lines.append( - "The failed request may have taken effect if its response was lost." + f"Check `rp flash env get {shlex.quote(environment['name'])} " + f"--app {shlex.quote(app.name)}`. Remaining workers may still incur charges." ) - lines.append( - f"Inspect this environment with `rp flash env get " - f"{shlex.quote(environment['name'])} --app {shlex.quote(app.name)}` " - "before retrying. " - "Endpoints with active workers can keep billing. " - "Delete unwanted endpoints or scale their minimum workers to 0." - ) + elif undo or activation_attempted: + lines.append("Changes rolled back.") raise EndpointReconciliationError("\n".join(lines)) from exc + obsolete = existing.keys() - provisionable.keys() + if obsolete: + _phase(events, "cleanup") + for name in sorted(obsolete): + endpoint_id = existing[name] + try: + if not await client.delete_endpoint(endpoint_id): + raise RuntimeError("endpoint deletion was not confirmed") + except Exception as exc: + raise EndpointReconciliationError( + f"Deployed, but couldn't remove '{name}' ({endpoint_id}): {exc}\n" + f"Check `rp flash env get {shlex.quote(environment['name'])} " + f"--app {shlex.quote(app.name)}`." + ) from exc + log.info("deleted removed endpoint %s (%s)", name, endpoint_id) + return endpoints @@ -627,6 +687,7 @@ async def deploy_app( environments = {e["name"]: e for e in remote_app.get("flashEnvironments") or []} environment = environments.get(env_name) + environment_created = environment is None if environment is None: environment = await client.create_environment(app_id, env_name) log.info("created environment %s (%s)", env_name, environment["id"]) @@ -652,18 +713,11 @@ async def deploy_app( ) log.info("uploaded build %s", build["id"]) - await client.deploy_build(environment["id"], build["id"]) - log.info("activated build %s on %s/%s", build["id"], app.name, env_name) - - _phase(events, "endpoints") - try: - endpoints = await reconcile_endpoints( - client, app, environment, build["id"], python_version, resolver=resolver - ) - except EndpointReconciliationError as exc: - raise EndpointReconciliationError( - f"build {build['id']} was activated on {app.name}/{env_name}.\n{exc}" - ) from exc + endpoints = await reconcile_endpoints( + client, app, environment, build["id"], python_version, + resolver=resolver, activate_build=True, + environment_created=environment_created, events=events, + ) endpoint_ready = getattr(events, "endpoint_ready", None) if endpoint_ready is not None: for name, endpoint_id in sorted(endpoints.items()): diff --git a/runpod/apps/discovery.py b/runpod/apps/discovery.py index 6211c985..efc5ce85 100644 --- a/runpod/apps/discovery.py +++ b/runpod/apps/discovery.py @@ -125,6 +125,9 @@ def _import_module(path: Path, project: Project) -> None: "or standard-library module." ) except (Exception, SystemExit) as exc: + if isinstance(exc, InvalidResourceError): + relative = path.relative_to(project.source_root) + raise DiscoveryError(f"{relative}: {exc}") from exc hint = "" if isinstance(exc, ModuleNotFoundError) and ( project.source_root != project.import_root diff --git a/runpod/apps/spec.py b/runpod/apps/spec.py index ead24ee7..7ec25bad 100644 --- a/runpod/apps/spec.py +++ b/runpod/apps/spec.py @@ -129,8 +129,8 @@ def normalize_cpu( supported = {flavor.value for flavor in CpuInstanceType} if unsupported := set(cpu) - supported: raise InvalidResourceError( - f"unsupported cpu instance ids: {', '.join(sorted(unsupported))}. " - f"expected one of {', '.join(sorted(supported))}" + f"unknown cpu {', '.join(repr(value) for value in sorted(unsupported))}. " + "Use a CpuInstanceType value." ) return list(cpu) @@ -258,8 +258,7 @@ def validate(self) -> None: or not 1 <= self.idle_timeout <= 3600 ): raise InvalidResourceError( - f"resource '{self.name}': idle_timeout must be an integer " - "between 1 and 3600 seconds" + f"'{self.name}': idle_timeout must be a whole number from 1 to 3600 seconds" ) if self.max_concurrency < 1: raise InvalidResourceError( diff --git a/runpod/rp_cli/console.py b/runpod/rp_cli/console.py index 423e2748..3cb24fb6 100644 --- a/runpod/rp_cli/console.py +++ b/runpod/rp_cli/console.py @@ -260,7 +260,10 @@ def reload_banner(module: str = "") -> None: "vendor": "resolving dependencies", "package": "packaging artifact", "upload": "uploading build", - "endpoints": "reconciling endpoints", + "endpoints": "deploying endpoints", + "snapshot": "saving current settings", + "activate": "activating build", + "cleanup": "removing unused endpoints", } @@ -371,7 +374,7 @@ def endpoint_ready(self, name: str, endpoint_id: str) -> None: def close(self, *, failed: bool = False) -> None: if failed and self._current is not None: - self._progress.update(self._current, failed=True, detail="[err]failed[/err]") + self._progress.update(self._current, failed=True, detail="") self._progress.stop_task(self._current) self._current = None else: diff --git a/tests/test_apps/test_deploy_rollback.py b/tests/test_apps/test_deploy_rollback.py new file mode 100644 index 00000000..77be76c4 --- /dev/null +++ b/tests/test_apps/test_deploy_rollback.py @@ -0,0 +1,354 @@ +"""endpoint snapshots and deployment rollback ordering.""" + +from copy import deepcopy +from unittest.mock import AsyncMock, patch + +import pytest + +from runpod.api.queries.apps import QUERY_ENDPOINT_SNAPSHOT +from runpod.api.mutations.apps import MUTATION_SAVE_ENDPOINT +from runpod.apps import App +from runpod.apps.api import AppsApiClient +from runpod.apps.app import _clear_registry +from runpod.apps.deploy import reconcile_endpoints +from runpod.apps.errors import EndpointReconciliationError + + +@pytest.fixture(autouse=True) +def clean_registry(): + _clear_registry() + yield + _clear_registry() + + +def snapshot(name="worker", endpoint_id="ep-old"): + return { + "id": endpoint_id, + "name": name, + "type": "QB", + "instanceIds": ["cpu3c-1-2"], + "gpuIds": None, + "gpuCount": 1, + "locations": "EU-RO-1", + "workersMin": 1, + "workersMax": 2, + "workersPFBTarget": 0, + "idleTimeout": 120, + "scalerType": "QUEUE_DELAY", + "scalerValue": 5, + "executionTimeoutMs": 5000, + "workerInitTimeoutMs": 10000, + "flashBootType": "OFF", + "minCudaVersion": None, + "allowedCudaVersions": "", + "modelReferences": [], + "env": [{"key": "CUSTOM", "value": "synthetic-private-value"}], + "networkVolumeIds": [{"networkVolumeId": "volume-old"}], + "volumes": [{"volumeId": "volume-old", "volumeType": "NETWORK_VOLUME"}], + "template": { + "name": "old-template", + "imageName": "old-image", + "containerDiskInGb": 10, + "dockerArgs": "old-command", + "env": [{"key": "RUNPOD_BUILD_ID", "value": "old-build"}], + "ports": None, + "containerRegistryAuthId": None, + "advancedStart": False, + "startScript": None, + "config": None, + }, + } + + +async def test_snapshot_is_a_restorable_copy_of_endpoint_and_template(): + original = snapshot() + client = AppsApiClient() + client._execute = AsyncMock(return_value={"myself": {"endpoint": original}}) + saved = await client.snapshot_endpoint("ep-old") + client._execute.assert_awaited_once_with( + QUERY_ENDPOINT_SNAPSHOT, {"id": "ep-old"}, retry=True + ) + assert saved == { + **original, + "template": {**original["template"], "containerRegistryAuthId": ""}, + "volumes": [], + } + saved["template"]["env"][0]["value"] = "modified" + assert original["template"]["env"][0]["value"] == "old-build" + assert original["template"]["containerRegistryAuthId"] is None + + +async def test_snapshot_retains_global_volumes_and_empty_values(): + original = snapshot() + original.update(env=None, modelReferences=None, networkVolumeIds=None) + original["template"].update(env=None, dockerArgs=None) + original["volumes"] = [ + {"volumeId": "global-old", "volumeType": "OBJECT_STORE_VOLUME"} + ] + client = AppsApiClient() + client._execute = AsyncMock(return_value={"myself": {"endpoint": original}}) + saved = await client.snapshot_endpoint("ep-old") + assert saved["volumes"] == original["volumes"] + assert saved["env"] == saved["modelReferences"] == saved["networkVolumeIds"] == [] + assert saved["template"]["env"] == [] + assert saved["template"]["dockerArgs"] == "" + + +async def test_snapshot_can_be_saved_through_the_control_plane_client(): + client = AppsApiClient() + client._execute = AsyncMock( + side_effect=[ + {"myself": {"endpoint": snapshot()}}, + {"saveEndpoint": {"id": "ep-old"}}, + ] + ) + saved = await client.snapshot_endpoint("ep-old") + assert await client.save_endpoint(saved) == {"id": "ep-old"} + client._execute.assert_awaited_with(MUTATION_SAVE_ENDPOINT, {"input": saved}) + + +@pytest.mark.parametrize( + "current", [None, {"activeBuildId": "new-build"}, {"activeBuildId": None}] +) +async def test_first_activation_failure_reports_unrestorable_build_state(current): + app = App("demo") + app.queue(name="worker", cpu="cpu3c-1-2")(lambda: None) + api = AsyncMock() + api.deploy_build.side_effect = TimeoutError("response lost") + api.get_environment_by_name.return_value = current + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, + app, + {"id": "env", "name": "prod"}, + "new-build", + "3.12", + activate_build=True, + ) + if current == {"activeBuildId": None}: + assert "Changes rolled back." in str(caught.value) + else: + assert "Rollback incomplete:\n build:" in str(caught.value) + api.save_endpoint.assert_not_awaited() + + +@pytest.mark.parametrize( + "endpoint", [None, {"id": "other"}, {"id": "ep-old", "template": None}] +) +async def test_snapshot_fails_closed_when_current_state_is_unavailable(endpoint): + client = AppsApiClient() + client._execute = AsyncMock(return_value={"myself": {"endpoint": endpoint}}) + with pytest.raises(RuntimeError, match="no restorable configuration"): + await client.snapshot_endpoint("ep-old") + + +def app_and_environment(): + app = App("demo") + for name in ("alpha", "beta", "gamma"): + app.queue(name=name, cpu="cpu3c-1-2")(lambda x: x) + environment = { + "id": "env", + "name": "prod", + "activeBuildId": "old-build", + "endpoints": [ + {"id": "ep-alpha", "name": "alpha"}, + {"id": "ep-beta", "name": "beta"}, + ], + } + return app, environment + + +async def test_restores_full_state_of_successful_and_failed_updates_in_reverse_order(): + app, environment = app_and_environment() + alpha = snapshot("alpha", "ep-alpha") + beta = snapshot("beta", "ep-beta") + api = AsyncMock() + api.snapshot_endpoint.side_effect = [deepcopy(alpha), deepcopy(beta)] + api.save_endpoint.side_effect = [ + {"id": "ep-alpha"}, + TimeoutError("response lost"), + {"id": "ep-beta"}, + {"id": "ep-alpha"}, + ] + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, app, environment, "new-build", "3.12", activate_build=True + ) + assert ( + str(caught.value) + == "Couldn't deploy 'beta': response lost\nChanges rolled back." + ) + calls = api.save_endpoint.await_args_list + assert calls[2].args[0] == {**beta, "flashEnvironmentId": "env"} + assert calls[3].args[0] == {**alpha, "flashEnvironmentId": "env"} + assert [call[0] for call in api.mock_calls[:3]] == [ + "snapshot_endpoint", + "snapshot_endpoint", + "deploy_build", + ] + assert [call.args for call in api.deploy_build.await_args_list] == [ + ("env", "new-build"), + ("env", "old-build"), + ] + assert api.mock_calls[-3][0] == "deploy_build" + api.delete_endpoint.assert_not_awaited() + assert "synthetic-private-value" not in str(caught.value) + + +async def test_snapshot_failure_stops_before_any_writes(): + app, environment = app_and_environment() + api = AsyncMock() + api.snapshot_endpoint.side_effect = [snapshot(), RuntimeError("access denied")] + with pytest.raises( + EndpointReconciliationError, match="Couldn't read 'beta': access denied" + ): + await reconcile_endpoints( + api, app, environment, "new-build", "3.12", activate_build=True + ) + api.save_endpoint.assert_not_awaited() + api.deploy_build.assert_not_awaited() + api.delete_endpoint.assert_not_awaited() + + +@pytest.mark.parametrize("restore_failure", [False, True]) +async def test_activation_failure_restores_endpoints_and_previous_build( + restore_failure, +): + app, environment = app_and_environment() + api = AsyncMock() + api.snapshot_endpoint.side_effect = [ + snapshot("alpha", "ep-alpha"), + snapshot("beta", "ep-beta"), + ] + api.save_endpoint.side_effect = lambda payload: { + "id": payload.get("id", "ep-gamma") + } + api.deploy_build.side_effect = [ + TimeoutError("activation timed out"), + RuntimeError("build restore denied") if restore_failure else {"id": "env"}, + ] + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, app, environment, "new-build", "3.12", activate_build=True + ) + assert [call.args for call in api.deploy_build.await_args_list] == [ + ("env", "new-build"), + ("env", "old-build"), + ] + api.delete_endpoint.assert_not_awaited() + api.save_endpoint.assert_not_awaited() + if restore_failure: + assert "Rollback incomplete:\n build: build restore denied" in str( + caught.value + ) + else: + assert ( + str(caught.value) + == "Couldn't activate the build: activation timed out\nChanges rolled back." + ) + + +async def test_restore_failure_does_not_skip_other_endpoints(): + app, environment = app_and_environment() + api = AsyncMock() + api.snapshot_endpoint.side_effect = [ + snapshot("alpha", "ep-alpha"), + snapshot("beta", "ep-beta"), + ] + api.save_endpoint.side_effect = [ + {"id": "ep-alpha"}, + RuntimeError("update failed"), + RuntimeError("restore denied"), + {"id": "ep-alpha"}, + ] + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints(api, app, environment, "new-build", "3.12") + assert "Rollback incomplete:\n beta (ep-beta): restore denied" in str(caught.value) + assert api.save_endpoint.await_args.args[0]["id"] == "ep-alpha" + + +async def test_lost_create_response_is_recovered_by_unique_template_name(): + app = App("demo") + app.queue(name="worker", cpu="cpu3c-1-2")(lambda: None) + environment = {"id": "env", "name": "prod", "endpoints": []} + api = AsyncMock() + api.save_endpoint.side_effect = TimeoutError("response lost") + api.get_environment_by_name.return_value = { + "endpoints": [ + { + "id": "ep-created", + "name": "worker", + "template": {"name": "demo-worker-template-token"}, + }, + { + "id": "ep-other", + "name": "worker", + "template": {"name": "another-deploy"}, + }, + ] + } + with patch("runpod.apps.deploy.secrets.token_hex", return_value="token"): + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints(api, app, environment, "new-build", "3.12") + api.delete_endpoint.assert_awaited_once_with("ep-created") + assert ( + str(caught.value) + == "Couldn't deploy 'worker': response lost\nChanges rolled back." + ) + + +async def test_activation_precedes_irreversible_cleanup(): + app, environment = app_and_environment() + environment["endpoints"].append({"name": "obsolete", "id": "ep-obsolete"}) + api = AsyncMock() + api.save_endpoint.side_effect = lambda payload: { + "id": payload.get("id", "ep-gamma") + } + await reconcile_endpoints( + api, app, environment, "new-build", "3.12", activate_build=True + ) + calls = [call[0] for call in api.mock_calls] + assert calls.index("deploy_build") < min( + i for i, name in enumerate(calls) if name == "save_endpoint" + ) + assert calls.index("delete_endpoint") > max( + i for i, name in enumerate(calls) if name == "save_endpoint" + ) + + +@pytest.mark.parametrize("cleanup_fails", [False, True]) +async def test_new_environment_is_removed_after_endpoint_rollback(cleanup_fails): + app = App("demo") + app.queue(name="alpha", cpu="cpu3c-1-2")(lambda: None) + app.queue(name="beta", cpu="cpu3c-1-2")(lambda: None) + environment = {"id": "env", "name": "prod", "endpoints": []} + api = AsyncMock() + api.save_endpoint.side_effect = [{"id": "ep-alpha"}, RuntimeError("quota exceeded")] + api.get_environment_by_name.return_value = { + "id": "env", + "endpoints": [], + "activeBuildId": "new-build", + } + if cleanup_fails: + api.delete_endpoint.side_effect = RuntimeError("delete denied") + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, + app, + environment, + "new-build", + "3.12", + activate_build=True, + environment_created=True, + ) + if cleanup_fails: + api.delete_environment.assert_not_awaited() + assert "Rollback incomplete:" in str(caught.value) + else: + api.delete_environment.assert_awaited_once_with("env") + calls = [call[0] for call in api.mock_calls] + assert calls.index("delete_endpoint") < calls.index("delete_environment") + assert ( + str(caught.value) + == "Couldn't deploy 'beta': quota exceeded\nChanges rolled back." + ) diff --git a/tests/test_apps/test_partial_deploy.py b/tests/test_apps/test_partial_deploy.py index 73cbbe5d..9c466560 100644 --- a/tests/test_apps/test_partial_deploy.py +++ b/tests/test_apps/test_partial_deploy.py @@ -32,7 +32,12 @@ def make_app(): def environment(endpoints=None): - return {"id": "env-1", "name": "prod", "endpoints": endpoints or []} + return { + "id": "env-1", + "name": "prod", + "endpoints": endpoints or [], + "activeBuildId": "build-old", + } @pytest.mark.parametrize("updating", [False, True]) @@ -41,7 +46,10 @@ def environment(endpoints=None): ) async def test_reports_confirmed_changes_and_failed_request(updating, cause): api = AsyncMock() - api.save_endpoint.side_effect = [{"id": "ep-ok"}, cause] + snapshot = {"id": "ep-ok", "name": "a_ok", "workersMin": 2} + api.snapshot_endpoint.return_value = snapshot.copy() + api.get_environment_by_name.return_value = environment() + api.save_endpoint.side_effect = [{"id": "ep-ok"}, cause, {"id": "ep-ok"}] existing = [{"name": "a_ok", "id": "ep-ok"}] if updating else [] existing.append({"name": "obsolete", "id": "ep-old"}) @@ -52,31 +60,33 @@ async def test_reports_confirmed_changes_and_failed_request(updating, cause): message = str(caught.value) assert caught.value.__cause__ is cause - assert "partial/prod (environment env-1, build build-1)" in message - status = ( - "updated endpoints (not rolled back)" - if updating - else "rolled back created endpoints" - ) - assert f"{status}: a_ok (ep-ok)" in message - assert f"failed to create b_bad: {cause}" in message - assert "deleted endpoints (not restored)" not in message - assert "response was lost" in message - assert "keep billing" in message - assert "rp flash env get prod --app partial" in message - assert [call.args[0]["name"] for call in api.save_endpoint.await_args_list] == [ + assert f"Couldn't deploy 'b_bad': {cause}" in message + if isinstance(cause, TimeoutError): + assert "Rollback incomplete:\n b_bad: creation status unknown" in message + assert "rp flash env get prod --app partial" in message + else: + assert ( + message == "Couldn't deploy 'b_bad': quota exceeded\nChanges rolled back." + ) + assert [call.args[0]["name"] for call in api.save_endpoint.await_args_list[:2]] == [ "a_ok", "b_bad", ] if updating: api.delete_endpoint.assert_not_awaited() + assert api.save_endpoint.await_args.args[0] == { + **snapshot, + "flashEnvironmentId": "env-1", + } else: api.delete_endpoint.assert_awaited_once_with("ep-ok") async def test_first_update_failure_identifies_existing_endpoint(): api = AsyncMock() - api.save_endpoint.side_effect = RuntimeError("rejected") + snapshot = {"id": "ep-old", "name": "a_ok", "workersMin": 2} + api.snapshot_endpoint.return_value = snapshot + api.save_endpoint.side_effect = [RuntimeError("rejected"), {"id": "ep-old"}] with pytest.raises(EndpointReconciliationError) as caught: await reconcile_endpoints( api, @@ -85,10 +95,10 @@ async def test_first_update_failure_identifies_existing_endpoint(): "build-1", "3.12", ) - assert "failed to update a_ok (ep-old): rejected" in str(caught.value) - assert "rolled back created endpoints" not in str(caught.value) + assert str(caught.value) == "Couldn't deploy 'a_ok': rejected\nChanges rolled back." api.delete_endpoint.assert_not_awaited() - assert api.save_endpoint.await_count == 1 + assert api.save_endpoint.await_count == 2 + assert api.save_endpoint.await_args.args[0] == snapshot async def test_preparation_failure_reports_earlier_changes(): @@ -100,9 +110,10 @@ async def test_preparation_failure_reports_earlier_changes(): ): with pytest.raises(EndpointReconciliationError) as caught: await reconcile_endpoints(api, make_app(), environment(), "build-1", "3.12") - assert "rolled back created endpoints: a_ok (ep-ok)" in str(caught.value) - assert "failed to prepare b_bad: volume unavailable" in str(caught.value) - assert "response was lost" not in str(caught.value) + assert ( + str(caught.value) + == "Couldn't prepare 'b_bad': volume unavailable\nChanges rolled back." + ) api.save_endpoint.assert_awaited_once() api.delete_endpoint.assert_awaited_once_with("ep-ok") @@ -119,17 +130,12 @@ async def test_deletion_failure_reports_creates_updates_and_deletes(): api, make_app(), environment(existing), "build-1", "3.12" ) message = str(caught.value) - assert ( - "rolled back created endpoints: c_later (ep-c_later), a_ok (ep-a_ok)" in message - ) - assert "updated endpoints (not rolled back): b_bad (ep-b_bad)" in message - assert "deleted endpoints (not restored): old1 (ep-old1)" in message - assert "failed to delete old2 (ep-old2): delete denied" in message + assert "Deployed, but couldn't remove 'old2' (ep-old2): delete denied" in message + assert "rolled back" not in message + assert api.save_endpoint.await_count == 3 assert [call.args for call in api.delete_endpoint.await_args_list] == [ ("ep-old1",), ("ep-old2",), - ("ep-c_later",), - ("ep-a_ok",), ] @@ -143,11 +149,10 @@ async def test_rollback_continues_after_cleanup_failure(cleanup_result): await reconcile_endpoints(api, make_app(), environment(), "build-1", "3.12") message = str(caught.value) assert caught.value.__cause__ is cause - assert "failed to create c_later: quota exceeded" in message - assert "rollback failed for b_bad (ep-b)" in message - assert "created endpoints still requiring cleanup: b_bad (ep-b)" in message - assert "rolled back created endpoints: a_ok (ep-a)" in message - assert "keep billing" in message + assert "Couldn't deploy 'c_later': quota exceeded" in message + assert "Rollback incomplete:\n b_bad (ep-b):" in message + assert "a_ok" not in message + assert "may still incur charges" in message assert [call.args for call in api.delete_endpoint.await_args_list] == [ ("ep-b",), ("ep-a",), @@ -198,13 +203,17 @@ def test_cli_reports_partial_deploy_and_failed_phase(tmp_path, monkeypatch): cli, ["flash", "deploy", str(tmp_path), "-e", "prod"] ) assert result.exit_code == 1, result.output - assert "build build-1 was activated on partial/prod" in result.output - assert "rolled back created endpoints: a_ok (ep-ok)" in result.output - assert "failed to create b_bad: quota exceeded" in result.output - assert "✗ reconciling endpoints" in buffer.getvalue() - assert "✓ reconciling endpoints" not in buffer.getvalue() + assert ( + result.output + == "Error: Couldn't deploy 'b_bad': quota exceeded\nChanges rolled back.\n" + ) + assert "✗ deploying endpoints" in buffer.getvalue() + assert "✓ deploying endpoints" not in buffer.getvalue() assert "is live" not in buffer.getvalue() - api.deploy_build.assert_awaited_once_with("env-1", "build-1") + assert [call.args for call in api.deploy_build.await_args_list] == [ + ("env-1", "build-1"), + ("env-1", "build-old"), + ] api.delete_endpoint.assert_awaited_once_with("ep-ok") @@ -231,9 +240,7 @@ def test_accepts_timeout_boundaries_and_supported_cpu_flavors(timeout, cpu): @pytest.mark.parametrize("cpu", ["cpu9z", ["cpu3c-1-2", "cpu9z"]]) def test_cpu_flavor_rejected_at_decoration(kind, cpu): app = App("invalid") - with pytest.raises( - InvalidResourceError, match="unsupported cpu instance ids: cpu9z" - ): + with pytest.raises(InvalidResourceError, match="unknown cpu 'cpu9z'"): getattr(app, kind)(name="worker", cpu=cpu)(lambda: None) From e80f886ffa641e3c1f55c02ebd75f962477e45e5 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:07:15 -0800 Subject: [PATCH 3/5] fix(cli): keep routine source exclusions in debug logs --- runpod/apps/source.py | 4 ++-- tests/test_apps/test_deploy.py | 9 +++++++++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/runpod/apps/source.py b/runpod/apps/source.py index 2130198b..e35bc80a 100644 --- a/runpod/apps/source.py +++ b/runpod/apps/source.py @@ -131,7 +131,7 @@ def iter_source_files( ) ): if not discovery: - log.warning("excluded source path %r from deployment artifact", rel) + log.debug("excluded source path %r from deployment artifact", rel) continue kept_dirs.append(name) dirs[:] = kept_dirs @@ -147,6 +147,6 @@ def iter_source_files( or (discovery and (path.suffix != ".py" or name == "setup.py")) ): if not discovery: - log.warning("excluded source path %r from deployment artifact", rel) + log.debug("excluded source path %r from deployment artifact", rel) continue yield path diff --git a/tests/test_apps/test_deploy.py b/tests/test_apps/test_deploy.py index 50d9187e..824e756f 100644 --- a/tests/test_apps/test_deploy.py +++ b/tests/test_apps/test_deploy.py @@ -196,6 +196,7 @@ def test_credential_like_source_paths_are_included(self, tmp_path): assert f"src/{name}" in tar.getnames() def test_excluded_paths_are_logged_without_contents(self, tmp_path, caplog): + caplog.set_level("DEBUG", logger="runpod.apps.source") for name in (".env", "private.key", "private.pem", "local.txt", ".aws/config"): path = tmp_path / name path.parent.mkdir(parents=True, exist_ok=True) @@ -210,6 +211,14 @@ def test_excluded_paths_are_logged_without_contents(self, tmp_path, caplog): ) assert "sensitive-file-contents" not in caplog.text + def test_excluded_paths_are_quiet_at_normal_log_levels(self, tmp_path, caplog): + (tmp_path / "__pycache__").mkdir() + (tmp_path / ".env").write_text("sensitive-file-contents") + with tarfile.open(package_project(tmp_path, {})) as tar: + assert set(tar.getnames()) == {"runpod_manifest.json"} + assert "excluded source path" not in caplog.text + assert "sensitive-file-contents" not in caplog.text + def test_vendored_env_included_under_env(self, tmp_path): _write_project(tmp_path) env_dir = tmp_path / "built-env" From 81d8f87da907933e5a17ed21bd6d35ddfea919d5 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:15:46 -0800 Subject: [PATCH 4/5] fix(apps): initialize deployment failure context --- runpod/apps/deploy.py | 1 + 1 file changed, 1 insertion(+) diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py index 4b3006ea..d116cf38 100644 --- a/runpod/apps/deploy.py +++ b/runpod/apps/deploy.py @@ -436,6 +436,7 @@ async def reconcile_endpoints( undo = [] activation_attempted = False creating = False + action = "prepare deployment" try: if provisionable.keys() & existing.keys(): _phase(events, "snapshot") From a666da854d4e19c9ff6003df09f3ecba3245a853 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Fri, 13 Nov 2026 16:41:00 -0800 Subject: [PATCH 5/5] fix(apps): preserve CPU compatibility and guard deployment rollback --- runpod/apps/deploy.py | 30 ++++- runpod/apps/spec.py | 15 +-- tests/test_apps/test_deploy_event_failure.py | 45 +++++++ .../test_apps/test_deploy_review_feedback.py | 113 ++++++++++++++++++ tests/test_apps/test_deploy_rollback.py | 2 + tests/test_apps/test_partial_deploy.py | 1 + 6 files changed, 196 insertions(+), 10 deletions(-) create mode 100644 tests/test_apps/test_deploy_event_failure.py create mode 100644 tests/test_apps/test_deploy_review_feedback.py diff --git a/runpod/apps/deploy.py b/runpod/apps/deploy.py index d116cf38..9f94d3ce 100644 --- a/runpod/apps/deploy.py +++ b/runpod/apps/deploy.py @@ -1,5 +1,6 @@ """deploy pipeline: discovered apps -> manifest -> artifact -> activated build.""" +import asyncio import json import logging import os @@ -493,7 +494,24 @@ async def reconcile_endpoints( except Exception as exc: lines = [f"Couldn't {action}: {str(exc) or type(exc).__name__}"] rollback_errors = [] - if activation_attempted and not environment_created: + rollback_allowed = True + if activation_attempted: + try: + current = await client.get_environment_by_name( + app.name, environment["name"] + ) + if not isinstance(current, dict) or "activeBuildId" not in current: + raise RuntimeError("activation status unknown") + if current["activeBuildId"] != build_id and ( + undo or current["activeBuildId"] != environment.get("activeBuildId") + ): + raise RuntimeError("environment changed during deployment") + except Exception as ownership_exc: + rollback_errors.append( + f"build: {str(ownership_exc) or type(ownership_exc).__name__}" + ) + rollback_allowed = False + if rollback_allowed and activation_attempted and not environment_created: try: previous_build = environment.get("activeBuildId") if previous_build: @@ -510,7 +528,11 @@ async def reconcile_endpoints( rollback_errors.append( f"build: {str(restore_exc) or type(restore_exc).__name__}" ) - if creating and isinstance(exc, (aiohttp.ClientError, OSError)): + if ( + rollback_allowed + and creating + and isinstance(exc, (aiohttp.ClientError, OSError, asyncio.TimeoutError)) + ): try: current = await client.get_environment_by_name( app.name, environment["name"] @@ -526,7 +548,9 @@ async def reconcile_endpoints( undo.append((name, matches[0]["id"], None)) except Exception: rollback_errors.append(f"{name}: creation status unknown") - for resource, endpoint_id, snapshot in reversed(undo): + for resource, endpoint_id, snapshot in ( + reversed(undo) if rollback_allowed else [] + ): try: if snapshot is not None: await client.save_endpoint(snapshot) diff --git a/runpod/apps/spec.py b/runpod/apps/spec.py index 7ec25bad..26d03391 100644 --- a/runpod/apps/spec.py +++ b/runpod/apps/spec.py @@ -13,7 +13,7 @@ from .errors import InvalidResourceError from .datacenter import DataCenter, datacenter_id -from .gpu import CpuInstanceType, GpuGroup, GpuLike, GpuType +from .gpu import GpuGroup, GpuLike, GpuType from .volume import Volume, normalize_mounts, validate_mounts log = logging.getLogger(__name__) @@ -126,12 +126,13 @@ def normalize_cpu( ) if not cpu or any(not isinstance(c, str) or not c.strip() for c in cpu): raise InvalidResourceError("cpu must contain non-empty instance ids") - supported = {flavor.value for flavor in CpuInstanceType} - if unsupported := set(cpu) - supported: - raise InvalidResourceError( - f"unknown cpu {', '.join(repr(value) for value in sorted(unsupported))}. " - "Use a CpuInstanceType value." - ) + from ..api.ctl_commands import _cpu_config + + for instance_id in cpu: + try: + _cpu_config(instance_id) + except ValueError as exc: + raise InvalidResourceError(f"unknown cpu {instance_id!r}. {exc}") from exc return list(cpu) diff --git a/tests/test_apps/test_deploy_event_failure.py b/tests/test_apps/test_deploy_event_failure.py new file mode 100644 index 00000000..2b3e379c --- /dev/null +++ b/tests/test_apps/test_deploy_event_failure.py @@ -0,0 +1,45 @@ +"""reconciliation errors before endpoint operations begin.""" + +from unittest.mock import AsyncMock + +import pytest + +from runpod.apps import App +from runpod.apps.app import _clear_registry +from runpod.apps.deploy import reconcile_endpoints +from runpod.apps.errors import EndpointReconciliationError + + +@pytest.fixture(autouse=True) +def clean_registry(): + _clear_registry() + yield + _clear_registry() + + +@pytest.mark.parametrize("existing", [False, True]) +async def test_phase_lookup_failure_preserves_original_error(existing): + error = RuntimeError("phase handler unavailable") + + class Events: + @property + def phase(self): + raise error + + app = App("demo") + app.queue(name="worker", cpu="cpu3c-1-2")(lambda: None) + api = AsyncMock() + environment = { + "id": "env", + "name": "prod", + "endpoints": [{"id": "ep-old", "name": "worker"}] if existing else [], + } + + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, app, environment, "new-build", "3.12", events=Events() + ) + + assert str(caught.value) == "Couldn't prepare deployment: phase handler unavailable" + assert caught.value.__cause__ is error + assert api.mock_calls == [] diff --git a/tests/test_apps/test_deploy_review_feedback.py b/tests/test_apps/test_deploy_review_feedback.py new file mode 100644 index 00000000..9bc48119 --- /dev/null +++ b/tests/test_apps/test_deploy_review_feedback.py @@ -0,0 +1,113 @@ +"""cpu compatibility and deployment rollback safety.""" + +import asyncio +from unittest.mock import AsyncMock, patch + +import pytest + +from runpod.apps import App +from runpod.apps.app import _clear_registry +from runpod.apps.deploy import reconcile_endpoints +from runpod.apps.errors import EndpointReconciliationError, InvalidResourceError + + +@pytest.fixture(autouse=True) +def clean_registry(): + _clear_registry() + yield + _clear_registry() + + +@pytest.mark.parametrize("kind", ["queue", "api", "task"]) +@pytest.mark.parametrize("cpu", ["cpu5g-4-16", "cpu3c-16-32", "cpu9z-32-64"]) +def test_cpu_instance_shapes_are_not_limited_to_enum_members(kind, cpu): + app = App("demo") + handle = getattr(app, kind)(name="worker", cpu=cpu)(lambda: None) + assert handle.spec.cpu == [cpu] + + +@pytest.mark.parametrize( + "cpu", + ["cpu5g", "cpu5g-0-16", "cpu5g-4-0", "cpu5g-4", "cpu5g-4-16-extra", " cpu5g-4-16"], +) +def test_malformed_cpu_instance_ids_fail_before_deployment(cpu): + app = App("demo") + with pytest.raises(InvalidResourceError, match="unknown cpu"): + app.queue(name="worker", cpu=cpu)(lambda: None) + + +async def test_asyncio_timeout_recovers_created_endpoint(): + app = App("demo") + app.queue(name="worker", cpu="cpu3c-1-2")(lambda: None) + api = AsyncMock() + error = asyncio.TimeoutError("response lost") + api.save_endpoint.side_effect = error + api.get_environment_by_name.return_value = { + "endpoints": [ + { + "id": "ep-created", + "name": "worker", + "template": {"name": "demo-worker-template-token"}, + } + ] + } + with patch("runpod.apps.deploy.secrets.token_hex", return_value="token"): + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, app, {"id": "env", "name": "prod"}, "new-build", "3.12" + ) + assert caught.value.__cause__ is error + assert ( + str(caught.value) + == "Couldn't deploy 'worker': response lost\nChanges rolled back." + ) + api.delete_endpoint.assert_awaited_once_with("ep-created") + + +@pytest.mark.parametrize("environment_created", [False, True]) +@pytest.mark.parametrize("current", [{"activeBuildId": "other-build"}, {}, None]) +async def test_unconfirmed_ownership_prevents_rollback_writes( + current, environment_created +): + app = App("demo") + for name in ("alpha", "beta", "gamma"): + app.queue(name=name, cpu="cpu3c-1-2")(lambda: None) + api = AsyncMock() + api.snapshot_endpoint.return_value = {"id": "ep-alpha", "name": "alpha"} + api.save_endpoint.side_effect = [ + {"id": "ep-alpha"}, + {"id": "ep-beta"}, + RuntimeError("quota exceeded"), + ] + api.get_environment_by_name.return_value = current + environment = { + "id": "env", + "name": "prod", + "activeBuildId": None if environment_created else "old-build", + "endpoints": ( + [] if environment_created else [{"id": "ep-alpha", "name": "alpha"}] + ), + } + with pytest.raises(EndpointReconciliationError) as caught: + await reconcile_endpoints( + api, + app, + environment, + "new-build", + "3.12", + activate_build=True, + environment_created=environment_created, + ) + message = str(caught.value) + assert "Rollback incomplete:" in message + assert "Changes rolled back." not in message + assert "rp flash env get prod --app demo" in message + assert "may still incur charges" in message + if current: + assert "environment changed during deployment" in message + else: + assert "activation status unknown" in message + api.deploy_build.assert_awaited_once_with("env", "new-build") + assert api.save_endpoint.await_count == 3 + api.delete_endpoint.assert_not_awaited() + api.delete_environment.assert_not_awaited() diff --git a/tests/test_apps/test_deploy_rollback.py b/tests/test_apps/test_deploy_rollback.py index 77be76c4..b8a9fcbb 100644 --- a/tests/test_apps/test_deploy_rollback.py +++ b/tests/test_apps/test_deploy_rollback.py @@ -163,6 +163,7 @@ async def test_restores_full_state_of_successful_and_failed_updates_in_reverse_o alpha = snapshot("alpha", "ep-alpha") beta = snapshot("beta", "ep-beta") api = AsyncMock() + api.get_environment_by_name.return_value = {"activeBuildId": "new-build"} api.snapshot_endpoint.side_effect = [deepcopy(alpha), deepcopy(beta)] api.save_endpoint.side_effect = [ {"id": "ep-alpha"}, @@ -216,6 +217,7 @@ async def test_activation_failure_restores_endpoints_and_previous_build( ): app, environment = app_and_environment() api = AsyncMock() + api.get_environment_by_name.return_value = {"activeBuildId": "new-build"} api.snapshot_endpoint.side_effect = [ snapshot("alpha", "ep-alpha"), snapshot("beta", "ep-beta"), diff --git a/tests/test_apps/test_partial_deploy.py b/tests/test_apps/test_partial_deploy.py index 9c466560..bc891637 100644 --- a/tests/test_apps/test_partial_deploy.py +++ b/tests/test_apps/test_partial_deploy.py @@ -189,6 +189,7 @@ def test_cli_reports_partial_deploy_and_failed_phase(tmp_path, monkeypatch): "objectKey": "key", } api.finalize_artifact_upload.return_value = {"id": "build-1"} + api.get_environment_by_name.return_value = {"activeBuildId": "build-1"} api.save_endpoint.side_effect = [{"id": "ep-ok"}, RuntimeError("quota exceeded")] buffer = io.StringIO() monkeypatch.setattr(ui, "console", Console(file=buffer, theme=ui.theme, width=120))