diff --git a/packages/reflex-build-sdk/README.md b/packages/reflex-build-sdk/README.md index 5825f9a4f3c..0fb657f7757 100644 --- a/packages/reflex-build-sdk/README.md +++ b/packages/reflex-build-sdk/README.md @@ -245,3 +245,7 @@ client = ReflexBuild( To use another HTTP library, implement the `Transport` or `AsyncTransport` protocol. +To wrap the synchronous SDK's default transport without choosing an HTTP library, +call `reflex_build_sdk.transports.default_transport()`. It selects httpx2 when +installed, otherwise httpx. Close the returned transport when finished; clients +leave explicitly supplied transports open. diff --git a/packages/reflex-build-sdk/news/+default-transport-factory.feature.md b/packages/reflex-build-sdk/news/+default-transport-factory.feature.md new file mode 100644 index 00000000000..ac028c35e12 --- /dev/null +++ b/packages/reflex-build-sdk/news/+default-transport-factory.feature.md @@ -0,0 +1 @@ +Expose `reflex_build_sdk.transports.default_transport()` so transport wrappers can preserve the SDK's choice of HTTP library. diff --git a/packages/reflex-build-sdk/src/reflex_build_sdk/transports/__init__.py b/packages/reflex-build-sdk/src/reflex_build_sdk/transports/__init__.py index bd526e4bf3c..62804036ccd 100644 --- a/packages/reflex-build-sdk/src/reflex_build_sdk/transports/__init__.py +++ b/packages/reflex-build-sdk/src/reflex_build_sdk/transports/__init__.py @@ -20,6 +20,7 @@ Transport, TransportError, ) +from reflex_build_sdk.transports._defaults import default_transport if TYPE_CHECKING: from reflex_build_sdk.transports._httpx import AsyncHttpxTransport, HttpxTransport @@ -47,6 +48,7 @@ "Response", "Transport", "TransportError", + "default_transport", ] diff --git a/packages/reflex-hosting-cli/README.md b/packages/reflex-hosting-cli/README.md index e55afd26d28..f654fd8a65a 100644 --- a/packages/reflex-hosting-cli/README.md +++ b/packages/reflex-hosting-cli/README.md @@ -1,6 +1,20 @@ # reflex-hosting-cli Hosting CLI for Reflex. +When an app is still scaling, `reflex deploy` waits 15 seconds between confirmed +scaling refusals, for up to 7 retries to update instance bounds and 11 retries +to submit the deployment. The CLI reports each wait, and you can cancel +with Ctrl-C. Submission retries reuse the uploaded build. The SDK's usual safe +retries can add two more submission attempts, for up to 14 attempts in total; +they do not restart the scaling wait budget. The scaling sleeps total at most +165 seconds for submission; requests and SDK backoff add to elapsed time. +Other failures, including a lost submission response, are reported without +replaying an uncertain write. +If an instance-bounds write has an uncertain outcome, including a server error, +the CLI warns that the bounds may have changed. +A success response that cannot be decoded also stops deployment with a separate +warning about bounds left changed. + If an access token is expired or rejected, run `reflex login` to authenticate again. If you supply a token explicitly, replace it with a valid token or remove the override before using the saved login: unset `REFLEX_ACCESS_TOKEN` or omit diff --git a/packages/reflex-hosting-cli/news/+retry-scaling-conflicts.bugfix.md b/packages/reflex-hosting-cli/news/+retry-scaling-conflicts.bugfix.md new file mode 100644 index 00000000000..479cfbaa946 --- /dev/null +++ b/packages/reflex-hosting-cli/news/+retry-scaling-conflicts.bugfix.md @@ -0,0 +1 @@ +`reflex deploy` now waits and retries confirmed scaling conflicts with progress messages, reusing the uploaded build and preserving support for custom backend URL prefixes. It stops on unrelated failures and uncertain writes, and warns when bounds may remain changed after a failed deployment, including server errors and unreadable success responses. diff --git a/packages/reflex-hosting-cli/pyproject.toml b/packages/reflex-hosting-cli/pyproject.toml index e86d4343ad0..7dba1d7d3d6 100644 --- a/packages/reflex-hosting-cli/pyproject.toml +++ b/packages/reflex-hosting-cli/pyproject.toml @@ -31,11 +31,10 @@ dependencies = [ # The SDK ships no HTTP library of its own; [httpx] points it at the one # `check_version` already pulls in, so the extra costs nothing to install. # The SDK is pre-1.0, so every release may break its API: cap at the patch - # and raise all three bounds when adopting a new one. The floor admits the - # workspace's dev build until the 0.1.0 alpha publishes and materialization - # lifts it; the cap admits 0.1.0 and the ``0.1.1.devN`` builds that follow - # its tag, but not 0.1.1 itself. - "reflex-build-sdk[httpx] >=0.1.0,<0.1.1.post,!=0.1.1", + # and raise all three bounds when adopting a new one. The default transport + # factory requires the next SDK release; materialization lifts this dev + # floor when it publishes. The cap also admits the following dev builds. + "reflex-build-sdk[httpx] >=0.1.1.dev0,<0.1.2.post,!=0.1.2", "rich >=13,<16", ] diff --git a/packages/reflex-hosting-cli/src/reflex_cli/utils/deploy.py b/packages/reflex-hosting-cli/src/reflex_cli/utils/deploy.py new file mode 100644 index 00000000000..7f655423613 --- /dev/null +++ b/packages/reflex-hosting-cli/src/reflex_cli/utils/deploy.py @@ -0,0 +1,196 @@ +"""Wait for confirmed scaling refusals without repeating build uploads.""" + +from __future__ import annotations + +import logging +import time +from collections.abc import Callable +from http import HTTPStatus +from typing import TypeVar + +from reflex_build_sdk import APIStatusError, ConflictError +from reflex_build_sdk.transports import Request, Response, Transport + +logger = logging.getLogger(__name__) + +_T = TypeVar("_T") +_SCALING_RETRY_DELAY = 15 +_DEPLOYMENT_SCALING_CODE = "app_scaling" +_LEGACY_DEPLOYMENT_SCALING_DETAIL = ( + "the app is currently being scaled; wait for the scale to finish, then deploy again" +) + + +def _is_scaling_conflict(error: APIStatusError, *, url: str, code: str) -> bool: + """Identify a refusal that guarantees a write was blocked by scaling. + + Args: + error: The SDK's typed refusal. + url: The endpoint's full URL on the configured backend. + code: The expected scaling refusal code. + + Returns: + Whether this request was explicitly refused before applying its write. + """ + if ( + error.status_code != HTTPStatus.CONFLICT + or error.request.method != "POST" + or error.request.url != url + ): + return False + # Older servers use app_busy for scaling, stopping, and another deployment. + # Only the exact scaling detail guarantees the submission can be retried. + return error.code == code or ( + code == _DEPLOYMENT_SCALING_CODE + and error.code == "app_busy" + and error.detail == _LEGACY_DEPLOYMENT_SCALING_DETAIL + ) + + +def _retry_scaling_conflicts( + operation: Callable[[], _T], *, url: str, code: str, action: str, attempts: int +) -> _T: + """Retry only writes the server explicitly refused because of scaling. + + Args: + operation: The SDK operation to attempt. + url: The endpoint's full URL on the configured backend. + code: The expected scaling refusal code. + action: The action described in progress messages. + attempts: The maximum number of calls, including the initial attempt. + + Returns: + The successful operation's result. + + Raises: + APIStatusError: If scaling persists or the refusal is unrelated. + """ + return _ScalingRetryBudget(attempts=attempts).run( + operation, url=url, code=code, action=action + ) + + +class _ScalingRetryBudget: + """Keep scaling waits bounded across the SDK's own safe request retries.""" + + def __init__(self, *, attempts: int) -> None: + """Create the scaling retry budget. + + Args: + attempts: The initial attempt plus the permitted scaling retries. + """ + self._retries = attempts - 1 + self._used = 0 + + def run( + self, operation: Callable[[], _T], *, url: str, code: str, action: str + ) -> _T: + """Retry a refused operation within the remaining scaling wait budget. + + Args: + operation: The operation to attempt. + url: The endpoint's full URL on the configured backend. + code: The expected scaling refusal code. + action: The action described in progress messages. + + Returns: + The successful operation's result. + + Raises: + APIStatusError: If scaling persists or the refusal is unrelated. + """ + while True: + try: + return operation() + except APIStatusError as ex: + if self._used >= self._retries or not _is_scaling_conflict( + ex, url=url, code=code + ): + raise + self._used += 1 + logger.info( + f"App is being scaled; waiting {_SCALING_RETRY_DELAY} seconds before " + f"retrying {action} (scaling retry {self._used}/{self._retries})." + ) + time.sleep(_SCALING_RETRY_DELAY) + + +class _DeploymentRetryTransport: + """Retry one deployment's refused submission using its uploaded archives. + + The CLI creates a fresh upload client and transport for each deployment. + Its scaling budget is shared across the SDK's safe retries of that submission. + """ + + def __init__(self, transport: Transport, *, url: str) -> None: + """Wrap the transport used by a deployment's upload client. + + Args: + transport: The underlying transport, closed by ``close``. + url: The exact control-plane deployment submission URL. + """ + self._transport = transport + self._url = url + self._budget = _ScalingRetryBudget(attempts=12) + + def send(self, request: Request) -> Response: + """Send a request, waiting only when a deployment is refused for scaling. + + Args: + request: The SDK's request, including its serialized body. + + Returns: + The response for the SDK to decode. + + Raises: + ConflictError: If scaling persists after eleven scaling waits. + """ + if ( + request.method != "POST" + or request.url != self._url + or not isinstance(request.content, bytes) + ): + return self._transport.send(request) + # Submit bodies are immutable bytes. Replaying this request keeps the + # stored_build_id and avoids reserving and uploading the archives again. + return self._budget.run( + lambda: self._send_submission(request), + url=self._url, + code=_DEPLOYMENT_SCALING_CODE, + action="deployment", + ) + + def _send_submission(self, request: Request) -> Response: + """Turn only a confirmed scaling response into a retryable SDK error. + + Args: + request: The replayable submission request. + + Returns: + Any response other than a confirmed scaling refusal. + + Raises: + ConflictError: If the server refused this submission for scaling. + """ + response = self._transport.send(request) + if response.status_code != HTTPStatus.CONFLICT: + return response + try: + body = response.json() + except ValueError: + return response + if not isinstance(body, dict): + return response + detail = body.get("detail") + error = ConflictError( + f"409 {response.reason_phrase}: {detail}", + response=response, + detail=detail, + ) + if _is_scaling_conflict(error, url=self._url, code=_DEPLOYMENT_SCALING_CODE): + raise error + return response + + def close(self) -> None: + """Release the wrapped transport's connections.""" + self._transport.close() diff --git a/packages/reflex-hosting-cli/src/reflex_cli/utils/hosting.py b/packages/reflex-hosting-cli/src/reflex_cli/utils/hosting.py index 08cb1c65869..4e045ce048d 100644 --- a/packages/reflex-hosting-cli/src/reflex_cli/utils/hosting.py +++ b/packages/reflex-hosting-cli/src/reflex_cli/utils/hosting.py @@ -32,6 +32,7 @@ from reflex_build_sdk import ( APIConnectionError, APIError, + APIResponseValidationError, APIStatusError, AuthenticationError, DeploymentFailedError, @@ -44,12 +45,18 @@ ) from reflex_build_sdk._decode import json_key from reflex_build_sdk._deploy import status_message_outcome +from reflex_build_sdk.transports import default_transport from reflex_build_sdk.types import DeploymentReport, LoginRequest, Me import reflex_cli.constants as constants from reflex_cli.core.config import Config, RegionOption from reflex_cli.utils import console, log from reflex_cli.utils.dependency import is_valid_url +from reflex_cli.utils.deploy import ( + _DeploymentRetryTransport, + _is_scaling_conflict, + _retry_scaling_conflicts, +) from reflex_cli.utils.exceptions import ( ResponseError, ScaleAppError, @@ -814,7 +821,8 @@ def identity_as_dict(me: Me) -> dict[str, Any]: } -def upload_client(client: AuthenticatedClient) -> ReflexBuild: +@contextlib.contextmanager +def upload_client(client: AuthenticatedClient) -> Iterator[ReflexBuild]: """Build a client whose timeouts suit pushing a build's archives. The SDK's defaults are sized for API calls. An archive is not one: it is @@ -824,15 +832,25 @@ def upload_client(client: AuthenticatedClient) -> ReflexBuild: Args: client: The authenticated client the deploy is running under. - Returns: - A client to submit the deployment with. The caller closes it. + Yields: + A client that retries confirmed scaling refusals without reuploading. """ - return ReflexBuild( - token=client.token, - base_url=constants.Hosting.HOSTING_SERVICE, - timeout=UPLOAD_IO_TIMEOUT.total_seconds(), - ) + base_url = constants.Hosting.HOSTING_SERVICE.rstrip("/") + with ( + contextlib.closing( + _DeploymentRetryTransport( + default_transport(), url=f"{base_url}/api/v1/deployments" + ) + ) as transport, + ReflexBuild( + token=client.token, + base_url=base_url, + timeout=UPLOAD_IO_TIMEOUT.total_seconds(), + transport=transport, + ) as uploader, + ): + yield uploader def validate_token(token: str) -> dict[str, Any]: @@ -1236,11 +1254,27 @@ def find_gcp_connection( return None +def _warn_unconfirmed_instance_bounds(app_name: str) -> None: + """Warn when a bounds write may have committed without acknowledgement. + + Args: + app_name: The app whose instance bounds were being updated. + """ + logger.warning( + f"Could not confirm the instance bounds of " + f"'{app_name}'; they may or may not have been applied. " + "Check the app in the Reflex Cloud dashboard before relying " + "on its scaling." + ) + + def set_instance_bounds( app_id: str, client: AuthenticatedClient, min_instances: int | None = None, max_instances: int | None = None, + *, + retry_scaling: bool = False, ) -> str | None: """Set the autoscaling instance bounds on an app. @@ -1254,23 +1288,74 @@ def set_instance_bounds( client: The authenticated client. min_instances: The minimum number of instances to keep running. max_instances: The maximum number of instances to scale out to. + retry_scaling: Whether to retry confirmed scaling refusals, up to eight + attempts with a 15-second wait between attempts. Returns: None on success, or a ``"set instance bounds failed: ..."`` string on - error (validation, unsupported platform, or a scale already running). + a non-retryable refusal (validation or unsupported platform). + + Raises: + APIStatusError: If scaling refused the write or a server error left its + outcome unknown. + APIConnectionError: If contact is lost while reading or updating bounds. + APIResponseValidationError: If the response could not be decoded. """ - try: + bounds_url = f"{client.api.base_url}/api/v1/apps/{app_id}/instance_bounds" + scaling_code = "instance_bounds_scale_conflict" + + def update_bounds() -> None: + """Update the requested bounds, preserving the latest unspecified bound.""" current = client.api.apps.get(app_id) - client.api.apps.set_instance_bounds( - app_id, - min_instances=current.min_instances - if min_instances is None - else min_instances, - max_instances=current.max_instances - if max_instances is None - else max_instances, - ) + try: + client.api.apps.set_instance_bounds( + app_id, + min_instances=current.min_instances + if min_instances is None + else min_instances, + max_instances=current.max_instances + if max_instances is None + else max_instances, + ) + except MissingTokenError: + raise + except APIResponseValidationError: + logger.warning( + f"The server reported success setting the instance bounds of " + f"'{current.name}', but its response could not be decoded. " + "The bounds may have changed even though this deploy stopped. " + "Check the app in the Reflex Cloud dashboard before relying " + "on its scaling." + ) + raise + except APIStatusError as ex: + if ex.status_code >= 500: + _warn_unconfirmed_instance_bounds(current.name) + raise + except BaseException: + _warn_unconfirmed_instance_bounds(current.name) + raise + + try: + if retry_scaling: + _retry_scaling_conflicts( + update_bounds, + url=bounds_url, + code=scaling_code, + action="the instance bounds update", + attempts=8, + ) + else: + update_bounds() + except (APIConnectionError, APIResponseValidationError): + raise + except APIStatusError as ex: + if ex.status_code >= 500 or _is_scaling_conflict( + ex, url=bounds_url, code=scaling_code + ): + raise + return f"set instance bounds failed: {error_message(ex)}" except ReflexBuildError as ex: return f"set instance bounds failed: {error_message(ex)}" return None diff --git a/packages/reflex-hosting-cli/src/reflex_cli/v2/cli.py b/packages/reflex-hosting-cli/src/reflex_cli/v2/cli.py index 7a93f880e29..f181f2d4576 100644 --- a/packages/reflex-hosting-cli/src/reflex_cli/v2/cli.py +++ b/packages/reflex-hosting-cli/src/reflex_cli/v2/cli.py @@ -1109,18 +1109,12 @@ def deploy( min_instances=min_instances, max_instances=max_instances, client=authenticated_client, + retry_scaling=True, ) - except BaseException: - # A dropped connection says nothing about whether the server - # applied the write, and the bounds are billable state, so hedge - # rather than report either outcome as fact. - logger.warning( - f"Lost contact while setting the instance bounds of " - f"'{app.name}'; they may or may not have been applied. " - "Check the app in the Reflex Cloud dashboard before relying " - "on its scaling." + except ReflexBuildError as ex: + hosting.exit_reporting( + ex, f"set instance bounds failed: {hosting.error_message(ex)}" ) - raise if bounds_error: logger.error(bounds_error) raise click.exceptions.Exit(1) diff --git a/tests/units/reflex_build_sdk/transports/test_defaults.py b/tests/units/reflex_build_sdk/transports/test_defaults.py index 6d52bcee0e1..c696914117a 100644 --- a/tests/units/reflex_build_sdk/transports/test_defaults.py +++ b/tests/units/reflex_build_sdk/transports/test_defaults.py @@ -14,11 +14,9 @@ HttpxTransport, Transport, _defaults, -) -from reflex_build_sdk.transports._defaults import ( - async_default_transport, default_transport, ) +from reflex_build_sdk.transports._defaults import async_default_transport def _without(monkeypatch: pytest.MonkeyPatch, *missing: str) -> None: diff --git a/tests/units/reflex_cli/sdk.py b/tests/units/reflex_cli/sdk.py index e2f07582f8f..51a51f308dd 100644 --- a/tests/units/reflex_cli/sdk.py +++ b/tests/units/reflex_cli/sdk.py @@ -46,7 +46,9 @@ def fake_client(**identity: Any) -> FakeClient: "tier": "Pro", } fields.update(identity) - return FakeClient(api=MagicMock(), me=Me(**fields)) + return FakeClient( + api=MagicMock(base_url="https://build.reflex.dev"), me=Me(**fields) + ) def patch_upload_client(mocker: Any, client: FakeClient) -> None: @@ -64,25 +66,37 @@ def patch_upload_client(mocker: Any, client: FakeClient) -> None: mocker.patch("reflex_cli.utils.hosting.upload_client", return_value=uploader) -def api_error(status_code: int, detail: str) -> APIStatusError: +def api_error( + status_code: int, + detail: str, + *, + code: str = "", + method: str = "GET", + path: str = "test", + base_url: str = "https://build.reflex.dev", +) -> APIStatusError: """Build the error the SDK raises for a refused request. Args: status_code: The status the API answered with. detail: The API's explanation, which the CLI reports. + code: The machine-readable refusal code. + method: The failed request's method. + path: The failed request's path below ``/api/v1/``. + base_url: The configured backend URL, including any path prefix. Returns: The error, to raise from a mocked call. """ request = Request( - method="GET", - url="https://build.reflex.dev/api/v1/test", + method=method, + url=f"{base_url.rstrip('/')}/api/v1/{path}", headers={"X-Request-ID": uuid.uuid4().hex}, ) response = Response( status_code=status_code, reason_phrase="", - headers={}, + headers={"x-reflex-error-code": code} if code else {}, content=json.dumps({"detail": detail}).encode(), request=request, ) diff --git a/tests/units/reflex_cli/utils/test_deploy.py b/tests/units/reflex_cli/utils/test_deploy.py new file mode 100644 index 00000000000..db6b1677c59 --- /dev/null +++ b/tests/units/reflex_cli/utils/test_deploy.py @@ -0,0 +1,383 @@ +"""Retries of confirmed deployment scaling refusals preserve uploaded builds.""" + +from __future__ import annotations + +import json +import logging +import uuid +from dataclasses import replace +from pathlib import Path +from typing import Literal +from unittest.mock import MagicMock, call +from urllib.parse import parse_qs + +import pytest +from reflex_build_sdk import APIConnectionError, ConflictError, ReflexBuild +from reflex_build_sdk.transports import Request, Response, TransportError +from reflex_cli.utils.deploy import _DeploymentRetryTransport, _retry_scaling_conflicts + +from tests.units.reflex_cli.sdk import api_error + +_URL = "https://build.reflex.dev/api/v1/deployments" +_DETAIL = ( + "the app is currently being scaled; wait for the scale to finish, then deploy again" +) + + +def _request() -> Request: + """Build a replayable deployment submission. + + Returns: + The request with its stored build id and trace id. + """ + return Request( + method="POST", + url=_URL, + headers={"X-Request-ID": "trace-id"}, + content=b"stored_build_id=original-build&app_id=app-1", + ) + + +def _response( + request: Request, + status: int = 409, + *, + detail: object = _DETAIL, + code: str = "app_busy", +) -> Response: + """Build an API response tied to its request. + + Args: + request: The request being answered. + status: The response status. + detail: The refusal's body detail. + code: The refusal's condition code. + + Returns: + The response. + """ + return Response( + request=request, + status_code=status, + reason_phrase="Conflict" if status == 409 else "", + headers={"x-reflex-error-code": code}, + content=json.dumps({"detail": detail}).encode(), + ) + + +@pytest.mark.parametrize( + ("code", "detail"), + [ + ("app_busy", _DETAIL), + ("app_scaling", "The scale is finishing."), + ("app_scaling", None), + ], +) +def test_deployment_retry_preserves_request(mocker, code: str, detail: object): + """A confirmed refusal repeats the same stored build and trace id.""" + request = _request() + success = _response(request, 200) + transport = MagicMock() + transport.send.side_effect = [ + _response(request, code=code, detail=detail), + success, + ] + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + + assert _DeploymentRetryTransport(transport, url=_URL).send(request) is success + + assert transport.send.call_args_list == [call(request), call(request)] + sleep.assert_called_once_with(15) + + +def test_deployment_retry_does_not_upload_archives_again(mocker, tmp_path: Path): + """The real SDK reserves and uploads once while its submit waits for scaling.""" + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + deployment_id = uuid.UUID(int=31) + requests: list[Request] = [] + submissions: list[Request] = [] + + def send(request: Request) -> Response: + requests.append(request) + if request.url == f"{_URL}/reserve": + body = { + "deployment_id": str(deployment_id), + "backend": {"url": "https://storage.test/backend", "headers": {}}, + "frontend": {"url": "https://storage.test/frontend", "headers": {}}, + } + elif request.method == "PUT": + body = None + else: + assert request.url == _URL + submissions.append(request) + if len(submissions) == 1: + return _response(request) + body = str(deployment_id) + return replace(_response(request, 200), content=json.dumps(body).encode()) + + transport = MagicMock() + transport.send.side_effect = send + backend, frontend = tmp_path / "backend.zip", tmp_path / "frontend.zip" + backend.write_bytes(b"backend") + frontend.write_bytes(b"frontend") + with ReflexBuild( + token="test-token", + transport=_DeploymentRetryTransport(transport, url=_URL), + max_retries=0, + ) as client: + result = client.deployments.create( + uuid.UUID(int=2), backend=backend, frontend=frontend + ) + + assert result == deployment_id + assert sum(request.url == f"{_URL}/reserve" for request in requests) == 1 + assert sum(request.method == "PUT" for request in requests) == 2 + assert len(submissions) == 2 + assert ( + submissions[0].headers["X-Request-ID"] == submissions[1].headers["X-Request-ID"] + ) + assert submissions[0].content == submissions[1].content + assert isinstance(submissions[0].content, bytes) + assert parse_qs(submissions[0].content.decode())["stored_build_id"] == [ + str(deployment_id) + ] + sleep.assert_called_once_with(15) + + +def test_deployment_retry_exhaustion_preserves_final_error(mocker): + """Twelve refused submissions sleep eleven times and surface the last refusal.""" + request = _request() + response = _response(request) + transport = MagicMock() + transport.send.return_value = response + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + + with pytest.raises(ConflictError) as exc: + _DeploymentRetryTransport(transport, url=_URL).send(request) + + assert exc.value.response is response + assert exc.value.detail == _DETAIL + assert transport.send.call_count == 12 + assert sleep.call_args_list == [call(15)] * 11 + + +@pytest.mark.parametrize("safe_failure", [408, 429, "unsent"]) +def test_deployment_retry_budget_survives_sdk_retries( + mocker, caplog, safe_failure: int | Literal["unsent"] +): + """Safe SDK retries cannot restart the submission's scaling wait budget.""" + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + caplog.set_level(logging.INFO, logger="reflex_cli.utils.deploy") + transport = MagicMock() + submissions: list[Request] = [] + + def send(request: Request) -> Response: + """Interleave safe SDK failures with confirmed scaling refusals. + + Args: + request: The attempted submission. + + Returns: + A safe refusal or scaling conflict. + + Raises: + TransportError: For the known-unsent failure case. + """ + submissions.append(request) + if len(submissions) in (6, 12): + if safe_failure == "unsent": + message = "not connected" + raise TransportError(message, request=request, sent=False) + return _response(request, safe_failure) + return _response(request) + + transport.send.side_effect = send + with ( + ReflexBuild( + token="test-token", transport=_DeploymentRetryTransport(transport, url=_URL) + ) as client, + pytest.raises(ConflictError), + ): + client._request("POST", "deployments", str, form={"stored_build_id": "build"}) + + assert len(submissions) == 14 + assert len({request.headers["X-Request-ID"] for request in submissions}) == 1 + assert all(request.content == submissions[0].content for request in submissions) + assert sleep.call_args_list.count(call(15)) == 11 + assert len(sleep.call_args_list) == 13 + assert caplog.messages[-1].endswith("(scaling retry 11/11).") + assert len(caplog.messages) == 11 + + +def test_deployment_retry_sdk_stops_after_uncertain_submission(mocker): + """A lost response after a scaling retry never grants another SDK attempt.""" + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + transport = MagicMock() + + def send(request: Request) -> Response: + """Lose the response after one explicit scaling refusal. + + Args: + request: The attempted submission. + + Returns: + The first scaling refusal. + + Raises: + TransportError: When the second submission's outcome is unknown. + """ + if transport.send.call_count == 1: + return _response(request) + message = "connection lost" + raise TransportError(message, request=request, sent=True) + + transport.send.side_effect = send + with ( + ReflexBuild( + token="test-token", transport=_DeploymentRetryTransport(transport, url=_URL) + ) as client, + pytest.raises(APIConnectionError), + ): + client._request("POST", "deployments", str, form={"stored_build_id": "build"}) + + assert transport.send.call_count == 2 + sleep.assert_called_once_with(15) + + +@pytest.mark.parametrize( + ("status", "code", "detail"), + [ + (400, "app_busy", _DETAIL), + (401, "app_busy", _DETAIL), + (403, "app_busy", _DETAIL), + (500, "app_busy", _DETAIL), + (400, "app_scaling", "The scale is finishing."), + (500, "app_scaling", "The scale is finishing."), + (409, "", _DETAIL), + (409, "unclassified", _DETAIL), + (409, "other_conflict", _DETAIL), + (409, "app_busy", "the app is currently being stopped"), + (409, "app_busy", "a deployment is already in progress for this app"), + (409, "app_busy", {"message": _DETAIL}), + ], +) +def test_deployment_retry_leaves_unrelated_responses_to_sdk( + mocker, status: int, code: str, detail: object +): + """Only the exact confirmed scaling refusal grants replay.""" + request = _request() + response = _response(request, status, code=code, detail=detail) + transport = MagicMock() + transport.send.return_value = response + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + + assert _DeploymentRetryTransport(transport, url=_URL).send(request) is response + + transport.send.assert_called_once_with(request) + sleep.assert_not_called() + + +@pytest.mark.parametrize( + "sdk_request", + [ + replace(_request(), url=f"{_URL}/reserve"), + replace(_request(), url="https://storage.test/api/v1/deployments"), + replace(_request(), url="https://build.reflex.dev/proxy/api/v1/deployments"), + replace(_request(), url=f"{_URL}?upload=true"), + replace(_request(), method="PUT"), + replace(_request(), content=iter([b"archive"])), + replace(_request(), content=None), + ], +) +def test_deployment_retry_does_not_replay_other_requests(mocker, sdk_request: Request): + """Reserve, storage, and streamed requests stay outside the retry boundary.""" + response = _response(sdk_request) + transport = MagicMock() + transport.send.return_value = response + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + + assert _DeploymentRetryTransport(transport, url=_URL).send(sdk_request) is response + + transport.send.assert_called_once_with(sdk_request) + sleep.assert_not_called() + + +@pytest.mark.parametrize("body", [b"not json", b"[]", b"null", b'"scaling"']) +def test_deployment_retry_leaves_malformed_refusal_to_sdk(mocker, body: bytes): + """An unrecognized refusal body cannot prove the submission was rejected.""" + request = _request() + response = replace(_response(request), content=body) + transport = MagicMock() + transport.send.return_value = response + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + + assert _DeploymentRetryTransport(transport, url=_URL).send(request) is response + + transport.send.assert_called_once_with(request) + sleep.assert_not_called() + + +def test_deployment_retry_propagates_uncertain_transport_error(mocker): + """A lost response could hide an accepted submit and must never be replayed.""" + request = _request() + error = TransportError("connection lost", request=request, sent=True) + transport = MagicMock() + transport.send.side_effect = error + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + + with pytest.raises(TransportError) as exc: + _DeploymentRetryTransport(transport, url=_URL).send(request) + + assert exc.value is error + transport.send.assert_called_once_with(request) + sleep.assert_not_called() + + +def test_deployment_retry_interrupt_stops_before_another_submit(mocker): + """Interrupting a scaling wait does not send another submission.""" + request = _request() + transport = MagicMock() + transport.send.return_value = _response(request) + mocker.patch("reflex_cli.utils.deploy.time.sleep", side_effect=KeyboardInterrupt) + + with pytest.raises(KeyboardInterrupt): + _DeploymentRetryTransport(transport, url=_URL).send(request) + + transport.send.assert_called_once_with(request) + + +def test_deployment_retry_close_delegates(): + """The caller can release the transport the decorator owns.""" + transport = MagicMock() + + _DeploymentRetryTransport(transport, url=_URL).close() + + transport.close.assert_called_once_with() + + +def test_bounds_retry_uses_typed_scaling_refusal(mocker): + """A bounds retry accepts its dedicated code without matching prose.""" + path = "apps/app-1/instance_bounds" + refusal = api_error( + 409, + "The scale is finishing.", + code="instance_bounds_scale_conflict", + method="POST", + path=path, + ) + operation = MagicMock(side_effect=[refusal, "updated"]) + sleep = mocker.patch("reflex_cli.utils.deploy.time.sleep") + + assert ( + _retry_scaling_conflicts( + operation, + url=f"https://build.reflex.dev/api/v1/{path}", + code="instance_bounds_scale_conflict", + action="instance bounds", + attempts=8, + ) + == "updated" + ) + + assert operation.call_count == 2 + sleep.assert_called_once_with(15) diff --git a/tests/units/reflex_cli/utils/test_hosting.py b/tests/units/reflex_cli/utils/test_hosting.py index db6f56ef099..f5596b5fb57 100644 --- a/tests/units/reflex_cli/utils/test_hosting.py +++ b/tests/units/reflex_cli/utils/test_hosting.py @@ -4,14 +4,22 @@ import json import logging import uuid +from pathlib import Path from unittest.mock import MagicMock, mock_open +from urllib.parse import parse_qs, urlsplit import click import pytest from pytest_mock import MockerFixture, MockFixture from reflex_base.utils.log import SUCCESS +from reflex_build_sdk import ( + APIConnectionError, + APIResponseValidationError, + APITimeoutError, +) from reflex_build_sdk.types import DeploymentReport, GcpConnection, GcpStatus from reflex_cli import constants +from reflex_cli.utils import hosting from reflex_cli.utils.exceptions import TokenAccessDeniedError, TokenValidationError from reflex_cli.utils.hosting import ( _WATCH_UNREACHABLE_GRACE, @@ -46,11 +54,117 @@ watch_deployment_status, ) +from tests.units.reflex_build_sdk.conftest import MockAPI, MockTransport, reply from tests.units.reflex_cli.sdk import api_error, fake_client _client = fake_client +def test_upload_client_uses_sdk_default_transport(mocker: MockerFixture): + """The upload client's retry wrapper preserves SDK transport selection. + + Args: + mocker: The pytest-mock fixture. + """ + transport = MagicMock() + default_transport = mocker.patch( + "reflex_cli.utils.hosting.default_transport", + return_value=transport, + create=True, + ) + + with hosting.upload_client(_client()): + default_transport.assert_called_once_with() + transport.close.assert_not_called() + + transport.close.assert_called_once_with() + + +@pytest.mark.parametrize("base_path", ["", "/proxy", "/nested/proxy/"]) +@pytest.mark.parametrize("refusal_code", ["app_busy", "app_scaling"]) +def test_upload_client_retries_submission_without_uploading_again( + mocker: MockerFixture, tmp_path: Path, base_path: str, refusal_code: str +): + """Scaling retries reuse the uploaded build and close their HTTP transport. + + Args: + mocker: The pytest-mock fixture. + tmp_path: The directory for the build archives. + base_path: The backend URL's optional path prefix. + refusal_code: The server's legacy or dedicated scaling refusal code. + """ + mocker.patch.object( + constants.Hosting, "HOSTING_SERVICE", f"https://build.example{base_path}" + ) + deployments_path = f"{base_path.rstrip('/')}/api/v1/deployments" + mock_api = MockAPI() + transport = MockTransport(mock_api) + mocker.patch("reflex_cli.utils.hosting.default_transport", return_value=transport) + sleep = mocker.patch("time.sleep") + client = _client() + client.api.token = "test-token" + deployment_id = str(uuid.UUID(int=41)) + archives = [tmp_path / name for name in ("backend.zip", "frontend.zip")] + for archive in archives: + archive.write_bytes(b"build") + mock_api.add("PUT", f"/{archive.name}", reply(200)) + mock_api.add( + "POST", + f"{deployments_path}/reserve", + reply( + 200, + json={ + "deployment_id": deployment_id, + "backend": { + "url": "https://storage.example/backend.zip", + "headers": {}, + }, + "frontend": { + "url": "https://storage.example/frontend.zip", + "headers": {}, + }, + "expires_in": 1800, + }, + ), + ) + mock_api.add( + "POST", + deployments_path, + reply( + 409, + headers={"x-reflex-error-code": refusal_code}, + json={ + "detail": ( + "the app is currently being scaled; wait for the scale to finish, then deploy again" + if refusal_code == "app_busy" + else "Scaling is underway. Please retry later." + ) + }, + ), + reply(201, json=deployment_id), + ) + + with hosting.upload_client(client) as uploader: + result = uploader.deployments.create( + uuid.UUID(int=51), backend=archives[0], frontend=archives[1] + ) + + assert str(result) == deployment_id + paths = [urlsplit(request.url).path for request in mock_api.requests] + assert paths.count(f"{deployments_path}/reserve") == 1 + assert paths.count("/backend.zip") == paths.count("/frontend.zip") == 1 + submits = mock_api.requests[-2:] + assert paths[-2:] == [deployments_path] * 2 + assert [request.method for request in submits] == ["POST"] * 2 + assert submits[0].url == submits[1].url + assert submits[0].headers["X-Request-ID"] == submits[1].headers["X-Request-ID"] + assert submits[0].content == submits[1].content + assert isinstance(submits[0].content, bytes) + assert parse_qs(submits[0].content.decode())["stored_build_id"] == [deployment_id] + sleep.assert_called_once_with(15) + assert mock_api.closed + + @pytest.mark.parametrize( "config_content, expected_token", [ @@ -827,6 +941,122 @@ def test_set_instance_bounds_error(status_code: int, detail: str): assert detail in result +def test_set_instance_bounds_preserves_scaling_refusal(): + """The deploy retry receives the typed refusal, including its request id.""" + client = _client() + error = api_error( + 409, + "the app is being scaled; change instance bounds after it finishes", + code="instance_bounds_scale_conflict", + method="POST", + path="apps/app-1/instance_bounds", + ) + client.api.apps.set_instance_bounds.side_effect = error + + with pytest.raises(type(error)) as raised: + set_instance_bounds("app-1", client, min_instances=2) + + assert raised.value is error + client.api.apps.set_instance_bounds.assert_called_once() + + +@pytest.mark.parametrize("status_code", [500, 502, 503, 504]) +def test_set_instance_bounds_preserves_server_errors( + status_code: int, caplog: pytest.LogCaptureFixture +): + """A server error cannot confirm whether the bounds write took effect. + + Args: + status_code: The server error returned after the write. + caplog: The captured log messages. + """ + client = _client() + error = api_error( + status_code, + "unavailable", + method="POST", + path="apps/app-1/instance_bounds", + ) + client.api.apps.set_instance_bounds.side_effect = error + + with pytest.raises(type(error)) as raised: + set_instance_bounds("app-1", client, min_instances=2) + + assert raised.value is error + client.api.apps.set_instance_bounds.assert_called_once() + assert any( + record.levelno == logging.WARNING + and "may or may not have been applied" in record.getMessage() + for record in caplog.records + ) + + +@pytest.mark.parametrize("error_type", [APIConnectionError, APITimeoutError]) +def test_set_instance_bounds_preserves_uncertain_outcomes( + error_type: type[APIConnectionError], +): + """A write without a usable answer must reach the deploy's uncertainty guard. + + Args: + error_type: The SDK failure that leaves the write's result unknown. + """ + client = _client() + response = api_error(200, "invalid response").response + error = error_type("lost response", request=response.request) + client.api.apps.set_instance_bounds.side_effect = error + + with pytest.raises(error_type) as raised: + set_instance_bounds("app-1", client, min_instances=2) + + assert raised.value is error + client.api.apps.set_instance_bounds.assert_called_once() + + +@pytest.mark.parametrize("during_write", [False, True]) +def test_set_instance_bounds_warns_about_undecodable_success( + during_write: bool, caplog: pytest.LogCaptureFixture +): + """An undecodable write response still carries the server's success status. + + Args: + during_write: Whether decoding failed after the write or preliminary read. + caplog: The captured log messages. + """ + client = _client() + client.api.apps.get.return_value.name = "my-app" + response = api_error( + 200, + "invalid response", + method="POST" if during_write else "GET", + path="apps/app-1/instance_bounds" if during_write else "apps/app-1", + ).response + error = APIResponseValidationError("invalid response", response=response) + operation = ( + client.api.apps.set_instance_bounds if during_write else client.api.apps.get + ) + operation.side_effect = error + + with pytest.raises(APIResponseValidationError) as raised: + set_instance_bounds("app-1", client, min_instances=2) + + assert raised.value is error + assert client.api.apps.set_instance_bounds.call_count == int(during_write) + warnings = [ + record.getMessage() + for record in caplog.records + if record.levelno == logging.WARNING + ] + if during_write: + assert len(warnings) == 1 + assert "server reported success" in warnings[0] + assert "response could not be decoded" in warnings[0] + assert "bounds may have changed" in warnings[0] + assert "my-app" in warnings[0] + assert "may or may not have been applied" not in warnings[0] + else: + assert not warnings + + def test_validate_token_names_the_product_it_logs_in_through( mocker: MockerFixture, ): diff --git a/tests/units/reflex_cli/v2/test_cli.py b/tests/units/reflex_cli/v2/test_cli.py index 1c3a45f4338..1d839ebe20c 100644 --- a/tests/units/reflex_cli/v2/test_cli.py +++ b/tests/units/reflex_cli/v2/test_cli.py @@ -1,5 +1,6 @@ from __future__ import annotations +import dataclasses import importlib.metadata import logging import os @@ -15,6 +16,12 @@ from reflex_base.config import Config, get_config from reflex_base.registry import RegistrationContext from reflex_base.utils.log import SUCCESS +from reflex_build_sdk import ( + APIConnectionError, + APIResponseValidationError, + APIStatusError, + APITimeoutError, +) from reflex_build_sdk.types import ( App, AppSummary, @@ -37,6 +44,9 @@ _RESERVATION = HostnameReservation( frontend_url="fake-hostname", backend_url="fake-server" ) +_SCALING_DETAIL = ( + "the app is currently being scaled; wait for the scale to finish, then deploy again" +) def _authed(mocker: MockFixture): @@ -1163,6 +1173,617 @@ def test_deploy_forwards_vmtype_to_create_deployment( assert recorder.create_deployment.call_args.kwargs["vm_type"] == "c2m2" +def _scaling_deploy_recorder( + mocker: MockerFixture, +) -> tuple[MagicMock, FakeClient]: + """Record a deployment using the real instance-bounds helper. + + Args: + mocker: The pytest-mock fixture. + + Returns: + The calls recorder and authenticated API client. + """ + client = _common_deploy_mocks(mocker) + mocker.patch("reflex_cli.utils.hosting.search_app", return_value=app_summary()) + client.api.apps.get.return_value = app(min_instances=1, max_instances=4) + recorder = MagicMock() + recorder.attach_mock(client.api.apps.set_instance_bounds, "bounds") + recorder.attach_mock(client.api.deployments.create, "submit") + recorder.attach_mock(mocker.patch("reflex_cli.utils.deploy.time.sleep"), "sleep") + return recorder, client + + +def _bounds_scaling_refusal( + base_url: str = "https://build.reflex.dev", +) -> APIStatusError: + """Build a refused bounds update because the app is being scaled. + + Args: + base_url: The configured backend URL, including any path prefix. + + Returns: + The SDK exception for the bounds scaling conflict. + """ + return api_error( + 409, + "a scale operation is already running", + code="instance_bounds_scale_conflict", + method="POST", + path=f"apps/{_APP_ID}/instance_bounds", + base_url=base_url, + ) + + +def _unknown_deploy_outcome(failure: str, refusal: APIStatusError) -> BaseException: + """Build an exception that leaves a write's outcome unknown. + + Args: + failure: The kind of unknown outcome. + refusal: A previous refusal supplying the request metadata. + + Returns: + The exception raised before the write is acknowledged. + """ + if failure == "connection": + return APIConnectionError("connection lost", request=refusal.request) + if failure == "timeout": + return APITimeoutError("request timed out", request=refusal.request) + if failure == "response": + return APIResponseValidationError( + "invalid response", + response=dataclasses.replace( + refusal.response, status_code=200, content=b"invalid JSON" + ), + ) + if failure == "interrupt": + return KeyboardInterrupt() + return ValueError("invalid JSON") + + +@pytest.mark.parametrize("max_instances", [None, 8]) +@pytest.mark.parametrize( + "backend_url", ["https://build.reflex.dev", "https://build.reflex.dev/proxy"] +) +def test_deploy_retries_instance_bounds_scaling_conflict( + mocker: MockerFixture, + mock_export_fn: MagicMock, + max_instances: int | None, + backend_url: str, +): + """A refused scaling operation waits and retries without exporting again. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + max_instances: An explicit maximum, or None to preserve the latest value. + backend_url: The configured backend URL, including any path prefix. + """ + recorder, client = _scaling_deploy_recorder(mocker) + client.api.base_url = backend_url + client.api.apps.set_instance_bounds.side_effect = [ + _bounds_scaling_refusal(backend_url), + None, + ] + client.api.apps.get.side_effect = [app(max_instances=4), app(max_instances=6)] + + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + max_instances=max_instances, + ) + + assert [call[0] for call in recorder.mock_calls] == [ + "bounds", + "sleep", + "bounds", + "submit", + ] + assert recorder.bounds.call_args_list[0].kwargs == { + "min_instances": 2, + "max_instances": 4 if max_instances is None else max_instances, + } + assert recorder.bounds.call_args_list[1].kwargs == { + "min_instances": 2, + "max_instances": 6 if max_instances is None else max_instances, + } + recorder.sleep.assert_called_once_with(15) + assert [call.args[3:5] for call in mock_export_fn.call_args_list] == [ + (False, True), + (True, False), + ] + + +def test_deploy_exhausts_bounds_scaling_conflicts_without_another_sleep( + mocker: MockerFixture, + mock_export_fn: MagicMock, + caplog: pytest.LogCaptureFixture, +): + """Persistent scaling conflicts stop at a bounded number of attempts. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + caplog: The captured log messages. + """ + recorder, _ = _scaling_deploy_recorder(mocker) + recorder.bounds.side_effect = _bounds_scaling_refusal() + + with pytest.raises(click.exceptions.Exit) as exc_info: + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + assert exc_info.value.exit_code == 1 + assert recorder.bounds.call_count == 8 + assert recorder.sleep.call_count == 7 + assert all(call.args == (15,) for call in recorder.sleep.call_args_list) + warnings = _log_messages(caplog, logging.WARNING) + assert not any( + "may or may not have been applied" in message for message in warnings + ) + assert not any("even though this deploy failed" in message for message in warnings) + recorder.submit.assert_not_called() + assert mock_export_fn.call_count == 2 + + +@pytest.mark.parametrize("previous_conflict", [False, True]) +@pytest.mark.parametrize( + ("field", "value"), + [ + ("status", 400), + ("code", "unrelated_conflict"), + ("code", ""), + ("method", "GET"), + ("path", "deployments/check"), + ("path", f"apps/{uuid.UUID(int=999)}/instance_bounds"), + ("base_url", "https://other.example"), + ("base_url", "https://build.reflex.dev/other-proxy"), + ], +) +@pytest.mark.parametrize( + "backend_url", ["https://build.reflex.dev", "https://build.reflex.dev/proxy"] +) +def test_deploy_does_not_retry_other_bounds_refusals( + mocker: MockerFixture, + mock_export_fn: MagicMock, + previous_conflict: bool, + field: str, + value: str | int, + backend_url: str, + caplog: pytest.LogCaptureFixture, +): + """Only the exact refused scaling operation permits a retry. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + previous_conflict: Whether a retryable refusal preceded the final error. + field: The response or request field that does not match. + value: The nonmatching field value. + backend_url: The configured backend URL, including any path prefix. + caplog: The captured log messages. + """ + recorder, client = _scaling_deploy_recorder(mocker) + client.api.base_url = backend_url + refusal = _bounds_scaling_refusal(backend_url) + other_refusal = api_error( + value if field == "status" and isinstance(value, int) else 409, + refusal.detail, + code=str(value) if field == "code" else refusal.code, + method=str(value) if field == "method" else "POST", + path=str(value) if field == "path" else f"apps/{_APP_ID}/instance_bounds", + base_url=str(value) if field == "base_url" else backend_url, + ) + recorder.bounds.side_effect = ( + [refusal, other_refusal] if previous_conflict else [other_refusal] + ) + + with pytest.raises(click.exceptions.Exit): + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + assert recorder.bounds.call_count == 1 + previous_conflict + assert recorder.sleep.call_count == int(previous_conflict) + assert not any( + "may or may not have been applied" in message + for message in _log_messages(caplog, logging.WARNING) + ) + recorder.submit.assert_not_called() + + +@pytest.mark.parametrize( + "detail", + [ + "the app already has a deployment in progress", + "a scale operation is already running", + _SCALING_DETAIL + ".", + "", + ], +) +def test_deploy_does_not_retry_other_app_busy_reasons( + mocker: MockerFixture, + mock_export_fn: MagicMock, + detail: str, +): + """A generic app-busy conflict does not prove that submission is safe to retry. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + detail: A busy explanation other than the exact scaling refusal. + """ + recorder, client = _scaling_deploy_recorder(mocker) + client.api.deployments.create.side_effect = api_error( + 409, detail, code="app_busy", method="POST", path="deployments" + ) + + with pytest.raises(click.exceptions.Exit): + cli.deploy(app_name="fake-app", export_fn=mock_export_fn, interactive=False) + + recorder.submit.assert_called_once() + recorder.sleep.assert_not_called() + + +@pytest.mark.parametrize("previous_conflict", [False, True]) +@pytest.mark.parametrize("failure", ["connection", "timeout", "decode"]) +def test_deploy_does_not_retry_unknown_bounds_outcomes( + mocker: MockerFixture, + mock_export_fn: MagicMock, + previous_conflict: bool, + failure: str, + caplog: pytest.LogCaptureFixture, +): + """An unanswered write is never repeated even after a confirmed refusal. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + previous_conflict: Whether a scaling refusal preceded the unanswered write. + failure: The kind of unknown response outcome. + caplog: The captured log messages. + """ + recorder, _ = _scaling_deploy_recorder(mocker) + refusal = _bounds_scaling_refusal() + error = _unknown_deploy_outcome(failure, refusal) + recorder.bounds.side_effect = [refusal, error] if previous_conflict else [error] + expected_error = type(error) if failure == "decode" else click.exceptions.Exit + + with pytest.raises(expected_error): + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + assert recorder.bounds.call_count == 1 + previous_conflict + assert recorder.sleep.call_count == int(previous_conflict) + recorder.submit.assert_not_called() + assert any( + "may or may not have been applied" in message + for message in _log_messages(caplog, logging.WARNING) + ) + + +@pytest.mark.parametrize("previous_conflict", [False, True]) +def test_deploy_stops_after_undecodable_bounds_success( + mocker: MockerFixture, + mock_export_fn: MagicMock, + previous_conflict: bool, + caplog: pytest.LogCaptureFixture, +): + """An undecodable success stops deployment without retrying changed bounds. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + previous_conflict: Whether a scaling refusal preceded the successful write. + caplog: The captured log messages. + """ + recorder, _ = _scaling_deploy_recorder(mocker) + refusal = _bounds_scaling_refusal() + error = _unknown_deploy_outcome("response", refusal) + recorder.bounds.side_effect = [refusal, error] if previous_conflict else [error] + + with pytest.raises(click.exceptions.Exit): + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + assert recorder.bounds.call_count == 1 + previous_conflict + assert recorder.sleep.call_count == int(previous_conflict) + recorder.submit.assert_not_called() + warnings = _log_messages(caplog, logging.WARNING) + assert len(warnings) == 1 + assert "server reported success" in warnings[0] + assert "response could not be decoded" in warnings[0] + assert "bounds may have changed" in warnings[0] + assert "may or may not have been applied" not in warnings[0] + + +@pytest.mark.parametrize("previous_conflict", [False, True]) +@pytest.mark.parametrize("during_write", [False, True]) +@pytest.mark.parametrize("status_code", [500, 502, 503, 504]) +def test_deploy_bounds_server_error_warns_only_after_write( + mocker: MockerFixture, + mock_export_fn: MagicMock, + previous_conflict: bool, + during_write: bool, + status_code: int, + caplog: pytest.LogCaptureFixture, +): + """Only a failed write has an uncertain outcome, and neither failure retries. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + previous_conflict: Whether a scaling refusal preceded the server error. + during_write: Whether the error came from the write or preliminary read. + status_code: The server error returned by the API. + caplog: The captured log messages. + """ + recorder, client = _scaling_deploy_recorder(mocker) + error = api_error( + status_code, + "unavailable", + method="POST" if during_write else "GET", + path=f"apps/{_APP_ID}/instance_bounds" if during_write else f"apps/{_APP_ID}", + ) + if during_write: + recorder.bounds.side_effect = ( + [_bounds_scaling_refusal(), error] if previous_conflict else [error] + ) + else: + client.api.apps.get.side_effect = [app(), error] if previous_conflict else error + recorder.bounds.side_effect = _bounds_scaling_refusal() + + with pytest.raises(click.exceptions.Exit): + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + assert recorder.bounds.call_count == int(previous_conflict) + during_write + assert recorder.sleep.call_count == int(previous_conflict) + recorder.submit.assert_not_called() + assert ( + any( + "may or may not have been applied" in message + for message in _log_messages(caplog, logging.WARNING) + ) + == during_write + ) + + +@pytest.mark.parametrize("previous_conflict", [False, True]) +@pytest.mark.parametrize( + "failure", ["connection", "timeout", "response", "decode", "interrupt"] +) +def test_deploy_failed_bounds_read_does_not_warn_about_a_write( + mocker: MockerFixture, + mock_export_fn: MagicMock, + previous_conflict: bool, + failure: str, + caplog: pytest.LogCaptureFixture, +): + """A failed read cannot apply bounds, even after a prior write was refused. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + previous_conflict: Whether a refused write preceded the failed read. + failure: The failure encountered while reading the app. + caplog: The captured log messages. + """ + recorder, client = _scaling_deploy_recorder(mocker) + error = _unknown_deploy_outcome( + failure, api_error(503, "unavailable", path=f"apps/{_APP_ID}") + ) + client.api.apps.get.side_effect = [app(), error] if previous_conflict else error + recorder.bounds.side_effect = _bounds_scaling_refusal() + expected_error = ( + type(error) if failure in {"decode", "interrupt"} else click.exceptions.Exit + ) + + with pytest.raises(expected_error): + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + assert recorder.bounds.call_count == int(previous_conflict) + assert recorder.sleep.call_count == int(previous_conflict) + recorder.submit.assert_not_called() + assert not _log_messages(caplog, logging.WARNING) + + +@pytest.mark.parametrize("during_wait", [False, True]) +def test_deploy_interrupted_bounds_does_not_retry_or_submit( + mocker: MockerFixture, + mock_export_fn: MagicMock, + during_wait: bool, + caplog: pytest.LogCaptureFixture, +): + """Interrupts preserve whether the last bounds write was refused or unanswered. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + during_wait: Whether interruption happens after the write was refused. + caplog: The captured log messages. + """ + recorder, _ = _scaling_deploy_recorder(mocker) + recorder.bounds.side_effect = ( + _bounds_scaling_refusal() if during_wait else KeyboardInterrupt() + ) + recorder.sleep.side_effect = KeyboardInterrupt() + + with pytest.raises(KeyboardInterrupt): + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + recorder.bounds.assert_called_once() + assert recorder.sleep.call_count == int(during_wait) + recorder.submit.assert_not_called() + warnings = _log_messages(caplog, logging.WARNING) + assert any( + "may or may not have been applied" in message for message in warnings + ) == (not during_wait) + assert not any("even though this deploy failed" in message for message in warnings) + + +@pytest.mark.parametrize( + "failure", ["connection", "timeout", "response", "decode", "interrupt"] +) +def test_deploy_does_not_resubmit_unknown_submission_outcomes( + mocker: MockerFixture, + mock_export_fn: MagicMock, + failure: str, + caplog: pytest.LogCaptureFixture, +): + """An SDK submit with an unknown outcome is called only once. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + failure: The kind of unknown submission outcome. + caplog: The captured log messages. + """ + recorder, _ = _scaling_deploy_recorder(mocker) + refusal = api_error( + 409, _SCALING_DETAIL, code="app_busy", method="POST", path="deployments" + ) + error = _unknown_deploy_outcome(failure, refusal) + recorder.submit.side_effect = error + expected_error = ( + type(error) if failure in {"decode", "interrupt"} else click.exceptions.Exit + ) + + with pytest.raises(expected_error): + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + recorder.bounds.assert_called_once() + recorder.submit.assert_called_once() + recorder.sleep.assert_not_called() + assert any( + "even though this deploy failed" in message + for message in _log_messages(caplog, logging.WARNING) + ) + + +def test_deploy_watcher_failure_does_not_resubmit( + mocker: MockerFixture, + mock_export_fn: MagicMock, + caplog: pytest.LogCaptureFixture, +): + """A lost watcher response cannot resubmit an accepted deployment. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + caplog: The captured log messages. + """ + recorder, _ = _scaling_deploy_recorder(mocker) + watch = mocker.patch( + "reflex_cli.utils.hosting.watch_deployment_status", + side_effect=APIConnectionError( + "connection lost", request=_bounds_scaling_refusal().request + ), + ) + + with pytest.raises(APIConnectionError): + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + min_instances=2, + ) + + recorder.bounds.assert_called_once() + recorder.submit.assert_called_once() + watch.assert_called_once() + recorder.sleep.assert_not_called() + assert not any( + "even though this deploy failed" in message + for message in _log_messages(caplog, logging.WARNING) + ) + + +@pytest.mark.parametrize("regions", [None, ["iad"]]) +def test_deploy_gcp_preserves_vmtype_and_region_notice_across_bounds_retry( + mocker: MockerFixture, + mock_export_fn: MagicMock, + regions: list[str] | None, + caplog: pytest.LogCaptureFixture, +): + """GCP retries keep VM sizing and drop only regions, with one region notice. + + Args: + mocker: The pytest-mock fixture. + mock_export_fn: The mocked build exporter. + regions: The regions explicitly requested by the caller. + caplog: The captured log messages. + """ + recorder, client = _scaling_deploy_recorder(mocker) + mocker.patch( + "reflex_cli.utils.hosting.search_app", return_value=app_summary(provider="gcp") + ) + client.api.apps.get.return_value = app( + provider="gcp", min_instances=1, max_instances=4 + ) + recorder.bounds.side_effect = [_bounds_scaling_refusal(), None] + + cli.deploy( + app_name="fake-app", + export_fn=mock_export_fn, + interactive=False, + regions=regions, + vmtype="c2m2", + min_instances=2, + ) + + assert recorder.bounds.call_count == 2 + recorder.submit.assert_called_once() + assert recorder.submit.call_args.kwargs["vm_type"] == "c2m2" + assert recorder.submit.call_args.kwargs["regions"] is None + assert client.api.deployments.check.call_args.kwargs["vm_type"] == "c2m2" + assert client.api.deployments.check.call_args.kwargs["regions"] is None + notices = [ + message + for message in _log_messages(caplog, logging.INFO) + if "Ignoring" in message + ] + assert len(notices) == int(bool(regions)) + assert all("--region" in message and "vmtype" not in message for message in notices) + assert mock_export_fn.call_count == 2 + + @pytest.mark.parametrize( ("min_instances", "max_instances"), [(1, 4), (2, None), (None, 8)], @@ -1192,6 +1813,7 @@ def test_deploy_sets_instance_bounds_before_submitting( assert bounds_kwargs["app_id"] == str(_APP_ID) assert bounds_kwargs["min_instances"] == min_instances assert bounds_kwargs["max_instances"] == max_instances + assert bounds_kwargs["retry_scaling"] is True def test_deploy_without_instance_bounds_flags_skips_the_call( @@ -1268,8 +1890,8 @@ def test_deploy_hedges_when_the_bounds_response_is_lost( caplog: pytest.LogCaptureFixture, ): """A dropped response leaves the outcome unknown, so neither is asserted.""" - recorder, _ = _deploy_call_recorder(mocker) - recorder.set_instance_bounds.side_effect = KeyError("no route") + recorder, _ = _scaling_deploy_recorder(mocker) + recorder.bounds.side_effect = KeyError("no route") with pytest.raises(KeyError): cli.deploy( @@ -1285,7 +1907,7 @@ def test_deploy_hedges_when_the_bounds_response_is_lost( if "instance bounds" in msg ) assert "may or may not have been applied" in warning - recorder.create_deployment.assert_not_called() + recorder.submit.assert_not_called() def test_deploy_does_not_warn_about_bounds_it_never_applied(