diff --git a/README.md b/README.md index 7e3c9d1..e752761 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,7 @@ index names: "version": "1.1.0", "biolink_model_toolkit_version": "v4.4.3", "babel": { + "checked_at": "2026-10-23T12:00:00+00:00", "default_version": "2026jul22", "available_versions": { "2025sep1": "https://github.com/NCATSTranslator/Babel/blob/main/releases/2025sep1/README.md", @@ -116,15 +117,16 @@ index names: ``` Index names in this example are illustrative. Each `indices` array contains the -concrete targets selected at startup and pinned until restart. The default release +concrete targets selected by the latest successful discovery snapshot. The default release uses every backing index of its configured alias; other releases select one index. Available releases come from Elasticsearch metadata, rather than a configured version list; these maps do not indicate backend health. ### Discovering available releases -At startup, the server uses its existing `ES_HOST` client to discover release indices -from `mappings._meta.src.nodenorm` metadata on the associated cluster. No version list, +At startup and every **10 minutes** in the background, the server uses its existing +`ES_HOST` client to discover release indices from `mappings._meta.src.nodenorm` metadata +on the associated cluster. No version list, version-to-index mapping, or per-release alias is required. The client's Elasticsearch permissions must allow reading mappings across that cluster. Discovery uses an explicit `version` field, or a dated release tag in the legacy source URL; it does not infer @@ -145,17 +147,32 @@ release and its exact concrete backing indices. All targets must identify the sa Babel release. The default keeps those targets even if newer builds exist elsewhere on the cluster; a multi-index default queries all its backing indices. Explicitly requesting the default Babel version uses those same alias targets. After the alias -moves to another release and the API restarts, the former default becomes a supported +moves to another release and discovery refreshes, the former default becomes a supported nondefault release and uses the build-date rule above. -The default release has no separate Babel configuration override. Configuration and -metadata discovery are loaded at startup. Adding a release or a newer build makes -it available after restart; adding a release does not automatically change the -default. To change the default, move the configured alias (or change the configured -default target) and restart. - -All selected concrete indices stay pinned until restart, so moving an alias cannot -change queries in a running process. Keep indices serving existing processes until -the restart rollout completes; removing an index earlier makes requests using it fail. -Verify `/status` and sample results before rollout; discovery relies on metadata -accurately describing each index's contents. +The default release has no separate Babel configuration override. Configuration is +loaded at startup. Adding a release or a newer build makes it available on the next +successful background refresh; adding a release does not automatically change the +default. Moving the configured default alias takes effect on the next successful +refresh, without restarting the API. Changing the configuration itself requires a restart. + +Each process refreshes independently, even without incoming requests. A refresh has a +10-second total deadline. It publishes the new release registry and default query +indices together, so `/status` and new requests use the same snapshot. Each request +keeps its selected concrete indices throughout processing, including conflation lookups. +Keep previous indices available until every process has refreshed and requests using +the previous snapshot have completed. Verify `/status` and sample results before +removing indices; discovery relies on metadata accurately describing each index's contents. + +`babel.checked_at` is the UTC time of the last successful discovery. A failed refresh +retains that snapshot and timestamp, adds `babel.refresh_error` with code +`discovery_failed`, and retries after 30 seconds. A successful retry clears the error. +Discovery failures alone do not determine service health. + +`GET /status` checks that every selected default index still exposes the expected +Babel source version. These separate health results are cached for **5 seconds** on +success and **1 second** on failure, with concurrent requests sharing one check and a +3-second total deadline. Changing the default version or its concrete indices causes +the next status request to use a new health check. A failed check returns HTTP 503; +an expired healthy result is never served after a failed check. Health checks do not +trigger Babel discovery. All status responses send `Cache-Control: no-store`. diff --git a/src/nodenorm/babel.py b/src/nodenorm/babel.py index 51a1242..d7959c0 100644 --- a/src/nodenorm/babel.py +++ b/src/nodenorm/babel.py @@ -80,6 +80,16 @@ def __init__(self, default_version: str, releases: dict, unavailable: dict = Non self._releases = {version: BabelRelease(version, index) for version, index in releases.items()} self._unavailable = dict(unavailable) + @classmethod + def from_default_mappings(cls, default_mappings, mappings): + """Build one registry from a validated default target and a cluster scan.""" + versions = {get_babel_version(mapping) for mapping in default_mappings.values()} + if len(versions) != 1: + raise ValueError("The default Elasticsearch target must contain exactly one Babel release") + backing_indices = tuple(default_mappings) + default_index = backing_indices[0] if len(backing_indices) == 1 else backing_indices + return cls.from_mappings(versions.pop(), {**mappings, **default_mappings}, default_index=default_index) + @classmethod def from_mappings(cls, default_version, mappings, default_index=None): """Discover release names and targets from source metadata on concrete indices.""" diff --git a/src/nodenorm/handlers/status.py b/src/nodenorm/handlers/status.py index 8c55650..8838dfe 100644 --- a/src/nodenorm/handlers/status.py +++ b/src/nodenorm/handlers/status.py @@ -2,12 +2,13 @@ import json from functools import lru_cache -from elasticsearch import AsyncElasticsearch +from tornado.web import HTTPError import nodenorm -from nodenorm.babel import get_babel_release_url, get_babel_version +from nodenorm.babel import get_babel_release_url from nodenorm.biolink import BIOLINK_MODEL_VERSION from nodenorm.handlers.base import NodeNormalizationBaseHandler +from nodenorm.status_health import StatusHealthCache @lru_cache(maxsize=None) @@ -23,23 +24,32 @@ class NodeNormHealthHandler(NodeNormalizationBaseHandler): name = "health" - async def get(self): - async_client: AsyncElasticsearch = self.biothings.elasticsearch.async_client - search_indices = self.biothings.elasticsearch.indices - - mapping_response = await async_client.indices.get_mapping(index=search_indices) - index_mapping = next(iter(mapping_response.values())) - # Validate live default metadata while reporting the releases resolved at startup. - get_babel_version(index_mapping) + def set_default_headers(self): + super().set_default_headers() + self.set_header("Cache-Control", "no-store") - releases = self.biothings.elasticsearch.babel + async def get(self): + elasticsearch = self.biothings.elasticsearch + releases = elasticsearch.babel babel_status = { "default_version": releases.default_version, "available_versions": {version: get_babel_release_url(version) for version in releases.available_versions}, "indices": {version: releases.resolve(version).search_indices for version in releases.available_versions}, + "checked_at": elasticsearch.babel_checked_at, } if releases.errors: babel_status["errors"] = releases.errors + refresh_error = elasticsearch.babel_refresh_error + if refresh_error is not None: + babel_status["refresh_error"] = refresh_error + + release = releases.resolve() + health = getattr(elasticsearch, "status_health_cache", None) + if health is None or health.configuration != (release.version, tuple(release.search_indices)): + health = StatusHealthCache(elasticsearch.async_client, release) + elasticsearch.status_health_cache = health + if not await health.check(): + raise HTTPError(503, reason="Elasticsearch health check failed") status_response = { "status": "running", diff --git a/src/nodenorm/namespace.py b/src/nodenorm/namespace.py index 08d642f..2aa518a 100644 --- a/src/nodenorm/namespace.py +++ b/src/nodenorm/namespace.py @@ -1,3 +1,4 @@ +import asyncio import importlib.resources import importlib.util import json @@ -7,6 +8,8 @@ import re import sys import types +from contextlib import suppress +from datetime import datetime, timezone import tornado.options @@ -14,10 +17,13 @@ from biothings.web.services.metadata import BiothingsESMetadata import nodenorm -from nodenorm.babel import BabelReleases, get_babel_version +from nodenorm.babel import BabelReleases logger = logging.getLogger(__name__) +BABEL_REFRESH_INTERVAL = 600 +BABEL_REFRESH_RETRY_INTERVAL = 30 +BABEL_REFRESH_TIMEOUT = 10 class NodeNormalizationAPINamespace: @@ -83,27 +89,71 @@ def configure_elasticsearch(self) -> types.SimpleNamespace: elasticsearch_namespace.async_client = connections.es.get_async_client( elasticsearch_configuration["ES_HOST"], **elasticsearch_configuration["ES_ARGS"] ) - # Preserve the alias's exact targets for the default release, even when - # the cluster contains a newer build of that Babel release. configured_indices = self._validate_elasticsearch_index(elasticsearch_namespace) - default_mappings = elasticsearch_namespace.client.indices.get_mapping(index=configured_indices) - versions = {get_babel_version(mapping) for mapping in default_mappings.values()} - if len(versions) != 1: - raise ValueError("The default Elasticsearch target must contain exactly one Babel release") - default_version = versions.pop() - backing_indices = tuple(default_mappings) - default_index = backing_indices[0] if len(backing_indices) == 1 else backing_indices - mappings = elasticsearch_namespace.client.indices.get_mapping(filter_path="*.mappings._meta") - # Preserve targets resolved above if the cluster-wide scan omits them. - # Prefer the scan's metadata when an index is present in both responses. - releases = BabelReleases.from_mappings( - default_version, {**default_mappings, **mappings}, default_index=default_index + default_mappings = elasticsearch_namespace.client.indices.get_mapping( + index=configured_indices, filter_path="*.mappings._meta" ) - # Pin every selected concrete index until restart. + mappings = elasticsearch_namespace.client.indices.get_mapping(filter_path="*.mappings._meta") + releases = BabelReleases.from_default_mappings(default_mappings, mappings) elasticsearch_namespace.indices = releases.resolve().search_indices elasticsearch_namespace.babel = releases + elasticsearch_namespace.babel_checked_at = datetime.now(timezone.utc).isoformat() + elasticsearch_namespace.babel_refresh_error = None return elasticsearch_namespace + async def _discover_babel_releases(self): + client = self.elasticsearch.async_client + configuration = self.config.elasticsearch + index = configuration["ES_INDEX"] + alias = configuration["ES_ALIAS"] + if index and await client.indices.exists(index=index): + target = index + elif await client.indices.exists_alias(name=alias): + target = alias + else: + raise RuntimeError("Unable to validate nodenorm elasticsearch index / alias") + default_mappings = await client.indices.get_mapping(index=[target], filter_path="*.mappings._meta") + mappings = await client.indices.get_mapping(filter_path="*.mappings._meta") + return BabelReleases.from_default_mappings(default_mappings, mappings) + + async def refresh_babel_releases(self): + """Publish a complete discovery snapshot, retaining the previous one on failure.""" + try: + releases = await asyncio.wait_for(self._discover_babel_releases(), timeout=BABEL_REFRESH_TIMEOUT) + except Exception: + logger.exception("Unable to refresh Babel releases; retaining the previous snapshot") + self.elasticsearch.babel_refresh_error = {"code": "discovery_failed"} + return False + + # There is no await between these writes: handlers see one complete + # snapshot, and requests that already selected a release keep that object. + self.elasticsearch.babel = releases + self.elasticsearch.indices = releases.resolve().search_indices + self.elasticsearch.babel_checked_at = datetime.now(timezone.utc).isoformat() + self.elasticsearch.babel_refresh_error = None + return True + + async def _run_babel_refresh(self): + delay = BABEL_REFRESH_INTERVAL + while True: + await asyncio.sleep(delay) + refreshed = await self.refresh_babel_releases() + delay = BABEL_REFRESH_INTERVAL if refreshed else BABEL_REFRESH_RETRY_INTERVAL + + def start_babel_refresh(self): + """Start one background discovery task on the server's running event loop.""" + task = getattr(self, "_babel_refresh_task", None) + if task is None or task.done(): + self._babel_refresh_task = asyncio.create_task(self._run_babel_refresh()) + + async def stop_babel_refresh(self): + task = getattr(self, "_babel_refresh_task", None) + if task is not None: + task.cancel() + with suppress(asyncio.CancelledError): + await task + self._babel_refresh_task = None + def _validate_elasticsearch_index(self, elasticsearch_namespace: types.SimpleNamespace) -> dict: """Validates the elasticsearch index / alias. diff --git a/src/nodenorm/server.py b/src/nodenorm/server.py index fa727ee..90e3905 100644 --- a/src/nodenorm/server.py +++ b/src/nodenorm/server.py @@ -61,11 +61,15 @@ def start(self): pformat(self.namespace.handlers, width=200), ) + loop = tornado.ioloop.IOLoop.instance() try: - loop = tornado.ioloop.IOLoop.instance() + loop.add_callback(self.namespace.start_babel_refresh) loop.start() except Exception as gen_exc: logger.exception(gen_exc) raise gen_exc finally: - loop.close() + try: + loop.run_sync(self.namespace.stop_babel_refresh) + finally: + loop.close() diff --git a/src/nodenorm/status_health.py b/src/nodenorm/status_health.py new file mode 100644 index 0000000..666139e --- /dev/null +++ b/src/nodenorm/status_health.py @@ -0,0 +1,48 @@ +"""Short health caching, independent of the ten-minute Babel discovery snapshot.""" + +import asyncio +import logging +from time import monotonic + +from nodenorm.babel import get_babel_version + + +logger = logging.getLogger(__name__) +HEALTH_CACHE_TTL = 5 +HEALTH_FAILURE_TTL = 1 +HEALTH_TIMEOUT = 3 + + +class StatusHealthCache: + def __init__(self, client, release): + self.client = client + self.configuration = (release.version, tuple(release.search_indices)) + self._healthy = None + self._expires_at = 0 + self._refresh = None + + async def check(self): + if self._healthy is not None and monotonic() < self._expires_at: + return self._healthy + if self._refresh is None or self._refresh.done(): + self._refresh = asyncio.create_task(self._load()) + # A disconnected probe must not cancel the check shared by other callers. + return await asyncio.shield(self._refresh) + + async def _probe(self): + version, indices = self.configuration + mappings = await self.client.indices.get_mapping( + index=list(indices), filter_path="*.mappings._meta.src.nodenorm" + ) + return set(mappings) == set(indices) and all( + get_babel_version(mapping) == version for mapping in mappings.values() + ) + + async def _load(self): + try: + self._healthy = await asyncio.wait_for(self._probe(), timeout=HEALTH_TIMEOUT) + except Exception: + logger.warning("NodeNorm Elasticsearch health check failed", exc_info=True) + self._healthy = False + self._expires_at = monotonic() + (HEALTH_CACHE_TTL if self._healthy else HEALTH_FAILURE_TTL) + return self._healthy diff --git a/src/nodenorm/webapp/openapi.json b/src/nodenorm/webapp/openapi.json index 8698873..2aa8c92 100644 --- a/src/nodenorm/webapp/openapi.json +++ b/src/nodenorm/webapp/openapi.json @@ -30,11 +30,16 @@ "get": { "summary": "Status information on this NodeNorm instance", "tags": ["health"], - "description": "Returns service status and version information for this NodeNorm instance. The babel object contains the default release name, available release names mapped to their release URLs, selected concrete Elasticsearch indices for each available release, and optional discovery errors. Discovery runs at startup using the configured Elasticsearch client and requires no configured version list. The default uses the exact concrete backing indices of its configured alias, including older builds. The default release has no separate Babel configuration override. Other supported releases select the index with the newest valid mappings._meta.build_date, compared in UTC with naive timestamps treated as UTC. Missing or malformed dates rank below valid dates; ties use the lexicographically greatest index name. Explicit selection of the default version uses the same alias targets. Release targets are pinned to concrete indices until restart; moving the default alias takes effect after restart. Adding a release does not automatically change the default.", + "description": "Returns service status and version information for this NodeNorm instance. Babel discovery runs at startup and every 600 seconds in the background, independently of incoming requests. It uses the configured Elasticsearch client and requires no configured version list. Successful discovery publishes the release registry and query indices together. Moving the configured default alias takes effect on the next successful refresh without restarting; each in-flight request keeps its selected concrete indices. The default uses every exact concrete backing index of its configured alias, including older builds, and has no separate Babel configuration override. Explicit selection of the default version uses those same targets. Other releases select the newest valid mappings._meta.build_date, compared in UTC with naive timestamps treated as UTC; missing or malformed dates rank below valid dates and ties use index name descending. Adding a release does not automatically change the default. Discovery has a 10-second total deadline; failure retains the previous snapshot and checked_at timestamp, exposes refresh_error, and retries after 30 seconds. Separate health checks validate the expected Babel source version on every default index, with a 3-second deadline, a 5-second success cache and a 1-second failure cache. Concurrent status requests share a health check. A changed default version or indices triggers a new check. Health failure returns HTTP 503. All status responses send Cache-Control: no-store.", "operationId": "status", "responses": { "200": { "description": "Successful Response", + "headers": { + "Cache-Control": { + "schema": {"type": "string", "const": "no-store"} + } + }, "content": { "application/json": { "schema": { @@ -74,7 +79,8 @@ "required": [ "default_version", "available_versions", - "indices" + "indices", + "checked_at" ], "properties": { "default_version": { @@ -103,12 +109,26 @@ "type": "string" } }, - "description": "Babel release names mapped to arrays of concrete Elasticsearch indices selected for requests at startup and pinned until restart. Has the same release keys as babel.available_versions. The default preserves every exact alias target even if older than other builds; explicit requests for the default version use those same targets. Other releases select one index by newest valid metadata build_date, with index name descending as the final tie-break. This map does not indicate backend health. Example index names are illustrative.", + "description": "Babel release names mapped to arrays of concrete Elasticsearch indices selected by the latest successful discovery snapshot, refreshed every 600 seconds in the background. Has the same release keys as babel.available_versions. The default preserves every exact alias target even if older than other builds; explicit requests for the default version use those same targets. Other releases select one index by newest valid metadata build_date, with index name descending as the final tie-break. Requests keep their selected targets throughout processing. This map does not indicate backend health. Example index names are illustrative.", "example": { "2025sep1": ["nodenorm_2025sep1"], "2026jul22": ["nodenorm_2026jul22"] } }, + "checked_at": { + "type": "string", + "format": "date-time", + "description": "UTC time of the last successful Babel discovery. Retained when a refresh fails; separate health checks do not change it.", + "example": "2026-10-23T12:00:00+00:00" + }, + "refresh_error": { + "type": "object", + "required": ["code"], + "properties": { + "code": {"type": "string", "const": "discovery_failed"} + }, + "description": "Present after a failed background discovery. The last valid registry remains active and discovery retries after 30 seconds. Cleared by a successful refresh. Contains no upstream connection details." + }, "errors": { "type": "object", "additionalProperties": { @@ -125,6 +145,14 @@ } } } + }, + "503": { + "description": "The default Elasticsearch health check failed or exceeded its 3-second deadline. Failed checks are cached for 1 second; an expired success is not served after failure.", + "headers": { + "Cache-Control": { + "schema": {"type": "string", "const": "no-store"} + } + } } } } diff --git a/tests/test_babel.py b/tests/test_babel.py index d868998..d3332d1 100644 --- a/tests/test_babel.py +++ b/tests/test_babel.py @@ -113,7 +113,9 @@ def test_legacy_configuration_resolves_configured_target_to_concrete_index(index assert elasticsearch.babel.available_versions == ["2025sep1", "2026jul22"] connections.get_client.assert_called_once_with("http://test:9200") connections.get_async_client.assert_called_once_with("http://test:9200") - assert client.indices.get_mapping.call_args_list[0].kwargs == {"index": [target]} + assert client.indices.get_mapping.call_args_list[0].kwargs == { + "index": [target], "filter_path": "*.mappings._meta" + } assert client.indices.get_mapping.call_args_list[1].kwargs == {"filter_path": "*.mappings._meta"} assert namespace.config.elasticsearch["ES_INDEX"] == "legacy" assert namespace.config.elasticsearch["ES_ALIAS"] == "alias" @@ -139,6 +141,7 @@ def test_alias_rollover_preserves_its_exact_default_and_selects_latest_builds_fo def get_mapping(*, index=None, filter_path=None): if index is not None: assert index == ["alias"] + assert filter_path == "*.mappings._meta" return {alias_index: all_mappings[alias_index]} assert filter_path == "*.mappings._meta" return all_mappings @@ -216,7 +219,9 @@ def test_multi_index_alias_preserves_all_default_backing_indices_despite_newer_b assert elasticsearch.babel.resolve("2025sep1").search_indices == ["old-a", "old-b"] assert elasticsearch.babel.resolve("2026jul22").index == "new-release-build" assert elasticsearch.babel.errors == {} - assert client.indices.get_mapping.call_args_list[0].kwargs == {"index": ["alias"]} + assert client.indices.get_mapping.call_args_list[0].kwargs == { + "index": ["alias"], "filter_path": "*.mappings._meta" + } assert client.indices.get_mapping.call_args_list[1].kwargs == {"filter_path": "*.mappings._meta"} diff --git a/tests/test_babel_endpoints.py b/tests/test_babel_endpoints.py index 1ea9b9b..8646181 100644 --- a/tests/test_babel_endpoints.py +++ b/tests/test_babel_endpoints.py @@ -1,16 +1,18 @@ """Offline endpoint coverage for request-scoped Babel release selection.""" +import asyncio import importlib.util import json import sys from pathlib import Path from types import ModuleType, SimpleNamespace -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, patch -from tornado.testing import AsyncHTTPTestCase +from tornado.testing import AsyncHTTPTestCase, gen_test from tornado.web import Application from nodenorm.babel import BabelReleases +from nodenorm.namespace import NodeNormalizationAPINamespace DEFAULT_VERSION = "2026jul22" @@ -122,6 +124,8 @@ def get_app(self): async_client=self.client, indices=[DEFAULT_INDEX], babel=BabelReleases(DEFAULT_VERSION, {OLD_VERSION: OLD_INDEX, DEFAULT_VERSION: DEFAULT_INDEX}), + babel_checked_at="2026-10-23T12:00:00+00:00", + babel_refresh_error=None, ) ) app = Application( @@ -297,6 +301,7 @@ def test_status_and_requests_share_metadata_discovered_releases(self): DEFAULT_VERSION: f"https://github.com/NCATSTranslator/Babel/blob/main/releases/{DEFAULT_VERSION}/README.md", }, "indices": {OLD_VERSION: [OLD_INDEX], DEFAULT_VERSION: [DEFAULT_INDEX]}, + "checked_at": "2026-10-23T12:00:00+00:00", } assert set(body) == {"status", "deployment_environment", "version", "biolink_model_toolkit_version", "babel"} assert body["deployment_environment"] == "unknown" @@ -304,7 +309,9 @@ def test_status_and_requests_share_metadata_discovered_releases(self): assert body["version"] == handlers.status.get_openapi_version() assert body["biolink_model_toolkit_version"] == "test" assert "elasticsearch" not in body - assert self.client.calls == [("get_mapping", {"index": [DEFAULT_INDEX]})] + assert self.client.calls == [ + ("get_mapping", {"index": [DEFAULT_INDEX], "filter_path": "*.mappings._meta.src.nodenorm"}) + ] selected = self.assert_selected_release( self.fetch(f"/get_setid?curie=INPUT:1&babel_version={OLD_VERSION}"), OLD_VERSION @@ -334,7 +341,9 @@ def mapping(version, build_date): assert set(body["babel"]["available_versions"]) == {OLD_VERSION, DEFAULT_VERSION} assert body["babel"]["indices"] == {OLD_VERSION: [OLD_INDEX], DEFAULT_VERSION: [DEFAULT_INDEX]} assert "errors" not in body["babel"] - assert self.client.calls == [("get_mapping", {"index": [DEFAULT_INDEX]})] + assert self.client.calls == [ + ("get_mapping", {"index": [DEFAULT_INDEX], "filter_path": "*.mappings._meta.src.nodenorm"}) + ] self.client.calls.clear() selected = self.assert_selected_release( @@ -407,7 +416,7 @@ def mapping(version, build_date): body = json.loads(response.body) assert body["babel"]["indices"] == {OLD_VERSION: [OLD_INDEX], DEFAULT_VERSION: default_indices} assert set(body["babel"]["indices"]) == set(body["babel"]["available_versions"]) - get_mapping.assert_awaited_once_with(index=default_indices) + get_mapping.assert_awaited_once_with(index=default_indices, filter_path="*.mappings._meta.src.nodenorm") def test_status_does_not_depend_on_cluster_node_access(self): self.client.cat_error = True @@ -424,7 +433,93 @@ def test_status_does_not_depend_on_cluster_node_access(self): assert "elasticsearch" not in body assert body["version"] == handlers.status.get_openapi_version() assert body["biolink_model_toolkit_version"] == "test" - assert self.client.calls == [("get_mapping", {"index": [DEFAULT_INDEX]})] + assert self.client.calls == [ + ("get_mapping", {"index": [DEFAULT_INDEX], "filter_path": "*.mappings._meta.src.nodenorm"}) + ] + + def test_status_reuses_health_check_and_keeps_request_headers_separate(self): + first = self.fetch("/status") + second = self.fetch("/status", headers={"Origin": "https://example.org"}) + + assert first.code == second.code == 200 + assert len(self.client.calls) == 1 + assert first.headers["Cache-Control"] == second.headers["Cache-Control"] == "no-store" + assert "Access-Control-Allow-Origin" not in first.headers + assert second.headers["Access-Control-Allow-Origin"] == "*" + + def test_status_exposes_retained_snapshot_after_discovery_failure(self): + self.namespace.elasticsearch.babel_checked_at = "2026-10-23T12:00:00+00:00" + self.namespace.elasticsearch.babel_refresh_error = {"code": "discovery_failed"} + + response = self.fetch("/status") + + assert response.code == 200 + babel = json.loads(response.body)["babel"] + assert babel["checked_at"] == "2026-10-23T12:00:00+00:00" + assert babel["refresh_error"] == {"code": "discovery_failed"} + assert babel["default_version"] == DEFAULT_VERSION + + def test_status_does_not_cache_success_for_the_ten_minute_discovery_interval(self): + with patch("nodenorm.status_health.monotonic", return_value=0): + assert self.fetch("/status").code == 200 + self.client.failed_index = DEFAULT_INDEX + with patch("nodenorm.status_health.monotonic", return_value=5): + failed = self.fetch("/status") + assert failed.code == 503 + assert failed.headers["Cache-Control"] == "no-store" + assert self.fetch("/status").code == 503 + assert len(self.client.calls) == 2 + self.client.failed_index = None + with patch("nodenorm.status_health.monotonic", return_value=6): + assert self.fetch("/status").code == 200 + + @gen_test + async def test_background_discovery_switches_status_and_routing_without_changing_inflight_request(self): + await self.http_client.fetch(self.get_url("/status")) + started = asyncio.Event() + release_request = asyncio.Event() + original_msearch = self.client.msearch + + async def delayed_msearch(**kwargs): + started.set() + await release_request.wait() + return await original_msearch(**kwargs) + + self.client.msearch = delayed_msearch + inflight = self.http_client.fetch(self.get_url("/get_setid?curie=INPUT:1")) + await started.wait() + old_snapshot = self.namespace.elasticsearch.babel + self.namespace.config.elasticsearch = {"ES_INDEX": "", "ES_ALIAS": "alias"} + self.client.indices.exists_alias = AsyncMock(return_value=True) + mappings = { + DEFAULT_INDEX: {"mappings": {"_meta": {"src": {"nodenorm": {"version": DEFAULT_VERSION}}}}}, + OLD_INDEX: {"mappings": {"_meta": {"src": {"nodenorm": {"version": OLD_VERSION}}}}}, + } + + async def get_mapping(*, index=None, **kwargs): + if index == ["alias"]: + return {OLD_INDEX: mappings[OLD_INDEX]} + return mappings if index is None else {name: mappings[name] for name in index} + + self.client.indices.get_mapping = AsyncMock(side_effect=get_mapping) + namespace = NodeNormalizationAPINamespace.__new__(NodeNormalizationAPINamespace) + namespace.config = self.namespace.config + namespace.elasticsearch = self.namespace.elasticsearch + assert await namespace.refresh_babel_releases() + assert old_snapshot.default_version == DEFAULT_VERSION + release_request.set() + previous = await inflight + assert self.assert_selected_release(previous, DEFAULT_VERSION)["normalized_curies"] == ["NEW:1"] + self.client.msearch = original_msearch + + current = await self.http_client.fetch(self.get_url("/get_setid?curie=INPUT:1")) + assert self.assert_selected_release(current, OLD_VERSION)["normalized_curies"] == ["OLD:1"] + status = await self.http_client.fetch(self.get_url("/status")) + babel = json.loads(status.body)["babel"] + assert babel["default_version"] == OLD_VERSION + assert babel["indices"][OLD_VERSION] == self.namespace.elasticsearch.indices == [OLD_INDEX] + assert babel["checked_at"] == self.namespace.elasticsearch.babel_checked_at + assert self.client.indices.get_mapping.await_args.kwargs["index"] == [OLD_INDEX] def test_status_reports_configured_environment_independently_of_request_host(self): for environment in ("CI", "Prod", "Test", "unknown"): diff --git a/tests/test_babel_refresh.py b/tests/test_babel_refresh.py new file mode 100644 index 0000000..d8e630f --- /dev/null +++ b/tests/test_babel_refresh.py @@ -0,0 +1,249 @@ +"""Offline coverage for timed Babel discovery and independent health checks.""" + +import asyncio +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest +from tornado.ioloop import IOLoop + +from nodenorm import namespace as namespace_module, status_health +from nodenorm.babel import BabelReleases +from nodenorm.namespace import NodeNormalizationAPINamespace +from nodenorm.server import NodeNormalizationWebServer +from nodenorm.status_health import StatusHealthCache + + +def mapping(version): + return {"mappings": {"_meta": {"src": {"nodenorm": {"version": version}}}}} + + +@pytest.fixture +def namespace(): + namespace = NodeNormalizationAPINamespace.__new__(NodeNormalizationAPINamespace) + namespace.config = SimpleNamespace(elasticsearch={"ES_INDEX": "", "ES_ALIAS": "alias"}) + namespace.elasticsearch = SimpleNamespace( + async_client=SimpleNamespace( + indices=SimpleNamespace( + exists=AsyncMock(return_value=False), + exists_alias=AsyncMock(return_value=True), + get_mapping=AsyncMock( + side_effect=[ + {"new": mapping("2026jul22")}, + {"old": mapping("2025sep1"), "new": mapping("2026jul22")}, + ] + ), + ) + ), + babel=BabelReleases("2025sep1", {"2025sep1": "old"}), + indices=["old"], + babel_checked_at="2026-10-23T12:00:00+00:00", + babel_refresh_error=None, + ) + return namespace + + +def test_server_starts_and_stops_background_discovery(namespace): + namespace.config.webserver = {"HOST": "localhost", "PORT": 9000, "ENABLE_CURL_CLIENT": False} + namespace.handlers = {} + loop = IOLoop(make_current=False) + loop.add_callback(loop.stop) + with ( + patch("nodenorm.server.tornado.httpserver.HTTPServer"), + patch("nodenorm.server.tornado.ioloop.IOLoop.instance", return_value=loop), + ): + NodeNormalizationWebServer(object(), namespace).start() + + assert namespace._babel_refresh_task is None + assert loop.asyncio_loop.is_closed() + + +async def test_refresh_follows_alias_and_publishes_matching_default_indices(namespace): + previous = namespace.elasticsearch.babel + + assert await namespace.refresh_babel_releases() + + current = namespace.elasticsearch.babel + assert current.default_version == "2026jul22" + assert current.resolve().search_indices == namespace.elasticsearch.indices == ["new"] + assert current.resolve("2025sep1").search_indices == ["old"] + assert previous.default_version == "2025sep1" + assert previous.resolve().search_indices == ["old"] + assert datetime.fromisoformat(namespace.elasticsearch.babel_checked_at).tzinfo is not None + assert namespace.elasticsearch.babel_refresh_error is None + indices = namespace.elasticsearch.async_client.indices + indices.exists.assert_not_awaited() + indices.exists_alias.assert_awaited_once_with(name="alias") + + +async def test_refresh_honors_configured_index_before_fallback_alias(namespace): + namespace.config.elasticsearch["ES_INDEX"] = "configured-index" + indices = namespace.elasticsearch.async_client.indices + indices.exists.return_value = True + + assert await namespace.refresh_babel_releases() + + indices.exists_alias.assert_not_awaited() + assert indices.get_mapping.await_args_list[0].kwargs["index"] == ["configured-index"] + + +@pytest.mark.parametrize( + "default_mappings", + [{}, {"old": mapping("2025sep1"), "new": mapping("2026jul22")}, {"bad": {"mappings": {}}}], +) +async def test_invalid_discovery_retains_entire_previous_snapshot(namespace, default_mappings): + previous = namespace.elasticsearch.babel + indices = namespace.elasticsearch.async_client.indices + indices.get_mapping.side_effect = [default_mappings, {"new": mapping("2026jul22")}] + + assert not await namespace.refresh_babel_releases() + + assert namespace.elasticsearch.babel is previous + assert namespace.elasticsearch.indices == ["old"] + assert namespace.elasticsearch.babel_checked_at == "2026-10-23T12:00:00+00:00" + assert namespace.elasticsearch.babel_refresh_error == {"code": "discovery_failed"} + + +async def test_discovery_failure_recovers_without_leaking_upstream_error(namespace): + indices = namespace.elasticsearch.async_client.indices + indices.get_mapping.side_effect = RuntimeError("private upstream connection details") + assert not await namespace.refresh_babel_releases() + assert namespace.elasticsearch.babel_refresh_error == {"code": "discovery_failed"} + indices.get_mapping.side_effect = [ + {"new": mapping("2026jul22")}, + {"new": mapping("2026jul22")}, + ] + + assert await namespace.refresh_babel_releases() + + assert namespace.elasticsearch.babel.default_version == "2026jul22" + assert namespace.elasticsearch.babel_refresh_error is None + + +async def test_background_refresh_runs_without_requests_and_retries_failure(namespace, monkeypatch): + monkeypatch.setattr(namespace_module, "BABEL_REFRESH_INTERVAL", 0.01) + monkeypatch.setattr(namespace_module, "BABEL_REFRESH_RETRY_INTERVAL", 0.001) + completed = asyncio.Event() + indices = namespace.elasticsearch.async_client.indices + + async def get_mapping(*, index=None, **kwargs): + if indices.get_mapping.await_count == 1: + raise RuntimeError("temporary outage") + return {"new": mapping("2026jul22")} + + indices.get_mapping.side_effect = get_mapping + refresh = namespace.refresh_babel_releases + + async def refresh_and_notify(): + refreshed = await refresh() + if refreshed: + completed.set() + return refreshed + + namespace.refresh_babel_releases = refresh_and_notify + namespace.start_babel_refresh() + task = namespace._babel_refresh_task + namespace.start_babel_refresh() + assert namespace._babel_refresh_task is task + try: + await asyncio.wait_for(completed.wait(), timeout=1) + assert namespace.elasticsearch.babel.default_version == "2026jul22" + assert namespace.elasticsearch.babel_refresh_error is None + assert indices.get_mapping.await_count == 3 + finally: + await namespace.stop_babel_refresh() + assert task.cancelled() + assert namespace._babel_refresh_task is None + + +async def test_background_shutdown_cancels_an_inflight_refresh(namespace, monkeypatch): + monkeypatch.setattr(namespace_module, "BABEL_REFRESH_INTERVAL", 0.001) + started = asyncio.Event() + cancelled = asyncio.Event() + + async def get_mapping(**kwargs): + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + indices = namespace.elasticsearch.async_client.indices + indices.get_mapping.side_effect = get_mapping + previous = namespace.elasticsearch.babel + namespace.start_babel_refresh() + try: + await asyncio.wait_for(started.wait(), timeout=1) + finally: + await namespace.stop_babel_refresh() + + assert cancelled.is_set() + assert indices.get_mapping.await_count == 1 + assert namespace.elasticsearch.babel is previous + assert namespace.elasticsearch.babel_refresh_error is None + + +async def test_discovery_deadline_retains_snapshot(namespace, monkeypatch): + monkeypatch.setattr(namespace_module, "BABEL_REFRESH_TIMEOUT", 0.001) + + async def never_returns(**kwargs): + await asyncio.Event().wait() + + namespace.elasticsearch.async_client.indices.get_mapping.side_effect = never_returns + previous = namespace.elasticsearch.babel + + assert not await asyncio.wait_for(namespace.refresh_babel_releases(), timeout=1) + assert namespace.elasticsearch.babel is previous + + +async def test_health_refresh_is_shared_and_survives_caller_cancellation(): + started = asyncio.Event() + release = asyncio.Event() + + async def get_mapping(**kwargs): + started.set() + await release.wait() + return {"old": mapping("2025sep1")} + + client = SimpleNamespace(indices=SimpleNamespace(get_mapping=AsyncMock(side_effect=get_mapping))) + cache = StatusHealthCache(client, BabelReleases("2025sep1", {"2025sep1": "old"}).resolve()) + first = asyncio.create_task(cache.check()) + second = asyncio.create_task(cache.check()) + await started.wait() + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + release.set() + + assert await second + assert await cache.check() + client.indices.get_mapping.assert_awaited_once() + + +@pytest.mark.parametrize( + "mappings", + [ + {}, + {"a": mapping("2025sep1")}, + {"a": mapping("2025sep1"), "b": mapping("2026jul22")}, + {"a": mapping("2025sep1"), "b": {"mappings": {}}}, + ], +) +async def test_health_validates_every_default_backing_index(mappings): + client = SimpleNamespace(indices=SimpleNamespace(get_mapping=AsyncMock(return_value=mappings))) + release = BabelReleases("2025sep1", {"2025sep1": ("a", "b")}).resolve() + + assert not await StatusHealthCache(client, release).check() + + +async def test_health_deadline_reports_failure(monkeypatch): + monkeypatch.setattr(status_health, "HEALTH_TIMEOUT", 0.001) + + async def never_returns(**kwargs): + await asyncio.Event().wait() + + client = SimpleNamespace(indices=SimpleNamespace(get_mapping=AsyncMock(side_effect=never_returns))) + release = BabelReleases("2025sep1", {"2025sep1": "old"}).resolve() + + assert not await asyncio.wait_for(StatusHealthCache(client, release).check(), timeout=1) diff --git a/tests/test_normalized_nodes_lookup.py b/tests/test_normalized_nodes_lookup.py index 92f14ae..2c25b71 100644 --- a/tests/test_normalized_nodes_lookup.py +++ b/tests/test_normalized_nodes_lookup.py @@ -538,7 +538,7 @@ def test_selected_release_also_routes_conflation_lookup(self): self.assert_release_response(response, "2026jul22", "NEW", conflated=True) assert [call["index"] for call in self.elasticsearch.calls] == [["nodenorm-new"], ["nodenorm-new"]] - def test_moving_default_alias_cannot_change_pinned_release_results(self): + def test_moving_default_alias_cannot_change_release_results_between_refreshes(self): def mapping(version): return {"mappings": {"_meta": {"src": {"nodenorm": {"version": version}}}}}