From 521c011f9f76b34900c02aafe873a991688f1f77 Mon Sep 17 00:00:00 2001 From: nilsmechtel Date: Sat, 5 Sep 2026 20:32:32 +0200 Subject: [PATCH 1/6] fix(apps): key redeploy on source content and verify recovered apps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Delete-before-rebuild only fired when the requested version or artifact_id differed from the tracked pin. That misses the two cases the pin cannot express: deploy_app(version=None) inherits the existing pin, and a re-staged version keeps its string. Both left the warm replica serving the module it imported at first start. Move the check after the build, where the version is resolved and the synced source is fingerprinted, and compare that fingerprint. It stays before _check_resources so the freed reservation is visible. Recovered apps carry no built_app, so _verify_running_identities returned early and both version_verified and the monitor's self-heal were blind for exactly the apps most likely to be stale. The verification needs the spec only to name the entry deployment, so run it regardless and guard the delete and _fire_redeploy on built_app instead — deleting a recovered app would strand it. Also surface code_verified in get_app_status: report-only, since a systematic hash disagreement would otherwise loop the monitor. --- bioengine/_app/bootstrap.py | 42 +++++--- bioengine/apps/builder.py | 11 +- bioengine/apps/manager.py | 198 +++++++++++++++++++++++++++--------- 3 files changed, 188 insertions(+), 63 deletions(-) diff --git a/bioengine/_app/bootstrap.py b/bioengine/_app/bootstrap.py index 8dc6932c..542b397d 100644 --- a/bioengine/_app/bootstrap.py +++ b/bioengine/_app/bootstrap.py @@ -21,6 +21,7 @@ import importlib import inspect import sys +from pathlib import Path from typing import Any, Dict, List, Optional from bioengine._app.errors import ( @@ -76,6 +77,25 @@ def _purge_stale_source_modules(source_root: str) -> None: importlib.invalidate_caches() +def hash_source_tree(source: Path) -> str: + """Content hash of a materialised app source tree (bytecode caches excluded). + + A version string can't distinguish same-version-different-content; this can. + The submit task bakes it onto each user class as ``code_hash`` so a running + replica reports the code it actually loaded, and the introspect task returns + it so the worker can tell a real content change from an unchanged redeploy. + Both must hash the same way for those two values to be comparable. + """ + import hashlib + + hasher = hashlib.md5() + for path in sorted(source.rglob("*")): + if path.is_file() and "__pycache__" not in path.parts: + hasher.update(path.relative_to(source).as_posix().encode()) + hasher.update(path.read_bytes()) + return hasher.hexdigest()[:16] + + # ───────────────────────────── introspection ───────────────────────────── @@ -139,7 +159,7 @@ def introspect_app_in_ray_task( ) -> Dict[str, Any]: """Phase-1 Ray task: download user source and introspect it. - Returns ``{"spec": …}``. + Returns ``{"spec": …, "source_signature": …}``. The download uses the Hypha ``BIOENGINE_ARTIFACT_FILES_URL`` (+ optional ``_DOWNLOAD_TOKEN``) env vars via ``_ensure_source``. Replicas materialise @@ -177,7 +197,12 @@ def introspect_app_in_ray_task( _purge_stale_source_modules(src_str) spec = introspect_app(entry_id) - return {"spec": spec} + + # Fingerprint what was actually synced, so the worker can tell a real + # content change from a version string that stayed the same (a re-staged + # version, or a redeploy at "latest"). The spec alone can't: a changed + # method body leaves qualnames and schemas identical. + return {"spec": spec, "source_signature": hash_source_tree(source)} def _walk( @@ -505,18 +530,7 @@ def build_and_run_application( sys.path.insert(0, src_str) _purge_stale_source_modules(src_str) - # Content hash of the materialised source — the code identity we bake into - # each user class below so a running replica reports what it *actually* - # loaded (a version string can't distinguish same-version-different-content; - # a content hash can). Excludes bytecode caches. - import hashlib as _hashlib - - _src_hasher = _hashlib.md5() - for _p in sorted(Path(src_str).rglob("*")): - if _p.is_file() and "__pycache__" not in _p.parts: - _src_hasher.update(_p.relative_to(src_str).as_posix().encode()) - _src_hasher.update(_p.read_bytes()) - source_hash = _src_hasher.hexdigest()[:16] + source_hash = hash_source_tree(head_source) head_artifact_id = replica_env_vars.get("BIOENGINE_ARTIFACT_ID") handles: Dict[str, Any] = {} diff --git a/bioengine/apps/builder.py b/bioengine/apps/builder.py index 9293f2dc..6878b5d8 100644 --- a/bioengine/apps/builder.py +++ b/bioengine/apps/builder.py @@ -312,8 +312,9 @@ async def _introspect_via_ray_task( """Submit :func:`introspect_app_in_ray_task` as a Ray task. The task syncs the user package from Hypha (token in ``env_vars``), - walks the type-hint composition graph, and returns the ``{spec}`` - payload. We never touch the worker's filesystem. + walks the type-hint composition graph, and returns the + ``{spec, source_signature}`` payload. We never touch the worker's + filesystem. Strips the ``_BIOENGINE_SECRET_*`` keys from ``env_vars`` before passing into the task to keep secrets out of Ray's logs; the @@ -669,11 +670,13 @@ async def build( # 3. Introspect the user package via a Ray task — the task syncs the # source from Hypha and walks the @bioengine.app composition. Returns - # spec only; replicas sync their own source per file from Hypha. + # the spec plus a content hash of the synced source; replicas sync their + # own source per file from Hypha. introspect_result = await self._introspect_via_ray_task( entry_id, env_vars, runtime_env ) spec = introspect_result["spec"] + source_signature = introspect_result.get("source_signature") self.logger.info(f"Introspect task returned for '{application_id}'") # Sanity check: format_version round-trip. @@ -731,6 +734,7 @@ async def build( "proxy_service_token_ttl_seconds": proxy_service_token_ttl_seconds, "entry": entry_id, "spec_hash": spec_hash, + "source_signature": source_signature, "display_name": manifest["name"], "description": manifest["description"], "artifact_id": artifact_id, @@ -779,6 +783,7 @@ async def build( "name": manifest["name"], "description": manifest["description"], "version": version, + "source_signature": source_signature, "resources": required_resources, "authorized_users": effective_authorized_users, "available_methods": available_methods, diff --git a/bioengine/apps/manager.py b/bioengine/apps/manager.py index 783ca792..b5e61ff5 100644 --- a/bioengine/apps/manager.py +++ b/bioengine/apps/manager.py @@ -248,6 +248,12 @@ def __init__( # moment an app is observed healthy again. See monitor_applications. self._redeploy_backoff: Dict[str, Dict[str, Any]] = {} + # Application ids already warned about an identity mismatch the monitor + # cannot self-heal. Both such conditions persist until a human + # redeploys, and the monitor ticks every ~10s — without this the log + # fills with the same line for as long as the worker lives. + self._identity_warned: set = set() + # Timestamp of the last in-place Serve-controller-loss recovery sweep, # used to rate-limit re-firing while a redeploy's serve.run is still # bootstrapping the controller. See _recover_from_controller_loss. @@ -681,6 +687,7 @@ async def _undeploy_application( # Remove from internal tracking after all cleanup operations complete self._deployed_applications.pop(application_id, None) self._redeploy_backoff.pop(application_id, None) + self._identity_warned.discard(application_id) self.logger.info(f"Undeployment of application '{application_id}' completed.") def _project_replicas( @@ -854,7 +861,7 @@ async def _verify_running_identities( application_id: str, application_details: Dict[str, Any], expected_version: str, - ) -> Tuple[Optional[str], Optional[bool]]: + ) -> Tuple[Optional[str], Optional[bool], Optional[bool]]: """Cross-reference each live replica's baked identity against the deployed version, entirely off the data plane. @@ -866,16 +873,28 @@ async def _verify_running_identities( deployment reports its *baked* stale version here, so the worker can detect and force a real restart without ever issuing an in-band request. - Returns ``(running_version, version_verified)``: the entry deployment's - running replica version, and whether every live replica booted the - expected version. Either is ``None`` when no live replica has a known - identity yet, so the caller never acts on partial data. + Returns ``(running_version, version_verified, code_verified)``: the entry + deployment's running replica version, whether every live replica booted + the expected version, and whether every live replica booted the expected + source content. Each is ``None`` when it can't be determined, so the + caller never acts on partial data. + + ``code_verified`` catches the case a version string cannot: the same + version re-staged with different files. It is reported only — a false + value never triggers a delete, because a systematic hash disagreement + would turn the monitor's self-heal into a redeploy loop on a healthy app. + + A recovered app carries no ``built_app``, so the entry deployment can't + be named and ``running_version`` stays ``None`` — but the verification + itself covers every deployment and does not need the spec. Bailing out + early on a missing spec would leave apps that survived a worker restart + permanently unverified, which is exactly when a warm replica is most + likely to be running code the version pin no longer describes. """ - info = self._deployed_applications.get(application_id) - built_app = info.get("built_app") if info else None - spec = getattr(built_app, "spec", None) - if not spec: - return None, None + info = self._deployed_applications.get(application_id) or {} + built_app = info.get("built_app") + expected_signature = info.get("source_signature") + spec = getattr(built_app, "spec", None) or {} entry_cid = spec.get("entry_id") classes = spec.get("classes") or {} entry_name = ( @@ -893,10 +912,11 @@ async def _verify_running_identities( self.logger.debug( f"Could not read replica identities for '{application_id}': {exc}" ) - return None, None + return None, None, None running_version = None verified = True + code_verified = None checked = 0 for deployment_name, deployment_info in ( application_details.get("deployments") or {} @@ -911,11 +931,16 @@ async def _verify_running_identities( checked += 1 if ident.get("version") != expected_version: verified = False + running_hash = ident.get("code_hash") + if expected_signature is not None and running_hash is not None: + code_verified = (running_hash == expected_signature) and ( + code_verified is not False + ) if deployment_name == entry_name and running_version is None: running_version = ident.get("version") if not checked: - return running_version, None - return running_version, verified + return running_version, None, None + return running_version, verified, code_verified async def _get_app_status( self, @@ -976,12 +1001,19 @@ async def _get_app_status( # version — a stale reused replica reads as "healthy" otherwise. # ``running_version`` shows the entry replica's baked version; # ``version_verified`` reflects EVERY live replica (a reused replica of - # any deployment, not just the entry, counts as unverified). Both are - # None when they can't be determined. + # any deployment, not just the entry, counts as unverified), and + # ``code_verified`` does the same for the source content, catching a + # re-staged version the version string alone cannot. All are None when + # they can't be determined. running_version = None version_verified = None + code_verified = None if status == "RUNNING": - running_version, version_verified = await self._verify_running_identities( + ( + running_version, + version_verified, + code_verified, + ) = await self._verify_running_identities( application_id, application_details, application_info["version"] ) @@ -1005,6 +1037,7 @@ async def _get_app_status( "version": application_info["version"] or "latest", "running_version": running_version, "version_verified": version_verified, + "code_verified": code_verified, "recovered_app": application_info["recovered_app"], "status": status, "message": message, @@ -1200,6 +1233,7 @@ async def recover_deployed_applications(self) -> None: "description": app_data["description"], "artifact_id": app_data["artifact_id"], "version": app_data["version"], + "source_signature": app_data.get("source_signature"), "application_kwargs": app_data["application_kwargs"], "application_env_vars": app_data["application_env_vars"], "hypha_token": None, @@ -1403,10 +1437,47 @@ async def monitor_applications(self) -> None: application_details = (instance_details.get("applications") or {}).get( application_id, {} ) - _, version_verified = await self._verify_running_identities( + ( + _, + version_verified, + code_verified, + ) = await self._verify_running_identities( application_id, application_details, application_info["version"] ) + if version_verified is not False and code_verified is not False: + self._identity_warned.discard(application_id) + if code_verified is False and version_verified is not False: + # Same version, different files. Report only: acting on this + # would delete on every tick if the two hashes ever disagree + # systematically. Redeploy explicitly to clear it. + if application_id not in self._identity_warned: + self._identity_warned.add(application_id) + self.logger.warning( + f"Application '{application_id}' reports RUNNING at " + f"the expected version " + f"{application_info['version']!r}, but a live " + f"replica loaded different source content than the " + f"deployed bundle. Redeploy it to load the current " + f"source." + ) if version_verified is False: + if application_info.get("built_app") is None: + # Recovered app: deleting it would strand it, since the + # redeploy path has no built application to resubmit. + # Report the mismatch and leave it serving — + # get_app_status carries version_verified=False so the + # split-brain is visible instead of silent. + if application_id not in self._identity_warned: + self._identity_warned.add(application_id) + self.logger.warning( + f"Application '{application_id}' reports RUNNING " + f"but a live replica loaded a version != " + f"{application_info['version']!r}. It was " + f"recovered from a previous worker, so it cannot " + f"be rebuilt here — redeploy it explicitly to " + f"load the pinned version." + ) + continue self.logger.warning( f"Application '{application_id}' reports RUNNING but a " f"live replica loaded a version != requested " @@ -1491,6 +1562,19 @@ def _fire_redeploy( attempt: int, ) -> None: """Schedule a redeploy task and log the attempt number.""" + if application_info.get("built_app") is None: + # Recovered from a previous worker: there is nothing to resubmit, + # and _deploy_application would only raise on the None built_app. + # Say so once per attempt instead; recovery is the liveness + # backstop's pod cycle, or an explicit deploy_app by a user. + self.logger.warning( + f"Application '{application_id}' for artifact " + f"'{application_info['artifact_id']}' is unhealthy but was " + f"recovered from a previous worker and carries no built " + f"application; skipping auto-redeploy (attempt #{attempt}). " + f"Redeploy it explicitly to bring it back under this worker." + ) + return self.logger.warning( f"Application '{application_id}' for artifact " f"'{application_info['artifact_id']}' is unhealthy; triggering " @@ -2418,35 +2502,6 @@ async def deploy_app( f"version '{version}'; kwargs: {kwargs_str}; env_vars: {env_vars_str}" ) await self._cancel_deployment_process(application_id=application_id) - - # A content change (new version, or a different artifact under - # this application_id) must reach every replica. serve.run's - # in-place update can silently reuse replicas — at - # num_replicas=1 with no surge headroom (e.g. a single GPU) - # the old replica keeps serving stale in-memory code. Delete - # first so the slot frees and fresh replicas load the new - # source. Clear is_deployed so the monitor loop doesn't fire a - # redundant redeploy during the delete→rebuild window. - content_changed = ( - version != existing_app["version"] - or artifact_id != existing_app["artifact_id"] - ) - if content_changed: - existing_app["is_deployed"].clear() - try: - await self.ray_cluster.call_with_reconnect( - serve.delete, application_id - ) - self.logger.info( - f"Deleted Ray Serve application '{application_id}' " - f"before redeploy so replicas are recreated with " - f"the new source." - ) - except Exception as delete_err: - self.logger.error( - f"Error deleting Ray Serve application " - f"'{application_id}' before redeploy: {delete_err}" - ) else: # Create a new application self.logger.info( @@ -2514,6 +2569,56 @@ async def deploy_app( f"{nr}; must be >= 0." ) + # A content change (new version, different artifact, or the same + # version string re-staged with different files) must reach every + # replica. serve.run's in-place update can silently reuse replicas — + # at num_replicas=1 with no surge headroom (e.g. a single GPU) the + # old replica keeps serving the module it imported at first start. + # Delete first so the slot frees and fresh replicas load the new + # source. Clear is_deployed so the monitor loop doesn't fire a + # redundant redeploy during the delete→rebuild window. + # + # This runs AFTER the build, not before it, because only the build + # resolves the request to a concrete version and fingerprints the + # files it actually synced. Comparing the raw request instead means + # deploy_app(version=None) — which inherits the old pin above — + # always looks unchanged, and a re-staged version always looks + # unchanged. It stays BEFORE _check_resources so the old app's + # reservation is already released when free capacity is measured. + if is_update: + new_signature = app.metadata.get("source_signature") + old_signature = existing_app.get("source_signature") + content_changed = ( + artifact_id != existing_app["artifact_id"] + or app.metadata["version"] != existing_app["version"] + # Both known and different: files changed under one version. + # Either unknown: fall back to the identity check above + # rather than restarting a healthy app on a config-only + # update (apps recovered from a pre-0.16.6 worker carry no + # signature). + or ( + new_signature is not None + and old_signature is not None + and new_signature != old_signature + ) + ) + if content_changed: + existing_app["is_deployed"].clear() + try: + await self.ray_cluster.call_with_reconnect( + serve.delete, application_id + ) + self.logger.info( + f"Deleted Ray Serve application '{application_id}' " + f"before redeploy so replicas are recreated with " + f"the new source." + ) + except Exception as delete_err: + self.logger.error( + f"Error deleting Ray Serve application " + f"'{application_id}' before redeploy: {delete_err}" + ) + # Check resources before creating deployment task await self._check_resources( application_id=application_id, @@ -2535,6 +2640,7 @@ async def deploy_app( "description": app.metadata["description"], "artifact_id": artifact_id, "version": app.metadata["version"], + "source_signature": app.metadata.get("source_signature"), "application_kwargs": app.metadata["application_kwargs"], "application_env_vars": app.metadata["application_env_vars"], "hypha_token": hypha_token, From 6c4b1e7077db289914d6097ac82df12270df8f3b Mon Sep 17 00:00:00 2001 From: nilsmechtel Date: Sat, 5 Sep 2026 23:18:47 +0200 Subject: [PATCH 2/6] docs(apps): correct the content-check comment deploy_app(version=None) on an update inherits the existing pin, so the pin genuinely does not advance and taking the in-place path is correct there. The case the fingerprint catches is a version whose content changed underneath it. --- bioengine/apps/manager.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/bioengine/apps/manager.py b/bioengine/apps/manager.py index b5e61ff5..5e1faf79 100644 --- a/bioengine/apps/manager.py +++ b/bioengine/apps/manager.py @@ -2578,12 +2578,11 @@ async def deploy_app( # source. Clear is_deployed so the monitor loop doesn't fire a # redundant redeploy during the delete→rebuild window. # - # This runs AFTER the build, not before it, because only the build - # resolves the request to a concrete version and fingerprints the - # files it actually synced. Comparing the raw request instead means - # deploy_app(version=None) — which inherits the old pin above — - # always looks unchanged, and a re-staged version always looks - # unchanged. It stays BEFORE _check_resources so the old app's + # This runs AFTER the build because only the build fingerprints the + # files it actually synced. A version string cannot see content that + # changed underneath it, so a re-staged version took the in-place + # serve.run path while _ensure_source had already refreshed the + # source on disk. It stays BEFORE _check_resources so the old app's # reservation is already released when free capacity is measured. if is_update: new_signature = app.metadata.get("source_signature") From 06240e922f7e5e2e69ae656ea364e2ddc28e82cb Mon Sep 17 00:00:00 2001 From: nilsmechtel Date: Sat, 5 Sep 2026 23:18:55 +0200 Subject: [PATCH 3/6] chore(release): bump version to 0.16.6 --- bioengine/_version.py | 2 +- pyproject.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/bioengine/_version.py b/bioengine/_version.py index 355a3ad1..14928a2e 100644 --- a/bioengine/_version.py +++ b/bioengine/_version.py @@ -13,4 +13,4 @@ Must stay in lock-step with ``pyproject.toml``'s ``version`` field. The ``version-check.yml`` CI workflow enforces the match. """ -__version__ = "0.16.5" +__version__ = "0.16.6" diff --git a/pyproject.toml b/pyproject.toml index 8fafd8b7..6e42d88a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "bioengine" -version = "0.16.5" +version = "0.16.6" description = "BioEngine — CLI and SDK for deploying and calling AI model services on BioEngine workers" requires-python = ">=3.11" authors = [ From 6261e1404e7f4d42bccf89f0b092a4bdac51ecb9 Mon Sep 17 00:00:00 2001 From: nilsmechtel Date: Sat, 12 Sep 2026 12:34:11 +0200 Subject: [PATCH 4/6] Revert "chore(release): bump version to 0.16.6" This reverts commit 06240e922f7e5e2e69ae656ea364e2ddc28e82cb. --- bioengine/_version.py | 2 +- pyproject.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/bioengine/_version.py b/bioengine/_version.py index 14928a2e..355a3ad1 100644 --- a/bioengine/_version.py +++ b/bioengine/_version.py @@ -13,4 +13,4 @@ Must stay in lock-step with ``pyproject.toml``'s ``version`` field. The ``version-check.yml`` CI workflow enforces the match. """ -__version__ = "0.16.6" +__version__ = "0.16.5" diff --git a/pyproject.toml b/pyproject.toml index 6e42d88a..8fafd8b7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "bioengine" -version = "0.16.6" +version = "0.16.5" description = "BioEngine — CLI and SDK for deploying and calling AI model services on BioEngine workers" requires-python = ">=3.11" authors = [ From a053cb0a4e51a800a02597aed2a16e36f5c360b6 Mon Sep 17 00:00:00 2001 From: nilsmechtel Date: Sat, 12 Sep 2026 14:09:58 +0200 Subject: [PATCH 5/6] test(apps): pin the three stale-actor holes The change shipped without tests, which is how the second hole survived review: a version-string check that returns early for any app without a built_app skips exactly the apps most likely to be running old code. Fifteen tests over the three holes and the constraint that bounds them: - verification now runs without a spec, so a recovered app whose replica booted 2.7.0 under a 2.8.0 pin reads as version_verified False instead of None; - a re-staged version is caught by source_signature, which the version string cannot see, and a missing signature reads as unknown rather than as a mismatch so pre-0.16.6 recovered apps do not all warn on the first tick after an upgrade; - a recovered app found stale is reported and left serving, while an app this worker built is still deleted -- that self-heal must not regress; - a content mismatch never deletes, and its warning fires once rather than once per ~10s tick, but rearms after a clean tick. Every one of them fails against origin/main. --- tests/apps/test_stale_actor_holes.py | 364 +++++++++++++++++++++++++++ 1 file changed, 364 insertions(+) create mode 100644 tests/apps/test_stale_actor_holes.py diff --git a/tests/apps/test_stale_actor_holes.py b/tests/apps/test_stale_actor_holes.py new file mode 100644 index 00000000..f6c19c40 --- /dev/null +++ b/tests/apps/test_stale_actor_holes.py @@ -0,0 +1,364 @@ +"""Pin the three holes through which a stale actor kept serving. + +The worker's guard against a reused Ray Serve replica was a version-string +comparison, made only for apps this worker had itself built. That left three +ways for an actor running old code to be reported as healthy: + +1. A version string cannot see content that changed underneath it. Re-staging + the same version with different files took the in-place ``serve.run`` path, + which may reuse the replica, and nothing compared what the replica had + actually loaded. +2. ``_verify_running_identities`` returned early when the app carried no + ``built_app`` — exactly the apps recovered from a previous worker, which are + the most likely to be running code the version pin no longer describes. +3. The recovered app that *was* found stale was deleted anyway, even though the + redeploy path has nothing to resubmit for it, so the delete stranded it. + +The counterpart constraint: a content mismatch is reported, never acted on. If +the two hashes ever disagreed systematically, deleting on it would turn the +monitor into a redeploy loop against a healthy app. +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from bioengine._app.bootstrap import hash_source_tree +from bioengine.apps.manager import AppsManager + +APP_ID = "model-runner" +VERSION = "2.8.0" +SIGNATURE = "0123456789abcdef" + + +def _built_app(entry_deployment: str = "ModelRunner"): + return SimpleNamespace( + spec={"entry_id": "cid0", "classes": {"cid0": {"qualname": entry_deployment}}} + ) + + +def _details(*replicas: tuple[str, str, str]) -> dict: + """Serve instance details for one app: (deployment, replica_id, state).""" + deployments: dict = {} + for deployment, replica_id, state in replicas: + deployments.setdefault(deployment, {"replicas": []})["replicas"].append( + {"replica_id": replica_id, "state": state} + ) + return {"deployments": deployments} + + +def _make_manager(*, built_app=_built_app(), source_signature=SIGNATURE) -> AppsManager: + is_deployed = asyncio.Event() + is_deployed.set() + + manager = object.__new__(AppsManager) + manager.logger = logging.getLogger("test.holes") + manager._redeploy_backoff = {} + manager._identity_warned = set() + manager._deployed_applications = { + APP_ID: { + "is_deployed": is_deployed, + "artifact_id": "bioimage-io/model-runner", + "version": VERSION, + "source_signature": source_signature, + "auto_redeploy": True, + "built_app": built_app, + "deployment_task": None, + } + } + + ray_cluster = MagicMock() + ray_cluster.check_connection = AsyncMock() + ray_cluster.proxy_actor_handle.get_serve_instance_details.remote = AsyncMock( + return_value={} + ) + ray_cluster.proxy_actor_handle.get_replica_identities.remote = AsyncMock( + return_value={} + ) + manager.ray_cluster = ray_cluster + manager._deploy_application = AsyncMock() + + return manager + + +def _warnings(caplog) -> list[str]: + return [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING] + + +# ───────────────────── hole 2: verification without a spec ───────────────────── + + +@pytest.mark.asyncio +async def test_a_recovered_app_is_verified_even_without_a_built_app() -> None: + # The whole point: an app that survived a worker restart has no built_app, + # and used to skip verification entirely. Its replicas are precisely the + # ones that have been alive longest. + manager = _make_manager(built_app=None) + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": "2.7.0", "code_hash": SIGNATURE}} + } + + running_version, verified, _ = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "RUNNING")), VERSION + ) + + assert verified is False, ( + "A recovered app running the wrong version must be detectable; " + "returning None here is what let it serve stale code unnoticed." + ) + # No spec means the entry deployment cannot be named, so this stays unknown + # rather than guessing — but that does not block the verdict above. + assert running_version is None + + +@pytest.mark.asyncio +async def test_a_stale_replica_of_a_non_entry_deployment_fails_the_check() -> None: + # Serve may reuse a replica of any deployment, not just the entry one. + manager = _make_manager() + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": VERSION, "code_hash": SIGNATURE}}, + "Helper": {"r2": {"version": "2.7.0", "code_hash": SIGNATURE}}, + } + + running_version, verified, _ = await manager._verify_running_identities( + APP_ID, + _details(("ModelRunner", "r1", "RUNNING"), ("Helper", "r2", "RUNNING")), + VERSION, + ) + + assert running_version == VERSION # the entry replica alone looks fine + assert verified is False + + +@pytest.mark.asyncio +async def test_nothing_to_check_yields_unknown_not_verified() -> None: + # A replica that has not yet pushed its identity must not read as proof. + manager = _make_manager() + + _, verified, code_verified = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "STARTING")), VERSION + ) + + assert verified is None + assert code_verified is None + + +# ───────────────────── hole 1: same version, different files ───────────────────── + + +@pytest.mark.asyncio +async def test_a_restaged_version_is_caught_by_content_not_by_version() -> None: + manager = _make_manager() + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": VERSION, "code_hash": "deadbeefdeadbeef"}} + } + + _, verified, code_verified = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "RUNNING")), VERSION + ) + + assert verified is True, "The version string genuinely matches — that is the trap." + assert code_verified is False + + +@pytest.mark.asyncio +async def test_matching_content_verifies() -> None: + manager = _make_manager() + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": VERSION, "code_hash": SIGNATURE}} + } + + _, verified, code_verified = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "RUNNING")), VERSION + ) + + assert (verified, code_verified) == (True, True) + + +@pytest.mark.asyncio +async def test_an_app_from_an_older_worker_has_no_signature_to_compare() -> None: + # Apps recovered from a pre-0.16.6 worker carry no source_signature. That + # must read as unknown, not as a mismatch, or every one of them would warn + # on the first tick after an upgrade. + manager = _make_manager(built_app=None, source_signature=None) + manager.ray_cluster.proxy_actor_handle.get_replica_identities.remote.return_value = { + "ModelRunner": {"r1": {"version": VERSION, "code_hash": "deadbeefdeadbeef"}} + } + + _, verified, code_verified = await manager._verify_running_identities( + APP_ID, _details(("ModelRunner", "r1", "RUNNING")), VERSION + ) + + assert verified is True + assert code_verified is None + + +# ───────────────────── hole 3: what the monitor does about it ───────────────────── + + +def _running_status(): + return SimpleNamespace( + applications={APP_ID: SimpleNamespace(status=SimpleNamespace(value="RUNNING"))} + ) + + +def _wire_monitor(manager: AppsManager, verdict: tuple) -> list: + """Route serve.status through call_with_reconnect; record serve.delete calls.""" + deletes: list = [] + + async def call_with_reconnect(fn, *args): + name = getattr(fn, "__name__", "") + if name == "delete": + deletes.append(args[0]) + return None + return _running_status() + + manager.ray_cluster.call_with_reconnect = AsyncMock(side_effect=call_with_reconnect) + manager._verify_running_identities = AsyncMock(return_value=verdict) + return deletes + + +@pytest.mark.asyncio +async def test_a_recovered_app_with_a_version_mismatch_is_reported_not_deleted( + caplog, +) -> None: + manager = _make_manager(built_app=None) + deletes = _wire_monitor(manager, (None, False, None)) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + await manager.monitor_applications() + + assert deletes == [], ( + "Deleting a recovered app strands it: _fire_redeploy has no built " + "application to resubmit, so nothing would bring it back." + ) + assert "recovered from a previous worker" in _warnings(caplog)[0] + + +@pytest.mark.asyncio +async def test_a_managed_app_with_a_version_mismatch_is_still_deleted() -> None: + # The pre-existing self-heal must not regress: an app this worker built can + # be deleted, because the next tick redeploys it from the cached built_app. + manager = _make_manager() + deletes = _wire_monitor(manager, (None, False, None)) + + await manager.monitor_applications() + + assert deletes == [APP_ID] + + +@pytest.mark.asyncio +async def test_a_content_mismatch_warns_but_never_deletes(caplog) -> None: + manager = _make_manager() + deletes = _wire_monitor(manager, (VERSION, True, False)) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + await manager.monitor_applications() + + assert deletes == [], ( + "A hash disagreement that turned out to be systematic would delete the " + "app on every tick. Report only." + ) + assert "different source content" in _warnings(caplog)[0] + + +@pytest.mark.asyncio +async def test_the_identity_warning_does_not_repeat_every_tick(caplog) -> None: + # The condition persists until a human redeploys and the monitor ticks every + # ~10 s, so without de-duplication this is one line per tick for the life of + # the worker. + manager = _make_manager() + _wire_monitor(manager, (VERSION, True, False)) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + for _ in range(5): + await manager.monitor_applications() + + assert len(_warnings(caplog)) == 1 + + +@pytest.mark.asyncio +async def test_the_warning_rearms_once_the_app_verifies_again(caplog) -> None: + # Suppression may not be permanent: a later, genuinely new mismatch after a + # clean period has to be visible. + manager = _make_manager() + _wire_monitor(manager, (VERSION, True, False)) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + await manager.monitor_applications() + manager._verify_running_identities = AsyncMock(return_value=(VERSION, True, True)) + await manager.monitor_applications() + manager._verify_running_identities = AsyncMock( + return_value=(VERSION, True, False) + ) + await manager.monitor_applications() + + assert len(_warnings(caplog)) == 2 + + +@pytest.mark.asyncio +async def test_a_recovered_app_is_never_auto_redeployed(caplog) -> None: + # _deploy_application would only raise on the None built_app; say so once + # per attempt instead of scheduling a task that cannot succeed. + manager = _make_manager(built_app=None) + + with caplog.at_level(logging.WARNING, logger="test.holes"): + manager._fire_redeploy(APP_ID, manager._deployed_applications[APP_ID], attempt=1) + await asyncio.sleep(0) + + manager._deploy_application.assert_not_awaited() + assert "skipping auto-redeploy" in _warnings(caplog)[0] + + +@pytest.mark.asyncio +async def test_undeploying_clears_the_warning_marker() -> None: + # Otherwise an application_id redeployed under the same name would inherit + # the previous instance's suppression and never warn. + manager = _make_manager() + manager._identity_warned.add(APP_ID) + _wire_monitor(manager, (VERSION, True, True)) + + await manager.monitor_applications() + + assert APP_ID not in manager._identity_warned + + +# ───────────────────── the fingerprint both sides must agree on ───────────────────── + + +def test_the_source_hash_tracks_content_and_ignores_bytecode(tmp_path: Path) -> None: + # The submit task bakes this onto each replica and the introspect task + # returns it to the worker; the two are only comparable if they hash the + # same way, which is why there is one function rather than two copies. + source = tmp_path / "src" + (source / "pkg").mkdir(parents=True) + (source / "pkg" / "main.py").write_text("x = 1\n") + + baseline = hash_source_tree(source) + + (source / "pkg" / "__pycache__").mkdir() + (source / "pkg" / "__pycache__" / "main.cpython-311.pyc").write_bytes(b"\x00\x01") + assert hash_source_tree(source) == baseline + + (source / "pkg" / "main.py").write_text("x = 2\n") + assert hash_source_tree(source) != baseline, ( + "A changed method body leaves qualnames and schemas identical — the " + "content hash is the only thing that sees it." + ) + + +def test_the_source_hash_tracks_file_names_not_just_bytes(tmp_path: Path) -> None: + a = tmp_path / "a" + b = tmp_path / "b" + a.mkdir() + b.mkdir() + (a / "one.py").write_text("x = 1\n") + (b / "two.py").write_text("x = 1\n") + + assert hash_source_tree(a) != hash_source_tree(b) From 5ab02fd8a187fb71761fb94d6d7f1ec18f6721ac Mon Sep 17 00:00:00 2001 From: nilsmechtel Date: Sat, 12 Sep 2026 14:12:20 +0200 Subject: [PATCH 6/6] test(apps): drive the redeploy skip through the monitor Calling _fire_redeploy directly pinned the test to that method's arity, which a sibling change to the same monitor widens with a `reason`. The public path is monitor_applications seeing UNHEALTHY, and testing it there is both signature-agnostic and closer to what actually happens. Also seeds the two monitor bookkeeping attributes the fixture omitted. --- tests/apps/test_stale_actor_holes.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/tests/apps/test_stale_actor_holes.py b/tests/apps/test_stale_actor_holes.py index f6c19c40..ea6b2a4f 100644 --- a/tests/apps/test_stale_actor_holes.py +++ b/tests/apps/test_stale_actor_holes.py @@ -61,6 +61,10 @@ def _make_manager(*, built_app=_built_app(), source_signature=SIGNATURE) -> Apps manager.logger = logging.getLogger("test.holes") manager._redeploy_backoff = {} manager._identity_warned = set() + # Bookkeeping the monitor keeps for its other branches; empty is the + # nothing-pending state these tests want. + manager._missing_from_status = {} + manager._deleted_pending_redeploy = set() manager._deployed_applications = { APP_ID: { "is_deployed": is_deployed, @@ -202,13 +206,13 @@ async def test_an_app_from_an_older_worker_has_no_signature_to_compare() -> None # ───────────────────── hole 3: what the monitor does about it ───────────────────── -def _running_status(): +def _status(state: str = "RUNNING"): return SimpleNamespace( - applications={APP_ID: SimpleNamespace(status=SimpleNamespace(value="RUNNING"))} + applications={APP_ID: SimpleNamespace(status=SimpleNamespace(value=state))} ) -def _wire_monitor(manager: AppsManager, verdict: tuple) -> list: +def _wire_monitor(manager: AppsManager, verdict: tuple, state: str = "RUNNING") -> list: """Route serve.status through call_with_reconnect; record serve.delete calls.""" deletes: list = [] @@ -217,7 +221,7 @@ async def call_with_reconnect(fn, *args): if name == "delete": deletes.append(args[0]) return None - return _running_status() + return _status(state) manager.ray_cluster.call_with_reconnect = AsyncMock(side_effect=call_with_reconnect) manager._verify_running_identities = AsyncMock(return_value=verdict) @@ -307,9 +311,10 @@ async def test_a_recovered_app_is_never_auto_redeployed(caplog) -> None: # _deploy_application would only raise on the None built_app; say so once # per attempt instead of scheduling a task that cannot succeed. manager = _make_manager(built_app=None) + _wire_monitor(manager, (None, None, None), state="UNHEALTHY") with caplog.at_level(logging.WARNING, logger="test.holes"): - manager._fire_redeploy(APP_ID, manager._deployed_applications[APP_ID], attempt=1) + await manager.monitor_applications() await asyncio.sleep(0) manager._deploy_application.assert_not_awaited()