Run health alert checks concurrently

quantara/web_app/contract_tools/mixins/alert.py

@@ -2,9 +2,12 @@
 This module contains the alert mixin class for health ratio monitoring.
 """
 
+import asyncio
 import logging
 import os
 
+from prometheus_client import Counter
+
 from web_app.telegram.notifications import send_health_ratio_notification
 from web_app.contract_tools.mixins.health_ratio import HealthRatioMixin
 from web_app.db.crud import UserDBConnector
@@ -13,6 +16,13 @@
 
 logger = logging.getLogger(__name__)
 ALERT_THRESHOLD = float(os.getenv("HEALTH_RATIO_ALERT_THRESHOLD", "1.1"))
+ALERT_BATCH_CONCURRENCY = int(os.getenv("HEALTH_ALERT_BATCH_CONCURRENCY", "16"))
+ALERT_USER_TIMEOUT_SECONDS = float(os.getenv("HEALTH_ALERT_USER_TIMEOUT_SECONDS", "10"))
+HEALTH_ALERT_SWEEP_RESULTS = Counter(
+    "health_alert_sweep_results_total",
+    "Health-ratio alert sweep outcomes by user check result.",
+    ["result"],
+)
 
 
 class AlertMixin:
@@ -21,6 +31,41 @@ class AlertMixin:
     Handles health ratio monitoring and notification dispatch.
     """
 
+    @classmethod
+    async def _check_single_user_health_ratio(
+        cls,
+        contract_address: str,
+        telegram_id: int,
+        client,
+    ) -> bool:
+        if not contract_address:
+            HEALTH_ALERT_SWEEP_RESULTS.labels(result="failure").inc()
+            return False
+
+        try:
+            health_ratio_level, _ = await asyncio.wait_for(
+                HealthRatioMixin.get_health_ratio_and_tvl(contract_address, client),
+                timeout=ALERT_USER_TIMEOUT_SECONDS,
+            )
+        except Exception as e:
+            logger.error(
+                "Failed to get health ratio for %s: %s", contract_address, e
+            )
+            HEALTH_ALERT_SWEEP_RESULTS.labels(result="failure").inc()
+            return False
+
+        health_value = float(health_ratio_level)
+        if health_value < ALERT_THRESHOLD:
+            logger.info(
+                "Health ratio level for user %s is %s",
+                contract_address,
+                health_ratio_level,
+            )
+            await cls.send_notification(telegram_id, health_ratio_level)
+
+        HEALTH_ALERT_SWEEP_RESULTS.labels(result="success").inc()
+        return True
+
     @classmethod
     async def check_users_health_ratio_level(cls) -> None:
         """
@@ -33,24 +78,30 @@ async def check_users_health_ratio_level(cls) -> None:
         client = get_stellar_client()
         user_number = len([user for user, _ in users_data])
         logger.info(f"Found number of users for notifications: {user_number}")
-        for contract_address, telegram_id in users_data:
-            if not contract_address:
-                continue
-            try:
-                health_ratio_level, _ = \
-                    await HealthRatioMixin.get_health_ratio_and_tvl(contract_address, client)
-            except Exception as e:
-                logger.error(
-                    "Failed to get health ratio for %s: %s", contract_address, e
-                )
-                continue
 
-            health_value = float(health_ratio_level)
-            if health_value < ALERT_THRESHOLD:
-                logger.info(
-                    f"Health ratio level for user {contract_address} is {health_ratio_level}"
+        semaphore = asyncio.Semaphore(ALERT_BATCH_CONCURRENCY)
+
+        async def check_with_limit(contract_address: str, telegram_id: int) -> bool:
+            async with semaphore:
+                return await cls._check_single_user_health_ratio(
+                    contract_address,
+                    telegram_id,
+                    client,
                 )
-                await cls.send_notification(telegram_id, health_ratio_level)
+
+        results = await asyncio.gather(
+            *(
+                check_with_limit(contract_address, telegram_id)
+                for contract_address, telegram_id in users_data
+            ),
+            return_exceptions=True,
+        )
+        successes = sum(result is True for result in results)
+        failures = len(results) - successes
+        logger.info(
+            "health_ratio_alert_sweep_complete",
+            extra={"success": successes, "failure": failures},
+        )
 
     @staticmethod
     async def send_notification(telegram_id: int, health_ratio: float):
@@ -64,4 +115,4 @@ async def send_notification(telegram_id: int, health_ratio: float):
         await send_health_ratio_notification(telegram_id, health_ratio)
         logger.info(
             f"Notification sent to user {telegram_id} with health ratio {health_ratio}"
-        )
+        )

quantara/web_app/tests/test_alert_mixin.py

@@ -0,0 +1,102 @@
+import asyncio
+from contextlib import ExitStack
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from web_app.contract_tools.mixins.alert import AlertMixin
+
+USER_CONNECTOR = "web_app.contract_tools.mixins.alert.UserDBConnector"
+STELLAR_CLIENT = "web_app.contract_tools.mixins.alert.get_stellar_client"
+HEALTH_RATIO = (
+    "web_app.contract_tools.mixins.alert."
+    "HealthRatioMixin.get_health_ratio_and_tvl"
+)
+
+
+@pytest.mark.asyncio
+async def test_alert_sweep_continues_after_user_failure():
+    users = [("contract-ok", 1001), ("contract-fail", 1002), ("contract-low", 1003)]
+
+    async def fake_health_ratio(contract_address, client):
+        if contract_address == "contract-fail":
+            raise RuntimeError("rpc failed")
+        if contract_address == "contract-low":
+            return "1.0", "100"
+        return "1.5", "100"
+
+    with ExitStack() as stack:
+        connector = stack.enter_context(patch(USER_CONNECTOR))
+        stack.enter_context(patch(STELLAR_CLIENT, return_value=MagicMock()))
+        stack.enter_context(patch(HEALTH_RATIO, side_effect=fake_health_ratio))
+        send_notification = stack.enter_context(
+            patch.object(AlertMixin, "send_notification", new_callable=AsyncMock)
+        )
+        connector.return_value.get_users_for_notifications.return_value = users
+
+        await AlertMixin.check_users_health_ratio_level()
+
+    send_notification.assert_awaited_once_with(1003, "1.0")
+
+
+@pytest.mark.asyncio
+async def test_alert_sweep_runs_health_checks_concurrently():
+    users = [(f"contract-{idx}", idx) for idx in range(4)]
+    started = 0
+    peak_started = 0
+    release = asyncio.Event()
+
+    async def fake_health_ratio(contract_address, client):
+        nonlocal started, peak_started
+        started += 1
+        peak_started = max(peak_started, started)
+        if started == len(users):
+            release.set()
+        await release.wait()
+        return "1.5", "100"
+
+    with ExitStack() as stack:
+        stack.enter_context(
+            patch(
+                "web_app.contract_tools.mixins.alert.ALERT_BATCH_CONCURRENCY",
+                len(users),
+            )
+        )
+        connector = stack.enter_context(patch(USER_CONNECTOR))
+        stack.enter_context(patch(STELLAR_CLIENT, return_value=MagicMock()))
+        stack.enter_context(patch(HEALTH_RATIO, side_effect=fake_health_ratio))
+        stack.enter_context(
+            patch.object(AlertMixin, "send_notification", new_callable=AsyncMock)
+        )
+        connector.return_value.get_users_for_notifications.return_value = users
+
+        await asyncio.wait_for(AlertMixin.check_users_health_ratio_level(), timeout=1)
+
+    assert peak_started == len(users)
+
+
+@pytest.mark.asyncio
+async def test_single_user_timeout_does_not_stop_successes():
+    users = [("contract-timeout", 1001), ("contract-ok", 1002)]
+
+    async def fake_health_ratio(contract_address, client):
+        if contract_address == "contract-timeout":
+            await asyncio.sleep(1)
+        return "1.5", "100"
+
+    with ExitStack() as stack:
+        stack.enter_context(
+            patch(
+                "web_app.contract_tools.mixins.alert.ALERT_USER_TIMEOUT_SECONDS",
+                0.01,
+            )
+        )
+        connector = stack.enter_context(patch(USER_CONNECTOR))
+        stack.enter_context(patch(STELLAR_CLIENT, return_value=MagicMock()))
+        stack.enter_context(patch(HEALTH_RATIO, side_effect=fake_health_ratio))
+        stack.enter_context(
+            patch.object(AlertMixin, "send_notification", new_callable=AsyncMock)
+        )
+        connector.return_value.get_users_for_notifications.return_value = users
+
+        await AlertMixin.check_users_health_ratio_level()