From e2b04204aed219e546b432f56d6f3aafd25f5076 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Thu, 6 Aug 2026 14:59:39 +0200 Subject: [PATCH] feat(telegram): route Envio indexer problems to a dedicated channel Indexer staleness, an unreachable GraphQL endpoint and GraphQL errors are all the same operational problem for whoever runs the indexer, so they now land in one chat (TELEGRAM_CHAT_ID_ENVIO) instead of being spread across the general errors feed. send_envio_error_message falls back to send_error_message when the chat id is unset, so visibility is never lost. Only Envio failures are rerouted; every other operational error still goes to the errors channel. Co-Authored-By: Claude Opus 5 (1M context) --- .env.example | 7 ++ monitoring.yaml | 2 +- protocols/timelock/timelock_alerts.py | 6 +- protocols/yearn/README.md | 2 +- protocols/yearn/alert_large_flows.py | 4 +- protocols/yearn/check_indexer_freshness.py | 12 ++-- tests/test_indexer_freshness.py | 4 +- tests/test_utils.py | 82 +++++++++++++++++++++- utils/telegram.py | 70 ++++++++++++++---- 9 files changed, 160 insertions(+), 29 deletions(-) diff --git a/.env.example b/.env.example index c0b3a9ab..c7718797 100644 --- a/.env.example +++ b/.env.example @@ -50,6 +50,13 @@ TELEGRAM_CHAT_ID_DEFAULT=your-default-chat-id # TELEGRAM_CHAT_ID_ERRORS=your-errors-chat-id # TELEGRAM_BOT_TOKEN_ERRORS=your-errors-bot-token # optional; falls back to DEFAULT bot +# Dedicated Envio channel — Envio indexer problems (staleness per chain, an +# unreachable GraphQL endpoint, GraphQL errors) are routed here instead of the +# general errors channel, since they're actioned by whoever runs the indexer. +# A standalone chat served by the DEFAULT bot; if unset, these fall back to the +# errors channel, and from there to the originating protocol's chat. +# TELEGRAM_CHAT_ID_ENVIO=your-envio-chat-id + # Protocol-specific Telegram settings (legacy per-protocol chats) TELEGRAM_BOT_TOKEN_AAVE=your-aave-bot-token TELEGRAM_CHAT_ID_AAVE=your-aave-chat-id diff --git a/monitoring.yaml b/monitoring.yaml index 4124c657..6896acd3 100644 --- a/monitoring.yaml +++ b/monitoring.yaml @@ -462,7 +462,7 @@ protocols: - protocols/yearn/check_indexer_freshness.py monitors: - name: "Indexer Freshness" - description: "Envio indexer lag per chain; alerts the errors channel when a chain's newest indexed block is older than 60 minutes or the GraphQL endpoint is down" + description: "Envio indexer lag per chain; alerts the Envio channel when a chain's newest indexed block is older than 60 minutes or the GraphQL endpoint is down" - name: "Large Flows" description: "Deposit/withdrawal flows >=$500k USD (Katana withdrawals >=$50k; or 10% of vault totalSupply fallback for unpriced tokens)" - name: "Timelock Delay" diff --git a/protocols/timelock/timelock_alerts.py b/protocols/timelock/timelock_alerts.py index 0c5166eb..c72f4e83 100644 --- a/protocols/timelock/timelock_alerts.py +++ b/protocols/timelock/timelock_alerts.py @@ -20,7 +20,7 @@ from utils.logger import get_logger from utils.proxy import build_diff_url, detect_proxy_upgrade, get_current_implementation from utils.safe_tx import unwrap_safe_exec_transaction -from utils.telegram import MAX_MESSAGE_LENGTH, escape_markdown, send_error_message, send_telegram_message +from utils.telegram import MAX_MESSAGE_LENGTH, escape_markdown, send_envio_error_message, send_telegram_message from utils.web3_wrapper import ChainManager load_dotenv() @@ -650,14 +650,14 @@ def main() -> None: msg = "⚠️ Timelock alerts: Envio API is unreachable after 3 retries" _logger.error(msg) try: - send_error_message(msg, "timelock") + send_envio_error_message(msg, "timelock") except Exception: _logger.exception("Failed to send Envio error alert") return if "errors" in response: msg = f"Timelock alerts: GraphQL errors: {response['errors']}" _logger.error(msg) - send_error_message(msg, "timelock") + send_envio_error_message(msg, "timelock") return data = response.get("data", {}) diff --git a/protocols/yearn/README.md b/protocols/yearn/README.md index 1fa87119..ea777084 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -262,7 +262,7 @@ Step 3 covers the inverse trap: an empty result set is not good news. If a chain ### Alerts -All alerts go to the errors channel (`TELEGRAM_*_ERRORS`) labelled `[yearn]`, alongside the other operational diagnostics: +This monitor only ever reports Envio indexer problems, so all of its alerts go to the Envio chat (`TELEGRAM_CHAT_ID_ENVIO`) labelled `[yearn]`, alongside the other indexer failures (large flows, timelock). Every other yearn monitor's operational error still goes to the errors channel. If `TELEGRAM_CHAT_ID_ENVIO` is unset these fall back to the errors channel, and from there to the protocol's own chat. - **Stale or missing chains** — one message listing every lagging chain with its lag and last indexed block, plus every expected chain the indexer reported no sync state for. - **Indexer unavailable** — the GraphQL endpoint is unset, unreachable, returned errors, or reported no chains. Sent on every run for as long as it lasts. diff --git a/protocols/yearn/alert_large_flows.py b/protocols/yearn/alert_large_flows.py index 7e5f7f0d..fb63daa1 100644 --- a/protocols/yearn/alert_large_flows.py +++ b/protocols/yearn/alert_large_flows.py @@ -16,7 +16,7 @@ from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file from utils.chains import EXPLORER_URLS, Chain from utils.defillama import fetch_prices -from utils.telegram import send_error_message +from utils.telegram import send_envio_error_message from utils.web3_wrapper import ChainManager load_dotenv() @@ -208,7 +208,7 @@ def gql_request(query: str, variables: dict) -> dict | None: try: return http_json(ENVIO_GRAPHQL_URL, method="POST", body=payload) except urllib.error.HTTPError as exc: - send_error_message( + send_envio_error_message( f"⚠️ Large Flow Alert: Envio GraphQL error (HTTP {exc.code}). Skipping this run.", PROTOCOL, ) diff --git a/protocols/yearn/check_indexer_freshness.py b/protocols/yearn/check_indexer_freshness.py index 9d3c3c68..06fd877d 100644 --- a/protocols/yearn/check_indexer_freshness.py +++ b/protocols/yearn/check_indexer_freshness.py @@ -9,7 +9,7 @@ This check reads `chain_metadata` from the indexer and, for every chain this repo monitors, resolves the wall-clock timestamp of `latest_processed_block` from an RPC. Any chain whose newest indexed block is older than the lag threshold -(default 60 minutes) is reported to the errors channel — as is any expected chain +(default 60 minutes) is reported to the envio channel — as is any expected chain the indexer reports no sync state for at all, since "nothing was stale" must never be mistaken for "everything is fresh". @@ -36,7 +36,7 @@ from utils.formatting import format_duration from utils.http_client import request_with_retry from utils.logger import get_logger -from utils.telegram import send_error_message +from utils.telegram import send_envio_error_message from utils.web3_wrapper import ChainManager load_dotenv() @@ -54,7 +54,7 @@ DEFAULT_MAX_LAG_MINUTES = 60 # Staleness persists for as long as the indexer takes to catch up (a re-sync can -# run for days), so re-alerting every hourly run would bury the errors channel. +# run for days), so re-alerting every hourly run would bury the envio channel. # Each chain alerts on the way into staleness, then at most once per cooldown # window, then once more when it recovers. DEFAULT_ALERT_COOLDOWN_HOURS = 6 @@ -282,7 +282,7 @@ def report_recovered(fresh: list[ChainFreshness]) -> None: if not recovered: return names = ", ".join(f"{chain.name} ({format_duration(chain.lag_seconds or 0)} behind)" for chain in recovered) - send_error_message(f"Envio indexer caught up: {names}", PROTOCOL, source="indexer_freshness") + send_envio_error_message(f"Envio indexer caught up: {names}", PROTOCOL, source="indexer_freshness") for chain in recovered: _set_last_alert_timestamp(chain.chain.chain_id, 0) @@ -299,7 +299,7 @@ def main() -> None: # The endpoint being down is itself the outage we are watching for, so it # alerts on every run rather than riding the per-chain cooldown. logger.error("Indexer unavailable: %s", exc) - send_error_message( + send_envio_error_message( f"Envio indexer unavailable: {exc}\nDashboard: {DASHBOARD_URL}", PROTOCOL, source="indexer_freshness", @@ -324,7 +324,7 @@ def main() -> None: logger.info("All %d unhealthy chain(s) already alerted within the cooldown window", len(stale) + len(missing)) return - send_error_message( + send_envio_error_message( build_alert_message(stale_to_alert, missing_to_alert, max_lag_seconds), PROTOCOL, source="indexer_freshness", diff --git a/tests/test_indexer_freshness.py b/tests/test_indexer_freshness.py index 599f0d08..011fb6b8 100644 --- a/tests/test_indexer_freshness.py +++ b/tests/test_indexer_freshness.py @@ -63,9 +63,9 @@ def envio_url(monkeypatch: pytest.MonkeyPatch) -> str: @pytest.fixture def sent(monkeypatch: pytest.MonkeyPatch) -> list[str]: - """Capture every message the monitor routes to the errors channel.""" + """Capture every message the monitor routes to the envio channel.""" messages: list[str] = [] - monkeypatch.setattr(freshness, "send_error_message", lambda msg, protocol, **kwargs: messages.append(msg)) + monkeypatch.setattr(freshness, "send_envio_error_message", lambda msg, protocol, **kwargs: messages.append(msg)) return messages diff --git a/tests/test_utils.py b/tests/test_utils.py index 85acdd8f..a43515cc 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -14,7 +14,7 @@ from utils.alert import Alert, AlertSeverity, register_alert_hook, send_alert from utils.config import Config, ProtocolConfig -from utils.telegram import TelegramError, send_error_message, send_telegram_message +from utils.telegram import TelegramError, send_envio_error_message, send_error_message, send_telegram_message from utils.web3_wrapper import ( MAX_BACKOFF_SECONDS, ProviderConnectionError, @@ -385,6 +385,86 @@ def test_falls_back_to_protocol_channel_when_unconfigured(self, mock_post): self.assertNotIn("parse_mode", json_body) # plain text +class TestSendEnvioErrorMessage(unittest.TestCase): + """Tests for utils.telegram.send_envio_error_message (dedicated envio channel).""" + + @staticmethod + def _ok_response(mock_post): + mock_response = unittest.mock.Mock() + mock_response.status_code = 200 + mock_response.raise_for_status = unittest.mock.Mock() + mock_post.return_value = mock_response + + @patch("utils.telegram.requests.post") + def test_routes_to_envio_chat_with_label_silent_plain(self, mock_post): + """With an envio chat configured, the message goes there labelled, silent, plain.""" + self._ok_response(mock_post) + + with patch.dict( + os.environ, + { + "TELEGRAM_TEST_CHAT_ID": "", + "TELEGRAM_CHAT_ID_ENVIO": "envio_chat_id", + # The errors channel must not win — envio problems have their own chat. + "TELEGRAM_CHAT_ID_ERRORS": "errors_chat_id", + "TELEGRAM_BOT_TOKEN_DEFAULT": "default_token", + "LOG_LEVEL": "INFO", + }, + ): + send_envio_error_message("Indexer stale on Mainnet", "yearn") + + url = mock_post.call_args[0][0] + json_body = mock_post.call_args[1]["json"] + self.assertIn("default_token", url) + self.assertEqual(json_body["chat_id"], "envio_chat_id") + self.assertNotIn("message_thread_id", json_body) # standalone chat, no topic + self.assertEqual(json_body["text"], "[yearn] Indexer stale on Mainnet") + self.assertTrue(json_body["disable_notification"]) + self.assertNotIn("parse_mode", json_body) # plain text + + @patch("utils.telegram.requests.post") + def test_labels_originating_protocol(self, mock_post): + """Every monitor's envio problems land in one chat, labelled by origin.""" + self._ok_response(mock_post) + + with patch.dict( + os.environ, + { + "TELEGRAM_TEST_CHAT_ID": "", + "TELEGRAM_CHAT_ID_ENVIO": "envio_chat_id", + "TELEGRAM_BOT_TOKEN_DEFAULT": "default_token", + "LOG_LEVEL": "INFO", + }, + ): + send_envio_error_message("GraphQL boom", "timelock") + + json_body = mock_post.call_args[1]["json"] + self.assertEqual(json_body["chat_id"], "envio_chat_id") + self.assertEqual(json_body["text"], "[timelock] GraphQL boom") + + @patch("utils.telegram.requests.post") + def test_falls_back_to_errors_channel_when_unconfigured(self, mock_post): + """With TELEGRAM_CHAT_ID_ENVIO unset, the alert routes to the errors channel.""" + self._ok_response(mock_post) + + with patch.dict( + os.environ, + { + "TELEGRAM_TEST_CHAT_ID": "", + "TELEGRAM_CHAT_ID_ENVIO": "", + "TELEGRAM_TOPIC_ID_ERRORS": "", + "TELEGRAM_CHAT_ID_ERRORS": "errors_chat_id", + "TELEGRAM_BOT_TOKEN_DEFAULT": "default_token", + "LOG_LEVEL": "INFO", + }, + ): + send_envio_error_message("GraphQL boom", "yearn") + + json_body = mock_post.call_args[1]["json"] + self.assertEqual(json_body["chat_id"], "errors_chat_id") + self.assertEqual(json_body["text"], "[yearn] GraphQL boom") + + class TestAlert(unittest.TestCase): """Tests for the Alert system.""" diff --git a/utils/telegram.py b/utils/telegram.py index d598929a..d66ed212 100644 --- a/utils/telegram.py +++ b/utils/telegram.py @@ -21,6 +21,14 @@ # optional TELEGRAM_BOT_TOKEN_ERRORS (falls back to the DEFAULT bot). ERROR_CHANNEL = "errors" +# Channel key for Envio indexer problems (staleness, unreachable GraphQL endpoint, +# GraphQL errors). The indexer is a shared dependency of several monitors, and its +# outages are actioned by whoever runs the indexer rather than by protocol owners, +# so they get their own chat instead of being mixed into the general errors feed. +# Destination is a standalone chat, TELEGRAM_CHAT_ID_ENVIO, served by the DEFAULT +# bot — no topic thread and no dedicated bot token. +ENVIO_CHANNEL = "envio" + # Matches `bot:` in Telegram API URLs. Used to scrub the bot # token out of exception messages — `requests.HTTPError.__str__()` includes # the full URL, so without this the token leaks into any log or alert that @@ -307,9 +315,22 @@ def _update_alert_delivery_safe( logger.debug("Failed to update alert delivery", exc_info=True) -def _error_channel_configured() -> bool: - """Return True if a dedicated errors destination (topic or chat id) is set.""" - return bool(os.getenv("TELEGRAM_TOPIC_ID_ERRORS") or os.getenv("TELEGRAM_CHAT_ID_ERRORS")) +def _channel_configured(channel: str) -> bool: + """Return True if a dedicated destination (topic or chat id) is set for a channel.""" + return bool(os.getenv(f"TELEGRAM_TOPIC_ID_{channel.upper()}") or os.getenv(f"TELEGRAM_CHAT_ID_{channel.upper()}")) + + +def _send_labelled(message: str, protocol: str, channel: str, disable_notification: bool, source: str) -> None: + """Send a `[protocol]`-labelled plain-text message to a shared channel.""" + send_telegram_message( + f"[{protocol}] {escape_markdown(message)}", + channel, + disable_notification, + plain_text=True, + source=source, + origin_protocol=protocol, + channel=channel, + ) def send_error_message( @@ -336,16 +357,8 @@ def send_error_message( and as the fallback channel when no errors destination is configured. disable_notification: If True (default), send silently. """ - if _error_channel_configured(): - send_telegram_message( - f"[{protocol}] {escape_markdown(message)}", - ERROR_CHANNEL, - disable_notification, - plain_text=True, - source=source, - origin_protocol=protocol, - channel=ERROR_CHANNEL, - ) + if _channel_configured(ERROR_CHANNEL): + _send_labelled(message, protocol, ERROR_CHANNEL, disable_notification, source) else: send_telegram_message( escape_markdown(message), @@ -358,6 +371,37 @@ def send_error_message( ) +def send_envio_error_message( + message: str, + protocol: str, + disable_notification: bool = True, + *, + source: str = "envio_error", +) -> None: + """Route an Envio indexer problem to the dedicated envio channel. + + Indexer staleness, an unreachable GraphQL endpoint and GraphQL errors are all + the same operational problem for whoever runs the indexer, so they land in one + chat instead of being spread across the per-protocol groups and the general + errors feed. The originating ``protocol`` is prefixed as a ``[label]`` so the + merged feed shows which monitor hit the problem. + + Falls back to :func:`send_error_message` (and from there to the protocol's own + channel) when ``TELEGRAM_CHAT_ID_ENVIO`` is unset, so visibility is never lost. + + Args: + message: The error/diagnostic text. + protocol: Originating protocol, used as the ``[label]`` prefix and for the + fallback routing. + disable_notification: If True (default), send silently. + source: Alert source tag recorded with the alert. + """ + if os.getenv(f"TELEGRAM_CHAT_ID_{ENVIO_CHANNEL.upper()}"): + _send_labelled(message, protocol, ENVIO_CHANNEL, disable_notification, source) + else: + send_error_message(message, protocol, disable_notification, source=source) + + def get_github_run_url() -> str: """Build a GitHub Actions run URL from environment variables, if available.""" run_url = os.getenv("GITHUB_RUN_URL", "")