diff --git a/bioengine/apps/manager.py b/bioengine/apps/manager.py index 783ca79..9e973ac 100644 --- a/bioengine/apps/manager.py +++ b/bioengine/apps/manager.py @@ -309,6 +309,56 @@ def _get_full_artifact_id(self, artifact_id: str) -> str: # If artifact_id does not contain a slash, prepend the workspace return f"{self.server.config.workspace}/{artifact_id}" + def _get_startup_version_pin( + self, application_id: str, artifact_id: str + ) -> Optional[str]: + """Return the version the startup config will restore on the next restart. + + None when the app is not a startup application, or is pinned without a + version — in that case a restart redeploys whatever is already running. + """ + for app_config in self.startup_applications: + if not isinstance(app_config, dict): + continue + if app_config.get("application_id") != application_id: + continue + config_artifact_id = app_config.get("artifact_id") + if not config_artifact_id: + continue + if self._get_full_artifact_id(config_artifact_id) == artifact_id: + return app_config.get("version") + return None + + def _warn_on_startup_pin_divergence(self, app_config: Dict[str, Any]) -> None: + """Report a startup pin that is about to move a running app to another version. + + ``recover_deployed_applications()`` runs first, so both numbers are in + hand here. Without this the revert is indistinguishable from a clean + start — same RUNNING status, same healthy probes, just older code. + """ + pinned_version = app_config.get("version") + application_id = app_config.get("application_id") + if not pinned_version or not application_id: + return + + running = self._deployed_applications.get(application_id) + if not running or running["version"] == pinned_version: + return + if running["artifact_id"] != self._get_full_artifact_id( + app_config["artifact_id"] + ): + return + + self.logger.warning( + f"Startup application '{application_id}' (artifact " + f"'{running['artifact_id']}') was found running version " + f"{running['version'] or 'latest'!r} but is pinned to " + f"{pinned_version!r} — deploying the pinned version. If " + f"{running['version'] or 'latest'!r} was rolled out over the API, " + f"that change is being reverted; update the worker's " + f"startup_applications config to keep it." + ) + async def _generate_application_id(self) -> str: """ Generate a unique identifier for a new application deployment. @@ -985,6 +1035,13 @@ async def _get_app_status( application_id, application_details, application_info["version"] ) + # The version a worker restart would restore. Differs from ``version`` + # only when the app was rolled out over the API without updating the + # worker's startup_applications config; None when it isn't pinned. + pinned_version = self._get_startup_version_pin( + application_id, application_info["artifact_id"] + ) + # Build static site URL with runtime config params so the frontend # knows which Hypha server and service to connect to. base_static_url = application_info.get("static_site_url") @@ -1005,6 +1062,7 @@ async def _get_app_status( "version": application_info["version"] or "latest", "running_version": running_version, "version_verified": version_verified, + "pinned_version": pinned_version, "recovered_app": application_info["recovered_app"], "status": status, "message": message, @@ -1292,6 +1350,8 @@ async def deploy_startup_applications(self) -> None: f"{', '.join(sorted(invalid_keys))}. Valid keys are: {', '.join(sorted(valid_keys))}" ) + self._warn_on_startup_pin_divergence(app_config) + if "hypha_token" not in app_config: app_config["hypha_token"] = startup_applications_token diff --git a/tests/apps/test_startup_pin_divergence.py b/tests/apps/test_startup_pin_divergence.py new file mode 100644 index 0000000..873456e --- /dev/null +++ b/tests/apps/test_startup_pin_divergence.py @@ -0,0 +1,241 @@ +"""Pin the reporting of a startup version pin that reverts a running app. + +A worker pod restart replays ``--startup-applications``, but an app rolled out +over the API only ever changed the worker's runtime state. When the pin still +carries the old version the restart quietly reinstates it: status RUNNING, +deployments HEALTHY, no diff against prior state — the app is simply running +older code. That happened on deNBI on 2026-09-01 (model-runner 2.7.2 -> 2.4.2) +and was found only because someone compared versions by hand. + +``recover_deployed_applications()`` runs before ``deploy_startup_applications()``, +so the worker holds both numbers at that moment. Two pins here: + +- ``deploy_startup_applications`` logs a WARNING naming both versions before it + redeploys at the pin. +- ``get_app_status`` returns ``pinned_version`` alongside the running one, so + the divergence is assertable over the API without reading cluster config. + +Neither changes what gets deployed. +""" + +from __future__ import annotations + +import asyncio +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from bioengine.apps.manager import AppsManager + +RUNNING_VERSION = "2.7.2" +PINNED_VERSION = "2.4.2" +ARTIFACT_ID = "bioimage-io/model-runner" +APP_ID = "model-runner" + + +def _make_manager( + *, + startup_applications: list, + running_version: str | None = RUNNING_VERSION, + running_artifact_id: str = ARTIFACT_ID, +) -> AppsManager: + """An AppsManager holding one recovered app plus the given startup pins.""" + is_deployed = asyncio.Event() + is_deployed.set() + + manager = object.__new__(AppsManager) + manager.logger = logging.getLogger("test.startup_pin") + manager.startup_applications = startup_applications + manager._deployed_applications = { + APP_ID: { + "is_deployed": is_deployed, + "display_name": "Model Runner", + "description": "Run bioimage.io models.", + "artifact_id": running_artifact_id, + "version": running_version, + "recovered_app": True, + "application_kwargs": {}, + "application_env_vars": {}, + "disable_gpu": False, + "application_resources": {}, + "authorized_users": ["*"], + "available_methods": ["infer"], + "max_ongoing_requests": 1, + "scaling": {}, + "static_site_url": None, + "started_at": 1_700_000_000.0, + "last_updated_at": 1_700_000_000.0, + "last_updated_by": "user@example.com", + "auto_redeploy": False, + "deployed_by_worker_client_id": "worker-abc", + "proxy_service_token_issued_at": None, + "proxy_service_token_ttl_seconds": None, + } + } + + server = MagicMock() + server.config.workspace = "bioimage-io" + server.generate_token = AsyncMock(return_value="startup-token") + manager.server = server + manager.admin_users = ["admin@example.com"] + + ray_cluster = MagicMock() + ray_cluster.proxy_actor_handle.get_deployment_replicas.remote = AsyncMock( + return_value={} + ) + manager.ray_cluster = ray_cluster + + # deploy_app is the thing under observation, not under test: replace it but + # keep the real __schema__, which deploy_startup_applications reads to + # validate config keys. + deploy_app = AsyncMock(return_value=APP_ID) + deploy_app.__schema__ = AppsManager.deploy_app.__schema__ + manager.deploy_app = deploy_app + + return manager + + +@pytest.mark.asyncio +async def test_divergent_pin_warns_with_both_versions(caplog) -> None: + manager = _make_manager( + startup_applications=[ + { + "artifact_id": ARTIFACT_ID, + "application_id": APP_ID, + "version": PINNED_VERSION, + } + ] + ) + + with caplog.at_level(logging.WARNING, logger="test.startup_pin"): + await manager.deploy_startup_applications() + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert warnings, ( + "A restart that reverts a running app to an older pin must say so. " + "Without this line the revert is indistinguishable from a clean start." + ) + message = warnings[0].getMessage() + assert RUNNING_VERSION in message and PINNED_VERSION in message, message + assert APP_ID in message + + # Observability only — the pinned version is still deployed. + manager.deploy_app.assert_awaited_once() + assert manager.deploy_app.await_args.kwargs["version"] == PINNED_VERSION + + +@pytest.mark.asyncio +async def test_matching_pin_is_silent(caplog) -> None: + manager = _make_manager( + startup_applications=[ + { + "artifact_id": ARTIFACT_ID, + "application_id": APP_ID, + "version": RUNNING_VERSION, + } + ] + ) + + with caplog.at_level(logging.WARNING, logger="test.startup_pin"): + await manager.deploy_startup_applications() + + assert [r for r in caplog.records if r.levelno == logging.WARNING] == [] + + +@pytest.mark.asyncio +async def test_pin_without_a_version_is_silent(caplog) -> None: + # An unversioned pin inherits whatever is already running, so there is + # nothing to revert and nothing to report. + manager = _make_manager( + startup_applications=[ + {"artifact_id": ARTIFACT_ID, "application_id": APP_ID}, + ] + ) + + with caplog.at_level(logging.WARNING, logger="test.startup_pin"): + await manager.deploy_startup_applications() + + assert [r for r in caplog.records if r.levelno == logging.WARNING] == [] + + +@pytest.mark.asyncio +async def test_pin_for_a_different_artifact_is_silent(caplog) -> None: + # Same application_id, different artifact: the versions belong to two + # different release lines and comparing them says nothing. + manager = _make_manager( + startup_applications=[ + { + "artifact_id": ARTIFACT_ID, + "application_id": APP_ID, + "version": PINNED_VERSION, + } + ], + running_artifact_id="bioimage-io/other-app", + ) + + with caplog.at_level(logging.WARNING, logger="test.startup_pin"): + await manager.deploy_startup_applications() + + assert [r for r in caplog.records if r.levelno == logging.WARNING] == [] + + +@pytest.mark.asyncio +async def test_status_surfaces_the_pin_next_to_the_running_version() -> None: + manager = _make_manager( + startup_applications=[ + { + "artifact_id": ARTIFACT_ID, + "application_id": APP_ID, + "version": PINNED_VERSION, + } + ] + ) + + status = await manager._get_app_status( + application_id=APP_ID, + instance_details={"applications": {}}, + n_previous_replica=0, + logs_tail=30, + ) + + assert status["version"] == RUNNING_VERSION + assert status["pinned_version"] == PINNED_VERSION + + +@pytest.mark.asyncio +async def test_status_reports_no_pin_for_an_unpinned_app() -> None: + manager = _make_manager(startup_applications=[]) + + status = await manager._get_app_status( + application_id=APP_ID, + instance_details={"applications": {}}, + n_previous_replica=0, + logs_tail=30, + ) + + assert status["pinned_version"] is None + + +@pytest.mark.asyncio +async def test_status_resolves_a_short_artifact_id_in_the_pin() -> None: + # Startup configs may name the artifact without its workspace prefix; the + # tracked app always carries the full form. + manager = _make_manager( + startup_applications=[ + { + "artifact_id": "model-runner", + "application_id": APP_ID, + "version": PINNED_VERSION, + } + ] + ) + + status = await manager._get_app_status( + application_id=APP_ID, + instance_details={"applications": {}}, + n_previous_replica=0, + logs_tail=30, + ) + + assert status["pinned_version"] == PINNED_VERSION diff --git a/tests/apps/test_status_app_missing_from_cluster.py b/tests/apps/test_status_app_missing_from_cluster.py index 76cdc99..f92edbd 100644 --- a/tests/apps/test_status_app_missing_from_cluster.py +++ b/tests/apps/test_status_app_missing_from_cluster.py @@ -67,6 +67,7 @@ def _make_manager(app_id: str, app_info: dict) -> AppsManager: manager = object.__new__(AppsManager) manager.logger = logging.getLogger("test") manager._deployed_applications = {app_id: app_info} + manager.startup_applications = [] ray_cluster = MagicMock() # No ProxyDeployment replicas exist once the head wiped Serve state;