From 9a5c0a80744832f71f64df9915fe6dc5423f142c Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 18:31:12 +0100 Subject: [PATCH 1/3] Make health endpoint check database and storage; add Docker HEALTHCHECK Co-Authored-By: Claude Sonnet 5.5 --- Dockerfile | 2 + server/controllers/api/v1/health.py | 126 +++++++++++++++++- server/test/controllers/api/v1/test_health.py | 78 ++++++++++- 3 files changed, 201 insertions(+), 5 deletions(-) diff --git a/Dockerfile b/Dockerfile index 5d693dce2..01fc8ad86 100644 --- a/Dockerfile +++ b/Dockerfile @@ -39,4 +39,6 @@ COPY --from=build_v3 /server/static /server/static WORKDIR /server RUN mkdir conf EXPOSE 8080 +HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \ + CMD ["python3", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/api/v1/health', timeout=4)"] ENTRYPOINT ["python3", "main.py"] diff --git a/server/controllers/api/v1/health.py b/server/controllers/api/v1/health.py index 030f18fbe..4e0cad5b6 100644 --- a/server/controllers/api/v1/health.py +++ b/server/controllers/api/v1/health.py @@ -1,10 +1,130 @@ +import asyncio +import os +import time +from typing import Optional + +from sqlalchemy import text +from tornado.ioloop import IOLoop + +from digi_server.logger import get_logger from digi_server.settings import get_version from utils.web.base_controller import BaseAPIController from utils.web.route import ApiRoute, ApiVersion -@ApiRoute("health", ApiVersion.V1) +# Upper bound on the database probe so a locked database cannot hang the endpoint. +DB_CHECK_TIMEOUT_SECONDS = 3.0 + +# Settings naming a file whose parent directory must be writable. +_STORAGE_FILE_SETTINGS = ("log_path", "db_log_path", "client_log_path") +# Settings naming a directory that must be writable. +_STORAGE_DIR_SETTINGS = ("draft_script_path", "compiled_script_path") + +_STARTED_AT = time.monotonic() + + +def _writable(path: str) -> bool: + """Check that ``path`` (or its nearest existing ancestor) is writable. + + Directories that have not been created yet are judged by the closest + ancestor that exists, since the server would create them on demand. + + :param path: Directory path to check. + :returns: ``True`` if the directory could be written to. + """ + path = os.path.abspath(path) + while not os.path.exists(path): + parent = os.path.dirname(path) + if parent == path: + return False + path = parent + return os.path.isdir(path) and os.access(path, os.W_OK | os.X_OK) + + +@ApiRoute("health", ApiVersion.V1, ignore_logging=True) class HealthController(BaseAPIController): + """Unauthenticated liveness/readiness probe. + + Responds ``200`` with ``status: "ok"`` when the database and storage + locations are usable, otherwise ``503`` with ``status: "degraded"``. + Failure detail is only ever logged server-side, never returned. + """ + + async def prepare(self): + """Skip authentication so a broken database cannot fail the probe itself.""" + + def _check_database_sync(self) -> None: + with self.make_session() as session: + session.execute(text("SELECT 1")) + + async def _check_database(self) -> bool: + """Run ``SELECT 1`` off the event loop with a bounded wait. + + :returns: ``True`` if the query succeeded in time. + """ + try: + await asyncio.wait_for( + IOLoop.current().run_in_executor(None, self._check_database_sync), + timeout=DB_CHECK_TIMEOUT_SECONDS, + ) + except asyncio.TimeoutError: + get_logger().error( + f"Health check: database probe timed out after " + f"{DB_CHECK_TIMEOUT_SECONDS}s" + ) + return False + except Exception as e: # noqa: BLE001 + get_logger().error(f"Health check: database probe failed: {e!r}") + return False + return True + + @staticmethod + def _setting(settings, key: str) -> Optional[str]: + try: + return settings.get_sync(key) + except KeyError: + return None + + def _check_storage(self) -> bool: + """Check the log, draft and compiled script locations are writable. + + :returns: ``True`` if every configured location is writable. + """ + settings = self.application.digi_settings + targets = [] + for key in _STORAGE_FILE_SETTINGS: + value = self._setting(settings, key) + if value: + targets.append((key, os.path.dirname(os.path.abspath(value)))) + for key in _STORAGE_DIR_SETTINGS: + value = self._setting(settings, key) + if value: + targets.append((key, value)) + + ok = True + for key, directory in targets: + if not _writable(directory): + get_logger().warning( + f"Health check: storage for '{key}' is not writable: {directory}" + ) + ok = False + return ok + async def get(self): - health_data = {"version": get_version(), "status": "ok"} - await self.finish(health_data) + db_ok = await self._check_database() + storage_ok = self._check_storage() + healthy = db_ok and storage_ok + + if not healthy: + self.set_status(503) + await self.finish( + { + "version": get_version(), + "status": "ok" if healthy else "degraded", + "checks": { + "database": "ok" if db_ok else "fail", + "storage": "ok" if storage_ok else "fail", + }, + "uptime_seconds": int(time.monotonic() - _STARTED_AT), + } + ) diff --git a/server/test/controllers/api/v1/test_health.py b/server/test/controllers/api/v1/test_health.py index d448f3cef..01567effd 100644 --- a/server/test/controllers/api/v1/test_health.py +++ b/server/test/controllers/api/v1/test_health.py @@ -2,9 +2,13 @@ Tests for health check endpoint. """ +import time +from unittest.mock import patch + from tornado import escape from tornado.testing import gen_test +from controllers.api.v1.health import HealthController from test.conftest import DigiScriptTestCase @@ -44,8 +48,11 @@ async def test_health_endpoint_does_not_expose_sensitive_data(self): data = escape.json_decode(response.body) - # Verify ONLY version and status are present - self.assertEqual(2, len(data), "Health endpoint should return exactly 2 fields") + self.assertEqual( + {"version", "status", "checks", "uptime_seconds"}, + set(data), + "Health endpoint should only return the documented fields", + ) # Verify sensitive fields are NOT present sensitive_fields = [ @@ -77,3 +84,70 @@ async def test_health_endpoint_version_format(self): # Each part should be numeric for part in parts: self.assertTrue(part.isdigit(), f"Version part '{part}' should be numeric") + + @gen_test + async def test_health_reports_checks_ok(self): + """Healthy server reports ok for every check.""" + response = await self.http_client.fetch(self.get_url("/api/v1/health")) + + self.assertEqual(200, response.code) + data = escape.json_decode(response.body) + self.assertEqual({"database": "ok", "storage": "ok"}, data["checks"]) + self.assertIsInstance(data["uptime_seconds"], int) + + @gen_test + async def test_health_database_failure_returns_503_without_leaking(self): + """A failing database yields 503/degraded and no exception text.""" + with patch.object( + HealthController, + "_check_database_sync", + side_effect=RuntimeError("secret-db-detail /var/secret/path"), + ): + response = await self.http_client.fetch( + self.get_url("/api/v1/health"), raise_error=False + ) + + self.assertEqual(503, response.code) + data = escape.json_decode(response.body) + self.assertEqual("degraded", data["status"]) + self.assertEqual("fail", data["checks"]["database"]) + self.assertEqual("ok", data["checks"]["storage"]) + self.assertIn("version", data) + self.assertNotIn("secret-db-detail", response.body.decode()) + self.assertNotIn("/var/secret/path", response.body.decode()) + + @gen_test + async def test_health_database_timeout_returns_503(self): + """A hung database probe is bounded by the timeout.""" + + def hang(_self): + time.sleep(0.5) + + with ( + patch.object(HealthController, "_check_database_sync", hang), + patch("controllers.api.v1.health.DB_CHECK_TIMEOUT_SECONDS", 0.05), + ): + response = await self.http_client.fetch( + self.get_url("/api/v1/health"), raise_error=False + ) + + self.assertEqual(503, response.code) + data = escape.json_decode(response.body) + self.assertEqual("fail", data["checks"]["database"]) + + @gen_test + async def test_health_unwritable_storage_returns_503(self): + """An unwritable storage directory yields 503 without leaking paths.""" + with patch("controllers.api.v1.health.os.access", return_value=False): + response = await self.http_client.fetch( + self.get_url("/api/v1/health"), raise_error=False + ) + + self.assertEqual(503, response.code) + data = escape.json_decode(response.body) + self.assertEqual("degraded", data["status"]) + self.assertEqual("fail", data["checks"]["storage"]) + self.assertEqual("ok", data["checks"]["database"]) + for key in ("log_path", "draft_script_path", "compiled_script_path"): + path = self._app.digi_settings.get_sync(key) + self.assertNotIn(path, response.body.decode()) From 2ef3edf6c05dafc46ed8eb7af5d79e6cfde024ad Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 23:16:07 +0100 Subject: [PATCH 2/3] Harden health endpoint: single-flight DB probe, never serve tracebacks, stronger tests Co-Authored-By: Claude Sonnet 5.5 --- Dockerfile | 3 + server/controllers/api/v1/health.py | 69 ++++++++- server/test/controllers/api/v1/test_health.py | 140 +++++++++++++++--- 3 files changed, 188 insertions(+), 24 deletions(-) diff --git a/Dockerfile b/Dockerfile index 01fc8ad86..a63790101 100644 --- a/Dockerfile +++ b/Dockerfile @@ -39,6 +39,9 @@ COPY --from=build_v3 /server/static /server/static WORKDIR /server RUN mkdir conf EXPOSE 8080 +# NOTE: the probe targets port 8080 (the main.py --port default). If the container +# is started with a different --port, override the check with `docker run --health-cmd` +# or the compose `healthcheck:` key, otherwise it will be reported unhealthy. HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \ CMD ["python3", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/api/v1/health', timeout=4)"] ENTRYPOINT ["python3", "main.py"] diff --git a/server/controllers/api/v1/health.py b/server/controllers/api/v1/health.py index 4e0cad5b6..3fc7f4101 100644 --- a/server/controllers/api/v1/health.py +++ b/server/controllers/api/v1/health.py @@ -1,10 +1,10 @@ import asyncio import os import time +from concurrent.futures import Future, ThreadPoolExecutor from typing import Optional from sqlalchemy import text -from tornado.ioloop import IOLoop from digi_server.logger import get_logger from digi_server.settings import get_version @@ -22,6 +22,11 @@ _STARTED_AT = time.monotonic() +# A single dedicated worker so a hung probe can never occupy more than one +# thread, nor starve the shared default executor. +_PROBE_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="health-probe") +_probe: Optional[Future] = None + def _writable(path: str) -> bool: """Check that ``path`` (or its nearest existing ancestor) is writable. @@ -54,17 +59,35 @@ async def prepare(self): """Skip authentication so a broken database cannot fail the probe itself.""" def _check_database_sync(self) -> None: + """Read the schema table of the database on a pooled connection. + + Querying ``sqlite_master`` forces SQLite to read the database file + header and schema, so a locked or corrupt file fails here, unlike a + bare ``SELECT 1`` which may never touch the file. + + :raises Exception: Any database error, handled by the caller. + """ with self.make_session() as session: - session.execute(text("SELECT 1")) + session.execute(text("SELECT 1 FROM sqlite_master LIMIT 1")) async def _check_database(self) -> bool: - """Run ``SELECT 1`` off the event loop with a bounded wait. + """Probe the database off the event loop with a bounded wait. + + Probes are single-flight on a dedicated one-thread executor: a timed + out probe cannot be cancelled (its thread keeps running), so while one + is still in flight further calls report failure without starting + another. This caps leaked threads and connections at one. :returns: ``True`` if the query succeeded in time. """ + global _probe # noqa: PLW0603 + if _probe is not None and not _probe.done(): + get_logger().error("Health check: previous database probe still running") + return False + _probe = _PROBE_EXECUTOR.submit(self._check_database_sync) try: await asyncio.wait_for( - IOLoop.current().run_in_executor(None, self._check_database_sync), + asyncio.shield(asyncio.wrap_future(_probe)), timeout=DB_CHECK_TIMEOUT_SECONDS, ) except asyncio.TimeoutError: @@ -73,13 +96,21 @@ async def _check_database(self) -> bool: f"{DB_CHECK_TIMEOUT_SECONDS}s" ) return False - except Exception as e: # noqa: BLE001 - get_logger().error(f"Health check: database probe failed: {e!r}") + except Exception as e: + get_logger().error( + f"Health check: database probe failed: {e!r}", exc_info=True + ) return False return True @staticmethod def _setting(settings, key: str) -> Optional[str]: + """Read a setting without raising for unknown keys. + + :param settings: The application settings object. + :param key: Setting name. + :returns: The setting value, or ``None`` if the setting is not defined. + """ try: return settings.get_sync(key) except KeyError: @@ -112,7 +143,11 @@ def _check_storage(self) -> bool: async def get(self): db_ok = await self._check_database() - storage_ok = self._check_storage() + try: + storage_ok = self._check_storage() + except Exception: + get_logger().error("Health check: storage check raised", exc_info=True) + storage_ok = False healthy = db_ok and storage_ok if not healthy: @@ -128,3 +163,23 @@ async def get(self): "uptime_seconds": int(time.monotonic() - _STARTED_AT), } ) + + def write_error(self, status_code: int, **kwargs) -> None: + """Never serve a traceback or error text, even with ``serve_traceback`` on. + + Server errors become a minimal degraded body; client errors (such as + 405) keep Tornado's default handling. + + :param status_code: HTTP status of the error. + """ + if status_code < 500: + super().write_error(status_code, **kwargs) + return + self.set_status(503) + self.finish( + { + "version": get_version(), + "status": "degraded", + "checks": {"database": "unknown", "storage": "unknown"}, + } + ) diff --git a/server/test/controllers/api/v1/test_health.py b/server/test/controllers/api/v1/test_health.py index 01567effd..6ebb0521e 100644 --- a/server/test/controllers/api/v1/test_health.py +++ b/server/test/controllers/api/v1/test_health.py @@ -2,12 +2,16 @@ Tests for health check endpoint. """ -import time +import os +import tempfile +import threading from unittest.mock import patch from tornado import escape +from tornado.httpclient import HTTPRequest from tornado.testing import gen_test +from controllers.api.v1 import health from controllers.api.v1.health import HealthController from test.conftest import DigiScriptTestCase @@ -117,37 +121,139 @@ async def test_health_database_failure_returns_503_without_leaking(self): self.assertNotIn("/var/secret/path", response.body.decode()) @gen_test - async def test_health_database_timeout_returns_503(self): - """A hung database probe is bounded by the timeout.""" + async def test_health_database_timeout_is_single_flight(self): + """A hung probe yields 503 and later polls do not start another probe.""" + release = threading.Event() + calls = [] def hang(_self): - time.sleep(0.5) + calls.append(1) + release.wait(5) - with ( - patch.object(HealthController, "_check_database_sync", hang), - patch("controllers.api.v1.health.DB_CHECK_TIMEOUT_SECONDS", 0.05), - ): + try: + with ( + patch.object(HealthController, "_check_database_sync", hang), + patch("controllers.api.v1.health.DB_CHECK_TIMEOUT_SECONDS", 0.05), + ): + first = await self.http_client.fetch( + self.get_url("/api/v1/health"), raise_error=False + ) + second = await self.http_client.fetch( + self.get_url("/api/v1/health"), raise_error=False + ) + finally: + release.set() + if health._probe is not None: + health._probe.result(timeout=5) + + for response in (first, second): + self.assertEqual(503, response.code) + data = escape.json_decode(response.body) + self.assertEqual("fail", data["checks"]["database"]) + self.assertEqual(1, len(calls), "second poll must not start another probe") + + # Once the stuck probe has finished, the endpoint recovers. + recovered = await self.http_client.fetch(self.get_url("/api/v1/health")) + self.assertEqual(200, recovered.code) + + @gen_test + async def test_health_unwritable_storage_returns_503(self): + """An unwritable storage directory yields 503 without leaking paths.""" + with patch("controllers.api.v1.health._writable", return_value=False): response = await self.http_client.fetch( self.get_url("/api/v1/health"), raise_error=False ) self.assertEqual(503, response.code) data = escape.json_decode(response.body) - self.assertEqual("fail", data["checks"]["database"]) + self.assertEqual("degraded", data["status"]) + self.assertEqual("fail", data["checks"]["storage"]) + self.assertEqual("ok", data["checks"]["database"]) + body = response.body.decode() + for key in ( + "log_path", + "db_log_path", + "client_log_path", + "draft_script_path", + "compiled_script_path", + ): + self.assertNotIn(self._app.digi_settings.get_sync(key), body) + + def test_writable_uses_os_access_on_nearest_ancestor(self): + """Missing directories are judged by their nearest existing ancestor.""" + with tempfile.TemporaryDirectory() as tmp: + missing = os.path.join(tmp, "a", "b") + self.assertTrue(health._writable(missing)) + with patch("controllers.api.v1.health.os.access", return_value=False): + self.assertFalse(health._writable(missing)) @gen_test - async def test_health_unwritable_storage_returns_503(self): - """An unwritable storage directory yields 503 without leaking paths.""" - with patch("controllers.api.v1.health.os.access", return_value=False): + async def test_health_missing_storage_setting_is_skipped(self): + """A storage setting that is not defined (KeyError) is ignored.""" + self.assertIsNone(HealthController._setting(self._app.digi_settings, "nope")) + real = self._app.digi_settings.get_sync + + def fake(key): + if key == "draft_script_path": + raise KeyError(key) + return real(key) + + with patch.object(self._app.digi_settings, "get_sync", side_effect=fake): + response = await self.http_client.fetch(self.get_url("/api/v1/health")) + self.assertEqual(200, response.code) + + @gen_test + async def test_health_uptime_non_decreasing(self): + """uptime_seconds is a non-negative, non-decreasing integer.""" + first = escape.json_decode( + (await self.http_client.fetch(self.get_url("/api/v1/health"))).body + ) + second = escape.json_decode( + (await self.http_client.fetch(self.get_url("/api/v1/health"))).body + ) + self.assertGreaterEqual(first["uptime_seconds"], 0) + self.assertGreaterEqual(second["uptime_seconds"], first["uptime_seconds"]) + + @gen_test + async def test_health_storage_exception_does_not_leak(self): + """An unexpected storage-check error is caught and not echoed back.""" + self._app.settings["serve_traceback"] = True + with patch.object( + HealthController, + "_check_storage", + side_effect=RuntimeError("boom /secret/path"), + ): response = await self.http_client.fetch( self.get_url("/api/v1/health"), raise_error=False ) + self.assertEqual(503, response.code) + body = response.body.decode() + self.assertNotIn("boom", body) + self.assertNotIn("/secret/path", body) + self.assertNotIn("Traceback", body) + self.assertEqual("fail", escape.json_decode(response.body)["checks"]["storage"]) + @gen_test + async def test_health_unexpected_error_never_serves_traceback(self): + """Even with serve_traceback on, an escaped error yields minimal JSON.""" + self._app.settings["serve_traceback"] = True + with patch( + "controllers.api.v1.health.get_version", + side_effect=[RuntimeError("boom /secret/path"), "1.2.3"], + ): + response = await self.http_client.fetch( + self.get_url("/api/v1/health"), raise_error=False + ) self.assertEqual(503, response.code) + body = response.body.decode() + for leaked in ("boom", "/secret/path", "Traceback", "RuntimeError"): + self.assertNotIn(leaked, body) data = escape.json_decode(response.body) self.assertEqual("degraded", data["status"]) - self.assertEqual("fail", data["checks"]["storage"]) - self.assertEqual("ok", data["checks"]["database"]) - for key in ("log_path", "draft_script_path", "compiled_script_path"): - path = self._app.digi_settings.get_sync(key) - self.assertNotIn(path, response.body.decode()) + + @gen_test + async def test_health_client_errors_keep_default_handling(self): + """Non-5xx errors (e.g. unsupported method) are not turned into 503.""" + request = HTTPRequest(self.get_url("/api/v1/health"), method="POST", body="{}") + response = await self.http_client.fetch(request, raise_error=False) + self.assertLess(response.code, 500) From b5a2ee5404d298e952bd53f275c591cc99708a34 Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 23:50:29 +0100 Subject: [PATCH 3/3] Health: overlapping polls share the in-flight DB probe instead of failing Co-Authored-By: Claude Sonnet 5.5 --- server/controllers/api/v1/health.py | 16 +++--- server/test/controllers/api/v1/test_health.py | 50 ++++++++++++++++--- 2 files changed, 52 insertions(+), 14 deletions(-) diff --git a/server/controllers/api/v1/health.py b/server/controllers/api/v1/health.py index 3fc7f4101..d1fa0618a 100644 --- a/server/controllers/api/v1/health.py +++ b/server/controllers/api/v1/health.py @@ -75,19 +75,21 @@ async def _check_database(self) -> bool: Probes are single-flight on a dedicated one-thread executor: a timed out probe cannot be cancelled (its thread keeps running), so while one - is still in flight further calls report failure without starting - another. This caps leaked threads and connections at one. + is in flight, overlapping callers await that same probe (shielded, so + one caller's timeout never cancels it) rather than starting another. + This caps leaked threads and connections at one without reporting a + healthy database as failed for merely overlapping polls. Every caller + is still bounded by its own timeout. :returns: ``True`` if the query succeeded in time. """ global _probe # noqa: PLW0603 - if _probe is not None and not _probe.done(): - get_logger().error("Health check: previous database probe still running") - return False - _probe = _PROBE_EXECUTOR.submit(self._check_database_sync) + if _probe is None or _probe.done(): + _probe = _PROBE_EXECUTOR.submit(self._check_database_sync) + probe = _probe try: await asyncio.wait_for( - asyncio.shield(asyncio.wrap_future(_probe)), + asyncio.shield(asyncio.wrap_future(probe)), timeout=DB_CHECK_TIMEOUT_SECONDS, ) except asyncio.TimeoutError: diff --git a/server/test/controllers/api/v1/test_health.py b/server/test/controllers/api/v1/test_health.py index 6ebb0521e..8640bdcb9 100644 --- a/server/test/controllers/api/v1/test_health.py +++ b/server/test/controllers/api/v1/test_health.py @@ -2,9 +2,11 @@ Tests for health check endpoint. """ +import asyncio import os import tempfile import threading +import time from unittest.mock import patch from tornado import escape @@ -121,8 +123,34 @@ async def test_health_database_failure_returns_503_without_leaking(self): self.assertNotIn("/var/secret/path", response.body.decode()) @gen_test - async def test_health_database_timeout_is_single_flight(self): - """A hung probe yields 503 and later polls do not start another probe.""" + async def test_health_concurrent_polls_share_slow_healthy_probe(self): + """Overlapping polls during a slow but healthy probe all report ok.""" + calls = [] + + def slow(_self): + calls.append(1) + time.sleep(0.3) + + with patch.object(HealthController, "_check_database_sync", slow): + responses = await asyncio.gather( + *( + self.http_client.fetch( + self.get_url("/api/v1/health"), raise_error=False + ) + for _ in range(3) + ) + ) + + for response in responses: + self.assertEqual(200, response.code) + self.assertEqual( + "ok", escape.json_decode(response.body)["checks"]["database"] + ) + self.assertEqual(1, len(calls), "overlapping polls must share one probe") + + @gen_test + async def test_health_hung_probe_fails_all_callers_without_new_thread(self): + """A hung probe fails every caller within the timeout; no second probe.""" release = threading.Event() calls = [] @@ -133,24 +161,32 @@ def hang(_self): try: with ( patch.object(HealthController, "_check_database_sync", hang), - patch("controllers.api.v1.health.DB_CHECK_TIMEOUT_SECONDS", 0.05), + patch("controllers.api.v1.health.DB_CHECK_TIMEOUT_SECONDS", 0.1), ): + started = time.monotonic() first = await self.http_client.fetch( self.get_url("/api/v1/health"), raise_error=False ) - second = await self.http_client.fetch( - self.get_url("/api/v1/health"), raise_error=False + later = await asyncio.gather( + *( + self.http_client.fetch( + self.get_url("/api/v1/health"), raise_error=False + ) + for _ in range(2) + ) ) + elapsed = time.monotonic() - started finally: release.set() if health._probe is not None: health._probe.result(timeout=5) - for response in (first, second): + for response in (first, *later): self.assertEqual(503, response.code) data = escape.json_decode(response.body) self.assertEqual("fail", data["checks"]["database"]) - self.assertEqual(1, len(calls), "second poll must not start another probe") + self.assertLess(elapsed, 2.0) + self.assertEqual(1, len(calls), "no second probe while one is hung") # Once the stuck probe has finished, the endpoint recovers. recovered = await self.http_client.fetch(self.get_url("/api/v1/health"))