Re-creating failed gateways with a single apply

src/dstack/_internal/cli/services/configurators/gateway.py

@@ -59,6 +59,13 @@ def apply_configuration(
             if plan.effective_spec.configuration.name is not None:
                 action_message += f"Gateway [code]{plan.effective_spec.configuration.name}[/] does not exist yet."
             confirm_message += "Create the gateway?"
+        elif plan.current_resource.status == GatewayStatus.FAILED:
+            action_message += (
+                f"Found gateway [code]{plan.effective_spec.configuration.name}[/]"
+                " in [error]FAILED[/] status."
+            )
+            confirm_message += "Re-create the gateway?"
+            delete_gateway_name = plan.current_resource.name
         else:
             action_message += f"Found gateway [code]{plan.effective_spec.configuration.name}[/]."
             diff = diff_gateway_configurations(

src/dstack/_internal/server/compatibility/gateways.py

@@ -2,11 +2,13 @@
 
 from packaging.version import Version
 
+from dstack._internal.core.errors import ServerClientError
 from dstack._internal.core.models.gateways import (
     Gateway,
     GatewayConfiguration,
     GatewayPlan,
     GatewaySpec,
+    GatewayStatus,
 )
 
 
@@ -47,6 +49,17 @@ def patch_gateway(gateway: Gateway, client_version: Optional[Version]) -> None:
 def patch_gateway_plan(plan: GatewayPlan, client_version: Optional[Version]) -> None:
     if client_version is None:
         return
+    if (
+        client_version < Version("0.22.7")
+        and plan.current_resource is not None
+        and plan.current_resource.status == GatewayStatus.FAILED
+    ):
+        # Pre-0.22.7 CLIs display a misleading message if current_resource is a failed gateway,
+        # so return an error the same way pre-0.22.7 servers used to do.
+        raise ServerClientError(
+            f"Gateway {plan.current_resource.name!r} is in FAILED status and"
+            " cannot be updated in-place. Delete it and re-apply."
+        )
     _patch_gateway_configuration(plan.spec.configuration, client_version)
     _patch_gateway_configuration(plan.effective_spec.configuration, client_version)
     if plan.current_resource is not None:

src/dstack/_internal/server/services/gateways/__init__.py

@@ -1064,18 +1064,16 @@ async def get_plan(
                 raise ServerClientError(
                     f"Gateway {effective_spec.configuration.name!r} is being deleted. Try again later."
                 )
-            if current_gateway_model.status == GatewayStatus.FAILED:
-                raise ServerClientError(
-                    f"Gateway {effective_spec.configuration.name!r} is in FAILED status and"
-                    " cannot be updated in-place. Delete it and re-apply."
-                )
             current_gateway = gateway_model_to_gateway(
                 current_gateway_model, default_gateway_id=project.default_gateway_id
             )
-            if _can_update_gateway_in_place(
-                diff_gateway_configurations(
-                    current_gateway.configuration,
-                    effective_spec.configuration,
+            if (
+                current_gateway_model.status != GatewayStatus.FAILED
+                and _can_update_gateway_in_place(
+                    diff_gateway_configurations(
+                        current_gateway.configuration,
+                        effective_spec.configuration,
+                    )
                 )
             ):
                 action = ApplyAction.UPDATE

src/tests/_internal/server/routers/test_gateways.py

@@ -1643,8 +1643,68 @@ async def test_get_plan_with_domain_change_is_update(
     @pytest.mark.asyncio
     @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
     @pytest.mark.parametrize("populate_configuration", [True, False])
-    async def test_get_plan_rejects_failed_gateway(
+    @pytest.mark.parametrize("client_version", [None, "0.22.7"])
+    async def test_get_plan_with_failed_gateway_is_create(
+        self,
+        test_db,
+        session: AsyncSession,
+        client: AsyncClient,
+        populate_configuration: bool,
+        client_version: Optional[str],
+    ):
+        gateway_name = "my-gateway"
+        user, project = await self._create_project_with_failed_gateway(
+            session, gateway_name, populate_configuration
+        )
+        headers = get_auth_headers(user.token)
+        if client_version is not None:
+            headers["X-API-Version"] = client_version
+        response = await client.post(
+            f"/api/project/{project.name}/gateways/get_plan",
+            json={"spec": self._failed_gateway_spec(gateway_name)},
+            headers=headers,
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["action"] == ApplyAction.CREATE
+        assert data["current_resource"]["name"] == gateway_name
+        assert data["current_resource"]["status"] == GatewayStatus.FAILED
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    @pytest.mark.parametrize("populate_configuration", [True, False])
+    async def test_get_plan_with_failed_gateway_rejected_for_old_client(
         self, test_db, session: AsyncSession, client: AsyncClient, populate_configuration: bool
+    ):
+        gateway_name = "my-gateway"
+        user, project = await self._create_project_with_failed_gateway(
+            session, gateway_name, populate_configuration
+        )
+        headers = get_auth_headers(user.token)
+        headers["X-API-Version"] = "0.22.6"
+        response = await client.post(
+            f"/api/project/{project.name}/gateways/get_plan",
+            json={"spec": self._failed_gateway_spec(gateway_name)},
+            headers=headers,
+        )
+        assert response.status_code == 400
+        assert "FAILED status" in response.json()["detail"][0]["msg"]
+
+    @staticmethod
+    def _failed_gateway_spec(name: str) -> dict:
+        return {
+            "configuration": {
+                "type": "gateway",
+                "name": name,
+                "backend": "aws",
+                "region": "us-east-1",
+                "domain": "new.example.com",
+            }
+        }
+
+    @staticmethod
+    async def _create_project_with_failed_gateway(
+        session: AsyncSession, name: str, populate_configuration: bool
     ):
         user = await create_user(session, global_role=GlobalRole.USER)
         project = await create_project(session)
@@ -1656,7 +1716,7 @@ async def test_get_plan_rejects_failed_gateway(
             session=session,
             project_id=project.id,
             backend_id=backend.id,
-            name="my-gateway",
+            name=name,
             region="us-east-1",
             wildcard_domain="old.example.com",
             status=GatewayStatus.FAILED,
@@ -1668,23 +1728,7 @@ async def test_get_plan_rejects_failed_gateway(
             gateway_id=gateway.id,
             populate_configuration=populate_configuration,
         )
-        response = await client.post(
-            f"/api/project/{project.name}/gateways/get_plan",
-            json={
-                "spec": {
-                    "configuration": {
-                        "type": "gateway",
-                        "name": "my-gateway",
-                        "backend": "aws",
-                        "region": "us-east-1",
-                        "domain": "new.example.com",
-                    }
-                }
-            },
-            headers=get_auth_headers(user.token),
-        )
-        assert response.status_code == 400
-        assert "FAILED status" in response.json()["detail"][0]["msg"]
+        return user, project
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)