Do not downgrade runner and shim by default

mkdocs/docs/reference/env.md

@@ -161,6 +161,8 @@ For more details on the options below, refer to the [server deployment](../guide
 - `DSTACK_SERVER_SSH_CONNECT_TIMEOUT`{ #DSTACK_SERVER_SSH_CONNECT_TIMEOUT } – The SSH `ConnectTimeout` for server-instance connections, in seconds. Defaults to `3`. Increase if there are high-latency links between the server and instances.
 - `DSTACK_SERVER_SSH_POOL_DISABLED`{ #DSTACK_SERVER_SSH_POOL_DISABLED } – Disables the reuse of server SSH connections to instances. If set, significantly decreases server RAM usage, but
 slows down processing and may cause CPU spikes due to frequent SSH-connection establishment.
+- `DSTACK_RUNNER_ALLOW_DOWNGRADE`{ #DSTACK_RUNNER_ALLOW_DOWNGRADE } – When set to any value, allows the server to install an older `dstack-runner` version over a newer one. By default, the server skips installation if a newer version is already installed, so that replicas running different server versions don't reinstall the runner over each other during a rolling deployment. Set this variable only while downgrading the server, and unset it afterwards — leaving it set brings the reinstall loop back.
+- `DSTACK_SHIM_ALLOW_DOWNGRADE`{ #DSTACK_SHIM_ALLOW_DOWNGRADE } – Same as `DSTACK_RUNNER_ALLOW_DOWNGRADE` but for `dstack-shim`.
 
 ??? info "Internal environment variables"
      The following environment variables are intended for development purposes:
@@ -169,10 +171,15 @@ slows down processing and may cause CPU spikes due to frequent SSH-connection es
      * `DSTACK_SERVER_UVICORN_LOG_LEVEL` – Sets uvicorn logger log level. Defaults to `ERROR`.
      * `DSTACK_SERVER_MAX_OFFERS_TRIED` - Sets how many instance offers to try when starting a job.
        Setting a high value can degrade server performance.
-     * `DSTACK_RUNNER_VERSION` – Sets exact runner version for debug. Defaults to `latest`. Ignored if `DSTACK_RUNNER_DOWNLOAD_URL` is set.
+     * `DSTACK_RUNNER_VERSION` – Overrides the `dstack-runner` version the server installs on instances. Release builds of the server
+      default to their own version; dev builds have no default version. Must be a valid [PyPA version](https://packaging.python.org/en/latest/specifications/version-specifiers/), for example, `0.20.1`.
+     * `DSTACK_RUNNER_VERSION_URL` – URL to fetch the `dstack-runner` version from. The response body must be the version string,
+      see `DSTACK_RUNNER_VERSION` for the format. Used only if `DSTACK_RUNNER_VERSION` is not set.
      * `DSTACK_RUNNER_DOWNLOAD_URL` – Overrides `dstack-runner` binary download URL. The URL can contain `{version}` and/or `{arch}` placeholders,
-      where `{version}` is `dstack` version in the `X.Y.Z` format or `latest`, and `{arch}` is either `amd64` or `arm64`, for example,
-      `https://dstack.example.com/{arch}/{version}/dstack-runner`.
+      where `{version}` is the `dstack-runner` version (see `DSTACK_RUNNER_VERSION`), or `latest` if the version cannot be determined,
+      and `{arch}` is either `amd64` or `arm64`, for example, `https://dstack.example.com/{arch}/{version}/dstack-runner`.
+     * `DSTACK_SHIM_VERSION` – Same as `DSTACK_RUNNER_VERSION` but for `dstack-shim`.
+     * `DSTACK_SHIM_VERSION_URL` – Same as `DSTACK_RUNNER_VERSION_URL` but for `dstack-shim`.
      * `DSTACK_SHIM_DOWNLOAD_URL` – Overrides `dstack-shim` binary download URL. The URL can contain `{version}` and/or `{arch}` placeholders,
       see `DSTACK_RUNNER_DOWNLOAD_URL` for the details.
      * `DSTACK_GATEWAY_PACKAGE_URL` – Overrides the URL the `dstack` package is installed from on gateway instances (used as `dstack[gateway] @ <URL>`).

src/dstack/_internal/core/backends/base/compute.py

@@ -969,13 +969,6 @@ def get_dstack_shim_version() -> Optional[str]:
         return version
     if version := settings.DSTACK_SHIM_VERSION:
         return version
-    if version := settings.DSTACK_RUNNER_VERSION:
-        logger.warning(
-            "DSTACK_SHIM_VERSION is not set, using DSTACK_RUNNER_VERSION."
-            " Future versions will not fall back to DSTACK_RUNNER_VERSION."
-            " Set DSTACK_SHIM_VERSION to supress this warning."
-        )
-        return version
     if version_url := settings.DSTACK_SHIM_VERSION_URL:
         return _fetch_version(version_url)
     if settings.DSTACK_USE_LATEST_FROM_BRANCH:

src/dstack/_internal/server/background/pipeline_tasks/instances/check.py

@@ -5,11 +5,13 @@
 from typing import Optional
 
 import gpuhunt
+import packaging.version
 import requests
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import joinedload
 
+from dstack._internal import settings
 from dstack._internal.core.backends.base.backend import Backend
 from dstack._internal.core.backends.base.compute import (
     get_dstack_runner_download_url,
@@ -46,6 +48,7 @@
 from dstack._internal.server.schemas.instances import InstanceCheck
 from dstack._internal.server.schemas.runner import (
     ComponentInfo,
+    ComponentName,
     ComponentStatus,
     InstanceHealthResponse,
 )
@@ -482,136 +485,244 @@ def _maybe_install_components(
     installation_requested = False
 
     if (runner_info := components.runner) is not None:
-        installation_requested |= _maybe_install_runner(instance_model, shim_client, runner_info)
+        installation_requested |= _maybe_install_runner(
+            instance_model=instance_model,
+            shim_client=shim_client,
+            runner_info=runner_info,
+            allow_downgrade=settings.DSTACK_RUNNER_ALLOW_DOWNGRADE,
+        )
     else:
-        logger.debug("Instance %s: no runner info", instance_model.name)
+        logger.debug(
+            "Instance %s: %s: no component info", instance_model.name, ComponentName.RUNNER.value
+        )
 
     if (shim_info := components.shim) is not None:
         if shim_info.status == ComponentStatus.INSTALLED:
             installed_shim_version = shim_info.version
-        installation_requested |= _maybe_install_shim(instance_model, shim_client, shim_info)
+        installation_requested |= _maybe_install_shim(
+            instance_model=instance_model,
+            shim_client=shim_client,
+            shim_info=shim_info,
+            allow_downgrade=settings.DSTACK_SHIM_ALLOW_DOWNGRADE,
+        )
     else:
-        logger.debug("Instance %s: no shim info", instance_model.name)
+        logger.debug(
+            "Instance %s: %s: no component info", instance_model.name, ComponentName.SHIM.value
+        )
 
-    # old shim without `dstack-shim` component and `/api/shutdown` support
-    # or the same version is already running
-    # or we just requested installation of at least one component
-    # or at least one component is already being installed
-    # or at least one shim task won't survive restart
     running_shim_version = shim_client.get_version_string()
+    # skip the restart if:
     if (
+        # old shim without `dstack-shim` component and `/api/shutdown` support
         installed_shim_version is None
+        # the same version is already running -- a string comparison on purpose: both sides come
+        # from the same `Version` build-time variable, one read off the on-disk binary, the other
+        # reported by the running process, so the question is whether it's the same build, not
+        # which version is newer
         or installed_shim_version == running_shim_version
+        # we just requested installation of at least one component
         or installation_requested
+        # at least one component is already being installed
         or any(component.status == ComponentStatus.INSTALLING for component in components)
+        # at least one shim task won't survive restart
         or not shim_client.is_safe_to_restart()
     ):
         return
 
     if shim_client.shutdown(force=False):
         logger.debug(
-            "Instance %s: restarting shim %s -> %s",
+            "Instance %s: %s: restarting %r -> %r",
             instance_model.name,
+            ComponentName.SHIM.value,
             running_shim_version,
             installed_shim_version,
         )
     else:
-        logger.debug("Instance %s: cannot restart shim", instance_model.name)
+        logger.debug(
+            "Instance %s: %s: cannot restart", instance_model.name, ComponentName.SHIM.value
+        )
 
 
 def _maybe_install_runner(
     instance_model: InstanceModel,
     shim_client: runner_client.ShimClient,
     runner_info: ComponentInfo,
+    allow_downgrade: bool,
 ) -> bool:
     # For developers:
     # * To install the latest dev build for the current branch from the CI,
     #   set DSTACK_USE_LATEST_FROM_BRANCH=1.
     # * To provide your own build, set DSTACK_RUNNER_VERSION_URL and DSTACK_RUNNER_DOWNLOAD_URL.
-    expected_version = get_dstack_runner_version()
-    if expected_version is None:
-        return False
-
+    name = runner_info.name
     installed_version = runner_info.version
+    expected_version = get_dstack_runner_version()
     logger.debug(
-        "Instance %s: runner status=%s installed_version=%s",
+        "Instance %s: %s: status=%s installed_version=%r expected_version=%r",
         instance_model.name,
+        name,
         runner_info.status.value,
-        installed_version or "(no version)",
+        installed_version,
+        expected_version,
     )
-    if runner_info.status == ComponentStatus.INSTALLING:
-        logger.debug("Instance %s: runner is already being installed", instance_model.name)
+
+    if not expected_version:
         return False
-    if installed_version and installed_version == expected_version:
-        logger.debug("Instance %s: expected runner version already installed", instance_model.name)
+
+    if not _should_install_component(
+        instance_model=instance_model,
+        component_info=runner_info,
+        expected_version=expected_version,
+        allow_downgrade=allow_downgrade,
+    ):
         return False
 
-    url = get_dstack_runner_download_url(
+    download_url = get_dstack_runner_download_url(
         arch=_get_instance_cpu_arch(instance_model),
         version=expected_version,
     )
     logger.debug(
-        "Instance %s: installing runner %s -> %s from %s",
+        "Instance %s: %s: installing %r -> %r from %s",
         instance_model.name,
-        installed_version or "(no version)",
+        name,
+        installed_version,
         expected_version,
-        url,
+        download_url,
     )
     try:
-        shim_client.install_runner(url)
+        shim_client.install_runner(download_url)
         return True
-    except requests.RequestException as exc:
-        logger.warning("Instance %s: shim.install_runner(): %s", instance_model.name, exc)
+    except requests.RequestException as e:
+        logger.warning("Instance %s: %s: failed to install: %s", instance_model.name, name, e)
     return False
 
 
 def _maybe_install_shim(
     instance_model: InstanceModel,
     shim_client: runner_client.ShimClient,
     shim_info: ComponentInfo,
+    allow_downgrade: bool,
 ) -> bool:
     # For developers:
     # * To install the latest dev build for the current branch from the CI,
     #   set DSTACK_USE_LATEST_FROM_BRANCH=1.
     # * To provide your own build, set DSTACK_SHIM_VERSION_URL and DSTACK_SHIM_DOWNLOAD_URL.
-    expected_version = get_dstack_shim_version()
-    if expected_version is None:
-        return False
-
+    name = shim_info.name
     installed_version = shim_info.version
+    expected_version = get_dstack_shim_version()
     logger.debug(
-        "Instance %s: shim status=%s installed_version=%s running_version=%s",
+        "Instance %s: %s: status=%s installed_version=%r expected_version=%r running_version=%r",
         instance_model.name,
+        name,
         shim_info.status.value,
-        installed_version or "(no version)",
+        installed_version,
+        expected_version,
         shim_client.get_version_string(),
     )
-    if shim_info.status == ComponentStatus.INSTALLING:
-        logger.debug("Instance %s: shim is already being installed", instance_model.name)
+
+    if not expected_version:
         return False
-    if installed_version and installed_version == expected_version:
-        logger.debug("Instance %s: expected shim version already installed", instance_model.name)
+
+    if not _should_install_component(
+        instance_model=instance_model,
+        component_info=shim_info,
+        expected_version=expected_version,
+        allow_downgrade=allow_downgrade,
+    ):
         return False
 
-    url = get_dstack_shim_download_url(
+    download_url = get_dstack_shim_download_url(
         arch=_get_instance_cpu_arch(instance_model),
         version=expected_version,
     )
     logger.debug(
-        "Instance %s: installing shim %s -> %s from %s",
+        "Instance %s: %s: installing %r -> %r from %s",
         instance_model.name,
-        installed_version or "(no version)",
+        name,
+        installed_version,
         expected_version,
-        url,
+        download_url,
     )
     try:
-        shim_client.install_shim(url)
+        shim_client.install_shim(download_url)
         return True
-    except requests.RequestException as exc:
-        logger.warning("Instance %s: shim.install_shim(): %s", instance_model.name, exc)
+    except requests.RequestException as e:
+        logger.warning("Instance %s: %s: failed to install: %s", instance_model.name, name, e)
     return False
 
 
+def _should_install_component(
+    instance_model: InstanceModel,
+    component_info: ComponentInfo,
+    expected_version: str,
+    allow_downgrade: bool,
+) -> bool:
+    """
+    Decides whether the component should be installed, comparing the installed and the expected
+    versions. Logs the reason if the installation is skipped.
+
+    Unless `allow_downgrade` is set, the component is not installed if the installed version is
+    newer than the expected one. This keeps server replicas running different versions from
+    reinstalling the component over each other during a rolling deployment.
+    """
+    name = component_info.name
+
+    if component_info.status == ComponentStatus.INSTALLING:
+        logger.debug("Instance %s: %s: already being installed", instance_model.name, name)
+        return False
+
+    expected_version_parsed = _parse_pypa_version(expected_version)
+    if expected_version_parsed is None:
+        logger.warning(
+            "Instance %s: %s: failed to parse expected_version: %r",
+            instance_model.name,
+            name,
+            expected_version,
+        )
+        return False
+
+    installed_version = component_info.version
+    if not installed_version:
+        return True
+
+    installed_version_parsed = _parse_pypa_version(installed_version)
+    if installed_version_parsed is None:
+        # Dev builds report `latest`; treat any unparseable version as the newest one
+        if not allow_downgrade:
+            logger.debug(
+                "Instance %s: %s: cannot parse installed_version, skipping the install",
+                instance_model.name,
+                name,
+            )
+            return False
+    elif installed_version_parsed == expected_version_parsed:
+        logger.debug(
+            "Instance %s: %s: expected version already installed", instance_model.name, name
+        )
+        return False
+    elif installed_version_parsed > expected_version_parsed and not allow_downgrade:
+        logger.debug("Instance %s: %s: newer version already installed", instance_model.name, name)
+        return False
+
+    return True
+
+
+def _parse_pypa_version(version_string: str) -> Optional[packaging.version.Version]:
+    """
+    Parses the version for comparing component versions, keeping the pre-release, dev, post-release,
+    and local segments, that is, `0.20.1rc1` is older than `0.20.1`.
+    Returns `None` if the version is not PyPA-conformant, e.g., `latest` reported by dev builds.
+
+    Not to be confused with `ShimClient.get_version_tuple()`, which is based on
+    `dstack._internal.server.services.runner.client._parse_version` -- it truncates the version to
+    `(major, minor, micro)` and treats unparseable versions as the latest, suiting feature gating
+    but not version comparison.
+    """
+    try:
+        return packaging.version.parse(version_string)
+    except packaging.version.InvalidVersion:
+        return None
+
+
 def _get_instance_cpu_arch(instance_model: InstanceModel) -> Optional[gpuhunt.CPUArchitecture]:
     job_provisioning_data = get_instance_provisioning_data(instance_model)
     if job_provisioning_data is None:

src/dstack/_internal/settings.py

@@ -13,9 +13,11 @@
 DSTACK_RUNNER_VERSION = os.getenv("DSTACK_RUNNER_VERSION")
 DSTACK_RUNNER_VERSION_URL = os.getenv("DSTACK_RUNNER_VERSION_URL")
 DSTACK_RUNNER_DOWNLOAD_URL = os.getenv("DSTACK_RUNNER_DOWNLOAD_URL")
+DSTACK_RUNNER_ALLOW_DOWNGRADE = os.getenv("DSTACK_RUNNER_ALLOW_DOWNGRADE") is not None
 DSTACK_SHIM_VERSION = os.getenv("DSTACK_SHIM_VERSION")
 DSTACK_SHIM_VERSION_URL = os.getenv("DSTACK_SHIM_VERSION_URL")
 DSTACK_SHIM_DOWNLOAD_URL = os.getenv("DSTACK_SHIM_DOWNLOAD_URL")
+DSTACK_SHIM_ALLOW_DOWNGRADE = os.getenv("DSTACK_SHIM_ALLOW_DOWNGRADE") is not None
 DSTACK_USE_LATEST_FROM_BRANCH = os.getenv("DSTACK_USE_LATEST_FROM_BRANCH") is not None
 DSTACK_GATEWAY_PACKAGE_URL = os.getenv("DSTACK_GATEWAY_PACKAGE_URL")
 

src/tests/_internal/server/background/pipeline_tasks/test_instances/conftest.py

@@ -15,6 +15,12 @@
 from dstack._internal.server.schemas.instances import InstanceCheck
 
 
+def pytest_configure(config: pytest.Config) -> None:
+    config.addinivalue_line(
+        "markers", "allow_downgrade: allow the server to install an older component version"
+    )
+
+
 @pytest.fixture
 def fetcher() -> InstanceFetcher:
     return InstanceFetcher(